Production RAG in Python

The main task of the project structure is to separate the transport layer, domain models, adapters to external services and orchestration services. Then the retrieval logic is not spread across endpoints, and the indexer does not depend on HTTP and UI.

EXAMPLE OF FILE TREE

app/
  main.py
  api/
    ask.py
    ingest.py
  core/
    config.py
    logging.py
    deps.py
  domain/
    models.py
    protocols.py
  adapters/
    embeddings_openai.py
    vector_store_qdrant.py
    llm_openai.py
    reranker_api.py
  services/
    chunking_service.py
    indexing_service.py
    retrieval_service.py
    prompt_service.py
    answer_service.py
  workers/
    reindex_worker.py
  tests/
    test_chunking.py
    test_retrieval.py
    test_answer_flow.py

`main.py` must not contain retrieval logic. His job is to create a FastAPI app, connect routers, configure lifespan hooks and, if necessary, warm up shared clients. The thinner `main.py`, the lower the risk of turning the project into one giant file.

app = FastAPI(...)

app.include_router(ask_router)
app.include_router(ingest_router)

@asynccontextmanager
async def lifespan(app):
  init_shared_clients()
  yield
  close_shared_clients()

Endpoint should not build embeddings itself, go to Qdrant and render the prompt. It accepts a request, receives the necessary services through dependency injection and delegates to orchestration. Otherwise, the code will quickly become unsupportable.

bad:
  endpoint = all business logic

good:
  endpoint -> service.answer(request)
  endpoint -> return response_model

Dependency injection in FastAPI is convenient to use not only for auth, but also for delivering settings, clients and services to endpoints. Then the handler does not create Qdrant/OpenAI clients for each request and does not know the wiring details.

def get_retrieval_service() -> RetrievalService:
  return RetrievalService(
    vector_store=get_vector_store(),
    embedder=get_embedder(),
    reranker=get_reranker(),
  )

@router.post("/ask")
async def ask(req: AskRequest, svc: RetrievalService = Depends(get_retrieval_service)):
  ...

If `0.72`, `20`, `3000` and `text-embedding-3-large` are scattered across different files, tuning and debugging will become a pain. Configuration should be centralized and typed, usually through Pydantic settings or a similar layer.

EMBED_MODEL
CHAT_MODEL
QDRANT_URL
COLLECTION_NAME
RETRIEVAL_TOP_K
RETRIEVAL_MIN_SCORE
RERANK_TOP_K
MAX_CONTEXT_TOKENS

Even if the project is small, it’s worth starting a model right away. The minimum set is: request/response for API, chunk model for indexing, search hit model for retrieval, packed context item and trace model for observability. This makes the code predictable and reduces the number of “formless vocabularies”.

AskRequest
AskResponse
ChunkRecord
SearchHit
RerankHit
ContextItem
RAGTrace

The API layer should be extremely clear: what exactly the client can send and what exactly it will receive in response. It’s better to immediately provide `filters`, `history`, `tenant_id`, `debug` and fields for citations than to break the contract later in battle.

class AskRequest(BaseModel):
    query: str
    history: list[str] = []
    tenant_id: str
    filters: dict[str, str] | None = None
    debug: bool = False

class AskResponse(BaseModel):
    answer: str
    citations: list[dict]
    trace_id: str | None = None

Once the search has returned candidates, all further logic revolves around `SearchHit`. This is where `score`, `payload`, `text`, `doc_id`, sometimes `rerank_score`, sometimes `source_label` live. It is better to explicitly store retrieval fields than to re-extract them from a raw Qdrant/SQL response each time.

class SearchHit(BaseModel):
    chunk_id: str
    doc_id: str
    text: str
    retrieval_score: float
    rerank_score: float | None = None
    metadata: dict = Field(default_factory=dict)

Indexing pipeline becomes easier if the chunk after parser/chunker already has an explicit form. Then the embedder, vector store adapter and worker do not talk through “something similar to a dict”, but through a normal object with ID, text and metadata.

