Mem0 写入管线 — `add()` 的 8 阶段

这章讲 Mem0 最核心的一条路径:你给 add() 一段对话,它怎么变成向量库里几条独立的记忆。读完你能讲清楚「单次抽取」「哈希去重」「批量落库」是怎么回事。

1. 它要解决的小问题

你不能把整段对话原样存下来当记忆——那既冗余又难检索。你要的是把对话蒸馏成一句句自包含的事实:「User 住在西雅图」「User 对花生过敏」。难点有三:

抽什么:哪些信息值得记?(口味、计划、人际、agent 给的建议……)
不重复:同一件事说了两遍,不能存两条。
够快:一轮对话别打十几次 LLM/embed。

Mem0 这版的答案是:一次 LLM 调用抽全部,然后用哈希和批处理把后续做廉价。

2. 思路/直觉:ADD-only + 批处理

关键设计决策是把「判断要不要改旧记忆」这件难事从写入端拿掉(见 index §3)。写入只做一件事:抽出新事实并存下来。冲突解决留给检索端的排序去自然处理。

这样写入就退化成一条纯管线:抽 → 去重 → embed → 存。管线里每一步都尽量批量做(一次 embed 一批、一次 insert 一批),把延迟压下去。

3. 全景:8 个阶段

_add_to_vector_store 注释自己把流程标成了「V3 PHASED BATCH PIPELINE」(mem0/memory/main.py:867)。从左到右是数据流向:

  对话 messages
     │
     ▼
 [0] 取会话上下文 ──── SQLite get_last_messages(session_scope, 10)
     │
     ▼
 [1] 捞已有相关记忆 ── 向量库 search(top_k=10) → 给 LLM 当参照 + 去重底
     │                  (UUID → "0"/"1"/... 整数映射,防 LLM 编 id)
     ▼
 [2] 单次 LLM 抽取 ─── ADDITIVE_EXTRACTION_PROMPT → JSON {"memory":[...]}
     │
     ▼
 [3] 批量 embed ────── embed_batch(所有抽出的 text)
     │
     ▼
 [4/5] 哈希去重 ────── md5(text):撞已有哈希 or 批内重复 → 丢
     │
     ▼
 [6] 批量持久化 ────── 向量库 insert(批) + SQLite batch_add_history(批)
     │
     ▼
 [7] 批量实体链接 ──── 见 03-entity-linking.md
     │
     ▼
 [8] 存回消息 ──────── SQLite save_messages(下一轮的上下文)

4. 逐阶段走读

4.0–4.1 上下文 + 捞已有记忆

session_scope 是从 scope id 拼出的确定性字符串,把记忆/消息按「谁的会话」隔开:

# mem0/memory/main.py:366 _build_session_scope —— 排序后拼成 user_id=alice&run_id=r1
for key in sorted(["user_id", "agent_id", "run_id"]):
    val = filters.get(key)
    if val:
        parts.append(f"{key}={val}")
return "&".join(parts)

然后 embed 新消息、向量库 search(top_k=10) 捞已有相关记忆(main.py:876-882)。这步有个反幻觉小技巧:把捞回来的记忆 UUID 重新编号成 "0"/"1"/... 给 LLM(main.py:885-889),让 LLM 在 linked_memory_ids 里引用短整数而不是去背长 UUID,事后再 uuid_mapping 映射回去。

4.2 单次 LLM 抽取(核心)

system prompt 是 ADDITIVE_EXTRACTION_PROMPT;如果是「纯 agent 记忆」(有 agent_id 但没 user_id)再拼上 AGENT_CONTEXT_SUFFIX 换个视角(main.py:892-895)。user prompt 由 generate_additive_extraction_prompt 拼装,把「已有记忆 / 最近消息 / 新消息 / 观测日期」分节塞进去(prompts.py:1016)。

# mem0/memory/main.py:907 —— 注意 response_format 强制 JSON
response = self.llm.generate_response(
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ],
    response_format={"type": "json_object"},
)

