Build an AI Chat API with Memory

Wire FastAPI, PostgreSQL, Redis, and LangChain into a containerized AI chat backend.

Introduction

30 Second Summary

Chat applications are everywhere. Every time you message a support bot or ask an AI a question, there's a backend managing the conversation, remembering what you said, and making sure no single user overwhelms the system.

In this project, you will build a Dockerized AI chat backend API that handles multi-turn conversations with persistent memory and enforces per-user rate limits.

What You'll Build

A multi-container backend application using FastAPI, PostgreSQL, Redis, LangChain, and Docker Compose that powers conversational AI with chat history and rate limiting.

By the end of this project, you'll have:

  • A Docker Compose setup orchestrating three services: a FastAPI app, PostgreSQL, and Redis.
  • A conversational AI API powered by LangChain and Groq that persists full chat history in PostgreSQL.
  • A Redis-based rate limiter using a sliding window to enforce 10 requests per minute per user.
  • Secret Mission: Add a streaming chat endpoint that delivers AI responses token-by-token via Server-Sent Events.

Are there any prerequisites?

This project assumes you are comfortable writing Python and have Docker Desktop installed on your machine. You will also need a free Groq account (for the AI model) and a GitHub account (for version control).

Set Up the Project and Docker Compose

Every backend service needs infrastructure to lean on. Your AI chat API will need a database to store conversations, a cache to enforce rate limits, and a web server to handle requests.

In this step, you'll define all three services as code using Docker Compose. By the end, you'll have PostgreSQL, Redis, and a FastAPI app container all orchestrated from a single configuration file.

In this step, get ready to:
  • Create the project folder and initialize Git.
  • Define the Docker Compose services and Dockerfile.
  • Verify PostgreSQL and Redis containers start with health checks.
Create the project folder and infrastructure files

Before writing any application code, you need a folder to hold the project and Git to track changes from the start.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder, move into it, and initialize a Git repository by running these commands:
mkdir ai-chat-api
cd ai-chat-api
git init
  • Confirm the folder exists by running:
ls

You should see the ai-chat-api folder listed on your Desktop. Inside it, Git is now tracking changes.

Command not found for git?

If git init fails, you may not have Git installed. Download it from git-scm.com and run the installer. Restart your terminal after installing.

Help me install Git

Now you'll create the Docker Compose file. This single file defines all three services your application needs: a PostgreSQL database, a Redis cache, and the FastAPI app container.

  • Create a new file called docker-compose.yml in your ai-chat-api folder and paste this content:
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: chatapp
      POSTGRES_PASSWORD: chatapp_secret
      POSTGRES_DB: chatdb
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U chatapp -d chatdb"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5

  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql+asyncpg://chatapp:chatapp_secret@postgres:5432/chatdb
      REDIS_URL: redis://redis:6379/0
      GROQ_API_KEY: ${GROQ_API_KEY}
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy

volumes:
  postgres_data:

What does this file do?

This file defines three services that Docker Compose will manage together:

  • postgres runs PostgreSQL 16 with a database called chatdb. The healthcheck block tells Docker how to confirm the database is accepting connections.
  • redis runs Redis 7 as an in-memory key-value store. Its health check uses redis-cli ping to verify it responds.
  • app builds your FastAPI container from the local Dockerfile. The depends_on block ensures it only starts after postgres and redis are healthy.
  • ${GROQ_API_KEY} tells Docker Compose to read your API key from the .env file in the project folder.
  • Save docker-compose.yml.

Next, you need a Dockerfile that tells Docker how to build the FastAPI app container.

  • Create a new file called Dockerfile (no file extension) in your ai-chat-api folder and paste this content:
FROM python:3.10-slim

WORKDIR /code

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ ./app/

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

What does the Dockerfile do?

Each line is a step in building your container image:

  • FROM python:3.10-slim starts from a lightweight Python 3.10 base image.
  • COPY requirements.txt and RUN pip install install your dependencies first. This layer gets cached, so rebuilds are fast when only your code changes.
  • CMD starts the FastAPI server with uvicorn and enables hot-reloading for development.
  • Save Dockerfile.

Finally, you need a placeholder for the Python app package so Docker can copy it during the build.

  • Create the app directory and an empty __init__.py file by running:
mkdir app
touch app/__init__.py

Windows users without touch

If touch is not recognized, use this PowerShell alternative:

mkdir app
New-Item app/__init__.py
  • Confirm the structure by running:
ls

You should see app, docker-compose.yml, and Dockerfile listed.

✔️ Awesome, I've got everything!

Great. Make sure all three files are saved before moving on.

ⓧ I'd like to double check the full code

Here is docker-compose.yml:

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: chatapp
      POSTGRES_PASSWORD: chatapp_secret
      POSTGRES_DB: chatdb
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U chatapp -d chatdb"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5

  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql+asyncpg://chatapp:chatapp_secret@postgres:5432/chatdb
      REDIS_URL: redis://redis:6379/0
      GROQ_API_KEY: ${GROQ_API_KEY}
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy

volumes:
  postgres_data:

Here is Dockerfile:

FROM python:3.10-slim

WORKDIR /code

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ ./app/

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
Add dependencies and environment configuration

Your Dockerfile installs packages from a requirements.txt file. You also need an .env file so Docker Compose can pass your Groq API key into the app container.

  • Create a new file called requirements.txt in your ai-chat-api folder and paste this content:
fastapi[standard]==0.136.3
uvicorn[standard]==0.34.0
sqlalchemy[asyncio]==2.0.50
asyncpg==0.31.0
redis==6.1.0
langchain-core==1.4.6
langchain-groq==1.1.2
pydantic-settings==2.9.1

What are these packages?

  • fastapi and uvicorn power the web server and API framework.
  • sqlalchemy and asyncpg handle async database operations with PostgreSQL.
  • redis is the Python client for Redis, used for rate limiting.
  • langchain-core and langchain-groq provide the LLM integration with Groq's API.
  • pydantic-settings loads environment variables with type validation.
  • Save requirements.txt.

