Mem0 源码解析系列（一）：记忆是如何被添加的-平芜编程栈

一、引言

Mem0（"mem-zero"）是一个为 AI 应用提供长期记忆层的开源项目。它能让 AI 助手记住用户偏好、适应个性化需求，并持续学习——非常适合客户支持聊天机器人、AI 助手和自主系统等场景。

在阅读源码之前，我一直好奇：

Mem0 是如何从对话中提取有价值的信息？
它如何决定是添加新记忆、更新旧记忆，还是删除过期记忆？
向量存储和图存储分别扮演什么角色？

带着这些问题，让我们开始探索。

二、整体架构

Mem0 的记忆添加流程可以概括为以下架构：

核心文件位于mem0/memory/main.py，主要涉及：

Memory.add()：入口方法
_add_to_vector_store()：向量存储逻辑
_add_to_graph()：图存储逻辑

三、入口方法：`Memory.add()`

让我们从入口方法开始（位于mem0/memory/main.py:281）：

def add( self, messages, *, user_id: Optional[str] = None, agent_id: Optional[str] = None, run_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, infer: bool = True, memory_type: Optional[str] = None, prompt: Optional[str] = None, ):

3.1 参数解析

messages：输入内容，可以是字符串、字典或消息列表
- 字符串："我喜欢喝咖啡"
- 单条消息：{"role": "user", "content": "我喜欢喝咖啡"}
- 消息列表：[{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
user_id/agent_id/run_id：会话标识符，用于隔离不同用户/会话的记忆
infer：是否使用 LLM 推理（默认 True）
- True：提取事实并智能管理记忆
- False：直接存储原始消息
memory_type：记忆类型（如 "procedural_memory" 用于程序性记忆）

3.2 核心流程

add()方法的核心逻辑（简化版）：

# 1. 构建元数据和过滤条件 processed_metadata, effective_filters = _build_filters_and_metadata( user_id=user_id, agent_id=agent_id, run_id=run_id, input_metadata=metadata ) # 2. 处理特殊记忆类型（如程序性记忆） if agent_id is not None and memory_type == MemoryType.PROCEDURAL.value: return self._create_procedural_memory(messages, metadata=processed_metadata) # 3. 并行执行向量存储和图存储添加 with concurrent.futures.ThreadPoolExecutor() as executor: future1 = executor.submit(self._add_to_vector_store, messages, processed_metadata, effective_filters, infer) future2 = executor.submit(self._add_to_graph, messages, effective_filters) concurrent.futures.wait([future1, future2]) vector_store_result = future1.result() graph_result = future2.result() # 4. 返回结果 return {"results": vector_store_result, "relations": graph_result}

这里有个关键设计：并行执行。向量存储和图存储的添加是独立进行的，提高了效率。

四、向量存储添加：`_add_to_vector_store()`

这是最核心的部分，位于mem0/memory/main.py:386。

4.1 两种模式

模式一：`infer=False`（直接存储）

当infer=False时，不经过 LLM 处理，直接存储原始消息：

if not infer: for message_dict in messages: # 跳过系统消息 if message_dict["role"] == "system": continue # 生成 embedding msg_embeddings = self.embedding_model.embed(msg_content, "add") # 直接创建记忆 mem_id = self._create_memory(msg_content, msg_embeddings, per_msg_meta) returned_memories.append({ "id": mem_id, "memory": msg_content, "event": "ADD" }) return returned_memories

模式二：`infer=True`（智能推理）

这是默认模式，流程更复杂：

4.2 步骤详解

步骤1：提取事实

使用 LLM 从对话中提取有价值的"事实"：

# 选择提示词（用户记忆 vs Agent记忆） is_agent_memory = self._should_use_agent_memory_extraction(messages, metadata) system_prompt, user_prompt = get_fact_retrieval_messages(parsed_messages, is_agent_memory) # 调用 LLM response = self.llm.generate_response( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], response_format={"type": "json_object"} ) # 解析 JSON 结果 new_retrieved_facts = json.loads(response)["facts"]

提示词示例（位于mem0/configs/prompts.py:14）：

FACT_RETRIEVAL_PROMPT = """You are a Personal Information Organizer... Types of Information to Remember: 1. Store Personal Preferences 2. Maintain Important Personal Details 3. Track Plans and Intentions ... Input: Hi, my name is John. I am a software engineer. Output: {"facts": ["Name is John", "Is a Software engineer"]} """

步骤2：搜索相似记忆

对每个新提取的事实，在向量数据库中搜索相似的记忆：

for new_mem in new_retrieved_facts: # 生成 embedding messages_embeddings = self.embedding_model.embed(new_mem, "add") # 搜索相似记忆（向量相似度） existing_memories = self.vector_store.search( query=new_mem, vectors=messages_embeddings, limit=5, filters=search_filters ) for mem in existing_memories: retrieved_old_memory.append({"id": mem.id, "text": mem.payload.get("data", "")})

步骤3：决定操作类型

使用 LLM 分析新旧记忆，决定操作类型：

决策提示词（位于mem0/configs/prompts.py:175）定义了四种操作：

# 构建决策提示词 function_calling_prompt = get_update_memory_messages( retrieved_old_memory, new_retrieved_facts ) # LLM 返回操作决策 response = self.llm.generate_response( messages=[{"role": "user", "content": function_calling_prompt}], response_format={"type": "json_object"} ) new_memories_with_actions = json.loads(response)

决策提示词（位于mem0/configs/prompts.py:175）定义了四种操作：

DEFAULT_UPDATE_MEMORY_PROMPT = """You are a smart memory manager... Compare newly retrieved facts with the existing memory. For each new fact, decide whether to: - 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 """

示例决策过程：

{ "memory": [ { "id": "0", "text": "Likes cheese and chicken pizza", // UPDATE: 更新口味偏好 "event": "UPDATE", "old_memory": "Likes cheese pizza" }, { "id": "1", "text": "Name is John", // NONE: 已存在，无需操作 "event": "NONE" }, { "id": "2", "text": "Dislikes cats", // ADD: 新增记忆 "event": "ADD" } ] }

步骤4：执行操作

根据决策结果执行相应的 CRUD 操作：

for resp in new_memories_with_actions.get("memory", []): action_text = resp.get("text") event_type = resp.get("event") if event_type == "ADD": memory_id = self._create_memory(action_text, existing_embeddings, metadata) elif event_type == "UPDATE": self._update_memory( memory_id=temp_uuid_mapping[resp.get("id")], data=action_text, existing_embeddings=existing_embeddings, metadata=metadata ) elif event_type == "DELETE": self._delete_memory(memory_id=temp_uuid_mapping[resp.get("id")]) elif event_type == "NONE": # 无需操作 pass

4.3 记忆创建细节

_create_memory()方法（mem0/memory/main.py:1075）：

def _create_memory(self, data, existing_embeddings, metadata=None): # 生成唯一 ID memory_id = str(uuid.uuid4()) # 构建 metadata metadata["data"] = data metadata["hash"] = hashlib.md5(data.encode()).hexdigest() metadata["created_at"] = datetime.now(pytz.timezone("US/Pacific")).isoformat() # 存入向量数据库 self.vector_store.insert( vectors=[embeddings], ids=[memory_id], payloads=[metadata] ) # 记录历史（SQLite） self.db.add_history(memory_id, None, data, "ADD", ...) return memory_id

五、图存储添加：`_add_to_graph()`

图存储用于存储实体和关系，适合处理复杂的知识网络。

5.1 入口方法

位于mem0/memory/main.py:599：

def _add_to_graph(self, messages, filters): if self.enable_graph: # 合并消息内容 data = "\n".join([msg["content"] for msg in messages if msg["role"] != "system"]) # 调用图存储添加 added_entities = self.graph.add(data, filters) return added_entities

5.2 图存储核心逻辑

位于mem0/memory/graph_memory.py:76，主要步骤：

def add(self, data, filters): # 1. 提取实体 entity_type_map = self._retrieve_nodes_from_data(data, filters) # 2. 建立实体关系 to_be_added = self._establish_nodes_relations_from_data(data, filters, entity_type_map) # 3. 搜索图数据库中的相似节点 search_output = self._search_graph_db(node_list=list(entity_type_map.keys()), filters=filters) # 4. 决定需要删除的实体（矛盾关系） to_be_deleted = self._get_delete_entities_from_search_output(search_output, data, filters) # 5. 执行删除和添加 deleted_entities = self._delete_entities(to_be_deleted, filters) added_entities = self._add_entities(to_be_added, filters, entity_type_map) return {"deleted_entities": deleted_entities, "added_entities": added_entities}

5.3 实体提取示例

使用 LLM 提取实体和类型：

entity_type_map = self._retrieve_nodes_from_data(data, filters) # 结果示例： # { # "john": "person", # "coffee": "drink", # "starbucks": "brand" # }

5.4 关系建立示例

使用 LLM 建立实体间的关系：

entities = self._establish_nodes_relations_from_data(data, filters, entity_type_map) # 结果示例： # [ # {"source": "john", "relationship": "likes", "destination": "coffee"}, # {"source": "coffee", "relationship": "brand", "destination": "starbucks"} # ]

最终存入 Neo4j 图数据库：

(john:Person) -[:LIKES]-> (coffee:Drink) -[:BRAND]-> (starbucks:Brand)

六、关键组件介绍

6.1 LLM 工厂模式

Mem0 使用工厂模式支持多种 LLM：

# mem0/utils/factory.py class LlmFactory: provider_to_class = { "openai": ("mem0.llms.openai.OpenAILLM", OpenAIConfig), "anthropic": ("mem0.llms.anthropic.AnthropicLLM", AnthropicConfig), "azure_openai": ("mem0.llms.azure_openai.AzureOpenAILLM", AzureOpenAIConfig), "gemini": ("mem0.llms.gemini.GeminiLLM", BaseLlmConfig), ... }

6.2 Embedding 生成

class EmbedderFactory: provider_to_class = { "openai": "mem0.embeddings.openai.OpenAIEmbedding", "huggingface": "mem0.embeddings.huggingface.HuggingFaceEmbedding", ... }

6.3 向量存储支持

class VectorStoreFactory: provider_to_class = { "qdrant": "mem0.vector_stores.qdrant.Qdrant", "chroma": "mem0.vector_stores.chroma.ChromaDB", "pinecone": "mem0.vector_stores.pinecone.PineconeDB", "milvus": "mem0.vector_stores.milvus.MilvusDB", "weaviate": "mem0.vector_stores.weaviate.Weaviate", ... }

七、完整流程示例

让我们用一个完整示例串联整个过程：

from mem0 import Memory memory = Memory() # 用户对话 messages = [ {"role": "user", "content": "我叫张三，我喜欢喝拿铁咖啡"}, {"role": "assistant", "content": "你好张三！拿铁是很受欢迎的咖啡"} ] # 添加记忆 result = memory.add(messages, user_id="zhangsan") # 结果示例 { "results": [ {"id": "abc123", "memory": "Name is 张三", "event": "ADD"}, {"id": "def456", "memory": "Likes 拿铁 coffee", "event": "ADD"} ], "relations": { "deleted_entities": [], "added_entities": [ {"source": "张三", "relationship": "likes", "target": "拿铁"} ] } }

流程拆解：

提取事实：
- LLM 从对话中提取：["Name is 张三", "Likes 拿铁 coffee"]
搜索相似记忆：
- 在向量数据库中搜索相似的记忆
- 结果：未找到相似记忆（假设是新用户）
决策操作：
- LLM 决定：两条事实都需要 ADD
执行操作：
- 创建两条记忆记录
- 生成 embedding 并存入向量数据库
图存储：
- 提取实体：{"张三": "person", "拿铁": "drink"}
- 建立关系：张三 -[:likes]-> 拿铁
- 存入 Neo4j

八、设计亮点

8.1 智能推理 vs 直接存储

Mem0 提供两种模式：

infer=True：智能提取、去重、更新，适合生产环境
infer=False：直接存储，适合需要完整保留原始对话的场景

8.2 双存储架构

向量存储：快速相似性搜索，适合检索场景
图存储：实体关系管理，适合复杂知识网络

两者互补，提供更全面的记忆能力。

8.3 并行处理

向量存储和图存储并行执行，提高效率：

with concurrent.futures.ThreadPoolExecutor() as executor: future1 = executor.submit(self._add_to_vector_store, ...) future2 = executor.submit(self._add_to_graph, ...)

8.4 提示词工程

Mem0 的提示词设计非常精细：

事实提取提示词包含详细的分类指南和示例
记忆更新提示词定义了清晰的 ADD/UPDATE/DELETE/NONE 规则
支持自定义提示词（custom_fact_extraction_prompt）

九、总结

通过这次源码解析，我们了解到：

Mem0 的记忆添加不只是简单的存储，而是包含了：
- 智能事实提取
- 相似性检索
- 增删改决策
- 双存储架构
核心设计理念：
- 用 LLM 智能管理记忆生命周期
- 向量存储处理事实检索
- 图存储处理实体关系
- 并行处理提升效率
提示词工程是 Mem0 的核心：
- 事实提取提示词定义了 7 种信息类型
- 记忆更新提示词定义了 4 种操作类型
- 这些精心设计的提示词是 Mem0 智能管理的关键

Mem0 源码解析系列（一）：记忆是如何被添加的

一、引言

二、整体架构

三、入口方法：`Memory.add()`

3.1 参数解析

3.2 核心流程

四、向量存储添加：`_add_to_vector_store()`

4.1 两种模式

模式一：`infer=False`（直接存储）

模式二：`infer=True`（智能推理）

4.2 步骤详解

步骤1：提取事实

步骤2：搜索相似记忆

步骤3：决定操作类型

步骤4：执行操作

4.3 记忆创建细节

五、图存储添加：`_add_to_graph()`

5.1 入口方法

5.2 图存储核心逻辑

5.3 实体提取示例

5.4 关系建立示例

六、关键组件介绍

6.1 LLM 工厂模式

6.2 Embedding 生成

6.3 向量存储支持

七、完整流程示例

流程拆解：

八、设计亮点

8.1 智能推理 vs 直接存储

8.2 双存储架构

8.3 并行处理

8.4 提示词工程

九、总结

基于大语言模型与视觉认知的UI自动化测试实践：OpenClaw与Qwen3-32B集成指南

逻辑回归前必须做的4步可视化诊断

hello_world

Linux的内存分页管理

从“猜词”到“理解世界”：2026年人工智能前沿进展速览

orcale的锁模式

一、引言

二、整体架构

三、入口方法：Memory.add()

3.1 参数解析

3.2 核心流程

四、向量存储添加：_add_to_vector_store()

4.1 两种模式

模式一：infer=False（直接存储）

模式二：infer=True（智能推理）

4.2 步骤详解

步骤1：提取事实

步骤2：搜索相似记忆

步骤3：决定操作类型

步骤4：执行操作

4.3 记忆创建细节

五、图存储添加：_add_to_graph()

5.1 入口方法

5.2 图存储核心逻辑

5.3 实体提取示例

5.4 关系建立示例

六、关键组件介绍

6.1 LLM 工厂模式

6.2 Embedding 生成

6.3 向量存储支持

七、完整流程示例

流程拆解：

八、设计亮点

8.1 智能推理 vs 直接存储

8.2 双存储架构

8.3 并行处理

8.4 提示词工程

九、总结

基于大语言模型与视觉认知的UI自动化测试实践：OpenClaw与Qwen3-32B集成指南

逻辑回归前必须做的4步可视化诊断

hello_world

Linux的内存分页管理

从“猜词”到“理解世界”：2026年人工智能前沿进展速览

orcale的锁模式

三、入口方法：`Memory.add()`

四、向量存储添加：`_add_to_vector_store()`

模式一：`infer=False`（直接存储）

模式二：`infer=True`（智能推理）

五、图存储添加：`_add_to_graph()`