解析做了三层兜底(main.py:919-931):先 remove_code_blocks 去掉 ```json 围栏,直接 json.loads;失败再 extract_json 抠出 JSON 子串重试;再失败就当空抽取。LLM 调用本身失败时直接返回 [](main.py:914-916)——不抛错,这轮就当没记到东西。

这个 prompt 本身是工程精华:它明确要求「疑则记(When in doubt, extract)」、casual 闲聊也算可记、多说话人各自归属、相对时间要锚到 Observation Date 变成绝对日期(prompts.py:535「'last week' 6 个月后就没用了」)。输出 schema 是 {id, text, attributed_to, linked_memory_ids}(prompts.py:931-936)。

4.3–4.5 批量 embed + 哈希去重

抽出的 text 一次性 embed_batch,失败才退化成逐条 embed(main.py:939-950)——这是「批量优先、单条兜底」的反复出现的模式。

去重用 md5 文本哈希,分两道:撞「已有记忆的哈希」丢、撞「本批已见哈希」丢:

# mem0/memory/main.py:967 —— 完全相同文本 = 同哈希 = 重复
mem_hash = hashlib.md5(text.encode()).hexdigest()
if mem_hash in existing_hashes or mem_hash in seen_hashes:
    continue  # 跳过重复
seen_hashes.add(mem_hash)

注意这是精确字符串去重,不是语义去重——「我住西雅图」和「我家在西雅图」哈希不同,都会存。语义层面的「别重复记」靠 prompt 让 LLM 自己跳过(prompts.py:511)。每条记忆还顺手算了 text_lemmatized(为 BM25 预处理,见 02 章)和时间戳,塞进 payload(main.py:973-982)。

4.6 批量持久化

向量、id、payload 三个并行列表,一把 insert;失败退化成逐条(main.py:997-1009)。历史同理走 batch_add_history(main.py:1012-1031)。历史库表结构见 storage.py:108,每条记忆变更记一行 event=ADD。

4.8 存回消息

最后 save_messages 把这轮原始消息写进 SQLite 的 messages 表,供下一轮 add 当「Last k Messages」上下文用(main.py:1140)。这是个滚动窗口缓冲。

5. 一条岔路:infer=False

如果调用方传 infer=False,整条 LLM 管线被跳过——每条消息原样存成一条记忆,不抽取、不去重(main.py:831-865)。适合你已经有结构化事实、不想让 LLM 再加工的场景。

6. 巧妙之处

UUID→整数映射防幻觉(main.py:885):让 LLM 引用 "0" 而非 32 位 UUID,显著降低 id 编造。
处处「批量优先、单条兜底」:embed/insert/history 三处都是 try 批量 except 逐条,既快又稳。
抽取失败也存消息(main.py:933-936):即便 LLM 没抽出东西,原始消息照样进缓冲,不丢上下文。
强制 JSON + 三层解析兜底:response_format 先压一道,解析再兜两道,对抗 LLM 输出格式抖动。

7. 边界与局限

只增不改:同一事实换种说法会产生冗余记忆;md5 去重只挡逐字重复。
写入要一次 LLM + 至少两次 embed(查询 embed + 批量 embed),延迟下限受 LLM 拖累;infer=False 可绕开。
memory_type 仅支持 procedural:其它值直接报 VALIDATION_002(main.py:782)。procedural 走单独的 _create_procedural_memory(main.py:1913)。

8. 代码地图

主题	文件	符号
写入管线	`mem0/memory/main.py`	`_add_to_vector_store` (:830)
公开入口	`mem0/memory/main.py`	`Memory.add` (:716)
会话 scope	`mem0/memory/main.py`	`_build_session_scope` (:366)
抽取提示词	`mem0/configs/prompts.py`	`ADDITIVE_EXTRACTION_PROMPT` (:468)
prompt 构造	`mem0/configs/prompts.py`	`generate_additive_extraction_prompt` (:1016)
历史/消息缓冲	`mem0/memory/storage.py`	`save_messages` (:257), `get_last_messages` (:298)

1. 它要解决的小问题​

2. 思路/直觉:ADD-only + 批处理​

3. 全景:8 个阶段​

4. 逐阶段走读​

4.0–4.1 上下文 + 捞已有记忆​

4.2 单次 LLM 抽取(核心)​

4.3–4.5 批量 embed + 哈希去重​

4.6 批量持久化​

4.8 存回消息​

5. 一条岔路:infer=False​

6. 巧妙之处​

7. 边界与局限​

8. 代码地图​