Production RAG
Why this matters
A RAG prototype that works on 1000 documents and 10 queries per day is a very different engineering problem from a RAG system serving 50,000 documents, 500 queries per minute, incremental document updates, and users who notice when answers take 4 seconds. The gap between demo and production is almost entirely in the operational concerns: caching, latency budgets, incremental indexing, cost visibility, and the monitoring that tells you when something is drifting before users notice.
If you are building RAG at work — even internal tooling — you will hit these problems faster than you expect. This lesson covers the patterns that turn a RAG script into a system you can actually operate.
Core concepts
Latency budget. A typical RAG call has this latency stack: embedding the query (50–200ms), vector search (10–100ms), optional reranking (100–500ms), LLM generation (300ms–3s). Total: 500ms to 4s. Identify which step dominates your latency, then optimize that step first. For most systems, the LLM generation step is the bottleneck. For very large indexes, vector search can become significant.
Query caching. Identical or near-identical queries should not hit the embedding model and vector DB every time. Cache the retrieved chunks (and optionally the final answer) keyed by a hash of the query. Redis with a TTL is the simplest implementation. For semantic deduplication, you can cache the embedding itself and use it to find cache-hit candidates by similarity.
Streaming. For end-user interfaces, stream the LLM response token-by-token instead of waiting for the full answer. Users tolerate 4 seconds of total response time much better when they see tokens arriving immediately. Most LLM clients support streaming with a few lines of code. Streaming does not reduce total latency but it dramatically improves perceived performance.
Incremental indexing. A naive approach is to re-index the entire corpus every time a document changes. For large corpora, this is hours of work and significant cost. Production systems use incremental indexing: track which documents have changed (by hash or modification timestamp), re-embed only changed documents, delete old vectors by document ID, and insert new vectors. This requires a document store alongside the vector DB to track state.
Cost at scale. At 1000 queries/day with 5 chunks of 512 tokens each retrieved per query: that is ~2.5M context tokens per day. At $0.10/1M tokens input, that is $0.25/day — negligible. But if your LLM generation uses a full 8K context window for each query, you are at 8M tokens/day, or $0.80/day — still fine. The costs compound when you add reranking API calls, expensive embedding models, and when query volume scales. Monitor cost per query from day one so you know what a 10x traffic spike costs before it happens.
Monitoring. The metrics that matter in production:
- Retrieval latency P50/P95 (separate from generation latency)
- Vector DB query time by index size
- Embedding API error rate and latency
- Cache hit rate (if you have query caching)
- Answer quality score from automated eval (sampled, not every query)
- User feedback signals (thumbs down, correction submissions)
Working example
from __future__ import annotations
import hashlib
import json
import time
from dataclasses import dataclass
from typing import Optional, Iterator
import redis
from openai import OpenAI
@dataclass
class RAGResponse:
answer: str
chunks: list[dict]
from_cache: bool
latency_ms: dict # {step: ms}
class ProductionRAGPipeline:
"""
A production-grade RAG pipeline with:
- Query result caching (Redis)
- Streaming generation
- Per-step latency tracking
- Incremental document indexing support
"""
def __init__(
self,
index, # Your RAGIndex from previous lesson
redis_client: redis.Redis,
cache_ttl_seconds: int = 3600,
top_k: int = 5,
system_prompt: str = "Answer the question using only the provided context.",
):
self.index = index
self.redis = redis_client
self.cache_ttl = cache_ttl_seconds
self.top_k = top_k
self.system_prompt = system_prompt
self.openai = OpenAI()
def _cache_key(self, query: str, where: Optional[dict] = None) -> str:
payload = json.dumps({"q": query, "w": where or {}}, sort_keys=True)
return "rag:v1:" + hashlib.sha256(payload.encode()).hexdigest()[:16]
def query(
self,
question: str,
where: Optional[dict] = None,
use_cache: bool = True,
) -> RAGResponse:
latencies: dict[str, float] = {}
cache_key = self._cache_key(question, where)
# Check cache
if use_cache:
t0 = time.monotonic()
cached = self.redis.get(cache_key)
latencies["cache_check_ms"] = (time.monotonic() - t0) * 1000
if cached:
data = json.loads(cached)
return RAGResponse(
answer=data["answer"],
chunks=data["chunks"],
from_cache=True,
latency_ms=latencies,
)
# Retrieve chunks
t0 = time.monotonic()
results = self.index.search(question, top_k=self.top_k, where=where)
latencies["retrieval_ms"] = (time.monotonic() - t0) * 1000
context = "\n\n---\n\n".join(
f"[Source: {r.metadata.get('source', 'unknown')}]\n{r.text}"
for r in results
)
chunks_data = [
{"text": r.text, "metadata": r.metadata, "score": r.score}
for r in results
]
# Generate answer
t0 = time.monotonic()
response = self.openai.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": self.system_prompt},
{
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {question}",
},
],
temperature=0.1,
max_tokens=512,
)
latencies["generation_ms"] = (time.monotonic() - t0) * 1000
answer = response.choices[0].message.content
# Cache and return
if use_cache:
self.redis.setex(
cache_key,
self.cache_ttl,
json.dumps({"answer": answer, "chunks": chunks_data}),
)
return RAGResponse(
answer=answer,
chunks=chunks_data,
from_cache=False,
latency_ms=latencies,
)
def stream_query(self, question: str, where: Optional[dict] = None) -> Iterator[str]:
"""Stream the answer token-by-token. Yields tokens as they arrive."""
results = self.index.search(question, top_k=self.top_k, where=where)
context = "\n\n---\n\n".join(r.text for r in results)
stream = self.openai.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
],
stream=True,
)
for chunk in stream:
token = chunk.choices[0].delta.content
if token:
yield token
class IncrementalIndexer:
"""
Track document versions and only re-index changed documents.
Uses a simple file-hash approach.
"""
def __init__(self, index, state_file: str = ".index_state.json"):
self.index = index
self.state_file = state_file
self._state: dict[str, str] = self._load_state()
def _load_state(self) -> dict:
try:
with open(self.state_file) as f:
return json.load(f)
except FileNotFoundError:
return {}
def _save_state(self) -> None:
with open(self.state_file, "w") as f:
json.dump(self._state, f)
def _doc_hash(self, content: str) -> str:
return hashlib.sha256(content.encode()).hexdigest()
def upsert_document(
self,
doc_id: str,
content: str,
chunks: list[dict], # Pre-chunked from your chunking pipeline
) -> bool:
"""
Index document only if content has changed.
Returns True if document was (re)indexed, False if skipped.
"""
new_hash = self._doc_hash(content)
if self._state.get(doc_id) == new_hash:
return False # Not changed — skip
# Delete old vectors for this document
try:
# Chroma: delete by where filter on metadata
self.index.collection.delete(where={"source_doc_id": doc_id})
except Exception:
pass # New document — nothing to delete
# Add chunks with doc_id in metadata
for chunk in chunks:
chunk.setdefault("metadata", {})["source_doc_id"] = doc_id
self.index.add_documents(chunks)
self._state[doc_id] = new_hash
self._save_state()
return True
Common mistakes
-
Re-indexing everything on every deploy. If you rebuild the index from scratch each time you update the ingestion pipeline, you will have downtime. Separate the index lifecycle from the deployment lifecycle. Production systems keep the old index live while the new one builds.
-
Not caching at all until it is a problem. Query caching is 3 lines of code (hash query, check Redis, set if miss). Add it before launch, not when latency becomes a user complaint.
-
Ignoring streaming for user-facing features. If users are waiting for answers, stream the generation. The perceived latency difference is dramatic and the implementation is trivial.
-
No cost attribution per feature. If multiple features share a RAG pipeline, you cannot tell which one is expensive without per-feature cost tagging. Add a
featurelabel to every LLM call from day one. -
Treating the index as append-only. Documents change. Employees leave. Contracts expire. Without soft-delete or document-level replacement, your index accumulates stale content that degrades answer quality over months.
Try it yourself
Add timing instrumentation to your RAG pipeline that logs retrieval latency, generation latency, and total latency for every query. Run 50 queries and compute P50 and P95. Which step is the bottleneck? Now add a simple in-memory cache (a dict keyed by query hash with a 5-minute TTL) and re-run the same 50 queries. What is your cache hit rate if queries repeat? What does this tell you about the query distribution for your use case?