《AI Agent 工程实战》6.2 语义缓存：相似问题复用与错配检测

6.2 语义缓存：相似问题复用与错配检测

6.1 的缓存 key 是查询指纹，只有逐字节相同的 query 才命中。可真实用户的提问方式千变万化：「Redis 的 TTL 怎么设」「缓存过期时间设多久」「TTL 该配多少」——三句话说的是同一件事，在 6.1 眼里却是三个毫不相干的 key，各自回源一次。要吃掉这种近义重复，就得让缓存认识「意思」，而不只是「字面」。

本节把 ResearchCopilot 推进到 researchcopilot/semantic_cache.py：把问题向量化，按相似度阈值命中历史答案，真跑出同义命中、无关不命中、相似误命中三种情况，并讲清阈值与错配检测的取舍。

6.2.1 字面缓存的天花板

先量化一下 6.1 漏掉了多少。假设一个 workspace 里 30% 的提问是「换个说法问同一件事」，字面缓存对这部分命中率为零——key 对不上就是 miss。语义缓存的目标就是把这块捡回来。两者的差别可以并排看：

最后一行是本节的全部难点：字面缓存不会误命中（key 相同，输入必然相同），而语义缓存必然会误命中——「相似」不等于「相同」，这是它的机制性缺陷，不是调参能消除的。所以语义缓存的设计重心不是「怎么命中更多」，而是怎么控制误命中的代价。

6.2.2 机制：嵌入 + 相似度阈值

语义缓存的流程只有三步：存的时候把问题和答案一起存，问题算成向量；查的时候把新问题也算成向量，跟历史向量比相似度；取相似度最高的那个，若超过阈值就返回它对应的答案。核心就一个判断：

hit = (max_i cos(embed(query), embed(q_i)) >= threshold)

这里复用卷一 9.1 的 HashingEmbedder——一个完全确定性、零依赖的哈希嵌入，把文本切成词元后映射进 256 维向量：

from __future__ import annotations
import hashlib, re
from dataclasses import dataclass
import numpy as np

_CJK = re.compile(r"[一-鿿]")
_WORD = re.compile(r"[a-zA-Z0-9_]+")

def tokenize(text: str) -> list[str]:
    tokens = _WORD.findall(text.lower())
    cjk = _CJK.findall(text)
    tokens += cjk
    tokens += [cjk[i] + cjk[i + 1] for i in range(len(cjk) - 1)]  # 汉字二元组
    return tokens

@dataclass
class HashingEmbedder:
    dim: int = 256

    @staticmethod
    def _hash(token: str) -> int:
        return int.from_bytes(hashlib.md5(token.encode("utf-8")).digest()[:8], "little")

    def embed(self, text: str) -> list[float]:
        vec = np.zeros(self.dim, dtype=np.float64)
        for tok in tokenize(text):
            h = self._hash(tok)
            vec[h % self.dim] += 1.0 if (h >> 63) & 1 == 0 else -1.0
        norm = float(np.linalg.norm(vec))
        if norm > 0.0:
            vec /= norm
        return vec.tolist()

它把文本切成词元（英文按单词、中文按单字 + 相邻汉字二元组），每个词元用 md5 确定性哈希落进 [0, 256) 的一个桶，桶里累加 +1 或 -1，最后 L2 归一化。用 md5 而不是 Python 内建 hash() 是刻意的：内建哈希带随机盐，进程重启后同一个词落进不同桶，结果不可复现。归一化之后，两个向量的余弦相似度就退化成一次点积，这也是 lookup 里能直接写 np.dot(q, vec) 的原因。

选它而不是真实 embedding 有两个理由：一是本机可离线复现，不需要联网、不花钱；二是它把语义缓存的机制暴露得最清楚，因为哈希嵌入没有真正的语义，它只能靠词面重叠来判断相似，恰好让我们看清「相似度阈值」这个机制本身能做什么、不能做什么。

