语义缓存：40 行代码省半费用

如果你的每一条用户请求都会调用一次 LLM API，那么你几乎肯定在为同一个问题重复付费。这并不是因为用户愚笨，而是因为人类会用十几种不同的方式询问同一件事，而普通缓存只能匹配完全相同的字符串。

对于 dict 或 Redis GET 来说，"how do I reset my password" 和 "reset password help pls" 完全没有关系。但对用户而言，它们就是同一个请求。每一次匹配失败，都意味着你为一次本可避免的模型调用支付了全价。

我们来解决这个问题。读完本文，你将拥有一个可以正常工作的语义缓存，准确理解它会在哪里失效，并知道如何修复这些问题。

具体来说，问题是什么

import redis
import hashlib

r = redis.Redis()

def cached_call(prompt):
    key = hashlib.sha256(prompt.encode()).hexdigest()
    cached = r.get(key)
    if cached:
        return cached.decode()
    response = call_llm(prompt)  # expensive
    r.set(key, response)
    return response

这段代码在演示环境中效果很好。但到了生产环境，面对真实用户流量时，缓存命中率会长期徘徊在接近零的位置。因为 hashlib.sha256 并不知道 "reset my password" 和 "help me reset password" 表达的是同一个意思。你构建的缓存，只有在用户连续两次原样复制粘贴同一个请求时才有用——而对于人类手动输入的内容来说，这种情况基本不会发生。

第一步：用 Embedding 代替 Hash

我们不再匹配完全相同的字符串，而是匹配语义。把查询转换为向量，再与已经回答过的查询向量进行比较。

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("all-MiniLM-L6-v2")  # fast, good enough for cache matching

def embed(text: str) -> np.ndarray:
    return model.encode(text, normalize_embeddings=True)

normalize_embeddings=True 非常重要——它意味着稍后可以直接使用简单的点积计算余弦相似度，不需要再额外执行归一化。

第二步：构建一个最小可用的向量存储

你不需要 Qdrant 或 Milvus 也能理解这种模式（生产环境要扩大规模时，你很可能会需要它们，后文会详细说明）。对于一个可运行的原型，使用内存列表完全足够：

cache_store = []  # list of dicts: {embedding, query, response}

def find_match(query_embedding, threshold=0.92):
    best_score, best_entry = 0, None
    for entry in cache_store:
        score = float(np.dot(query_embedding, entry["embedding"]))
        if score > best_score:
            best_score, best_entry = score, entry
    if best_score > threshold:
        return best_entry
    return None

第三步：把它们串联起来

def semantic_cached_call(prompt: str) -> str:
    query_embedding = embed(prompt)
    match = find_match(query_embedding)

    if match:
        print(f"cache hit (score matched threshold)")
        return match["response"]

    response = call_llm(prompt)  # your actual OpenAI/Anthropic call
    cache_store.append({
        "embedding": query_embedding,
        "query": prompt,
        "response": response,
    })
    return response

全部代码就是这些。用一批经过改写的测试查询运行它，你会立刻看到：那些传统字符串缓存完全无法识别的查询，现在可以命中缓存了。

第四步：它会在哪里失效（以及为什么你仍然需要它）

print(semantic_cached_call("How do I cancel my subscription?"))
print(semantic_cached_call("How do I reactivate my subscription?"))

具体结果取决于你使用的 Embedding 模型和阈值，但第二个查询确实有可能获得关于取消订阅的回答。因为 "cancel" 和 "reactivate" 所在的句子几乎共享了其他所有单词，而 Embedding 模型往往更容易根据共有词汇进行聚类，却不太擅长区分这种表达指令的动词。

这是语义缓存最常见的失效模式。也正因如此，如果没有保护机制，你不能直接把上面这个 40 行版本部署到生产环境。

解决办法如下：针对已知的语义对立词，添加一条轻量级拒绝规则。在相似度匹配完成之后、信任匹配结果之前执行这项检查：

OPPOSING_PAIRS = [
    ("cancel", "reactivate"), ("enable", "disable"),
    ("add", "remove"), ("increase", "decrease"),
]

def has_conflicting_intent(query: str, cached_query: str) -> bool:
    q, c = query.lower(), cached_query.lower()
    for a, b in OPPOSING_PAIRS:
        if (a in q and b in c) or (b in q and a in c):
            return True
    return False

def safer_semantic_call(prompt: str) -> str:
    query_embedding = embed(prompt)
    match = find_match(query_embedding)

    if match and not has_conflicting_intent(prompt, match["query"]):
        return match["response"]

    response = call_llm(prompt)
    cache_store.append({"embedding": query_embedding, "query": prompt, "response": response})
    return response

它无法捕获所有边缘情况——如果想做到这一点，你需要加入真正的 cross-encoder 重排序步骤，或者使用一个小型意图分类器——但它几乎没有增加多少复杂度，却能消除最令人尴尬的那一类 Bug。对于周末项目或内部工具来说，这是一个不错的收尾点。

第五步：让它达到生产可用水平

内存列表适合演示。面对真实流量时，应当将其替换为真正的向量数据库。查询模式保持不变，只需要把 for entry in cache_store 换成 ANN 索引：

# Using Qdrant as an example
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance

client = QdrantClient(":memory:")  # swap for a real host in prod
client.create_collection(
    collection_name="semantic_cache",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE),
)

def find_match_qdrant(query_embedding, threshold=0.92):
    results = client.search(
        collection_name="semantic_cache",
        query_vector=query_embedding.tolist(),
        limit=1,
    )
    if results and results[0].score > threshold:
        return results[0].payload
    return None

逻辑完全相同，只是现在有了索引作为支撑，可以扩展到数百万条缓存记录，同时将查询耗时控制在一毫秒以内。

上线之前，我会提醒你的常见错误

错误 1：所有查询类型共用一个全局阈值。为 "how do I use the API" 调整好的阈值，用在 "what's my refund eligibility" 上可能过于宽松。如果你的应用横跨多个领域，请使用不同的阈值，或者完全禁止对敏感类别进行缓存。

错误 2：缓存所有内容，并且永不过期。一旦底层事实发生变化，例如价格、政策或功能可用性有了调整，缓存中的回答就会过时。为每条缓存记录添加 created_at，并对所有非长期有效的内容采取更激进的过期策略。

错误 3：只相信相似度分数。正如前面展示的那样，相似度高并不代表意图相同。在返回缓存响应之前，一定要为向量搜索搭配至少一道轻量级保护机制。

错误 4：没有按类别统计命中率。总体命中率只是一个虚荣指标。一个客服 Bot 在 "how do I..." 类问题上的命中率可能达到 70%，但在账户相关问题上却接近 0%。你必须分清不同类别，才能决定哪些内容真正值得缓存。

拿起上面的 40 行版本，把它连接到你正在调用的任意 LLM，然后基于一天的真实流量记录缓存命中和未命中的情况。对于任何存在重复用户意图的应用，例如客服、FAQ 或内部工具，我真心愿意打赌，你至少能获得 30%～40% 的命中率。每一次命中，都意味着你少支付了一次完整 LLM 调用的费用，也不必再等待它返回结果。

如果你想参考一份更加完整、内置保护机制和失效逻辑的实现，我做了一个名为 Remem 的开源 package（pip install remem-ai）。等你构建完自己的版本，想要对比那些更棘手的边缘情况时，它值得一看。

现在轮到你了：构建这个 40 行版本，用你自己的日志或一套经过改写的测试数据运行它，然后在评论区告诉我你的命中率。我想知道，在我自己的流量模式之外，30%～40% 的命中率是否依然成立。

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。