Build an AI Document Q&A API

Build a containerized RAG API with FastAPI, pgvector, and Redis caching.

Introduction

30 Second Summary

When you ask a question about a long document, you don't read the whole thing from start to finish. You jump to the most relevant section first. That's exactly how modern AI systems answer questions about large collections of text.

In this project, you will build a containerized AI-powered Q&A API that lets users upload documents and ask questions about them. The system finds the most relevant passages, sends them to an AI model for reasoning, and caches answers for instant repeat lookups.

What You'll Build

A Docker Compose application with FastAPI, PostgreSQL (pgvector), and Redis that ingests documents, stores vector embeddings, and answers questions using Groq's Llama model with semantic retrieval and response caching.

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

  • A FastAPI REST API with document upload and AI-powered Q&A endpoints.
  • A Docker Compose configuration orchestrating three services (app, PostgreSQL+pgvector, Redis).
  • A RAG pipeline using LangChain that chunks text, generates embeddings, retrieves relevant context, and generates answers via Groq.
  • A Redis caching layer so identical questions return instantly without hitting the LLM.
  • Secret Mission: Add an analytics endpoint that tracks cache hit rate, total queries, and average response time.

Are there any prerequisites?

You'll need basic Python knowledge and Docker Desktop installed on Windows. You'll also need a free Groq account with an API key.

Groq's free tier includes access to all models with no credit card required. This project costs nothing to run.

Set Up the Project and Docker Environment

A RAG API involves multiple moving parts: a web server, a vector database, and a cache layer. Docker Compose lets you define all of these services in a single file and spin them up together with one command.

In this step, you will create the project folder, pin every Python dependency, and write the Docker configuration files that orchestrate your three-service stack.

In this step, get ready to:
  • Create the project directory and a pinned dependency file.
  • Write the Dockerfile for the FastAPI application.
  • Configure Docker Compose to orchestrate the app, PostgreSQL, and Redis services.
Create the project directory and dependencies file

Before writing any application code, you need a folder to hold the project and a file that lists every Python package the container will install.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir ai-doc-qa-api
cd ai-doc-qa-api
  • Create the app subdirectory by running:
mkdir app
  • Confirm the folder structure exists by running:
ls

You should see the app folder listed in the output.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Select the ai-doc-qa-api folder on your Desktop, then click Open.

You should see the app folder in the Explorer sidebar on the left.

  • Create an empty file called __init__.py inside the app folder. Right-click the app folder in the Explorer sidebar, select New File, and type __init__.py.
  • Leave __init__.py empty and save it. This file tells Python that app is a package.
  • Create a new file in the root of your ai-doc-qa-api folder called requirements.txt. Right-click the empty space below app in the Explorer sidebar, select New File, and type requirements.txt.
  • Paste the following dependency list into requirements.txt:
fastapi==0.136.3
uvicorn==0.34.0
langchain==1.3.8
langchain-groq==1.1.3
langchain-core==1.4.6
langchain-community==0.3.24
sentence-transformers==4.1.0
psycopg2-binary==2.9.10
sqlalchemy==2.0.40
pgvector==0.4.0
redis==5.2.1
python-multipart==0.0.20
pydantic==2.11.3

What do these packages do?

  • fastapi and uvicorn provide the web server and REST API framework.
  • langchain, langchain-groq, langchain-core, and langchain-community form the LangChain framework that orchestrates the RAG pipeline.
  • sentence-transformers generates vector embeddings from text using the all-MiniLM-L6-v2 model locally.
  • psycopg2-binary, sqlalchemy, and pgvector connect to PostgreSQL and enable vector similarity search.
  • redis is the Python client for the Redis cache layer.
  • Save requirements.txt with Cmd+S (macOS) or Ctrl+S (Windows).

File not appearing in the sidebar?

Make sure you created requirements.txt at the root of ai-doc-qa-api, not inside the app subfolder. If it landed in the wrong place, drag it to the correct level in the Explorer sidebar.

Help me figure out why my file isn't showing up.

Write the Dockerfile for the FastAPI application

A Dockerfile describes how to build your application into a container image. It declares the base image, installs dependencies, copies your code, and sets the startup command.

  • Create a new file in the root of your ai-doc-qa-api folder called Dockerfile (no extension). Right-click the empty space in the Explorer sidebar, select New File, and type Dockerfile.
  • Paste the following into Dockerfile:
FROM python:3.10-slim

WORKDIR /app

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 this Dockerfile do?

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

✔️ Awesome, I've got everything!

Great. Make sure Dockerfile is saved at the root of ai-doc-qa-api, right next to requirements.txt.

ⓧ I'd like to double check the full code

FROM python:3.10-slim

WORKDIR /app

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"]
Write the docker-compose.yml to orchestrate all services

Your API depends on two external services: PostgreSQL (with the pgvector extension for storing embeddings) and Redis (for caching responses). Docker Compose defines all three services in one file so a single command starts the entire stack.

First, you need a .env file so Docker Compose can pass your Groq API key into the container without hardcoding it.

  • Create a new file in the root of ai-doc-qa-api called .env. Right-click in the Explorer sidebar, select New File, and type .env.
  • Paste the following into .env and replace your-groq-api-key-here with your actual Groq API key:
GROQ_API_KEY=your-groq-api-key-here

Where do I find my Groq API key?

Log into console.groq.com, navigate to API Keys, and create a new key. Copy the value and paste it into your .env file.

  • Save .env.
  • Create another new file in the root of ai-doc-qa-api called docker-compose.yml.
  • Paste the following into docker-compose.yml:
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/ragdb
      - REDIS_URL=redis://redis:6379
      - GROQ_API_KEY=${GROQ_API_KEY}
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
    volumes:
      - ./app:/app/app

  db:
    image: pgvector/pgvector:0.8.2-pg17
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=ragdb
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      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

volumes:
  pgdata:

What does this configuration do?

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

  • app builds your FastAPI image from the Dockerfile. It exposes port 8000 and mounts your local app/ folder so code changes appear without rebuilding.
  • db runs PostgreSQL 17 with the pgvector extension pre-installed using the pgvector/pgvector:0.8.2-pg17 image.
  • redis runs a lightweight Redis instance using redis:7-alpine.

The healthcheck blocks ensure PostgreSQL and Redis are fully ready before the app container starts. The depends_on condition service_healthy enforces this startup order.

  • Save docker-compose.yml.

✔️ Awesome, I've got everything!

Great. Your project folder should now contain: .env, Dockerfile, docker-compose.yml, requirements.txt, and the app/ folder with __init__.py inside it.

ⓧ I'd like to double check the full code

Here is each file as it should look right now:

fastapi==0.136.3
uvicorn==0.34.0
langchain==1.3.8
langchain-groq==1.1.3
langchain-core==1.4.6
langchain-community==0.3.24
sentence-transformers==4.1.0
psycopg2-binary==2.9.10
sqlalchemy==2.0.40
pgvector==0.4.0
redis==5.2.1
python-multipart==0.0.20
pydantic==2.11.3
FROM python:3.10-slim

WORKDIR /app

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"]
GROQ_API_KEY=your-groq-api-key-here
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/ragdb
      - REDIS_URL=redis://redis:6379
      - GROQ_API_KEY=${GROQ_API_KEY}
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
    volumes:
      - ./app:/app/app

  db:
    image: pgvector/pgvector:0.8.2-pg17
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=ragdb
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      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

volumes:
  pgdata:

Now verify that Docker can build your image from these files.

  • Switch back to your terminal. Make sure you are inside the ai-doc-qa-api folder.
  • Build the Docker image by running this command:
docker compose build

This downloads the base image, installs all Python packages, and produces your application image. It may take a few minutes the first time.

When the build finishes, you should see output ending with a line similar to => exporting to image and no red error text.

Build failing with an error?

  • Make sure Docker Desktop is running. Open Docker Desktop if it is not already active.
  • Check that Dockerfile has no file extension. Some editors add .txt automatically. Rename it if needed.
  • Confirm requirements.txt is in the same folder as Dockerfile. The COPY requirements.txt . instruction fails if the file is missing from the build context.

Help me debug my Docker build failure.

Your Docker environment is ready. Next up, you will write the FastAPI application code and connect it to both databases.

Build the FastAPI Application with Database Connection

Your Docker environment is ready to run containers. Now you need to build the actual application that lives inside those containers.

In this step, you will create the FastAPI application that connects to PostgreSQL and Redis, sets up the vector storage table, and exposes the API endpoints your RAG pipeline will use.