Seeing a typo in a package name?

Every package name and version must match exactly. A single typo will cause the Docker build to fail. Double check that fastapi[standard] has square brackets and that all version numbers use == (double equals).

Help me fix my requirements.txt

Now create the environment file that Docker Compose reads to inject your Groq API key.

  • Create a new file called .env in your ai-chat-api folder and paste this content:
GROQ_API_KEY=your-api-key-here
  • Replace your-api-key-here with your actual Groq API key from console.groq.com.
  • Save .env.

Why a .env file?

Docker Compose automatically reads variables from a .env file in the same folder as docker-compose.yml. The ${GROQ_API_KEY} in your compose file gets replaced with the real value at runtime. This keeps secrets out of version-controlled files.

Time to start the containers. Make sure Docker Desktop is running before you proceed.

  • Build and start all services by running:
docker compose up --build -d
  • Wait about 30 seconds for the images to download and containers to start, then check their status by running:
docker compose ps

You should see postgres and redis with a status of healthy. The app container will build successfully but show as restarting. That is expected because you have not written the FastAPI application code yet.

Why is the app container restarting?

The Dockerfile tells uvicorn to run app.main:app, but that file does not exist yet. Docker keeps restarting the container hoping it will succeed.

In the next step, you will create app/main.py and all three containers will run smoothly.

Docker compose command not found?

Make sure Docker Desktop is running. You should see the Docker icon in your system tray (Windows) or menu bar (macOS). If the command still fails, restart Docker Desktop and try again.

Help me troubleshoot Docker

✔️ Awesome, I've got everything!

Great. Make sure you see postgres and redis as healthy in the output.

ⓧ I'd like to double check the full code

Here is requirements.txt:

fastapi[standard]==0.136.3
uvicorn[standard]==0.34.0
sqlalchemy[asyncio]==2.0.50
asyncpg==0.31.0
redis==6.1.0
langchain-core==1.4.6
langchain-groq==1.1.2
pydantic-settings==2.9.1

Here is .env:

GROQ_API_KEY=your-api-key-here
Make your initial commit

Your infrastructure is defined. Lock it in with a Git commit so you have a clean starting point to return to.

  • Stage all project files and commit them by running these commands:
git add .
git commit -m "Initial project setup with Docker Compose"
  • Confirm the commit was created by running:
git log --oneline

You should see one commit with the message Initial project setup with Docker Compose.

Git asking for your name and email?

If Git prompts you to set your identity, run these two commands first (replacing the placeholder values with your own name and email):

git config --global user.email "you@example.com"
git config --global user.name "Your Name"

Then re-run the git commit command above.

Help me configure Git

Your project skeleton is ready. Next up, you'll build the FastAPI application with a health check endpoint that proves the app container can talk to both PostgreSQL and Redis.

Build the FastAPI App and Health Check

Your Docker containers are running, but the FastAPI app container has no code to serve yet. It needs a way to load configuration, connect to PostgreSQL and Redis, and prove those connections work.

In this step, you'll create a settings module that reads environment variables, set up database connections, and build a health check endpoint that confirms all three services can talk to each other.

In this step, get ready to:
  • Create a settings configuration that loads environment variables.
  • Build the FastAPI app with a health check endpoint.
  • Rebuild the Docker image and verify both databases respond.
Create the settings configuration

Your app needs to know where PostgreSQL, Redis, and the Groq API live. Instead of hardcoding connection strings, you'll use pydantic-settings to load them from environment variables automatically.

  • Create a new file called app/config.py inside your app/ folder.
  • Add the following code to app/config.py:
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Connection URLs for PostgreSQL and Redis
    database_url: str = "postgresql+asyncpg://chatapp:chatapp_secret@localhost:5432/chatdb"
    redis_url: str = "redis://localhost:6379/0"
    # Groq API key for the LLM
    groq_api_key: str = ""
    # Rate limiting configuration
    rate_limit_requests: int = 10
    rate_limit_window_seconds: int = 60

    class Config:
        env_file = ".env"


settings = Settings()

What does this code do?

The Settings class defines every configuration value your app needs, with sensible defaults for local development.

  • database_url and redis_url tell the app where to find each database.
  • rate_limit_requests and rate_limit_window_seconds control how many requests a user can make per time window.
  • The Config inner class tells pydantic-settings to also read from the .env file.

When Docker Compose passes environment variables like DATABASE_URL, pydantic-settings automatically maps them to the matching field names (case-insensitive).

  • Save app/config.py.

Why pydantic-settings?

FastAPI's own documentation recommends pydantic-settings for managing environment variables with type validation. It gives you type checking, default values, and .env file support in one package.

Create the FastAPI app and health endpoint

The health endpoint needs to reach PostgreSQL and Redis. Before writing main.py, you'll create two small connection modules that it depends on.

  • Create a new file called app/database.py and add this code:
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.config import settings

# Create the async database engine using the URL from settings
engine = create_async_engine(settings.database_url, echo=False)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_db() -> AsyncSession:
    async with async_session() as session:
        yield session

What does this code do?

  • create_async_engine creates an async connection pool to PostgreSQL using the asyncpg driver.
  • async_sessionmaker is a factory that produces database sessions for each request.
  • get_db is a FastAPI dependency that opens a session, yields it to the endpoint, then closes it automatically.
  • Save app/database.py.
  • Create another new file called app/rate_limiter.py and add this code:
import redis.asyncio as aioredis
from app.config import settings

# Create the async Redis client using the URL from settings
redis_client = aioredis.from_url(settings.redis_url, decode_responses=True)

What does this code do?

This creates a Redis client connected to your Redis container. The decode_responses=True flag means Redis returns strings instead of raw bytes.

You'll expand this file with rate limiting logic in a later step. For now, the health endpoint just needs the client to call ping().

  • Save app/rate_limiter.py.
  • Now create app/main.py with the FastAPI application and health check endpoint:
from fastapi import FastAPI, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.rate_limiter import redis_client

app = FastAPI(title="AI Chat API")


@app.get("/health")
async def health(db: AsyncSession = Depends(get_db)):
    # Check PostgreSQL connectivity
    postgres_ok = False
    redis_ok = False

    try:
        await db.execute(text("SELECT 1"))
        postgres_ok = True
    except Exception:
        pass

    # Check Redis connectivity
    try:
        await redis_client.ping()
        redis_ok = True
    except Exception:
        pass

    return {"postgres": postgres_ok, "redis": redis_ok}

What does this code do?

  • Depends(get_db) injects a database session into the endpoint automatically. FastAPI handles opening and closing it.
  • text("SELECT 1") runs a trivial SQL query to prove PostgreSQL is reachable.
  • redis_client.ping() sends a PING command to Redis and expects a PONG back.
  • If either connection fails, the try/except block catches the error and leaves the status as False instead of crashing the entire endpoint.
  • Save app/main.py.

✔️ Awesome, I've got everything!

Great. Make sure all four files are saved before moving on.

ⓧ I'd like to double check the full code

Here is app/config.py:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://chatapp:chatapp_secret@localhost:5432/chatdb"
    redis_url: str = "redis://localhost:6379/0"
    groq_api_key: str = ""
    rate_limit_requests: int = 10
    rate_limit_window_seconds: int = 60

    class Config:
        env_file = ".env"


settings = Settings()

Here is app/database.py:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.config import settings

engine = create_async_engine(settings.database_url, echo=False)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_db() -> AsyncSession:
    async with async_session() as session:
        yield session

Here is app/rate_limiter.py:

import redis.asyncio as aioredis
from app.config import settings

redis_client = aioredis.from_url(settings.redis_url, decode_responses=True)

Here is app/main.py:

from fastapi import FastAPI, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.rate_limiter import redis_client

app = FastAPI(title="AI Chat API")


@app.get("/health")
async def health(db: AsyncSession = Depends(get_db)):
    postgres_ok = False
    redis_ok = False

    try:
        await db.execute(text("SELECT 1"))
        postgres_ok = True
    except Exception:
        pass

    try:
        await redis_client.ping()
        redis_ok = True
    except Exception:
        pass

    return {"postgres": postgres_ok, "redis": redis_ok}
Rebuild and verify connectivity

Your Docker image was built before these files existed. You need to rebuild it so the container picks up your new code.

  • In your terminal, stop the running containers and rebuild by running these commands:
docker compose down
docker compose up --build

What do these commands do?

  • docker compose down stops and removes the running containers.
  • docker compose up --build rebuilds the app image with your new Python files and starts all three services.
  • Wait until you see log output confirming all services are healthy.
  • Open a new terminal tab and test the health endpoint by running:
curl http://localhost:8000/health
  • You should see this JSON response:
{"postgres":true,"redis":true}

Both values showing true confirms your FastAPI app can reach PostgreSQL and Redis inside the Docker network.

Seeing false for one or both connections?

  • Check that all three containers are running by running docker compose ps in another terminal. All services should show a "healthy" or "running" state.
  • Make sure the DATABASE_URL in docker-compose.yml uses postgres as the hostname (the service name), not localhost.
  • If you see an import error, confirm you have app/__init__.py (even if empty). Python needs it to treat app/ as a package.

Still stuck? Help me debug my /health endpoint returning false for postgres or redis

Your API is alive and connected. Next up, you'll define the database models for storing conversations and build endpoints to create and retrieve them.

Create PostgreSQL Models and Conversation Endpoints

Your health endpoint proves that FastAPI can talk to PostgreSQL. Now it's time to give the database something meaningful to store.

In this step, you'll define the data structure for conversations and messages using SQLAlchemy models. Then you'll create API endpoints that let users start new conversations and retrieve message history.

In this step, get ready to:
  • Create SQLAlchemy models for conversation and message data.
  • Build REST endpoints to create conversations and list messages.
  • Verify that conversation data persists across container restarts.
Create the database models

An ORM (Object-Relational Mapper) lets you define database tables as Python classes. Each class maps to a table, and each attribute maps to a column. SQLAlchemy's async mode works natively with FastAPI's event loop.

  • Create a new file called app/database.py in your app folder and add the following code:
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.config import settings

engine = create_async_engine(settings.database_url, echo=False)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_db() -> AsyncSession:
    async with async_session() as session:
        yield session

What does this code do?

This module creates the connection layer between your FastAPI app and PostgreSQL.

  • create_async_engine opens an async connection pool to PostgreSQL using the URL from your config.
  • async_sessionmaker creates a factory for database sessions. Each session is a short-lived conversation with the database.
  • get_db is a FastAPI dependency that hands each endpoint its own session, then closes it automatically.
  • Save app/database.py.

You won't see a visible change yet. This file provides the connection that the models will use next.

  • Create a new file called app/models.py in your app folder and add the Conversation model:
import uuid
from datetime import datetime, timezone
from sqlalchemy import String, Text, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class Conversation(Base):
    __tablename__ = "conversations"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    title: Mapped[str] = mapped_column(String(255), default="New Conversation")
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    messages: Mapped[list["Message"]] = relationship(back_populates="conversation", order_by="Message.created_at")

What does the Conversation model do?

  • Base is the parent class that tells SQLAlchemy "everything below me is a database table."
  • Each Conversation gets a UUID as its primary key, a title, and a timestamp.
  • The messages relationship links a conversation to all its messages, ordered by creation time.

You won't see a change yet. The next chunk adds the Message model that Conversation references.

  • Below the Conversation class in app/models.py, add the Message model:

class Message(Base):
    __tablename__ = "messages"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    conversation_id: Mapped[str] = mapped_column(ForeignKey("conversations.id"))
    role: Mapped[str] = mapped_column(String(20))
    content: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    conversation: Mapped["Conversation"] = relationship(back_populates="messages")

