← Cleiton FragaBlog
← Voltar para o blog
·10 min de leitura

The Whole Machine: A FastAPI RAG Service, End to End

Part 5 of a 5-part series building a production-shaped RAG API on FastAPI + MongoDB Atlas

The Whole Machine: A FastAPI RAG Service, End to End

Across four posts we built up the pieces in isolation: the OpenAI-compatible LLM client, chunking and vector storage, hybrid search and reranking, and the Atlas indexes underneath. This final post wires them into the actual application — a FastAPI service with two RAG endpoints, JWT auth, and a clean request lifecycle — and then does the thing tutorials usually skip: catalogs exactly what you'd fix before trusting it with real traffic.

The repo is small on purpose. That's its value: every architectural decision in a production RAG system is visible here without the noise of a framework. Read it once and the shape of the whole category clicks.

The map

app/
├── main.py                 # FastAPI app; mounts routers under /api/v1
├── core/security.py        # JWT (HS256) + bcrypt
├── deps/auth.py            # dependency: extract user id from Bearer token
├── db/mongo.py             # Motor async client + get_db()
├── llm/openai_client.py    # embed_texts() + chat_completion()   [Part 1]
├── rag/
│   ├── chunk.py            # simple_chunk()                       [Part 2]
│   ├── store.py            # upsert + vector + hybrid search       [Parts 2–3]
│   └── rerank.py           # cross-encoder rerank + fallback      [Part 3]
├── schemas/                # Pydantic request/response models
└── api/v1/
    ├── auth.py             # login / logout
    ├── users.py            # user CRUD
    └── rag.py              # /rag/documents + /rag/ask            ← the orchestration

main.py is four lines of substance — create the app, include three routers under /api/v1. The layering is the conventional FastAPI split: routers handle HTTP, dependencies handle cross-cutting concerns (auth, the DB handle), schemas validate the wire format, and the rag/ and llm/ modules hold the domain logic. Routes stay thin; logic stays testable.

Auth, briefly, because RAG without ownership is a data leak

Every RAG route is gated by a JWT bearer token, and that isn't ceremony — it's the mechanism that keeps one user's documents out of another user's answers.

core/security.py does the standard two jobs: hash passwords with bcrypt (so the database never holds a plaintext password) and mint/verify HS256 JWTs. Login verifies the password and returns a signed token; deps/auth.py exposes get_current_user_id, a dependency that pulls the sub (subject = user id) out of the token on every protected request. Drop current_user_id = Depends(get_current_user_id) into a route and you've got the authenticated user's id, or a 401 if the token is missing or invalid.

That current_user_id is the thread that runs through the entire RAG flow. Watch where it goes.

Endpoint 1 — POST /rag/documents: ingestion

This is the write path: a PDF goes in, chunks-with-embeddings come out, owned by the uploader. The route is the orchestration layer over machinery from Parts 1–2.

# app/api/v1/rag.py  (abridged)
@router.post("/rag/documents", response_model=UploadDocResponse)
async def upload_document(
    db=Depends(get_db),
    current_user_id: str = Depends(get_current_user_id),   # ← auth → owner
    file: UploadFile = File(...),
    title: str = Form(...),
):
    # 1) validate it's a PDF
    if file.content_type not in {"application/pdf", "application/x-pdf",
                                 "binary/octet-stream"} \
            and not file.filename.lower().endswith(".pdf"):
        raise HTTPException(400, "Send a valid PDF")

    data = await file.read()
    if not data:
        raise HTTPException(400, "Empty PDF")

    # 2) extract text (capped at 15 pages)
    text = _extract_text_from_pdf_bytes(data)
    if not text or not text.strip():
        raise HTTPException(400, "Could not extract text from the PDF")

    # 3) chunk + embed + store, tagged with the owner
    res = await upsert_document(db, title, text, owner_id=current_user_id)
    return UploadDocResponse(**res)

The pipeline: validate → extract → chunk → embed → store. Text extraction uses pypdf and is deliberately capped at 15 pages — a sane guardrail for a demo that also caps embedding cost and latency. Then upsert_document (Part 2) does the heavy lifting: it writes the parent documents row stamped with owner: current_user_id, splits the text with simple_chunk, embeds every chunk in one batched call, and inserts the chunks. The response is just { doc_id, chunks } — the document's id and how many chunks it produced.

