Production RAG Pipelines: Building Retrieval-Augmented Generation That Actually Works

I built my first RAG pipeline in early 2023, shortly after the GPT-4 API became available. It was for a legal document analysis system—a law firm wanted to query thousands of case files and get accurate, cited answers. The prototype worked beautifully in demos. In production, it was a disaster: hallucinated citations, irrelevant retrievals, answers that sounded confident but were wrong, and retrieval latency that made the system unusable for interactive queries.

That failure taught me more about RAG than any tutorial or paper. Over the next two years, I built and deployed RAG systems for legal analysis, insurance claims processing, technical documentation, and internal knowledge bases. Each deployment taught me something new about what makes RAG work in production—where “production” means accuracy, latency, cost, and reliability all matter simultaneously.

This is a practitioner’s guide to production RAG pipelines. It covers the full stack: document processing, chunking strategies, embedding models, vector stores, retrieval optimization, generation strategies, hallucination prevention, and evaluation. Every recommendation is grounded in real deployment experience.

The RAG Architecture

At its core, RAG has two phases:

1. Indexing (offline): Process documents, chunk them, generate embeddings, store in a vector database.
2. Querying (online): Embed the user query, retrieve relevant chunks, construct a prompt with retrieved context, generate an answer with an LLM.

Simple in concept. Complex in production. Let me walk through each component.

Document Processing and Chunking

The quality of your RAG pipeline is bounded by the quality of your document processing. Garbage in, garbage out—this is the most underestimated part of the pipeline.

Document Parsing

Documents come in many formats: PDF, Word, HTML, Markdown, scanned images, spreadsheets. Each requires different parsing strategies.

import fitz  # PyMuPDF
from docx import Document
from bs4 import BeautifulSoup

def parse_pdf(file_path: str) -> list[dict]:
    """Extract text with structure from PDF."""
    doc = fitz.open(file_path)
    pages = []
    for page_num, page in enumerate(doc):
        text = page.get_text("text")
        # Extract structure: headers, paragraphs, tables
        blocks = page.get_text("dict")["blocks"]
        structured_content = []
        for block in blocks:
            if block["type"] == 0:  # Text block
                for line in block["lines"]:
                    text = " ".join(span["text"] for span in line["spans"])
                    font_size = line["spans"][0]["size"]
                    is_bold = line["spans"][0]["flags"] & 2**4
                    structured_content.append({
                        "text": text,
                        "font_size": font_size,
                        "is_bold": bool(is_bold),
                        "page": page_num + 1,
                    })
        pages.append({
            "page": page_num + 1,
            "content": structured_content,
            "raw_text": page.get_text("text"),
        })
    return pages

def parse_html(file_path: str) -> str:
    """Extract clean text from HTML, preserving structure."""
    with open(file_path) as f:
        soup = BeautifulSoup(f.read(), "html.parser")

    # Remove scripts, styles, nav, footer
    for tag in soup(["script", "style", "nav", "footer", "header"]):
        tag.decompose()

    # Preserve headers as markers
    for h in soup.find_all(["h1", "h2", "h3", "h4"]):
        h.replace_with(f"\n\n## {h.get_text().strip()}\n\n")

    return soup.get_text(separator="\n", strip=True)

Chunking Strategies

Chunking is the most consequential decision in your RAG pipeline. Too small, and you lose context. Too large, and you dilute relevance and exceed context limits.

Strategy 1: Fixed-Size Chunking (Baseline)

def chunk_fixed_size(text: str, chunk_size: int = 512, overlap: int = 64) -> list[str]:
    """Simple fixed-size chunking with overlap."""
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if len(chunk.strip()) > 50:  # Minimum chunk size
            chunks.append(chunk)
    return chunks

Strategy 2: Semantic Chunking

Split based on topic changes using embedding similarity:

import numpy as np
from sentence_transformers import SentenceTransformer

