Building a Production-Ready RAG Pipeline

Retrieval-Augmented Generation (RAG) grounds Large Language Models on domain-specific, dynamic, or private data without requiring model fine-tuning. While a naive prototype takes only a few lines of code, production systems require careful attention to chunk boundaries, vector indexing, cross-encoder reranking, and context window assembly.

Architecture Overview

A production RAG architecture separates the Ingestion Pipeline (offline/asynchronous) from the Inference Pipeline (live query latency path).

1. Document Ingestion & Cleaning

Raw files (PDFs, Markdown, Notion docs, HTML) must be stripped of boilerplate, navigation elements, and corrupted encodings before processing.

2. Semantic Chunking

Fixed-size character chunking often splits critical context across boundaries. Prefer:

• Recursive character splitting: Breaks on paragraphs (`

), then sentences ( ), then words ( `).

• Chunk size & overlap: A standard baseline is 512 tokens with an overlap of 50–100 tokens to preserve sentence context across chunk borders.

3. Dense Retrieval vs. Hybrid Search

Vector search alone struggles with exact keyword matches (e.g., product IDs, SKU numbers, or specific names). Production systems blend:

• Dense vectors: Semantic similarity via bi-encoder models.
• Sparse vectors (BM25): Exact token and lexical matching.
• Reciprocal Rank Fusion (RRF): Combines rank lists from both retrievers.

4. Cross-Encoder Reranking

Vector search retrieves candidates quickly (Top-K ≈ 25–50), but bi-encoders compute document and query embeddings independently. Passing the candidates through a Cross-Encoder Reranker (such as bge-reranker-large or Cohere Rerank) scores query-document pairs jointly, filtering the list down to the most relevant Top-N ≈ 3–5 chunks.

5. Context Window Assembly & Generation

Inject retrieved passages with source metadata into a structured system prompt, instructing the model to cite sources and explicitly refuse to answer when the context does not contain sufficient information.

End-to-End Implementation

The following Python implementation demonstrates the full query-time path: dense retrieval, reranking, and grounded generation using standard client libraries.

import os
from typing import List, Dict, Any
from dataclasses import dataclass
from sentence_transformers import CrossEncoder
import chromadb
from openai import OpenAI

@dataclass
class RetrievedDocument:
    id: str
    text: str
    metadata: Dict[str, Any]
    score: float

class RAGPipeline:
    def __init__(
        self,
        collection_name: str = "kb_articles",
        reranker_model: str = "BAAI/bge-reranker-base",
    ):
        self.chroma_client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.chroma_client.get_or_create_collection(name=collection_name)
        self.reranker = CrossEncoder(reranker_model)
        self.llm_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

    def retrieve(self, query: str, top_k: int = 25) -> List[RetrievedDocument]:
        """Performs initial dense vector retrieval."""
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            include=["documents", "metadatas", "distances"]
        )

        documents = []
        if not results["documents"] or not results["documents"][0]:
            return documents

        for doc_id, doc_text, meta, dist in zip(
            results["ids"][0],
            results["documents"][0],
            results["metadatas"][0],
            results["distances"][0]
        ):
            documents.append(
                RetrievedDocument(
                    id=doc_id,
                    text=doc_text,
                    metadata=meta,
                    score=1.0 - dist,  # Convert distance to similarity
                )
            )
        return documents

    def rerank(
        self, 
        query: str, 
        documents: List[RetrievedDocument], 
        top_n: int = 3
    ) -> List[RetrievedDocument]:
        """Applies a cross-encoder to compute query-passage interaction scores."""
        if not documents:
            return []

        pairs = [[query, doc.text] for doc in documents]
        rerank_scores = self.reranker.predict(pairs)

        for doc, score in zip(documents, rerank_scores):
            doc.score = float(score)

        # Sort descending by cross-encoder score
        reranked = sorted(documents, key=lambda x: x.score, reverse=True)
        return reranked[:top_n]

    def generate_answer(self, query: str, context_docs: List[RetrievedDocument]) -> str:
        """Constructs grounded context and queries the LLM."""
        if not context_docs:
            return "No relevant context found to answer the query."

        # Format retrieved snippets with citations
        formatted_context = "

".join(
            [f"[{i+1}] {doc.text}" for i, doc in enumerate(context_docs)]
        )

        system_prompt = (
            "You are a factual assistant. Answer questions strictly using the provided context.
"
            "If the answer cannot be determined from the context, say 'I cannot find that in the documents.'
"
            "Cite source references using the bracketed tags provided."
        )

        user_content = f"Context:
{formatted_context}

Question: {query}"

        response = self.llm_client.chat.completions.create(
            model="gpt-4o-mini",
            temperature=0.0,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_content},
            ],
        )

        return response.choices[0].message.content

    def run(self, query: str) -> str:
        """Full pipeline execution: Retrieve -> Rerank -> Generate."""
        initial_candidates = self.retrieve(query, top_k=20)
        reranked_docs = self.rerank(query, initial_candidates, top_n=3)
        return self.generate_answer(query, reranked_docs)

# Example Usage
if __name__ == "__main__":
    rag = RAGPipeline()
    query = "What is the warranty policy for enterprise hardware?"
    answer = rag.run(query)
    print(answer)