I noticed that a significant fraction of queries to our RAG pipeline were semantically equivalent to queries asked in the prior hour. Not identical — users phrased them differently — but answerable with the same retrieved context. Each one triggered a full retrieval cycle and a full LLM generation pass.
Exact-match caching catches nothing here. "How do I reset my password?" and "What is the password reset process?" share no tokens in common. But they require the same answer.
Semantic caching indexes previous query-response pairs by their embeddings. When a new query arrives, it is embedded and compared against the cache. If a sufficiently similar previous query exists, the cached answer is returned directly (no retrieval, no generation). The only cost is an embedding call and a vector lookup.
This post covers how to implement semantic caching in a self-hosted pipeline, how to set the similarity threshold, what to cache and what not to, and how to measure cache effectiveness.
How Semantic Caching Works
The cache is a vector store of (query_embedding, response) pairs. For each new query:
- Embed the query
- Search the cache for the nearest stored query embedding
- If the nearest neighbor's similarity score exceeds a threshold, return the cached response
- If not, run the full RAG pipeline, then store (query_embedding, response) in the cache
The critical variable is the similarity threshold. Too high: few cache hits, little benefit. Too low: semantically different queries get the same answer, reducing accuracy.
Implementation
We will use Qdrant as the cache store, since we already have it in the stack for retrieval.
Cache Collection Setup
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance
client = QdrantClient(host="localhost", port=6333)
client.create_collection(
collection_name="semantic_cache",
vectors_config=VectorParams(size=768, distance=Distance.COSINE)
)
Cache Lookup and Storage
import ollama
import hashlib
import time
def embed(text: str) -> list[float]:
return ollama.embeddings(model="nomic-embed-text", prompt=text)["embedding"]
def cache_lookup(
query: str,
threshold: float = 0.95,
max_age_seconds: int = 3600
) -> str | None:
query_vec = embed(query)
results = client.search(
collection_name="semantic_cache",
query_vector=query_vec,
limit=1,
with_payload=True,
score_threshold=threshold
)
if not results:
return None
result = results[0]
cached_at = result.payload.get("cached_at", 0)
if time.time() - cached_at > max_age_seconds:
return None # expired
return result.payload.get("response")
def cache_store(query: str, response: str) -> None:
query_vec = embed(query)
doc_id = int(hashlib.md5(query.encode()).hexdigest(), 16) % (10**15)
client.upsert(
collection_name="semantic_cache",
points=[{
"id": doc_id,
"vector": query_vec,
"payload": {
"query": query,
"response": response,
"cached_at": time.time()
}
}]
)
Full Pipeline with Cache
def rag_with_cache(
query: str,
threshold: float = 0.95,
max_age_seconds: int = 3600
) -> dict:
# Check cache first
cached = cache_lookup(query, threshold=threshold, max_age_seconds=max_age_seconds)
if cached:
return {"response": cached, "source": "cache"}
# Full RAG pipeline
chunks = retrieve(query, k=10)
context = "\n\n".join(c["text"] for c in chunks)
response = generate(query, context)
# Store in cache
cache_store(query, response)
return {"response": response, "source": "pipeline"}
Setting the Similarity Threshold
The threshold determines when two queries are "similar enough" to share an answer. There is no universal value: it depends on how factual and query-specific your answers are.
High-specificity answers (the answer changes significantly with small query changes): use a threshold of 0.95 or higher. Example: documentation queries about specific configuration parameters. "What is the timeout for gateway A?" and "What is the timeout for gateway B?" are highly similar but have different answers.
Low-specificity answers (the answer is stable across semantically similar queries): you can lower the threshold to 0.88 or 0.90. Example: conceptual questions. "How does HNSW indexing work?" and "Can you explain HNSW index structure?" deserve the same answer.
Measure threshold impact on your actual query distribution:
def eval_threshold(
eval_pairs: list[dict], # [{"query": str, "expected_response": str}]
threshold: float
) -> dict:
hits = 0
correct_hits = 0
for i, item in enumerate(eval_pairs):
# Use other queries as cache population
for j, other in enumerate(eval_pairs):
if i != j:
cache_store(other["query"], other["expected_response"])
result = cache_lookup(item["query"], threshold=threshold)
if result:
hits += 1
# A hit is correct if the cached response is semantically similar
# to the expected response (simplified check)
if item["expected_response"][:100] in result:
correct_hits += 1
return {
"threshold": threshold,
"hit_rate": hits / len(eval_pairs),
"precision": correct_hits / hits if hits > 0 else 0
}
What to Cache and What Not To
Good cache candidates:
- Conceptual questions ("how does X work")
- Policy and procedure questions ("what is the process for Y")
- Stable factual lookups that don't change frequently
Poor cache candidates:
- Queries with explicit time references ("what happened today", "latest status")
- Queries that depend on user context or permissions
- Queries where the answer changes faster than your TTL
Use the max_age_seconds parameter to evict stale entries. Set it to match how often your underlying data changes: hourly for frequently updated corpora, daily or longer for stable documentation.
Cache Invalidation
When your source documents change, cached answers may be stale. Two approaches:
TTL-based: set a maximum age and let entries expire. Simple, no coordination needed. Works when you can tolerate some staleness.
Event-based: clear relevant cache entries when specific documents update. Requires tracking which cache entries depended on which source documents. Implement by storing source document IDs in the cache payload:
def cache_store_with_sources(
query: str,
response: str,
source_doc_ids: list[str]
) -> None:
query_vec = embed(query)
doc_id = int(hashlib.md5(query.encode()).hexdigest(), 16) % (10**15)
client.upsert(
collection_name="semantic_cache",
points=[{
"id": doc_id,
"vector": query_vec,
"payload": {
"query": query,
"response": response,
"cached_at": time.time(),
"source_doc_ids": source_doc_ids
}
}]
)
def invalidate_by_source(source_doc_id: str) -> int:
from qdrant_client.models import Filter, FieldCondition, MatchValue
result = client.delete(
collection_name="semantic_cache",
points_selector=Filter(
must=[
FieldCondition(
key="source_doc_ids",
match=MatchValue(value=source_doc_id)
)
]
)
)
return result.deleted if hasattr(result, "deleted") else 0
Measuring Cache Effectiveness
Track three metrics in production:
class CacheMetrics:
def __init__(self):
self.total_queries = 0
self.cache_hits = 0
self.latency_pipeline = []
self.latency_cache = []
def record(self, source: str, latency_ms: float) -> None:
self.total_queries += 1
if source == "cache":
self.cache_hits += 1
self.latency_cache.append(latency_ms)
else:
self.latency_pipeline.append(latency_ms)
def report(self) -> dict:
return {
"hit_rate": self.cache_hits / self.total_queries if self.total_queries else 0,
"avg_latency_cache_ms": sum(self.latency_cache) / len(self.latency_cache) if self.latency_cache else 0,
"avg_latency_pipeline_ms": sum(self.latency_pipeline) / len(self.latency_pipeline) if self.latency_pipeline else 0,
}
A cache hit rate in the low single digits usually means the threshold is too high or your query distribution has too little repetition. A precision that is clearly degraded usually means the threshold is too low.
When Semantic Caching Is Worth the Complexity
Caching adds a vector store, an embedding call on every query, and an invalidation problem. It is worth it when:
- You have repeated or near-repeated queries (user support, FAQ-style RAG)
- Generation latency is the bottleneck (large model, slow hardware)
- Your corpus is stable enough that a TTL-based strategy works
If your queries are highly varied (no two users ever ask similar things) or your corpus changes frequently, caching adds complexity without benefit. Profile your query logs for semantic repetition before building a cache layer.
Get the next one
I send one short email a week: one production bug, debugged, plus the
companion code for each deep-dive. No spam, unsubscribe anytime.
Reader challenge: are you running any form of caching in your RAG pipeline? What hit rate are you seeing and what threshold works for your query distribution? Reply with what you found.
Sources
- Qdrant search with score threshold: https://qdrant.tech/documentation/concepts/search/#filtering-results-by-score
- nomic-embed-text on Ollama: https://ollama.com/library/nomic-embed-text
- Qdrant payload filtering for cache invalidation: https://qdrant.tech/documentation/concepts/filtering/
About the Author
Toc Am
Founder of AmtocSoft. Writing practical deep-dives on AI engineering, cloud architecture, and developer tooling. Previously built backend systems at scale. Reviews every post published under this byline.
Published: 2026-07-27 · Written with AI assistance, reviewed by Toc Am.
Get These In Your Inbox
Weekly deep-dives on AI engineering, no fluff. Join the newsletter →
Or grab the book ($39, ~100 pages) · Buy me a coffee
☕ Buy Me a Coffee · 🔔 YouTube · 💼 LinkedIn · 🐦 X/Twitter
No comments:
Post a Comment