Production RAG: Index Refresh, Caching & a Full Pipeline in Code

Incremental Re-Indexing: Only Update What Changed

A full re-index reprocesses every document from scratch — simple, but wasteful once a collection grows large and only a small fraction of documents actually change between runs. Incremental re-indexing tracks a content hash per document and compares it on each update pass:

import hashlib

def content_hash(text):
    return hashlib.sha256(text.encode()).hexdigest()[:12]

previous_index = {
    "policy.md": content_hash("Remote work is allowed up to 3 days per week."),
    "faq.md": content_hash("Contact support at help@example.com for assistance."),
    "pricing.md": content_hash("The starter plan costs $29 per month."),
}

# New crawl of the same documents -- only policy.md actually changed.
current_documents = {
    "policy.md": "Remote work is allowed up to 4 days per week.",  # CHANGED
    "faq.md": "Contact support at help@example.com for assistance.",
    "pricing.md": "The starter plan costs $29 per month.",
}

def find_changed_documents(previous_index, current_documents):
    changed = []
    for name, text in current_documents.items():
        if previous_index.get(name) != content_hash(text):
            changed.append(name)
    return changed

changed = find_changed_documents(previous_index, current_documents)
print("Documents needing re-embedding:", changed)
# Documents needing re-embedding: ['policy.md']
print("Documents skipped (unchanged):", [d for d in current_documents if d not in changed])
# Documents skipped (unchanged): ['faq.md', 'pricing.md']

Only policy.md — the one document whose content actually differs — gets re-chunked and re-embedded. The untouched documents are skipped entirely, which is exactly what keeps re-indexing fast as a collection scales.

There’s a deeper limitation this doesn’t fully solve: vector similarity itself has no built-in sense of time — nothing about a stored embedding distinguishes a document indexed yesterday from one indexed a year ago. Hash-based re-indexing keeps content accurate; it doesn’t automatically tell a retrieval system to prefer fresher information when several similar-scoring chunks exist.

Semantic Caching: Skipping the Work Entirely

Many real queries repeat, or closely resemble ones already asked. Semantic caching embeds each incoming query and checks it against cached query embeddings — if similarity clears a threshold (commonly 0.85–0.95), the cached response returns immediately, skipping retrieval and generation altogether. One documented system reduced API calls by up to 68.8% using exactly this approach, with a reported cache-hit reliability above 97%.

Source: “GPT Semantic Cache: Reducing LLM Costs and Latency via Semantic Embedding Caching,” arXiv:2411.05276

A Complete Pipeline, End to End

Putting every stage together — ingest, chunk, embed (mocked here, standing in for a real embedding model), retrieve, and assemble the final prompt:

import numpy as np

documents = {
    "battery.md": "The XR-400 sensor operates for approximately 18 months "
                  "on a single CR2032 battery.",
    "connectivity.md": "The XR-400 sensor supports Bluetooth 5.2 connectivity "
                       "with a 30 meter range.",
    "installation.md": "Installation requires no tools and takes under five minutes.",
}

def chunk_document(text, max_words=20):
    words = text.split()
    return [" ".join(words[i:i+max_words]) for i in range(0, len(words), max_words)]

def mock_embed(text):
    if "battery" in text.lower() or "months" in text.lower():
        return np.array([0.9, 0.1, 0.05])
    if "bluetooth" in text.lower() or "connectivity" in text.lower():
        return np.array([0.1, 0.9, 0.05])
    return np.array([0.05, 0.05, 0.9])

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# --- Indexing time ---
index = []
for doc_name, text in documents.items():
    for chunk in chunk_document(text):
        index.append({"source": doc_name, "text": chunk, "embedding": mock_embed(chunk)})
print(f"Indexed {len(index)} chunks from {len(documents)} documents.")
# Indexed 3 chunks from 3 documents.

# --- Query time ---
query = "How long does the battery last?"
query_embedding = mock_embed(query)
scored = sorted(index, key=lambda c: cosine_similarity(query_embedding, c["embedding"]), reverse=True)
top_chunk = scored[0]
print(f"Top retrieved chunk (from {top_chunk['source']}): \"{top_chunk['text']}\"")
# Top retrieved chunk (from battery.md): "The XR-400 sensor operates for
# approximately 18 months on a single CR2032 battery."

prompt = f"""Answer the question using only the context below.

Context: {top_chunk['text']}
Source: {top_chunk['source']}

Question: {query}"""
print(prompt)

Three documents, three chunks, and a query about battery life correctly retrieves the battery chunk — not the connectivity or installation chunks — before the final prompt gets assembled with its source attached, exactly the grounding behavior covered in RAG Advanced.

A Pipeline Failure Checklist

Every stage in this pipeline has its own way of quietly breaking. When a RAG system gives a bad answer, this is roughly where to look first:

Fun Fact

A high semantic-cache hit rate can mask a real problem: if the cache is serving stale answers because the underlying documents changed but the cache wasn’t invalidated, the same freshness gap that affects the vector index affects the cache too — both need their own invalidation strategy, not just one shared one.

Test Yourself