一、引言Mem0mem-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:281def 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 推理默认 TrueTrue提取事实并智能管理记忆False直接存储原始消息memory_type记忆类型如 procedural_memory 用于程序性记忆3.2 核心流程add()方法的核心逻辑简化版# 1. 构建元数据和过滤条件 processed_metadata, effective_filters _build_filters_and_metadata( user_iduser_id, agent_idagent_id, run_idrun_id, input_metadatametadata ) # 2. 处理特殊记忆类型如程序性记忆 if agent_id is not None and memory_type MemoryType.PROCEDURAL.value: return self._create_procedural_memory(messages, metadataprocessed_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 两种模式模式一inferFalse直接存储当inferFalse时不经过 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模式二inferTrue智能推理这是默认模式流程更复杂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:14FACT_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( querynew_mem, vectorsmessages_embeddings, limit5, filterssearch_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_idtemp_uuid_mapping[resp.get(id)], dataaction_text, existing_embeddingsexisting_embeddings, metadatametadata ) elif event_type DELETE: self._delete_memory(memory_idtemp_uuid_mapping[resp.get(id)]) elif event_type NONE: # 无需操作 pass4.3 记忆创建细节_create_memory()方法mem0/memory/main.py:1075def _create_memory(self, data, existing_embeddings, metadataNone): # 生成唯一 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:599def _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_entities5.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_listlist(entity_type_map.keys()), filtersfilters) # 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_idzhangsan) # 结果示例 { 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 提供两种模式inferTrue智能提取、去重、更新适合生产环境inferFalse直接存储适合需要完整保留原始对话的场景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 智能管理的关键