Multi-Layer RAG Pipeline Caching

A RAG pipeline has three expensive stages:

1. Vector search -- embedding the query and searching a vector index for relevant chunks.
2. Prompt compilation -- assembling the chunks, the system prompt, and the user query into a context window.
3. LLM call -- sending the compiled prompt to a model and getting a response.

Each stage runs on every request. For identical or similar queries, you are repeating work that already produced the same result.

With inhouse, you cache each stage independently. When your knowledge base updates, bump a version number and the whole pipeline refreshes naturally.

Layer 1: Cache vector search results

Vector search is the most expensive stage. Same query, same corpus, same results. Cache them.

from inhouse import inhouse_cache
from inhouse.rag import rag_cache

# Layer 1: cache the raw vector search results
@inhouse_cache(ttl_seconds=300)
async def search_chunks(query: str, top_k: int = 5, corpus_version: str = "") -> list[dict]:
    query_embedding = await embed(query)
    results = await vector_index.search(query_embedding, top_k=top_k)
    return [{"text": r.text, "source": r.source, "score": r.score} for r in results]

Pass corpus_version as a hash of your knowledge base state. When you reindex, compute a new hash. All cache keys change. Every query for that corpus misses, and fresh results get cached under the new key.

Layer 2: Cache the compiled prompt

Once you have chunks, you assemble them into a prompt. This varies by user query but often repeats for popular questions.

@rag_cache(ttl_seconds=600)
async def compile_prompt(
    query: str,
    chunks: list[dict],
    system_style: str = "concise",
) -> str:
    context = "\n\n".join(
        f"[Source: {c['source']}]\n{c['text']}" for c in chunks
    )
    return f"""You are a helpful assistant. Be {system_style}.

Context:
{context}

Question: {query}
"""

The chunks argument is a list of dicts from the search layer. inhouse's recursive key freezing handles lists of dicts deterministically, so the same chunks in the same order produce the same cache key.

Layer 3: Cache the LLM response (optional)

If you have queries that repeat identically (FAQs, common support questions), cache the final LLM response too.

@rag_cache(ttl_seconds=120)
async def answer_question(
    compiled_prompt: str,
    model: str = "gpt-4",
    temperature: float = 0.3,
) -> str:
    response = await llm_client.complete(
        model=model,
        messages=[{"role": "user", "content": compiled_prompt}],
        temperature=temperature,
    )
    return response.choices[0].message.content

Three layers of caching. The first hit runs everything. The second hit for the same query skips the LLM call. A similar query with different chunks skips vector search but recompiles the prompt.

Tying it together in a FastAPI endpoint

from fastapi import FastAPI
from inhouse import MemoryStore
from inhouse.fastapi import create_lifespan

app = FastAPI(lifespan=create_lifespan(MemoryStore(default_ttl=60)))

@app.post("/rag/ask")
async def ask(query: str, kb_version: str = ""):
    chunks = await search_chunks(query, corpus_version=kb_version)
    prompt = await compile_prompt(query, chunks)
    answer = await answer_question(prompt)
    return {"answer": answer, "sources": [c["source"] for c in chunks]}

Context versioning

When you reindex your knowledge base, everything downstream needs to refresh. The corpus_version argument makes this automatic.

# After a document ingestion run:
NEW_VERSION = hashlib.sha256(open("corpus.db", "rb").read()).hexdigest()[:12]

# Every subsequent request with this version gets fresh results.
# Requests without a version string or with an old version also miss.

Because corpus_version is part of the cache key for search_chunks, changing it invalidates that layer. The downstream layers (compile_prompt and answer_question) depend on the chunk data and compiled prompt, so they cascade naturally -- different chunks means different cache keys for those layers too.

If you need to evict everything at once

search_chunks.cache_clear()
compile_prompt.cache_clear()
answer_question.cache_clear()

Call this after a reindex if you want to skip the versioning approach and just nuke all cached RAG data. The next request for each query repopulates the cache fresh.

The cache independence rule

Each layer has its own TTL. Vector search results are cached for 5 minutes. Compiled prompts for 10 minutes. LLM responses for 2 minutes. Short TTLs on the expensive outer layers, longer on the cheaper inner layers.

Layer	TTL	Why
Vector search	300s	Most expensive, cache longest
Prompt compilation	600s	Medium cost, medium cache
LLM response	120s	Least expensive, shortest cache

Adjust based on your traffic and how often your corpus changes. If you never reindex, crank all TTLs up. If you reindex every minute, use corpus_version exclusively and set TTLs to a large value.