《AI Agent 工程实战》11.2 检索注入:长期记忆接入 run 并量化收益

记忆写进来不等于用得上。本节把长期记忆接进一次 run:检索、排序、按 token 预算截断、再注入,并讲清 system、首条 user、工具返回三种注入位置对模型注意力的影响。最后用一个 10 题的评测小集真跑对照,量化「不注入」与「注入」的得分差。

11.2 检索注入:长期记忆接入 run 并量化收益

11.1 把记忆可靠地写进了库,但库里躺着再多的记忆,模型在生成时也看不见——它只看得见这一次请求的上下文。记忆要从「存储」变成「能力」,必须经过一道注入:在 run 开始前把相关的记忆检索出来、拼进提示词,模型才用得上。这一节做这件事,并且量化它到底带来多少收益——「感觉有用」不算数,得分表才算数。

本节把 ResearchCopilot 推进到「记忆接入 run」:检索 → 排序 → 按 token 预算截断 → 注入,并真跑一个 10 题评测小集,对比「不注入记忆」与「注入记忆」的得分。

11.2.1 从检索到注入的四步

一条记忆从库里到模型眼前,走四步,每一步都有取舍:

步骤做什么关键取舍
检索用查询找相关记忆用向量还是关键词(第 6 章混合检索)
排序按相关度降序相似度单一打分还是加时间衰减
截断按 token 预算裁到能放下预算给多少、超了砍谁
注入拼进提示词放 system、首条 user 还是工具返回

四步里最容易忽视的是截断。检索通常返回「越多越好」,但上下文窗口是硬的:塞进去十条记忆,可能挤掉本该给工具结果的位置,也可能把模型注意力稀释到无关条目上。所以注入前必须有一道预算闸门。

检索与排序复用第 6.2 节的思路:查询与记忆都算向量,余弦相似度降序。本节沿用 11.1 的本地哈希向量,保证零密钥可复现:

# memory/inject.py
MEMORIES = [
    {"id": 1, "slot": "语言",     "content": "用户偏好用 Python 写 Agent"},
    {"id": 2, "slot": "项目",     "content": "用户在 acme 团队,workspace 是 ws_acme"},
    {"id": 3, "slot": "报告",     "content": "用户的报告必须带引用来源"},
    {"id": 4, "slot": "项目",     "content": "用户在做本地知识库问答项目"},
    {"id": 5, "slot": "回答风格", "content": "用户希望用简体中文回复"},
    {"id": 6, "slot": "回答风格", "content": "用户偏好简洁的代码示例"},
    {"id": 7, "slot": "工具",     "content": "用户平时用 VS Code"},
]

def retrieve(query: str, k: int = 3) -> list[dict]:
    qv = embed(query)
    scored = [(float(np.dot(qv, embed(m["content"]))), m) for m in MEMORIES]
    scored.sort(key=lambda x: x[0], reverse=True)
    return [m for _, m in scored[:k]]

真实系统里「排序」往往不止看相似度。一个常见改进是加时间衰减:同样相关的两条记忆,更新的那条更该被召回——用户上周的偏好比去年的偏好更可信。做法是把相似度乘一个随年龄递减的系数(score = cosine * exp(-age / tau)),tau 是半衰期。本节为保持对照干净,只用纯相似度排序,把衰减留作生产侧的增强。

11.2.2 按 token 预算截断

截断不能拍脑袋,要可解释:从相关度最高的记忆开始放,放不下就停,而不是放不下就硬塞或随机砍。这样「为什么这条没进去」永远有答案——它排在预算边界之外。

def build_memory_block(query: str, budget: int = 60, k: int = 3) -> tuple[str, list[dict]]:
    """检索 → 排序 → 按 token 预算截断 → 拼成注入块。"""
    picked, used = [], 0
    for m in retrieve(query, k):                 # retrieve 已按相似度降序
        line = f"- {m['content']}"
        cost = tok(line)
        if used + cost > budget:                 # 超预算就停,保证截断可解释
            break
        picked.append(m)
        used += cost
    return "\n".join(f"- {m['content']}" for m in picked), picked

tok 是粗略的 token 计数(CJK 每字一个、拉丁词每词一个):

import re

def tok(text: str) -> int:
    """粗略 token 计数:CJK 每字 1 个,拉丁词每词 1 个。"""
    return len(re.findall(r"[一-鿿]|[A-Za-z0-9]+", text))

真实系统里要用对应模型的分词器精确计数,但结构不变:预算是硬上限,超了就截。预算值本身是调参项——给太小,关键记忆进不来;给太大,挤占工具结果与对话历史的位置。一个务实起点是总窗口的 10%~20%。

11.2.3 注入位置的三种取舍

同一块记忆,放在提示词的不同位置,对模型注意力的影响不同:

注入位置形状对注意力的影响适用
system追加到系统提示词全局、稳定,但位置靠前易被长上下文稀释稳定的长期偏好、身份
首条 user拼在用户问题前离问题最近,注意力最强与当前问题强相关的记忆
工具返回作为 recall 结果回灌模型主动取,最相关但不确定会不会取可能有用、不必强塞的记忆

