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

极客资料网推荐Mem0 源码解析系列（一）：记忆是如何被添加的这篇文章给大家，欢迎收藏极客资料网享受知识的乐趣

这是 Mem0 源码解析系列的第一篇文章。我们将深入探讨 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": "拿铁"}
        ]
    }
}

流程拆解：

1. 提取事实：

   • LLM 从对话中提取：["Name is 张三", "Likes 拿铁 coffee"]

2. 搜索相似记忆：

   • 在向量数据库中搜索相似的记忆
   • 结果：未找到相似记忆（假设是新用户）

3. 决策操作：

   • LLM 决定：两条事实都需要 ADD

4. 执行操作：

   • 创建两条记忆记录
   • 生成 embedding 并存入向量数据库

5. 图存储：

   • 提取实体：{"张三": "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）

九、总结

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

1. Mem0 的记忆添加不只是简单的存储，而是包含了：

   • 智能事实提取
   • 相似性检索
   • 增删改决策
   • 双存储架构

2. 核心设计理念：

   • 用 LLM 智能管理记忆生命周期
   • 向量存储处理事实检索
   • 图存储处理实体关系
   • 并行处理提升效率

3. 提示词工程是 Mem0 的核心：

   • 事实提取提示词定义了 7 种信息类型
   • 记忆更新提示词定义了 4 种操作类型
   • 这些精心设计的提示词是 Mem0 智能管理的关键

十、系列预告

通过这篇文章，了解了 Mem0 的记忆添加流程，包括智能事实提取、相似性检索、增删改决策以及双存储架构。接下来，我们将深入探讨 Mem0 的提示词工程，了解如何通过精心设计的提示词来实现智能管理。

• 第二篇：提示词工程的深度解析

敬请期待！

---

相关代码文件：

• mem0/memory/main.py：核心 Memory 类
• mem0/memory/graph_memory.py：图存储实现
• mem0/configs/prompts.py：提示词定义
• mem0/utils/factory.py：各种工厂类

文章来源:https://www.cnblogs.com/wylrcx/p/19842964
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：jacktools123@163.com进行投诉反馈，一经查实，立即删除！