End-to-End RAG Serving

Individual RAG components — embedding, retrieval, reranking, generation — are well-understood. The hard part is orchestrating them into a reliable, low-latency production pipeline. This chapter covers the end-to-end architecture: latency budgets, parallelization, caching, failure handling, and quality monitoring.

---

1. End-to-End Architecture

User query
  │
  ├─→ Embed query (10ms)
  │     │
  │     ├─→ Vector search (20ms)           ─┐
  │     └─→ BM25 keyword search (15ms)      │── Parallel
  │                                         ─┘
  ├─→ Fuse results (RRF, 1ms)
  │
  ├─→ Rerank top-K (30ms)
  │
  ├─→ Build prompt (context + query) (2ms)
  │
  └─→ LLM generation (500-2000ms)
        │
        └─→ Stream tokens to user

Total latency: 65ms retrieval + LLM generation. The retrieval pipeline should be <100ms to keep TTFT low.

---

2. Latency Budget Allocation

Every millisecond in the retrieval pipeline delays the first token. Allocate a budget and enforce it:

class LatencyBudget:
    """Track and enforce per-stage latency budgets."""
    
    BUDGETS = {
        "embed_query":     15,   # ms
        "vector_search":   30,   # ms
        "keyword_search":  25,   # ms
        "fusion":           2,   # ms
        "rerank":          40,   # ms
        "prompt_build":     5,   # ms
        "total_retrieval": 100,  # ms (hard ceiling before LLM call)
    }
    
    def __init__(self):
        self.timings = {}
        self.start = time.monotonic()
    
    def record(self, stage: str, duration_ms: float):
        self.timings[stage] = duration_ms
        budget = self.BUDGETS.get(stage)
        if budget and duration_ms > budget * 1.5:  # 50% over budget = warning
            logger.warning(f"Stage {stage} took {duration_ms:.1f}ms (budget: {budget}ms)")
    
    def remaining(self) -> float:
        elapsed = (time.monotonic() - self.start) * 1000
        return max(0, self.BUDGETS["total_retrieval"] - elapsed)
    
    def should_skip_rerank(self) -> bool:
        """Skip reranking if we've already blown the budget."""
        return self.remaining() < self.BUDGETS["rerank"]

---

3. The RAG Orchestrator

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

class RAGOrchestrator:
    def __init__(self, embedder, vector_db, bm25_index, reranker, llm):
        self.embedder = embedder
        self.vector_db = vector_db
        self.bm25_index = bm25_index
        self.reranker = reranker
        self.llm = llm
    
    async def query(self, user_query: str, top_k: int = 5) -> AsyncIterator[str]:
        budget = LatencyBudget()
        
        # Stage 1: Embed query
        t0 = time.monotonic()
        query_vector = self.embedder.encode(user_query)
        budget.record("embed_query", elapsed_ms(t0))
        
        # Stage 2: Parallel retrieval (vector + keyword)
        t0 = time.monotonic()
        vector_task = asyncio.create_task(
            self.vector_db.search(query_vector, top_k=top_k * 4)
        )
        keyword_task = asyncio.create_task(
            self.bm25_index.search(user_query, top_k=top_k * 4)
        )
        vector_results, keyword_results = await asyncio.gather(
            vector_task, keyword_task
        )
        budget.record("parallel_retrieval", elapsed_ms(t0))
        
        # Stage 3: Fuse results
        t0 = time.monotonic()
        fused = reciprocal_rank_fusion(vector_results, keyword_results, top_k=top_k * 2)
        budget.record("fusion", elapsed_ms(t0))
        
        # Stage 4: Rerank (skip if over budget)
        if not budget.should_skip_rerank():
            t0 = time.monotonic()
            passages = [r.text for r in fused]
            reranked = self.reranker.rerank(user_query, passages, top_k=top_k)
            budget.record("rerank", elapsed_ms(t0))
            context_passages = [p.text for p in reranked]
        else:
            logger.warning("Skipping rerank — over latency budget")
            context_passages = [r.text for r in fused[:top_k]]
        
        # Stage 5: Build prompt
        prompt = self._build_prompt(user_query, context_passages)
        
        # Stage 6: Stream LLM response
        async for token in self.llm.stream(prompt):
            yield token
    
    def _build_prompt(self, query: str, passages: list[str]) -> str:
        context = "\n\n---\n\n".join(passages)
        return f"""Answer the question based on the provided context. If the context doesn't contain enough information, say so.

Context:
{context}

Question: {query}

Answer:"""

