RAG in Production: Build a Reliable Pipeline

Building a RAG prototype takes an afternoon. Making it production-ready takes weeks of engineering. The gap between demo and production is filled with chunking edge cases, retrieval failures, hallucination monitoring, and ingestion pipelines that break on the 47th PDF. This guide covers the engineering decisions that turn a prototype into a reliable system.

Production RAG Architecture

A production RAG system has four stages: ingestion, retrieval, augmentation, and generation. Each stage needs error handling, monitoring, and the ability to be tested independently.

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import logging
import time

logger = logging.getLogger("rag_pipeline")

@dataclass
class Document:
    """A source document before chunking."""
    id: str
    content: str
    metadata: dict = field(default_factory=dict)
    source: str = ""

@dataclass
class Chunk:
    """A chunk of a document ready for embedding."""
    id: str
    content: str
    document_id: str
    metadata: dict = field(default_factory=dict)
    embedding: Optional[list[float]] = None

@dataclass
class RetrievalResult:
    """A retrieved chunk with its relevance score."""
    chunk: Chunk
    score: float
    retrieval_method: str = "vector"

@dataclass
class RAGResponse:
    """The final RAG response with provenance."""
    answer: str
    sources: list[RetrievalResult]
    model: str
    latency_ms: float
    token_usage: dict = field(default_factory=dict)

Stage 1: Document Ingestion

The ingestion pipeline converts raw files into chunks stored in your vector database. This is where most production issues originate.

import hashlib
from pathlib import Path
from typing import Generator

class DocumentLoader:
    """Load documents from various sources with error handling."""
    
    def __init__(self):
        self.supported_formats = {".txt", ".md", ".pdf", ".html", ".csv", ".json"}
    
    def load_file(self, file_path: str) -> Document:
        """Load a single file into a Document."""
        path = Path(file_path)
        
        if path.suffix not in self.supported_formats:
            raise ValueError(f"Unsupported format: {path.suffix}")
        
        if path.suffix == ".pdf":
            content = self._load_pdf(path)
        elif path.suffix == ".html":
            content = self._load_html(path)
        else:
            content = path.read_text(encoding="utf-8", errors="replace")
        
        # Generate deterministic ID from content
        doc_id = hashlib.sha256(content.encode()).hexdigest()[:16]
        
        return Document(
            id=doc_id,
            content=content,
            metadata={
                "filename": path.name,
                "file_type": path.suffix,
                "file_size": path.stat().st_size,
            },
            source=str(path)
        )
    
    def _load_pdf(self, path: Path) -> str:
        """Extract text from PDF with fallback."""
        try:
            import pymupdf
            doc = pymupdf.open(str(path))
            text = ""
            for page in doc:
                text += page.get_text() + "\n"
            doc.close()
            return text
        except Exception as e:
            logger.warning(f"pymupdf failed for {path}, trying pdfplumber: {e}")
            try:
                import pdfplumber
                with pdfplumber.open(str(path)) as pdf:
                    return "\n".join(page.extract_text() or "" for page in pdf.pages)
            except Exception as e2:
                logger.error(f"All PDF parsers failed for {path}: {e2}")
                raise
    
    def _load_html(self, path: Path) -> str:
        """Extract text from HTML."""
        from bs4 import BeautifulSoup
        html = path.read_text(encoding="utf-8", errors="replace")
        soup = BeautifulSoup(html, "html.parser")
        for tag in soup(["script", "style", "nav", "footer", "header"]):
            tag.decompose()
        return soup.get_text(separator="\n", strip=True)
    
    def load_directory(self, dir_path: str) -> Generator[Document, None, None]:
        """Load all supported files from a directory."""
        path = Path(dir_path)
        for file_path in sorted(path.rglob("*")):
            if file_path.suffix in self.supported_formats and file_path.is_file():
                try:
                    yield self.load_file(str(file_path))
                except Exception as e:
                    logger.error(f"Failed to load {file_path}: {e}")
                    continue

Stage 2: Chunking

Chunking strategy has the single biggest impact on retrieval quality. Too small and you lose context. Too large and you dilute relevance.

import re
import tiktoken

