Capsule 04: Retrieval, Generation, and API

Capsule description

You're going to take the end-to-end RAG pipeline you built in M4/11 (a minimal ask() function in ~150 lines) and expose it as a professional REST API with FastAPI: documented endpoints, input validation, error handling, structured citations, and a trace_id for debugging.

Key prerequisite (M4/11): you already have the core RAG logic — query → retrieve top-k → build an anti-hallucination prompt → generate with GPT → answer with sources. This capsule doesn't reinvent that logic; it encapsulates it behind an API that third parties (frontend, mobile, other services) can consume.

What's new in this capsule is the service layer: stable API contracts, validation with Pydantic, observability (latency per endpoint, error tracking, request tracing), and the details that separate a Python script from a service a team can operate.

---

API endpoints

Method	Path	Description
POST	/ingest	Triggers ingestion of a directory (async or sync by design)
GET	/search	Similarity search (query, top_k, optional filters)
POST	/ask	RAG question: retrieval + generation, answer with sources
GET	/collections	Lists collections in ChromaDB
DELETE	/collections/{name}	Deletes a collection
GET	/health	Health check for Docker/K8s

---

/ask endpoint contract

{
  "answer": "Text generated based on the retrieved context.",
  "sources": [
    {
      "doc_id": "doc_001",
      "source": "/path/to/doc.txt",
      "title": "Sample Document",
      "score": 0.87
    }
  ],
  "confidence": 0.82,
  "trace_id": "550e8400-e29b-41d4-a716-446655440000",
  "fallback_reason": null
}

If there is not enough evidence: fallback_reason = "insufficient retrieval", confidence = 0, sources = [].

---

Project structure

project/
├── api/
│   ├── __init__.py
│   ├── main.py          # FastAPI app
│   ├── routes/
│   │   ├── ask.py
│   │   ├── search.py
│   │   └── ingest.py
│   └── services/
│       ├── retrieval.py
│       └── generation.py
├── ingestion/           # (from capsule 03)
├── config.py
└── run_ingestion.py

---

Complete code

1. Configuration

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

CHROMA_PATH = os.getenv("CHROMA_PATH", "./chroma_data")
COLLECTION_NAME = os.getenv("COLLECTION_NAME", "rag_docs")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")
LLM_MODEL = os.getenv("LLM_MODEL", "gpt-3.5-turbo")
TOP_K = int(os.getenv("TOP_K", "5"))
SCORE_THRESHOLD = float(os.getenv("SCORE_THRESHOLD", "0.5"))
MAX_QUESTION_LENGTH = int(os.getenv("MAX_QUESTION_LENGTH", "2000"))

---

2. Retrieval service

# api/services/retrieval.py
import chromadb
from chromadb.config import Settings
from openai import OpenAI
from typing import List, Optional
import os

client_openai = OpenAI()
chroma_client = chromadb.PersistentClient(path=os.getenv("CHROMA_PATH", "./chroma_data"))


def embed_text(text: str, model: str = "text-embedding-3-small") -> List[float]:
    """Generate an embedding for a single text (query)."""
    response = client_openai.embeddings.create(model=model, input=[text])
    return response.data[0].embedding


def retrieve(
    query: str,
    collection_name: str = "rag_docs",
    top_k: int = 5,
    score_threshold: float = 0.5,
    where: Optional[dict] = None,
) -> dict:
    """
    Retrieval: embed the query, similarity search, filter by score.
    Returns: documents, metadatas, ids, distances (lower = more similar).
    """
    collection = chroma_client.get_collection(collection_name)
    query_embedding = embed_text(query)

    kwargs = {
        "query_embeddings": [query_embedding],
        "n_results": top_k,
        "include": ["documents", "metadatas", "distances"],
    }
    if where:
        kwargs["where"] = where

    result = collection.query(**kwargs)

    docs = result["documents"][0] if result["documents"] else []
    metas = result["metadatas"][0] if result["metadatas"] else []
    ids = result["ids"][0] if result["ids"] else []
    distances = result["distances"][0] if result.get("distances") else []

    # ChromaDB uses L2 by default; convert to similarity score (1 / (1 + distance))
    # Or if you use cosine, distances may be cosine distance (1 - similarity)
    scores = [1 / (1 + d) if d is not None else 0 for d in distances]

    # Filter by threshold
    filtered = [
        (doc, meta, id_, score)
        for doc, meta, id_, score in zip(docs, metas, ids, scores)
        if score >= score_threshold
    ]
    if not filtered:
        return {"documents": [], "metadatas": [], "ids": [], "scores": []}

    docs_f, metas_f, ids_f, scores_f = zip(*filtered)
    return {
        "documents": list(docs_f),
        "metadatas": list(metas_f),
        "ids": list(ids_f),
        "scores": list(scores_f),
    }

