01-Mem0 的 ADD-Only 记忆提取流水线
Mem0 的 ADD-Only 记忆提取流水线
学习目标
本章聚焦 Mem0 如何将”用户说了一句话”转化为”持久化记忆”的完整过程。你将了解：
- Mem0 v3 的 8 阶段提取流水线如何工作
- 为什么 Mem0 从 v2 的 ADD/UPDATE/DELETE 三操作切换到 ADD-Only
- Hash 去重、批量 Embedding、实体链接的具体实现
- 流水线中每个阶段的失败处理与回退策略
前置知识
本章涉及 ADD-Only 记忆提取模式的通用原理，建议先阅读：
下文假设你已理解 ADD-Only 的基本概念，直接聚焦 Mem0 的具体实现。
项目实践
入口：Memory.add() 方法
Mem0 的 add() 方法是整个提取流水线的入口：
# 伪代码memory = Memory() # 初始化，加载配置并创建 Provider 实例result = memory.add( messages="我喜欢 Python 和机器学习", # 单条消息或多条消息 user_id="alice", # 作用域标识符 infer=True, # 是否使用 LLM 提取)当 infer=True（默认）时，add() 调用 _add_to_vector_store() 进入 8 阶段流水线。当 infer=False 时，消息直接嵌入并存储，跳过 LLM 提取。
8 阶段流水线详解
Phase 0: 上下文收集
session_scope = _build_session_scope(filters) # "user_id=alice"last_messages = self.db.get_last_messages(session_scope, limit=10)parsed_messages = parse_messages(messages)从 SQLite 获取最近 10 条对话消息作为上下文，帮助 LLM 理解指代消解和时间引用。
Phase 1: 现有记忆检索
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)先检索 top-10 相关记忆，目的是：
- 为 LLM 提供上下文（避免提取已存在的记忆）
- 为 Hash 去重准备已有记忆的 Hash 集合
- 为
linked_memory_ids准备现有记忆 UUID 映射
Phase 2: LLM 单次提取
这是整个流水线的核心。Mem0 使用 ADDITIVE_EXTRACTION_PROMPT（900+ 行提示词）引导 LLM 一次性提取所有新记忆：
response = self.llm.generate_response( messages=[ {"role": "system", "content": ADDITIVE_EXTRACTION_PROMPT}, {"role": "user", "content": generate_additive_extraction_prompt( existing_memories=existing_memories, new_messages=parsed_messages, last_k_messages=last_messages, )}, ], response_format={"type": "json_object"},)关键设计：
- 单次 LLM 调用，不做多轮对话
response_format={"type": "json_object"}确保输出可解析- Agent 场景下追加
AGENT_CONTEXT_SUFFIX，指示 LLM 从 Agent 视角提取
回退机制：如果 LLM 响应解析失败，尝试从代码块中提取 JSON；如果仍失败，返回空列表。
Phase 3: 批量 Embedding
mem_texts = [m.get("text", "") for m in extracted_memories]mem_embeddings_list = self.embedding_model.embed_batch(mem_texts, "add")embed_map = dict(zip(mem_texts, mem_embeddings_list))对所有提取的记忆文本批量 Embedding，建立文本→向量的映射。如果批量 Embedding 失败，回退到逐条 Embedding。
Phase 4: Hash 去重
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)对每条记忆计算 MD5 Hash，与现有记忆 Hash 集合和当前批次内 Hash 集合比对，跳过重复项。
同时进行的处理：
- 词形还原（
lemmatize_for_bm25）：为后续 BM25 搜索准备 - 元数据填充：
data、hash、created_at、updated_at、attributed_to
Phase 5-6: 批量持久化
# 批量插入向量存储self.vector_store.insert( vectors=all_vectors, ids=all_ids, payloads=all_payloads)
# 批量记录历史日志self.db.batch_add_history(history_records)批量插入所有新记忆到向量存储，并在 SQLite history 表中记录所有 ADD 操作。如果批量插入失败，回退到逐条插入。
Phase 7: 批量实体链接
这是 Mem0 最有特色的阶段：
# 全局去重：收集所有记忆中的唯一实体global_entities = {} # normalized_key -> (entity_type, entity_text, set of memory_ids)
# 批量 Embedding 所有唯一实体entity_embeddings = self.embedding_model.embed_batch(entity_texts, "add")
# 批量搜索现有实体existing_matches = self.entity_store.search_batch(...)
# 分类：更新现有实体 vs 插入新实体# 批量插入所有新实体self.entity_store.insert(vectors=to_insert_vectors, ids=to_insert_ids, payloads=to_insert_payloads)优化亮点：
- 跨所有记忆全局去重实体，避免重复 Embedding
- 批量搜索 + 批量插入，减少 Entity Store 交互次数
- 单条失败不影响整体（逐条回退 + 异常捕获）
Phase 8: 保存消息
将原始对话消息保存到 SQLite messages 表，为下一次提取提供上下文。消息表自动裁剪，只保留最近 10 条。
问题与规避
| 问题 | 对策 |
|---|---|
| LLM 响应为空或解析失败 | 尝试从代码块提取 JSON；仍失败则返回空列表，流水线不中断 |
| 批量 Embedding 失败 | 回退到逐条 Embedding，逐条捕获异常 |
| 向量存储批量插入失败 | 回退到逐条插入，每条失败单独记录日志 |
| 实体链接全部失败 | 异常被 logger.warning 捕获，不影响主流程的记忆存储 |
| Hash 去重无法捕获语义重复 | 语义重复不在去重阶段处理——LLM 提示词中已要求跳过已存在的语义内容 |
设计取舍
为什么选 8 阶段流水线而非简单循环？
优势：
- 批量操作（Embedding、插入、实体链接）大幅减少存储交互次数
- 每个阶段有独立的回退策略，流水线整体鲁棒性强
- 阶段间数据传递清晰（如
embed_map连接 Phase 3 和 Phase 4）
代价：
- 流水线长，单次
add()调用涉及多次存储和 LLM 交互 - 实体链接的批量优化增加了代码复杂度
为什么 LLM 提示词 900+ 行？
ADDITIVE_EXTRACTION_PROMPT 包含 12 个 Few-Shot 示例、详细的提取规则、完整性检查清单。这看似冗，但 Mem0 的基准测试提升（LoCoMo +20 分）证明：精心设计的提示词对提取质量的提升远超额外 Token 的成本。
参考来源
- Mem0 论文：arXiv:2504.19413
- Mem0 OSS v2→v3 迁移指南：https://docs.mem0.ai/migration/oss-v2-to-v3