RAG Pipeline Architecture: From Prototype to Production

Retrieval Augmented Generation (RAG) is the most practical way to ground LLM responses in your organization's actual data. But the gap between a RAG prototype and a production system is enormous. Most teams get a demo working in a day, then spend months trying to make it reliable.

This guide covers what we've learned deploying RAG systems across healthcare, finance, and enterprise knowledge management.

Why RAG Matters

LLMs are trained on general knowledge with a fixed cutoff date. RAG solves four critical limitations:

1. Grounding: Responses are based on your actual data, not the model's training data
2. Freshness: New documents are immediately available without retraining
3. Attribution: Every answer can cite its source documents
4. Cost: Far cheaper than fine-tuning for most knowledge-retrieval use cases

The RAG Pipeline Architecture

A production RAG system has five core stages:

Document Ingestion → Chunking → Embedding → Retrieval → Generation

Each stage has critical decisions that compound into overall system quality.

Stage 1: Document Ingestion

Production document pipelines must handle diverse formats reliably:

class DocumentIngestionPipeline:
    def __init__(self):
        self.parsers = {
            ".pdf": PDFParser(ocr_enabled=True),
            ".docx": DocxParser(),
            ".html": HTMLParser(extract_tables=True),
            ".md": MarkdownParser(),
            ".csv": CSVParser(),
            ".xlsx": ExcelParser(),
        }

    async def ingest(self, source: DataSource) -> list[Document]:
        documents = []
        for file in source.list_files():
            parser = self.parsers.get(file.extension)
            if parser:
                doc = await parser.parse(file)
                doc.metadata = self.extract_metadata(file, doc)
                documents.append(doc)
        return documents

Key Ingestion Considerations

• OCR for scanned PDFs: Many enterprise documents are scanned images, not text
• Table extraction: Tables contain critical structured data that naive parsers miss
• Metadata preservation: Document dates, authors, and categories are essential for filtering
• Incremental updates: Only re-process documents that have changed
• Deduplication: Detect and handle duplicate or near-duplicate documents

Stage 2: Chunking Strategy

Chunking is where most RAG systems fail. The wrong chunking strategy leads to irrelevant retrieval and hallucinated answers.

Semantic Chunking

Instead of splitting by fixed character count, split by semantic boundaries:

class SemanticChunker:
    def __init__(self, embedding_model, similarity_threshold=0.75):
        self.embedding_model = embedding_model
        self.threshold = similarity_threshold

    def chunk(self, document: Document) -> list[Chunk]:
        sentences = self.split_sentences(document.text)
        embeddings = self.embedding_model.encode(sentences)

        chunks = []
        current_chunk = [sentences[0]]

        for i in range(1, len(sentences)):
            similarity = cosine_similarity(embeddings[i-1], embeddings[i])
            if similarity < self.threshold:
                # Semantic break detected - start new chunk
                chunks.append(Chunk(
                    text="\n".join(current_chunk),
                    metadata=document.metadata
                ))
                current_chunk = [sentences[i]]
            else:
                current_chunk.append(sentences[i])

        return chunks

Chunking Best Practices

• Chunk size: 256-512 tokens works best for most use cases. Too small loses context, too large dilutes relevance
• Overlap: 10-20% overlap between chunks prevents information loss at boundaries
• Preserve structure: Keep headers, lists, and tables intact within chunks
• Parent-child chunks: Store small chunks for retrieval but return the parent section for context
• Document-aware splitting: Never split mid-sentence or mid-paragraph

Stage 3: Embedding Models

The embedding model determines how well your system understands semantic similarity.

Model Selection

Embedding Best Practices

• Instruction-prefixed embeddings: Use query prefixes like "Represent this question for retrieving relevant passages:" to improve retrieval quality
• Batch processing: Embed documents in batches, not one at a time
• Dimension reduction: For cost optimization, reduce dimensions with Matryoshka embeddings
• Re-embedding schedule: Re-embed when you upgrade models, not continuously

Stage 4: Vector Database and Retrieval

Choosing a Vector Database

