Mem0 深度解析：开源本地 AI 智能体长期记忆系统原理与实战

def extract_and_store_memory(messages, user_id, metadata=None):
    # 使用 LLM 提取关键信息
    extracted_info = llm.extract_key_information(messages)
    
    # 结构化为记忆单元
    memory_unit = {
        "content": extracted_info,
        "source_messages": messages,
        "user_id": user_id,
        "timestamp": datetime.now(),
        "metadata": metadata or {}
    }
    
    # 向量化记忆
    vector = embedder.embed(extracted_info)
    
    # 存储在向量数据库中
    vector_store.add(vector, memory_unit)
    
    return memory_unit

def retrieve_memories(query, user_id, limit=5):
    # 向量化查询
    query_vector = embedder.embed(query)
    
    # 在向量空间中搜索相似记忆
    similar_memories = vector_store.search(
        query_vector, 
        filter={"user_id": user_id},
        limit=limit
    )
    
    # 根据相关性、时效性和重要性排序
    ranked_memories = rank_memories(similar_memories)
    
    return ranked_memories

ollama pull mxbai-embed-large
# 验证
curl http://localhost:11434/api/embeddings -d '{
  "model": "mxbai-embed-large",
  "prompt": "Represent this sentence for searching relevant passages: The sky is blue because of Rayleigh scattering"
}'

docker pull qdrant/qdrant
docker run -p 6333:6333 --name qdrant -d qdrant/qdrant

git clone https://github.com/qdrant/qdrant-web-ui.git
cd qdrant-web-ui
npm install
npm start

uv add dotenv langchain langgraph mem0ai ollama

import os
import json
from mem0 import Memory
from mem0.configs.base import MemoryConfig, LlmConfig, EmbedderConfig

# 设置 API 密钥
os.environ["DEEPSEEK_API_KEY"] = "sk-……" # 替换为你的 DeepSeek API 密钥

# 创建配置
config = {
    "llm": {
        "provider": "deepseek",
        "model": "deepseek-chat",
        "temperature": 0.2,
        "max_tokens": 2000,
        "top_p": 1.0
    },
    "embedder": {
        "provider": "ollama",
        "model": "mxbai-embed-large"
    },
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "collection_name": "mem0_new",
            "embedding_model_dims": 768,
            "host": "localhost",
            "port": 6333,
        }
    },
    "reset_vector_store": True
}

# 创建 Memory 实例
m = Memory.from_config(config)

# 添加对话记忆
messages = [
    {"role": "user", "content": "我喜欢科幻电影，尤其是星际穿越。"},
    {"role": "assistant", "content": "星际穿越确实是一部经典的科幻电影！我会记住你喜欢这类电影。"},
    {"role": "user", "content": "我也喜欢克里斯托弗·诺兰的其他电影。"},
    {"role": "assistant", "content": "诺兰的作品确实很出色！除了《星际穿越》，他还导演了《盗梦空间》、《信条》、《记忆碎片》等经典作品。您对这些电影有什么特别的看法吗？"}
]

# 添加记忆，并附加元数据
m.add(messages, user_id="alice", metadata={"category": "movies"})

# 检索记忆
memories = m.search(query="我喜欢什么类型的电影？", user_id="alice")
print(json.dumps(memories, indent=2, ensure_ascii=False))

# 获取所有记忆
all_memories = m.get_all(user_id="alice")
print(json.dumps(all_memories, indent=2, ensure_ascii=False))