class ProductionChunker:
    """Chunk documents with multiple strategies."""
    
    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50, model: str = "gpt-4o"):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoding = tiktoken.encoding_for_model(model)
    
    def chunk_document(self, document: Document) -> list[Chunk]:
        """Chunk a document using the best strategy for its type."""
        file_type = document.metadata.get("file_type", ".txt")
        
        if file_type in (".md", ".html"):
            return self._chunk_by_headers(document)
        else:
            return self._chunk_by_paragraphs(document)
    
    def _chunk_by_paragraphs(self, document: Document) -> list[Chunk]:
        """Split on paragraph boundaries, respecting token limits."""
        paragraphs = re.split(r'\n\s*\n', document.content)
        paragraphs = [p.strip() for p in paragraphs if p.strip()]
        
        chunks = []
        current_paragraphs = []
        current_tokens = 0
        
        for para in paragraphs:
            para_tokens = len(self.encoding.encode(para))
            
            # If a single paragraph exceeds chunk size, split it by sentences
            if para_tokens > self.chunk_size:
                if current_paragraphs:
                    chunks.append(self._create_chunk(
                        document, "\n\n".join(current_paragraphs), len(chunks)
                    ))
                    current_paragraphs = []
                    current_tokens = 0
                
                sentence_chunks = self._split_long_paragraph(para)
                for sc in sentence_chunks:
                    chunks.append(self._create_chunk(document, sc, len(chunks)))
                continue
            
            if current_tokens + para_tokens > self.chunk_size and current_paragraphs:
                chunks.append(self._create_chunk(
                    document, "\n\n".join(current_paragraphs), len(chunks)
                ))
                
                # Keep last paragraph for overlap
                if self.chunk_overlap > 0 and current_paragraphs:
                    overlap_para = current_paragraphs[-1]
                    current_paragraphs = [overlap_para]
                    current_tokens = len(self.encoding.encode(overlap_para))
                else:
                    current_paragraphs = []
                    current_tokens = 0
            
            current_paragraphs.append(para)
            current_tokens += para_tokens
        
        if current_paragraphs:
            chunks.append(self._create_chunk(
                document, "\n\n".join(current_paragraphs), len(chunks)
            ))
        
        return chunks
    
    def _chunk_by_headers(self, document: Document) -> list[Chunk]:
        """Split markdown/HTML by headers for topical coherence."""
        sections = re.split(r'(^#{1,3}\s+.+$)', document.content, flags=re.MULTILINE)
        
        chunks = []
        current_header = ""
        current_body = []
        
        for section in sections:
            if re.match(r'^#{1,3}\s+', section):
                if current_body:
                    body_text = "\n".join(current_body).strip()
                    if body_text:
                        full_text = f"{current_header}\n{body_text}" if current_header else body_text
                        token_count = len(self.encoding.encode(full_text))
                        
                        if token_count > self.chunk_size:
                            sub_chunks = self._split_long_paragraph(full_text)
                            for sc in sub_chunks:
                                chunks.append(self._create_chunk(document, sc, len(chunks)))
                        else:
                            chunks.append(self._create_chunk(document, full_text, len(chunks)))
                
                current_header = section.strip()
                current_body = []
            else:
                current_body.append(section)
        
        if current_body:
            body_text = "\n".join(current_body).strip()
            if body_text:
                full_text = f"{current_header}\n{body_text}" if current_header else body_text
                chunks.append(self._create_chunk(document, full_text, len(chunks)))
        
        return chunks
    
    def _split_long_paragraph(self, text: str) -> list[str]:
        """Split long text by sentences when it exceeds chunk size."""
        sentences = re.split(r'(?<=[.!?])\s+', text)
        
        result_chunks = []
        current = []
        current_tokens = 0
        
        for sentence in sentences:
            sent_tokens = len(self.encoding.encode(sentence))
            if current_tokens + sent_tokens > self.chunk_size and current:
                result_chunks.append(" ".join(current))
                current = []
                current_tokens = 0
            current.append(sentence)
            current_tokens += sent_tokens
        
        if current:
            result_chunks.append(" ".join(current))
        
        return result_chunks
    
    def _create_chunk(self, document: Document, content: str, index: int) -> Chunk:
        """Create a Chunk with proper ID and metadata."""
        chunk_id = f"{document.id}_chunk_{index}"
        return Chunk(
            id=chunk_id,
            content=content,
            document_id=document.id,
            metadata={
                **document.metadata,
                "chunk_index": index,
                "token_count": len(self.encoding.encode(content)),
            }
        )

Stage 3: Retrieval

Production retrieval goes beyond simple vector search. Hybrid search, reranking, and query transformation significantly improve result quality.

from openai import OpenAI
import numpy as np

client = OpenAI()

