"Understanding RAG Pipelines: From Chunking to Production"

What is RAG?

Retrieval-Augmented Generation (RAG) is a technique that enhances LLM outputs by first retrieving relevant information from a knowledge base, then feeding that context to the model for generation. It addresses the fundamental limitation of LLMs: they only know what they were trained on.

The Full Pipeline

Stage 1: Document Ingestion

Before any retrieval can happen, documents must be processed and indexed. This involves:

class IngestionPipeline:
    def __init__(self):
        self.parser = DocumentParser()
        self.chunker = ChunkingStrategy()
        self.embedder = EmbeddingModel()
        self.indexer = VectorIndex()

    async def process(self, document: Document) -> None:
        # Parse document into raw text
        text = await self.parser.parse(document)

        # Split into chunks
        chunks = self.chunker.chunk(text)

        # Generate embeddings
        vectors = await self.embedder.embed_batch(chunks)

        # Index into vector store
        await self.indexer.add(vectors)

Stage 2: Chunking Strategies

Chunking quality is the single most important factor in RAG performance. Here are the strategies I've tested:

Fixed-Size Chunking

The simplest approach — split text every N tokens with optional overlap:

def fixed_chunk(text: str, size: int = 512, overlap: int = 64) -> list[Chunk]:
    tokens = tokenize(text)
    chunks = []
    start = 0
    while start < len(tokens):
        end = min(start + size, len(tokens))
        chunks.append(Chunk(tokens[start:end]))
        start += size - overlap
    return chunks

Pros: Simple, predictable. Cons: Cuts through sentences and paragraphs.

Semantic Chunking

Splits at natural boundaries like paragraphs and sections:

def semantic_chunk(text: str) -> list[Chunk]:
    # Split by double newlines (paragraphs)
    paragraphs = re.split(r'\n\n+', text)

    # Merge small paragraphs
    chunks = []
    current = ""
    for para in paragraphs:
        if len(current) + len(para) < MAX_CHUNK_SIZE:
            current += "\n\n" + para
        else:
            chunks.append(Chunk(current.strip()))
            current = para
    if current:
        chunks.append(Chunk(current.strip()))
    return chunks

Pros: Preserves semantic boundaries. Cons: Can produce very large chunks.

Stage 3: Embedding Models

Choosing the right embedding model involves trade-offs:

| Model | Dimensions | Quality | Speed | Cost | |-------|-----------|---------|-------|------| | text-embedding-3-small | 1536 | High | Fast | Low | | text-embedding-3-large | 3072 | Very High | Medium | Medium | | BAAI/bge-large-en-v1.5 | 1024 | High | Fast | Free | | intfloat/e5-mistral-7b | 4096 | Best | Slow | Free (GPU) |

For production, I recommend starting with text-embedding-3-small — it offers the best quality-to-cost ratio.

Stage 4: Retrieval Strategies

Hybrid Search

Hybrid search combines dense (semantic) and sparse (keyword) retrieval:

class HybridSearch:
    def __init__(self):
        self.dense = VectorStore()
        self.sparse = BM25Index()

    def search(self, query: str, top_k: int = 10):
        # Dense: semantic similarity
        q_vec = embed(query)
        dense_results = self.dense.search(q_vec, top_k * 2)

        # Sparse: keyword matching
        sparse_results = self.sparse.search(query, top_k * 2)

        # RRF fusion
        return self.rrf_fusion(dense_results, sparse_results, top_k)

    def rrf_fusion(self, dense, sparse, k=60):
        scores = {}
        for rank, doc in enumerate(dense + sparse):
            doc_id = doc.id
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank)
        return sorted(scores.items(), key=lambda x: -x[1])[:top_k]

Stage 5: Generation

The generation stage combines retrieved context with the user's question:

RAG_PROMPT = """You are a helpful assistant. Use the following context to answer the question.

Context:
{context}

Question: {question}

Instructions:
1. Only use information from the provided context
2. If the context does not contain the answer, say so
3. Cite your sources using [Source: filename]
4. Keep answers concise and accurate

Answer:"""

Advanced Techniques

Query Transformation

Raw user queries often perform poorly for retrieval. Common transformation strategies:

1. Query Rewriting — Rephrase the query for better retrieval
2. Query Decomposition — Split complex questions into sub-queries
3. HyDE — Generate a hypothetical document to search with

class QueryTransformer:
    def rewrite(self, query: str) -> list[str]:
        prompt = f"Generate 3 search queries for: {query}"
        return llm.generate(prompt).split('\n')

Re-ranking

A second-pass re-ranker significantly improves results:

class ReRanker:
    def rerank(self, query: str, documents: list[Doc]) -> list[Doc]:
        pairs = [(query, doc.text) for doc in documents]
        scores = cross_encoder.predict(pairs)
        return [doc for _, doc in sorted(zip(scores, documents), reverse=True)]

Evaluation

Building an evaluation pipeline is the single most important investment:

class RAGEvaluator:
    def __init__(self):
        self.metrics = {
            "retrieval_precision": PrecisionMetric(),
            "retrieval_recall": RecallMetric(),
            "answer_relevance": RelevanceMetric(),
            "citation_accuracy": CitationMetric(),
        }

    def evaluate(self, test_set: list[TestCase]):
        results = {}
        for query, expected in test_set:
            response = rag_pipeline(query)
            for name, metric in self.metrics.items():
                results[name] = metric(response, expected)
        return results

Production Considerations

Caching

Caching significantly reduces latency and cost:

class CachedRetriever:
    def __init__(self):
        self.cache = RedisCache(ttl=3600)
        self.retriever = HybridRetriever()

    async def search(self, query: str):
        # Check cache first
        cached = await self.cache.get(query)
        if cached:
            return cached

        # Miss — retrieve and cache
        results = await self.retriever.search(query)
        await self.cache.set(query, results)
        return results

Monitoring

Every production RAG system needs monitoring:

• Latency — P50/P95/P99 for each pipeline stage
• Retrieval quality — Tracking top-1/top-5 accuracy
• Hallucination rate — Percentage of responses with unsupported claims
• User feedback — Thumbs up/down on responses

Conclusion

Building a production RAG system requires careful consideration at every stage. The key takeaway: start simple with fixed chunking and dense retrieval, measure performance, then iteratively optimize. Most teams over-engineer their first iteration and miss the fundamentals.

The most impactful improvements I've found in practice are:

1. Hybrid search (dense + sparse)
2. Query transformation
3. Evaluation pipeline
4. Caching layer

These four components consistently produce the biggest quality improvements with the least engineering effort.