{
  "results": [
    {
      "id": "b052c579-23dd-4d89-933e-4c6ceb4e8958",
      "memory": "喜欢克里斯托弗·诺兰的其他电影",
      "hash": "a46f8ee3dcfded0fd39d1b5006810adc",
      "metadata": {
        "category": "movies"
      },
      "score": 0.6853715,
      "created_at": "2025-06-24T07:36:23.381311-07:00",
      "updated_at": null,
      "user_id": "alice"
    },
    {
      "id": "2fae3fa7-0fa7-4164-b521-5c0d821d1c7a",
      "memory": "喜欢科幻电影，尤其是星际穿越",
      "hash": "55309121dad25e817ea0b3f132db1e6c",
      "metadata": {
        "category": "movies"
      },
      "score": 0.6162164,
      "created_at": "2025-06-24T07:36:23.367111-07:00",
      "updated_at": null,
      "user_id": "alice"
    }
  ]
}

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from mem0 import Memory
from langchain_core.messages import HumanMessage, AIMessage
from langchain.chat_models import init_chat_model

# 设置 API 密钥
api_key = os.getenv("DEEPSEEK_API_KEY", "sk-……") # 请替换为你的 DeepSeek API 密钥
os.environ["DEEPSEEK_API_KEY"] = api_key

# 创建配置
config = {
    "llm": {
        "provider": "deepseek",
        "model": "deepseek-chat",
        "temperature": 0.2,
        "max_tokens": 2000,
        "top_p": 1.0
    },
    "embedder": {
        "provider": "ollama",
        "model": "mxbai-embed-large"
    },
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "collection_name": "mem0_new",
            "embedding_model_dims": 768,
            "host": "localhost",
            "port": 6333,
        }
    },
    "reset_vector_store": True
}

# 初始化 Mem0
m = Memory.from_config(config)

llm = init_chat_model(
    "deepseek-chat",  # 使用DeepSeek模型
    api_key=api_key
)

# 定义状态类型
class State(TypedDict):
    messages: list
    user_id: str
    memories: list

# 定义记忆检索函数
def retrieve_memories(state: State):
    query = state["messages"][-1]["content"]
    user_id = state.get("user_id", "alice")
    memories = m.search(query=query, user_id=user_id)
    return {"memories": memories["results"]}

# 定义 LLM 节点
def llm_node(state: State):
    # 构建系统提示，包含记忆
    memories_str = "\n".join(f"- {m['memory']}" for m in state["memories"])
    system_prompt = f"你是一个有记忆能力的助手。以下是用户的相关记忆：\n{memories_str}\n请基于这些记忆回答问题。"
    
    # 准备消息
    messages = [{"role": "system", "content": system_prompt}]
    messages.extend(state["messages"])
    
    # 调用 LLM
    response = llm.invoke(messages)
    
    # 更新消息列表，转换为 dict
    new_message = {"role": "assistant", "content": response.content}
    return {"messages": state["messages"] + [new_message]}

# 定义记忆存储函数
def store_memories(state: State):
    user_id = state.get("user_id", "alice")
    # 现在 state["messages"] 已全为 dict
    m.add(state["messages"], user_id=user_id)
    return {}

# 构建图
graph_builder = StateGraph(State)
graph_builder.add_node("retrieve_memories", retrieve_memories)
graph_builder.add_node("llm", llm_node)
graph_builder.add_node("store_memories", store_memories)

# 添加边
graph_builder.add_edge(START, "retrieve_memories")
graph_builder.add_edge("retrieve_memories", "llm")
graph_builder.add_edge("llm", "store_memories")
graph_builder.add_edge("store_memories", END)

# 编译图
graph = graph_builder.compile()

# 使用图处理查询
result = graph.invoke({
    "messages": [{"role": "user", "content": "我最喜欢什么电影？"}],
    "user_id": "alice",
    "memories": []
})

print(result)