def chunk_semantic(text: str, model: SentenceTransformer,
                   similarity_threshold: float = 0.75) -> list[str]:
    """Split text where topic changes significantly."""
    sentences = split_into_sentences(text)
    embeddings = model.encode(sentences)

    chunks = []
    current_chunk = [sentences[0]]
    current_embedding = embeddings[0]

    for i in range(1, len(sentences)):
        similarity = np.dot(current_embedding, embeddings[i]) / (
            np.linalg.norm(current_embedding) * np.linalg.norm(embeddings[i])
        )

        if similarity < similarity_threshold:
            # Topic changed, start new chunk
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentences[i]]
            current_embedding = embeddings[i]
        else:
            current_chunk.append(sentences[i])
            # Update chunk embedding (running average)
            current_embedding = 0.8 * current_embedding + 0.2 * embeddings[i]

    if current_chunk:
        chunks.append(" ".join(current_chunk))

    return chunks

Strategy 3: Structure-Aware Chunking (Best for Production)

Respect document structure—headers, sections, paragraphs:

def chunk_by_structure(pages: list[dict], max_chunk_size: int = 1000,
                       min_chunk_size: int = 100) -> list[dict]:
    """Chunk based on document structure (headers, sections)."""
    chunks = []
    current_chunk = {"text": "", "metadata": {}}

    for page in pages:
        for block in page["content"]:
            text = block["text"].strip()
            if not text:
                continue

            # New section on bold text or large font
            if block["is_bold"] or block["font_size"] > 14:
                if len(current_chunk["text"]) >= min_chunk_size:
                    chunks.append(current_chunk)
                current_chunk = {
                    "text": text,
                    "metadata": {
                        "page": block["page"],
                        "section_header": text,
                    },
                }
            else:
                current_chunk["text"] += " " + text

            # Force split if too large
            if len(current_chunk["text"]) > max_chunk_size:
                chunks.append(current_chunk)
                current_chunk = {"text": "", "metadata": {}}

    if current_chunk["text"].strip():
        chunks.append(current_chunk)

    return chunks

My Chunking Recommendation

After extensive experimentation, I use a hybrid approach:

1. Parse documents with structure awareness (preserve headers, sections, tables)
2. Chunk by logical sections (respect document structure)
3. Add overlap between chunks (preserve cross-section context)
4. Attach rich metadata (page number, section header, document title)
5. Store parent-child relationships (retrieve child chunks but include parent for context)

class ParentChildChunker:
    """Create overlapping chunks with parent context."""

    def __init__(self, parent_size=2000, child_size=400, overlap=100):
        self.parent_size = parent_size
        self.child_size = child_size
        self.overlap = overlap

    def chunk(self, text: str, metadata: dict) -> list[dict]:
        # Create parent chunks
        parent_chunks = chunk_fixed_size(text, self.parent_size, 0)

        all_chunks = []
        for parent_idx, parent in enumerate(parent_chunks):
            # Create child chunks within parent
            child_chunks = chunk_fixed_size(parent, self.child_size, self.overlap)

            for child_idx, child in enumerate(child_chunks):
                all_chunks.append({
                    "text": child,
                    "metadata": {
                        **metadata,
                        "parent_text": parent,
                        "parent_idx": parent_idx,
                        "child_idx": child_idx,
                    },
                })

        return all_chunks

Embedding Models

The embedding model is the heart of your retrieval system. Choosing the right one matters enormously.

Model Selection

For production RAG, I recommend:

1. text-embedding-3-large (OpenAI): Best quality, good for most use cases. 3072 dimensions.
2. bge-large-en-v1.5 (BAAI): Best open-source model. Excellent performance, runs locally.
3. cohere embed-v3: Good balance of quality and cost, with built-in search optimization.

Embedding Generation

from sentence_transformers import SentenceTransformer
import numpy as np