三种形状的角色序列:

system       角色序列: ['system', 'user']
first-user   角色序列: ['system', 'user']
tool-return  角色序列: ['system', 'user', 'tool']

system 位置的具体构造——把记忆块追加到系统提示词末尾,再原样交给 run:

block, _ = build_memory_block(user_input)
system = "你是 ResearchCopilot 的研究助理。\n\n[长期记忆]\n" + block
result = run(model, registry, user_input, system=system)

run 是卷一立下的循环契约,签名一字未改。记忆注入对引擎是透明的:引擎只看到「一个更长的 system 字符串」,不需要知道里面哪几行来自长期记忆。这个设计的好处是——记忆能力不侵入引擎,第 8 章的并发、第 10 章的工具都感知不到它的存在。

一个经验法则:「必须出现」的硬约束放 system(用户身份、安全规则),「跟这轮强相关」的放首条 user(离问题近、注意力足),「可能有用」的走工具(交给模型按需取,对应卷一 9.3 的「记忆即工具」)。本节对照实验统一用 system 位置,因为它是确定性最强、最容易测的一种。

位置为什么重要?因为真实模型对上下文里的信息不是均匀关注的——「中间遗忘(lost in the middle)」现象表明,模型对开头和结尾的内容记得最牢,夹在中间的容易被忽略。把记忆塞进一段很长的 system 提示词中部,可能还没放在用户问题正上方管用。所以「放哪」不是格式问题,是注意力预算问题。三种位置没有绝对优劣,只有匹配:稳定的东西放远处,易变的东西放近处。

11.2.4 真跑:一个「有记忆才能答对」的评测小集

要量化收益,得先有一把尺子。构造 10 个问题的评测小集,5 个必须靠记忆才能答对(记忆题),另外 5 个靠通用知识就能答(自足题)。记忆题的正解只存在于记忆库里,不注入就无从得知:

@dataclass
class Q:
    qid: int
    text: str
    kind: str            # "memory" | "self"
    answer: str          # 判定正确的关键子串
    need: str = ""       # memory 题:上下文里必须出现的事实子串

QUESTIONS = [
    Q(1, "用户偏好用什么编程语言写 Agent?", "memory", "Python", "Python"),
    Q(2, "用户所在的 workspace 标识是什么?", "memory", "ws_acme", "ws_acme"),
    Q(3, "用户的报告需要满足什么?", "memory", "引用", "引用"),
    Q(4, "用户目前在做什么项目?", "memory", "知识库", "知识库"),
    Q(5, "用户希望我用什么语言回复?", "memory", "简体中文", "简体中文"),
    Q(6, "2 加 2 等于几?", "self", "4"),
    Q(7, "Python 里列表和元组的主要区别是什么?", "self", "可变"),
    Q(8, "HTTP 200 状态码表示什么?", "self", "成功"),
    Q(9, "JSON 对象的键必须是什么类型?", "self", "字符串"),
    Q(10, "TCP 和 UDP 哪个提供可靠传输?", "self", "TCP"),
]

假模型按「上下文里有没有所需事实」作答:记忆题所需事实只在注入块里出现,没注入就答「无法确定」。这样得分只由是否注入决定,排除了其它变量:

SELF_ANSWERS = {6: "4", 7: "列表可变、元组不可变", 8: "请求成功",
                9: "字符串", 10: "TCP 提供可靠传输"}

def fake_model(q: Q, context: str) -> str:
    if q.kind == "self":
        return SELF_ANSWERS[q.qid]
    return q.answer if q.need in context else "无法确定"

def score(inject: bool) -> tuple[int, list[tuple[int, str, str]]]:
    hit, detail = 0, []
    for q in QUESTIONS:
        ctx = build_memory_block(q.text)[0] if inject else ""
        resp = fake_model(q, ctx)
        ok = q.answer in resp
        hit += ok
        detail.append((q.qid, "OK " if ok else "MISS", resp))
    return hit, detail

score(inject) 是唯一的开关:inject=True 时给模型拼上检索块,inject=False 时上下文为空。先看问题 1 的注入块(真连本机 Postgres 读取):

注入块:
- 用户偏好用 Python 写 Agent
- 用户偏好简洁的代码示例
- 用户在做本地知识库问答项目
命中记忆 id: [1, 6, 4]  token: 32

检索按相似度取前 3 条,拼成 32 token 的注入块。再看两个条件的完整对照:

== 对照实验:不注入 vs 注入 ==

[不注入记忆] 得分 5/10
  Q1  MISS  无法确定
  Q2  MISS  无法确定
  Q3  MISS  无法确定
  Q4  MISS  无法确定
  Q5  MISS  无法确定
  Q6  OK   4
  Q7  OK   列表可变、元组不可变
  Q8  OK   请求成功
  Q9  OK   字符串
  Q10 OK   TCP 提供可靠传输