一个必须提前说清的局限：哈希嵌入认的是词面，不是意思。所以本节演示的「同义命中」，本质是带词面重叠的改写（比如都含「缓存」「TTL」），而真正零重叠的同义句（「对话太长怎么办」对「上下文压缩」）它会直接抓瞎——那是真实 embedding 才有的能力。这个局限不是缺陷，而是让我们把注意力放在阈值和错配检测上。

6.2.3 实现：SemanticCache

缓存对象是一串 (问题, 问题向量, 答案) 三元组，查询时线性扫一遍取最高分：

from __future__ import annotations
from dataclasses import dataclass, field
import numpy as np

@dataclass
class SemanticCache:
    emb: HashingEmbedder
    threshold: float = 0.60
    entries: list[tuple[str, list[float], str]] = field(default_factory=list)

    def put(self, question: str, answer: str) -> None:
        self.entries.append((question, self.emb.embed(question), answer))

    def lookup(self, query: str):
        q = self.emb.embed(query)
        best = None
        for question, vec, answer in self.entries:
            score = float(np.dot(q, vec))         # 向量已归一化，点积即余弦
            if best is None or score > best[1]:
                best = (question, score, answer)
        q_text, score, answer = best
        # 关键：只有分数过阈值才返回答案，否则当作未命中
        return (q_text, score, answer if score >= self.threshold else None)

三个设计点。第一，阈值判定和最近邻查找分开：即使找到了最近邻，分数不够也不返回答案——这是语义缓存区别于普通 KNN 检索的地方，它宁可不命中也不乱命中。第二，存的是原问题而不是向量本身，因为错配排查时需要看到「命中了哪条历史问题」，光有向量没法人工判断。第三，线性扫描在这个量级足够（几百条历史问题），真实系统才需要向量库的 ANN 索引。

lookup 的返回值里同时带上 q_text 和 score，是为可观测性留的口子——每次命中都记一条「谁命中了谁、分数多少」的日志，这就是下一节错配检测的数据来源。

6.2.4 真跑：三种情况

往缓存里放三条问答，再用三个查询去撞，覆盖「该命中」「该不命中」「不该命中却命中」：

cache = SemanticCache(HashingEmbedder(dim=256), threshold=0.60)
cache.put("Redis 缓存的 TTL 该怎么设置？", "检索结果 5 分钟、嵌入 24 小时、会话 30 分钟。")
cache.put("本地开发用不用开 Redis 持久化？", "开发环境可以关掉，重启即清空，省事。")
cache.put("Postgres 慢查询怎么定位？", "先 EXPLAIN ANALYZE，看是否走了索引。")

实测输出（Python 3.14.6，numpy 2.5.3）：

[同义改写（应命中）] Redis 缓存的 TTL 该设多少？
    最近邻: Redis 缓存的 TTL 该怎么设置？
    相似度: +0.6842  阈值: 0.60  -> 命中
    返回答案: 检索结果 5 分钟、嵌入 24 小时、会话 30 分钟。

[无关问题（应不命中）] 怎么给 Agent 加限流？
    最近邻: Redis 缓存的 TTL 该怎么设置？
    相似度: +0.3311  阈值: 0.60  -> 不命中

[相似但答案相反（误命中）] 生产环境用不用开 Redis 持久化？
    最近邻: 本地开发用不用开 Redis 持久化？
    相似度: +0.6805  阈值: 0.60  -> 命中
    返回答案: 开发环境可以关掉，重启即清空，省事。

三种情况恰好演示了语义缓存的全貌：

• 同义改写（该设多少 vs 该怎么设置）：共享「Redis」「缓存」「TTL」等词元，相似度 +0.6842 过线，命中。这是语义缓存想要的——换个说法拿到同一个答案。
• 无关问题（怎么给 Agent 加限流）：跟三条历史问题都没什么词面重叠，最高才 +0.3311，远低于阈值，正确地不命中，回源处理。
• 相似但答案相反（生产环境用不用开 vs 本地开发用不用开）：两句几乎一字之差，相似度 +0.6805——比正确的同义改写还高一点点——于是命中，返回了「开发环境可以关掉」这个对生产环境完全错误的答案。