What does the Message model do?

  • conversation_id is a foreign key linking each message back to its parent conversation.
  • role tracks whether the message came from the user or the AI assistant.
  • content stores the actual text of the message with no length limit.
  • The conversation relationship is the reverse side, letting you navigate from a message back to its conversation.
  • Save app/models.py.

The models define the schema, but they won't create actual database tables until the app uses them at startup.

✔️ Awesome, I've got everything!

Great. Make sure you've saved app/models.py.

ⓧ I'd like to double check the full code

import uuid
from datetime import datetime, timezone
from sqlalchemy import String, Text, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class Conversation(Base):
    __tablename__ = "conversations"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    title: Mapped[str] = mapped_column(String(255), default="New Conversation")
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    messages: Mapped[list["Message"]] = relationship(back_populates="conversation", order_by="Message.created_at")


class Message(Base):
    __tablename__ = "messages"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    conversation_id: Mapped[str] = mapped_column(ForeignKey("conversations.id"))
    role: Mapped[str] = mapped_column(String(20))
    content: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    conversation: Mapped["Conversation"] = relationship(back_populates="messages")
Add conversation endpoints to the API

Now you'll update app/main.py to use these models. The app needs a startup event to create the database tables, plus two new endpoints for managing conversations.

  • In app/main.py, replace the import section at the top of the file and add a startup event. The top of your file (everything above the health endpoint) should look like this:
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel

from app.database import get_db, engine
from app.models import Base, Conversation, Message
from app.rate_limiter import redis_client

app = FastAPI(title="AI Chat API")


@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

What does the startup event do?

The @app.on_event("startup") decorator runs a function once when the app first boots. Inside it, Base.metadata.create_all tells SQLAlchemy to create the conversations and messages tables in PostgreSQL if they don't already exist.

The new imports bring in select for queries, BaseModel for request validation, and the engine from database.py for the startup connection.

You won't see a change yet. The next chunk adds the first endpoint.

  • After the health endpoint in app/main.py, add the ConversationCreate model and the POST endpoint:

class ConversationCreate(BaseModel):
    title: str = "New Conversation"


@app.post("/conversations")
async def create_conversation(body: ConversationCreate, db: AsyncSession = Depends(get_db)):
    conversation = Conversation(title=body.title)
    db.add(conversation)
    await db.commit()
    await db.refresh(conversation)
    return {"id": conversation.id, "title": conversation.title, "created_at": conversation.created_at.isoformat()}

What does this endpoint do?

  • ConversationCreate is a Pydantic model that validates incoming request bodies. It accepts an optional title field with a default value.
  • The POST endpoint creates a new Conversation row, commits it to PostgreSQL, and returns the new conversation's ID, title, and timestamp.
  • db.refresh(conversation) reloads the object from the database so you get the server-generated defaults (like the UUID and timestamp).

You won't see a change yet. The next chunk adds the messages endpoint.

  • Below the POST endpoint in app/main.py, add the GET messages endpoint:

@app.get("/conversations/{conversation_id}/messages")
async def get_messages(conversation_id: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(Message)
        .where(Message.conversation_id == conversation_id)
        .order_by(Message.created_at)
    )
    messages = result.scalars().all()
    return [
        {"id": m.id, "role": m.role, "content": m.content, "created_at": m.created_at.isoformat()}
        for m in messages
    ]

What does this endpoint do?

This endpoint fetches all messages belonging to a specific conversation, ordered by creation time. It uses SQLAlchemy's select to build a query that filters by conversation_id and returns a list of message dictionaries.

  • Save app/main.py.
  • Rebuild and restart your containers by running:
docker compose up --build -d
  • Wait a few seconds for all services to start, then test the new endpoint by running:
curl -X POST http://localhost:8000/conversations -H "Content-Type: application/json" -d '{"title": "My First Chat"}'

You should see a JSON response containing an id, title, and created_at field.

Not getting a response?

  • Run docker compose logs app to check for import errors or typos in your Python files.
  • Make sure you saved both app/models.py and app/main.py before rebuilding.
  • If you see a connection error, wait 10 seconds and try again. PostgreSQL may still be starting.

help me debug why my POST /conversations endpoint isn't working

✔️ Awesome, I've got everything!

Great. Make sure you've saved app/main.py and your containers are rebuilt.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel

from app.database import get_db, engine
from app.models import Base, Conversation, Message
from app.rate_limiter import redis_client

app = FastAPI(title="AI Chat API")


@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.get("/health")
async def health(db: AsyncSession = Depends(get_db)):
    postgres_ok = False
    redis_ok = False

    try:
        await db.execute(text("SELECT 1"))
        postgres_ok = True
    except Exception:
        pass

    try:
        await redis_client.ping()
        redis_ok = True
    except Exception:
        pass

    return {"postgres": postgres_ok, "redis": redis_ok}


class ConversationCreate(BaseModel):
    title: str = "New Conversation"


@app.post("/conversations")
async def create_conversation(body: ConversationCreate, db: AsyncSession = Depends(get_db)):
    conversation = Conversation(title=body.title)
    db.add(conversation)
    await db.commit()
    await db.refresh(conversation)
    return {"id": conversation.id, "title": conversation.title, "created_at": conversation.created_at.isoformat()}


@app.get("/conversations/{conversation_id}/messages")
async def get_messages(conversation_id: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(Message)
        .where(Message.conversation_id == conversation_id)
        .order_by(Message.created_at)
    )
    messages = result.scalars().all()
    return [
        {"id": m.id, "role": m.role, "content": m.content, "created_at": m.created_at.isoformat()}
        for m in messages
    ]
Test data persistence across restarts

Docker containers are ephemeral by default. However, your docker-compose.yml maps PostgreSQL's data to a named volume called postgres_data. This means your conversations survive container restarts.

  • Create a test conversation by running:
curl -X POST http://localhost:8000/conversations -H "Content-Type: application/json" -d '{"title": "Persistence Test"}'
  • Copy the id value from the response. You'll need it after the restart.
  • Stop and remove all containers by running:
docker compose down
  • Start everything back up by running:
docker compose up --build -d
  • Wait a few seconds for services to become healthy, then verify the conversation still exists by running:
curl http://localhost:8000/conversations/[[CONVERSATION_ID="the id value from the earlier response"]]/messages

You should see an empty list [] in the response. The empty list confirms the conversation exists in the database (otherwise you'd get an empty response for a non-existent ID). The list is empty because you haven't sent any messages to this conversation yet.

Why does the data survive?

The postgres_data volume in your docker-compose.yml stores PostgreSQL's files on your host machine. When you run docker compose down (without the -v flag), the volume stays intact. Only docker compose down -v deletes volumes and wipes your data.

Getting a connection error after restart?

  • PostgreSQL takes a few seconds to start. Wait 10 seconds and try the curl command again.
  • Run docker compose ps to check that all three services show "healthy" or "running" status.
  • If services show "restarting", check logs with docker compose logs app for errors.

help me figure out why my data didn't persist after restarting Docker containers

Your database layer is locked in. Conversations persist, and you have clean endpoints to create and query them. Next up, you'll wire in LangChain and Groq so the AI can actually respond to messages inside these conversations.

Integrate LangChain and the Chat Endpoint

Your API can store conversations and list messages. Now it needs the ability to actually think.

In this step, you will connect LangChain to the Groq API so the app can generate AI responses. Each time a user sends a message, the endpoint will load the full conversation history from PostgreSQL, pass it to the LLM, and store the AI's reply back in the database.

In this step, get ready to:
  • Create the LangChain chat module that calls Groq's Llama 3.1 8B model with full conversation context.
  • Build a POST endpoint that accepts user messages, generates AI replies, and saves both to the database.
  • Test a multi-turn conversation and confirm the AI remembers earlier messages.
Create the LangChain chat module

The chat module is the brain of your API. It initializes the LLM client, defines a system prompt for consistent behaviour, and exposes a function that other parts of the app can call.

  • Create a new file app/chat.py in the app/ folder.
  • Add the LLM setup code by pasting this into app/chat.py:
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models import Message

# Initialize the Groq LLM with Llama 3.1 8B
llm = ChatGroq(
    model="llama-3.1-8b-instant",
    temperature=0.7,
    max_tokens=1024,
    api_key=settings.groq_api_key,
)

# System prompt sets the AI's personality for every conversation
SYSTEM_PROMPT = "You are a helpful AI assistant. Be concise and informative in your responses."

What does this code do?

The ChatGroq class from langchain-groq creates a client that talks to Groq's inference API. The model parameter selects llama-3.1-8b-instant, which is fast, free-tier eligible, and good enough for conversational tasks.

The SYSTEM_PROMPT constant gets prepended to every conversation so the model maintains a consistent tone. You can customise this text later to change the AI's personality.

Now add the generate_response function below the SYSTEM_PROMPT line in app/chat.py:


async def generate_response(conversation_id: str, user_message: str, db: AsyncSession) -> str:
    # Load all previous messages for this conversation from the database
    result = await db.execute(
        select(Message)
        .where(Message.conversation_id == conversation_id)
        .order_by(Message.created_at)
    )
    history = result.scalars().all()

    # Build the message list LangChain expects: system prompt + history + new message
    messages = [SystemMessage(content=SYSTEM_PROMPT)]
    for msg in history:
        if msg.role == "user":
            messages.append(HumanMessage(content=msg.content))
        elif msg.role == "assistant":
            messages.append(AIMessage(content=msg.content))

    messages.append(HumanMessage(content=user_message))

    # Call the LLM asynchronously and return its text response
    response = await llm.ainvoke(messages)
    return response.content

What does this code do?

The generate_response function is called every time a user sends a message. It queries the database for all previous messages in the conversation, converts them into LangChain's HumanMessage and AIMessage types, and prepends the SystemMessage.

By sending the full history each time, the model understands context from earlier in the conversation. The ainvoke method calls the LLM asynchronously so it does not block other API requests while waiting for Groq's response.

  • Save app/chat.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/chat.py before moving on.

ⓧ I'd like to double check the full code

from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models import Message

llm = ChatGroq(
    model="llama-3.1-8b-instant",
    temperature=0.7,
    max_tokens=1024,
    api_key=settings.groq_api_key,
)

SYSTEM_PROMPT = "You are a helpful AI assistant. Be concise and informative in your responses."


async def generate_response(conversation_id: str, user_message: str, db: AsyncSession) -> str:
    result = await db.execute(
        select(Message)
        .where(Message.conversation_id == conversation_id)
        .order_by(Message.created_at)
    )
    history = result.scalars().all()

    messages = [SystemMessage(content=SYSTEM_PROMPT)]
    for msg in history:
        if msg.role == "user":
            messages.append(HumanMessage(content=msg.content))
        elif msg.role == "assistant":
            messages.append(AIMessage(content=msg.content))

    messages.append(HumanMessage(content=user_message))

    response = await llm.ainvoke(messages)
    return response.content
Wire up the chat endpoint in main.py

With the chat module ready, you need to connect it to a new API endpoint. The endpoint will accept a user's message, run it through LangChain, and persist both the user message and the AI reply in the database.

  • In app/main.py, find this import line near the top:
from app.rate_limiter import redis_client
  • Replace it with these two lines so the app imports the chat function and rate limiter utilities:
from app.chat import generate_response
from app.rate_limiter import check_rate_limit, get_rate_limit_status, redis_client

Now add the request model and the new endpoint. Scroll to the bottom of app/main.py and paste this after the get_messages function:


class MessageCreate(BaseModel):
    content: str
    user_id: str = "default_user"


@app.post("/conversations/{conversation_id}/messages")
async def send_message(conversation_id: str, body: MessageCreate, db: AsyncSession = Depends(get_db)):
    # Verify the conversation exists
    conversation = await db.get(Conversation, conversation_id)
    if not conversation:
        raise HTTPException(status_code=404, detail="Conversation not found")

    # Check rate limit before processing
    rate_limit = await check_rate_limit(body.user_id)
    if not rate_limit["allowed"]:
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded", "retry_after": rate_limit["retry_after"]},
            headers={"Retry-After": str(rate_limit["retry_after"])},
        )

    # Save the user's message to the database
    user_msg = Message(conversation_id=conversation_id, role="user", content=body.content)
    db.add(user_msg)
    await db.commit()

    # Generate AI response using LangChain with full conversation history
    ai_content = await generate_response(conversation_id, body.content, db)

    # Save the AI's response to the database
    ai_msg = Message(conversation_id=conversation_id, role="assistant", content=ai_content)
    db.add(ai_msg)
    await db.commit()
    await db.refresh(ai_msg)

    return {
        "id": ai_msg.id,
        "role": "assistant",
        "content": ai_content,
        "created_at": ai_msg.created_at.isoformat(),
        "rate_limit_remaining": rate_limit["remaining"],
    }

