02 - 写入路径:什么该被记下来
前置:01 篇的三种记忆分型与四个动作。
本篇回答:一段对话结束后,具体是谁、在什么时候、按什么规则决定哪几句话值得留下来。这一步决定了检索的天花板 —— 抽取时丢掉的信息,任何检索策略都找不回来。
本篇会用到的词:
| 词 | 意思 |
|---|---|
| 抽取(extraction) | 用一次模型调用把对话变成若干条独立的事实陈述 |
| 巩固(consolidation) | 新抽出的事实和已有记忆比对,决定是新增、改写、作废还是丢弃 |
| 自包含(self-contained) | 一条记忆脱离原对话也读得懂。「他也喜欢」不自包含,「用户的哥哥也喜欢滑雪」自包含 |
| 观测日期 / 当前日期 | 抽取时用到的两个不同时间锚点:对话实际发生的日期,和跑抽取的日期。混用会把相对时间解析错 |
| 幻觉 ID | 让模型直接操作 UUID 时,它会编出一个格式正确但不存在的 ID。规避手段见 2.2 |
一、两条路线:抽取还是原样存
MemDelta 那个数字值得单独说一句:在受控对照下,让 Agent 自己决定记什么(42%),反而不如把对话原样切片做向量检索(47%)。这不是说抽取式没用,而是说抽取带来的收益不像宣传材料里那么确定,而它的成本是确定的。完整实验设计在 07 篇第四节。
二、mem0 的写入流水线
mem0 是抽取式路线里代码最完整、被读得最多的实现。下面按主分支 mem0/memory/main.py 的 _add_to_vector_store 逐段拆。
2.1 阶段②:检索已有记忆的两个约束
# mem0/memory/main.py,_add_to_vector_store 内
# Phase 1: Existing memory retrieval
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
query_embedding = self.embedding_model.embed(parsed_messages, "search")
existing_results = self.vector_store.search(
query=parsed_messages,
vectors=query_embedding,
top_k=10, # 硬编码:模型最多能看到十条已有记忆
filters=search_filters, # 三个租户维度,缺一个就会跨范围检索
)
两点要看清楚:
filters决定隔离边界。user_id/agent_id/run_id三个维度,调用方漏传就退化成全库检索。多租户系统里这是一个真实的越界风险,07 篇第三节展开- 查询用的是整段对话的嵌入,不是逐条事实的嵌入。对话涉及多个话题时,这个"平均向量"可能哪个话题都不太像,检索出来的十条已有记忆跟真正要写的事实关系不大
2.2 阶段③:UUID 要先映射成整数
这段代码上面挂着一行注释 # Map UUIDs to integers (anti-hallucination):
existing_memories = []
uuid_mapping = {}
for idx, mem in enumerate(existing_results):
uuid_mapping[str(idx)] = mem.id # 记下 "0" → "a3f1c8e2-..." 的对照
existing_memories.append({"id": str(idx), "text": mem.payload.get("data", "")})
# 给模型看到的是 [{"id": "0", "text": "..."}, ...],不是真正的 UUID
为什么必须这么做:直接把 UUID 给模型,让它回答"这条新事实和哪条旧记忆有关",它会返回一个格式完全正确、但库里不存在的 UUID。32 位十六进制串对模型来说是高熵噪声,复制过程中出错的概率远高于复制一个个位数。
映射成 "0"–"9" 之后,模型的输出空间被压到十个值,越界立刻能校验出来。这是一条通用经验:任何需要模型引用现有实体的场合,都不要让它直接搬运长标识符。
2.3 infer=False:绕过抽取的旁路
def _add_to_vector_store(self, messages, metadata, filters, infer, prompt=None):
if not infer:
# 不调模型,每条非 system 消息原样存一条记忆
for message_dict in messages:
if message_dict["role"] == "system":
continue
msg_embeddings = self.embedding_model.embed(message_dict["content"], "add")
mem_id = self._create_memory(message_dict["content"], {...}, per_msg_meta)
return returned_memories
# ...以下才是抽取流水线
这条旁路就是第一节的路线 B。它在两种场景下有用:
- 写入的内容已经是结构化事实(比如从业务事件流灌进来的),再抽一遍没有意义还会引入误差
- 成本敏感的批量灌数:一百万条历史消息全走抽取,是一百万次模型调用
三、从四操作巩固退回纯追加
mem0 论文里的写法是两阶段:抽取出候选事实之后,再让模型对照已有记忆决定做什么。提示词至今还在 mem0/configs/prompts.py 里:
DEFAULT_UPDATE_MEMORY_PROMPT = """You are a smart memory manager which controls the memory of a system.
You can perform four operations: (1) add into the memory, (2) update the memory,
(3) delete from the memory, and (4) no change.
...
- ADD: Add it to the memory as a new element
- UPDATE: Update an existing memory element
- DELETE: Delete an existing memory element
- NONE: Make no change (if the fact is already present or irrelevant)
"""
而主分支的默认路径已经不走它了。同一个文件里新增了:
# V3 Additive Extraction Prompt (ADD-only with memory linking)
ADDITIVE_EXTRACTION_PROMPT = """
# ROLE
You are a Memory Extractor ... Your sole operation is ADD: identify every piece of
memorable information and produce self-contained, contextually rich factual statements.
...
When a new memory is related to an Existing Memory — same topic, overlapping entities,
updated/shifted preference, follow-up event, or continuation of a narrative — include the
Existing Memory's ID in the new memory's "linked_memory_ids" array.
"""
为什么会退回去,从代码本身能读出三条理由:
- UPDATE 和 DELETE 是不可逆的。模型在只看到十条相关记忆的情况下判断"这条该删",判错就没有第二次机会
- 四操作要求模型同时做两件事:识别新事实、并对每条做决策。合并任务会同时拉低两者的准确率
- 纯追加天然幂等。同一段对话重跑一遍,最坏结果是多几条被哈希或语义去重挡掉的重复;四操作重跑一遍可能把上次刚建的条目删掉
代价被明确接受了:库只增不减,冲突并存。这就把压力全部转移到检索侧和作废机制上 —— 也就是 03 篇和 04 篇的主题。
四、写入时机:热路径还是后台
LangMem 的概念文档把这两种时机叫 active(热路径)和 background(后台),并列出了各自的代价:
| 时机 | 延迟影响 | 记忆生效速度 | 处理位置 | 适合 |
|---|---|---|---|---|
| 热路径 | 高 | 立即 | 生成回复的同一轮内 | 关键上下文更新 |
| 后台 | 无 | 延迟 | 两次调用之间或之后 | 模式分析、摘要 |