这就是语义缓存的死穴，值得盯着看：正确的命中（0.6842）和错误的命中（0.6805）分数几乎一样，没有任何阈值能把它们分开。把阈值调到 0.70 会同时杀掉这两个；调到 0.68 会同时放过这两个。阈值这个旋钮，在「语义相似度」这个标尺上，根本区分不了「同义改写」和「一问之差但答案相反」——因为对词袋式的嵌入来说，这两者在几何上就是同一件事。

把全部九个相似度列成矩阵，看得更清楚（行是查询，列 S1/S2/S3 是三条历史问题）：

query                               S1      S2      S3
Redis 缓存的 TTL 该设多少？         +0.6842 +0.1800 +0.1147
怎么给 Agent 加限流？              +0.3311 -0.0566 +0.2887
生产环境用不用开 Redis 持久化？         +0.0468 +0.6805 +0.0000

矩阵揭示了三条规律。第一，对角线上的高分是「该命中」的：第 1 行第 1 列 +0.6842、第 3 行第 2 列 +0.6805，都是跟「同主题」的历史问题撞上了。第二，同一行的其余列都塌到 0.3 以下，说明无关问题确实被拉开了距离——阈值确实在起作用。第三，也是最要命的：第 3 行那个 +0.6805 恰恰是错的。它跟第 1 行的正确命中 +0.6842 只差 0.0037，比任何有意义的阈值间距都小。矩阵告诉我们：分数只能回答「像不像」，回答不了「对不对」——这是语义缓存的机制上限，不是实现 bug。

6.2.5 错配检测：阈值只是第一道闸

既然阈值分不开对错，那语义缓存岂不是不能用？不是，但必须承认：语义缓存是一个「命中即正确」概率小于 100% 的缓存，工程上要围绕这个概率做三件事。

第一，阈值要按具体嵌入模型实测，不能照搬。 哈希嵌入的分数分布跟真实模型完全不同：真实模型的自我相似度接近 1、语义相近常在 0.7–0.9、不相关也有 0.2–0.4 的底噪。所以「阈值设 0.60」这句话，离开具体模型就毫无意义。定阈值的正确做法是拿一批标注数据扫一遍：把「该命中」和「不该命中」的分数画成两条分布，取让误命中率降到可接受水平的那个点。宁可设高、宁可多漏几次，也不要设低、把错误答案喂给用户。

第二，必须留人工抽查的口子。 再好的阈值也会有漏网之鱼，所以真实系统一定会记录每一次语义命中的 (query, 命中的历史问题, 分数, 返回的答案)，让运营能事后翻账。做法通常是：把阈值以上的命中分两档——高分档（如 ≥0.9）直接返回，低分档（阈值到 0.9 之间）只作提示（「你问的是不是这个？」）而不直接作答。这个「两档」设计把「机器不确定的边界」交回给人，是语义缓存能不能上生产的关键工程决策：

def lookup_two_tier(self, query: str, high: float = 0.90) -> tuple[str, str | None, float]:
    q_text, score, answer = self.lookup(query)
    if answer is None:
        return q_text, None, score            # 未命中，回源
    if score >= high:                         # 高分：直接返回
        return q_text, answer, score
    return q_text, None, score                # 低分：命中但只作提示，不直接作答

把 high 设成一个更高的门槛（比如 0.90），系统就有了一条「机器自己拿不准」的中间带——这一带的命中会进入抽查队列，而不是直接喂给用户。能不能把不确定的边界暴露出来，是「玩具」和「能上生产」的分水岭。

第三，从源头拦住不该缓存的问题。 见下一节——有些问题根本不该进语义缓存，再高的阈值也救不了。

一句话总结错配检测：阈值决定「敢不敢命中」，抽查决定「命错了能不能发现」，白名单决定「哪些根本不该进缓存」。三者缺一不可。

6.2.6 哪些问题绝不能走语义缓存

上面那个「生产 vs 开发」的误命中不是偶然，它暴露了一整类绝对不能缓存的问题。判断标准只有一条：这个问题的正确答案是否随「不在 query 文本里」的上下文变化。若是，语义缓存必然出错。三类最典型：

