6.1 Redis 缓存：检索结果与嵌入缓存命中率
第 5 章把会话、消息、运行、记忆都落进了 Postgres，查询也调优了，可服务一旦真跑起来，你会发现钱和时间仍花在重复的计算上：同一个用户反复问「Redis 的 TTL 怎么设」，Agent 每次都重新嵌入问题、重新跑一遍多跳检索、重新拼上下文。检索和嵌入都是纯函数——输入相同、输出必相同，却没有任何一层记住过结果。这一节就把这层「记住」补上。
本节把 ResearchCopilot 推进到
researchcopilot/cache.py：用 Redis 给检索结果与嵌入向量加一层缓存，key 带上workspace_id与查询指纹，真连本机 Redis 跑出命中率与耗时对比，并定下 TTL 与三大缓存故障的对策。
6.1.1 Agent 服务的两种重复
缓存不是「有就好」，得先看清 Agent 服务里到底在重复什么。有两类重复，量级和性质完全不同：
| 重复类型 | 例子 | 相同输入？ | 适合缓存？ |
|---|---|---|---|
| 完全重复 | 同一 workspace 反复问同一个问题 | 是，逐字节相同 | 强适合，key 直接命中 |
| 近义重复 | 换个说法问同一个问题 | 否，字面不同 | 弱适合，交给 6.2 的语义缓存 |
本节只解决第一类——字面完全相同的重复。别小看它：企业用户往往把同一个问题粘给多个会话，Agent 的 search 工具在固定语料上反复被同样的关键词打，后台 digest 每天用同样的模板跑同样的检索。这些请求的字面输入一模一样，却每次都在重算。
Agent 服务里最值得缓存的三样东西是：检索结果（search 工具的输出，一次多跳检索可能要几百毫秒）、嵌入向量（把问题或文档变成向量，最贵的一步）、会话摘要（长对话压缩后的中间产物）。三者的读多写少程度递减，缓存价值也递减，本节重点做前两个。
把 ResearchCopilot 处理一次提问的路径摊开，重复点就清楚了。一次 run 的骨架是：嵌入问题 → 向量检索 → 重排 → 可能再检索一轮 → 拼上下文 → 模型生成。其中前四步不含任何模型调用，是纯计算，而且对同一 (workspace_id, query) 的输出完全确定；只有最后一步「模型生成」是不确定的（有采样、有温度）。于是结论很明确：把确定的那几段缓存起来，把不确定的那一段留给模型。这一刀切下去，缓存层要管的就只是检索和嵌入，逻辑干净、失效简单——这正是本节只做这两样的原因。
6.1.2 缓存 key 必须带 workspace_id
这是本节最容易被漏、后果最严重的一点：缓存 key 里必须带 workspace_id。4.2 花了整节讲租户隔离——记忆表按 workspace_id 分区、查询强制带租户谓词——可一旦在缓存层忘了它，隔离就从前门挡住了、后门又漏了。
设想一个没有租户维度的 key：ret:<query_hash>:3。租户 A 问「我们 Q3 的营收是多少」，命中并缓存；租户 B 问一模一样的字符串，直接拿到 A 的检索结果——这不是性能问题，是跨租户数据泄漏。所以 key 的形态必须是：
ret:<workspace_id>:<sha256(query)[:16]>:<top_k>
emb:<workspace_id>:<sha256(text)[:16]>
三个组成部分各有职责：workspace_id 保证租户隔离，sha256 指纹保证同 query 稳定映射到同一个 key（且长度可控），top_k 保证不同召回条数不会被串味——top_k=3 和 top_k=10 的结果集不同，必须分开存。
用 sha256 而不是把原始 query 塞进 key，还有一个现实原因：Redis key 是有长度和字符限制的，中文 query 直接进 key 又长又难读，而且一旦 query 里带了 : 就会把 key 结构冲乱。指纹化之后，key 永远是定长的十六进制串。
6.1.3 起 Redis 与缓存层实现
本机需要一个 Redis。书稿专用容器映射在非默认端口 16380（6379 已被占用），等价命令：
docker run -d --name ab-redis -p 16380:6379 redis:7-alpine
连接串是 redis://localhost:16380/0，服务端实测 Redis 7.4.11，Python 客户端 redis 8.1.0。若连不上，先 docker ps | grep ab-redis 确认容器在跑。没起容器时，本节代码可以把 redis.Redis 换成 fakeredis 或直接用一个 dict 降级，但那样就没有 TTL 与原子性的真实验证了。
缓存层的核心是「查缓存 → 未命中则回源 → 写回缓存」三段式：
from __future__ import annotations
import hashlib, json, time
from dataclasses import dataclass
import redis
R = redis.from_url("redis://localhost:16380/0", decode_responses=True)
def qhash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
@dataclass
class CachedRetriever:
r: redis.Redis
ttl_retrieval: int = 300 # 检索结果 5 分钟
ttl_embedding: int = 86400 # 嵌入 24 小时
ret_hits: int = 0
ret_misses: int = 0
emb_hits: int = 0
emb_misses: int = 0
def embed(self, workspace_id: str, text: str) -> list[float]:
key = f"emb:{workspace_id}:{qhash(text)}"
raw = self.r.get(key)
if raw is not None:
self.emb_hits += 1
return json.loads(raw)
self.emb_misses += 1
vec = self._embed_source(text) # 昂贵的回源
self.r.set(key, json.dumps(vec), ex=self.ttl_embedding)
return vec
def retrieve(self, workspace_id: str, query: str, top_k: int = 3):
key = f"ret:{workspace_id}:{qhash(query)}:{top_k}"
raw = self.r.get(key)
if raw is not None:
self.ret_hits += 1
return json.loads(raw), "hit"
self.ret_misses += 1
result = self._retrieve_source(workspace_id, query, top_k)
self.r.set(key, json.dumps(result), ex=self.ttl_retrieval)
return result, "miss"
@staticmethod
def _embed_source(text: str) -> list[float]:
time.sleep(0.08) # 模拟一次真实嵌入调用
return [b / 255.0 for b in hashlib.md5(text.encode()).digest()]
@staticmethod
def _retrieve_source(workspace_id: str, query: str, top_k: int):
time.sleep(0.03) # 模拟向量检索 + 重排
return [{"doc": f"{workspace_id}-doc-{i}", "score": round(0.9 - 0.1 * i, 2)}
for i in range(top_k)]
三个设计点值得展开。第一，回源函数是「慢」的模拟——_embed_source 睡 80ms、_retrieve_source 睡 30ms，代表真实世界里一次嵌入 API 调用和一次向量检索的耗时；缓存的价值就是把这几十上百毫秒压成一次网络往返。第二，写缓存用 ex= 而不是永久 key——所有缓存都必须有 TTL，下一节细讲。第三，命中计数器内建在类里，这样命中率是一个可观测指标，而不是拍脑袋的感觉。
6.1.4 真跑：未命中 → 回源 → 命中
先看单次调用的对比——同一 workspace、同一 query 连跑两次：
cache = CachedRetriever(R)
WS = "ws-acme"
t0 = time.perf_counter()
res1, st1 = cache.retrieve(WS, "Redis 缓存 TTL 怎么设", 3)
t1 = time.perf_counter()
res2, st2 = cache.retrieve(WS, "Redis 缓存 TTL 怎么设", 3)
t2 = time.perf_counter()
print(f"retrieve #1: {st1:4s} {(t1 - t0) * 1000:6.1f} ms -> {res1[0]['doc']}")
print(f"retrieve #2: {st2:4s} {(t2 - t1) * 1000:6.1f} ms -> {res2[0]['doc']}")
e0 = time.perf_counter(); cache.embed(WS, "Redis 缓存 TTL 怎么设"); e1 = time.perf_counter()
cache.embed(WS, "Redis 缓存 TTL 怎么设"); e2 = time.perf_counter()
print(f"embed #1: miss {(e1 - e0) * 1000:6.1f} ms")
print(f"embed #2: hit {(e2 - e1) * 1000:6.1f} ms")
实测输出（Python 3.14.6，redis 8.1.0，服务端 7.4.11）：
retrieve #1: miss 51.8 ms -> ws-acme-doc-0
retrieve #2: hit 4.6 ms -> ws-acme-doc-0
embed #1: miss 95.0 ms
embed #2: hit 4.3 ms
差距一目了然：第一次 51.8ms（回源检索 + 一次 Redis 写），第二次 4.6ms（一次 Redis 读），快了约 11 倍。嵌入更夸张：95.0ms → 4.3ms，约 22 倍。这正是缓存层的意义——把「昂贵的计算」换成「一次内存查表」。
再跑一批有真实感的请求流：8 次请求里只有 3 个不同的问题，看聚合命中率：
BATCH = [
"Redis 缓存 TTL 怎么设", "Redis 缓存 TTL 怎么设", "Postgres 慢查询怎么优化",
"Redis 缓存 TTL 怎么设", "Postgres 慢查询怎么优化", "怎么给 Agent 加限流",
"怎么给 Agent 加限流", "Postgres 慢查询怎么优化",
]
t = time.perf_counter()
for q in BATCH:
cache.retrieve(WS, q, 3)
elapsed = time.perf_counter() - t
total = cache.ret_hits + cache.ret_misses
print(f"批量 {len(BATCH)} 次请求: 命中 {cache.ret_hits} / 未命中 {cache.ret_misses}"
f" 命中率 {cache.ret_hits / total:.0%} 总耗时 {elapsed * 1000:.0f} ms")
etotal = cache.emb_hits + cache.emb_misses
print(f"嵌入: 命中 {cache.emb_hits} / 未命中 {cache.emb_misses} 命中率 {cache.emb_hits / etotal:.0%}")
_, st3 = cache.retrieve("ws-other", "Redis 缓存 TTL 怎么设", 3)
print(f"跨 workspace 同 query: {st3}（key 带 workspace_id，故不命中）")
for k in sorted(R.keys("*")):
print(f" {k} ttl={R.ttl(k)}s")
实测输出：
批量 8 次请求: 命中 7 / 未命中 3 命中率 70% 总耗时 154 ms
嵌入: 命中 1 / 未命中 1 命中率 50%
跨 workspace 同 query: miss（key 带 workspace_id，故不命中）
emb:ws-acme:eaa7c783f8ad98a6 ttl=86400s
ret:ws-acme:c162db3574b5ca61:3 ttl=300s
ret:ws-acme:d7f7cc5fd20cd386:3 ttl=300s
ret:ws-acme:eaa7c783f8ad98a6:3 ttl=300s
ret:ws-other:eaa7c783f8ad98a6:3 ttl=300s
注意这里的命中数是 7，但总请求是 10（前两次单测 + 批量的 8 次），因为计数器是跨批次累积的。批量里 3 个不同问题各首跑一次（3 miss）、其余 5 次命中（5 hit），加上前面单测的 1 miss + 1 hit，合计 7 hit / 3 miss = 70%。这个数字说明：只要问题有重复，缓存就能吃掉大部分回源。
最后一行是租户隔离的验证：ws-other 用完全相同的 query 请求，结果 miss。因为 key 里带了 workspace_id，ret:ws-acme:... 和 ret:ws-other:... 是两个不同的 key，谁也命中不了谁。翻到 KEYS * 的输出也能看到两条 eaa7c783f8ad98a6 并排存在、各自带自己的 TTL——指纹相同、前缀不同，隔离生效。
6.1.5 缓存粒度：缓存零件，不缓存整机
新手最容易犯的错，是直接把整个 run 的结果当缓存对象：run:<workspace>:<query_hash> -> 最终答案。这看起来最省事——一次命中直接返回报告，连模型都不用调——但它踩了三个坑。
第一，粒度太粗，命中率骤降。整机结果的 key 必须包含全部输入：query、检索到的语料版本、工具返回、甚至模型名与温度。任何一个变了，key 就不同，缓存就 miss。而零件缓存（比如嵌入）的 key 只依赖一个变量（文本），稳定得多，命中率高得多。
第二，答案不该被缓存。模型生成是不确定的，同一问题两次回答可能措辞不同；更重要的是，缓存一个旧答案会让 Agent 的「思考」完全冻结，用户改了后续追问、语料更新了，它还在吐旧报告。缓存中间产物（检索、嵌入）没有这个问题——它们是确定的事实，不是生成的判断。
第三，越粗的缓存越难失效。语料更新时，你希望失效的是「检索结果」这一层；可如果只缓存了整机结果，你就只能整份作废，连带着没变的嵌入也一起丢了。分层的零件缓存让失效可以精准到最小单位。
| 缓存粒度 | key 依赖 | 命中率 | 失效难度 | 建议 |
|---|---|---|---|---|
| 整机 run 结果 | 全部输入 | 低 | 高，只能整份作废 | 不推荐 |
| 检索结果 | workspace + query + top_k | 中 | 低，语料更新时删对应 key | 推荐 |
| 嵌入向量 | workspace + text | 高 | 极低，几乎不用失效 | 强烈推荐 |
一句话：缓存零件，不缓存整机。零件是确定的事实、依赖少、命中高、失效准；整机是生成的判断、依赖多、命中低、失效糙。
6.1.6 TTL 策略表
缓存不能永久，TTL 就是「这份数据多久后算过期」的承诺。三类缓存的时效性诉求完全不同：
| 缓存对象 | 建议 TTL | 依据 | 过期后果 |
|---|---|---|---|
| 嵌入向量 | 24 小时 | 文本→向量是纯函数，几乎不变 | 无，重算即可 |
| 检索结果 | 5 分钟 | 语料可能更新、索引可能重建 | 可能返回略旧的召回集 |
| 会话摘要 | 30 分钟 | 对话在推进，摘要随时变旧 | 上下文不一致 |
三行的规律是：越「纯函数」的越可以长 TTL，越「随状态变」的越要短。嵌入是纯粹的 text -> vector，同一文本永远同一个向量，24 小时（甚至更久）都安全；检索结果依赖语料和索引，语料一更新旧结果就错了，所以只能给分钟级；会话摘要是对话状态的快照，对话每多一轮它就该重算，TTL 最短。
TTL 还有两个工程细节。其一，TTL 要带抖动——如果所有 key 都在同一时刻批量写入、又设同一个 TTL，它们会同时过期，形成「雪崩」，所以真实系统会给 TTL 加 ±10% 的随机偏移。其二，TTL 是兜底不是精度——别指望「TTL 一到数据就准」，主动失效（语料更新时显式删 key）才是保证新鲜度的正路，TTL 只负责清理那些没人管、已经烂掉的条目。
6.1.7 穿透、击穿、雪崩在 Agent 场景的形态
教科书上缓存三大故障是抽象概念，落到 Agent 服务里各有具体长相：
| 故障 | 通用定义 | Agent 场景的具体形态 | 对策 |
|---|---|---|---|
| 穿透 | 查不存在的数据，次次打到后端 | 用户问语料里根本没有的问题，key 永远 miss | 缓存空结果（短 TTL），或布隆过滤器挡 |
| 击穿 | 某个热 key 过期瞬间，大量请求同时回源 | 热门问题缓存刚过期，一波请求全去跑嵌入 | 单飞（single-flight）：只放一个请求回源，其余等结果 |
| 雪崩 | 大量 key 同时过期，回源洪峰压垮后端 | 早高峰批量写入的缓存同时到期 | TTL 加随机抖动，分批预热 |
穿透在 Agent 场景特别常见：用户问的很多问题语料里就是没有答案，检索回空，可如果空结果不缓存，每次问都白跑一遍昂贵的检索。对策是把「空」也缓存起来（给一个更短的 TTL，比如 30 秒），这样同一个无效问题短时间内只回源一次。
击穿的典型是「热门问题刚好过期」：一个被高频问的问题，缓存 TTL 到点，此刻恰好有几个并发请求同时发现 miss，一起冲向嵌入服务——这就是缓存最不该造成的负载放大。对策是单飞锁：miss 时不是每个请求都回源，而是抢一把锁，抢到的那个回源并写缓存，其余请求短暂等待后重读缓存。Redis 的 SET key value NX 天然适合做这把锁：
def get_or_load(self, key: str, loader, ttl: int):
raw = self.r.get(key)
if raw is not None:
return json.loads(raw), "hit"
lock = f"{key}:lock"
# SET NX 抢锁，锁自身也带 TTL 防死锁
if self.r.set(lock, "1", nx=True, ex=5):
try:
value = loader() # 只有一个请求回源
self.r.set(key, json.dumps(value), ex=ttl)
return value, "miss"
finally:
self.r.delete(lock)
# 没抢到锁：短暂等待后重读，仍 miss 则自己回源兜底
time.sleep(0.05)
raw = self.r.get(key)
if raw is not None:
return json.loads(raw), "hit-after-wait"
return loader(), "miss-fallback"
单飞的本质是把并发的回源压成一次：锁只保护「回源」这一小段，写缓存后立刻释放，其余请求要么等到结果、要么在兜底路径上自己回源。注意兜底那行——锁不能成为单点故障，如果持锁的请求崩了，等待的请求必须能自己顶上，否则一次崩溃会拖垮整个热门查询。
雪崩的诱因在 Agent 里往往是「预热点」：服务启动时批量预热了一堆热门 query 的缓存，都设了同样的 TTL，于是它们会在同一秒集体过期。对策是给 TTL 加抖动，让过期时间摊开；再配合分批预热，避免洪峰。
一句话记牢：穿透是「查了不存在的东西」，击穿是「一个热点突然失效」，雪崩是「一片缓存同时失效」——三者的对策分别是「缓存空值」「单飞锁」「TTL 抖动」。
真实环境会怎样：本节的回源是 time.sleep 模拟的，真实耗时由嵌入 API 的网络往返和向量库的检索规模决定，可能是几百毫秒到几秒，缓存收益更大。真实系统里还有三个本机演示没覆盖的点：一是缓存要落指标（命中率、回源耗时、key 数量）进 13 章的看板，否则命中率掉了没人知道；二是写缓存要考虑一致性，语料更新时要主动失效对应的检索 key，否则用户会拿到旧结果；三是Redis 通常不是单点，生产上是哨兵或集群，key 的分布与故障转移都要考虑。本节刻意只做「检索 + 嵌入」两个最纯的缓存，正是因为它们的失效模型最简单——把最简的做对，再去碰有状态的会话缓存。
阅读导航：上一节：5.3 查询优化：EXPLAIN 定位慢查询并加索引 · 下一节：6.2 语义缓存：相似问题复用与错配检测 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。