Anatomy of RAG Systems

Building production-ready Retrieval-Augmented Generation (RAG) systems requires moving past naive lookup loops to structured, multi-stage pipelines. To guarantee low latency, factual grounding, and secure execution, developers must understand the core architectural layers that form a modern RAG system.

🏗️ 1. The Multi-Stage RAG Pipeline

A production-grade RAG system separates data ingestion from runtime query orchestration. While a naive RAG system operates as a single-turn database lookup, enterprise RAG applications execute a multi-stage retrieval and routing pipeline to filter, score, and isolate relevant context.

Naive vs. Production RAG Architecture

Ingestion vs. Query Data Flow

Component Execution Performance Matrix

📥 2. Ingestion & Document Processing

The ingestion pipeline parses raw files into structured markdown text blocks. Rather than extracting flat text strings, production ingestion utilizes Layout-Aware Parsers (e.g. Azure Document Intelligence, LlamaParse) to preserve the structural relationship of headers, paragraphs, and multi-column tables.

Structured Schema Metadata Extraction

Document processing should extract key metadata fields and tabular arrays into structured schemas at the ingestion boundary. Below is a Python pattern illustrating metadata extraction using Pydantic schemas:

from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import date
 
class TableCell(BaseModel):
    row_index: int
    column_index: int
    content: str
 
class ExtractedTable(BaseModel):
    table_id: str = Field(description="Unique table identifier in the document")
    headers: List[str] = Field(description="Headers of the table columns")
    cells: List[TableCell] = Field(description="Individual cells mapping table data")
 
class DocumentMetadataSchema(BaseModel):
    title: str = Field(description="The formal title of the document")
    document_date: Optional[date] = Field(description="The signing or effective date of the document")
    signatories: List[str] = Field(description="Parties executing the contract or agreement")
    governing_law: str = Field(description="Jurisdiction governing the document terms")
    extracted_tables: List[ExtractedTable] = Field(description="List of all tables detected in the document")
 
# Production implementation:
# response = client.beta.chat.completions.parse(
#     model="gpt-4o",
#     messages=[{"role": "user", "content": "Extract structured details from page..."}],
#     response_format=DocumentMetadataSchema
# )

✂️ 3. Chunking & Vectorization

Chunking Topologies

Splitting parsed text into discrete segments requires balancing context retention with retrieval granularity:

• Fixed-size Overlapping: Splitting text strictly by character or token limits (e.g., 512 tokens with a 10% overlap). While computationally cheap, it frequently cuts sentences in half, severing semantic context.
• Semantic Chunking: Splitting documents dynamically based on shifts in semantic distance between sequential sentences, maintaining cohesive context blocks.
• Hierarchical Parent-Child Chunking: Storing tiny, granular snippets (child chunks, e.g., 128 tokens) for vector similarity matching, but mapping them back to a larger parent chunk (e.g., 1024 tokens) containing the full context. When a child matches, the system injects the parent’s text into the LLM context.

Vectorization & Matryoshka Embeddings

Convert text chunks into numerical vectors using embedding models. To optimize memory footprint and search latency, leverage Matryoshka Representation Learning (MRL) models (e.g. OpenAI text-embedding-3). MRL trains embeddings to store the most critical semantic data in the first few dimensions, allowing you to slice vectors (e.g. compressing 1536 dimensions down to 512) while retaining up to 98% of the retrieval accuracy.

🔍 4. Advanced Retrieval & Search Fusion

Hybrid Search & Reciprocal Rank Fusion (RRF)

Standard vector search struggles with exact keyword matching (e.g. searching for a specific product ID like X-8902). Production RAG must utilize Hybrid Search, combining dense vector cosine similarity with sparse keyword searches (BM25). The retrieved candidate lists are merged using the Reciprocal Rank Fusion (RRF) algorithm:

RRF_Score(d ∈ D) = \sum_{m ∈ M} \frac{1}{k + r_m(d)}

Where (M) is the set of retrievers, and (r_m(d)) is the rank of document (d) in retriever (m).

def reciprocal_rank_fusion(sparse_results: list, dense_results: list, k: int = 60) -> list:
    """
    Combines search results from sparse (BM25) and dense (Vector) retrievers
    using the Reciprocal Rank Fusion (RRF) algorithm.
    
    sparse_results: List of chunk IDs returned by sparse search, ordered by rank.
    dense_results: List of chunk IDs returned by dense search, ordered by rank.
    k: Constant smoothing parameter (default 60).
    """
    rrf_scores = {}
    
    # 1. Score results from the sparse retriever
    for rank, doc_id in enumerate(sparse_results):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (k + rank + 1))
        
    # 2. Score results from the dense retriever
    for rank, doc_id in enumerate(dense_results):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (k + rank + 1))
        
    # 3. Sort document candidates by combined RRF score in descending order
    sorted_docs = sorted(rrf_scores.items(), key=lambda item: item[1], reverse=True)
    return sorted_docs

Database Indexing & Metadata Pre-Filtering (pgvector)

