AI 记忆系统 Mem0 快速入门
原
GrissomFI约 11 分钟
GrissomFI约 11 分钟AI 记忆系统 Mem0 快速入门
1. 项目概述
Mem0（发音为 "mem-zero"）是一个为大型语言模型（LLM）应用设计的自改进记忆层，它通过提供持久化、个性化的记忆能力来增强 AI 助手和智能体的表现。该项目同时提供托管平台服务和开源解决方案，让开发者能够为 AI 应用添加上下文记忆功能。
1.1 核心优势
• 准确性提升：在 LOCOMO 基准测试中比 OpenAI Memory 准确率提升 26% • 响应速度：相比全上下文方法响应速度提升 91% • 成本优化：相比全上下文方法 token 使用量降低 90% • 检索性能：亚 50 毫秒的快速记忆查找
1.2 应用场景
• AI 助手：提供一致、上下文丰富的对话体验 • 客户支持：回忆历史工单和用户历史记录，提供个性化帮助 • 医疗保健：跟踪患者偏好和历史记录，提供个性化护理 • 生产力和游戏：基于用户行为自适应工作流程和环境
2. 快速入门
Mem0 提供两种集成方式：托管平台（推荐）和开源自托管版本。
2.1 托管平台（推荐）
2.1.1 安装和配置
# Python
pip install mem0ai
# Node.js
npm install mem0ai1. 访问 Mem0 平台 2. 注册账号并获取 API 密钥
2.1.2 基本使用
from mem0 import MemoryClient
# 初始化客户端
client = MemoryClient(api_key="your-api-key")
# 添加记忆
messages = [
{"role": "user", "content": "我喜欢印度菜，但对奶酪过敏不能吃披萨"},
{"role": "assistant", "content": "我会记住您的饮食偏好"}
]
client.add(messages, user_id="user123")
# 搜索记忆
results = client.search("饮食偏好", user_id="user123")
print(results)
# 获取所有记忆
all_memories = client.get_all(user_id="user123")2.2 开源自托管版本
2.2.1 基础配置
import os
from mem0 import Memory
# 设置 OpenAI API 密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
# 初始化记忆系统
memory = Memory()
# 添加记忆
memory.add([
{"role": "user", "content": "我是程序员，喜欢写 Python 代码"},
{"role": "assistant", "content": "已记录您的职业信息和技术偏好"}
], user_id="developer001")
# 搜索相关记忆
results = memory.search("技术栈", user_id="developer001")2.2.2 高级配置（使用 Qdrant 向量数据库）
import os
from mem0 import Memory
# 启动 Qdrant 服务
# docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant
config = {
"vector_store": {
"provider": "qdrant",
"config": {
"host": "localhost",
"port": 6333,
"collection_name": "mem0"
}
},
"llm": {
"provider": "openai",
"config": {
"api_key": "your-openai-api-key",
"model": "gpt-4"
}
}
}
memory = Memory.from_config(config)3. 技术原理与架构
3.1 分层架构设计
Mem0 采用模块化的分层架构设计，通过工厂模式实现组件的灵活配置和扩展：
• 应用层：提供统一的 Memory 和 AsyncMemory API 接口，支持同步和异步操作 • 记忆管理层：负责记忆的生命周期管理，包括提取、存储、更新和检索 • 存储层：支持 20+ 种向量数据库（Qdrant、Chroma、Pinecone 等）和图数据库（Neo4j、Kuzu 等） • 模型层：集成 15+ 种 LLM 提供商（OpenAI、Anthropic、Ollama 等）和多种嵌入模型
3.2 核心组件架构
3.2.1 Memory 核心类
class Memory(MemoryBase):
def __init__(self, config: MemoryConfig = MemoryConfig()):
# 通过工厂模式创建各组件
self.embedding_model = EmbedderFactory.create(
self.config.embedder.provider,
self.config.embedder.config,
self.config.vector_store.config,
)
self.vector_store = VectorStoreFactory.create(
self.config.vector_store.provider,
self.config.vector_store.config
)
self.llm = LlmFactory.create(
self.config.llm.provider,
self.config.llm.config
)
self.db = SQLiteManager(self.config.history_db_path)
# 可选的图存储支持
if self.config.graph_store.config:
self.graph = GraphStoreFactory.create(
self.config.graph_store.provider,
self.config
)3.2.2 工厂模式组件管理
• LlmFactory：支持 15+ LLM 提供商，包括 OpenAI、Anthropic、Ollama、Azure OpenAI 等 • EmbedderFactory：支持多种嵌入模型，包括 OpenAI、HuggingFace、Ollama 等 • VectorStoreFactory：支持 20+ 向量数据库，从本地 FAISS 到云端 Pinecone • GraphStoreFactory：支持图数据库，包括 Neo4j、Kuzu、Memgraph 等
3.2.3 配置系统
class MemoryConfig(BaseModel):
vector_store: VectorStoreConfig # 向量存储配置
llm: LlmConfig # LLM 配置
embedder: EmbedderConfig # 嵌入模型配置
graph_store: GraphStoreConfig # 图存储配置（可选）
history_db_path: str # 历史数据库路径
custom_fact_extraction_prompt: Optional[str] # 自定义提取提示
custom_update_memory_prompt: Optional[str] # 自定义更新提示3.3 智能记忆处理流程
3.3.1 记忆添加流程（_add_to_vector_store）
1. 消息解析：使用 parse_messages()解析输入消息格式2. 事实提取：通过 LLM 和自定义提示提取关键事实 # 使用 JSON 格式确保结构化输出
response = self.llm.generate_response(
messages=[{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}],
response_format={"type": "json_object"}
)3. 向量化处理：将提取的事实转换为高维向量表示 4. 相似性检索：在现有记忆中搜索相关内容，避免重复存储 5. 智能更新：根据相似性决定创建新记忆或更新现有记忆 6. 多存储同步：同时更新向量存储、图存储和历史数据库
3.3.2 记忆检索流程（search）
1. 查询向量化：将搜索查询转换为向量表示 2. 向量相似性搜索：在向量空间中找到最相关的记忆 3. 阈值过滤：根据相似性阈值过滤结果 4. 元数据过滤：支持 user_id、agent_id、run_id 等多维度过滤 5. 结果排序：按相似性分数排序返回结果
3.3.3 图关系处理（可选）
当启用图存储时，Mem0 会额外处理记忆间的关系：
if self.enable_graph:
# 将记忆添加到图数据库，建立实体关系
self._add_to_graph(messages, filters)3.4 API 接口设计
3.4.1 核心 API 接口
Mem0 提供了简洁而强大的 API 接口，遵循 RESTful 设计原则：
# 记忆管理核心接口
class Memory:
def add(self, messages, user_id=None, agent_id=None, run_id=None, metadata=None):
"""添加新记忆到系统中"""
pass
def get(self, memory_id):
"""根据 ID 获取特定记忆"""
pass
def get_all(self, user_id=None, agent_id=None, run_id=None):
"""获取指定范围内的所有记忆"""
pass
def search(self, query, user_id=None, agent_id=None, run_id=None, limit=100):
"""基于语义相似度搜索相关记忆"""
pass
def update(self, memory_id, data):
"""更新现有记忆内容"""
pass
def delete(self, memory_id):
"""删除特定记忆"""
pass
def delete_all(self, user_id=None, agent_id=None, run_id=None):
"""批量删除记忆"""
pass
# 实体管理接口
def create_user(data): """创建用户实体"""
def get_user(user_id): """获取用户信息"""
def update_user(user_id, data): """更新用户信息"""
def delete_user(user_id): """删除用户"""
# 智能体管理接口
def create_agent(data): """创建智能体"""
def get_agent(agent_id): """获取智能体信息"""
def update_agent(agent_id, data): """更新智能体配置"""
def delete_agent(agent_id): """删除智能体"""3.4.2 配置系统架构
Mem0 采用分层配置设计，支持多种存储后端和模型提供商：
# 完整配置示例
config = {
# 向量存储配置
"vector_store": {
"provider": "qdrant", # 支持: qdrant, pinecone, weaviate, chroma, faiss
"config": {
"host": "localhost",
"port": 6333,
"collection_name": "mem0",
"embedding_model_dims": 1536 # 嵌入维度
}
},
# 大语言模型配置
"llm": {
"provider": "openai", # 支持: openai, anthropic, google, ollama
"config": {
"api_key": "your-api-key",
"model": "gpt-4",
"temperature": 0.1, # 控制输出随机性
"max_tokens": 1000 # 最大输出长度
}
},
# 嵌入模型配置
"embedder": {
"provider": "openai",
"config": {
"api_key": "your-api-key",
"model": "text-embedding-3-small",
"dimensions": 1536 # 嵌入向量维度
}
},
# 图数据库配置（可选）
"graph_store": {
"provider": "neo4j",
"config": {
"url": "bolt://localhost:7687",
"username": "neo4j",
"password": "password"
}
}
}3.4.3 API 设计原则
1. 简洁性：最小化必需参数，提供合理默认值 2. 一致性：统一的命名规范和返回格式 3. 扩展性：支持元数据和自定义配置 4. 性能优化：批量操作和异步支持 5. 错误处理：详细的错误信息和状态码
4. 与其他记忆系统对比
4.1 主要系统对比
| 类型 | |||||
| 核心特点 | |||||
| 架构设计 | |||||
| 性能 | |||||
| 准确性 | |||||
| 部署方式 | |||||
| 隐私保护 | |||||
| 集成难度 |
4.2 性能基准测试
根据 LOCOMO 基准测试结果：
| 记忆准确性 | ||||||
| F1 分数提升 | ||||||
| BLEU-1 提升 | ||||||
| 响应速度（平均） | ||||||
| Token 使用量 | ||||||
| 成本效率 |
*基于 LoCoMo 基准测试，相对于基线模型的提升
4.3 选择建议
选择 Mem0 的场景：
• 需要高性能、低延迟的记忆系统 • 希望获得自改进的智能记忆能力 • 需要跨会话的持久化记忆 • 成本敏感的应用
选择其他系统的场景：
• MemoryOS：需要极致性能和隐私保护，长期对话场景，本地部署优先 • Zep：企业级应用，需要丰富的元数据支持 • Letta：需要复杂的多代理系统 • LangChain Memory：已深度使用 LangChain 生态
4.4 MemoryOS 技术特色
分层记忆架构：
# MemoryOS 三层记忆结构
class MemoryOS:
def __init__(self):
self.short_term_memory = [] # 短期记忆 (STM)
self.mid_term_memory = [] # 中期记忆 (MTM)
self.long_term_persona = {} # 长期个人记忆 (LPM)
def update_memory_hierarchy(self, dialogue):
# 基于热度评分的动态更新机制
heat_score = self.calculate_heat_score(dialogue)
if heat_score > self.mid_term_threshold:
self.promote_to_mid_term(dialogue)核心优势：
• 操作系统级设计：借鉴 OS 内存管理原理，实现分层存储和动态更新 • 热度评分机制：自动识别重要信息，智能管理记忆生命周期 • 本地优先部署：确保数据隐私和确定性，避免云端依赖 • MCP 协议支持：通过 OpenMemory MCP 实现多工具记忆共享
5. 集成示例
5.1 与 LangChain 集成
from langchain.memory import ConversationBufferMemory
from langchain.schema import BaseMemory
from mem0 import Memory
from typing import Any, Dict, List
class Mem0LangChainMemory(BaseMemory):
"""将 Mem0 集成到 LangChain 的自定义记忆类"""
def __init__(self, user_id: str, **kwargs):
super().__init__(**kwargs)
self.mem0 = Memory()
self.user_id = user_id
self.memory_key = "history"
@property
def memory_variables(self) -> List[str]:
return [self.memory_key]
def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""从 Mem0 加载相关记忆"""
query = inputs.get("input", "")
# 搜索相关记忆
memories = self.mem0.search(query, user_id=self.user_id, limit=5)
# 格式化记忆内容
memory_content = "\n".join([
f"记忆: {memory['memory']}" for memory in memories
])
return {self.memory_key: memory_content}
def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
"""保存对话上下文到 Mem0"""
messages = [
{"role": "user", "content": inputs["input"]},
{"role": "assistant", "content": outputs["output"]}
]
self.mem0.add(messages, user_id=self.user_id)
def clear(self) -> None:
"""清除用户记忆"""
self.mem0.delete_all(user_id=self.user_id)5.2 与 FastAPI 集成
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from mem0 import Memory
from typing import List, Optional
app = FastAPI(title="Mem0 API Service")
memory = Memory()
class ChatMessage(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
messages: List[ChatMessage]
user_id: str
metadata: Optional[dict] = None
@app.post("/chat")
async def chat_with_memory(request: ChatRequest):
"""处理聊天并管理记忆"""
try:
# 转换消息格式
messages = [msg.dict() for msg in request.messages]
# 添加到记忆
result = memory.add(
messages,
user_id=request.user_id,
metadata=request.metadata
)
return {"status": "success", "memories_added": len(result)}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/memories/{user_id}")
async def get_user_memories(user_id: str):
"""获取用户所有记忆"""
try:
memories = memory.get_all(user_id=user_id)
return {"memories": memories}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))6. 总结
Mem0 通过其创新的记忆层设计，为 AI 应用提供了强大的个性化能力。无论是选择托管平台还是自托管方案，开发者都能快速集成并获得显著的性能提升。其灵活的架构设计支持多种存储后端和 LLM 提供商，使其能够适应各种应用场景和规模需求。
7. 参考资料
7.1 官方资源
• Mem0 官方文档 - 完整的产品文档和使用指南 • Mem0 GitHub 仓库 - 开源代码和社区贡献 • Mem0 API 参考 - 详细的 API 接口文档 • Mem0 平台 - 托管服务平台 • Mem0 社区论坛 - 开发者交流社区
7.2 技术论文与研究
• MemoryOS: Memory OS of AI Agent - 分层记忆架构的学术研究 • LoCoMo Benchmark - 长期对话记忆基准测试 • RAG 与记忆系统结合研究 - 检索增强生成相关论文 • 向量数据库技术综述 - 向量存储技术研究
7.3 相关开源项目
• MemoryOS - 分层记忆操作系统 • Zep - 长期记忆存储解决方案 • Letta (MemGPT) - 虚拟上下文管理系统 • LangChain Memory - LangChain 记忆组件 • ChromaDB - 开源向量数据库 • Qdrant - 高性能向量搜索引擎
7.4 集成框架与工具
• LangChain - LLM 应用开发框架 • LlamaIndex - 数据框架和索引工具 • FastAPI - 现代 Web API 框架 • Streamlit - 快速构建数据应用 • Gradio - 机器学习模型界面工具
7.5 性能基准与评估
• LOCOMO 基准测试 - 长期对话记忆评估 • MTEB Leaderboard - 文本嵌入基准 • VectorDBBench - 向量数据库性能测试 • AI Memory Benchmark - AI 记忆能力评估
7.7 学习资源
• Building LLM Applications with Memory - DeepLearning.AI 课程 • RAG 系统设计最佳实践 - 检索增强生成指南 • Memory in AI Systems - AI 系统中的记忆机制