Anatomy of a production RAG architecture

Most teams build a RAG demo in an afternoon. Getting a production RAG architecture to survive real traffic, stale data, and adversarial queries takes months. The difference isn’t better embeddings or a fancier reranker — it’s the infrastructure choices that make the system observable, correctable, and resilient when the happy path breaks.

This post walks through the anatomy of a production-grade system: ingestion pipelines that don’t silently corrupt context, retrieval that degrades gracefully, generation that cites sources, and the observability that lets you debug why the answer was wrong at 3 AM.

The thesis: RAG is a data engineering problem, not an LLM problem

The industry treats RAG as “vector search + LLM call.” That mental model produces demos. In production, the LLM is the easiest component to swap. The hard parts are: keeping the index consistent with source truth, handling documents that don’t fit in context, attributing every claim to evidence, and detecting drift before users do.

A production RAG architecture has four distinct layers, each with different SLAs, failure modes, and scaling characteristics:

1. Ingestion & normalization — transforms raw sources into searchable units with metadata
2. Retrieval & ranking — finds relevant context under latency and freshness constraints
3. Generation & grounding — produces answers that cite evidence and refuse when uncertain
4. Observability & feedback — closes the loop between production behavior and offline evaluation

Let’s dissect each.

Ingestion: the silent correctness killer

Most teams start with a simple pattern: chunk documents, embed chunks, upsert to vector store. This works until it doesn’t.

Chunking strategy determines retrieval ceiling

Fixed-size chunking (512 tokens, 50 overlap) is the default because it’s easy. It fails on:

• Code: splits functions mid-logic, separates docstrings from implementations
• Tables: destroys row/column relationships
• Legal/regulatory: splits defined terms from their definitions

Production systems use structure-aware chunking with fallback:

def chunk_document(doc: Document) -> list[Chunk]:
    # Try structure-aware first
    if doc.mime_type == "text/markdown":
        return chunk_by_heading(doc, max_tokens=512)
    if doc.mime_type == "application/pdf":
        return chunk_by_layout(doc, max_tokens=512)  # preserves tables
    if doc.mime_type in CODE_MIME_TYPES:
        return chunk_by_ast(doc, max_tokens=512)     # preserves functions/classes
    
    # Fallback: semantic chunking with overlap
    return semantic_chunk(doc, target_tokens=512, overlap_tokens=50)

Metadata is not optional

Every chunk needs: source_id, source_version, chunk_index, section_path, last_modified, access_control_tags. Without source_version, you cannot invalidate stale chunks when a document updates. Without access_control_tags, you leak data across tenants.

@dataclass
class Chunk:
    id: str                    # deterministic: hash(source_id + chunk_index + version)
    text: str
    embedding: list[float]
    metadata: ChunkMetadata

@dataclass
class ChunkMetadata:
    source_id: str
    source_version: str        # content hash or version number
    chunk_index: int
    section_path: str          # e.g. "Chapter 3 > Section 2.1 > Paragraph 4"
    last_modified: datetime
    access_tags: frozenset[str]  # e.g. {"pii", "finance", "tenant:acme-corp"}
    token_count: int

Idempotent upserts prevent index corruption

Re-ingestion happens constantly: schema changes, embedding model upgrades, bug fixes. Your upsert must be idempotent and versioned:

async def upsert_chunks(chunks: list[Chunk], index: VectorIndex) -> IngestionResult:
    # Group by source_id for atomic per-document updates
    by_source = groupby(chunks, key=lambda c: c.metadata.source_id)
    
    results = []
    for source_id, source_chunks in by_source:
        # Delete old version atomically
        await index.delete_by_filter({"source_id": source_id})
        
        # Upsert new version with version tag
        version = source_chunks[0].metadata.source_version
        await index.upsert([
            {**c.to_dict(), "source_version": version} 
            for c in source_chunks
        ])
        results.append(IngestedSource(source_id, version, len(source_chunks)))
    
    return IngestionResult(results)

Tradeoff: Per-document atomicity means a 10k-chunk document blocks ingestion for its duration. For massive docs, use chunk-level upserts with a generation counter and a background compaction job that removes stale generations.

Retrieval: latency, freshness, and the hybrid reality

Vector search alone fails on exact-match queries (error codes, SKUs, function names). Keyword search alone fails on semantic queries. Production systems run hybrid retrieval with reciprocal rank fusion (RRF) or learned fusion.

Hybrid retrieval with RRF