In this step, get ready to:
  • Create the FastAPI app with database and Redis connections.
  • Define the pgvector table schema for storing document embeddings.
  • Bring up the full stack and verify the health endpoint.
Create the FastAPI app with health check endpoint and database connection setup

Your API needs to connect to two services on startup: PostgreSQL for storing document embeddings, and Redis for caching responses. SQLAlchemy handles the database connection, while the redis Python client handles the cache connection.

  • Create a new file called main.py inside the app folder of your project.
  • Add the following imports and connection setup code to app/main.py:
import time
import json
import os

from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, text
import redis

from app.ingestion import ingest_document
from app.rag import ask_question

# Create the FastAPI application instance
app = FastAPI(title="AI Document Q&A API")

# Read connection strings from environment variables set in docker-compose.yml
DATABASE_URL = os.getenv("DATABASE_URL")
REDIS_URL = os.getenv("REDIS_URL")

# Initialize the database engine and Redis client
engine = create_engine(DATABASE_URL)
redis_client = redis.from_url(REDIS_URL, decode_responses=True)

What does this code do?

This sets up the two external connections your API depends on.

  • os.getenv("DATABASE_URL") reads the PostgreSQL connection string injected by Docker Compose.
  • create_engine(DATABASE_URL) creates a SQLAlchemy engine that manages a pool of database connections.
  • redis.from_url(REDIS_URL, decode_responses=True) creates a Redis client that automatically decodes byte responses to strings.
  • The two imports at the top (ingest_document and ask_question) reference modules you will build in later steps.

The app imports ingest_document and ask_question from modules that do not exist yet. You need to create placeholder files so the app can start without import errors.

  • Create a placeholder app/ingestion.py file with this temporary stub:
# Placeholder - full implementation coming in Step 3
def ingest_document(content, filename, engine):
    return 0
  • Create a placeholder app/rag.py file with this temporary stub:
# Placeholder - full implementation coming in Step 4
def ask_question(question, engine):
    return {"answer": "Not implemented yet", "source_chunks": []}

Why create placeholder files?

Python will raise an ImportError if the app tries to import from modules that do not exist. These stubs let the app start cleanly while you build the real implementations in Steps 3 and 4.

Create the pgvector table schema for storing document chunks and their embeddings

When the app starts, it needs to ensure the database has the pgvector extension enabled and a table ready to store document chunks alongside their vector embeddings.

  • Add the startup event, Pydantic models, and all three endpoints below the redis_client line in app/main.py:

@app.on_event("startup")
def startup():
    with engine.connect() as conn:
        conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS documents (
                id SERIAL PRIMARY KEY,
                content TEXT NOT NULL,
                embedding vector(384),
                metadata JSONB DEFAULT '{}'
            )
        """))
        conn.commit()


class QuestionRequest(BaseModel):
    question: str


class AnswerResponse(BaseModel):
    answer: str
    source_chunks: list[str]
    cached: bool


@app.get("/health")
def health():
    try:
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        redis_client.ping()
        return {"status": "healthy", "database": "connected", "redis": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=str(e))


@app.post("/documents/upload")
async def upload_document(file: UploadFile = File(...)):
    if not file.filename.endswith(".txt"):
        raise HTTPException(status_code=400, detail="Only .txt files are supported")

    content = await file.read()
    text_content = content.decode("utf-8")

    num_chunks = ingest_document(text_content, file.filename, engine)

    return {"message": f"Document '{file.filename}' ingested successfully", "chunks": num_chunks}


@app.post("/ask", response_model=AnswerResponse)
def ask(request: QuestionRequest):
    start_time = time.time()

    cache_key = f"qa:{request.question.lower().strip()}"
    cached_response = redis_client.get(cache_key)

    if cached_response:
        redis_client.incr("stats:cache_hits")
        redis_client.incr("stats:total_queries")
        elapsed = (time.time() - start_time) * 1000
        redis_client.incrbyfloat("stats:total_time_ms", elapsed)
        data = json.loads(cached_response)
        data["cached"] = True
        return AnswerResponse(**data)

    redis_client.incr("stats:cache_misses")
    redis_client.incr("stats:total_queries")

    result = ask_question(request.question, engine)

    elapsed = (time.time() - start_time) * 1000
    redis_client.incrbyfloat("stats:total_time_ms", elapsed)

    response_data = {
        "answer": result["answer"],
        "source_chunks": result["source_chunks"],
        "cached": False,
    }

    redis_client.setex(cache_key, 3600, json.dumps({
        "answer": result["answer"],
        "source_chunks": result["source_chunks"],
    }))

    return AnswerResponse(**response_data)

What does this code do?

This chunk handles four responsibilities:

  • The startup() function runs once when the app boots. It enables the pgvector extension and creates a documents table with a 384-dimension vector column matching the embedding model you will use later.
  • The /health endpoint tests both database and Redis connectivity. If either fails, it returns a 503 error.
  • The /documents/upload endpoint accepts .txt files and passes them to the ingestion pipeline (currently a stub).
  • The /ask endpoint checks Redis for a cached answer before calling the RAG pipeline. It also tracks timing and query statistics in Redis for analytics.

What is vector(384)?

The vector(384) column type comes from the pgvector extension. It stores a 384-dimensional floating-point array. This dimension matches the all-MiniLM-L6-v2 embedding model you will use in Step 3 to convert text into numerical vectors.

  • Save app/main.py, app/ingestion.py, and app/rag.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files before moving on.

ⓧ I'd like to double check the full code

Here is the complete app/main.py file:

import time
import json
import os

from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, text
import redis

from app.ingestion import ingest_document
from app.rag import ask_question

app = FastAPI(title="AI Document Q&A API")

DATABASE_URL = os.getenv("DATABASE_URL")
REDIS_URL = os.getenv("REDIS_URL")

engine = create_engine(DATABASE_URL)
redis_client = redis.from_url(REDIS_URL, decode_responses=True)


@app.on_event("startup")
def startup():
    with engine.connect() as conn:
        conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS documents (
                id SERIAL PRIMARY KEY,
                content TEXT NOT NULL,
                embedding vector(384),
                metadata JSONB DEFAULT '{}'
            )
        """))
        conn.commit()


class QuestionRequest(BaseModel):
    question: str


class AnswerResponse(BaseModel):
    answer: str
    source_chunks: list[str]
    cached: bool


@app.get("/health")
def health():
    try:
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        redis_client.ping()
        return {"status": "healthy", "database": "connected", "redis": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=str(e))


@app.post("/documents/upload")
async def upload_document(file: UploadFile = File(...)):
    if not file.filename.endswith(".txt"):
        raise HTTPException(status_code=400, detail="Only .txt files are supported")

    content = await file.read()
    text_content = content.decode("utf-8")

    num_chunks = ingest_document(text_content, file.filename, engine)

    return {"message": f"Document '{file.filename}' ingested successfully", "chunks": num_chunks}


@app.post("/ask", response_model=AnswerResponse)
def ask(request: QuestionRequest):
    start_time = time.time()

    cache_key = f"qa:{request.question.lower().strip()}"
    cached_response = redis_client.get(cache_key)

    if cached_response:
        redis_client.incr("stats:cache_hits")
        redis_client.incr("stats:total_queries")
        elapsed = (time.time() - start_time) * 1000
        redis_client.incrbyfloat("stats:total_time_ms", elapsed)
        data = json.loads(cached_response)
        data["cached"] = True
        return AnswerResponse(**data)

    redis_client.incr("stats:cache_misses")
    redis_client.incr("stats:total_queries")

    result = ask_question(request.question, engine)

    elapsed = (time.time() - start_time) * 1000
    redis_client.incrbyfloat("stats:total_time_ms", elapsed)

    response_data = {
        "answer": result["answer"],
        "source_chunks": result["source_chunks"],
        "cached": False,
    }

    redis_client.setex(cache_key, 3600, json.dumps({
        "answer": result["answer"],
        "source_chunks": result["source_chunks"],
    }))

    return AnswerResponse(**response_data)

Seeing a DeprecationWarning about on_event?

You may see a warning in your logs mentioning that on_event is deprecated. This is safe to ignore. The @app.on_event("startup") decorator still works correctly. The newer alternative (lifespan context managers) adds complexity that is not needed for this project.

help me understand the on_event deprecation warning

Bring up the stack and verify the health endpoint

With all files in place, you can start the full Docker Compose stack and confirm your API connects to both PostgreSQL and Redis.

  • Start all three services by running this command in your terminal from the project root:
docker compose up --build

What does this command do?

The --build flag rebuilds the app image to include your new main.py file. Docker Compose then starts PostgreSQL, Redis, and your FastAPI app in the correct order (thanks to the healthcheck dependencies you configured in Step 1).

Wait until you see a log line containing Uvicorn running on http://0.0.0.0:8000 in your terminal output. This means the app is ready.

  • Open a new terminal window and test the health endpoint by running:
curl http://localhost:8000/health

You should see a JSON response confirming both connections are healthy:

{"status": "healthy", "database": "connected", "redis": "connected"}

Getting a connection refused error?

  • Make sure all three containers are running. Check the Docker Compose terminal for error messages.
  • If you see an ImportError, confirm you saved both app/ingestion.py and app/rag.py with the placeholder functions.
  • If the database container is still starting, wait a few seconds and try again. The healthcheck retries 5 times at 5-second intervals.

help me debug why my Docker Compose stack is not starting

Your API skeleton is running with live database and cache connections. Next up, you will build the document ingestion pipeline that turns uploaded text files into searchable vector embeddings.

Build the Document Ingestion Pipeline

Your API is running and the health endpoint confirms connectivity to both PostgreSQL and Redis. But the /documents/upload endpoint currently calls a function that doesn't exist yet.

In this step, you'll build the document ingestion pipeline. This pipeline takes raw text, splits it into smaller chunks, converts each chunk into a vector embedding, and stores everything in pgvector so your API can search it later.

In this step, get ready to:
  • Review the existing upload endpoint and understand what the ingestion function needs.
  • Create the ingestion module with text chunking and embedding generation.
  • Test the full pipeline by uploading a sample document.
Review the document upload endpoint

The /documents/upload endpoint already exists in app/main.py. It reads the uploaded file, decodes the text, and calls ingest_document to handle all the processing.

  • In app/main.py, find the upload endpoint to see how it calls the ingestion function:
num_chunks = ingest_document(text_content, file.filename, engine)

What does the upload endpoint expect?

The ingest_document function receives three arguments: the raw text content of the file, the original filename (for metadata), and the SQLAlchemy database engine.

It must return an integer representing how many chunks were stored. Your job now is to create the module that provides this function.

Create the ingestion module

The ingestion pipeline has three jobs: split text into chunks, generate vector embeddings for each chunk, and store both in the documents table.

  • Create a new file called app/ingestion.py inside your app folder and paste the following code:
from langchain_community.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
from sqlalchemy import text

# Load the embedding model once at module level (384-dimension vectors)
model = SentenceTransformer("all-MiniLM-L6-v2")


def ingest_document(content: str, filename: str, engine) -> int:
    # Configure the text splitter with chunk size and overlap
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        length_function=len,
    )

    # Split the raw text into smaller, searchable chunks
    chunks = splitter.split_text(content)
    # Generate a vector embedding for each chunk
    embeddings = model.encode(chunks)

    # Store each chunk and its embedding in PostgreSQL
    with engine.connect() as conn:
        for chunk, embedding in zip(chunks, embeddings):
            conn.execute(
                text("""
                    INSERT INTO documents (content, embedding, metadata)
                    VALUES (:content, :embedding, :metadata)
                """),
                {
                    "content": chunk,
                    "embedding": embedding.tolist().__str__().replace("'", ""),
                    "metadata": f'{{"filename": "{filename}"}}',
                },
            )
        conn.commit()

    return len(chunks)

What does this code do?

RecursiveCharacterTextSplitter from LangChain breaks long documents into 500-character chunks with 50 characters of overlap. The overlap ensures that sentences split across chunk boundaries still appear in at least one chunk completely.

SentenceTransformer loads the all-MiniLM-L6-v2 model, which converts text into 384-dimensional vectors. These vectors capture the meaning of each chunk so the database can find semantically similar content later.

The function then inserts each chunk alongside its embedding into the documents table using pgvector's vector(384) column type.

  • Save app/ingestion.py.

Why load the model at module level?

Loading a machine learning model takes a few seconds. By placing it at the top of the module (outside the function), Python loads it once when the app starts. Every subsequent call to ingest_document reuses the already-loaded model instead of reloading it each time.

Seeing an import error in the container logs?

