如何实现长期记忆的去重和合并？

Mem0 是一个开源的 AI 记忆管理层，专为大模型应用设计，支持向量存储、图存储和历史追踪。它能够帮助 AI Agent 记住用户的偏好、历史对话和关键事实。

Mem0 长期记忆实现教程

概述

Mem0 是一个开源的 AI 记忆管理层，专为大模型应用设计，支持向量存储、图存储和历史追踪。它能够帮助 AI Agent 记住用户的偏好、历史对话和关键事实。

1. 核心架构

1.1 存储层次

存储类型 用途 技术栈
向量存储 语义搜索、相似度匹配 Chroma, Faiss, Pinecone, Qdrant, Weaviate, Redis, Milvus, PGVector 等
图存储 关系知识图谱 Neo4j, AWS Neptune, Memgraph, Apache AGE
历史存储 变更历史追踪 SQLite

1.2 核心类

类 位置 说明
Memory mem0/memory/main.py:244 同步记忆实现
AsyncMemory mem0/memory/main.py:1348 异步记忆实现
MemoryBase mem0/memory/base.py 抽象基类
MemoryClient mem0/client/main.py API 客户端

2. 记忆写入流程

2.1 写入数据结构的定义

# 存入向量存储的 metadata
{
    "data": "用户喜欢喝咖啡",           # 记忆内容文本
    "hash": "a1b2c3...",                # 内容哈希（去重用）
    "created_at": "2026-03-29T...",    # 创建时间
    "user_id": "user_123",             # 用户标识
    "agent_id": "agent_456",           # Agent 标识
    "run_id": "run_789",               # 运行标识
    "role": "user",                    # 消息角色
    "metadata": {...}                   # 自定义元数据
}

2.2 SQLite 历史表结构

CREATE TABLE history (
    id           TEXT PRIMARY KEY,
    memory_id    TEXT,              -- 关联的记忆ID
    old_memory   TEXT,              -- 更新前的值
    new_memory   TEXT,              -- 更新后的值
    event        TEXT,              -- ADD|UPDATE|DELETE
    created_at   DATETIME,
    updated_at   DATETIME,
    is_deleted   INTEGER,           -- 是否已删除
    actor_id     TEXT,
    role         TEXT
)

2.3 写入流程详解

入口: Memory.add() 方法 (mem0/memory/main.py:353-456)

用户消息
    ↓
parse_messages() 格式化消息
    ↓
LLM 提取事实 (get_fact_retrieval_messages)
    ↓
向量搜索相似记忆 (vector_store.search)
    ↓
LLM 决定操作 (get_update_memory_messages)
    ↓
执行 ADD/UPDATE/DELETE
    ↓
写入向量存储 + SQLite 历史

3. 去重和更新机制

Mem0 通过三层机制实现智能去重和更新。

3.1 第一层：向量相似度搜索

代码位置: mem0/memory/main.py:548-558

for new_mem in new_retrieved_facts:
    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,
    )

作用: 通过向量相似度找到语义上相关的已有记忆，为 LLM 判断提供上下文。

3.2 第二层：LLM 智能决策

代码位置: mem0/configs/prompts.py:175-459

function_calling_prompt = get_update_memory_messages(
    retrieved_old_memory,    # 已有记忆
    new_retrieved_facts,     # 新事实
)

response = self.llm.generate_response(
    messages=[{"role": "user", "content": function_calling_prompt}],
    response_format={"type": "json_object"},
)
# LLM 返回: {"memory": [{"id": "...", "text": "...", "event": "ADD|UPDATE|DELETE|NONE"}]}

LLM 决策规则

操作 条件 示例
ADD 新信息，记忆中不存在 记忆为空，新事实 "Name is John"
UPDATE 语义不同，需要更新 "User likes cricket" → "Loves to play cricket with friends"
DELETE 矛盾或过时信息 旧记忆需要被删除
NONE 语义相同，仅表述不同 "Likes cheese pizza" vs "Loves cheese pizza"

UPDATE 示例

旧记忆:
[
  {"id": "0", "text": "User is a software engineer"},
  {"id": "1", "text": "User lives in Beijing"}
]

新事实:
["User is a backend engineer", "User moved to Shanghai"]

LLM 决策:
{
  "memory": [
    {"id": "0", "text": "User is a backend engineer", "event": "UPDATE", "old_memory": "User is a software engineer"},
    {"id": "1", "text": "User lives in Shanghai", "event": "UPDATE", "old_memory": "User lives in Beijing"}
  ]
}