class EmbeddingService:
    def __init__(self, model_name: str = "BAAI/bge-large-en-v1.5"):
        self.model = SentenceTransformer(model_name)
        self.dimension = self.model.get_sentence_embedding_dimension()

    def embed(self, texts: list[str], batch_size: int = 32) -> np.ndarray:
        """Generate embeddings for a list of texts."""
        return self.model.encode(
            texts,
            batch_size=batch_size,
            show_progress_bar=True,
            normalize_embeddings=True,  # For cosine similarity
        )

    def embed_query(self, query: str) -> np.ndarray:
        """Embed a single query with instruction prefix."""
        # BGE models perform better with instruction prefix
        query_with_instruction = f"Represent this sentence for searching relevant passages: {query}"
        return self.model.encode(
            [query_with_instruction],
            normalize_embeddings=True,
        )[0]

Embedding Optimization Tips

1. Normalize embeddings: Always normalize to unit vectors. This enables fast cosine similarity via dot product.
2. Use instruction prefixes: Many embedding models (BGE, E5) perform better when queries are prefixed with an instruction.
3. Batch processing: Generate embeddings in batches for 3-5x speedup over individual calls.
4. Quantize for storage: Use float16 or int8 quantization to reduce storage and improve retrieval speed with minimal quality loss.

Vector Stores

Choosing the right vector store depends on your scale, latency requirements, and operational preferences.

Chroma (Development/Small Scale)

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"},
)

# Add documents
collection.add(
    ids=[f"chunk_{i}" for i in range(len(chunks))],
    documents=[c["text"] for c in chunks],
    embeddings=embeddings.tolist(),
    metadatas=[c["metadata"] for c in chunks],
)

# Query
results = collection.query(
    query_embeddings=[query_embedding.tolist()],
    n_results=10,
    where={"document_type": "policy"},
)

Pinecone (Managed, Production)

import pinecone

pinecone.init(api_key="YOUR_API_KEY", environment="us-east1-gcp")
index = pinecone.Index("rag-documents")

# Upsert in batches
def upsert_batch(index, chunks, embeddings, batch_size=100):
    for i in range(0, len(chunks), batch_size):
        batch_ids = [f"chunk_{j}" for j in range(i, min(i + batch_size, len(chunks)))]
        batch_embeddings = embeddings[i:i + batch_size].tolist()
        batch_metadata = [c["metadata"] for c in chunks[i:i + batch_size]]
        index.upsert(zip(batch_ids, batch_embeddings, batch_metadata))

# Query with metadata filter
results = index.query(
    vector=query_embedding.tolist(),
    top_k=10,
    filter={"document_type": {"$eq": "policy"}},
    include_metadata=True,
)

Qdrant (Self-Hosted, Production)

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(host="localhost", port=6333)

# Create collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1024,
        distance=Distance.COSINE,
    ),
)

# Upsert
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=i,
            vector=embeddings[i].tolist(),
            payload=chunks[i]["metadata"],
        )
        for i in range(len(chunks))
    ],
)

# Query
results = client.search(
    collection_name="documents",
    query_vector=query_embedding.tolist(),
    limit=10,
    query_filter={
        "must": [
            {"key": "document_type", "match": {"value": "policy"}},
        ]
    },
)

Retrieval Optimization

Raw vector similarity search is rarely sufficient for production RAG. Several optimization techniques can dramatically improve retrieval quality.

Hybrid Search: Combining Semantic and Keyword Search

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, chunks, embeddings, embedding_model):
        self.chunks = chunks
        self.embeddings = embeddings
        self.embedding_model = embedding_model

        # BM25 index for keyword search
        tokenized_chunks = [chunk["text"].lower().split() for chunk in chunks]
        self.bm25 = BM25Okapi(tokenized_chunks)

    def retrieve(self, query: str, top_k: int = 10,
                 alpha: float = 0.7) -> list[dict]:
        """Hybrid retrieval combining semantic and keyword search."""
        # Semantic search
        query_embedding = self.embedding_model.embed_query(query)
        semantic_scores = np.dot(self.embeddings, query_embedding)

        # Keyword search
        tokenized_query = query.lower().split()
        bm25_scores = self.bm25.get_scores(tokenized_query)

        # Normalize scores to [0, 1]
        semantic_scores = (semantic_scores - semantic_scores.min()) / (
            semantic_scores.max() - semantic_scores.min() + 1e-8
        )
        bm25_scores = (bm25_scores - bm25_scores.min()) / (
            bm25_scores.max() - bm25_scores.min() + 1e-8
        )

        # Combine scores
        combined_scores = alpha * semantic_scores + (1 - alpha) * bm25_scores

        # Get top-k indices
        top_indices = np.argsort(combined_scores)[::-1][:top_k]

        results = []
        for idx in top_indices:
            results.append({
                **self.chunks[idx],
                "score": float(combined_scores[idx]),
            })

        return results