Metadata filtering restricts searches by user roles or tenant IDs. For performance, apply pre-filtering (filtering database rows before executing similarity calculations) rather than post-filtering. Below is a PostgreSQL pgvector configuration showcasing tenant-isolated Row-Level Security (RLS) and an HNSW index:

-- 1. Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;
 
-- 2. Create document chunks table with tenant RLS metadata
CREATE TABLE document_chunks (
    chunk_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id VARCHAR(255) NOT NULL,
    parent_id UUID,
    content TEXT NOT NULL,
    embedding VECTOR(512), -- Matryoshka dimension-aligned embedding
    metadata JSONB NOT NULL, -- Flexible tags: { "role_access": ["admin", "staff"] }
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
 
-- 3. Create HNSW index on the vector column using Cosine Distance
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops);
 
-- 4. Enable Row-Level Security on the chunks table
ALTER TABLE document_chunks ENABLE ROW LEVEL SECURITY;
 
-- 5. Create a security policy for tenant isolation
CREATE POLICY tenant_isolation_policy ON document_chunks
    FOR ALL
    USING (tenant_id = current_setting('app.current_tenant_id', true));
 
-- 6. Query executing similarity search with metadata pre-filtering
-- In application context: SET app.current_tenant_id = 'tenant-908';
SELECT chunk_id, content, 
       (1 - (embedding <=> '[0.015, -0.043, 0.12, ...]')) AS similarity_score
FROM document_chunks
WHERE metadata -> 'role_access' ?| ARRAY['staff', 'compliance-auditor']
ORDER BY embedding <=> '[0.015, -0.043, 0.12, ...]'
LIMIT 5;

🔄 5. Post-Retrieval Reranking & Context Compression

Retrieving too many chunks inflates token consumption and degrades model focus (due to attention dispersion). Post-retrieval optimizes the context window:

1. Cross-Encoder Reranking: Bi-encoders (used during initial vector search) calculate similarity scores for candidate documents independently. A Cross-Encoder Reranker (e.g. Cohere Rerank, BGE-Reranker) analyzes the query and retrieved document chunk together, capturing deeper semantic relevance and ordering candidates with high precision.
2. Context Token Compression: Strip redundant phrases and low-information words from the top-ranked text blocks (e.g., using LLMLingua) before formatting the prompt.
3. Prompt Context Isolation: Inject retrieved text blocks inside clear XML tags (e.g. <context_chunk> and </context_chunk>). Instruct the system prompt to treat information within these tags strictly as untrusted data, mitigating indirect prompt injection.

📊 6. Production Observability & Evaluation

Distributed Tracing Spans (OpenTelemetry)

Production observability requires logging parent-child trace spans conforming to OpenTelemetry GenAI Semantic Conventions to isolate bottlenecks:

Parent Trace: RAG Query Lifecycle (Trace ID: f0a9b2...)
  ├── Span 1: Query Expansion / Rewrite (LLM latency & token cost)
  ├── Span 2: Vector Search DB Query (pgvector execution I/O)
  ├── Span 3: Cross-Encoder Rerank (GPU execution latency)
  └── Span 4: Final LLM Generation (Time-to-first-token & usage metrics)

RAGAS Evaluation Framework

Evaluate RAG application performance programmatically using RAGAS metrics:

• Faithfulness (Groundedness): Measures if the generated response is derived only from the retrieved context.
• Answer Relevance: Evaluates if the generated response directly addresses the user’s initial question.
• Context Recall: Checks if the retrieval system fetched all required information points to answer the query.
• Context Precision: Assesses the ratio of relevant chunks to irrelevant chunks in the retrieved context block.

from openai import OpenAI
 
client = OpenAI()
 
def evaluate_groundedness(query: str, retrieved_context: str, generated_response: str) -> float:
    """
    Evaluates faithfulness/groundedness by checking if the response is derived 
    solely from the retrieved context without hallucinations.
    """
    system_prompt = (
        "You are an evaluator. Rate the Groundedness of the generated response "
        "relative to the retrieved context on a scale of 0.0 to 1.0.\n"
        "0.0 means the response contains complete hallucinations not supported by the context.\n"
        "1.0 means every factual statement in the response is strictly derived from the context.\n"
        "Output your score in the exact format: SCORE: <value>"
    )
    
    user_content = (
        f"User Query: {query}\n\n"
        f"Retrieved Context:\n{retrieved_context}\n\n"
        f"Generated Response:\n{generated_response}"
    )
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_content}
        ],
        temperature=0.0
    )
    
    output = response.choices[0].message.content.strip()
    try:
        score = float(output.split("SCORE:")[1].strip())
    except (IndexError, ValueError):
        score = 0.0
    return score

🔗 Related Sections

• Agentic Document Workflows (ADW) — Pipeline configurations for unstructured file queues.
• Agent Memory Systems — Truncating context windows for chat histories.
• Agent State Management — Persistent checkpoints and transaction safety.
• Agent Observability & Tracing — Setting up OpenTelemetry traces for tracking tool latencies.
• Agent Security & Guardrails — Direct and indirect prompt injection defenses.