# Building a Robust RAG Pipeline Architecture for Production

> Source: <https://dev.to/ayush_kumar_085a0f2c54e3f/building-a-robust-rag-pipeline-architecture-for-production-3j25>
> Published: 2026-07-14 13:47:18+00:00

**Answer up front:** A RAG pipeline architecture is a set of connected services that ingest raw documents, turn them into embeddings, store them in a vector database, retrieve the most relevant chunks, and finally feed those chunks to a language model for generation. In practice you need a modular design, solid chunking, a fast vector store with hybrid search, and observability that lets you spot bottlenecks before they break your service.

Below I walk through each piece of that puzzle, share the code I run in production, and point out the trade-offs that kept me up at night.

A RAG pipeline architecture typically consists of:

`text-embedding-ada-002`

or a local BERT) and produces dense vectors.
In my last project I ran each component as a small Docker container behind a Cloud Run service. The biggest pain point was the **session leak** in the SQLAlchemy layer that silently ate DB connections after a few thousand requests. I fixed it by scoping sessions to the request lifecycle – see my write-up on [FastAPI SQLAlchemy Session Leak Detection](https://www.logiclooptech.dev/fastapi-session-leak-detection-sqlalchemy-long-running/) for the exact steps.

LangChain already gives you the building blocks – `DocumentLoaders`

, `TextSplitters`

, `Embedders`

, and `Retrievers`

. The trick is to keep the configuration external so you can swap out a component without rebuilding the whole service.

```
# config.yaml
pipeline:
  loader: "pdf"
  splitter:
    type: "RecursiveCharacter"
    chunk_size: 1000
    chunk_overlap: 200
  embedder:
    provider: "openai"
    model: "text-embedding-ada-002"
  vector_store: "pinecone"
  retriever:
    top_k: 5
    use_hybrid: true
python
# app/pipeline.py
import yaml
from fastapi import Depends
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
from langchain.chains import RetrievalQA

def load_config():
    with open("config.yaml") as f:
        return yaml.safe_load(f)

def build_retriever(cfg=Depends(load_config)):
    # 1. loader
    loader = PyPDFLoader(cfg["pipeline"]["loader_path"])
    docs = loader.load()

    # 2. splitter
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=cfg["pipeline"]["splitter"]["chunk_size"],
        chunk_overlap=cfg["pipeline"]["splitter"]["chunk_overlap"],
    )
    chunks = splitter.split_documents(docs)

    # 3. embedder
    embedder = OpenAIEmbeddings(model=cfg["pipeline"]["embedder"]["model"])

    # 4. vector store
    vect = Pinecone.from_documents(
        chunks,
        embedder,
        index_name="rag-index",
        namespace="prod",
    )

    # 5. retriever
    retriever = vect.as_retriever(
        search_type="mmr" if cfg["pipeline"]["retriever"]["use_hybrid"] else "similarity",
        search_kwargs={"k": cfg["pipeline"]["retriever"]["top_k"]},
    )
    return retriever

def get_qa_chain(retriever=Depends(build_retriever)):
    return RetrievalQA.from_chain_type(
        llm=OpenAI(temperature=0),
        chain_type="stuff",
        retriever=retriever,
    )
```

The FastAPI endpoint then just calls `qa_chain.run(query)`

. Because every step is instantiated on demand, you can replace the `PyPDFLoader`

with a `SQLLoader`

or swap Pinecone for a self-hosted Qdrant without touching the rest of the code.

Chunk size is a classic source of bugs. Too small and you lose context; too large and the embedding vector becomes a blurry average of unrelated sentences. In my experiments:

| Chunk Size | Overlap | Retrieval Latency (ms) | Answer Quality |
|---|---|---|---|
| 200 | 50 | 120 | Misses multi-sentence facts |
| 500 | 100 | 85 | Good balance |
| 1000 | 200 | 70 | Best for long technical docs |

I settled on 500-character chunks with 20 % overlap. The overlap ensures that a phrase split across a boundary still appears in at least one chunk.

Embedding choice matters too. OpenAI's `text-embedding-ada-002`

is cheap and works well for English, but for multilingual corpora I switched to a `sentence-transformers`

model hosted on a GPU-enabled inference server. The cost difference was stark: $0.0004 per 1k tokens vs $0.0012 for the transformer, but the recall improvement for non-Latin scripts was worth it.

**When not to use a dense embedder:** If your corpus is under 10 k short snippets, a classic TF-IDF vectorizer can be faster and cheaper. You lose semantic matching, but for exact keyword queries the trade-off may be acceptable.

Vector stores differ on latency, scalability, and hybrid capabilities.

| Store | Cloud-native | Hybrid (BM25 + ANN) | Cost (per GB/mo) | Known Pitfalls |
|---|---|---|---|---|
| Pinecone | Yes | ✅ | $0.30 | Cold start latency |
| Qdrant | Self-hosted | ✅ (via `qdrant-client` ) |
$0 (self) | Need to manage replicas |
| Weaviate | Yes | ✅ (built-in) | $0.25 | Schema migrations can be tricky |
| Milvus | Self-hosted | ❌ (requires external BM25) | $0 (self) | Complex config for large clusters |

I prefer Pinecone for quick SaaS spin-up, but Qdrant gave me tighter control over latency when I moved the vector store into the same VPC as the FastAPI service. The hybrid query looks like this:

``` python
# hybrid_retriever.py
def hybrid_search(vector_store, query, top_k=5):
    # 1. embed the query
    query_vec = OpenAIEmbeddings().embed_query(query)

    # 2. perform ANN search
    ann_results = vector_store.query(
        vector=query_vec,
        top_k=top_k,
        include_metadata=True,
    )

    # 3. BM25 fallback (if vector store supports it)
    bm25_results = vector_store.hybrid_search(
        query=query,
        top_k=top_k,
    )

    # 4. merge and re‑rank
    combined = ann_results + bm25_results
    combined.sort(key=lambda r: r["score"], reverse=True)
    return combined[:top_k]
```

A common failure mode is **score mismatch**: the ANN scores are in the range `[0, 2]`

while BM25 scores can be a hundred-fold larger. Normalizing scores before merging prevents the BM25 side from drowning out the semantic matches.

Two families of metrics matter:

**Retrieval effectiveness** – measured with `Recall@k`

, `Mean Reciprocal Rank (MRR)`

, and `Precision@k`

. I generate a test set of 200 queries with known ground-truth passages and run the pipeline nightly.

**Generation quality** – `BLEU`

, `ROUGE-L`

, and more importantly **human-rated factuality**. I run a lightweight A/B test where half the traffic hits the new pipeline and the other half hits the previous version. A simple Google Form collects annotator scores.

Automated latency monitoring is a must. In production I instrument each FastAPI route with OpenTelemetry and push metrics to Cloud Monitoring. A typical latency breakdown looks like:

If any component spikes beyond its 95th percentile, an alert fires. I once saw a sudden 2× increase in embedding latency caused by a throttling rule on the OpenAI API key. The fix was to add exponential back-off and a fallback local embedder.

Each stage lives in its own container:

`ingestor`

– FastAPI + Celery worker, pulls new docs every hour.`embedder`

– tiny FastAPI service that only does `POST /embed`

.`vector-store`

– managed Pinecone or a self-hosted Qdrant cluster.`api-gateway`

– the public FastAPI endpoint that runs retrieval and generation.I use **Google Cloud Run** for the stateless services because it auto-scales to zero and handles TLS out of the box. The vector store runs on a managed GKE node pool with **node-autoscaling** enabled.

Every push to `main`

triggers a GitHub Actions workflow that builds Docker images, runs unit tests, and pushes to Artifact Registry. Then a Cloud Run deployment rolls out the new revision. My CI/CD pipeline is described in detail in [Automating Production: A CI/CD Pipeline for Google Cloud Run with GitHub Actions](https://www.logiclooptech.dev/automating-production-a-cicd-pipeline-for-google-cloud-run-with-github-actions/).

`temperature=0`

and `max_tokens`

limits to keep per-request spend predictable.`/healthz`

that runs a quick vector store ping and a tiny embed test. Cloud Run will automatically replace unhealthy instances.**How many chunks should I store per document?**

It depends on document length and the chosen chunk size. For a 10 k word technical manual, 500-character chunks yield roughly 40–50 chunks, which balances retrieval relevance and storage cost.

**Can I use a RAG pipeline without a vector store?**

Yes, if your use-case is tiny (under 1 k documents) and you only need keyword search, a relational DB with full-text indexes can suffice. You lose semantic matching, though.

**What’s the simplest way to add hybrid search to an existing vector store?**

If your store lacks native hybrid capability, run a parallel SQLite FTS5 index on the same metadata and merge the results as shown in the `hybrid_search`

function above.

**Do I need a separate service for embeddings?**

Not strictly. For low traffic you can embed inline in the ingestion step. For high-throughput pipelines, a dedicated embedder service isolates latency spikes and lets you cache results.

That’s the blueprint I follow for every RAG pipeline I ship to production. It’s not magical, but it’s reliable enough to keep my services up 99.9 % while staying within a predictable budget. Happy building!