If your container restarts after saving the file, check the Docker Compose logs for the exact error. The most common cause is a typo in the import path. Confirm your first line reads from langchain_community.text_splitter import RecursiveCharacterTextSplitter exactly.

help me debug an import error in my Docker container for the ingestion module

✔️ Awesome, I've got everything!

Great. Double check you've saved app/ingestion.py and the container logs show no import errors.

ⓧ I'd like to double check the full code

from langchain_community.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
from sqlalchemy import text

model = SentenceTransformer("all-MiniLM-L6-v2")


def ingest_document(content: str, filename: str, engine) -> int:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        length_function=len,
    )

    chunks = splitter.split_text(content)
    embeddings = model.encode(chunks)

    with engine.connect() as conn:
        for chunk, embedding in zip(chunks, embeddings):
            conn.execute(
                text("""
                    INSERT INTO documents (content, embedding, metadata)
                    VALUES (:content, :embedding, :metadata)
                """),
                {
                    "content": chunk,
                    "embedding": embedding.tolist().__str__().replace("'", ""),
                    "metadata": f'{{"filename": "{filename}"}}',
                },
            )
        conn.commit()

    return len(chunks)
Test the document upload pipeline

Now that the ingestion module exists, you can upload a real document and verify the full pipeline works end-to-end.

  • Create a sample text file called sample.txt in your project root (the same folder as docker-compose.yml) with some test content:
echo "Docker is a platform for building and running containerized applications. Containers package code and dependencies together so applications run consistently across environments. Docker Compose orchestrates multi-container applications using a single YAML configuration file." > sample.txt
  • Upload the sample file to your API by running this command:
curl -X POST http://localhost:8000/documents/upload -F "file=@sample.txt"
  • You should see a JSON response confirming the upload succeeded. It will include the filename and the number of chunks created.

What should I see?

The response looks like this:

{"message":"Document 'sample.txt' ingested successfully","chunks":1}

The number of chunks depends on how long your text is. Since the sample text is under 500 characters, it produces 1 chunk. Longer documents produce more chunks.

Getting a connection error or 500 response?

If curl can't connect, make sure Docker Compose is still running. Check with docker compose ps to see if all three services are healthy.

If you get a 500 error, check the app container logs with docker compose logs app for a traceback. The most common cause is a missing app/ingestion.py file or a typo in the function signature.

help me debug a 500 error when uploading a document to my FastAPI Docker container

Your ingestion pipeline is working. Documents go in as raw text and come out as searchable vector embeddings stored in PostgreSQL. Next up, you'll build the Q&A endpoint that searches these embeddings and generates answers using Groq.

Build the RAG Q&A Endpoint with Redis Caching

Your ingestion pipeline can store documents as vector embeddings in PostgreSQL. Now you need the retrieval side.

When a user asks a question, the system converts it into an embedding, finds the most similar document chunks in the database, sends those chunks as context to an LLM, and returns a generated answer. Redis caching ensures repeated questions return instantly without hitting the LLM again.

In this step, get ready to:
  • Create the RAG module with vector search and LLM answer generation.
  • Walk through the /ask endpoint's Redis caching layer.
  • Test the full Q&A flow and verify cache behaviour.
Create the vector search and LLM setup module

Your app/main.py already imports ask_question from app.rag. That module does not exist yet, so the app container is currently failing to start. You will create it now.

  • Create a new file called rag.py inside the app folder.
  • Add the imports, embedding model, LLM client, and prompt template by pasting this code:
import os

from langchain_groq import ChatGroq
from langchain_core.prompts import ChatPromptTemplate
from sentence_transformers import SentenceTransformer
from sqlalchemy import text

# Load the same embedding model used during ingestion
model = SentenceTransformer("all-MiniLM-L6-v2")

# Initialize the Groq LLM client
llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    temperature=0,
    api_key=os.getenv("GROQ_API_KEY"),
)

# Define the prompt that sends retrieved context + the user's question
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant that answers questions based on the provided context. If the context does not contain enough information to answer, say so clearly."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])

What does this code do?

This module sets up the core retrieval-augmented generation components.

  • SentenceTransformer("all-MiniLM-L6-v2") loads the same 384-dimension embedding model you used during ingestion. Matching models is critical because the question embedding must live in the same vector space as the stored document embeddings.
  • ChatGroq connects to Groq's free API to run llama-3.3-70b-versatile in the cloud. No local GPU needed.
  • ChatPromptTemplate defines the system instruction and human message format. The {context} placeholder will be filled with retrieved document chunks at query time.