---

3. Generation service

# api/services/generation.py
from openai import OpenAI
from typing import List, Optional

client = OpenAI()


def build_context(documents: List[str], metadatas: List[dict]) -> str:
    """Assemble the context for the prompt from the retrieved chunks."""
    parts = []
    for i, (doc, meta) in enumerate(zip(documents, metadatas), 1):
        source = meta.get("source", "unknown")
        title = meta.get("title", meta.get("doc_id", "Doc"))
        parts.append(f"[Source {i}: {title}]\n{doc}")
    return "\n\n---\n\n".join(parts)


def generate_answer(
    question: str,
    documents: List[str],
    metadatas: List[dict],
    model: str = "gpt-3.5-turbo",
    temperature: float = 0.2,
) -> str:
    """
    Generate an answer using the retrieved context.
    If there are no documents, return a fallback message without calling the LLM.
    """
    if not documents or not metadatas:
        return "I don't have enough information in my knowledge base to answer this question."

    context = build_context(documents, metadatas)
    prompt = f"""You are an assistant that answers questions based ONLY on the provided context.
Do not make up information. If the context does not contain the answer, say you don't have that information.

Context:
{context}

Question: {question}

Answer clearly and concisely. Cite the sources when relevant."""

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "You answer based only on the given context. Do not make up data."},
            {"role": "user", "content": prompt},
        ],
        temperature=temperature,
    )
    return response.choices[0].message.content


def build_sources_response(metadatas: List[dict], scores: List[float]) -> List[dict]:
    """Format the sources for the response contract."""
    return [
        {
            "doc_id": m.get("doc_id", ""),
            "source": m.get("source", ""),
            "title": m.get("title", m.get("doc_id", "Doc")),
            "score": round(s, 2),
        }
        for m, s in zip(metadatas, scores)
    ]

---

4. FastAPI routes

# api/routes/ask.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from uuid import uuid4
import os

from api.services.retrieval import retrieve
from api.services.generation import generate_answer, build_sources_response

router = APIRouter(prefix="/ask", tags=["RAG"])
MAX_QUESTION_LENGTH = int(os.getenv("MAX_QUESTION_LENGTH", "2000"))


class AskRequest(BaseModel):
    question: str = Field(..., min_length=1, max_length=MAX_QUESTION_LENGTH)


class SourceItem(BaseModel):
    doc_id: str
    source: str
    title: str
    score: float


class AskResponse(BaseModel):
    answer: str
    sources: list[SourceItem]
    confidence: float
    trace_id: str
    fallback_reason: str | None = None


@router.post("", response_model=AskResponse)
def ask(req: AskRequest):
    trace_id = str(uuid4())
    question = req.question.strip()

    # Retrieval
    result = retrieve(
        query=question,
        top_k=int(os.getenv("TOP_K", "5")),
        score_threshold=float(os.getenv("SCORE_THRESHOLD", "0.5")),
    )
    docs = result["documents"]
    metas = result["metadatas"]
    scores = result["scores"]

    # Fallback if there is not enough evidence
    if not docs or not metas:
        return AskResponse(
            answer="I didn't find relevant information to answer your question.",
            sources=[],
            confidence=0.0,
            trace_id=trace_id,
            fallback_reason="insufficient retrieval",
        )

    # Compute confidence (normalized average of scores)
    confidence = round(sum(scores) / len(scores), 2) if scores else 0.0

    # Generation
    answer = generate_answer(
        question=question,
        documents=docs,
        metadatas=metas,
        model=os.getenv("LLM_MODEL", "gpt-3.5-turbo"),
    )
    sources = build_sources_response(metas, scores)

    return AskResponse(
        answer=answer,
        sources=[SourceItem(**s) for s in sources],
        confidence=confidence,
        trace_id=trace_id,
        fallback_reason=None,
    )

---