[注入记忆] 得分 10/10
  Q1  OK   Python
  Q2  OK   ws_acme
  Q3  OK   引用
  Q4  OK   知识库
  Q5  OK   简体中文
  Q6  OK   4
  Q7  OK   列表可变、元组不可变
  Q8  OK   请求成功
  Q9  OK   字符串
  Q10 OK   TCP 提供可靠传输

汇总: 不注入 5/10  ->  注入 10/10  (提升 5 题)

数字很干净:不注入 5/10,注入 10/10,提升 5 题。而且提升的恰好是 5 道记忆题(Q1~Q5),自足题(Q6~Q10)两个条件下都对——这正说明记忆注入只影响它该影响的部分,没有引入噪声、没有让自足题反而变差。

11.2.5 一次 run 里检索几次

还有一个容易被忽略的决策:注入发生在 run 的哪个时刻、发生几次。三种常见节奏:

节奏做法代价
run 前一次用原始问题检索一次,注入后不再动多跳 run 后半段问题变了,记忆可能过时
每个 step每轮都用当前消息检索准,但每轮一次向量计算与 IO
按需(工具)交给模型调 recall省,但不保证被调用

本节对照实验用的是第一种:run 开始前用用户问题检索一次。它实现最简单、行为最确定,适合「用户偏好」这类整轮都成立的记忆。多跳检索场景(ResearchCopilot 的典型形态)更适合第二种——第一跳拿到线索后,问题焦点会漂移,用最初的问题去检索后面几跳的记忆可能已经不对题。而第三种就是卷一 9.3 的「记忆即工具」,把决定权交给模型,代价是漏调。

工程上的常见组合是前两种叠加:run 前用问题做一次粗注入(保证关键偏好一定在场),每个 step 用当前查询再做一次精注入(跟住焦点漂移)。多出来的检索成本可以用第 6.1 节的缓存压下去——同一 workspace 下相似查询命中缓存,就不必重复算向量。

11.2.6 为什么这个对照可信

一个「注入后分数变高」的结论,很容易是假象。这个对照之所以可信,是因为它只改了一个变量:

受控项两个条件是否相同
问题集相同(同一份 QUESTIONS)
假模型脚本相同(同一个 fake_model 函数)
评分器相同(同一套 q.answer in resp 判定)
检索实现相同(同一个 retrieve)
注入开关唯一变量(inject 布尔)

除了 inject 这个布尔,两次跑用的是同一份代码、同一批数据、同一个模型。所以得分差只能归因于注入本身。这比「用两个不同模型比一比」可信得多——后者你分不清是记忆起了作用,还是模型换了。

也要清醒地看到这把尺子的边界:假模型是脚本化的,它的「答对」只是字符串包含判定,衡量的是注入链路是否把正确记忆送到了模型面前,而不是模型会不会用。真实模型可能拿到了记忆却答偏(幻觉、忽略),那是第 12 章评测门禁要管的事。所以本节量化的结论要精确表述:注入链路本身是通的、有效的;真实模型下的端到端收益需要用真模型重跑同一套问题集。

11.2.7 真实环境与验收清单

真实环境里,四步都会更复杂:检索换成第 6 章的混合检索(关键词 + 向量 + 重排),排序加入时间衰减与来源可信度,截断用真分词器精确计费,注入位置按「硬约束 / 强相关 / 可能有用」分流。但四步的结构和「只改一个变量」的评测纪律不变。

还要防两个反向风险。一是注入有害:塞进去的记忆如果与当前问题无关,可能把模型带偏,让本来能答对的题反而答错——所以评测集里一定要放自足题当对照组,它们的分数不该因注入下降(本节里 Q6~Q10 两条件都是 5/5,就是这个保障)。二是注入泄露:多租户下检索必须带 workspace_id(第 4.2 节),否则会把别的租户的记忆注进这次 run,这既是隐私事故也是评分污染。注入前过一遍租户过滤,是硬性要求。

验收清单:

  • 记忆在 run 开始前被检索并注入,而不是只躺在库里。
  • 注入前有 token 预算闸门,超预算按相关度降序截断。
  • 注入位置有明确规则:硬约束进 system,强相关进首条 user,可能有用走工具。
  • 评测集包含「有记忆才能答对」的题,且对照只改注入这一个变量。
  • 真实模型下用同一套问题集复跑,确认端到端收益。

记忆能写、能用之后,还剩最后一个、也是最容易被工程团队拖到上线才想起的问题:记忆是用户数据。哪些内容根本不该被记?记下的什么时候该过期?用户说「删掉我的数据」时,怎么保证真的删干净?下一节把这三件事收进一套治理机制。

阅读导航:上一节:11.1 写入流水线:抽取、去重、冲突消解 · 下一节:11.3 记忆治理:TTL、隐私与按用户删除 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「agent」更多文章

  1. 《AI Agent 高级编程》目录
  2. 《AI Agent 高级编程》11.3 全景回顾:从零到框架的完整路径与展望
  3. 《AI Agent 高级编程》11.2 真实 API 冒烟:假模型 vs 真模型量化差距