async def retrieve(
    query: str,
    top_k: int = 20,
    filters: dict | None = None,
    alpha: float = 0.5  # weight for vector vs keyword
) -> list[RetrievedChunk]:
    # Parallel fan-out
    vector_results, keyword_results = await asyncio.gather(
        vector_search(query, top_k * 2, filters),
        keyword_search(query, top_k * 2, filters),
    )
    
    # Reciprocal Rank Fusion
    scores: dict[str, float] = defaultdict(float)
    for rank, chunk in enumerate(vector_results, 1):
        scores[chunk.id] += alpha / (rank + 60)
    for rank, chunk in enumerate(keyword_results, 1):
        scores[chunk.id] += (1 - alpha) / (rank + 60)
    
    # Re-fetch top-k by fused score
    top_ids = sorted(scores, key=scores.get, reverse=True)[:top_k]
    return await fetch_chunks(top_ids)

Reranking is non-negotiable

A cross-encoder reranker (e.g., bge-reranker-v2-m3, jina-reranker-v2) adds 50-150ms but dramatically improves precision@k. Run it on the top 50-100 fused results, not the full index.

async def rerank(
    query: str,
    chunks: list[RetrievedChunk],
    top_k: int = 10,
    model: str = "bge-reranker-v2-m3"
) -> list[RerankedChunk]:
    if not chunks:
        return []
    
    pairs = [(query, c.text) for c in chunks]
    scores = await cross_encoder_score(pairs, model=model)
    
    reranked = [
        RerankedChunk(chunk=c, rerank_score=s)
        for c, s in zip(chunks, scores)
    ]
    reranked.sort(key=lambda x: x.rerank_score, reverse=True)
    return reranked[:top_k]

Freshness via filtered search, not re-indexing

Don’t re-index for every document update. Instead, store source_version in the index and filter at query time:

def build_freshness_filter(known_versions: dict[str, str]) -> dict:
    # known_versions: {source_id: expected_version}
    # Returns filter that only matches current versions
    return {
        "$and": [
            {"source_id": {"$in": list(known_versions.keys())}},
            {"source_version": {"$in": list(known_versions.values())}}
        ]
    }

This lets you update the version map in Redis (sub-ms) while the vector index catches up asynchronously.

Generation: grounding, citations, and refusal

The generation layer has three jobs: synthesize an answer, cite every claim, and refuse when evidence is insufficient.

Structured prompting with citation enforcement

SYSTEM_PROMPT = """You are a precise answerer. Rules:
1. Answer ONLY using the provided context chunks.
2. Every factual claim MUST end with a citation like [chunk_id].
3. If context is insufficient, respond with exactly: "INSUFFICIENT_EVIDENCE: <reason>"
4. Do not hedge. Do not add disclaimers beyond the citation format.
"""

def build_generation_prompt(query: str, chunks: list[RerankedChunk]) -> list[dict]:
    context_blocks = []
    for i, chunk in enumerate(chunks):
        context_blocks.append(
            f"[CHUNK {chunk.id}]\n"
            f"Source: {chunk.metadata.source_id} (v{chunk.metadata.source_version})\n"
            f"Section: {chunk.metadata.section_path}\n"
            f"{chunk.text}\n"
        )
    
    context = "\n---\n".join(context_blocks)
    
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"}
    ]

Streaming with citation validation

Stream the response but validate citations post-hoc:

async def generate_with_citations(
    prompt: list[dict],
    valid_chunk_ids: set[str],
    model: str = "gpt-4o-mini"
) -> GenerationResult:
    full_text = ""
    async for token in stream_chat(prompt, model=model):
        full_text += token
        yield token  # stream to client
    
    # Validate citations after generation
    citations = extract_citations(full_text)  # regex for [chunk_id]
    invalid = [c for c in citations if c not in valid_chunk_ids]
    
    if invalid:
        # Log for evaluation, optionally trigger regeneration
        logger.warning("hallucinated_citations", invalid=invalid, response=full_text)
    
    return GenerationResult(
        text=full_text,
        citations=citations,
        invalid_citations=invalid,
        refused=full_text.startswith("INSUFFICIENT_EVIDENCE")
    )

When to refuse

Refusal triggers:

• Top rerank score < threshold (e.g., 0.3 for cross-encoder)
• No chunks pass keyword overlap filter for entity-heavy queries
• Generated response contains INSUFFICIENT_EVIDENCE

def should_refuse(reranked: list[RerankedChunk], query: str) -> RefusalDecision:
    if not reranked:
        return RefusalDecision(refuse=True, reason="no_chunks_retrieved")
    
    top_score = reranked[0].rerank_score
    if top_score < 0.25:
        return RefusalDecision(refuse=True, reason=f"low_relevance_score={top_score:.3f}")
    
    # Entity check: if query has proper nouns, demand keyword overlap
    entities = extract_entities(query)
    if entities:
        has_overlap = any(
            any(e.lower() in c.text.lower() for e in entities)
            for c in reranked[:5]
        )
        if not has_overlap:
            return RefusalDecision(refuse=True, reason="entity_mismatch")
    
    return RefusalDecision(refuse=False)

Tradeoff: Aggressive refusal improves precision but frustrates users. Track refusal rate by query type and adjust thresholds per domain.