The ownership stamp is the important part. Because each document records its owner, the read path can later prove that this user is allowed to query this document.

Endpoint 2 — POST /rag/ask: the full RAG lifecycle

This is the whole series in one function. Every part shows up, in order:

# app/api/v1/rag.py  (abridged)
@router.post("/rag/ask", response_model=AnswerResponse)
async def ask(
    body: AskRequest,
    db=Depends(get_db),
    current_user_id: str = Depends(get_current_user_id),
):
    # 0) ownership: does this user own this document?
    oid = await get_owned_document_oid(db, body.doc_id, current_user_id)
    if not oid:
        raise HTTPException(404, "Document not found or not owned by user")

    # 1) embed the question                                   [Part 1]
    q_emb = embed_texts([body.question])[0]

    # 2) hybrid search → wide candidate set                   [Parts 3–4]
    candidates = await hybrid_search_chunks(
        db, body.doc_id, body.question, q_emb, k=body.candidates
    )
    if not candidates:
        raise HTTPException(404, "No context found")

    # 3) rerank → top-k                                        [Part 3]
    results = rerank(body.question, candidates, top_n=body.k)

    # 4) build grounded prompt + generate                      [Part 1]
    context = "".join(r.get("chunk", "") for r in results)
    system = ("You are an assistant that answers based ONLY on the provided "
              "CONTEXT. If there isn't enough information, say you don't know.")
    user = f"CONTEXT: {context} QUESTION: {body.question}"
    answer = chat_completion(system, user)

    # 5) return answer + provenance
    sources = [f"chunk_id={r.get('_id')}" for r in results]
    return AnswerResponse(answer=answer, sources=sources)

Trace it against the series:

  1. Ownership check — get_owned_document_oid looks up the document by id and owner. If the user doesn't own it, 404. This runs first, before any expensive work, and it's the security boundary that makes a shared multi-user system safe.
  2. Embed the question — same model that embedded the chunks (Part 1). It has to be the same model, or the query vector lives in a different space than the document vectors and similarity is meaningless.
  3. Hybrid search — hybrid_search_chunks runs vector + BM25 under $rankFusion, scoped to this document, returning ~candidates (default 30) chunks tuned for recall (Parts 3–4).
  4. Rerank — a cross-encoder reorders the candidates and keeps the top k (default 5), tuned for precision — or, if no rerank key is configured or the call fails, gracefully returns the first k unchanged (Part 3).
  5. Generate — the surviving chunks become the CONTEXT, and the system prompt hard-constrains the model to answer only from that context, and to admit ignorance otherwise. This instruction is the difference between RAG and a chatbot that confidently hallucinates: it's what keeps answers grounded in your documents.
  6. Return with provenance — the answer ships alongside the chunk_ids it was built from, so a caller can show sources and a developer can debug a bad answer by inspecting exactly which chunks fed it.

The two tuning knobs from Part 3 — candidates and k — are request fields on AskRequest, so callers dial the recall/precision/cost trade-off per query without redeploying.

That's the complete lifecycle: auth → ownership → embed → hybrid retrieve → rerank → grounded generation → cited answer. Five subsystems, one coherent request.

Configuration: the abstraction made concrete

Everything that varies between environments lives in .env, which is what makes the same code run three different ways:

# MongoDB
MONGODB_URI=mongodb+srv://<user>:<pass>@<cluster>/?retryWrites=true&w=majority
MONGODB_DB=curso_api

# LLM — hosted OpenAI (leave OPENAI_BASE_URL unset)
OPENAI_API_KEY=sk-...
OPENAI_EMBEDDING_MODEL=text-embedding-3-large   # 3072 dims → index must match
OPENAI_CHAT_MODEL=gpt-4o-mini                    # swap for a current model freely
# OPENAI_BASE_URL=http://localhost:11434/v1      # ← uncomment to run on Ollama

# Reranking — optional; absent ⇒ rerank degrades to a no-op
RERANK_API_KEY=...
RERANK_BASE_URL=https://api.cohere.com
RERANK_MODEL=rerank-v3.5

# JWT
SECRET_KEY=change-this
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=60