What does this endpoint do?

The POST /conversations/{conversation_id}/messages endpoint is the core of the chat API. It first checks whether the user has exceeded their rate limit via Redis.

If allowed, it saves the user's message, calls generate_response to get an AI reply (which loads full history from the DB), saves the AI response, and returns it with the remaining rate limit count.

  • Save app/main.py.

Seeing an import error?

If you see ImportError when the container restarts, double check that from app.chat import generate_response appears on its own line, directly above the from app.rate_limiter import.

Also confirm that app/chat.py is saved in the app/ directory (not the project root).

help me fix an ImportError in my FastAPI Docker container

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py before continuing.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel

from app.database import get_db, engine
from app.models import Base, Conversation, Message
from app.chat import generate_response
from app.rate_limiter import check_rate_limit, get_rate_limit_status, redis_client

app = FastAPI(title="AI Chat API")


@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.get("/health")
async def health(db: AsyncSession = Depends(get_db)):
    postgres_ok = False
    redis_ok = False

    try:
        await db.execute(text("SELECT 1"))
        postgres_ok = True
    except Exception:
        pass

    try:
        await redis_client.ping()
        redis_ok = True
    except Exception:
        pass

    return {"postgres": postgres_ok, "redis": redis_ok}


class ConversationCreate(BaseModel):
    title: str = "New Conversation"


@app.post("/conversations")
async def create_conversation(body: ConversationCreate, db: AsyncSession = Depends(get_db)):
    conversation = Conversation(title=body.title)
    db.add(conversation)
    await db.commit()
    await db.refresh(conversation)
    return {"id": conversation.id, "title": conversation.title, "created_at": conversation.created_at.isoformat()}


@app.get("/conversations/{conversation_id}/messages")
async def get_messages(conversation_id: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(Message)
        .where(Message.conversation_id == conversation_id)
        .order_by(Message.created_at)
    )
    messages = result.scalars().all()
    return [
        {"id": m.id, "role": m.role, "content": m.content, "created_at": m.created_at.isoformat()}
        for m in messages
    ]


class MessageCreate(BaseModel):
    content: str
    user_id: str = "default_user"


@app.post("/conversations/{conversation_id}/messages")
async def send_message(conversation_id: str, body: MessageCreate, db: AsyncSession = Depends(get_db)):
    conversation = await db.get(Conversation, conversation_id)
    if not conversation:
        raise HTTPException(status_code=404, detail="Conversation not found")

    rate_limit = await check_rate_limit(body.user_id)
    if not rate_limit["allowed"]:
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded", "retry_after": rate_limit["retry_after"]},
            headers={"Retry-After": str(rate_limit["retry_after"])},
        )

    user_msg = Message(conversation_id=conversation_id, role="user", content=body.content)
    db.add(user_msg)
    await db.commit()

    ai_content = await generate_response(conversation_id, body.content, db)

    ai_msg = Message(conversation_id=conversation_id, role="assistant", content=ai_content)
    db.add(ai_msg)
    await db.commit()
    await db.refresh(ai_msg)

    return {
        "id": ai_msg.id,
        "role": "assistant",
        "content": ai_content,
        "created_at": ai_msg.created_at.isoformat(),
        "rate_limit_remaining": rate_limit["remaining"],
    }
Rebuild and test a multi-turn conversation

Since the Docker container copies your app files at build time, you need to rebuild the image so it picks up app/chat.py and the updated app/main.py.

  • Rebuild and restart the containers by running:
docker compose up --build -d
  • Wait for all 3 containers to report healthy. Confirm with:
docker compose ps

You should see the app, postgres, and redis services all running.

  • Create a new conversation by running:
curl -s -X POST http://localhost:8000/conversations -H "Content-Type: application/json" -d '{"title": "Memory Test"}'

What should I see?

You will get a JSON response containing an id field. Copy that UUID. You will use it in the next commands.

  • Send a first message to the conversation. Replace the conversation ID with the UUID you just copied:
curl -s -X POST http://localhost:8000/conversations/[[CONV_ID="your-conversation-id"]]/messages -H "Content-Type: application/json" -d '{"content": "My name is Alex and I live in Melbourne", "user_id": "test_user"}'
  • Send a second message that tests whether the AI remembers the first:
curl -s -X POST http://localhost:8000/conversations/[[CONV_ID="your-conversation-id"]]/messages -H "Content-Type: application/json" -d '{"content": "What is my name and where do I live?", "user_id": "test_user"}'
  • Send a third message to further confirm multi-turn memory:
curl -s -X POST http://localhost:8000/conversations/[[CONV_ID="your-conversation-id"]]/messages -H "Content-Type: application/json" -d '{"content": "Suggest a restaurant near where I live", "user_id": "test_user"}'