{
	'messages': [{
		'role': 'user',
		'content': '我最喜欢什么电影？'
	}, {
		'role': 'assistant',
		'content': '根据我们的记忆，您最喜欢的电影是《星际穿越》！这部由克里斯托弗·诺兰执导的科幻杰作确实令人难忘，特别是其中关于时间、空间和人类情感的深刻探讨。您还喜欢诺兰的其他作品，比如《盗梦空间》和《黑暗骑士》三部曲等。需要聊聊最近重温《星际穿越》的新发现吗？'
	}],
	'user_id': 'alice',
	'memories': [{
		'id': '2fae3fa7-0fa7-4164-b521-5c0d821d1c7a',
		'memory': '喜欢科幻电影，尤其是星际穿越',
		'hash': '55309121dad25e817ea0b3f132db1e6c',
		'metadata': {
			'category': 'movies'
		},
		'score': 0.62142074,
		'created_at': '2025-06-24T07:36:23.367111-07:00',
		'updated_at': None,
		'user_id': 'alice'
	}, {
		'id': 'b052c579-23dd-4d89-933e-4c6ceb4e8958',
		'memory': '喜欢克里斯托弗·诺兰的其他电影',
		'hash': 'a46f8ee3dcfded0fd39d1b5006810adc',
		'metadata': {
			'category': 'movies'
		},
		'score': 0.61610943,
		'created_at': '2025-06-24T07:36:23.381311-07:00',
		'updated_at': None,
		'user_id': 'alice'
	}]
}

你是一个有记忆能力的助手。以下是用户的相关记忆：
{{memories}}
请基于这些记忆回答问题。

你是我的得力助手。
请将以下内容作为你的知识储备，放在 <context></context> 标签内。
<context>
{{#context#}}
</context>
根据用户的问题，检索已有记忆的结果是：
<memory>
{{#1747805606313.text#}}
</memory>
当回答用户问题时：
若不知道答案，就直接说不知道。
若不确定，就向用户澄清问题。
避免提及信息来源于上下文。
根据用户提问的语言进行回答。

# 添加带标签的记忆
m.add(messages, user_id="alice", metadata={
    "category": "movies",
    "tags": ["sci-fi", "preferences"],
    "importance": "high"
})

# 按类别检索
memories = m.search(
    query="科幻电影", 
    user_id="alice", 
    filter={"category": "movies"}
)

# 按标签检索
memories = m.search(
    query="科幻电影", 
    user_id="alice", 
    filter={"tags": "sci-fi"}
)

# 按重要性检索
memories = m.search(
    query="科幻电影", 
    user_id="alice", 
    filter={"importance": "high"}
)

# 导出所有记忆
all_memories = m.get_all(user_id="alice")
import json
with open("alice_memories.json", "w") as f:
    json.dump(all_memories, f, ensure_ascii=False, indent=2)

# 按标签和类别检索
memories = m.search(
    query="深度学习",
    user_id="alice",
    filter={
        "category": "ml",
        "tags": ["neural-network", "research"]
    },
    limit=10
)

# 按重要性检索
memories = m.search(
    query="紧急事项",
    user_id="alice",
    filter={"importance": "high"}
)

# 批量导出所有记忆（适合备份/迁移）
all_memories = m.get_all(user_id="alice")
with open("alice_memories.json", "w") as f:
    import json
    json.dump(all_memories, f, ensure_ascii=False, indent=2)

# 批量导入记忆
with open("alice_memories.json", "r") as f:
    memories_data = json.load(f)
for memory in memories_data:
    m.add([memory["source_messages"]], user_id="alice", metadata=memory.get("metadata"))

# 创建组织并分配用户
org_id = m.create_organization(name="Acme Corp")
m.add_user_to_organization(user_id="alice", org_id=org_id, role="admin")

# 以组织维度设置隔离
memories = m.search(query="合规", user_id="alice", org_id=org_id)

POST /search_memory
{
  "query": "项目进度",
  "user_id": "alice",
  "org_id": "acme",
  "filter": {"category": "project", "tags": "progress"},
  "limit": 5
}

from mem0.plugins import register_preprocessor

def custom_preprocessor(memory_item):
    # 增加自定义元数据
    memory_item["metadata"]["custom_flag"] = True
    return memory_item

register_preprocessor(custom_preprocessor)