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.

`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.

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.

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.

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.

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”.

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.

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.

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.

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.

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.

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.

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.

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`.

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.

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.

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”.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.