Production RAG

A prototype RAG pipeline has one job: retrieve documents and generate an answer. Production adds five more.

Three layers to build on Lesson 5: Cache (embedding, query results, answers), Index (versioned, model-tagged), Ops (monitoring, alerting, rollback).

RAG has three expensive operations -- embedding (~10-50 ms), vector search (~20-200 ms), LLM generation (~500 ms+) -- each cacheable with separate TTLs.

Embedding cache (1 hr) -- SHA-256 the query, store raw float32 bytes in Redis. Hit: np.frombuffer(r.get(key), dtype=np.float32). Miss: embed, then r.setex(key, 3600, vec.tobytes()).

Query result cache (5 min) -- key on the vector rounded to 3 decimal places; store the serialised hit list as JSON.

Answer cache (30 min) -- only for stable queries; skip for time-relative or user-specific ones.

Nightly full rebuilds break down at scale. The production pattern: re-embed only documents whose content changed.

Content-hash gating -- SHA-256 the text; if the hash matches the stored record, skip re-embedding entirely. When 5% of docs change daily, this cuts ingestion cost by 95%.

import hashlib

def update_document(doc_id, text, collection, registry, chunker, embedder):
    h = hashlib.sha256(text.encode()).hexdigest()
    if doc_id in registry and registry[doc_id]["hash"] == h:
        return  # unchanged
    if doc_id in registry:
        collection.delete(ids=registry[doc_id]["chunk_ids"])
    chunks = chunker(text)
    ids = [f"{doc_id}::{i}" for i in range(len(chunks))]
    collection.add(ids=ids, documents=chunks,
                   embeddings=[embedder(c).tolist() for c in chunks],
                   metadatas=[{"doc_id": doc_id, "i": i} for i in range(len(chunks))])
    registry[doc_id] = {"hash": h, "chunk_ids": ids,
                        "version": registry.get(doc_id, {}).get("version", 0) + 1}

Deletion handling -- skipping leaves deleted docs in the index forever. Call collection.delete(ids=registry[doc_id]["chunk_ids"]) then del registry[doc_id].

Versioning answers the question that has no answer without it: "Last Tuesday's answer was wrong -- what documents did my system retrieve?"

What to version:

Minimal PostgreSQL event log:

CREATE TABLE rag_events (
    event_id BIGSERIAL PRIMARY KEY, occurred_at TIMESTAMPTZ DEFAULT NOW(),
    event_type TEXT NOT NULL, doc_id TEXT, doc_version INT,
    content_hash TEXT, model_id TEXT, chunk_count INT
);
CREATE INDEX ON rag_events (doc_id, occurred_at DESC);

One row per update_document call lets you answer: at time T, what version of doc X was indexed and which model produced its vectors?

Embedding model migration -- new model vectors are geometrically incompatible with old ones. Safe procedure: build a parallel collection, shadow-test at 5%, compare precision@5 and MRR, ramp to 100%, retire the old.

Quality drift kills user trust slowly -- new topics appear in questions before documents are indexed, or popular docs are deleted without replacement.

Expose three Prometheus metrics (scrape with Grafana) and add a 1% faithfulness-sampling loop (LLM-as-judge, Lesson 7) writing scores to a time-series table.

from prometheus_client import Counter, Histogram, Gauge, start_http_server
start_http_server(9000)
LATENCY = Histogram("rag_retrieval_latency_seconds", "Latency",
                    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0])
NO_RESULTS = Counter("rag_no_results_total", "Zero-result queries")
TOP_SCORE  = Gauge("rag_top_chunk_score", "Top chunk cosine similarity")

Wrap retrieval calls: record latency, increment NO_RESULTS when empty, set TOP_SCORE on hits.

Alert thresholds:

Single-node stores become the bottleneck around 1-5 million chunks or hundreds of QPS -- three levers: shard, replicate, batch ingest.

Vector store options at scale:

Batch ingestion -- one-chunk-at-a-time embedding is the most common performance mistake; batch encoding is 10-50x faster.

embed_model = SentenceTransformer("BAAI/bge-small-en-v1.5")  # loaded once

def batch_ingest(chunks, metas, coll, client, bs=256):
    for i in range(0, len(chunks), bs):
        b, m = chunks[i:i+bs], metas[i:i+bs]
        vecs = embed_model.encode(b, batch_size=64, normalize_embeddings=True)
        pts = [PointStruct(id=str(uuid.uuid4()), vector=vecs[j].tolist(),
                           payload={"text": b[j], **m[j]}) for j in range(len(b))]
        client.upsert(collection_name=coll, points=pts, wait=False)

wait=False returns immediately while Qdrant indexes in the background. For pgvector, use USING hnsw ... WITH (m = 16, ef_construction = 64) and tune ef_search per session.

Production RAG has failure modes absent from prototypes.

Graceful degradation -- return a static fallback when retrieval fails or top-chunk score is below threshold:

FALLBACK = "Unable to retrieve relevant documents. Please try again."

def safe_rag_query(query, collection, generate_fn):
    try:
        hits = instrumented_retrieve(query, collection)
        if not hits or hits[0]["score"] < 0.45:  # calibrate via Lesson 7 eval
            return {"answer": FALLBACK, "status": "no_results"}
        return {"answer": answer_with_cache(query, lambda q: generate_fn(q, hits)),
                "status": "ok"}
    except Exception as exc:
        return {"answer": FALLBACK, "status": "error", "detail": str(exc)}