class ProductionRetriever:
    """Retriever with hybrid search and reranking."""
    
    def __init__(self, vector_store, reranker=None):
        self.vector_store = vector_store
        self.reranker = reranker
    
    def retrieve(
        self,
        query: str,
        top_k: int = 5,
        use_hyde: bool = False,
        filters: dict = None
    ) -> list[RetrievalResult]:
        """Retrieve relevant chunks with optional HyDE and reranking."""
        
        search_query = query
        
        # HyDE: Generate a hypothetical answer and search with that instead
        if use_hyde:
            search_query = self._generate_hypothetical_answer(query)
            logger.info(f"HyDE query: {search_query[:100]}...")
        
        # Retrieve more candidates than needed for reranking
        fetch_k = top_k * 3 if self.reranker else top_k
        
        # Vector search
        results = self.vector_store.search(
            query=search_query,
            top_k=fetch_k,
            filters=filters
        )
        
        retrieval_results = [
            RetrievalResult(
                chunk=Chunk(id=r["id"], content=r["text"], document_id=r.get("document_id", "")),
                score=r["score"],
                retrieval_method="vector"
            )
            for r in results
        ]
        
        # Rerank if available
        if self.reranker and retrieval_results:
            retrieval_results = self._rerank(query, retrieval_results, top_k)
        else:
            retrieval_results = retrieval_results[:top_k]
        
        return retrieval_results
    
    def _generate_hypothetical_answer(self, query: str) -> str:
        """HyDE: Generate a hypothetical document that would answer the query."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": "Write a short paragraph that would be the ideal answer to the following question. Be factual and specific."
                },
                {"role": "user", "content": query}
            ],
            temperature=0.3,
            max_tokens=200
        )
        return response.choices[0].message.content
    
    def _rerank(
        self,
        query: str,
        results: list[RetrievalResult],
        top_k: int
    ) -> list[RetrievalResult]:
        """Rerank results using a cross-encoder or LLM."""
        # Using an LLM-based reranker (simpler alternative to cross-encoder)
        documents = [r.chunk.content for r in results]
        
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Given a query and a list of documents, rank the documents by relevance. "
                        "Return a JSON list of indices ordered from most to least relevant."
                    )
                },
                {
                    "role": "user",
                    "content": f"Query: {query}\n\nDocuments:\n" + "\n".join(
                        f"[{i}] {doc[:200]}" for i, doc in enumerate(documents)
                    )
                }
            ],
            response_format={"type": "json_object"},
            temperature=0
        )
        
        import json
        ranking = json.loads(response.choices[0].message.content)
        indices = ranking.get("indices", list(range(len(results))))
        
        reranked = []
        for rank, idx in enumerate(indices[:top_k]):
            if idx < len(results):
                result = results[idx]
                result.score = 1.0 - (rank / len(indices))  # Normalize score
                result.retrieval_method = "reranked"
                reranked.append(result)
        
        return reranked

Stage 4: Generation

The generation stage combines retrieved context with the user query and sends it to the LLM. The prompt template and context formatting matter more than you might expect.

class RAGGenerator:
    """Generate answers with source attribution and confidence."""
    
    def __init__(self, model: str = "gpt-4o"):
        self.model = model
        self.system_prompt = """You are a helpful assistant that answers questions based on the provided context.

Rules:
1. Only use information from the provided context to answer
2. If the context doesn't contain enough information, say so clearly
3. Cite your sources using [Source N] notation
4. Be concise but thorough"""
    
    def generate(
        self,
        query: str,
        retrieval_results: list[RetrievalResult]
    ) -> RAGResponse:
        """Generate a response using retrieved context."""
        start_time = time.time()
        
        # Format context with source numbers
        context_parts = []
        for i, result in enumerate(retrieval_results, 1):
            source_info = result.chunk.metadata.get("filename", "Unknown source")
            context_parts.append(
                f"[Source {i}] (from {source_info}, relevance: {result.score:.2f}):\n"
                f"{result.chunk.content}"
            )
        
        context = "\n\n---\n\n".join(context_parts)
        
        user_message = f"""Context:
{context}

Question: {query}