Re-ranking

After initial retrieval, re-rank results using a cross-encoder for higher precision:

from sentence_transformers import CrossEncoder

class ReRanker:
    def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"):
        self.model = CrossEncoder(model_name)

    def rerank(self, query: str, candidates: list[dict],
               top_k: int = 5) -> list[dict]:
        """Re-rank candidates using cross-encoder."""
        pairs = [(query, c["text"]) for c in candidates]
        scores = self.model.predict(pairs)

        # Sort by cross-encoder score
        scored_candidates = list(zip(candidates, scores))
        scored_candidates.sort(key=lambda x: x[1], reverse=True)

        return [
            {**c, "rerank_score": float(s)}
            for c, s in scored_candidates[:top_k]
        ]

Query Expansion

Expand the user query to improve recall:

def expand_query(query: str, llm_client) -> list[str]:
    """Generate query variations for better recall."""
    prompt = f"""Generate 3 different search queries that would help find
    information to answer this question. Each query should approach the
    topic from a different angle.

    Question: {query}

    Queries (one per line):"""

    response = llm_client.generate(prompt)
    expanded = [q.strip() for q in response.strip().split("\n") if q.strip()]
    return [query] + expanded

Metadata Filtering

Use metadata to narrow search scope before vector search:

def retrieve_with_filters(self, query: str, filters: dict,
                          top_k: int = 10) -> list[dict]:
    """Retrieve with metadata filters."""
    # First apply metadata filters
    filtered_chunks = [
        (i, chunk) for i, chunk in enumerate(self.chunks)
        if all(chunk["metadata"].get(k) == v for k, v in filters.items())
    ]

    if not filtered_chunks:
        return []

    # Then do vector search on filtered subset
    filtered_indices = [i for i, _ in filtered_chunks]
    filtered_embeddings = self.embeddings[filtered_indices]

    query_embedding = self.embedding_model.embed_query(query)
    scores = np.dot(filtered_embeddings, query_embedding)

    top_indices = np.argsort(scores)[::-1][:top_k]

    return [
        {**filtered_chunks[i][1], "score": float(scores[idx])}
        for idx, i in enumerate(top_indices)
    ]

Generation Strategies

The generation phase is where most RAG systems fail. The LLM must synthesize retrieved context into accurate, cited answers.

Prompt Engineering for RAG

RAG_SYSTEM_PROMPT = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Only answer based on the provided context. If the context doesn't contain
   enough information, say "I don't have enough information to answer this question."
2. Always cite your sources using [Source: document_name, page X] format.
3. If multiple sources provide conflicting information, note the conflict.
4. Do not make assumptions or add information not present in the context.
5. If the question is ambiguous, ask for clarification.

Context:
{context}

Question: {question}