Add the question-answering function

The ask_question function ties everything together. It embeds the user's question, queries pgvector for the closest document chunks, and feeds those chunks to the LLM.

  • Add this function below the prompt_template definition in app/rag.py:
def ask_question(question: str, engine) -> dict:
    # Convert the question into a vector embedding
    question_embedding = model.encode([question])[0]
    embedding_str = question_embedding.tolist().__str__().replace("'", "")

    # Query pgvector for the 3 closest document chunks
    with engine.connect() as conn:
        result = conn.execute(
            text("""
                SELECT content, embedding <-> :embedding AS distance
                FROM documents
                ORDER BY embedding <-> :embedding
                LIMIT 3
            """),
            {"embedding": embedding_str},
        )
        rows = result.fetchall()

    # Handle the case where no documents exist
    if not rows:
        return {"answer": "No documents have been uploaded yet.", "source_chunks": []}

    # Join the top chunks into a single context string
    context = "\n\n".join([row[0] for row in rows])
    source_chunks = [row[0] for row in rows]

    # Run the LLM chain with the retrieved context
    chain = prompt_template | llm
    response = chain.invoke({"context": context, "question": question})

    return {"answer": response.content, "source_chunks": source_chunks}

What does this code do?

This function implements the full RAG retrieval-and-generation loop.

  • model.encode([question])[0] converts the question text into a 384-dimension vector.
  • The SQL query uses pgvector's <-> operator to compute L2 (Euclidean) distance between the question embedding and every stored embedding. The 3 rows with the smallest distance are the most relevant chunks.
  • prompt_template | llm creates a LangChain chain. The pipe operator connects the prompt formatter to the LLM, so invoke() formats the prompt then sends it to Groq in one call.
  • Save app/rag.py.

Seeing an import error in the container logs?

  • Check that you named the file exactly rag.py inside the app folder. The import in main.py is from app.rag import ask_question so the path must match.
  • Confirm you pasted both code blocks (module setup and the ask_question function) into the same file.

help me fix import errors in my Docker container

✔️ Awesome, I've got everything!

Great. Double check you have saved app/rag.py and the container is no longer showing import errors.

ⓧ I'd like to double check the full code

import os

from langchain_groq import ChatGroq
from langchain_core.prompts import ChatPromptTemplate
from sentence_transformers import SentenceTransformer
from sqlalchemy import text

model = SentenceTransformer("all-MiniLM-L6-v2")

llm = ChatGroq(
    model="llama-3.3-70b-versatile",
    temperature=0,
    api_key=os.getenv("GROQ_API_KEY"),
)

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant that answers questions based on the provided context. If the context does not contain enough information to answer, say so clearly."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])


def ask_question(question: str, engine) -> dict:
    question_embedding = model.encode([question])[0]
    embedding_str = question_embedding.tolist().__str__().replace("'", "")

    with engine.connect() as conn:
        result = conn.execute(
            text("""
                SELECT content, embedding <-> :embedding AS distance
                FROM documents
                ORDER BY embedding <-> :embedding
                LIMIT 3
            """),
            {"embedding": embedding_str},
        )
        rows = result.fetchall()

    if not rows:
        return {"answer": "No documents have been uploaded yet.", "source_chunks": []}

    context = "\n\n".join([row[0] for row in rows])
    source_chunks = [row[0] for row in rows]

    chain = prompt_template | llm
    response = chain.invoke({"context": context, "question": question})

    return {"answer": response.content, "source_chunks": source_chunks}
Walk through the Redis caching layer and test the Q&A flow

The /ask endpoint in app/main.py already wraps ask_question with a Redis caching layer. Now that rag.py exists, the full flow is active.

How does the Redis caching work?

Open app/main.py and look at the ask function. Here is the caching logic:

  • Before calling the LLM, the endpoint builds a cache key from the normalised question: f"qa:{request.question.lower().strip()}".
  • If that key exists in Redis, the cached JSON is returned immediately with cached: True. No embedding, no database query, no LLM call.
  • If the key does not exist, the full RAG pipeline runs. The response is stored in Redis with a 3600-second (1 hour) TTL via redis_client.setex.
  • Every request increments Redis counters (stats:total_queries, stats:cache_hits or stats:cache_misses) for analytics.