Provide a thorough answer based on the context above. Cite sources using [Source N] notation."""
        
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_message}
            ],
            temperature=0.3
        )
        
        latency = (time.time() - start_time) * 1000
        
        return RAGResponse(
            answer=response.choices[0].message.content,
            sources=retrieval_results,
            model=self.model,
            latency_ms=latency,
            token_usage={
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            }
        )

Monitoring and Observability

You cannot improve what you do not measure. Track these metrics for every RAG query:

import json
from datetime import datetime

class RAGMonitor:
    """Monitor RAG pipeline performance and quality."""
    
    def __init__(self, log_file: str = "rag_metrics.jsonl"):
        self.log_file = log_file
    
    def log_query(self, query: str, response: RAGResponse, retrieval_results: list[RetrievalResult]):
        """Log metrics for a single RAG query."""
        metrics = {
            "timestamp": datetime.utcnow().isoformat(),
            "query": query,
            "answer_length": len(response.answer),
            "num_sources": len(response.sources),
            "latency_ms": response.latency_ms,
            "model": response.model,
            "token_usage": response.token_usage,
            "retrieval_scores": [r.score for r in retrieval_results],
            "avg_retrieval_score": (
                sum(r.score for r in retrieval_results) / len(retrieval_results)
                if retrieval_results else 0
            ),
            "top_score": retrieval_results[0].score if retrieval_results else 0,
            "score_gap": (
                retrieval_results[0].score - retrieval_results[-1].score
                if len(retrieval_results) > 1 else 0
            ),
        }
        
        # Low retrieval score warning
        if metrics["top_score"] < 0.5:
            metrics["warning"] = "low_retrieval_relevance"
            logger.warning(f"Low retrieval score for query: {query[:100]}")
        
        with open(self.log_file, "a") as f:
            f.write(json.dumps(metrics) + "\n")
        
        return metrics
    
    def get_summary(self, last_n: int = 100) -> dict:
        """Get summary statistics from recent queries."""
        records = []
        with open(self.log_file, "r") as f:
            for line in f:
                records.append(json.loads(line))
        
        recent = records[-last_n:]
        
        if not recent:
            return {"message": "No data available"}
        
        latencies = [r["latency_ms"] for r in recent]
        scores = [r["top_score"] for r in recent]
        tokens = [r["token_usage"]["total_tokens"] for r in recent]
        
        return {
            "total_queries": len(recent),
            "avg_latency_ms": sum(latencies) / len(latencies),
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
            "avg_top_retrieval_score": sum(scores) / len(scores),
            "low_relevance_rate": sum(1 for s in scores if s < 0.5) / len(scores),
            "avg_tokens_per_query": sum(tokens) / len(tokens),
            "warnings": sum(1 for r in recent if "warning" in r),
        }

Putting the Pipeline Together

Here is the complete pipeline wired together:

class ProductionRAGPipeline:
    """Complete production RAG pipeline."""
    
    def __init__(self, vector_store, model: str = "gpt-4o"):
        self.loader = DocumentLoader()
        self.chunker = ProductionChunker(chunk_size=512, chunk_overlap=50)
        self.retriever = ProductionRetriever(vector_store, reranker=True)
        self.generator = RAGGenerator(model=model)
        self.monitor = RAGMonitor()
    
    def ingest(self, source_path: str) -> int:
        """Ingest documents from a file or directory."""
        path = Path(source_path)
        documents = []
        
        if path.is_file():
            documents = [self.loader.load_file(str(path))]
        elif path.is_dir():
            documents = list(self.loader.load_directory(str(path)))
        
        total_chunks = 0
        for doc in documents:
            chunks = self.chunker.chunk_document(doc)
            
            # Generate embeddings and store
            for chunk in chunks:
                embedding = self._get_embedding(chunk.content)
                self.retriever.vector_store.add_documents(
                    documents=[chunk.content],
                    ids=[chunk.id],
                    metadatas=[chunk.metadata]
                )
            
            total_chunks += len(chunks)
            logger.info(f"Ingested {doc.source}: {len(chunks)} chunks")
        
        return total_chunks
    
    def query(self, question: str, top_k: int = 5) -> RAGResponse:
        """Run the full RAG pipeline for a question."""
        # Retrieve
        results = self.retriever.retrieve(question, top_k=top_k)
        
        if not results:
            return RAGResponse(
                answer="I couldn't find any relevant information to answer your question.",
                sources=[],
                model=self.generator.model,
                latency_ms=0
            )
        
        # Generate
        response = self.generator.generate(question, results)
        
        # Monitor
        self.monitor.log_query(question, response, results)
        
        return response
    
    def _get_embedding(self, text: str) -> list[float]:
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

Common Production Issues and Fixes

Problem: Ingestion fails on certain PDFs. PDFs with scanned images, complex layouts, or encryption will break text extraction. Use fallback parsers and log failures for manual review.

Problem: Retrieval returns irrelevant chunks. This usually means your chunks are too large (diluted relevance) or too small (missing context). Experiment with chunk sizes between 256 and 1024 tokens. Also check that your query and document embeddings use the same model.

Problem: The LLM ignores retrieved context. Your system prompt may not be strong enough. Add explicit instructions like “ONLY use information from the provided context” and “If the context does not contain the answer, say ‘I don’t have enough information’.”

Problem: Latency is too high. Profile each stage independently. Common bottlenecks: embedding generation (batch your requests), vector search (check index configuration), and LLM generation (use streaming for perceived speed).

Wrapping Up

A production RAG pipeline is an engineering system, not just a prompt wrapper. Each stage, ingestion, chunking, retrieval, and generation, needs independent testing, error handling, and monitoring. Start with the simplest version that works: load documents, chunk by paragraphs, embed with OpenAI, store in Chroma, and generate with a clear prompt. Then improve iteratively: add reranking when retrieval quality is insufficient, add hybrid search when keyword matching would help, add HyDE when users ask abstract questions. Monitor your retrieval scores and latencies continuously, because a RAG system that silently returns irrelevant context is worse than one that says “I don’t know.”