RAG Production

RAG in Production

Incremental Indexing

Never reindex the whole corpus on every change. Three trigger sources:

1. Change Data Capture (CDC) from the source database (Debezium, native Postgres logical replication).
2. Document diffing on filesystem/S3: store sha256 + last_modified per doc; reprocess when changed.
3. Event-driven: producer emits doc.created | doc.updated | doc.deleted on a queue.

# Doc diffing example
import hashlib, json, pathlib

STATE = pathlib.Path(".index_state.json")
state = json.loads(STATE.read_text()) if STATE.exists() else {}

def scan_and_upsert(root: pathlib.Path):
    seen = set()
    for p in root.rglob("*.md"):
        content = p.read_bytes()
        h = hashlib.sha256(content).hexdigest()
        key = str(p)
        seen.add(key)
        if state.get(key) != h:
            reindex_document(key, content.decode())
            state[key] = h
    for key in list(state.keys()) - seen:
        delete_from_index(key)
        state.pop(key, None)
    STATE.write_text(json.dumps(state))

At chunk level, store doc_id + chunk_hash; on doc update, delete chunks where chunk_hash is no longer present, insert new ones.

Blue-Green Re-indexing (Full Rebuild)

Required when changing embedding model, chunking strategy, or schema. Never rebuild in place.

# Pinecone: use different index names and swap at the application layer
# Weaviate/Qdrant/Elasticsearch: use index aliases

# Elasticsearch
client.indices.create(index="docs_v42", body=mapping)
# ... reindex into docs_v42 ...
client.indices.update_aliases(actions=[
    {"remove": {"index": "docs_v41", "alias": "docs"}},
    {"add":    {"index": "docs_v42", "alias": "docs"}},
])
# Keep docs_v41 around for fast rollback, delete after N days

Validation gate before switching the alias: run a golden eval set against both indexes, require new index recall@5 >= old - 1%.

Hot-Swap of Embedding Models

Changing embedding model invalidates the entire index. Procedure:

1. Build index_v{new} offline with new embeddings.
2. Shadow: send 10% of live queries to both indexes, log differences.
3. Gate on eval metrics + shadow diff rate.
4. Flip alias; keep old index warm for 72h.
5. Invalidate semantic caches (they contain old embeddings).

Track embedding_model_version in .dev-suite-manifest.json style metadata alongside the index.

Index Versioning and Aliases

Always deploy through the alias/env var. Never hardcode index names in application code.

Cost Optimization

Matryoshka Embedding Truncation

OpenAI text-embedding-3-* and Nomic nomic-embed-text-v1.5 expose dimensions parameter; truncate then L2-renormalize.

import numpy as np
def truncate_renorm(v: list[float], d: int = 512) -> list[float]:
    arr = np.array(v[:d], dtype=np.float32)
    return (arr / np.linalg.norm(arr)).tolist()

SLA Targets

Measure per stage. Streaming (SSE) the LLM response lets you hide a 2s generation behind first-token latency of ~400ms.

Capacity Planning

Simple model for ANN capacity:

memory_bytes ~= num_vectors * (4 * dim + 24)   # float32 + HNSW overhead
                + index_metadata (~20%)

QPS_per_replica ~= 1000ms / p95_query_ms * parallelism_factor(0.6-0.8)
replicas = ceil(target_qps / QPS_per_replica) + 1 (headroom)

Rules of thumb:

• Keep per-replica index size < 50 GB for in-memory HNSW (else shard).
• Plan for 2x query volume spikes (traffic + rebuild catch-up).
• Reserve 30% CPU headroom on embedding/rerank servers; GPU utilization target 60-70%.

Horizontal Scaling Patterns

Multi-Tenant Strategies

Strategy        Isolation    Cost     Use case
-----------------------------------------------------------
Per-tenant DB   strongest    $$$$     Regulated / sovereignty
Namespace       strong       $$       Most SaaS B2B
Filter only     weakest      $        Small tenants, shared budget

# Pinecone namespace pattern
pc_index.namespace(f"tenant_{tenant_id}").query(vector=v, top_k=5)

# Metadata filter pattern (avoid for >100 tenants due to selectivity)
pc_index.query(vector=v, top_k=5, filter={"tenant_id": tenant_id})

# Row-level security in pgvector
# Role granted SELECT ... USING (tenant_id = current_setting('app.tenant_id')::int)

Never rely on filter-only isolation when tenants have very different document counts; rebalance to namespaces.

Async Ingestion Pipelines

Source -> CDC/producer -> Kafka/Pub-Sub -> Ingest worker
  -> Chunk -> Embed (batched) -> Upsert -> Update manifest
  -> Publish `index.updated` event for cache invalidation

# Example Kafka worker skeleton
from aiokafka import AIOKafkaConsumer
import asyncio, orjson

async def run():
    consumer = AIOKafkaConsumer("docs.events",
        bootstrap_servers="kafka:9092", group_id="rag-ingest",
        enable_auto_commit=False)
    await consumer.start()
    try:
        batch = []
        async for msg in consumer:
            batch.append(orjson.loads(msg.value))
            if len(batch) >= 32:
                await ingest_batch(batch)
                await consumer.commit()
                batch.clear()
    finally:
        await consumer.stop()

Key properties:

• At-least-once with idempotent upserts (keyed by doc_id + chunk_idx).
• DLQ for malformed docs; alert if DLQ > N/hour.
• Backpressure: pause consumption if embedding API is rate-limited.
• Ordering: partition by doc_id so updates to the same doc don't race.

Embedding API Rate-Limit Handling

import asyncio, random
from openai import AsyncOpenAI, RateLimitError

async def embed_batch(texts, client: AsyncOpenAI, max_retries=6):
    for attempt in range(max_retries):
        try:
            r = await client.embeddings.create(model="text-embedding-3-small", input=texts)
            return [d.embedding for d in r.data]
        except RateLimitError:
            await asyncio.sleep((2 ** attempt) + random.random())
    raise RuntimeError("embedding rate-limit budget exhausted")

Anti-Patterns

Production Checklist

• Incremental ingestion (CDC or doc diff) wired
• Blue-green reindex tested in staging
• Index alias used by all readers (no hardcoded names)
• Manifest records embedding model, chunk params, index version
• Per-stage latency budgets defined and measured
• Multi-tenant isolation strategy chosen and enforced in tests
• Async ingest DLQ + alerts configured
• Embedding / rerank / LLM rate-limit retry with backoff
• Rollback runbook (swap alias back, invalidate cache)
• Cost dashboards per stage (embed, vector, rerank, generation)