Mem0源码解析：记忆添加机制深度测评

2026-06-09阅读 0热度 0

机器人

Mem0 源码深度拆解（一）：记忆写入机制全解析

在AI应用落地中，让模型“记住”用户说过的话、偏好某种风格，一直是技术难点。Mem0（读作“mem-zero”）正是为解决此问题而生的开源项目——它为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,):

参数解析

messages：输入对话内容，格式灵活——可以是纯字符串（如 "我喜欢喝咖啡"）、单条消息字典（{"role": "user", "content": "我喜欢喝咖啡"}），或包含多轮角色的消息列表。
user_id/agent_id/run_id：会话标识符，用于隔离不同用户或会话的记忆，避免交叉污染。
infer：是否启用LLM推理，默认 True。启用时系统利用大模型提取事实、智能管理记忆；关闭则直接存入原始消息，不做任何处理。
memory_type：记忆类型标记，例如 "procedural_memory" 用于存储程序性记忆。

核心流程

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}

关键设计：向量存储与图存储的添加是并行执行的。两者相互独立，利用 ThreadPoolExecutor 同步提交，提升效率。

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

这是整个流程的核心，文件位置 mem0/memory/main.py:386。

两种模式

模式一：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（智能推理）

默认模式，流程更复杂：

下面逐步骤拆解。

步骤1：提取事实

第一步让LLM从对话中提取有保留价值的事实。系统根据消息角色是用户还是Agent来选用不同的提取提示词：

# 选择提示词（用户记忆 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"]}
"""

提示词定义了详细的分类指南，LLM据此输出结构化的“事实”。

步骤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", "")})

默认取前5条相似记忆作为后续决策依据。

步骤3：决定操作类型

结合新旧记忆，让LLM决策如何处置新事实：

# 构建决策提示词
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", "event": "UPDATE"},
    {"id": "1", "text": "Name is John", "event": "NONE"},
    {"id": "2", "text": "Dislikes cats", "event": "ADD"}
  ]
}

此例中：第一条原有记忆“喜欢芝士披萨”需更新为“喜欢芝士和鸡肉披萨”；第二条“名字叫John”已存在，无需操作；第三条“讨厌猫”为全新信息，直接新增。

步骤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

记忆创建细节

_create_memory() 方法（位于 mem0/memory/main.py:1075）完成以下工作：生成唯一ID、构建元数据（包括原文、MD5哈希值、创建时间）、将embedding和元数据写入向量数据库，同时将操作历史记录到SQLite。

def _create_memory(self, data, existing_embeddings, metadata=None):
    memory_id = str(uuid.uuid4())
    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])
    self.db.add_history(memory_id, None, data, "ADD", ...)
    return memory_id

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

向量存储负责事实检索，而图存储管理实体与关系，擅长处理复杂知识网络。

入口方法

位于 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

图存储核心逻辑

图存储位于 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}

它先用LLM从数据中抽取实体及其类型，然后建立关系，再对比图数据库已有节点，决定哪些需新增、哪些需删除（如发现矛盾关系）。最后执行实际写入操作。

例如，用户说“张三喜欢喝星巴克的拿铁”，提取结果如下：

实体：{ "张三": "person", "拿铁": "drink", "星巴克": "brand" }
关系：[ {"source": "张三", "relationship": "likes", "destination": "拿铁"}, {"source": "拿铁", "relationship": "brand", "destination": "星巴克"} ]

最终存入Neo4j的三元组示例：

(张三:Person) -[:LIKES]-> (拿铁:Drink) -[:BRAND]-> (星巴克:Brand)

五、关键组件介绍

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),
        ...
    }

类似地，Embedding模型与向量存储也采用相应工厂类，覆盖主流供应商和开源方案：OpenAI、HuggingFace、Qdrant、Chroma、Pinecone、Milvus、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提取两条事实→在向量数据库搜索（新用户，无相似记忆）→LLM决定全部ADD→向量库创建两条记录→图存储提取实体{“张三”: “person”, “拿铁”: “drink”}并建立关系。

七、设计亮点

智能推理 vs 直接存储。两种模式对应不同场景：生产环境推荐 infer=True，实现智能提取、去重、更新；若需保留完整原始对话，infer=False 更直接。

双存储架构。向量存储擅长快速相似性搜索，图存储擅长管理实体间网络关系——两者互补，缺一不可。

并行处理。向量存储与图存储写入互不影响，通过 ThreadPoolExecutor 并行提交，从架构上避免串行等待。

提示词工程。这是Mem0实现“聪明”记忆管理的核心——事实提取提示词定义7种信息类型，记忆更新提示词定义4种操作类型，且支持用户自定义。提示词质量直接影响记忆管理效果。

八、总结

回顾整个添加记忆流程，并非简单的“存进去”，而是一套完整的智能决策链路：

首先让LLM从对话中提取有价值的事实
再到向量数据库中检索已有相似记忆
然后由LLM决定新事实应新增、更新、删除还是不处理
最后同时写入向量存储和图存储

核心设计理念清晰：用LLM管理记忆生命周期，向量存储负责事实级检索，图存储负责实体关系建模，并行处理保证系统效率。

当然，这一切的起点是那些精心编写的提示词。下一篇将专门拆解 Mem0 的提示词工程，分析提示词的设计方式及其对记忆管理智能程度的影响。

第二篇预告：提示词工程深度解析

Mem0源码解析：记忆添加机制深度测评

Mem0 源码深度拆解（一）：记忆写入机制全解析

一、整体架构

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

参数解析

核心流程

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

两种模式

步骤1：提取事实

步骤2：搜索相似记忆

步骤3：决定操作类型

步骤4：执行操作

记忆创建细节

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

入口方法

图存储核心逻辑

五、关键组件介绍

LLM 工厂模式

六、完整流程示例

七、设计亮点

八、总结

相关阅读

最新教程

最新资讯

Mem0 源码深度拆解（一）：记忆写入机制全解析

一、整体架构

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

参数解析

核心流程

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

两种模式

步骤1：提取事实

步骤2：搜索相似记忆

步骤3：决定操作类型

步骤4：执行操作

记忆创建细节

四、图存储添加：_add_to_graph()

入口方法

图存储核心逻辑

五、关键组件介绍

LLM 工厂模式

六、完整流程示例

七、设计亮点

八、总结

相关阅读

最新教程

最新资讯

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

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

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