@app.post("/rag")
async def rag_endpoint(query: str):
    orchestrator = RAGOrchestrator(embedder, vector_db, bm25_index, reranker, llm)
    return StreamingResponse(
        orchestrator.query(query),
        media_type="text/event-stream",
    )

---

4. Parallelization Strategies

What Can Run in Parallel

Sequential (must wait):         Parallelizable:
  embed → search (needs vector)   vector_search ║ keyword_search
  search → rerank (needs results)  multi-source retrieval (DB1 ║ DB2 ║ DB3)
  rerank → LLM (needs context)    embed_query ║ check_cache

Multi-Source Retrieval

For complex RAG, retrieve from multiple sources simultaneously:

async def multi_source_retrieve(query_vector, query_text, sources):
    """Retrieve from multiple sources in parallel."""
    tasks = []
    for source in sources:
        if source.type == "vector":
            tasks.append(source.db.search(query_vector, top_k=20))
        elif source.type == "keyword":
            tasks.append(source.index.search(query_text, top_k=20))
        elif source.type == "knowledge_graph":
            tasks.append(source.kg.query(query_text, max_hops=2))
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Handle partial failures — use whatever succeeded
    valid_results = []
    for source, result in zip(sources, results):
        if isinstance(result, Exception):
            logger.error(f"Source {source.name} failed: {result}")
            continue
        valid_results.extend(result)
    
    return valid_results

---

5. Caching Architecture

Three levels of caching in a RAG pipeline:

Cache level	What’s cached	Hit rate	Savings
Query embedding cache	Embed(query) → vector	10-30% (repeated queries)	Skip embedding model
Retrieval cache	Query → retrieved passages	15-40% (FAQ-heavy workloads)	Skip vector search + rerank
Response cache	(Query, context) → LLM response	5-20% (exact duplicates)	Skip entire LLM call

class RAGCache:
    def __init__(self, redis_client, ttl=3600):
        self.redis = redis_client
        self.ttl = ttl
    
    async def get_or_retrieve(self, query: str, retrieve_fn):
        """Check cache before retrieval."""
        cache_key = f"rag:retrieval:{hashlib.md5(query.encode()).hexdigest()}"
        
        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)  # Cache hit — skip retrieval + reranking
        
        # Cache miss — run full pipeline
        results = await retrieve_fn(query)
        await self.redis.setex(cache_key, self.ttl, json.dumps(results))
        return results
    
    async def get_or_generate(self, query: str, context: list[str], generate_fn):
        """Check cache before LLM generation."""
        # Hash both query and context — different context = different answer
        content = query + "||" + "||".join(context)
        cache_key = f"rag:response:{hashlib.md5(content.encode()).hexdigest()}"
        
        cached = await self.redis.get(cache_key)
        if cached:
            return cached  # Full response from cache
        
        response = await generate_fn(query, context)
        await self.redis.setex(cache_key, self.ttl, response)
        return response

Cache Invalidation

When documents change, cached results become stale:

async def invalidate_on_doc_update(doc_id: str):
    """Invalidate all cache entries that used this document."""
    # Option 1: Nuke all retrieval cache (simple, aggressive)
    await redis.delete_pattern("rag:retrieval:*")
    
    # Option 2: Track which cache entries used which docs (precise, complex)
    affected_keys = await redis.smembers(f"rag:doc_refs:{doc_id}")
    if affected_keys:
        await redis.delete(*affected_keys)