Answer:"""

def build_context(retrieved_chunks: list[dict]) -> str:
    """Build context string from retrieved chunks."""
    context_parts = []
    for i, chunk in enumerate(retrieved_chunks):
        source = chunk["metadata"].get("source", "unknown")
        page = chunk["metadata"].get("page", "unknown")
        context_parts.append(
            f"[{i+1}] (Source: {source}, Page {page})\n{chunk['text']}"
        )
    return "\n\n".join(context_parts)

Multi-Turn Conversation

For conversational RAG, maintain context across turns:

class ConversationalRAG:
    def __init__(self, retriever, llm_client, max_history=5):
        self.retriever = retriever
        self.llm_client = llm_client
        self.history = []
        self.max_history = max_history

    def query(self, question: str) -> dict:
        # Reformulate question with conversation history
        reformulated = self._reformulate(question)

        # Retrieve relevant chunks
        chunks = self.retriever.retrieve(reformulated)

        # Build prompt with history
        context = build_context(chunks)
        history_text = "\n".join(
            f"User: {h['question']}\nAssistant: {h['answer']}"
            for h in self.history[-self.max_history:]
        )

        prompt = f"""Previous conversation:
{history_text}

Context:
{context}

Current question: {question}

Answer based on the context. If the context doesn't contain the answer, say so."""

        answer = self.llm_client.generate(prompt)

        # Update history
        self.history.append({"question": question, "answer": answer})

        return {"answer": answer, "sources": chunks}

    def _reformulate(self, question: str) -> str:
        """Reformulate question to be standalone."""
        if not self.history:
            return question

        prompt = f"""Given the conversation history, reformulate the follow-up
question to be a standalone question.

History:
{chr(10).join(f'User: {h["question"]}' for h in self.history[-3:])}

Follow-up: {question}

Standalone question:"""

        return self.llm_client.generate(prompt)

Hallucination Prevention

The biggest risk in RAG is hallucination—the LLM generating plausible but incorrect information. Here are the strategies that work:

1. Citation Enforcement

Force the model to cite specific sources. If it cannot cite, it should say it does not know.

def validate_citations(answer: str, sources: list[dict]) -> dict:
    """Validate that citations in the answer match retrieved sources."""
    import re

    # Extract citations from answer
    citations = re.findall(r'\[Source: ([^\]]+)\]', answer)

    # Check each citation exists in sources
    source_names = {
        f"{s['metadata'].get('source', 'unknown')}, page {s['metadata'].get('page', '?')}"
        for s in sources
    }

    valid_citations = [c for c in citations if any(c in s for s in source_names)]
    invalid_citations = [c for c in citations if c not in valid_citations]

    return {
        "valid_citations": valid_citations,
        "invalid_citations": invalid_citations,
        "has_hallucinated_citations": len(invalid_citations) > 0,
    }

2. Faithfulness Scoring

Use an LLM to evaluate whether the answer is faithful to the retrieved context:

def score_faithfulness(answer: str, context: str, llm_client) -> float:
    """Score how faithful the answer is to the context."""
    prompt = f"""Evaluate whether the following answer is supported by the context.
Score from 0 (completely unsupported) to 1 (fully supported).

Context:
{context}

Answer: {answer}

Score (0-1) and explanation:"""

    response = llm_client.generate(prompt)
    # Parse score from response
    try:
        score = float(response.split()[0])
        return min(max(score, 0), 1)
    except ValueError:
        return 0.5  # Default if parsing fails

3. Self-Consistency Checking

Generate multiple answers and check for consistency:

def self_consistency_check(question: str, context: str,
                           llm_client, n_samples: int = 3) -> dict:
    """Generate multiple answers and check consistency."""
    answers = []
    for _ in range(n_samples):
        answer = llm_client.generate(
            f"Context: {context}\n\nQuestion: {question}\n\nAnswer:",
            temperature=0.7,
        )
        answers.append(answer)

    # Check consistency (simple approach: compare key claims)
    prompt = f"""Are these answers consistent? List any contradictions.

Answers:
{chr(10).join(f'{i+1}. {a}' for i, a in enumerate(answers))}

Contradictions (or "None"):"""

    contradictions = llm_client.generate(prompt, temperature=0)

    return {
        "answers": answers,
        "contradictions": contradictions,
        "is_consistent": "none" in contradictions.lower(),
    }

Evaluation

RAG evaluation requires assessing both retrieval quality and generation quality.

Retrieval Metrics

