Building & Optimizing Production-Grade RAG Pipelines

Introduction to Modern RAG

Retrieval-Augmented Generation (RAG) has transformed how enterprise Large Language Models (LLMs) access dynamic, proprietary, or domain-specific knowledge without costly model fine-tuning. However, naive RAG architectures—typically consisting of basic fixed-size text chunking followed by cosine similarity searches—often degrade significantly when deployed in production environments.

In this article, we explore the architectural enhancements necessary to build resilient, hallucination-resistant RAG systems capable of accurately synthesizing multi-hop queries across diverse document formats.

flowchart LR
    A[Raw Documents] --> B[Semantic Chunking]
    B --> C[Vector + Keyword Indexing]
    D[User Query] --> E[Query Expansion / Multi-Query]
    E --> F[Hybrid Search: Dense + BM25]
    F --> G[Cross-Encoder Reranker]
    G --> H[LLM Synthesis & Citation Grounding]
    H --> I[Validated Output]

---

1. Advanced Chunking Strategies

Standard character-count or token-length chunking often fractures context across sentence boundaries or table structures. To maintain cohesive semantic units, modern pipelines utilize Semantic Chunking and Hierarchical Chunking.

Semantic Boundary Chunking in Python

Semantic chunking monitors embedding distance between adjacent sentences, inserting break points only when semantic variance exceeds a determined threshold:

from typing import List
import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

def semantic_chunk(text: str, similarity_threshold: float = 0.75) -> List[str]:
    sentences = [s.strip() for s in text.split(". ") if s.strip()]
    if not sentences:
        return []
    
    embeddings = model.encode(sentences)
    chunks, current_chunk = [], [sentences[0]]
    
    for i in range(1, len(sentences)):
        # Compute cosine similarity between consecutive sentences
        sim = np.dot(embeddings[i-1], embeddings[i]) / (
            np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i])
        )
        if sim < similarity_threshold:
            chunks.append(". ".join(current_chunk) + ".")
            current_chunk = [sentences[i]]
        else:
            current_chunk.append(sentences[i])
            
    if current_chunk:
        chunks.append(". ".join(current_chunk) + ".")
    return chunks

---

2. Hybrid Retrieval: Combining Dense and Sparse Vectors

While dense vector representations capture semantic intent and synonyms effectively, they frequently struggle with exact-match identifiers, acronyms, or specific product codes. Combining Dense Embeddings (e.g., OpenAI text-embedding-3, BGE) with Sparse Lexical Search (e.g., BM25) ensures both semantic breadth and keyword precision.

Reciprocal Rank Fusion (RRF) Formula

$RRF(d) = \sum_{m \in M} \frac{1}{k + r_m(d)}$

Where:

• $M$ is the set of ranking models (Dense + BM25).
• $r_m(d)$ is the rank of document $d$ in system $m$.
• $k$ is a smoothing constant (typically set to 60).

---

3. Two-Stage Retrieval with Cross-Encoder Rerankers

Bi-encoders generate independent vector representations for query and documents, making them fast for initial vector indexing but susceptible to false positives. Deploying a Cross-Encoder Reranker (such as bge-reranker-large or Cohere Rerank) as a second-stage pass jointly computes self-attention across the query-document pair, dramatically improving Top-3 precision.

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-large")

def rerank_results(query: str, retrieved_docs: List[str], top_k: int = 3):
    pairs = [[query, doc] for doc in retrieved_docs]
    scores = reranker.predict(pairs)
    
    # Sort docs descending by cross-encoder score
    ranked_docs = [doc for _, doc in sorted(zip(scores, retrieved_docs), reverse=True)]
    return ranked_docs[:top_k]

---

4. Hallucination Mitigation & Citation Grounding

To guarantee production safety:

• Strict Context Prompting: Explicitly instruct the model to respond with “I do not have sufficient information in the provided context” when facts are absent.
• Source Attributions: Enforce structured outputs where every factual claim is keyed to an indexed chunk identifier [Doc ID: X].
• Self-Correction & Critic Loops: Run secondary lightweight evaluation prompts or G-Eval pipelines to verify faithfulness before rendering responses to end users.

---

Conclusion

A resilient RAG system is far more than a simple vector store lookup. By structuring content with semantic chunking, executing hybrid searches, refining candidates through rerankers, and enforcing strict citation grounding, you can build production AI systems that deliver high accuracy and enterprise-grade reliability.