class ChunkRecord(BaseModel):
    chunk_id: str
    doc_id: str
    text: str
    section: str | None = None
    token_count: int
    metadata: dict = Field(default_factory=dict)

If RetrievalService directly depends on a specific `QdrantClient` and a specific `OpenAI` SDK, the project becomes tightly coupled. It is much cleaner to describe small protocol interfaces: `Embedder`, `VectorStore`, `LLMClient`, `Reranker`. Then the library changes, but the orchestration code remains.

class Embedder(Protocol):
    def embed_query(self, text: str) -> list[float]: ...

class VectorStore(Protocol):
    def search(self, vector: list[float], limit: int, filters: dict | None) -> list[SearchHit]: ...

class LLMClient(Protocol):
    def generate(self, prompt: str) -> str: ...

Chunking is one of the most sensitive layers of RAG, so it is best to isolate it. Then you can separately test the cutting rules for Markdown, PDF, code blocks and overlap. And most importantly, the chunking logic version becomes explicit, and not hidden in a random utility function.

class ChunkingService:
    def split(self, doc: ParsedDocument) -> list[ChunkRecord]:
        ...

chunker = ChunkingService(
    max_tokens=300,
    overlap_tokens=40,
)

IndexingService is where parser, chunker, embedder and vector store come together. It receives a raw document, builds chunk records, batches embedding requests and makes an upsert into the index. This is a separate use-case orchestration, and not a job in one endpoint.

class IndexingService:
    def ingest_document(self, doc: RawDocument) -> int:
        parsed = self.parser.parse(doc)
        chunks = self.chunker.split(parsed)
        vectors = self.embedder.embed_documents([c.text for c in chunks])
        self.vector_store.upsert(chunks, vectors)
        return len(chunks)

Services generally do not need to know whether `query_points`, `upsert` or SQL is being called via pgvector. Therefore, the adapter must convert internal domain objects into the required SDK format and back. Then migration from one vector DB to another does not break the upper layers.

class QdrantVectorStore(VectorStore):
    def upsert(self, chunks: list[ChunkRecord], vectors: list[list[float]]) -> None:
        points = [
            models.PointStruct(
                id=chunk.chunk_id,
                vector=vec,
                payload={"doc_id": chunk.doc_id, "text": chunk.text, **chunk.metadata},
            )
            for chunk, vec in zip(chunks, vectors)
        ]
        self.client.upsert(collection_name=self.collection, points=points)

Indexing documents can be time-consuming and expensive. Therefore, production systems often move ingestion to a separate worker: Celery/RQ/Dramatiq/Arq/cron-job/queue consumer. The HTTP endpoint only queues the task and returns `202 Accepted`.

@router.post("/ingest")
async def ingest(req: IngestRequest):
    job_id = enqueue_reindex(req.doc_id)
    return {"job_id": job_id, "status": "queued"}

When changing the embedding model, chunking logic or metadata schema, old and new points cannot be quietly mixed. It is practically useful to store `embedding_model_version`, `chunker_version` and be able to safely reindex a document or an entire collection without double garbage.

It is RetrievalService that most often answers the user’s question “what happens after cosine similarity?” It builds query embedding, calls the vector store, filters the results, decides about rerank, does dedup and prepares context candidates for the next layer.

class RetrievalService:
    def retrieve(self, query: str, filters: dict | None) -> list[SearchHit]:
        q_emb = self.embedder.embed_query(query)
        hits = self.vector_store.search(q_emb, limit=self.top_k, filters=filters)
        hits = self.apply_thresholds(hits)
        hits = self.dedup(hits)
        if self.reranker:
            hits = self.rerank(query, hits)
        return hits

Inside the SDK adapter, the Qdrant/pgvector/FAISS response is turned into normalized `SearchHit` objects. It is important that a single format emerges. Then RetrievalService does not know whether the score came as `distance`, `similarity` or the `result.score` field.

