8.2 记忆检索注入
8.1 节的 BufferMemory 与 SummaryMemory 解决的是「怎么留住最近的对话」,但它们的共同前提是按时间取舍:留最近 N 条、压最老的一批。这个前提有个漏洞——用户真正需要的那条信息,未必在最近,也未必在最老,它可能藏在几百轮之前的某个角落。一句「你记得我上次说的代码示例偏好吗」,缓冲版答不上来,摘要版也可能因为那轮太早被压没了。
本节把 minagent 推进到
minagent/retrieval.py:把记忆存成带文本的条目列表,用关键词与中文二元组给条目打分,实现retrieve(query, k)按相关度召回,再把命中的记忆拼进本轮上下文。本章的检索只用字符串匹配,不上向量——嵌入与向量索引是第 9 章的主题。
8.2.1 从「按时间」到「按相关度」
换个视角看记忆:它不该只是一串按时间排的消息,而是一个可以被查询的集合。每条记忆是一个独立条目,检索时拿「当前的问题」去和每条记忆比相关度,挑出最高的几条注入上下文。
这个转变带来两个好处:
- 不再依赖时间顺序。上周说的话和昨天说的话地位平等,谁相关谁被召回。
- 只注入相关的。缓冲版无论问什么都带上最近 N 条,检索版只带和当前问题有关的几条,token 花得更值。
两者放在一起对比,差别一目了然:
| 维度 | 缓冲 / 摘要(8.1) | 检索(本节) |
|---|---|---|
| 取舍依据 | 时间(新近优先) | 相关度(命中优先) |
| 输入 | 一段对话流 | 一个可查询的条目集合 |
| 问「上周的偏好」 | 答不上来 | 能召回 |
| 算法复杂度 | 一个滑动窗口 | 打分 + 排序 |
| 是否依赖外部模型 | 摘要版需要 | 本节不需要(向量版需要) |
代价是:检索需要一套「相关度」的算法。工业界的主流答案是向量检索——把每条记忆编码成向量,用余弦相似度比对语义。但它要嵌入模型、要向量库、要网络或本地模型文件,属于第 9 章的内容。本节先用一个零依赖的近似:关键词与子串匹配。它不懂语义,但足够说明「检索注入」这套流程的骨架,而且完全确定、可断言。
把检索的骨架先讲清楚,还有一个工程上的理由:检索、排序、注入这三步,无论用关键词还是向量,形状是完全一样的。 差别只在「怎么算分」这一个函数。所以本章先把流程跑通,第 9 章换掉打分实现即可,不必推倒重来。
8.2.2 关键词打分:不上向量的检索
先定义记忆条目。它至少要有文本,最好还有一个时间戳用于并列时排序:
# minagent/retrieval.py
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class MemoryItem:
text: str
ts: int = 0
打分的关键在于「把一句话切成什么单元去比对」。英文好办,按空白和标点切词即可;中文没有空格,直接按字符比对噪声太大。这里用一个折中:英文取单词,中文取相邻两字的二元组(bigram)。「代码示例偏好」会被切成 代码 / 码示 / 示例 / 例偏 / 偏好,其中 示例、偏好 就是有效的语义单元。
import re
_WORD = re.compile(r"[a-z0-9]+|[一-鿿]+")
def terms(text: str) -> list[str]:
out: list[str] = []
for tok in _WORD.findall(text.lower()):
if "一" <= tok[0] <= "鿿":
if len(tok) == 1:
out.append(tok)
else:
out.extend(tok[i:i + 2] for i in range(len(tok) - 1))
else:
out.append(tok)
return out
terms 对中英混排的输入各切各的,实测输出:
terms("代码示例偏好") -> ['代码', '码示', '示例', '例偏', '偏好']
terms("Python Agent 项目") -> ['python', 'agent', '项目']
可以看到 码示、例偏 这类跨词边界的二元组是「副产品」——它们几乎不会和别的句子撞上,因此对打分的贡献接近零,不影响判断。真正起作用的是 示例、偏好 这种有意义的相邻字对。英文部分则原样按单词切分并转小写,Python 与 python 会被视作同一个词。
有了切分,打分就是查询词与记忆词的集合交集大小:
def score(query: str, item: MemoryItem) -> int:
q = set(terms(query))
m = set(terms(item.text))
return len(q & m)
用集合交集而不是子串计数,是为了避免「同一关键词出现十次就加十倍分」——那会让一条啰嗦的长记忆压过一条精准的短记忆。交集大小天然把重复词压成 1,更接近「命中了几个不同的概念」。
8.2.3 retrieve:排序、过滤、截断
打分有了,检索就是「算分、丢掉零分、按分排序、取前 k」四步:
def retrieve(query: str, memories: list[MemoryItem], k: int = 3) -> list[MemoryItem]:
scored = [(score(query, it), it) for it in memories]
scored = [(s, it) for s, it in scored if s > 0]
scored.sort(key=lambda pair: (-pair[0], -pair[1].ts))
return [it for _, it in scored[:k]]
四处细节:
- 过滤零分(
if s > 0):完全不相干的记忆不该被注入,否则等于给模型添噪声。 - 排序键是元组
(-score, -ts):先按分数降序,分数相同时按时间戳降序(新的优先)。用一个元组就把「相关性优先、新鲜度次之」两个规则表达完了。 k截断:召回条数必须封顶。检索的诱惑是「多带几条总没坏处」,但每条记忆都占 token,而且塞太多会让模型分不清主次。k通常取 2~5。- 返回原条目,不是分数。调用方只关心「召回了什么」,不关心分数是多少——分数是内部实现。
k 该取多少,没有标准答案,但有可参照的量级:
| 场景 | 建议 k | 理由 |
|---|---|---|
| 闲聊型助手 | 1–2 | 多数问题用不到历史事实,多了是噪声 |
| 个人助理 | 3–5 | 需要综合偏好、背景、近期结论 |
| 知识密集任务 | 5–10 | 召回即检索,宁可多带再由模型筛 |
k 太大最典型的坏处是稀释:模型面对一屏记忆,反而不知道该用哪条。真实的经验是「少而准」胜过「多而全」,先从 3 起步,观察漏召回再往上调。
retrieve 是整个模块唯一需要测试的逻辑,它接受一个纯列表、返回一个纯列表,没有模型、没有网络、没有状态。这种「纯函数检索」正是本章坚持不用向量的一个额外好处:可以拿一组固定的记忆和查询,断言召回结果逐条一致。
8.2.4 注入:拼进本轮上下文
检索出记忆后,要把它们变成模型能读的消息。做法是每条记忆包成一条 system 消息,放在用户问题之前:
def inject(query: str, memories: list[MemoryItem], k: int = 3) -> list[Message]:
hits = retrieve(query, memories, k)
msgs: list[Message] = []
for it in hits:
msgs.append(Message(role="system", content=f"相关记忆:{it.text}"))
msgs.append(Message(role="user", content=query))
return msgs
为什么用 system 而不是 user?因为记忆是背景事实,不是用户此刻在说的话。混进 user 角色会让模型以为用户在重复陈述这些内容,干扰对话的自然感。用 system 明确标注「这是我之前了解到的信息」,模型更容易把它当参考资料用。
注入的条数是有上限的:k 条记忆 + 1 条用户消息。这个列表接着可以交给 Agent.run(),与它自己的 history 拼在一起。检索注入与 history 不冲突——history 管「这一轮对话的连续上下文」,检索管「跨会话的长期事实」,两者叠加才是完整的上下文。
把两层拼起来,组装请求的逻辑大致是:
def build_messages(system: str, history: list[Message],
memory: list[MemoryItem], query: str,
k: int = 3) -> list[Message]:
msgs = [Message(role="system", content=system)]
msgs.extend(history) # 短期:本会话的连续上下文
msgs.extend(inject(query, memory, k)) # 长期:跨会话召回的事实
return msgs
顺序上,system 在最前、history 居中、召回的记忆紧挨着用户问题,是常见排布。把召回内容放在离问题最近的位置,模型的注意力更容易落在它上面。注意这里 inject 返回的列表已经包含用户问题,所以调用方不必再单独追加一次——重复追加会让模型看到两个一模一样的问题。
8.2.5 跑一遍:召回上周的偏好
造五条记忆,其中一条是用户上周说的偏好,其余是干扰项。拿一句「你记得我上次说的代码示例偏好吗」去检索:
# demo_retrieval.py
memories = [
MemoryItem(text="用户偏好简洁的代码示例,不喜欢大段解释。", ts=1),
MemoryItem(text="用户在做一个 Python 本地知识库问答的 Agent 项目。", ts=2),
MemoryItem(text="用户对咖啡的偏好是拿铁。", ts=3),
MemoryItem(text="用户说上周调试过向量检索的性能问题。", ts=4),
MemoryItem(text="用户希望回答里带可运行的最小示例。", ts=5),
]
query = "你记得我上次说的代码示例偏好吗?"
hits = retrieve(query, memories, k=2)
print("命中条数:", len(hits))
for h in hits:
print(" -", h.text)
context = inject(query, memories, k=2)
print("注入消息条数:", len(context))
fake = FakeModel([
"记得,你偏好简洁、可运行的最小示例,我会保持这个风格。",
])
resp = fake.complete(context, [])
print("模型回答:", resp.message.content)
print("无关问题的命中:", len(retrieve("今天天气怎么样", memories, k=2)))
实测输出(Python 3.14.6):
命中条数: 2
- 用户偏好简洁的代码示例,不喜欢大段解释。
- 用户希望回答里带可运行的最小示例。
注入消息条数: 3
模型回答: 记得,你偏好简洁、可运行的最小示例,我会保持这个风格。
无关问题的命中: 0
逐行读一遍:查询里含 代码 / 示例 / 偏好 等词,命中了第 1 条(含「代码示例」「偏好」)和第 5 条(含「示例」),两条都是关于示例偏好的;关于咖啡的拿铁、关于向量检索性能的第 4 条都没被召回——它们和查询没有共同词。注入后是 2 条记忆 + 1 条用户问题 = 3 条消息,模型据此给出「记得,你偏好简洁、可运行的最小示例」。
最后一行是反例验证:换一个「今天天气怎么样」,命中数是 0。这说明检索确实在按相关度过滤,而不是无脑返回全部记忆。零命中时 inject 只返回用户那条消息,Agent 会正常回答「我不知道」——检索的「找不到」是一个合法结果,不该报错。
8.2.6 关键词检索的三个陷阱
关键词打分便宜、确定,但它的短板必须在用之前心里有数:
| 陷阱 | 表现 | 缓解 |
|---|---|---|
| 不懂同义 | 查「示例」召回不到写着「例子」的记忆 | 维护同义词表,或上向量检索(第 9 章) |
| 中文切分粗 | 二元组会造出 例偏 这种无意义片段 | 只影响打分权重,通常不致命;可用分词库替换 |
| 关键词越长越吃亏 | 长句查询词多,交集反而被稀释 | 归一化分数,或只取查询里的实词 |
第一个陷阱最要命。用户问「有没有推荐的做法」,记忆里写的是「用户偏好简洁风格」——两句话一个共同词都没有,关键词检索完全召回不到,尽管它们在语义上高度相关。这正是向量检索存在的理由:它比的是语义相近,不是字面重合。第 9 章会用一个本地嵌入模型把这件事补上,届时本节的 retrieve 只需换掉 score 的实现,inject 与调用方一行都不用改。
第三个陷阱也值得展开:分数是「交集大小」,没有除以查询长度。所以一个 30 个词的查询,即使命中 5 个词,也未必比一个 3 个词的查询命中 2 个词更「相关」。要修正,可以改成 Jaccard 相似度 |q ∩ m| / |q ∪ m|,代价是短记忆容易被长查询稀释。没有普适最优,只有按场景调。
对付这些陷阱最实用的手段是写一组固定的召回用例:准备若干「查询 → 期望命中的记忆」,改一次打分函数就跑一遍,看有没有回归。这正是纯函数检索的价值——它不依赖模型,所以可以像测普通函数一样测它。第 16 章的评测会把这套思路正式化。
8.2.7 真实模型下会怎样
本节的召回结果由 score 完全决定,给定记忆和查询,每次结果逐字节一致,非常适合写测试。真实场景里,检索这一层会换上向量模型,差异集中在三点:
- 召回靠语义而非字面。「有没有推荐的做法」与「用户偏好简洁风格」在向量空间里接近,能被召回;关键词版则漏掉。这是从关键词到向量最大的收益。
- 相似度是浮点数,有阈值问题。关键词检索的「命中/不命中」是二元的(交集是否为 0),向量检索要设一个相似度阈值,太低会召回噪声,太高会漏掉相关项,需要按数据调。
- 召回条数
k与 token 预算联动。向量召回的记忆往往更长,k设大了会挤占上下文。真实系统常把「召回条数」和「记忆长度」一起纳入预算,必要时对召回内容再摘要。
还有一点:真实模型对注入的 system 记忆并非照单全收。它可能忽略、也可能过度依赖某条记忆。要验证「记忆到底有没有起作用」,得靠第 16 章的评测——用假模型固定召回结果,断言模型在注入前后拿到的 messages 确实不同。本节已经把这个可断言的接缝留好了,只要替换 score 的实现,整套注入流程无需改动。
检索解决了「怎么把记忆取出来」,但记忆条目本身从哪来?如果什么都往记忆里塞,检索会被垃圾淹没;如果塞得太少,又会漏掉关键信息。下一节 8.3 记忆写入策略 回答「什么值得记」。想回看短期记忆是怎么留住近期对话的,见 8.1 对话缓冲与摘要记忆 。
阅读导航:上一节:8.1 对话缓冲与摘要记忆 · 下一节:8.3 记忆写入策略 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。