The AI should reference "Alex" and "Melbourne" in its responses, proving it remembers the full conversation history.

AI not remembering previous messages?

Make sure you are using the same conversation ID in all three curl commands. Each conversation has its own isolated history.

If you get a 404, the conversation ID might have been lost when you restarted containers. Create a fresh conversation and try again.

help me debug why my AI chat endpoint is not remembering conversation history

Your API now has a working brain. It can hold a conversation, remember what was said, and generate relevant responses. Next up, you will add rate limiting to protect the endpoint from abuse.

Add Redis Rate Limiting and Push to GitHub

Your AI chat endpoint is working, but there is nothing stopping a single user from flooding it with hundreds of requests per minute. Without a rate limiter, one abusive client could exhaust your Groq API quota and block everyone else.

In this step, you will build a per-user rate limiter powered by Redis sorted sets. Each request timestamp gets stored in a sorted set, old entries get pruned automatically, and the count of remaining entries decides whether to allow or reject. After verifying the limiter works, you will push the finished project to GitHub.

In this step, get ready to:
  • Create a rate limiter module that tracks requests per user with a 60-second sliding window.
  • Verify the API returns HTTP 429 after 10 rapid requests.
  • Commit and push the finished project to GitHub.
Create the rate limiter module

The sliding window pattern stores each request's timestamp as a score in a Redis sorted set. On every new request, entries older than the window are pruned and the remaining count is compared against the limit.

  • Create a new file app/rate_limiter.py inside your app folder.
  • Add the following code for the Redis client connection and the check_rate_limit function:
import time
import redis.asyncio as aioredis
from app.config import settings

# Connect to Redis using the URL from settings
redis_client = aioredis.from_url(settings.redis_url, decode_responses=True)


async def check_rate_limit(user_id: str) -> dict:
    # Build the Redis key for this user's rate limit set
    key = f"rate_limit:{user_id}"
    now = time.time()
    window_start = now - settings.rate_limit_window_seconds

    # Remove expired timestamps and count remaining entries
    pipe = redis_client.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)
    pipe.zcard(key)
    pipe.execute()
    results = await pipe.execute()

    request_count = results[1]

    # If over the limit, deny the request
    if request_count >= settings.rate_limit_requests:
        ttl = await redis_client.ttl(key)
        return {"allowed": False, "remaining": 0, "retry_after": ttl if ttl > 0 else settings.rate_limit_window_seconds}

    # Otherwise, record this request and allow it
    pipe = redis_client.pipeline()
    pipe.zadd(key, {str(now): now})
    pipe.expire(key, settings.rate_limit_window_seconds)
    await pipe.execute()

    return {"allowed": True, "remaining": settings.rate_limit_requests - request_count - 1, "retry_after": 0}

What does this code do?

The redis_client creates an async connection to your Redis container using the URL from app/config.py.

  • zremrangebyscore prunes all entries with timestamps older than the sliding window boundary.
  • zcard counts how many request timestamps remain in the current window.
  • If the count is at or above the limit (10), the function returns allowed: False with a retry_after value.
  • If the count is below the limit, zadd records the new request timestamp and the function returns allowed: True with the number of remaining requests.

Why a sliding window instead of a fixed counter?

A fixed window (one counter that resets every 60 seconds) allows a burst of 20 requests if a user sends 10 at the end of one window and 10 at the start of the next. A sliding window tracks exact timestamps so the limit is always enforced over any continuous 60-second period.

  • Now add the get_rate_limit_status function below check_rate_limit in the same file:
async def get_rate_limit_status(user_id: str) -> dict:
    # Query how many requests are in the current window
    key = f"rate_limit:{user_id}"
    now = time.time()
    window_start = now - settings.rate_limit_window_seconds

    await redis_client.zremrangebyscore(key, 0, window_start)
    request_count = await redis_client.zcard(key)

    return {
        "user_id": user_id,
        "requests_made": request_count,
        "requests_remaining": max(0, settings.rate_limit_requests - request_count),
        "window_seconds": settings.rate_limit_window_seconds,
    }

What does this code do?

This is a read-only companion to check_rate_limit. It prunes expired entries and returns a summary showing how many requests have been made and how many remain. The /rate-limit-status endpoint in app/main.py calls this function so users can check their own quota.

  • Save app/rate_limiter.py.

Seeing an import error when the container restarts?

Make sure the file is saved as app/rate_limiter.py (with an underscore, not a hyphen). Python cannot import modules with hyphens in their names.

Also confirm app/__init__.py exists (even if empty). Without it, Python will not recognise app as a package.

help me debug a module import error in my Docker container

✔️ Awesome, I've got everything!

Great. Double check you have saved app/rate_limiter.py with both functions.

ⓧ I'd like to double check the full code

import time
import redis.asyncio as aioredis
from app.config import settings

redis_client = aioredis.from_url(settings.redis_url, decode_responses=True)


async def check_rate_limit(user_id: str) -> dict:
    key = f"rate_limit:{user_id}"
    now = time.time()
    window_start = now - settings.rate_limit_window_seconds

    pipe = redis_client.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)
    pipe.zcard(key)
    pipe.execute()
    results = await pipe.execute()

    request_count = results[1]

    if request_count >= settings.rate_limit_requests:
        ttl = await redis_client.ttl(key)
        return {"allowed": False, "remaining": 0, "retry_after": ttl if ttl > 0 else settings.rate_limit_window_seconds}

    pipe = redis_client.pipeline()
    pipe.zadd(key, {str(now): now})
    pipe.expire(key, settings.rate_limit_window_seconds)
    await pipe.execute()

    return {"allowed": True, "remaining": settings.rate_limit_requests - request_count - 1, "retry_after": 0}


async def get_rate_limit_status(user_id: str) -> dict:
    key = f"rate_limit:{user_id}"
    now = time.time()
    window_start = now - settings.rate_limit_window_seconds

    await redis_client.zremrangebyscore(key, 0, window_start)
    request_count = await redis_client.zcard(key)

    return {
        "user_id": user_id,
        "requests_made": request_count,
        "requests_remaining": max(0, settings.rate_limit_requests - request_count),
        "window_seconds": settings.rate_limit_window_seconds,
    }
