给大模型加上“记忆”，深入探索 Mem0 项目

背景介绍

在之前的软件应用中，我们总会在应用中保留大量的用户历史操作记录，方便用户下次使用时可以快速查看和复用，甚至基于这些用户记录可以为用户提供个性化的服务。而这些记录往往都保存在传统的结构化或非结构化数据库中。

在大模型的应用，特别是助手类的大模型应用中，我们往往需要处理大量语义化的文本或多模态的信息，方便后续快速匹配，从而提供个性化的服务。为了支持这种语义检索的需求，往往会将数据保存至向量数据库中。

向量数据库执行语义检索相对方便，但是包含语义内容的文本管理比较困难，特别是涉及到相关内容的更新和替换。在这种背景下，Mem0 应运而生。Mem0 可以给大模型应用提供智能记忆层，可以记住用户偏好，适应个人需求，并随着时间的推移不断改进，从而方便应用提供更加个性化的体验和服务。
请添加图片描述

从开源依赖，Github 上的 star 数量一路飙升，截止目前已经 20.6k 的 star 了。

上手体验

Mem0 提供了两种接入方案，一种是使用官方的托管平台，另一种是使用官方提供的 python 依赖包 mem0ai执行。下面以本地依赖包形式了解下 Mem0 的使用。

依赖配置

首先需要通过 pip 安装所需的依赖包:

pip install mem0ai

Mem0 需要依赖大模型才能执行，默认使用的 gpt-4o, 因此需要配置对应的 ChatGPT API 秘钥，需要将秘钥配置至环境变量中:

import os
os.environ["OPENAI_API_KEY"] = "sk-xxx"

动手实践

Mem0 实际使用在大模型应用中，我们主要关注下面的能力：

针对特定用户添加非结构化的文本信息；
特定用户根据查询问题进行文本检索；

针对上面的关注的能力，使用 Mem0 可以简单满足需求：

from mem0 import Memory# 初始化记忆对象m = Memory()# 1. 针对特定用户添加非结构化的文本信息result = m.add("I am working on improving my tennis skills. Suggest some online courses.", user_id="alice", metadata={"category": "hobbies"})# 2. 特定用户根据查询问题进行文本检索related_memories = m.search(query="What are Alice's hobbies?", user_id="alice")

可以看到常规的功能使用比较简单，看起来几行代码就可以快速为大模型应用加上“记忆”。

实现方案

官方方案概述

首先可以从官方描述中大致了解 Mem0 的实现方案，从官方文档了解到下面的信息：

Mem0 利用混合数据库方法来管理和检索人工智能代理和助手的长期记忆。每个内存都与一个唯一的标识符相关联，例如用户 ID 或代理 ID，允许 Mem0 组织和访问特定于个人或上下文的内存。

当使用 add() 方法将消息添加到 Mem0 时，系统会提取相关事实和偏好并将其存储在数据存储中：矢量数据库、键值数据库和图形数据库。这种混合方法可确保以最有效的方式存储不同类型的信息，从而使后续搜索快速有效。

当人工智能代理或LLM需要回忆记忆时，它会使用 search() 方法。然后 Mem0 在这些数据存储中执行搜索，从每个源检索相关信息。然后，该信息通过评分层，根据相关性、重要性和新近度评估其重要性。这确保了只显示最个性化和最有用的上下文。

可以大致了解到 Mem0 是基于混合数据库来管理，同时支持向量、键值和图形数据库，但是具体的实现细节依旧不清楚，这种情况下 Talk is cheap, show me the code.，下面我们通过源码来了解 Mem0 的实现方案。

方案深入解析

文本检索

Mem0 中的文本检索是通过调用 Memory.search() 方法实现，文本检索的主要流程如下所示：

检索文本向量化，构造过滤条件，方便基于用户 id 等进行过滤；
基于向量数据库进行检索和过滤，默认使用的向量数据库为 Qdrant;
可选基于知识图谱的检索（从 v1.1 开始支持）；

从上面的流程来看，主流程就是简单的向量检索，同时利用向量数据库提供的元信息过滤的能力进行用户区分的封装，实现如下所示：

filters = filters or {}
if user_id:filters["user_id"] = user_id
if agent_id:filters["agent_id"] = agent_id
if run_id:filters["run_id"] = run_id# 查询向量化embeddings = self.embedding_model.embed(query)# 向量检索memories = self.vector_store.search(query=embeddings, limit=limit, filters=filters
)# 可选的知识图谱检索if self.version == "v1.1":if self.enable_graph:graph_entities = self.graph.search(query)

文本添加

Mem0 中非结构化文本添加是通过调用 Memory.add() 方法实现，文本添加的主要流程如下所示：

将添加的文本进行向量化，从向量数据库中检索相关内容；
将检索的相关内容与新增的文本提供给大模型，大模型判断所需采取的动作列表，可选动作包含 新增文本, 更新原有的文本, 删除原有文本；
根据大模型给出的动作列表依次执行；

可以看到上面的流程中主要基于大模型提供的能力处理历史信息的更新问题，对应的实现如下所示：

# 从向量库中检索已有的文本内容embeddings = self.embedding_model.embed(data)
existing_memories = self.vector_store.search(query=embeddings,limit=5,filters=filters,
)# 转换检索到的文本内容，构造成特定格式，方便调用大模型existing_memories = [MemoryItem(id=mem.id,score=mem.score,metadata=mem.payload,memory=mem.payload["data"],)for mem in existing_memories
]
serialized_existing_memories = [item.model_dump(include={"id", "memory", "score"})for item in existing_memories
]
messages = get_update_memory_messages(serialized_existing_memories, extracted_memories
)# 调用大模型，确定需要采取的动作列表tools = [ADD_MEMORY_TOOL, UPDATE_MEMORY_TOOL, DELETE_MEMORY_TOOL]
response = self.llm.generate_response(messages=messages, tools=tools)tool_calls = response["tool_calls"]response = []
if tool_calls:available_functions = {"add_memory": self._create_memory_tool,"update_memory": self._update_memory_tool,"delete_memory": self._delete_memory_tool,}# 依次执行动作列表，可选动作为 `新增文本`， `更新原有的文本`, `删除原有文本`for tool_call in tool_calls:function_name = tool_call["name"]function_to_call = available_functions[function_name]function_args = tool_call["arguments"]if function_name in ["add_memory", "update_memory"]:function_args["metadata"] = metadatafunction_result = function_to_call(**function_args)response.append({"id": function_result,"event": function_name.replace("_memory", ""),"data": function_args.get("data"),})