def evaluate_retrieval(queries: list[dict], retriever, k: int = 10) -> dict:
    """Evaluate retrieval quality."""
    recall_at_k = []
    precision_at_k = []
    mrr = []

    for query_data in queries:
        query = query_data["question"]
        relevant_ids = set(query_data["relevant_chunk_ids"])

        results = retriever.retrieve(query, top_k=k)
        retrieved_ids = [r["id"] for r in results]

        # Recall@K
        hits = len(relevant_ids & set(retrieved_ids))
        recall_at_k.append(hits / len(relevant_ids) if relevant_ids else 0)

        # Precision@K
        precision_at_k.append(hits / k)

        # MRR (Mean Reciprocal Rank)
        for rank, rid in enumerate(retrieved_ids, 1):
            if rid in relevant_ids:
                mrr.append(1.0 / rank)
                break
        else:
            mrr.append(0.0)

    return {
        f"recall@{k}": np.mean(recall_at_k),
        f"precision@{k}": np.mean(precision_at_k),
        "mrr": np.mean(mrr),
    }

End-to-End Metrics

def evaluate_rag_pipeline(test_cases: list[dict], pipeline) -> dict:
    """Evaluate full RAG pipeline."""
    results = {
        "faithfulness": [],
        "relevance": [],
        "correctness": [],
    }

    for case in test_cases:
        output = pipeline.query(case["question"])

        # Faithfulness: Is the answer supported by context?
        faith = score_faithfulness(
            output["answer"],
            build_context(output["sources"]),
            pipeline.llm_client,
        )
        results["faithfulness"].append(faith)

        # Relevance: Does the answer address the question?
        relevance = score_relevance(
            case["question"], output["answer"], pipeline.llm_client
        )
        results["relevance"].append(relevance)

        # Correctness: Is the answer factually correct?
        if "expected_answer" in case:
            correctness = score_correctness(
                output["answer"], case["expected_answer"], pipeline.llm_client
            )
            results["correctness"].append(correctness)

    return {k: np.mean(v) for k, v in results.items() if v}

Production Architecture

A production RAG system needs more than the core pipeline:

User Query
    ↓
Query Router (classify intent, select pipeline)
    ↓
Query Rewriter (expand, decompose, clarify)
    ↓
Hybrid Retriever (vector + keyword + metadata)
    ↓
Re-ranker (cross-encoder)
    ↓
Context Builder (deduplicate, truncate, format)
    ↓
LLM Generator (with citation enforcement)
    ↓
Faithfulness Checker
    ↓
Response Formatter
    ↓
User Response

Each component should be independently deployable, monitorable, and replaceable.

Lessons Learned

1. Chunking is the highest-leverage optimization: Getting chunking right improves RAG quality more than switching embedding models or LLMs.
2. Hybrid search is non-negotiable: Pure vector search misses keyword matches. Always combine semantic and keyword search.
3. Re-ranking is cheap and effective: A cross-encoder re-ranker improves precision by 10-20% with minimal latency impact.
4. Hallucination prevention requires multiple layers: No single technique prevents all hallucination. Use citations, faithfulness scoring, and self-consistency checking together.
5. Evaluation must be continuous: RAG quality degrades as your document corpus changes. Set up automated evaluation pipelines that run regularly.
6. Start simple, optimize incrementally: A basic RAG pipeline with good chunking and a strong embedding model beats a complex pipeline with poor fundamentals.

Conclusion

Building a production RAG pipeline is an engineering challenge, not just a modeling challenge. The quality of your document processing, chunking strategy, retrieval optimization, and generation strategy all matter—and they interact in complex ways.

Start with the fundamentals: clean document parsing, sensible chunking, a strong embedding model, and a well-engineered prompt. Then layer on optimization: hybrid search, re-ranking, query expansion, and hallucination prevention. Evaluate continuously, and iterate based on real user queries and failure modes.

The RAG systems that work in production are not the most clever ones—they are the most well-engineered ones.