def search(self, vector, limit, filters):
    result = self.client.query_points(
        collection_name=self.collection,
        query=vector,
        limit=limit,
        query_filter=to_qdrant_filter(filters),
        with_payload=True,
    )
    return [self._to_hit(point) for point in result.points]

Once the vector store has returned `score`, the next step almost always looks like pure Python logic. This is where the decision is made to “leave a hit”, “ask for clarification”, “return the FAQ directly”, “run a reranker” or “go to LLM”.

def apply_thresholds(self, hits: list[SearchHit]) -> list[SearchHit]:
    if not hits:
        return []

    if hits[0].retrieval_score < self.min_score:
        return []

    return [h for h in hits if h.retrieval_score >= self.per_hit_score]

If retrieval returns three almost identical adjacent pieces, the service must correct this before the prompt layer. Therefore, in real code there are almost always `dedup_hits`, `group_by_doc`, `expand_neighbors`, `fetch_parent_blocks` or similar methods.

def dedup(self, hits):
    seen = set()
    out = []
    for hit in hits:
        key = (hit.doc_id, hit.chunk_id)
        if key in seen:
            continue
        seen.add(key)
        out.append(hit)
    return out

Reranker should not be “another line inside the endpoint”. It’s logical to move it into the adapter and call it from RetrievalService so that you can easily turn off rerank, change the provider, and log the original retrieval scores separately from the rerank scores.

def rerank(self, query: str, hits: list[SearchHit]) -> list[SearchHit]:
    scored = self.reranker.score(query, [h.text for h in hits])
    for hit, rerank_score in zip(hits, scored):
        hit.rerank_score = rerank_score
    return sorted(hits, key=lambda h: h.rerank_score or 0, reverse=True)

If you want dense + BM25 + metadata filters, it is better to build it as two separate retrieval strategies and then combine them via fusion. This way the code remains testable: `dense_retriever`, `sparse_retriever`, `fusion_service`, and not one 200-line function with a bunch of flags.

dense_hits = self.dense_retriever.retrieve(query, filters)
sparse_hits = self.sparse_retriever.retrieve(query, filters)
hits = self.fusion_service.rrf(dense_hits, sparse_hits)
return hits[:self.top_k]

The Prompt layer receives a ready-made `SearchHit` or `ContextItem` list and only renders the context, system instructions, refusal policy and citation labels. If the prompt builder suddenly went into the database for additional pieces, the boundaries of the layers begin to flow.

class PromptService:
    def build(self, query: str, context_items: list[ContextItem]) -> str:
        ...

Packing is almost always needed as a separate function or service method. It counts tokens, limits length, assigns source labels, and turns `SearchHit` into `ContextItem`. This is where the real cut-off of the context often occurs, and not at the vector DB level.

def pack_context(hits: list[SearchHit], max_tokens: int) -> list[ContextItem]:
    items = []
    budget = max_tokens
    for i, hit in enumerate(hits, start=1):
        n = estimate_tokens(hit.text)
        if n > budget:
            continue
        items.append(ContextItem(
            label=f"SOURCE {i}",
            chunk_id=hit.chunk_id,
            text=hit.text,
            token_count=n,
            metadata=hit.metadata,
        ))
        budget -= n
    return items

If the RetrievalService is responsible for the search, then the AnswerService is usually responsible for the complete happy path of the request. It calls retrieval, decides to refuse, packs the context, builds a prompt, calls LLM and collects the final response object.

class AnswerService:
    def answer(self, req: AskRequest) -> AskResponse:
        hits = self.retrieval.retrieve(req.query, req.filters)
        if not hits:
            return self.refusal("Insufficient data in sources")
        context = self.pack_context(hits)
        prompt = self.prompt_service.build(req.query, context)
        raw = self.llm.generate(prompt)
        return self.postprocess(raw, context)

The final text of the response rarely goes to the client “as is”. Post-processing can pull out citation labels, substitute real URLs, check JSON schema, remove junk markdown, and even fall back to a safer response if the model produced something outside the contract.