Observability: the only way to improve

You cannot fix what you don’t measure. A production RAG architecture emits structured events at every stage.

The minimal event schema

{
  "event": "rag_query",
  "trace_id": "abc-123",
  "timestamp": "2024-01-15T03:42:11.234Z",
  "user_id": "user_456",
  "tenant_id": "tenant_789",
  "query": "How do I configure TLS termination?",
  "retrieval": {
    "vector_latency_ms": 42,
    "keyword_latency_ms": 18,
    "fusion_latency_ms": 5,
    "rerank_latency_ms": 87,
    "chunks_retrieved": 50,
    "chunks_reranked": 20,
    "top_rerank_score": 0.84,
    "chunks_selected": 8,
    "filters_applied": {"tenant_id": "tenant_789", "access_tags": ["!pii"]}
  },
  "generation": {
    "model": "gpt-4o-mini",
    "prompt_tokens": 3421,
    "completion_tokens": 287,
    "latency_ms": 1240,
    "citations_emitted": 6,
    "invalid_citations": 0,
    "refused": false
  },
  "outcome": {
    "user_feedback": null,  // filled later via thumbs up/down
    "escalated_to_human": false
  }
}

Offline evaluation pipeline

Production logs feed a nightly evaluation job:

async def run_nightly_eval(date: date) -> EvalReport:
    events = await fetch_events(date, event_type="rag_query")
    
    # Sample for human annotation (stratified by refusal, feedback)
    annotated = await sample_for_annotation(events, n=200)
    
    # Automated metrics on full set
    metrics = {
        "refusal_rate": mean(e.generation.refused for e in events),
        "invalid_citation_rate": mean(
            len(e.generation.invalid_citations) / max(1, len(e.generation.citations))
            for e in events
        ),
        "latency_p50": percentile(e.retrieval.total_latency_ms + e.generation.latency_ms, 50),
        "latency_p99": percentile(..., 99),
    }
    
    # LLM-as-judge on annotated subset
    judge_results = await llm_judge(annotated, criteria=[
        "groundedness", "completeness", "conciseness", "tone"
    ])
    
    return EvalReport(date, metrics, judge_results)

Drift detection

Track embedding drift by re-embedding a fixed validation set weekly:

VALIDATION_QUERIES = [
    "What is the refund policy for enterprise customers?",
    "How do I rotate API keys?",
    # ... 50 representative queries
]

async def detect_embedding_drift(current_model: str) -> DriftReport:
    current_embeddings = await embed_batch(VALIDATION_QUERIES, current_model)
    baseline_embeddings = await fetch_baseline_embeddings()
    
    similarities = [
        cosine_sim(c, b) for c, b in zip(current_embeddings, baseline_embeddings)
    ]
    
    return DriftReport(
        model=current_model,
        mean_similarity=mean(similarities),
        min_similarity=min(similarities),
        drifted_queries=[
            q for q, s in zip(VALIDATION_QUERIES, similarities) if s < 0.85
        ]
    )

A drop below 0.9 mean similarity warrants investigation — often a provider silently changed their embedding model.

Failure modes and mitigations

Circuit breaker pattern for retrieval

class RetrievalCircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 30):
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.timeout_seconds = timeout_seconds
        self.last_failure: datetime | None = None
        self._lock = asyncio.Lock()
    
    async def call(self, fn: Callable, fallback: Callable):
        async with self._lock:
            if self.is_open:
                return await fallback()
        
        try:
            result = await fn()
            async with self._lock:
                self.failures = 0
            return result
        except Exception as e:
            async with self._lock:
                self.failures += 1
                self.last_failure = datetime.utcnow()
                if self.failures >= self.failure_threshold:
                    logger.error("circuit_breaker_opened", failures=self.failures)
            return await fallback()
    
    @property
    def is_open(self) -> bool:
        if self.failures < self.failure_threshold:
            return False
        if self.last_failure and (datetime.utcnow() - self.last_failure).total_seconds() > self.timeout_seconds:
            self.failures = 0  # half-open
            return False
        return True

The decisive takeaway

A production RAG architecture is not a vector store plus an LLM call. It is a data pipeline with strict contracts between stages: ingestion emits versioned, access-controlled chunks; retrieval honors freshness and tenant boundaries; generation validates its own citations; observability closes the loop with automated and human evaluation.

The teams that ship reliable RAG systems invest first in:

1. Idempotent, versioned ingestion — so re-indexing is safe and frequent
2. Hybrid retrieval with reranking — because vector-only fails on entities
3. Citation-enforced generation with refusal — because ungrounded answers destroy trust
4. Structured logging from day one — because you cannot evaluate what you don’t record

Everything else — choice of embedding model, vector database, LLM provider — is swappable infrastructure. The architecture is the contracts between these layers. Get the contracts right, and the components become interchangeable. Get them wrong, and no model upgrade will save you.