三类的共同点是：query 相同，正确答案却不同——而语义缓存的整个假设恰恰是「query 相似 → 答案可复用」。这个假设对「客观知识类」问题成立（「Redis TTL 怎么设」的答案对谁都一样），对上面三类彻底失效。

所以工程上的落地方式是白名单，而不是黑名单：不指望识别出「哪些问题不能缓存」，而是只对明确安全的问题类型开启语义缓存——通常是通用知识问答、文档解释、代码示例这类「答案与用户无关」的请求。涉及「我的」「今天」「最新」这类词、或需要查库/查权限的请求，直接绕过语义缓存：

VOLATILE = ("今天", "现在", "最新", "我的", "我们公司", "当前")
def cacheable(query: str) -> bool:
    # 命中时效/个性化/权限词的一律不进语义缓存
    return not any(w in query for w in VOLATILE)

白名单思路的本质是把「判断责任」放在最确定的地方：与其让一个相似度分数去猜「这个问题能不能缓存」（它猜不准），不如用一条显式的规则在入口处拦掉已知的危险类型。规则会漏、会误杀，但它可控、可解释、可审计——而这三点恰恰是缓存这种「命中即返回」的机制最需要的品质。

6.2.7 失效与两级叠加

语义缓存还有两个绕不开的工程问题。

第一，语料更新了怎么办。 语义缓存存的是「问题 → 答案」，可答案本身来自检索。语料一更新，旧答案就过时了，可它的 key 是问题文本，跟语料毫无关系——光靠 TTL 等它自然过期，中间这段时间用户拿到的就是错答案。解决办法是给缓存加版本号：把语料版本（或索引构建时间戳）编进缓存条目的元数据，版本一变，整批语义缓存视为失效。这比「逐条判断哪条答案受语料影响」简单得多，也安全得多。

@dataclass
class VersionedCache:
    emb: HashingEmbedder
    threshold: float = 0.60
    corpus_version: str = "v1"                # 语料一变就换这个值
    entries: list[tuple[str, list[float], str]] = field(default_factory=list)

    def invalidate(self, new_version: str) -> None:
        self.corpus_version = new_version
        self.entries.clear()                  # 整体作废，不做逐条判断

第二，为什么语义缓存通常和字面缓存叠成两级。 语义缓存虽好，但它贵（每次都要算一次嵌入）且有误命中风险。所以真实系统的顺序是：先查 6.1 的字面缓存（一次 GET，零误判、几乎免费），不中再查语义缓存（要算嵌入、有误判风险）。这样绝大多数「完全重复」的请求被第一级用最低成本吃掉，只有真正的「近义改写」才落到第二级。两级叠起来，既省了算力，又把误命中的暴露面压到了最小——因为高频的完全重复根本不会走到有风险的那一级。

一句话：L1 兜住量大且安全的重复，L2 才去碰量小但有风险的重写——顺序不能反，否则等于把最贵的判断用在了最不需要它的请求上。

真实环境会怎样：本节用的是哈希嵌入，换成真实 embedding（text-embedding-3-small、bge-small-zh）后，相似度分布、阈值、误命中模式都会变，但机制和风险完全一样——真实模型更懂语义，意味着它能命中更多真正的同义句，但同样会把「生产 vs 开发」这类表面高度相似的问题判为命中。真实系统里还有两个本机演示没有的点：一是缓存要带版本号，语料或提示词更新时整体作废；二是语义缓存通常和 6.1 的字面缓存叠成两级——先查字面（便宜、零误判），不中再查语义（贵、有误判风险），这样既省算力又把误命中面压到最小。语义缓存不是「更好的缓存」，它是「用一点正确性换更多命中的缓存」——用之前先想清楚这笔交换划不划算。

阅读导航：上一节：6.1 Redis 缓存：检索结果与嵌入缓存命中率 · 下一节：6.3 限流与配额：按租户 token 预算的令牌桶 。