def postprocess(self, raw: str, context: list[ContextItem]) -> AskResponse:
    citations = extract_citations(raw)
    citations = map_labels_to_sources(citations, context)
    answer = cleanup_markdown(raw)
    return AskResponse(answer=answer, citations=citations)

If the failure is scattered across the design in different lines, the behavior of the system quickly becomes inconsistent. It is more practical to have one function `refusal_response(reason=...)` so that the UI, analytics, and trace understand the failure in the same way.

def refusal(self, reason: str) -> AskResponse:
    return AskResponse(
        answer="I don’t see sufficient evidence in the available sources.",
        citations=[],
        trace_id=None,
    )

If trace is collected into a separate model, it can be logged, stored in the database, shown in the debug UI and used in golden tests. This is much better than just printing random lines about score into the log. The Trace should reflect the entire path of the request through the system.

class RAGTrace(BaseModel):
    trace_id: str
    query: str
    filters: dict
    retrieval_hits: list[SearchHit]
    packed_chunk_ids: list[str]
    prompt_tokens: int
    answer_preview: str
    latency_ms: dict[str, int]

A good service has not only HTTP latency, but also metrics of retrieval and generation layers: average top-1 score, percentage of empty retrieval, rerank latency, packed token count, refusal rate, cache hit rate. These numbers help you understand what is broken even before reading the logs.

In production code, the cache is usually useful in three places: embeddings of identical chunks during ingestion, retrieval-only shortcut / semantic cache for similar questions, and prompt/result cache for very expensive LLM calls. But the cache must be strictly tied to the index version, otherwise you will return the outdated truth.

embedding_cache_key =
  sha256(embedding_model + chunk_text)

semantic_cache_key =
  tenant_id + nearest_query_cluster + index_version

answer_cache_key =
  prompt_hash + model_name

It is better to cover Production RAG in layers: unit tests for chunking and thresholds, integration tests for retrieval adapter, golden tests for end-to-end answer flow, and separate regression cases for known failures. Then the bug is localized to a specific function faster.

unit:
  test_chunker_splits_headers()
  test_apply_thresholds_refuses_low_scores()

integration:
  test_qdrant_adapter_maps_hits()

golden:
  test_answer_flow_for_known_query()

It is very useful to be able to locally simulate failures: empty retrieval, slow reranker, broken LLM output, timeouts vector store, outdated index version. Then you can see how the service really degrades: it honestly refuses, crashes 500 times, or begins to hallucinate.

When everything is put together correctly, the request flows through the project almost like a dependency diagram. This is the mental model that is useful to keep in mind when reading or writing production RAG code.

END-TO-END FLOW

app/api/ask.py
  ask(req, answer_service)
        ↓
app/services/answer_service.py
  answer(req)
        ↓
app/services/retrieval_service.py
  retrieve(query, filters)
        ↓
app/adapters/embeddings_openai.py
  embed_query(query)
        ↓
app/adapters/vector_store_qdrant.py
  search(q_emb, top_k, filters)
        ↓
app/services/retrieval_service.py
  thresholds → dedup → rerank
        ↓
app/services/prompt_service.py
  build(query, context)
        ↓
app/adapters/llm_openai.py
  generate(prompt)
        ↓
app/services/answer_service.py
  postprocess → AskResponse

In production RAG, it is not the framework or the endpoint that most often change. The following changes: chunking rules, metadata filters, thresholds, fusion weights, rerank policy, prompt template, refusal rules and a set of logged trace fields. Therefore, the project should be built so that these pieces are small and isolated.

“Production RAG in Python” is not “connected FastAPI and Qdrant”. This is neatly laid out code: APIs, contracts, adapters, orchestration services, workers, traces and tests. The sooner the project takes this form, the easier it is to explain later why, after `cosine similarity = 0.81`, the system went to `rerank`, and not immediately to LLM.