Python — FastAPI Development
FastAPI for RAG API Endpoints
Direct answer
A production RAG API in FastAPI usually needs three surfaces: an ingestion endpoint that accepts documents and enqueues processing, a query endpoint that embeds the question, retrieves chunks, and streams a grounded answer with citations, and status plus feedback endpoints around them. The rules that matter: keep chunking and embedding out of the request path, stream generation over server-sent events, and return typed citations so the client can show sources.
I build RAG backends for clients regularly, and the difference between a demo and a dependable service is almost entirely API design — contracts, streaming, and what happens off the request path. Here is the endpoint architecture that holds up.
Key facts, with sources
- In the JetBrains Python Developers Survey 2024, which collected responses from more than 30,000 Python developers, FastAPI usage jumped from 29% to 38%, overtaking Django (35%) and Flask (34%) as the most-used Python web framework. (JetBrains Python Developers Survey 2024)
- The 2025 Stack Overflow Developer Survey shows FastAPI at 14.8% of respondents doing extensive work with it, edging out Flask at 14.4% and Django at 12.6%. (Stack Overflow Developer Survey 2025)
- FastAPI's official documentation cites independent TechEmpower benchmarks showing FastAPI applications running under Uvicorn as one of the fastest Python frameworks available, ranked only below Starlette and Uvicorn themselves. (FastAPI official documentation)
- FastAPI surpassed Flask in GitHub stars in December 2025, reaching roughly 88,000 stars compared to Flask's 68,400. (DZone)
- Industry analysis of FastAPI's 2025 growth reports about 40% year-over-year growth in job mentions and production adoption at companies including Uber, Netflix, and Microsoft. (byteiota)
The endpoint surface a RAG service actually needs
Every RAG backend I ship converges on the same small surface. Ingestion: accept a document, persist the raw file, enqueue processing, return a job ID with a 202. Query: take a question, run retrieval and generation, stream the answer. Job status: let clients poll ingestion progress. Feedback: record thumbs up or down against a specific answer, because that log becomes your evaluation set later.
Resist the temptation to expose retrieval internals as public endpoints early — a raw "search chunks" route becomes a contract you must maintain. Keep the public API at the level of questions and documents, and keep the vector store an implementation detail you can swap.
Typed contracts with citations built in
The response schema is where RAG APIs earn trust. An answer without sources is a liability in any serious deployment, so citations are first-class in my contracts: which document, which chunk, what relevance score. Clients render them as links or footnotes, and support teams use them to debug why the system said what it said.
On the request side, validate aggressively: bound the question length, cap top_k so a caller cannot request five hundred chunks and detonate your token budget, and type the metadata filters you support instead of accepting arbitrary dicts.
from pydantic import BaseModel, Field
class AskRequest(BaseModel):
question: str = Field(min_length=3, max_length=2000)
top_k: int = Field(default=5, ge=1, le=20)
filters: dict[str, str] | None = None
class Citation(BaseModel):
document_id: str
title: str
chunk_index: int
score: floatStream the answer, send sources first
Retrieval finishes in tens or hundreds of milliseconds; generation takes seconds. So I stream over server-sent events and send the citations as the first event, before any answer tokens. The client renders the source list immediately — useful perceived progress — and then tokens flow in. A final done event closes the stream cleanly so clients can distinguish completion from a dropped connection.
Named SSE events make this trivial to parse client-side, and the pattern degrades gracefully: the same handler logic can back a non-streaming JSON variant for callers that cannot consume SSE.
import json
from fastapi.responses import StreamingResponse
@app.post("/ask")
async def ask(req: AskRequest):
query_vector = await embed(req.question)
chunks = await vector_store.search(
query_vector, limit=req.top_k, filters=req.filters
)
async def stream():
sources = [c.model_dump() for c in build_citations(chunks)]
yield f"event: sources\ndata: {json.dumps(sources)}\n\n"
async for token in generate_answer(req.question, chunks):
yield f"data: {json.dumps(token)}\n\n"
yield "event: done\ndata: {}\n\n"
return StreamingResponse(stream(), media_type="text/event-stream")Ingestion is a pipeline, not a request handler
Parsing a PDF, chunking it, embedding a few hundred chunks, and upserting vectors can take minutes for a large document. None of that belongs in an HTTP request. The ingestion endpoint should do the minimum synchronously — validate the file type and size, persist the raw bytes, create a job record, enqueue — and return 202 with the job ID.
The worker does the heavy lifting and updates job status through stages so the status endpoint can report real progress: parsing, chunking, embedding, indexing. Make ingestion idempotent by hashing document content — re-uploading the same file should not create duplicate chunks that then show up twice in every retrieval, which is one of the most common data-quality bugs I find in existing RAG systems I'm brought in to fix.
Latency budget and cost controls
A query request pays three sequential costs: embedding the question, searching the vector store, and generation. Generation dominates, which is why streaming matters most — but the first two are where cheap wins live. Caching question embeddings helps when users repeat common queries, and retrieval itself is usually fast enough that top_k caps are about token cost, not search speed.
Enforce a per-request token ceiling on generation and set timeouts around every upstream call with a clean error event on the stream when one trips — a stream that dies silently is far worse for clients than one that reports failure. Log token counts per request tied to the API key or user, because RAG cost problems typically arrive as one power user, not as uniform growth.
When to hire senior help
Bring in senior help when your API needs to handle real concurrency, when you are designing service boundaries and auth for the first time, or when an existing FastAPI codebase mixes sync and async code and latency is degrading. An experienced engineer can usually diagnose event-loop blocking and connection-pool misconfiguration in days, which is far cheaper than re-architecting after launch. If your stack includes React Native + Python + AI, a senior engineer who owns the full product beats coordinating multiple juniors.
Bottom line
Dhairya Senjaliya ships Python — FastAPI Development projects worldwide — book a scoping call to discuss your specific situation.
Common pitfalls to avoid
- ✕Calling blocking libraries (classic SQLAlchemy sessions, requests, heavy file I/O) inside async def endpoints, which stalls the event loop and erases FastAPI's concurrency advantage
- ✕Deploying a single Uvicorn process with no process manager or worker scaling, leaving most CPU cores idle under production load
- ✕Treating the auto-generated OpenAPI docs as a versioning strategy, then breaking mobile and partner clients when response schemas change
- ✕Running large payloads through deeply nested Pydantic models on every request and response, adding serialization latency that shows up only at scale
Frequently asked questions
How do I return citations from a RAG API?
Make citations a typed part of the response contract — document ID, title, chunk index, and relevance score per source — rather than asking the model to embed references in prose. In streaming responses, send the citation list as the first server-sent event before answer tokens, so clients render sources immediately and can attribute the answer even if generation is later cut off.
Should document ingestion happen in the API request?
No. Parsing, chunking, and embedding a document can take minutes, far beyond a reasonable HTTP timeout. The endpoint should validate the upload, persist the raw file, enqueue a background job, and return 202 with a job ID that clients poll for status. Hash document content for idempotency so repeated uploads never create duplicate chunks polluting retrieval.
Why stream RAG responses instead of returning JSON?
Generation takes seconds while retrieval takes milliseconds, so a non-streaming endpoint leaves users staring at a spinner for the full duration. Streaming over server-sent events delivers sources immediately and the first answer tokens within roughly a second, which users experience as fast. Keep a non-streaming JSON variant only for machine-to-machine callers that cannot consume SSE.
Is FastAPI mature enough for production?
Yes. It was the most-used Python web framework in the JetBrains 2024 survey at 38%, and companies including Uber, Netflix, and Microsoft run it in production. The ecosystem for auth, ORMs, and testing is now well established.
How much faster is FastAPI than Flask or Django really?
Independent TechEmpower benchmarks place FastAPI among the fastest Python frameworks, and published comparisons show several times Flask's throughput on I/O-bound endpoints. For CPU-bound work or database-bottlenecked apps, the framework choice matters far less than query and infrastructure design.
Should we pick FastAPI or Django for a new SaaS backend?
FastAPI suits API-first products, microservices, and ML model serving because of async support and automatic OpenAPI docs. Django ships batteries included (admin, ORM, auth) and is often faster to launch a conventional CRUD product. Many teams run both, per the JetBrains finding that a third of Django developers also use Flask or FastAPI.
Bottom line: Dhairya Senjaliya ships Python — FastAPI Development projects worldwide. Book a scoping call at https://dhairyasenjaliya.com/#book-call.