3.3 第三层：MD5 Hash

代码位置: mem0/memory/main.py:1185

metadata["hash"] = hashlib.md5(data.encode()).hexdigest()

作用: 即使前两层漏过，Hash 可用于应用层去重检查。

4. 记忆读取流程

4.1 读取方法

方法 位置 功能
search() 行 840-938 基于查询字符串语义搜索
get() 行 692-733 通过记忆 ID 获取单条
get_all() 行 735-791 获取用户所有记忆

4.2 读取返回数据结构

{
    "results": [
        {
            "id": "mem-uuid-xxx",
            "memory": "用户喜欢喝咖啡",
            "score": 0.95,              # 相似度分数
            "created_at": "2026-03-29T...",
            "metadata": {...}
        },
        ...
    ]
}

5. 使用示例

5.1 Personal Assistant 场景

from mem0 import MemoryClient

client = MemoryClient()

def chat_user(user_input: str, user_id: str = "user_123"):
    # 1. 搜索相关记忆
    memories = client.search(user_input, user_id=user_id)

    # 2. 格式化为上下文字符串
    memory_context = "\n".join(f"- {m['memory']}" for m in memories)

    # 3. 注入到 prompt
    prompt = f"""
You are a helpful personal assistant.

Here is what you remember about the user:
{memory_context}

User question: {user_input}
"""
    response = agent.run(prompt)

    # 4. 将对话存入记忆
    client.add(
        f"User: {user_input}\nAssistant: {response.content}",
        user_id=user_id
    )

    return response.content

5.2 对话轮次示例

轮次 用户输入 记忆操作 结果
第 1 轮 "我叫张三" 写: 存入 "用户名叫张三" "好的，张三！"
第 2 轮 "我喜欢跑步" 写: 存入 "用户喜欢跑步" "好的，记录下来了！"
第 3 轮 "我叫什么？" 读: 搜索到 "用户名叫张三" "你叫张三！"

6. 核心 API 总结

from mem0 import MemoryClient

client = MemoryClient()

# 添加记忆
client.add(
    messages=[{"role": "user", "content": "Hello"}],
    user_id="user_123"
)

# 搜索记忆
results = client.search(
    query="What did I say my name was?",
    user_id="user_123",
    limit=10
)

# 获取单条记忆
memory = client.get(memory_id="mem-uuid-xxx")

# 获取所有记忆
all_memories = client.get_all(user_id="user_123")

# 更新记忆
client.update(memory_id="mem-uuid-xxx", data="Updated content")

# 删除记忆
client.delete(memory_id="mem-uuid-xxx")

# 获取历史
history = client.history(memory_id="mem-uuid-xxx")

7. 配置说明

7.1 支持的向量存储

# Chroma (本地)
config = {"vector_store": {"provider": "chroma"}}

# Pinecone
config = {"vector_store": {"provider": "pinecone", "config": {"api_key": "...", "environment": "..."}}}

# Qdrant
config = {"vector_store": {"provider": "qdrant", "config": {"host": "...", "port": "..."}}}

# Redis
config = {"vector_store": {"provider": "redis", "config": {"host": "...", "port": "..."}}}

7.2 初始化示例

from mem0 import Memory

# 使用默认配置
memory = Memory()

# 使用自定义配置
memory = Memory(
    config={
        "vector_store": {"provider": "chroma"},
        "embedding_model": {"provider": "openai", "config": {"model": "text-embedding-3-small"}}
    }
)

8. 设计思想总结

  1. 事实提取而非原始对话: LLM 从对话中提取结构化事实，而非存储原始对话
  2. 语义去重: 通过向量相似度 + LLM 语义理解实现智能去重
  3. 增量更新: 支持 ADD/UPDATE/DELETE，而非简单覆盖
  4. 历史追踪: SQLite 记录完整变更历史，支持审计和回溯
  5. 多存储后端: 支持多种向量数据库，灵活适配不同场景

9. 相关文件索引

文件 说明
mem0/memory/main.py 核心记忆实现
mem0/memory/base.py 抽象基类
mem0/memory/graph_memory.py 图结构记忆
mem0/memory/storage.py SQLite 存储管理
mem0/configs/prompts.py LLM 提示词模板
mem0/client/main.py API 客户端
examples/misc/personal_assistant_agno.py 个人助手示例
examples/misc/study_buddy.py 学习伙伴示例
Next ReAct、Plan-and-Execute、Plan-Execute-RePlan 三种 Agent 范式分别适合什么场景？