Rebuild and test the rate limiter

Your app/main.py already imports check_rate_limit and get_rate_limit_status from app.rate_limiter. Now that the module exists, the app container can load it.

  • Rebuild the app container to pick up the new file by running:
docker compose up --build -d
  • Wait a few seconds for all containers to become healthy, then verify with:
docker compose ps

You should see all three services (postgres, redis, app) listed as healthy or running.

  • First, create a conversation to test against by running:
curl -s -X POST http://localhost:8000/conversations -H "Content-Type: application/json" -d '{"title": "Rate limit test"}'

Copy the id value from the JSON response. You will use it in the next command.

  • Send 11 rapid requests to the chat endpoint using a loop. Replace the conversation ID placeholder with your actual ID:
for i in $(seq 1 11); do
  curl -s -X POST http://localhost:8000/conversations/[[CONV_ID="your conversation id"]]/messages \
    -H "Content-Type: application/json" \
    -d "{\"content\": \"Hello $i\", \"user_id\": \"test_user\"}" | head -c 120
  echo
done

The first 10 requests should return normal AI responses. The 11th request should return a JSON body containing "detail": "Rate limit exceeded" with HTTP status 429.

  • Confirm the rate limit status endpoint works by running:
curl -s http://localhost:8000/rate-limit-status?user_id=test_user

You should see requests_remaining at 0 and requests_made at 10.

Not seeing 429 on the 11th request?

If all 11 requests succeed, check whether your loop ran slowly enough for the window to partially expire. Try sending requests closer together or reduce rate_limit_window_seconds temporarily in app/config.py to test faster.

If you see a 500 error instead, check the container logs with docker compose logs app for import or connection errors.

help me debug why my rate limiter is not returning 429

Commit and push to GitHub

The project is feature-complete. Time to commit the final code and push it to a GitHub repository.

  • Stage and commit all changes by running:
git add -A
git commit -m "Add Redis rate limiting and status endpoint"
  • Create a new repository on GitHub. Go to github.com/new in your browser.
  • Name the repository ai-chat-api (or any name you prefer).
  • Leave the repository empty (do not add a README or .gitignore).
  • Click Create repository.
  • Back in your terminal, add the remote and push by running:
git remote add origin [[YOUR_REPO_URL="https://github.com/your-username/ai-chat-api.git"]]
git branch -M main
git push -u origin main
  • Refresh your GitHub repository page in the browser.
  • You should see all your project files listed including docker-compose.yml, Dockerfile, requirements.txt, and the app/ directory.

Push rejected or authentication error?

If you see a permission denied error, GitHub requires a personal access token (PAT) instead of a password. Generate one at github.com/settings/tokens and use it as your password when prompted.

help me push my code to GitHub

Your AI chat API is fully operational with rate limiting, conversation memory, and version control. Next up, tackle the Secret Mission and add real-time token streaming to your chat endpoint.

Secret mission

Streaming Chat with Server-Sent Events

Your chat endpoint works, but users have to wait for the entire AI response before seeing anything. Real chat applications stream tokens as they arrive, just like ChatGPT. In this mission, you'll add a streaming endpoint that delivers AI tokens in real-time using Server-Sent Events.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local machine using Docker containers. There are no ongoing cloud costs. The containers consume system resources (CPU, memory) while running, but you will not be charged anything.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker containers (PostgreSQL, Redis, FastAPI app) running locally via Docker Compose.
  • Docker volume postgres_data storing your PostgreSQL database.
  • Project folder on your Desktop containing all source files.
  • GitHub repository with your pushed project code.

Keep everything running

No action needed. Choose this if you are still actively building or testing your API.

  • Your Docker containers will continue running and consuming system resources (around 500MB of RAM total).
  • You can send requests to http://localhost:8000 at any time while the containers are running.
  • Your conversation history in PostgreSQL and rate limit data in Redis will remain intact.

Pause - I'll come back to this later

Stop the running containers to free up memory without deleting any data. You can restart everything later with a single command.

  • Open your terminal and navigate to your project folder.
  • Stop all containers by running:
docker compose stop
  • This stops the containers without removing them or deleting your data volumes.
  • When you want to resume, run:
docker compose start
  • Your conversation history and all project files will be exactly as you left them.

Delete - I don't want to use this again

Remove all project resources and start fresh. This permanently deletes your conversation history and all container data.

  • Open your terminal and navigate to your project folder.
  • Stop and remove all containers, networks, and data volumes by running:
docker compose down -v
  • Delete the project folder from your Desktop by running:
cd ~/Desktop
rm -rf ai-chat-api

Windows users

If you are on Windows, use this command instead to remove the project folder:

cd ~\Desktop
rmdir /s /q ai-chat-api
  • To delete your GitHub repository, open the repository page on GitHub.
  • Click Settings in the top navigation bar.
  • Scroll down to the Danger Zone section and click Delete this repository.
  • Follow the confirmation prompts and type your repository name to confirm deletion.

Nice Work!

Nice Work!

Nice work! You just built a fully containerized AI chat backend from scratch, wiring together FastAPI, PostgreSQL, Redis, LangChain, and Docker Compose into a production-style service.

You've learned how to:

  • Orchestrate a multi-container application with Docker Compose, coordinating FastAPI, PostgreSQL, and Redis with health checks and dependency ordering.
  • Build an async REST API with LangChain and Groq that maintains conversation memory by storing chat history in PostgreSQL via SQLAlchemy.
  • Implement per-user rate limiting using Redis sorted sets with a sliding window pattern, returning HTTP 429 responses when limits are exceeded.
  • Secret Mission: Added a streaming chat endpoint that delivers AI response tokens in real time as Server-Sent Events.

Ready to quiz yourself?