记忆系统
一句话定义：让 Agent 跨步骤、跨会话保持信息，分短期、长期、工作记忆三层，是长任务不"失忆"的关键。
1. 为什么需要记忆
- LLM 上下文窗口有限，长任务信息易溢出。
- 跨会话需持久化用户偏好、历史决策。
- 无记忆的 Agent 每次从零开始，无法积累。
2. 三层记忆
| 层级 | 内容 | 存储 | 生命周期 |
|---|---|---|---|
| 短期记忆 | 本次会话上下文与中间结果 | 对话历史/上下文 | 会话内 |
| 工作记忆 | 当前任务草稿/scratchpad | 临时变量 | 任务内 |
| 长期记忆 | 跨会话复用的事实/偏好/决策 | 向量库/DB/KV | 持久 |
3. 长期记忆实现
向量记忆
- 把记忆片段嵌入向量库，按语义检索召回。
- 适合"回忆相关往事"。
- 代表：Mem0、LangChain Memory + 向量库。
结构化记忆
- 用 KV/关系库存结构化事实（用户偏好、实体属性）。
- 适合"精确查询"。
- 代表：知识图谱、JSON profile。
摘要记忆
- 定期把会话摘要存档，避免全量保留。
- 平衡保真与省 token。
4. 记忆管理挑战
- 过期与纠错：记忆可能过时或错误，需更新/失效机制。
- 隐私：敏感信息不应长期留存，需脱敏/删除策略。
- 检索质量：向量召回不准会引入错误记忆。
- 上下文膨胀：记忆注入过多会淹没当前任务。
- 一致性：多记忆片段可能冲突，需仲裁。
5. 设计要点
- 分层：短期放对话，长期落库，按需检索注入。
- 摘要压缩：长会话定期摘要，保留关键决策与硬约束。
- 元数据：记忆带时间/来源/置信度，便于过滤。
- 遗忘机制：设过期或重要性衰减，避免无限堆积。
- 检索 + 注入：只注入与当前任务相关的记忆，而非全量。
6. 架构设计与 LLM 交互流程
6.1 整体架构
┌─────────────────────────────────────────────────┐
│ Agent 主循环 │
│ (感知 → 检索记忆 → 规划 → 行动 → 写回记忆) │
└───────────────┬─────────────────────────────────┘
│
┌───────▼────────┐
│ Memory Manager │ ← 核心：读写/检索/更新/遗忘
└───┬─────┬──────┘
│ │
┌────────▼─┐ ┌─▼──────────────┐
│ 注入器 │ │ 提取器 │
│ (拼prompt)│ │ (从回复抽记忆) │
└──────────┘ └────────────────┘
│
┌────────────▼────────────────────────┐
│ Memory Store (存储层) │
├──────────────┬──────────┬───────────┤
│ 短期(对话历史)│工作(scratch)│长期(向量库)│
└──────────────┴──────────┴───────────┘
6.2 与 LLM 交互的标准流程（每个 Agent 步骤）
用户输入
│
▼
1. 检索：用当前输入/任务 去长期记忆库做向量检索，召回 Top-K 相关记忆
│
▼
2. 组装 Prompt：
┌───────────────────────────────────┐
│ [System] 你是XX助手，遵循XX约束 │
│ [Memory] ← 注入召回的相关长期记忆 │ ← 关键
│ [Scratchpad] ← 工作记忆(任务草稿/中间结果)│
│ [History] ← 短期记忆(最近N轮对话) │
│ [User] 当前用户输入 │
└───────────────────────────────────┘
│
▼
3. LLM 生成回复
│
▼
4. 提取：从回复/工具调用结果中识别"值得长期记住"的信息
(用户偏好、事实、决策、关键结论)
│
▼
5. 写回：带元数据 {time, source, type, confidence} 存入向量库
│
▼
6. 维护：短期记忆超阈值时触发摘要压缩 → 落到长期；旧记忆按重要性衰减
6.3 关键设计要点
| 环节 | 设计 | 为什么 |
|---|---|---|
| 检索 | 向量检索 + 元数据过滤（时间/类型） | 只注入相关记忆，避免上下文膨胀 |
| 注入 | 记忆放 system 之后、history 之前 | LLM 对靠前指令敏感，记忆作为"背景知识" |
| 提取 | 用 LLM 做"记忆抽取"而非正则 | 自然语言事实难以规则匹配 |
| 更新 | 新旧冲突时用 LLM 仲裁或时间戳覆盖 | 解决"我现在可以吃花生了"这种纠错 |
| 遗忘 | 重要性分数 = 频次 × 时间衰减 × 显式标记 | 防止无限堆积 |
| 摘要 | 每 N 轮把旧对话压成摘要 | 平衡保真与省 token |
6.4 最小实现骨架（伪代码）
class SimpleMemory:
def __init__(self):
self.short_term = [] # 对话历史
self.working = "" # scratchpad
self.long_term = VectorDB() # 长期记忆库
def retrieve(self, query, k=3):
# 1. 向量召回 + 元数据过滤
return self.long_term.search(query, top_k=k)
def build_prompt(self, system, user_input):
memories = self.retrieve(user_input)
return [
{"role":"system","content": system},
{"role":"system","content": f"[相关记忆]\n{format(memories)}"}, # 注入
{"role":"system","content": f"[草稿]\n{self.working}"},
*self.short_term[-10:], # 最近10轮
{"role":"user","content": user_input},
]
def extract_and_store(self, user_input, reply):
# 2. 用 LLM 抽取值得记的事实
facts = llm_extract(user_input, reply)
for f in facts:
self.long_term.upsert(
text=f.text,
meta={"time": now(), "type": f.type, "conf": f.conf}
)
self.short_term.append({"role":"user","content":user_input})
self.short_term.append({"role":"assistant","content":reply})
# 3. 超长则摘要压缩
if len(self.short_term) > 20:
self.summarize_old()
# Agent 主循环
mem = SimpleMemory()
while True:
user = input()
prompt = mem.build_prompt(SYSTEM, user)
reply = llm.chat(prompt)
mem.extract_and_store(user, reply)
print(reply)
6.5 两种记忆注入模式
- 检索增强式（RAG-style）：每轮检索相关记忆拼进 prompt —— 最常用，适合事实型记忆。
- 摘要滚动式：把旧对话压成摘要常驻 prompt 头部 —— 适合连续会话，保证连贯性。
实际系统通常两者结合：摘要保证上下文连贯，检索补充跨会话的长期事实。
6.6 记忆匹配机制详解：新问题如何命中旧记忆
核心问题：用户存了一条记忆后，下次问新问题时，Agent 如何从记忆库中"找到"该用哪条？
匹配流程（四步）
用户新问题
│
▼
1. Query 改写（可选但关键）
原问题可能不是好的检索词，需用 LLM/规则改写
│
▼
2. 向量化：把 query 转成 embedding 向量
│
▼
3. 召回：在记忆库做相似度搜索（余弦/内积），取 Top-K
+ 元数据过滤（时间/类型/置信度）
│
▼
4. 重排（Rerank，可选）：用更精细的模型对 Top-K 再排序
剔除"看起来像但实际无关"的记忆
│
▼
注入到 prompt
关键：为什么不能直接用原问题检索？
| 用户原问题 | 直接检索的问题 | 改写后的 query |
|---|---|---|
| "今天午饭吃啥" | 检不到"花生过敏"那条记忆 | "用户饮食偏好 过敏 午餐推荐" |
| "帮我订张票" | 太泛，召回一堆无关记忆 | "用户出行偏好 购票 目的地" |
Query 改写的两种方式：
- HyDE：让 LLM 先假设一个答案，用假设答案去检索（答案比问题更像记忆内容）
- 多 query 生成：让 LLM 生成 3-5 个不同角度的检索词，并行召回后合并去重
具体例子：从存入到匹配命中
Step 1 - 存入记忆（上次会话）：
用户："我对花生过敏"
→ 提取器抽取事实
→ 向量化存入记忆库：
{
id: "mem_001",
text: "用户对花生过敏",
embedding: [0.12, -0.34, ...], ← 用 embedding 模型算出
meta: {type: "健康偏好", time: "2026-06-28", conf: 0.95}
}
Step 2 - 新会话提问（几天后）：
用户："推荐个午餐"
Step 3 - 匹配过程：
# 1. Query 改写（LLM 生成）
raw_query = "推荐个午餐"
rewritten = llm_rewrite(raw_query)
# → ["用户饮食偏好", "用户过敏食物", "午餐推荐约束"]
# 2. 多路召回
candidates = []
for q in rewritten:
emb = embed(q) # 向量化
hits = vector_db.search(emb, top_k=5, # 余弦相似度搜索
filter={"type":"健康偏好"}) # 元数据过滤
candidates.extend(hits)
# 3. 去重 + 重排
unique = dedupe(candidates)
ranked = rerank(raw_query, unique) # 精排模型打分
# 4. 取 Top-3
top_memories = ranked[:3]
# 结果命中：mem_001 "用户对花生过敏" (score: 0.87)
Step 4 - 注入 prompt：
[System] 你是午餐推荐助手
[Memory]
- 用户对花生过敏 (2026-06-28, 健康偏好) ← 命中的记忆
[History] ...
[User] 推荐个午餐
Step 5 - LLM 生成：
"推荐你试试宫保鸡丁（用腰果替代花生）、番茄炒蛋配米饭..."
相似度计算的原理
记忆匹配的本质是语义相似度，不是关键词匹配：
"我对花生过敏" ──embed──► [0.12, -0.34, 0.56, ...]
"推荐个午餐" ──embed──► [0.08, -0.31, 0.49, ...]
↑
余弦相似度 cos(θ) ≈ 0.87（高度相关）
"我喜欢编程" ──embed──► [0.91, 0.22, -0.15, ...]
↑
余弦相似度 cos(θ) ≈ 0.12（不相关，被过滤）
为什么"午餐"能匹配到"过敏"？因为 embedding 模型在训练时学到了"饮食"这个语义空间内"过敏"和"午餐推荐"经常共现——推荐午餐时必须考虑过敏，这是语义关联而非字面匹配。
匹配失败的常见原因与对策
| 问题 | 现象 | 对策 |
|---|---|---|
| Query 太泛 | 召回一堆无关记忆 | 改写 query + 元数据过滤缩范围 |
| 语义鸿沟 | 用户用"忌口"问，记忆存的是"过敏" | 用更强的 embedding 模型 / 同义词扩展 |
| 记忆太碎 | 单条记忆信息不足，匹配不上 | 存入时做"事件聚合"，一条记忆含完整上下文 |
| 召回噪音 | Top-K 里混入不相关记忆 | 加 Rerank 精排环节 |
| 时效失效 | 旧记忆已过时但分数高 | 元数据带时间，相似度 × 时间衰减因子 |
6.7 召回（Recall）是什么
一句话定义
召回 = 从海量记忆库中，按相似度快速粗筛出一批"可能相关"的候选记忆，供后续精排使用。
类比图书馆找书：你不会一本本翻完所有书（太慢），而是先用索引/分类快速挑出十几本"看起来相关"的，再细看目录决定借哪本。前一步叫召回，后一步叫精排/重排（Rerank）。
为什么需要召回这个环节
记忆库可能有成千上万条记忆，如果每条都用复杂模型算一遍相关性，太慢太贵。所以检索分两阶段：
记忆库 (10万条)
│
▼ [召回] 快 + 粗，用向量相似度
候选集 (Top-50)
│
▼ [重排] 慢 + 精，用交叉编码器
最终结果 (Top-3)
│
▼ 注入 prompt
| 阶段 | 方法 | 特点 | 目标 |
|---|---|---|---|
| 召回 | 向量检索(Bi-Encoder) | 快(毫秒级)、粗 | 宁可多召回，不能漏（高召回率） |
| 重排 | 交叉编码器(Cross-Encoder) | 慢、精 | 剔除噪音，精确排序（高精度） |
核心权衡：召回追求"别漏"，重排追求"别错"。召回的目的是把 10 万条缩小到几十条，让昂贵的精排模型算得过来。
召回的具体工作原理
1. Bi-Encoder 双塔模型（召回主流方式）
把 query 和记忆分别编码成向量，再算余弦相似度：
Query "推荐午餐" ──Encoder──► q_vec ┐
├─► cos(q_vec, m_vec) = 0.87
记忆 "花生过敏" ──Encoder──► m_vec ┘
为什么快：记忆向量可以预先算好存库（离线编码），查询时只需算一次 query 向量，然后和库里的向量做矩阵乘法，GPU/专用索引(ANN)毫秒级返回 Top-K。
存入时：text → encode → 存向量 (一次性成本)
查询时：query → encode → 和库里所有向量比相似度 (只算1次编码)
2. 近似最近邻搜索（ANN）加速
当记忆库很大（>10万），暴力遍历算相似度仍慢，用 ANN 索引近似搜索：
| 算法 | 原理 | 代表 |
|---|---|---|
| HNSW | 构建分层图，沿边快速跳转 | FAISS / Milvus 默认 |
| IVF | 聚类分桶，只搜最近的几个桶 | FAISS |
| PQ | 向量压缩，牺牲精度换内存 | FAISS IVFPQ |
代价：ANN 会损失少量精度（可能漏掉个别相关记忆），换取百倍速度。生产系统通常设
ef_search等参数平衡召回率与速度。
3. 多路召回（Recall 多路融合）
单路向量召回容易漏，实际系统常多路并行召回再合并：
┌─► 向量召回 (语义相似) ──► Top-20
Query ──────────┼─► BM25 召回 (关键词) ──► Top-20
├─► 元数据召回 (同类型) ──► Top-20
└─► 时间召回 (近期记忆) ──► Top-20
│
▼
合并去重 → 候选集 (约50条)
│
▼
Rerank 精排
| 召回路 | 擅长抓什么 | 弱点 |
|---|---|---|
| 向量召回 | 语义相关（"忌口"≈"过敏"） | 专有名词/数字不准 |
| BM25 召回 | 精确关键词（"花生"命中"花生"） | 不懂同义改写 |
| 元数据召回 | 同类记忆（健康偏好聚一起） | 无语义判断 |
| 时间召回 | 近期记忆（刚说过的事） | 与相关性无关 |
多路融合的价值：互补。向量召回漏了"花生"但 BM25 抓到了；BM25 不懂"忌口"但向量召回懂。合并后召回率显著提升。
召回的质量评估指标
召回好不好，看两个指标（常矛盾，需权衡）：
相关记忆
┌────────┐
召回的 ────┤ 命中 ├──── 召回率 = 命中 / 总相关
候选集 │ │
(50条) │ 漏掉 │ ← 召回率低 = 漏了该有的
└────────┘
不相关
命中
精确率 = ─────────── = 命中 / 召回总数
候选集(50)
召回率：该记住的有多少被找回来了（别漏）
精确率：找回来的有多少是对的（别错）
| 指标 | 定义 | 重要性 |
|---|---|---|
| 召回率 Recall | 相关记忆中被找回的比例 | 召回阶段最重要，漏了就彻底没了 |
| 精确率 Precision | 找回的候选中相关的比例 | 次要，噪音可交给 Rerank 过滤 |
经验值：召回阶段追求 Recall@50 ≥ 0.95（50 条候选里几乎不漏），Precision 可低些；Rerank 阶段再追求 Precision@3 ≥ 0.9。
一个完整的召回例子
场景：用户问"我最近能吃坚果吗"，记忆库有 1 万条记忆。
query = "我最近能吃坚果吗"
# 路1: 向量召回（语义）
q_vec = embed(query) # "坚果" ≈ "花生/核桃/杏仁"
vec_hits = vector_db.search(q_vec, top_k=20)
# → 命中 "用户对花生过敏"(0.85), "喜欢核桃酥"(0.72), ...
# 路2: BM25 召回（关键词）
bm25_hits = bm25_index.search(query, top_k=20)
# → 命中 "上周买了坚果礼盒"(精确匹配"坚果")
# 路3: 元数据召回（同类型）
meta_hits = vector_db.search(filter={"type":"健康偏好"}, top_k=10)
# → 命中 所有健康相关记忆
# 合并去重
candidates = dedupe(vec_hits + bm25_hits + meta_hits) # 约 45 条
# 交给 Rerank 精排
ranked = rerank(query, candidates)
top_3 = ranked[:3]
# 最终: "用户对花生过敏"(0.97), "花生是常见坚果"(0.89), "上周买了坚果礼盒"(0.82)
关键点：单路向量召回会漏掉"上周买了坚果礼盒"（因为语义上"买礼盒"和"能不能吃"相关性低），但 BM25 抓到了。多路融合保证不漏。
召回 vs 检索 vs 重排 的关系
这几个词常混用，精确区分：
| 术语 | 范围 | 含义 |
|---|---|---|
| 检索 Retrieval | 最广义 | 从库中找信息的总称，含召回+重排 |
| 召回 Recall | 检索的第一阶段 | 快速粗筛候选集 |
| 重排 Rerank | 检索的第二阶段 | 对候选集精排 |
日常说"记忆检索"通常指整个流程；说"召回"特指第一阶段粗筛。
实践建议
- 召回宁多勿漏：Top-K 设大些（如 50），交给 Rerank 剔噪音，比漏掉强。
- 多路融合：至少向量 + BM25 两路，互补提升召回率。
- ANN 参数调优：
ef_search/nprobe调大提升召回率，牺牲速度。 - 离线评测：构建评测集(查询+相关记忆标注)，监控 Recall@K 退化。
- 记忆质量 > 检索算法：存入时做好事件聚合、去重，比调召回参数有效得多。
6.8 跨 Embedding 模型的检索问题
核心矛盾：记忆是用 LLM 1 的 embedding 模型存的，现在用 LLM 2 提问，两者的向量空间不兼容——维度可能不同，即使维度相同语义几何也不一致，直接算余弦相似度无意义。
为什么不能跨模型直接检索
存入时（LLM 1 / OpenAI text-embedding-3）：
"我对花生过敏" → [0.12, -0.34, 0.56, ...] (1536 维)
提问时（LLM 2 / BGE-large）：
"推荐个午餐" → [0.45, 0.21, -0.08, ...] (1024 维)
↑
维度不同，且语义空间几何完全不同
cos(openAI_vec, BGE_vec) = 无意义数值
不同 embedding 模型是在不同语料、不同目标下训练的，"花生过敏"在 OpenAI 空间和 BGE 空间里的坐标方向没有对应关系。就像用经纬度和用 XY 坐标系描述同一个点——数值不能直接比较。
解决方案（按推荐度排序）
方案 1：统一 embedding 模型（最佳实践）
永远用同一个 embedding 模型做存入和检索，元数据记录所用模型：
# 存入
memory.upsert(
text="用户对花生过敏",
embedding=embed_model_A("用户对花生过敏"), # 用模型 A
meta={
"type": "健康偏好",
"time": "2026-06-28",
"embed_model": "text-embedding-3", # ← 记录用了哪个模型
}
)
# 检索（必须用同一个模型 A）
query_vec = embed_model_A("推荐个午餐") # 仍用模型 A，不是 LLM 2 自带的
hits = vector_db.search(query_vec, top_k=5)
关键认知：Embedding 模型和对话用的 LLM 是两个独立的东西。你可以用 GPT-4 对话 + OpenAI embedding 检索，也可以用 Claude 对话 + BGE embedding 检索。记忆库的 embedding 模型一旦确定就不要换，它不随对话 LLM 变化。
方案 2：模型迁移时全量 Re-embed
当确实要换 embedding 模型时，必须重新编码所有历史记忆：
# 迁移脚本
old_memories = vector_db.list_all()
for mem in old_memories:
new_vec = embed_model_B(mem.text) # 用新模型重新编码
vector_db.update(mem.id, embedding=new_vec,
meta={**mem.meta, "embed_model": "bge-large"})
# 迁移完成后，查询也统一用模型 B
代价：全量重算 embedding，量大时耗时费钱；但这是保证语义一致性的唯一正确方式。
方案 3：双写双检索（过渡期方案）
迁移期间新旧并存，存入时双写，检索时双路召回再合并：
# 存入：同时用两个模型编码
def store(text):
db_a.upsert(embed_A(text), text, meta={"model":"A"})
db_b.upsert(embed_B(text), text, meta={"model":"B"})
# 检索：两路并行召回，分数归一化后合并
def retrieve(query, k=5):
hits_a = db_a.search(embed_A(query), top_k=k) # 用 A 检 A 的库
hits_b = db_b.search(embed_B(query), top_k=k) # 用 B 检 B 的库
# 各自归一化到 [0,1] 再合并排序
return merge_and_rerank(
normalize(hits_a), normalize(hits_b)
).top_k(k)
注意：不能用 embed_A 生成 query 去检索 db_b，必须同模型配套。
方案 4：文本兜底（BM25 关键词检索）
当 embedding 不可靠或不一致时，退回关键词检索，它不依赖向量模型：
# 向量检索失败/不可用时，用 BM25
bm25_hits = bm25_index.search(query, top_k=10)
# 或混合检索：向量 + BM25 加权融合
final = fuse(vector_hits, bm25_hits, weights=[0.7, 0.3])
决策流程图
是否要切换 embedding 模型？
├── 否 → 沿用原模型，存查一致（方案1）
└── 是 → 能否全量 re-embed？
├── 是 → 执行迁移脚本，切完统一用新模型（方案2）
└── 否 → 过渡期双写双检索（方案3）+ BM25 兜底（方案4）
架构上的防呆设计
生产系统应在 Memory Manager 层强制约束"存查同模型"：
class MemoryManager:
def __init__(self, embed_model_name):
self.embed_model = load_model(embed_model_name)
self.model_name = embed_model_name
def store(self, text):
vec = self.embed_model.encode(text)
self.db.upsert(vec, text, meta={"embed_model": self.model_name})
def retrieve(self, query, k=5):
# 防呆：校验库中记忆的模型一致性
vec = self.embed_model.encode(query)
hits = self.db.search(vec, top_k=k)
for h in hits:
assert h.meta["embed_model"] == self.model_name, \
"Embedding 模型不一致，需先迁移"
return hits
实践建议
- Embedding 模型独立选型：不要绑定对话 LLM，按"中文/英文/多语言 + 维度 + 成本"独立选（如 BGE-M3 中文强、OpenAI 通用性好）。
- 元数据必带
embed_model：为未来迁移留接口。 - 版本化记忆库：换模型 = 新版本库，旧库保留备份，灰度切换。
- 混合检索兜底：向量 + BM25 双路，任何一路失效另一路补位。
7. 实战示例
场景：个人助理 Agent 跨会话记住用户偏好。
- 用户说"我对花生过敏" → 写入长期记忆（结构化：
{fact:花生过敏, type:健康}）。 - 下次用户问"推荐午餐" → 检索记忆 → 过滤含花生菜品 → 推荐。
- 用户改口"我现在可以吃花生了" → 更新记忆（失效旧条目）。
8. 学习要点
- 记忆让 Agent 从"金鱼"变"有经验者"。
- 三层分工：短期对话、工作草稿、长期落库。
- 检索质量与遗忘机制是长期记忆的关键。
9. 参考资料
- "Generative Agents: Interactive Simulacra of Human Behavior"（记忆/反思机制）
- Mem0、LangChain Memory 文档
- "Cognitive Architectures for Language Agents"（CoALA，记忆章节）
- BM25：Best Matching 25，最优匹配算法第 25 版，是现在工业界最主流的关键词相关性打分算法，Lucene、ES、rank_bm25、Whoosh 底层全在用它。
评论
评论加载中…