Three independent decisions, all configuration, all without touching code: where the LLM runs (hosted vs Ollama vs any compatible gateway, via OPENAI_BASE_URL), whether reranking is on (set a key or don't), and which models do the embedding and chat. That flexibility is the entire reward for building on the OpenAI-compatible interface and treating reranking as an optional layer.

Standing it up

Two non-obvious setup steps the README spells out:

Create the search indexes. The pipeline is dead on arrival without them — a vector_index for $vectorSearch and a text_index for the BM25 half of hybrid search, both on the chunks collection (full definitions and the legacy-vs-current syntax caveat are in Part 4). And the recurring reminder: the vector index's dimension must match your embedding model — 3072 for text-embedding-3-large.

Bootstrap the first user. Every user route requires a token, but you can't log in without a user — a chicken-and-egg problem. The README's answer is to insert one user directly into Mongo with a bcrypt hash generated at the REPL, then log in normally. A cleaner production answer is a one-off seed script or an admin-only registration path.

Then it's the usual uvicorn app.main:app --reload, with interactive docs at /docs.

What you'd harden before production

The repo is refreshingly honest about its own rough edges — its ContextProject.md reads like a code-review TODO. The ones that actually matter, roughly in priority order:

  1. The OpenAI client is synchronous, called from async routes. embed_texts and chat_completion make blocking network calls inside async def handlers, which stalls the event loop for the whole round-trip and quietly serializes concurrent requests. This is the first thing to fix for any real concurrency: move to the SDK's AsyncOpenAI client, or offload the blocking calls to a thread pool. (Recall from Part 4 that the database layer is already correctly async via Motor — it's only the LLM calls that block.)
  2. "Upsert" only ever inserts. As covered in Part 2, upsert_document always creates a new document; re-uploading the same file silently duplicates it. Implement a real upsert keyed on something stable (e.g. owner + title) before users can re-upload.
  3. Add the operational indexes. A unique index on users.email, plus plain indexes on documents.owner and chunks.document_id, keep correctness and performance from degrading as data grows. A startup hook that ensures these indexes exist is the clean way to guarantee it.
  4. Centralize configuration. load_dotenv() is called from several modules independently. A single settings module (e.g. a Pydantic BaseSettings) imported everywhere removes the duplication, makes the config testable, and stops environment variables from being read ad hoc across the codebase.
  5. Tighten input validation. The PDF MIME check accepts binary/octet-stream, where the conventional value is application/octet-stream — a small bug that lets some odd content types through and rejects some valid ones.
  6. Add tests. With no transactions around the multi-step ingestion and several external services on the request path, a test suite with a fixture database and mocked OpenAI/rerank calls is how you keep the RAG flow honest as it evolves.
  7. Reach for transactions where partial writes hurt. Ingestion does an insert, a delete, and an insert with no atomicity; a mid-flight crash can leave inconsistent state. Wrap the multi-document write in a transaction, or make it idempotently retryable.

None of these are reasons not to learn from the repo — they're the syllabus for turning a clear reference implementation into a production service, and the fact that they're enumerated rather than hidden is exactly what makes it a good teaching codebase.

The takeaway

A modern RAG system is not one clever model call. It's a pipeline of cheap, well-separated stages — embed, retrieve broadly, rerank narrowly, generate under tight grounding — each tuned for one job, with sensible failure behavior when a piece is missing. fastapi-basic-rag makes that pipeline legible: 700-odd lines where you can see the LLM abstraction, the chunking, the two-phase retrieval, the Atlas indexes, and the auth boundary all at once, and trace a single question from HTTP request to cited answer.

Clone it, point it at Ollama to run it for free, then swap in hosted OpenAI and turn on reranking and watch the answers sharpen. That progression — local to hosted, vector to hybrid to reranked — is the practical curriculum for building retrieval systems that actually work.

The series

  1. One Client, Two Backends: OpenAI Embeddings and Chat (and Swapping in Ollama)
  2. From Document to Vector: Chunking, Upsert, and Vector Search
  3. Two-Phase Retrieval: Hybrid Search with $rankFusion and Cross-Encoder Reranking
  4. One Database, Two Search Engines: MongoDB Atlas Search for RAG
  5. The Whole Machine: A FastAPI RAG Service, End to End ← you are here

References