Why simple key-value caching instead of semantic caching?

Semantic caching would embed each new question and compare it against cached question embeddings. That adds complexity and another embedding call per request.

Simple string-match caching is fast, predictable, and teaches the core caching pattern clearly. If you need fuzzy matching later, you can layer semantic caching on top.

Now test the full flow. First, make sure the containers are running with your new code.

  • Restart the stack to pick up the new rag.py module by running:
docker compose up --build -d
  • Wait about 30 seconds for the sentence-transformers model to download on first boot.
  • Verify the app is healthy by running:
curl http://localhost:8000/health
  • You should see a JSON response with "status": "healthy", "database": "connected", and "redis": "connected".
  • Upload a sample text file so the system has documents to search. Create a file called sample.txt with any content (a few paragraphs of text), then upload it by running:
curl -X POST http://localhost:8000/documents/upload -F "file=@sample.txt"
  • You should see a JSON response confirming the document was ingested with a chunk count.
  • Ask a question about your uploaded document by running:
curl -X POST http://localhost:8000/ask -H "Content-Type: application/json" -d "{\"question\": \"What is this document about?\"}"
  • You should see a JSON response with answer, source_chunks, and "cached": false.
  • Run the exact same curl command again.
  • This time the response should return much faster and show "cached": true. The answer came directly from Redis without touching the LLM.

Getting a connection error or 503?

  • Run docker compose logs app to check for Python import errors or missing environment variables.
  • Confirm your .env file contains your Groq API key. The GROQ_API_KEY variable must be set for the LLM call to work.
  • If you see a timeout on the first /ask call, the embedding model may still be downloading. Wait 30 seconds and retry.

help me debug my /ask endpoint returning an error

Your RAG API is fully operational. Documents go in, questions come back with LLM-generated answers, and Redis ensures repeat questions are lightning-fast. Next up, you will tackle the Secret Mission and add analytics tracking.

Secret mission

Add an Analytics Dashboard Endpoint

Your API is answering questions and caching responses, but how do you know if the cache is actually helping? Build a /stats endpoint that tracks total queries, cache hit rate, and average response time so you can measure your API's real-world performance.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally via Docker, so there are no ongoing cloud costs.

Resources you used:

  • Docker containers (app, db, redis) running locally.
  • Docker volume (pgdata) storing PostgreSQL data.
  • Project folder on your Desktop.

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with the API.

  • Your Docker containers continue running and the API remains accessible at http://localhost:8000.
  • Uploaded documents and cached responses persist in the PostgreSQL volume and Redis.
  • Your .env file still contains your Groq API key. Keep it safe and avoid committing it to version control.

Pause - I'll come back to this later

Stop the containers to free up memory without removing your data. You can restart everything later with a single command.

  • Stop all containers by running this command from your project folder:
  • When you're ready to resume, run docker compose up to bring everything back.
docker compose stop

Your PostgreSQL volume and uploaded documents are preserved. Redis data also remains until the container is removed.

Delete - I don't want to use this again

Remove all containers, volumes, and project files to start completely fresh.

  • Stop and remove all containers and volumes by running this command from your project folder:
docker compose down -v

What does this command do?

The -v flag removes the named volumes (including pgdata) along with the containers. Without it, your PostgreSQL data would persist even after the containers are gone.

  • Delete the project folder from your Desktop. On macOS or Linux:
cd ~/Desktop
rm -rf ai-document-qa-api
  • On Windows:
cd ~\Desktop
rmdir /s /q ai-document-qa-api
  • Your Groq account remains active on the free tier. No action is needed unless you want to revoke your API key from the Groq console.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-style AI-powered document Q&A API, fully containerized and running across three services with Docker Compose.

You've learned how to:

  • Build a FastAPI REST API with document upload and AI-powered Q&A endpoints.
  • Orchestrate 3 Docker services (app, PostgreSQL, Redis) with a single docker-compose.yml file.
  • Implement a full RAG pipeline from end to end: text chunking, vector embedding, similarity search in pgvector, and LLM-powered answer generation via Groq.
  • Secret Mission: Built an analytics endpoint that tracks cache hit rate, total queries, and average response time.

Ready to quiz yourself?