---

6. Failure Handling

RAG has multiple failure modes. Handle each gracefully:

Failure	Impact	Fallback
Embedding service down	Can’t search	Return cached results or fall back to keyword search
Vector DB timeout	No retrieval results	Fall back to keyword search only
Reranker timeout	Unranked results	Skip reranking, use vector search order
All retrieval fails	No context	Generate without context (warn user)
LLM timeout	No response	Retry with shorter context, or return “try again”

async def resilient_rag(query: str, orchestrator: RAGOrchestrator):
    """RAG with cascading fallbacks."""
    try:
        # Full pipeline
        return await orchestrator.query(query, top_k=5)
    except RetrievalTimeout:
        logger.warning("Retrieval timeout — trying keyword-only")
        try:
            # Fallback: keyword search only (no vector, no rerank)
            results = await orchestrator.bm25_index.search(query, top_k=5)
            context = [r.text for r in results]
            return await orchestrator.llm.generate(
                orchestrator._build_prompt(query, context)
            )
        except Exception:
            pass
    except Exception as e:
        logger.error(f"RAG pipeline failed: {e}")
    
    # Last resort: generate without context
    return await orchestrator.llm.generate(
        f"Answer this question (no additional context available): {query}"
    )

---

7. Quality Monitoring

RAG quality degrades silently — you need continuous measurement:

Retrieval Quality

# Track retrieval metrics per query
retrieval_metrics = {
    "num_results": len(retrieved),             # Should be ≥ top_k
    "avg_relevance_score": mean(scores),       # Should be stable
    "min_relevance_score": min(scores),        # Watch for garbage results
    "retrieval_latency_ms": retrieval_time,    # Watch for degradation
    "cache_hit": was_cached,                   # Track hit rate
}

Answer Quality (Automated)

async def assess_rag_quality(query, context, answer):
    """Use LLM-as-judge to assess RAG answer quality."""
    assessment = await judge_llm.generate(f"""
    Rate this answer on two dimensions (1-5):
    
    FAITHFULNESS: Does the answer only use information from the context?
    RELEVANCE: Does the answer address the question?
    
    Question: {query}
    Context: {context[:2000]}
    Answer: {answer}
    
    Respond as JSON: {{"faithfulness": X, "relevance": X}}
    """)
    return json.loads(assessment)

# Run on 1-5% of production traffic
if random.random() < 0.03:
    quality = await assess_rag_quality(query, context, answer)
    metrics.record("faithfulness", quality["faithfulness"])
    metrics.record("relevance", quality["relevance"])

Hallucination Detection

The most dangerous RAG failure: the LLM ignores context and makes things up.

def detect_hallucination(answer: str, context_passages: list[str]) -> float:
    """Estimate hallucination risk by checking answer against context."""
    answer_sentences = split_sentences(answer)
    hallucination_score = 0
    
    for sentence in answer_sentences:
        # Check if sentence is supported by any context passage
        max_support = max(
            semantic_similarity(sentence, passage)
            for passage in context_passages
        )
        if max_support < 0.5:  # Low support = potentially hallucinated
            hallucination_score += 1
    
    return hallucination_score / len(answer_sentences)

# Alert if hallucination rate exceeds threshold
if hallucination_rate > 0.15:
    alert("RAG hallucination rate above 15% — check retrieval quality")

---

Self-Check

• Design a RAG pipeline with latency budget allocation per stage
• Implement parallel retrieval from multiple sources with failure handling
• Build a three-level caching strategy (embedding, retrieval, response)
• Describe cascading fallbacks when RAG components fail
• Set up automated quality monitoring with faithfulness and relevance scoring

---

References

• Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks (Lewis et al., 2020)
• RAGAS: Automated Evaluation of Retrieval Augmented Generation
• LlamaIndex RAG Documentation
• For reranker details, see Reranker Serving
• For embedding model selection, see Embedding Pipeline