# api/routes/search.py
from fastapi import APIRouter, Query
from pydantic import BaseModel
import os

from api.services.retrieval import retrieve

router = APIRouter(prefix="/search", tags=["Search"])


class SearchResponse(BaseModel):
    documents: list[str]
    metadatas: list[dict]
    ids: list[str]
    scores: list[float]


@router.get("", response_model=SearchResponse)
def search(
    q: str = Query(..., min_length=1),
    top_k: int = Query(5, ge=1, le=20),
    score_threshold: float = Query(0.0, ge=0, le=1),
):
    result = retrieve(
        query=q,
        top_k=top_k,
        score_threshold=score_threshold,
    )
    return SearchResponse(
        documents=result["documents"],
        metadatas=result["metadatas"],
        ids=result["ids"],
        scores=result["scores"],
    )

---

# api/routes/ingest.py
from fastapi import APIRouter, HTTPException, BackgroundTasks
from pydantic import BaseModel

router = APIRouter(prefix="/ingest", tags=["Ingestion"])


class IngestRequest(BaseModel):
    dir_path: str


@router.post("")
def trigger_ingest(req: IngestRequest, background_tasks: BackgroundTasks):
    """Trigger asynchronous ingestion of a directory."""
    from pathlib import Path
    if not Path(req.dir_path).exists():
        raise HTTPException(status_code=400, detail="Directory not found")
    from ingestion.pipeline import ingest_directory
    from ingestion.config import IngestionConfig
    config = IngestionConfig()
    def run():
        ingest_directory(req.dir_path, config=config)
    background_tasks.add_task(run)
    return {"status": "ingestion started", "dir_path": req.dir_path}

---

# api/routes/collections.py
from fastapi import APIRouter
import chromadb
import os

router = APIRouter(prefix="/collections", tags=["Collections"])
chroma_client = chromadb.PersistentClient(path=os.getenv("CHROMA_PATH", "./chroma_data"))


@router.get("")
def list_collections():
    cols = chroma_client.list_collections()
    return [{"name": c.name, "count": c.count()} for c in cols]


@router.delete("/{name}")
def delete_collection(name: str):
    try:
        chroma_client.delete_collection(name)
        return {"status": "deleted", "collection": name}
    except Exception as e:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail=str(e))

---

# api/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from api.routes.ask import router as ask_router
from api.routes.search import router as search_router
from api.routes.ingest import router as ingest_router
from api.routes.collections import router as collections_router

app = FastAPI(
    title="RAG API",
    description="RAG system with ChromaDB and OpenAI",
    version="1.0.0",
    docs_url="/docs",   # Swagger
    redoc_url="/redoc", # ReDoc
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(ask_router)
app.include_router(search_router)
app.include_router(collections_router)


@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/")
def root():
    return {
        "message": "RAG API - Use /docs for Swagger, /redoc for ReDoc",
        "endpoints": ["/ask", "/search", "/ingest", "/collections", "/health"],
    }

---

5. Running the API

# Install dependencies
pip install fastapi uvicorn chromadb openai python-dotenv pydantic

# Run
uvicorn api.main:app --reload --host 0.0.0.0 --port 8000

Access:

• Swagger: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

---

Answer quality rules

Rule	Implementation
Don't make things up without evidence	Explicit fallback when retrieval is empty or scores are low
Include sources	sources with doc_id, source, title, score
Avoid answers without context	Don't call the LLM if there are no chunks; return a standard message
Traceability	trace_id in every response for correlation with logs
Input validation	max_length on the question to avoid abuse

---

Usage example

POST /ask

curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"question": "What is a vector database?"}'

Expected response:

{
  "answer": "A vector database is a storage system optimized for...",
  "sources": [
    {
      "doc_id": "intro_vectors",
      "source": "/docs/intro.txt",
      "title": "intro",
      "score": 0.87
    }
  ],
  "confidence": 0.85,
  "trace_id": "550e8400-e29b-41d4-a716-446655440000",
  "fallback_reason": null
}

---

GET /search

curl "http://localhost:8000/search?q=vector%20database&top_k=3"

---

Practical exercises

Exercise 1: Extend /ask with metadata filtering

Add an optional doc_id or source parameter to the request to filter the search to only certain documents.

