使用嵌入实现语义缓存
构建语义缓存，通过将查询嵌入与历史请求嵌入缓存进行比较，为语义相似但并不完全相同的查询检索已存储的响应。
使用嵌入实现语义缓存 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
The Limitation of Exact Caching
Exact caching only helps when users send byte-for-byte identical requests. In reality, users phrase the same question differently: 'How do I cancel my subscription?', 'What is the process to unsubscribe?', and 'Can I stop my plan?' all intend the same question but produce different cache keys. Exact caching misses all these variants. Semantic caching solves this by matching similar queries instead of identical ones, dramatically increasing cache hit rates.
How Semantic Caching Works
A semantic cache stores the embedding of each cached query alongside the cached response. When a new query arrives, embed it and search the cache for a previously seen query with high cosine similarity. If the closest cached query is above a similarity threshold (typically 0.95+), return its cached response. If no match is found, call the LLM, store the new response, and add the new query's embedding to the cache index for future lookups.
# Semantic cache flow
# 1. New query arrives: 'How do I cancel my subscription?'
# 2. Embed it: embed_query = embed('How do I cancel my subscription?')
# 3. Search cache index for nearest cached query embedding
# 4. Find cached: 'What is the process to unsubscribe?' (similarity=0.97)
# 5. 0.97 >= threshold (0.95) → cache HIT, return cached response
# 6. If 0.82 < threshold → cache MISS, call LLM, cache result, add embedding to indexIn-Memory Semantic Cache with NumPy
For small applications or prototypes, implement semantic caching in memory using NumPy for cosine similarity computation. Store cached query embeddings in a 2D array and responses in a parallel list. On each new query, compute cosine similarity between the new embedding and all cached embeddings, and return the closest match if it exceeds the threshold.
import numpy as np
from openai import OpenAI
client = OpenAI()
class InMemorySemanticCache:
def __init__(self, threshold: float = 0.95):
self.threshold = threshold
self.embeddings = [] # list of np.ndarray
self.responses = [] # list of str
self.queries = [] # list of str (for inspection)
def _embed(self, text: str) -> np.ndarray:
resp = client.embeddings.create(model='text-embedding-3-small', input=text)
return np.array(resp.data[0].embedding)
def get(self, query: str) -> str | None:
if not self.embeddings:
return None
q_emb = self._embed(query)
cache_matrix = np.array(self.embeddings)
# Cosine similarity: dot product of normalized vectors
norms = np.linalg.norm(cache_matrix, axis=1)
q_norm = np.linalg.norm(q_emb)
sims = (cache_matrix @ q_emb) / (norms * q_norm + 1e-8)
best_idx = int(np.argmax(sims))
if sims[best_idx] >= self.threshold:
print(f'[SEMANTIC HIT] sim={sims[best_idx]:.3f} matched: {self.queries[best_idx]!r}')
return self.responses[best_idx]
return None
def set(self, query: str, response: str):
emb = self._embed(query)
self.embeddings.append(emb)
self.responses.append(response)
self.queries.append(query)Semantic Cache with Redis and Pinecone
For production semantic caching, store query embeddings in a vector database for fast approximate nearest neighbor search, and store responses in Redis keyed by a unique ID. When a new query arrives, search the vector database for the closest cached query, retrieve the response from Redis using the ID in the vector metadata, and return it — all without calling the LLM.
import redis
from pinecone import Pinecone
import hashlib
r = redis.Redis(decode_responses=True)
pc = Pinecone(api_key='YOUR_KEY')
index = pc.Index('semantic-cache')
SIMILARITY_THRESHOLD = 0.95
def semantic_cache_get(query: str) -> str | None:
q_emb = embed(query) # from earlier lesson
results = index.query(vector=q_emb, top_k=1, include_metadata=True)
if not results.matches:
return None
best = results.matches[0]
if best.score >= SIMILARITY_THRESHOLD:
response_key = best.metadata.get('response_key')
return r.get(response_key)
return None
def semantic_cache_set(query: str, response: str):
q_emb = embed(query)
entry_id = hashlib.sha256(query.encode()).hexdigest()[:16]
response_key = f'sem_cache_resp:{entry_id}'
r.setex(response_key, 86400, response) # 24h TTL
index.upsert(vectors=[{
'id': entry_id,
'values': q_emb,
'metadata': {'query': query[:200], 'response_key': response_key},
}])GPTCache: A Ready-Made Semantic Cache
GPTCache is an open-source library that implements semantic caching for LLM applications. It wraps the OpenAI client, embeds queries automatically, checks a vector similarity cache, and falls back to the real API on misses. It supports multiple vector stores (FAISS, Milvus, Redis) and multiple embedding models out of the box, making it a fast way to add semantic caching to an existing application.
# pip install gptcache
from gptcache import cache
from gptcache.adapter import openai
from gptcache.embedding import OpenAI as EmbeddingOpenAI
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation
# Configure GPTCache with FAISS vector store
cache.init(
embedding_func=EmbeddingOpenAI().to_embeddings,
data_manager=get_data_manager(
CacheBase('sqlite'),
VectorBase('faiss', dimension=1536),
),
similarity_evaluation=SearchDistanceEvaluation(),
)
# Now use the wrapped openai client — caching is transparent
response = openai.ChatCompletion.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': 'What is RAG?'}],
)Choosing the Similarity Threshold
The similarity threshold is the most important hyperparameter for semantic caching. Too high (0.99+) and you miss most paraphrase variants. Too low (0.85-) and you return wrong cached answers for different but superficially similar queries. Validate your threshold empirically by sampling query pairs and checking whether queries above the threshold truly have the same intended answer. Typical values: 0.92-0.97 for factual Q&A, 0.98+ for code generation where small differences matter a lot.
def validate_threshold(cache, query_pairs_with_labels):
'''
query_pairs_with_labels: list of (query1, query2, should_match: bool)
'''
true_pos = true_neg = false_pos = false_neg = 0
for q1, q2, should_match in query_pairs_with_labels:
e1, e2 = embed(q1), embed(q2)
sim = cosine_similarity(e1, e2)
matched = sim >= cache.threshold
if should_match and matched: true_pos += 1
elif not should_match and not matched: true_neg += 1
elif not should_match and matched: false_pos += 1
else: false_neg += 1
precision = true_pos / (true_pos + false_pos) if (true_pos + false_pos) else 0
recall = true_pos / (true_pos + false_neg) if (true_pos + false_neg) else 0
print(f'Precision: {precision:.3f}, Recall: {recall:.3f}')Semantic Cache Scope: System Prompt Matters
A critical detail: semantic caching must account for the system prompt. Two identical user queries produce different answers if the system prompt differs (different personas, different knowledge bases, different response formats). Always include the system prompt in the embedding input or create separate cache namespaces per system prompt. A clean pattern is to hash the system prompt and use it as a cache namespace prefix.
import hashlib
def make_semantic_cache_namespace(system_prompt: str) -> str:
return 'sc:' + hashlib.md5(system_prompt.encode()).hexdigest()[:8]
def semantic_cache_get_namespaced(system_prompt: str, user_query: str) -> str | None:
namespace = make_semantic_cache_namespace(system_prompt)
q_emb = embed(user_query)
# Search only within this namespace
results = index.query(
vector=q_emb,
top_k=1,
filter={'namespace': namespace},
include_metadata=True,
)
if results.matches and results.matches[0].score >= SIMILARITY_THRESHOLD:
return r.get(results.matches[0].metadata['response_key'])
return NoneSemantic Cache Hit Rate Analysis
After deploying semantic caching, analyze hit rates segmented by query cluster. Use the cached query embeddings themselves — cluster them with K-means and compute hit rate per cluster. High-hit clusters represent common question themes where caching pays off most. Low-hit clusters of diverse unique queries may not benefit from caching at all and could be excluded from the cache to reduce index size and embedding costs.
from sklearn.cluster import KMeans
import numpy as np
def analyze_cache_clusters(cache, n_clusters=10):
if len(cache.embeddings) < n_clusters:
print('Not enough cache entries to cluster')
return
matrix = np.array(cache.embeddings)
kmeans = KMeans(n_clusters=n_clusters, n_init=10, random_state=42)
labels = kmeans.fit_predict(matrix)
from collections import Counter
cluster_sizes = Counter(labels)
print('Query clusters by size:')
for cluster_id, count in cluster_sizes.most_common():
representative = cache.queries[labels.tolist().index(cluster_id)]
print(f' Cluster {cluster_id}: {count} queries, e.g. {representative!r}')Semantic Cache Security Considerations
Semantic caching introduces a privacy risk: if user A asks a sensitive question, its response might be returned to user B who asks a similar question. This is acceptable for public knowledge bases but not for applications with user-specific data or sensitive content. Apply strict namespace isolation per user or organization, and consider excluding queries that match patterns like personal information from the cache entirely.
import re
PII_PATTERNS = [
r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', # phone numbers
r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b', # emails
r'\b\d{9}\b', # SSN-like
]
def should_cache(query: str) -> bool:
for pattern in PII_PATTERNS:
if re.search(pattern, query, re.IGNORECASE):
return False # do not cache queries with PII
return True
def secure_semantic_completion(user_id: str, query: str) -> str:
if should_cache(query):
cached = semantic_cache_get_namespaced(f'user:{user_id}', query)
if cached:
return cached
result = call_llm_api(query) # actual API call
if should_cache(query):
semantic_cache_set_namespaced(f'user:{user_id}', query, result)
return resultCombining Exact and Semantic Caching
The most efficient caching strategy uses both exact and semantic caching in a two-tier hierarchy. Check exact cache first (fastest, zero embedding cost) and return immediately on hit. If exact cache misses, check semantic cache (requires one embedding API call). If semantic cache misses, call the LLM. This ordering minimizes both latency and cost per request.
async def two_tier_cached_completion(messages: list[dict], model: str = 'gpt-4o-mini') -> str:
user_query = messages[-1].get('content', '')
system_prompt = messages[0].get('content', '') if messages and messages[0]['role'] == 'system' else ''
# Tier 1: exact cache (instant, free)
exact_key = make_cache_key(messages, model, temperature=0.0)
exact_cached = await async_r.get(exact_key)
if exact_cached:
return json.loads(exact_cached)
# Tier 2: semantic cache (one embedding call ~5ms)
sem_result = semantic_cache_get_namespaced(system_prompt, user_query)
if sem_result:
# Backfill exact cache to avoid embedding next time
await async_r.setex(exact_key, 3600, json.dumps(sem_result))
return sem_result
# Tier 3: actual LLM call
response = await async_client.chat.completions.create(
model=model, messages=messages, temperature=0.0
)
result = response.choices[0].message.content
await async_r.setex(exact_key, 3600, json.dumps(result))
semantic_cache_set_namespaced(system_prompt, user_query, result)
return resultCache Warming for Cold Start
A fresh semantic cache provides zero benefit until it is populated. For applications with predictable traffic patterns, pre-warm the cache at startup by embedding and caching responses for the most frequently asked questions from your historical query logs. This eliminates the cold-start period where every user during the first hours of operation gets a cache miss and incurs full API cost.
async def warm_semantic_cache(faq_list: list[dict], system_prompt: str):
print(f'Warming cache with {len(faq_list)} FAQ entries...')
for entry in faq_list:
cached = semantic_cache_get_namespaced(system_prompt, entry['question'])
if cached:
print(f' Already cached: {entry["question"][:50]}')
continue
# Generate and cache the response
response = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': entry['question']},
],
temperature=0.0,
)
answer = response.choices[0].message.content
semantic_cache_set_namespaced(system_prompt, entry['question'], answer)
print(f' Cached: {entry["question"][:50]}')
print('Cache warming complete')Quick Check
Test your understanding of semantic caching from this lesson.
Lesson Recap
In this lesson you learned: semantic caching matches similar but non-identical queries by comparing query embeddings against a vector store of cached query embeddings, the similarity threshold controls the trade-off between hit rate and answer correctness, and system prompt namespacing prevents incorrect cross-context cache hits. A two-tier architecture checking exact cache before semantic cache minimizes both latency and embedding costs. Next up we leverage OpenAI's built-in prompt prefix caching.
常见问题解答
「使用嵌入实现语义缓存」课时是免费的吗？
是的 — 「使用嵌入实现语义缓存」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「使用嵌入实现语义缓存」这节课中我会学到什么？
构建语义缓存，通过将查询嵌入与历史请求嵌入缓存进行比较，为语义相似但并不完全相同的查询检索已存储的响应。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「使用嵌入实现语义缓存」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。