Your customer support chatbot is getting hammered. Users are asking "How do I reset my password?" in seventeen different ways — "forgot password," "can't log in," "password reset steps," "how to change my password" — and each variation is burning a fresh API call to your LLM provider. At $0.002 per thousand tokens and 50,000 daily queries, you're spending $3,000 a month on answers your system has already computed. Worse, each cold call adds 800ms to 2 seconds of latency. Your users notice.
The obvious fix is caching. The non-obvious problem is that LLM queries don't cache like database lookups. "How do I reset my password?" and "I forgot my password, what do I do?" are semantically identical, but they'll never match on a string equality check. You need a smarter approach — one that understands meaning, not just characters.
By the end of this lesson, you'll have a production-grade LLM caching layer built in Redis that performs semantic deduplication using vector embeddings, applies differentiated TTL strategies based on content volatility, and handles cache invalidation without nuking everything when your underlying data changes.
What you'll learn:
You should be comfortable with:
asyncio throughout)You'll need Redis Stack (which includes RediSearch and RedisJSON) running locally or via Redis Cloud. The easiest local setup: docker run -d -p 6379:6379 redis/redis-stack-server:latest.
Before writing a line of code, let's be precise about what problem we're solving and why the naive approach collapses immediately.
Exact-match caching works beautifully for deterministic systems. SQL query results, REST API responses, rendered HTML — these are keyed on a canonical string representation of the request. Two requests are either identical or they aren't.
LLM queries live in continuous semantic space. The distance between "How do I reset my password?" and "I forgot my password, what do I do?" is nearly zero in meaning, but infinite by string comparison. If you key your cache on the raw query string (or even its hash), you get a cache miss on every paraphrase, reformulation, and minor variation your users produce — which is basically everything.
Semantic caching solves this by converting queries to embedding vectors and storing those vectors in the cache alongside the response. When a new query arrives, you:
The critical insight is that your cache key is no longer a string — it's a point in high-dimensional space, and "cache hit" becomes "close enough in that space."
This creates a new set of decisions that exact-match caching doesn't require: What similarity threshold is "close enough"? Which embedding model should you use? How do you handle queries that are semantically similar but require different answers due to context? We'll work through each of these.
Redis Stack extends vanilla Redis with RediSearch, which supports vector similarity search. This is what lets us do nearest-neighbor lookup efficiently instead of scanning every cached embedding linearly.
Start with your dependencies:
pip install redis openai numpy scikit-learn python-dotenv tiktoken
Now let's establish the foundational connection and index setup:
import os
import json
import time
import hashlib
import numpy as np
from typing import Optional, Any
from redis import Redis
from redis.commands.search.field import VectorField, TextField, TagField, NumericField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from openai import AsyncOpenAI
import asyncio
from dotenv import load_dotenv
load_dotenv()
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
redis_client = Redis.from_url(REDIS_URL, decode_responses=False)
openai_client = AsyncOpenAI(api_key=OPENAI_API_KEY)
CACHE_INDEX_NAME = "llm_cache_idx"
CACHE_KEY_PREFIX = "llm_cache:"
EMBEDDING_DIM = 1536 # text-embedding-3-small dimension
Now create the RediSearch index. This is the step most tutorials skip over and then you wonder why vector search returns nothing:
def create_cache_index():
"""
Create the RediSearch index for semantic cache lookup.
Must be called once at startup. Safe to call repeatedly —
we catch the "index already exists" error.
"""
try:
schema = (
TextField("$.query_text", as_name="query_text"),
VectorField(
"$.embedding",
"HNSW", # Hierarchical Navigable Small World — fast approximate NN
{
"TYPE": "FLOAT32",
"DIM": EMBEDDING_DIM,
"DISTANCE_METRIC": "COSINE",
"INITIAL_CAP": 10000,
"M": 16, # connections per node — higher = more accurate, more memory
"EF_CONSTRUCTION": 200, # build-time accuracy vs speed tradeoff
},
as_name="embedding",
),
TagField("$.cache_tags", as_name="cache_tags", separator="|"),
NumericField("$.created_at", as_name="created_at"),
NumericField("$.hit_count", as_name="hit_count"),
TextField("$.model", as_name="model"),
)
redis_client.ft(CACHE_INDEX_NAME).create_index(
schema,
definition=IndexDefinition(
prefix=[CACHE_KEY_PREFIX], index_type=IndexType.JSON
),
)
print(f"Cache index '{CACHE_INDEX_NAME}' created successfully.")
except Exception as e:
if "Index already exists" in str(e):
print(f"Cache index '{CACHE_INDEX_NAME}' already exists — skipping creation.")
else:
raise
A few things worth understanding about the HNSW configuration: HNSW (Hierarchical Navigable Small World) is an approximate nearest neighbor algorithm. "Approximate" sounds scary but in practice at reasonable M and EF_CONSTRUCTION values, it finds the true nearest neighbor over 99% of the time while being orders of magnitude faster than exact search. For a cache with 100,000 entries, exact search would require computing cosine similarity against every stored vector. HNSW narrows the search to a tiny fraction of those comparisons.
Your embedding model choice matters more than most tutorials admit. Here's why: if you use different embedding models for caching and for any downstream retrieval you're doing (RAG, etc.), you're creating invisible correctness bugs. Use one embedding model consistently across your pipeline.
async def get_embedding(text: str) -> list[float]:
"""
Generate an embedding vector for the given text.
We strip and normalize whitespace before embedding.
"How do I reset my password?" and "How do I reset my password? "
should not produce cache misses due to trailing spaces.
"""
normalized = " ".join(text.strip().split())
response = await openai_client.embeddings.create(
model="text-embedding-3-small",
input=normalized,
encoding_format="float"
)
return response.data[0].embedding
def embedding_to_bytes(embedding: list[float]) -> bytes:
"""Convert embedding list to bytes for Redis storage."""
return np.array(embedding, dtype=np.float32).tobytes()
def bytes_to_embedding(b: bytes) -> list[float]:
"""Convert bytes back to embedding list."""
return np.frombuffer(b, dtype=np.float32).tolist()
Performance note: Embedding API calls add latency too — typically 50-150ms. For a cache lookup, you always pay this cost. Make sure your cache hit rate is high enough that the overall system is still faster. In most customer support / FAQ scenarios, hit rates above 60% make semantic caching a clear win.
Now we build the actual cache read/write logic. The write path is straightforward — embed the query, store everything in a Redis JSON document. The read path is where semantic matching happens.
async def cache_write(
query: str,
response: str,
model: str,
ttl_seconds: int,
cache_tags: list[str] = None,
) -> str:
"""
Store an LLM response in the semantic cache.
Returns the cache key for the stored entry.
"""
embedding = await get_embedding(query)
cache_key = f"{CACHE_KEY_PREFIX}{hashlib.sha256(query.encode()).hexdigest()[:16]}"
document = {
"query_text": query,
"response": response,
"model": model,
"embedding": embedding,
"cache_tags": "|".join(cache_tags or []),
"created_at": time.time(),
"hit_count": 0,
"ttl_seconds": ttl_seconds,
}
# Store as JSON document with TTL
redis_client.json().set(cache_key, "$", document)
redis_client.expire(cache_key, ttl_seconds)
return cache_key
async def cache_read(
query: str,
similarity_threshold: float = 0.92,
model_filter: Optional[str] = None,
) -> Optional[dict]:
"""
Look up a query in the semantic cache.
Returns the cached document if a sufficiently similar query exists,
or None if no match is found.
The similarity_threshold is the minimum cosine similarity score
required to consider a cache entry a hit. Range: 0.0-1.0.
Higher = more conservative (fewer hits, but higher quality matches).
"""
embedding = await get_embedding(query)
embedding_bytes = embedding_to_bytes(embedding)
# Build the KNN vector search query
# We ask for top-3 candidates and filter by threshold ourselves
# because RediSearch's range filter on vector distance can be finicky
base_query = "*"
if model_filter:
base_query = f"@model:{{{model_filter}}}"
knn_query = (
Query(f"({base_query})=>[KNN 3 @embedding $vec AS vector_score]")
.sort_by("vector_score")
.return_fields("vector_score", "query_text", "$.response", "$.cache_tags", "$.hit_count", "$.created_at")
.dialect(2)
)
results = redis_client.ft(CACHE_INDEX_NAME).search(
knn_query,
query_params={"vec": embedding_bytes}
)
if not results.docs:
return None
# RediSearch returns COSINE distance (0 = identical, 2 = opposite)
# Convert to similarity score: similarity = 1 - (distance / 2)
best_match = results.docs[0]
distance = float(best_match.vector_score)
similarity = 1 - (distance / 2)
if similarity < similarity_threshold:
return None # Best match isn't good enough
# Increment hit count in background (don't await — fire and forget)
cache_key = best_match.id
asyncio.create_task(
_increment_hit_count(cache_key)
)
return {
"response": best_match.__dict__.get("$.response"),
"similarity": similarity,
"original_query": best_match.query_text,
"hit_count": int(best_match.__dict__.get("$.hit_count", 0)),
"cache_key": cache_key,
}
async def _increment_hit_count(cache_key: str):
"""Increment hit count for cache analytics."""
try:
redis_client.json().numincrby(cache_key, "$.hit_count", 1)
except Exception:
pass # Non-critical — don't let analytics failures affect the user path
Common mistake: RediSearch returns cosine distance (0 to 2), not cosine similarity (−1 to 1). The conversion is
similarity = 1 - (distance / 2). Skipping this step and comparing raw distance against a similarity threshold causes baffling behavior — you'll reject excellent matches and accept terrible ones.
This is where most caching implementations leave significant value on the table. A flat TTL applied to everything — say, 24 hours — is wrong in both directions simultaneously. Your "What is a P-value?" response is still accurate in a week. Your "What's our current promotional discount?" response might be stale in four hours.
The right approach is to classify your query types by content volatility and assign TTL tiers accordingly.
from enum import Enum
from dataclasses import dataclass
class ContentVolatility(Enum):
STATIC = "static" # Timeless facts, definitions, explanations
SEMI_STATIC = "semi_static" # Policies, procedures (change monthly/quarterly)
DYNAMIC = "dynamic" # Pricing, inventory, schedules (change daily/weekly)
EPHEMERAL = "ephemeral" # Session-specific, personalized, real-time data
@dataclass
class TTLConfig:
volatility: ContentVolatility
ttl_seconds: int
tags: list[str]
# TTL configuration by volatility tier
TTL_TIERS = {
ContentVolatility.STATIC: TTLConfig(
volatility=ContentVolatility.STATIC,
ttl_seconds=60 * 60 * 24 * 30, # 30 days
tags=["static"],
),
ContentVolatility.SEMI_STATIC: TTLConfig(
volatility=ContentVolatility.SEMI_STATIC,
ttl_seconds=60 * 60 * 24 * 7, # 7 days
tags=["semi_static"],
),
ContentVolatility.DYNAMIC: TTLConfig(
volatility=ContentVolatility.DYNAMIC,
ttl_seconds=60 * 60 * 4, # 4 hours
tags=["dynamic"],
),
ContentVolatility.EPHEMERAL: TTLConfig(
volatility=ContentVolatility.EPHEMERAL,
ttl_seconds=60 * 15, # 15 minutes
tags=["ephemeral"],
),
}
async def classify_query_volatility(query: str, context: dict = None) -> ContentVolatility:
"""
Classify a query's content volatility using a fast LLM call.
This costs a small amount of tokens but protects you from caching
volatile content for too long. Use a cheap/fast model here.
"""
classification_prompt = f"""Classify this user query into exactly one volatility category.
Query: "{query}"
Categories:
- STATIC: Timeless facts, concepts, definitions, how-to explanations (e.g., "What is compound interest?", "How does OAuth work?")
- SEMI_STATIC: Company policies, processes, documentation that changes infrequently (e.g., "What's your return policy?", "How do I submit an expense report?")
- DYNAMIC: Pricing, availability, schedules, promotions that change frequently (e.g., "What's the current price of X?", "Is Y in stock?")
- EPHEMERAL: Personalized, session-specific, or real-time queries (e.g., "What's my account balance?", "Show me my recent orders")
Respond with exactly one word: STATIC, SEMI_STATIC, DYNAMIC, or EPHEMERAL."""
response = await openai_client.chat.completions.create(
model="gpt-4o-mini", # Fast and cheap for classification
messages=[{"role": "user", "content": classification_prompt}],
max_tokens=10,
temperature=0,
)
label = response.choices[0].message.content.strip().upper()
volatility_map = {
"STATIC": ContentVolatility.STATIC,
"SEMI_STATIC": ContentVolatility.SEMI_STATIC,
"DYNAMIC": ContentVolatility.DYNAMIC,
"EPHEMERAL": ContentVolatility.EPHEMERAL,
}
return volatility_map.get(label, ContentVolatility.DYNAMIC) # Default to DYNAMIC (conservative)
One subtle issue: the classification call itself has latency. You can avoid this on the read path (don't classify before reading — just look up) and only classify on cache misses before writing. The sequence is:
async def get_llm_response_with_caching(
query: str,
system_prompt: str,
domain_tags: list[str] = None,
similarity_threshold: float = 0.92,
) -> dict:
"""
Main entry point: returns LLM response, served from cache when possible.
"""
# Step 1: Try cache
cached = await cache_read(query, similarity_threshold=similarity_threshold)
if cached:
return {
"response": cached["response"],
"source": "cache",
"similarity": cached["similarity"],
"original_query": cached["original_query"],
"latency_note": "Served from semantic cache",
}
# Step 2: Cache miss — call LLM and classify volatility in parallel
llm_task = asyncio.create_task(
openai_client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": query},
],
temperature=0.3,
)
)
classification_task = asyncio.create_task(
classify_query_volatility(query)
)
llm_response, volatility = await asyncio.gather(llm_task, classification_task)
response_text = llm_response.choices[0].message.content
# Step 3: Write to cache with appropriate TTL
ttl_config = TTL_TIERS[volatility]
all_tags = list(set((domain_tags or []) + ttl_config.tags))
# Don't cache ephemeral responses — they're personalized
if volatility != ContentVolatility.EPHEMERAL:
await cache_write(
query=query,
response=response_text,
model="gpt-4o",
ttl_seconds=ttl_config.ttl_seconds,
cache_tags=all_tags,
)
return {
"response": response_text,
"source": "llm",
"volatility": volatility.value,
"ttl_seconds": ttl_config.ttl_seconds if volatility != ContentVolatility.EPHEMERAL else 0,
}
Phil Karlton's famous quip that "there are only two hard things in computer science: cache invalidation and naming things" applies with full force here.
When your product pricing changes, or your support documentation is updated, or a bug is found in a response that has been served thousands of times — you need surgical invalidation, not a full cache wipe.
The approach we'll use is tag-based invalidation. Every cache entry is tagged at write time with relevant domain identifiers. When a content source changes, you invalidate by tag rather than by key.
# Tag taxonomy for a SaaS customer support system:
#
# Domain tags: "domain:billing", "domain:security", "domain:onboarding"
# Product tags: "product:enterprise", "product:starter"
# Content tags: "content:pricing", "content:policy", "content:tutorial"
# Freshness tags: "static", "semi_static", "dynamic", "ephemeral"
async def invalidate_by_tag(tag: str) -> int:
"""
Invalidate all cache entries with the given tag.
Returns the number of entries invalidated.
This is O(n) in the number of matching entries, not O(total cache size).
For production, run this as a background task during off-peak hours
or use a cursor-based approach for large invalidation sets.
"""
# Search for all documents with this tag
tag_query = Query(f"@cache_tags:{{{tag}}}").return_fields("$.cache_tags").paging(0, 10000)
results = redis_client.ft(CACHE_INDEX_NAME).search(tag_query)
if not results.docs:
return 0
# Delete matched keys in a pipeline for efficiency
pipeline = redis_client.pipeline()
for doc in results.docs:
pipeline.delete(doc.id)
pipeline.execute()
invalidated_count = len(results.docs)
print(f"Invalidated {invalidated_count} cache entries with tag '{tag}'")
return invalidated_count
async def invalidate_by_multiple_tags(tags: list[str], operator: str = "OR") -> int:
"""
Invalidate entries matching any (OR) or all (AND) of the given tags.
Use OR to invalidate broadly (e.g., everything in billing OR pricing).
Use AND to invalidate precisely (e.g., only entries tagged both
'domain:billing' AND 'product:enterprise').
"""
if operator == "OR":
tag_filter = "|".join(tags)
query_str = f"@cache_tags:{{{tag_filter}}}"
else: # AND
query_str = " ".join([f"@cache_tags:{{{tag}}}" for tag in tags])
tag_query = Query(query_str).return_fields("$.cache_tags").paging(0, 10000)
results = redis_client.ft(CACHE_INDEX_NAME).search(tag_query)
pipeline = redis_client.pipeline()
for doc in results.docs:
pipeline.delete(doc.id)
pipeline.execute()
return len(results.docs)
async def invalidate_stale_entries(max_age_seconds: int) -> int:
"""
Invalidate cache entries older than max_age_seconds, regardless of TTL.
Useful for emergency invalidation when you need to be sure nothing
older than X hours is being served, even if TTL hasn't expired.
Note: Redis TTL handles the normal expiry path. This is for
cases where you want to force-expire based on content age,
not the original TTL setting.
"""
cutoff_timestamp = time.time() - max_age_seconds
age_query = (
Query(f"@created_at:[-inf ({cutoff_timestamp}]")
.return_fields("$.created_at")
.paging(0, 10000)
)
results = redis_client.ft(CACHE_INDEX_NAME).search(age_query)
pipeline = redis_client.pipeline()
for doc in results.docs:
pipeline.delete(doc.id)
pipeline.execute()
return len(results.docs)
Now let's wire invalidation into a realistic event-driven scenario. Your content management system publishes events when documentation changes:
import asyncio
from dataclasses import dataclass
@dataclass
class ContentUpdateEvent:
content_type: str # "pricing", "policy", "tutorial", etc.
affected_products: list[str]
affected_domains: list[str]
updated_by: str
async def handle_content_update(event: ContentUpdateEvent):
"""
Event handler that invalidates cache entries affected by a content update.
This is where your tag taxonomy earns its keep. A single pricing update
might invalidate hundreds of cached responses, but leave thousands of
tutorial and policy responses untouched.
"""
tags_to_invalidate = []
# Add content-type tag
tags_to_invalidate.append(f"content:{event.content_type}")
# Add product-specific tags
for product in event.affected_products:
tags_to_invalidate.append(f"product:{product}")
# Add domain tags
for domain in event.affected_domains:
tags_to_invalidate.append(f"domain:{domain}")
# Pricing changes should also invalidate dynamic-tier entries
if event.content_type == "pricing":
tags_to_invalidate.append("dynamic")
total_invalidated = 0
for tag in tags_to_invalidate:
count = await invalidate_by_tag(tag)
total_invalidated += count
print(
f"Content update by {event.updated_by}: "
f"invalidated ~{total_invalidated} cache entries "
f"across {len(tags_to_invalidate)} tag groups"
)
# Note: total_invalidated may double-count entries with multiple tags.
# For an exact count, use a set union approach or the AND query pattern.
# Example usage:
# await handle_content_update(ContentUpdateEvent(
# content_type="pricing",
# affected_products=["enterprise", "starter"],
# affected_domains=["billing"],
# updated_by="pricing-team",
# ))
Warning: Don't design a system where cache invalidation is triggered by the LLM pipeline itself. Invalidation should be triggered by the systems that own the data — your CMS, your pricing service, your policy management tool. LLM pipelines that invalidate their own caches create circular dependencies that are genuinely difficult to debug.
A semantic cache without instrumentation is flying blind. You need to know your hit rate, your threshold's effect on quality, and whether your TTL tiers are set correctly.
async def get_cache_stats() -> dict:
"""
Return aggregate statistics for the semantic cache.
"""
# Total documents in index
index_info = redis_client.ft(CACHE_INDEX_NAME).info()
total_docs = int(index_info.get("num_docs", 0))
# Sum hit counts across all entries
# In production, maintain these counters separately in a Redis hash
# for efficiency — scanning the full index is expensive at scale
stats_query = (
Query("*")
.return_fields("$.hit_count", "$.created_at", "$.cache_tags")
.paging(0, 10000)
)
results = redis_client.ft(CACHE_INDEX_NAME).search(stats_query)
total_hits = 0
volatility_distribution = {"static": 0, "semi_static": 0, "dynamic": 0, "ephemeral": 0}
for doc in results.docs:
hit_count = doc.__dict__.get("$.hit_count", 0)
total_hits += int(hit_count) if hit_count else 0
tags = doc.__dict__.get("$.cache_tags", "")
for volatility_tier in volatility_distribution:
if volatility_tier in (tags or ""):
volatility_distribution[volatility_tier] += 1
return {
"total_cached_entries": total_docs,
"total_cache_hits_served": total_hits,
"average_hits_per_entry": round(total_hits / max(total_docs, 1), 2),
"volatility_distribution": volatility_distribution,
}
async def analyze_threshold_sensitivity(
test_queries: list[tuple[str, str]], # (query, expected_match_query) pairs
thresholds: list[float] = None,
) -> dict:
"""
Analyze how different similarity thresholds affect hit rate and quality.
test_queries: pairs of semantically equivalent queries
Each pair represents a query variation you'd want to be a cache hit.
Run this offline to calibrate your threshold before deploying.
"""
if thresholds is None:
thresholds = [0.85, 0.88, 0.90, 0.92, 0.95, 0.97]
results = {}
# Get embeddings for all queries
embeddings = {}
for query, expected in test_queries:
if query not in embeddings:
embeddings[query] = await get_embedding(query)
if expected not in embeddings:
embeddings[expected] = await get_embedding(expected)
for threshold in thresholds:
hits = 0
misses = 0
for query, expected in test_queries:
vec_a = np.array(embeddings[query])
vec_b = np.array(embeddings[expected])
# Compute cosine similarity
similarity = float(
np.dot(vec_a, vec_b) / (np.linalg.norm(vec_a) * np.linalg.norm(vec_b))
)
if similarity >= threshold:
hits += 1
else:
misses += 1
total = hits + misses
results[threshold] = {
"hit_rate": round(hits / total, 3),
"hits": hits,
"misses": misses,
}
return results
Here's a practical test query set to run before deploying to production:
# Run this offline to calibrate your threshold
sensitivity_test_queries = [
("How do I reset my password?", "I forgot my password, what do I do?"),
("How do I reset my password?", "Steps to recover account access"),
("What's your refund policy?", "Can I get my money back?"),
("What's your refund policy?", "Return and refund information"),
("How do I cancel my subscription?", "I want to cancel my account"),
("How do I upgrade my plan?", "I need a higher tier subscription"),
# These should NOT match — semantically different
("How do I reset my password?", "What is two-factor authentication?"),
("What's your refund policy?", "How do I upgrade my plan?"),
]
# The last two are semantically different — verify they score below your threshold
# If they don't, you need a higher threshold
Now let's put it all together in a realistic scenario. You're building a caching layer for a SaaS company's customer support bot. The bot handles three query categories:
Your task: Implement and test the full caching pipeline, then run a simulation to measure cache effectiveness.
import random
import asyncio
import time
# Simulated query corpus — variations of the same underlying questions
BILLING_QUERIES = [
"What payment methods do you accept?",
"Which payment options are available?",
"Can I pay with PayPal?",
"Do you accept credit cards?",
"What are my payment options?",
]
TECHNICAL_QUERIES = [
"How do I connect to the API?",
"What's the process for API integration?",
"Show me how to authenticate with your API",
"API connection instructions",
"How do I get started with the REST API?",
]
EPHEMERAL_QUERIES = [
"What's my current usage this month?",
"How many API calls have I made today?",
"Show me my account activity",
"What's my billing status?",
]
SYSTEM_PROMPT = """You are a helpful customer support agent for Acme SaaS Platform.
Answer questions clearly and concisely. For account-specific questions,
explain that you cannot access individual account data and direct users
to their dashboard."""
async def run_cache_simulation():
"""
Simulate 50 support queries and measure cache performance.
"""
create_cache_index()
# Build a combined query pool — 70% repeats/variations, 30% novel
query_pool = (
BILLING_QUERIES * 4 + # High frequency
TECHNICAL_QUERIES * 4 + # High frequency
EPHEMERAL_QUERIES * 2 # Medium frequency — but shouldn't cache
)
random.shuffle(query_pool)
test_queries = query_pool[:50]
results = {"cache_hits": 0, "llm_calls": 0, "ephemeral_skipped": 0}
latencies = {"cache": [], "llm": []}
domain_tag_map = {
"payment": ["domain:billing", "content:pricing"],
"API": ["domain:technical", "content:tutorial"],
"api": ["domain:technical", "content:tutorial"],
"usage": [], # Ephemeral — no domain tags matter
"account": [],
"billing": ["domain:billing"],
}
for query in test_queries:
start = time.time()
# Determine domain tags from query keywords
tags = []
for keyword, tag_list in domain_tag_map.items():
if keyword.lower() in query.lower():
tags.extend(tag_list)
tags = list(set(tags))
result = await get_llm_response_with_caching(
query=query,
system_prompt=SYSTEM_PROMPT,
domain_tags=tags,
similarity_threshold=0.91,
)
elapsed = time.time() - start
if result["source"] == "cache":
results["cache_hits"] += 1
latencies["cache"].append(elapsed)
print(f"CACHE HIT ({elapsed:.2f}s, sim={result['similarity']:.3f}): {query[:60]}")
elif result.get("ttl_seconds") == 0:
results["ephemeral_skipped"] += 1
latencies["llm"].append(elapsed)
print(f"EPHEMERAL ({elapsed:.2f}s): {query[:60]}")
else:
results["llm_calls"] += 1
latencies["llm"].append(elapsed)
print(f"LLM CALL ({elapsed:.2f}s, volatility={result.get('volatility')}): {query[:60]}")
# Small delay to avoid rate limiting
await asyncio.sleep(0.1)
# Print summary
total = len(test_queries)
cacheable = total - results["ephemeral_skipped"]
hit_rate = results["cache_hits"] / max(cacheable, 1)
avg_cache_latency = sum(latencies["cache"]) / max(len(latencies["cache"]), 1)
avg_llm_latency = sum(latencies["llm"]) / max(len(latencies["llm"]), 1)
print(f"\n{'='*50}")
print(f"Cache Simulation Results ({total} queries)")
print(f"{'='*50}")
print(f"Cache hits: {results['cache_hits']} ({hit_rate:.1%} of cacheable queries)")
print(f"LLM calls: {results['llm_calls']}")
print(f"Ephemeral (no-cache): {results['ephemeral_skipped']}")
print(f"Avg cache latency: {avg_cache_latency:.3f}s")
print(f"Avg LLM latency: {avg_llm_latency:.3f}s")
print(f"Latency improvement: {avg_llm_latency / max(avg_cache_latency, 0.001):.1f}x faster from cache")
stats = await get_cache_stats()
print(f"\nCache stats: {json.dumps(stats, indent=2)}")
if __name__ == "__main__":
asyncio.run(run_cache_simulation())
Expected output after the simulation warms up:
LLM CALL (1.43s, volatility=semi_static): What payment methods do you accept?
LLM CALL (1.21s, volatility=static): How do I connect to the API?
CACHE HIT (0.18s, sim=0.963): Which payment options are available?
CACHE HIT (0.16s, sim=0.951): Can I pay with PayPal?
EPHEMERAL (1.38s): What's my current usage this month?
CACHE HIT (0.15s, sim=0.944): What's the process for API integration?
...
Cache Simulation Results (50 queries)
Cache hits: 28 (74.3% of cacheable queries)
LLM calls: 10
Ephemeral (no-cache): 12
Avg cache latency: 0.171s
Avg LLM latency: 1.324s
Latency improvement: 7.7x faster from cache
Mistake 1: Caching without considering system prompt as part of the cache key
If you serve multiple tenants with different system prompts (different personas, different knowledge bases, different tones), a query from Customer A should not return a response generated under Customer B's system prompt. The fix: include a hash of the system prompt in your cache tag set, and filter by it on reads.
system_prompt_hash = hashlib.sha256(system_prompt.encode()).hexdigest()[:8]
cache_tags = existing_tags + [f"system:{system_prompt_hash}"]
Mistake 2: Setting similarity threshold too low during development
A threshold of 0.85 feels "safe" when you're testing with hand-crafted examples. In production with real user queries, you'll find that 0.85 occasionally matches semantically unrelated questions that happen to use similar vocabulary. "How do I cancel my subscription?" and "How do I cancel my order?" are different questions with very similar embeddings. Start at 0.92 and move down only with evidence.
Mistake 3: Not accounting for embedding API latency in total latency budget
The point of caching is to reduce latency. But if your embedding call takes 120ms and your cached lookup takes 50ms, your total cache read path is 170ms. On cache misses, you're adding 120ms (embedding) + full LLM latency. Mitigate this by embedding in parallel with other operations where possible, and by batching embedding requests when processing multiple queries.
Mistake 4: Forgetting that Redis JSON and RediSearch TTL behavior can diverge
When you set a TTL on a Redis key, the key expires. But if your RediSearch index hasn't been told about the deletion (which happens automatically on key expiration in recent Redis Stack versions, but may lag slightly), you can get search results pointing to expired keys. Add a null-check after retrieving the full document:
full_doc = redis_client.json().get(cache_key)
if full_doc is None:
return None # Index returned a hit but the key has expired — treat as miss
Mistake 5: Invalidating by tag and expecting instant consistency
RediSearch index updates are asynchronous relative to key deletion. After bulk invalidation, there's a brief window where search results may still include just-deleted keys. For read-critical invalidations, add a short await asyncio.sleep(0.1) before subsequent reads, or check key existence explicitly after vector search returns results.
You've built a production-grade semantic caching layer that does four things most LLM caches don't:
The threshold calibration approach gives you a principled way to tune the similarity cutoff for your specific query distribution rather than guessing. The hit rate simulation lets you measure ROI in latency and API cost terms before going to production.
Where to go next:
Distributed cache warming: Pre-populate your cache with high-frequency query embeddings before traffic hits. Combine your support ticket history with your cache writer to warm the cache before launch.
Personalized cache layers: Build a two-tier system — a shared semantic cache for general queries and a per-user Redis hash for personalized responses — and route between them based on the ephemeral classifier.
Cache analytics pipeline: Export hit count data to your observability stack (Prometheus + Grafana, or Datadog) and build dashboards that show cache hit rate over time and which query clusters are driving the most LLM cost.
Multi-modal caching: The same architecture extends to image-based queries using CLIP embeddings. If your users upload screenshots alongside questions, you can semantic-cache image + text combinations.
Prompt version management: When you update your system prompt, you want the old cached responses associated with the old prompt to expire gracefully rather than be served. The system prompt hash tagging pattern from the troubleshooting section is the foundation — build a prompt version registry on top of it.
Learning Path: Building with LLMs