Hybrid Search

Pure vector search misses exact keyword matches. Hybrid search combines vector similarity with BM25 keyword search:

class HybridRetriever:
    def __init__(self, vector_store, bm25_index, alpha=0.7):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.alpha = alpha  # Weight for vector vs keyword

    async def retrieve(self, query: str, top_k: int = 10) -> list[Chunk]:
        # Vector search
        vector_results = await self.vector_store.search(query, top_k=top_k*2)

        # BM25 keyword search
        keyword_results = self.bm25_index.search(query, top_k=top_k*2)

        # Reciprocal Rank Fusion
        return self.rrf_merge(vector_results, keyword_results, top_k)

    def rrf_merge(self, vec_results, kw_results, top_k, k=60):
        scores = {}
        for rank, doc in enumerate(vec_results):
            scores[doc.id] = self.alpha * (1 / (k + rank + 1))
        for rank, doc in enumerate(kw_results):
            scores[doc.id] = scores.get(doc.id, 0) + (1 - self.alpha) * (1 / (k + rank + 1))

        sorted_ids = sorted(scores, key=scores.get, reverse=True)[:top_k]
        return [self.get_doc(id) for id in sorted_ids]

Advanced Retrieval Techniques

• Re-ranking: Use a cross-encoder model to re-rank the top-k results for better precision
• Metadata filtering: Filter by date, category, or access level before vector search
• Multi-query retrieval: Generate multiple query variations and merge results
• Contextual compression: Compress retrieved chunks to only include relevant portions

Stage 5: Generation with Retrieved Context

Prompt Engineering for RAG

The generation prompt is critical. A well-structured RAG prompt:

You are an expert assistant. Answer the user's question based ONLY on the provided context.
If the context doesn't contain enough information to answer, say so clearly.
Always cite your sources using [Source N] notation.

Context:
{retrieved_chunks_with_source_numbers}

Question: {user_query}

Answer:

Handling Edge Cases

• No relevant results: Detect when retrieval quality is low and respond honestly
• Contradictory sources: When documents conflict, present both perspectives
• Multi-hop questions: Some questions require chaining multiple retrievals
• Out of scope: Reject questions that are clearly outside your knowledge base

Evaluation Framework

You can't improve what you can't measure. RAG evaluation requires both retrieval and generation metrics:

Retrieval Metrics

• Recall@K: What percentage of relevant documents are in the top K results?
• Mean Reciprocal Rank (MRR): How high is the first relevant result ranked?
• Precision@K: What percentage of top K results are actually relevant?

Generation Metrics

• Faithfulness: Does the answer only use information from the retrieved context?
• Answer relevance: Does the answer actually address the question?
• Completeness: Does the answer cover all aspects of the question?

Automated Evaluation Pipeline

class RAGEvaluator:
    def __init__(self, judge_model):
        self.judge = judge_model

    async def evaluate(self, question, answer, context, ground_truth=None):
        faithfulness = await self.judge.score_faithfulness(answer, context)
        relevance = await self.judge.score_relevance(answer, question)

        metrics = {
            "faithfulness": faithfulness,
            "relevance": relevance,
        }

        if ground_truth:
            metrics["correctness"] = await self.judge.score_correctness(
                answer, ground_truth
            )

        return metrics

Production Deployment

Performance Optimization

• Caching: Cache embeddings for frequent queries
• Streaming: Stream the generation response while retrieval happens
• Async retrieval: Run vector search and keyword search in parallel
• Connection pooling: Reuse database connections across requests

Monitoring

• Track retrieval latency, generation latency, and end-to-end latency
• Monitor retrieval quality with automated evaluation on a sample of queries
• Alert on embedding model degradation or vector store performance issues
• Log all queries and responses for continuous improvement

Getting Started

Building a production RAG system requires careful attention to each pipeline stage. The difference between a demo and a production system is in the details: robust document parsing, smart chunking, hybrid search, comprehensive evaluation, and continuous monitoring.

Contact Cloudpush to discuss how we can help you build a RAG system that works reliably at scale.