Solution: In AskRequest add doc_id: str | None = None. In the retrieve call, if doc_id is present, use where={"doc_id": doc_id}.

---

Exercise 2: "I have no evidence" answer when confidence < 0.5

If the average of the scores is less than 0.5, don't call the LLM. Return the fallback message and fallback_reason="low_confidence".

Solution: Before generate_answer, compute confidence. If confidence < 0.5, return AskResponse(..., fallback_reason="low_confidence") without calling the LLM.

---

Exercise 3: Include trace_id in logs

Configure a middleware or dependency that captures the trace_id and adds it to every log of the request. Use structlog or logging with context.

Solution: A middleware that generates a trace_id at the start of the request and stores it in request.state.trace_id. In each endpoint, you pass that value to the response. For logs, use logging.LoggerAdapter with extra={"trace_id": trace_id}.

---

Exercise 4: Timeout and retry for OpenAI

Add a 30s timeout to the OpenAI calls and retry (1 attempt) if it fails on timeout.

Solution: client.chat.completions.create(..., timeout=30). Wrap in try/except; on Timeout or APIConnectionError, retry once with time.sleep(2).

---

Exercise 5: POST /ingest endpoint

Create an endpoint that receives {"dir_path": "/path/to/docs"} and runs the ingestion pipeline from capsule 03. To avoid blocking, consider FastAPI's BackgroundTasks.

Solution:

from fastapi import BackgroundTasks
from ingestion.pipeline import ingest_directory

@router.post("/ingest")
def trigger_ingest(payload: dict, bg: BackgroundTasks):
    dir_path = payload.get("dir_path")
    if not dir_path:
        raise HTTPException(400, "dir_path required")
    def run():
        ingest_directory(dir_path)
    bg.add_task(run)
    return {"status": "ingestion started", "dir": dir_path}

---

Exercise 6: Error documentation

Define a standard error schema: {"error": str, "trace_id": str, "detail": str}. Use a global exception handler to return it on 4xx/5xx.

Solution:

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
def global_exception_handler(request: Request, exc: Exception):
    trace_id = getattr(request.state, "trace_id", "unknown")
    return JSONResponse(
        status_code=500,
        content={"error": "internal_error", "trace_id": trace_id, "detail": str(exc)},
    )

---

Troubleshooting

"Correct answers but no traceability"

Cause: Sources and trace_id are not included.

Solution: Make sure AskResponse always includes sources (even if empty) and trace_id. Generate the trace_id at the start of the request and pass it through the whole flow.

---

"Made-up answers when there is no context"

Cause: The LLM is called even with empty retrieval or very low scores.

Solution: Add validation: if len(docs) == 0 or max(scores) < 0.5, don't call generate_answer. Return an explicit fallback with fallback_reason.

---

"Unstable API under load"

Cause: No limits, timeouts, or concurrency control.

Solution: Limit question length (max_length=2000). Add a timeout to OpenAI. Consider rate limiting with slowapi or similar. Review ChromaDB connection pooling.

---

"Swagger/ReDoc don't show correct schemas"

Cause: Pydantic models without response_model or with undocumented complex types.

Solution: Use response_model=AskResponse in the decorator. Define all models with BaseModel and typed fields. FastAPI generates the schemas automatically.

---

"500 errors without useful detail"

Cause: Uncaught exceptions, insufficient logs.

Solution: A global exception handler that logs the traceback and returns a generic message to the client (don't reveal internals). In development, debug=True can help; in production, structured logs with a trace_id.

---

Summary

• The API exposes /ask, /search, /collections, /health with FastAPI.
• The /ask endpoint integrates retrieval (ChromaDB + OpenAI embeddings) and generation (GPT-3.5-turbo).
• Response contract: answer, sources, confidence, trace_id, fallback_reason.
• Explicit fallback when there is not enough evidence; answers are not made up.
• Swagger (/docs) and ReDoc (/redoc) document the API automatically.
• Separating retrieval and generation into independent functions makes testing and evolution easier.
• Input validation, timeout, and error handling improve robustness.

---

Additional resources

• FastAPI Tutorial
• FastAPI Background Tasks
• ChromaDB Query
• OpenAI Chat Completions
• Pydantic Models
• Uvicorn
• RAG Prompt Engineering
• Structured Logging (Python)

---

Estimated time: 35-40 minutes
Next: 05-testing-and-evaluation.md