Why this matters

Shipping a RAG prototype is the easy part. Keeping it accurate, fast, and affordable as your document corpus grows from hundreds to hundreds of thousands of documents — and as query volume grows from tens to tens of thousands per day — is the hard part. Most teams discover these problems after launch, when they are already under pressure. This lesson covers the operational patterns that production RAG systems require: index lifecycle management, multi-layer caching, retrieval quality monitoring, and cost control.

If you are building RAG for real users at work, these are the engineering decisions that determine whether the system is maintainable in six months.

Core concepts

Index management: incremental updates and versioning

The naive approach is to rebuild the entire vector index whenever documents change. This works at 1,000 documents. At 100,000 documents with daily updates, a full rebuild takes hours, requires your index to be offline during reindexing, and wastes embedding API budget re-embedding unchanged documents.

Incremental indexing only re-embeds changed documents. The pattern: hash each document's content, compare to stored hashes, re-embed only documents where the hash changed, delete old vectors by document ID, insert new vectors. This requires maintaining a document registry alongside the vector store.

Index versioning runs in shadow mode: build the new index version while the current version serves traffic, validate quality metrics on the new version, then switch traffic atomically. This eliminates indexing-related downtime. Chroma, Pinecone, and Weaviate all support multiple named collections for exactly this pattern.

Soft deletes. When a document is removed, immediately mark it deleted in the document registry and filter it out of retrieval results. Batch the actual vector deletion to avoid blocking the write path. Stale content left in the index will surface in answers for weeks if you skip this step.

Caching strategies

A production RAG system has three distinct caching opportunities:

Query cache — cache the full retrieval + generation result, keyed by a hash of the query and any filter parameters. Redis with a TTL of 1-24 hours is the standard implementation. Even a 20% hit rate meaningfully reduces latency and cost.

Embedding cache — cache the query embedding keyed by query text. A simple in-memory LRU cache with 10,000 entries covers most repeated queries in a session and eliminates redundant embedding API calls.

Result cache with semantic deduplication — for near-duplicate queries ("what is the refund policy" vs "what's the refund policy"), exact hash matching misses the cache. Semantic deduplication stores query embeddings alongside cached results and returns a cache hit if a new query is within a similarity threshold of a cached query. More complex, but dramatically improves effective hit rate for paraphrased queries.

Monitoring retrieval quality in production

Unlike traditional software where bugs throw exceptions, RAG quality degrades silently. Retrieval can get worse as your document corpus changes and nobody notices until users stop trusting the system. Monitoring is how you catch this before it becomes a crisis.

Metrics to track per query:

• Retrieval latency (P50, P95, P99)
• Average cosine similarity of top-k results (low similarity = off-topic retrieval)
• Cache hit rate
• Answer generation latency

Metrics to track on a sampled basis:

• Automated faithfulness score via LLM judge (sampled 5-10% of queries)
• Citation accuracy (does the answer cite the right sources?)
• User feedback signals (thumbs down, correction submissions)

Drift detection. Compare rolling average similarity scores week-over-week. A sustained drop usually means new documents were added that differ significantly from the query distribution, or the query distribution shifted.

Cost optimization

Embedding cache. For a system with 30% query repetition, an in-process LRU cache gives a free 30% reduction in embedding API calls.

Smaller models for routing. If your RAG system serves multiple document collections, use a small fast model (or keyword matching) to route the query to the right collection before embedding and retrieval. This avoids querying large indexes for queries that only match a small subset.

Retrieve more, rerank down. Retrieve top-20 candidates cheaply with vector search, then rerank to top-5 with a cross-encoder. Often cheaper than using an expensive similarity metric throughout, and produces better results.

Context window sizing. Do not pass all top-k chunks into the context window. Use similarity thresholds to filter chunks below a minimum score before including them in the prompt. Shorter prompts mean lower generation costs.

Working example

from __future__ import annotations
import hashlib
import json
import logging
import time
from dataclasses import dataclass, field
from functools import lru_cache
from typing import Optional

import redis
from openai import OpenAI

logger = logging.getLogger(__name__)
client = OpenAI()


# In-process embedding cache

@lru_cache(maxsize=10_000)
def get_embedding_cached(text: str, model: str = "text-embedding-3-small") -> tuple[float, ...]:
    """LRU cache for query embeddings. Returns tuple (hashable) not list."""
    resp = client.embeddings.create(model=model, input=text)
    return tuple(resp.data[0].embedding)


# Redis query cache

class QueryCache:
    """Cache full RAG results in Redis with TTL-based expiry."""

    def __init__(self, redis_client: redis.Redis, ttl_seconds: int = 3600, prefix: str = "rag:v1"):
        self.redis = redis_client
        self.ttl = ttl_seconds
        self.prefix = prefix

    def _key(self, query: str, filters: Optional[dict] = None) -> str:
        payload = json.dumps({"q": query, "f": filters or {}}, sort_keys=True)
        return f"{self.prefix}:{hashlib.sha256(payload.encode()).hexdigest()[:16]}"

    def get(self, query: str, filters: Optional[dict] = None) -> Optional[dict]:
        raw = self.redis.get(self._key(query, filters))
        if raw:
            logger.info("cache_hit", extra={"query": query[:80]})
            return json.loads(raw)
        return None

    def set(self, query: str, result: dict, filters: Optional[dict] = None) -> None:
        self.redis.setex(self._key(query, filters), self.ttl, json.dumps(result))

    def invalidate_prefix(self, prefix_override: str) -> int:
        """Bulk invalidate all keys for a version prefix. Use after index updates."""
        keys = self.redis.keys(f"{prefix_override}:*")
        return self.redis.delete(*keys) if keys else 0


# Retrieval quality logger

