RAG Caching

What to Cache Where

Rule: cache deterministic steps with stable inputs (embedding, exact retrieval) eagerly; cache probabilistic outputs (generation) only with semantic matching and a short TTL.

Hierarchical L1/L2 Cache

L1 exact hash  (Redis GET on sha256(prompt))  -> ~1ms
L2 semantic    (vector similarity in Redis)    -> ~10ms
L3 origin LLM                                  -> 500-3000ms

import hashlib, json, redis
from openai import OpenAI

r = redis.Redis()
oai = OpenAI()

def l1_get(prompt: str):
    return r.get(f"l1:{hashlib.sha256(prompt.encode()).hexdigest()}")

def l1_set(prompt: str, answer: str, ttl: int = 3600):
    r.setex(f"l1:{hashlib.sha256(prompt.encode()).hexdigest()}", ttl, answer)

def serve(prompt: str, semantic_lookup, origin_call):
    if hit := l1_get(prompt):
        return hit.decode(), "L1"
    if hit := semantic_lookup(prompt):   # defined below
        l1_set(prompt, hit)
        return hit, "L2"
    answer = origin_call(prompt)
    l1_set(prompt, answer)
    semantic_index(prompt, answer)
    return answer, "ORIGIN"

Redis-Based Semantic Cache

Requires Redis 7.2+ with the RediSearch module (Redis Stack).

import redis, numpy as np
from redis.commands.search.field import TextField, VectorField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query

r = redis.Redis(decode_responses=False)
DIM = 1536

try:
    r.ft("semcache").create_index(
        fields=[
            TextField("prompt"),
            TextField("answer"),
            VectorField("vec", "HNSW",
                {"TYPE": "FLOAT32", "DIM": DIM, "DISTANCE_METRIC": "COSINE"}),
        ],
        definition=IndexDefinition(prefix=["semcache:"], index_type=IndexType.HASH),
    )
except redis.ResponseError:
    pass  # already exists

def embed(text: str) -> np.ndarray:
    v = oai.embeddings.create(model="text-embedding-3-small", input=text).data[0].embedding
    return np.array(v, dtype=np.float32)

def semantic_get(prompt: str, threshold: float = 0.92):
    v = embed(prompt).tobytes()
    q = (Query("*=>[KNN 1 @vec $vec AS score]")
         .sort_by("score").return_fields("answer", "score").dialect(2))
    res = r.ft("semcache").search(q, query_params={"vec": v})
    if res.docs and (1 - float(res.docs[0].score)) >= threshold:
        return res.docs[0].answer.decode() if isinstance(res.docs[0].answer, bytes) \
               else res.docs[0].answer
    return None

def semantic_put(prompt: str, answer: str, ttl: int = 900):
    key = f"semcache:{hashlib.sha256(prompt.encode()).hexdigest()}"
    r.hset(key, mapping={"prompt": prompt, "answer": answer,
                          "vec": embed(prompt).tobytes()})
    r.expire(key, ttl)

Tune threshold: 0.95+ for factual QA, 0.88-0.92 for chat; log false-positive rate and adjust.

GPTCache

# pip install gptcache
from gptcache import cache
from gptcache.adapter import openai as gpt_openai
from gptcache.embedding import OpenAI as OpenAIEmbed
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation

embedding = OpenAIEmbed()
data_manager = get_data_manager(
    CacheBase("sqlite"),
    VectorBase("faiss", dimension=embedding.dimension),
)
cache.init(
    embedding_func=embedding.to_embeddings,
    data_manager=data_manager,
    similarity_evaluation=SearchDistanceEvaluation(),
)
cache.set_openai_key()

# Drop-in replacement for openai.ChatCompletion
resp = gpt_openai.ChatCompletion.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": user_q}],
    cache_obj=cache,
)

LangChain Cache

from langchain.globals import set_llm_cache
from langchain_community.cache import RedisSemanticCache
from langchain_openai import OpenAIEmbeddings

set_llm_cache(RedisSemanticCache(
    redis_url="redis://localhost:6379",
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    score_threshold=0.1,  # distance (1 - cosine_sim)
))
# All LangChain LLM calls now check cache transparently

Partial Caching (Cache Retrieval, Not Generation)

For chat apps where answers are personalized but the knowledge base is shared, cache retrieval so every user pays only once for the embedding + vector search.

def retrieve_cached(query: str, filters: dict, k: int = 8):
    key = "rtr:" + hashlib.sha256(
        json.dumps({"q": query, "f": filters, "k": k}, sort_keys=True).encode()
    ).hexdigest()
    if cached := r.get(key):
        ids = json.loads(cached)
        return load_chunks_by_id(ids)
    hits = vectorstore.similarity_search_with_score(query, k=k, filter=filters)
    r.setex(key, 600, json.dumps([h[0].metadata["id"] for h in hits]))
    return [h[0] for h in hits]

Anthropic Prompt Caching

Claude supports ephemeral cache breakpoints on system prompts, tools, and any message content block. Cached prefix reads cost ~10% of base input tokens. Minimum prefix: ~1024 tokens (Sonnet/Opus) or ~2048 (Haiku). Default TTL 5 minutes; ttl: "1h" available.

import anthropic
claude = anthropic.Anthropic()

resp = claude.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system=[
        {"type": "text", "text": "You are a support assistant. Follow policy."},
        {
            "type": "text",
            "text": LARGE_POLICY_DOCUMENT,  # 10k+ tokens, reused across requests
            "cache_control": {"type": "ephemeral", "ttl": "1h"},
        },
    ],
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": RETRIEVED_CONTEXT,
                 "cache_control": {"type": "ephemeral"}},  # per-conversation cache
                {"type": "text", "text": user_question},
            ],
        }
    ],
)

# Inspect hits vs misses
print(resp.usage.cache_read_input_tokens, resp.usage.cache_creation_input_tokens)

Ordering matters: put the most stable content first (system instructions), then slower-changing (retrieved context), then the volatile user turn. You can stack up to 4 breakpoints.

OpenAI Prompt Caching

Automatic for prompts >= 1024 tokens on supported models (gpt-4o, gpt-4o-mini, o1, o3, gpt-4.1). No explicit breakpoints; cache key is the exact prefix. Keep static content (system, tools, few-shots) at the front, user input last.

# Inspect cached_tokens in usage
r = oai.chat.completions.create(model="gpt-4o-mini", messages=[...])
print(r.usage.prompt_tokens_details.cached_tokens)

Cache Invalidation

Two strategies; pick one explicitly.

# Strategy A: version prefix — bump on any index change, old keys expire naturally
INDEX_VERSION = "v42"
def key(q): return f"semcache:{INDEX_VERSION}:{sha(q)}"

# Strategy B: tag-based invalidation on doc updates
# When doc_id X changes, delete all cache entries whose retrieval included X
def invalidate_doc(doc_id: str):
    for key in r.sscan_iter(f"cache_by_doc:{doc_id}"):
        r.delete(key)
    r.delete(f"cache_by_doc:{doc_id}")

Tag-based is more precise but requires tracking which cache entries depend on which doc.

Anti-Patterns

Production Checklist

• L1 exact-match cache (Redis hash) enabled
• L2 semantic cache with tuned threshold and per-query-type metrics
• Embedding cache keyed by sha256 + model
• Retrieval cache keyed by query + filter + k
• Provider prompt caching configured (static-first ordering)
• Invalidation strategy chosen (version prefix OR tag-based)
• Hit/miss/layer metrics dashboarded
• PII redaction before caching user inputs
• Cache size budget + eviction policy (LRU/LFU) set