深入浅出分析最近火热的Mem0个性化AI记忆层
最近Mem0横空出世，官方称之为PA的记忆层，The memory layer for Personalized AI，有好事者还称这个是RAG的替代者，Mem0究竟为何物，背后的原理是什么，我们今天来一探究竟。
Mem0 介绍
开源地址：
https://github.com/mem0ai/mem0
官方介绍为：
Mem0 provides a smart, self-improving memory layer for Large Language Models, enabling personalized AI experiences across applications.
关键点，是为LLM提供的
智能的，可自我改进
的记忆层，从而可以实现在各种应用中提供
更加个性化的和连贯一致
的用户体验。
主要特点
- 多层次记忆：支持用户级、会话级和AI代理级的记忆保留。
- 自适应个性化：根据用户交互不断改进，提供精准个性化记忆。
- 开发者友好API：简单易用，易于集成。
- 跨平台一致性：保持不同设备上的行为一致性。
- 托管服务：简化部署和维护。
快速开始
安装：通过pip安装mem0ai。
pip install mem0ai
基本用法：
import os
from mem0 import Memory
# 依赖LLM提取记忆，所以需要open ai
os.environ["OPENAI_API_KEY"] = "xxx"
# 吃石化 Mem0
m = Memory()
# 通过add方法，存储非结构化的记忆，metadata提供schema定义
result = m.add("I am working on improving my tennis skills. Suggest some online courses.", user_id="alice", metadata={"category": "hobbies"})
print(result)
# Created memory: Improving her tennis skills. Looking for online suggestions.
# Retrieve memories
all_memories = m.get_all()
print(all_memories)
# 搜索记忆 Search memories
related_memories = m.search(query="What are Alice's hobbies?", user_id="alice")
print(related_memories)
# 更新记忆 Update a memory
result = m.update(memory_id="m1", data="Likes to play tennis on weekends")
print(result)
# Get memory history
history = m.history(memory_id="m1")
print(history)
上述的示例代码展示了如何添加记忆、检索记忆、搜索、更新和获取记忆历史。
注意代码里的
metadata
, 这里相当于定义了一个schema，让LLM从非结构化数据里提取相关的记忆信息。
原理分析
透过上面的示例代码，我们先来猜测下mem0的原理：
- 通过LLM+制定的metadata，抽取记忆信息，这里雷士知识图谱抽取，重点是定制一个合适的prompt来抽取有效信息
- 相关记忆信息通过向量化存储，因此可以支持记忆信息检索
- 记忆支持更新，例如demo里的hobbies更新
我们下载代码一探究竟。
记忆管理
def add(
self,
data,
user_id=None,
agent_id=None,
run_id=None,
metadata=None,
filters=None,
prompt=None,
):
"""
Create a new memory.
Args:
data (str): Data to store in the memory.
user_id (str, optional): ID of the user creating the memory. Defaults to None.
agent_id (str, optional): ID of the agent creating the memory. Defaults to None.
run_id (str, optional): ID of the run creating the memory. Defaults to None.
metadata (dict, optional): Metadata to store with the memory. Defaults to None.
filters (dict, optional): Filters to apply to the search. Defaults to None.
Returns:
str: ID of the created memory.
"""
if metadata is None:
metadata = {}
embeddings = self.embedding_model.embed(data)
filters = filters or {}
if user_id:
filters["user_id"] = metadata["user_id"] = user_id
if agent_id:
filters["agent_id"] = metadata["agent_id"] = agent_id
if run_id:
filters["run_id"] = metadata["run_id"] = run_id
if not prompt:
prompt = MEMORY_DEDUCTION_PROMPT.format(user_input=data, metadata=metadata)
extracted_memories = self.llm.generate_response(
messages=[
{
"role": "system",
"content": "You are an expert at deducing facts, preferences and memories from unstructured text.",
},
{"role": "user", "content": prompt},
]
)
existing_memories = self.vector_store.search(
name=self.collection_name,
query=embeddings,
limit=5,
filters=filters,
)
existing_memories = [
MemoryItem(
id=mem.id,
score=mem.score,
metadata=mem.payload,
text=mem.payload["data"],
)
for mem in existing_memories
]
serialized_existing_memories = [
item.model_dump(include={"id", "text", "score"})
for item in existing_memories
]
logging.info(f"Total existing memories: {len(existing_memories)}")
messages = get_update_memory_messages(
serialized_existing_memories, extracted_memories
)
# Add tools for noop, add, update, delete memory.
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:
# Create a new memory
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"]
logging.info(
f"[openai_func] func: {function_name}, args: {function_args}"
)
# Pass metadata to the function if it requires it
if function_name in ["add_memory", "update_memory"]:
function_args["metadata"] = metadata
function_result = function_to_call(**function_args)
# Fetch the memory_id from the response
response.append(
{
"id": function_result,
"event": function_name.replace("_memory", ""),
"data": function_args.get("data"),
}
)
capture_event(
"mem0.add.function_call",
self,
{"memory_id": function_result, "function_name": function_name},
)
capture_event("mem0.add", self)
return response
这里的逻辑比较简单
- 参数的判断、处理
- 通过 MEMORY_DEDUCTION_PROMPT 结合用户的data，抽取记忆，得到extracted_memories
- 然后通过data查询相关的existing_memories
- 然后将extracted_memories、existing_memories 拼接到一起，交予大模型，让大模型调用合适的tool来更新记忆，tools ：
[ADD_MEMORY_TOOL, UPDATE_MEMORY_TOOL, DELETE_MEMORY_TOOL] - 根据function call的结果，调用tool_calls更新记忆
本质上全部委托给大模型，通过prompt做了一定的约束。
相关prompt设计
ps，我们来看下相关度的prompt设计。
MEMORY_DEDUCTION_PROMPT = """
Deduce the facts, preferences, and memories from the provided text.
Just return the facts, preferences, and memories in bullet points:
Natural language text: {user_input}
User/Agent details: {metadata}
Constraint for deducing facts, preferences, and memories:
- The facts, preferences, and memories should be concise and informative.
- Don't start by "The person likes Pizza". Instead, start with "Likes Pizza".
- Don't remember the user/agent details provided. Only remember the facts, preferences, and memories.
Deduced facts, preferences, and memories:
从提供的文本中推断出事实、偏好和记忆。
仅以项目符号形式返回事实、偏好和记忆：
自然语言文本：{用户输入}
用户/代理详细信息：{元数据}
推断事实、偏好和记忆的约束：
- 事实、偏好和记忆应简洁且信息丰富。
- 不要以“此人喜欢披萨”开头。而是以“喜欢披萨”开头。
- 不要记住提供的用户/代理详细信息。只记住事实、偏好和记忆。
推断出的事实、偏好和记忆
再来看更新记忆的prompt：
You are an expert at merging, updating, and organizing memories. When provided with existing memories and new information, your task is to merge and update the memory list to reflect the most accurate and current information. You are also provided with the matching score for each existing memory to the new information. Make sure to leverage this information to make informed decisions about which memories to update or merge.
Guidelines:
- Eliminate duplicate memories and merge related memories to ensure a concise and updated list.
- If a memory is directly contradicted by new information, critically evaluate both pieces of information:
- If the new memory provides a more recent or accurate update, replace the old memory with new one.
- If the new memory seems inaccurate or less detailed, retain the original and discard the old one.
- Maintain a consistent and clear style throughout all memories, ensuring each entry is concise yet informative.
- If the new memory is a variation or extension of an existing memory, update the existing memory to reflect the new information.
Here are the details of the task:
- Existing Memories:
{existing_memories}
- New Memory: {memory}
您是合并、更新和组织记忆的专家。当您获得现有记忆和新信息时，您的任务是合并和更新记忆列表，以反映最准确和最新的信息。您还会获得每个现有记忆与新信息的匹配分数。请务必利用这些信息，就更新或合并哪些记忆做出明智的决策。
指南：
- 消除重复记忆并合并相关记忆，以确保列表简洁且是最新的。
- 如果新信息与某一记忆直接矛盾，请仔细评估这两部分信息：
- 如果新记忆提供了更新或更准确的更新内容，用新记忆替换旧记忆。
- 如果新记忆似乎不准确或不够详细，则保留原始记忆并丢弃新记忆。
- 在所有记忆中保持一致、清晰的风格，确保每个条目都简洁且内容丰富。
- 如果新记忆是现有记忆的变体或扩展，请更新现有记忆以反映新信息。
以下是任务的详细信息：
- 现有记忆：
{现有记忆}
- 新记忆：{记忆}
Mem0 点评
Mem0 是RAG的杀手?
- NO, Mem0 是RAG的助手，可以帮助提供更个性化的内容。
Mem0 有什么用处？
- Mem0可以显著提升个性化AI的能力。通过记住用户的偏好等用户画像信息，AI产品就可以提供更加个性化服务，有较好的想象空间。传统的用户画像依赖于产研去设计schema，只能挖掘存储设计好的一些特征，而
Mem0通过大模型，可以提供schame base和大模型自己挖掘的记忆，提供了一条更通用的方案 - PA产品、泛娱乐、教育等各个领域，都可以发挥出作用
Mem0 有什么不足？
- mem0当前未看到提供一些通用的schema，提供graph base的管理支持
- 用户的memory应该区分短中长期记忆，mem0当前是未区分的，需要有解决方案
- 自定义的记忆管理规则
- 支持结合RAG 和用户对话历史，初始化记忆
- 完全依赖LLM，成本较大
我们也可以看下mem0的roadmap，有规划提供一些自定义规则支持：
- Integration with various LLM providers
- Support for LLM frameworks
- Integration with AI Agents frameworks
- Customizable memory creation/update rules
- Hosted platform support