@dataclass
class RetrievalMetrics:
    query: str
    top_k_scores: list[float]
    retrieval_latency_ms: float
    generation_latency_ms: float
    from_cache: bool
    filters: dict = field(default_factory=dict)

    @property
    def avg_score(self) -> float:
        return sum(self.top_k_scores) / len(self.top_k_scores) if self.top_k_scores else 0.0

    def to_log_dict(self) -> dict:
        return {
            "event": "rag_query",
            "avg_retrieval_score": round(self.avg_score, 4),
            "min_retrieval_score": round(min(self.top_k_scores, default=0.0), 4),
            "retrieval_latency_ms": round(self.retrieval_latency_ms, 1),
            "generation_latency_ms": round(self.generation_latency_ms, 1),
            "from_cache": self.from_cache,
            "result_count": len(self.top_k_scores),
        }


def log_retrieval_metrics(metrics: RetrievalMetrics) -> None:
    log_data = metrics.to_log_dict()
    if metrics.avg_score < 0.35:
        logger.warning("low_retrieval_quality", extra=log_data)
    else:
        logger.info("rag_query_complete", extra=log_data)


# Production RAG service with caching and quality logging

class ProductionRAGService:
    """
    RAG service with in-process embedding cache, Redis query cache,
    per-query retrieval quality logging, and score-threshold filtering.
    """

    def __init__(
        self,
        index,
        redis_client: redis.Redis,
        cache_ttl: int = 3600,
        top_k: int = 10,
        score_threshold: float = 0.3,
        max_context_chunks: int = 5,
    ):
        self.index = index
        self.cache = QueryCache(redis_client, ttl_seconds=cache_ttl)
        self.top_k = top_k
        self.score_threshold = score_threshold
        self.max_context_chunks = max_context_chunks

    def query(self, question: str, filters: Optional[dict] = None) -> dict:
        # 1. Check cache
        cached = self.cache.get(question, filters)
        if cached:
            log_retrieval_metrics(RetrievalMetrics(
                query=question, top_k_scores=[], retrieval_latency_ms=0,
                generation_latency_ms=0, from_cache=True, filters=filters or {},
            ))
            return {**cached, "from_cache": True}

        # 2. Embed (in-process LRU cache)
        embedding = list(get_embedding_cached(question))

        # 3. Retrieve
        t0 = time.monotonic()
        raw_results = self.index.query(embedding, top_k=self.top_k, where=filters)
        retrieval_ms = (time.monotonic() - t0) * 1000

        # 4. Filter low-quality chunks to reduce prompt size and cost
        filtered = [r for r in raw_results if r["score"] >= self.score_threshold]
        context_chunks = filtered[: self.max_context_chunks]
        scores = [r["score"] for r in raw_results]

        if not context_chunks:
            log_retrieval_metrics(RetrievalMetrics(
                query=question, top_k_scores=scores, retrieval_latency_ms=retrieval_ms,
                generation_latency_ms=0, from_cache=False, filters=filters or {},
            ))
            return {
                "answer": "I don't have enough information in the knowledge base to answer this question.",
                "chunks": [],
                "from_cache": False,
                "retrieval_quality": "low",
            }

        # 5. Generate
        context = "\n\n---\n\n".join(
            f"[Source: {r['metadata'].get('source', 'unknown')}]\n{r['text']}"
            for r in context_chunks
        )
        t0 = time.monotonic()
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Answer using only the provided context."},
                {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
            ],
            temperature=0.1,
            max_tokens=512,
        )
        generation_ms = (time.monotonic() - t0) * 1000
        answer = resp.choices[0].message.content

        # 6. Log and cache
        metrics = RetrievalMetrics(
            query=question, top_k_scores=scores, retrieval_latency_ms=retrieval_ms,
            generation_latency_ms=generation_ms, from_cache=False, filters=filters or {},
        )
        log_retrieval_metrics(metrics)

        result = {
            "answer": answer,
            "chunks": [{"text": r["text"], "score": r["score"], "metadata": r["metadata"]} for r in context_chunks],
            "from_cache": False,
            "retrieval_quality": "high" if metrics.avg_score >= 0.5 else "medium",
        }
        self.cache.set(question, result, filters)
        return result

Common mistakes

1. No score threshold. Passing all top-k chunks regardless of score bloats the prompt with irrelevant content and increases generation costs. A simple score >= 0.3 filter on cosine similarity removes the worst chunks for free.

2. Single-layer cache only. Teams often add Redis caching for full query results but skip embedding caching. The LRU embedding cache is 5 lines of code and eliminates redundant embedding API calls for repeated queries in the same process.

3. Logging without alerting. Logging low retrieval quality scores is only useful if someone looks at the logs. Set up an alert on avg_retrieval_score < 0.3 sustained for more than 100 queries — that is the signal that something has changed in your corpus or query distribution.

4. Rebuilding the cache on every deploy. Query cache invalidation should be scoped to what actually changed. If you update the prompt but not the index, do not invalidate retrieval caches. Use a version prefix in cache keys (rag:v1:, rag:v2:) to do bulk invalidation cheaply when the index changes.

5. No shadow index for updates. If you take the index offline to rebuild it, you have a production outage. Always build new index versions in parallel and switch traffic atomically. Most vector databases support multiple named collections for exactly this reason.

Try it yourself

Instrument your RAG pipeline with the RetrievalMetrics logger from this lesson. Run 100 varied queries through it (or simulate them from a question dataset). Plot the distribution of avg_retrieval_score across all queries. What percentage of queries fall below 0.35? For those queries, what do they have in common — are they longer, more specific, more ambiguous? Now add the score threshold filter and measure how often it triggers. Does excluding low-score chunks improve or hurt answer quality for borderline queries?