文档中心

记忆库 Mem0

快速入门

通过 SDK 连接记忆库 Mem0

记忆库 Mem0 基于开源 Mem0 提供了完整的 Python SDK 示例代码，安装 SDK 后，即可通过 SDK 请求记忆项目服务，并为目标记忆项目快速添加、修改和查询记忆。

前提条件

已成功创建 Mem0 记忆项目。具体操作步骤，请参见创建记忆项目。

已获取 Mem0 记忆项目的连接地址。具体操作步骤，请参见查看连接地址。

已获取 Mem0 记忆项目的 API Key。具体操作步骤，请参见查看 API key。

说明

记忆项目创建成功后，记忆库 Mem 0 会提供一个默认的 API Key 用于请求 Mem0 服务，您也可以根据业务需要创建 API Key。更多详情，请参见创建 API Key。

已在 ECS 实例或本地设备上安装 Python 3.10 或以上版本。

说明

您可以通过 python --version 命令检查当前 Python 的版本。

操作步骤

安装 Python SDK

参考如下操作步骤安装 Python SDK：

登录已安装了 Python 工具的设备，例如 ECS 实例或本地设备。

打开本地设备的命令行窗口或命令终端工具，执行如下命令安装 0.1.118 版本的 Python SDK。

pip install mem0ai==0.1.118

创建 Python 脚本文件

在命令行窗口，执行如下命令将示例代码保存为 Python 脚本文件。

说明

请根据您的实际情况，修改示例代码中的如下参数：

<Python file name>：Python 文件名称。本文示例中使用的 Python 文件名为 test_mem0.py。

api_key="<Your API Key value>"：将 <Your API Key value> 替换为您 Mem0 记忆项目中的真实 API Key。您可以在 Mem0 控制台上查看 API Key，具体操作步骤，请参见查看 API key。

host="<Your Mem0 connection address>:8000"：将 <Your Mem0 connection address> 替换为您 Mem0 记忆项目中的真实连接地址。您可以在 Mem0 控制台上查看连接地址，具体操作步骤，请参见查看连接地址。

cat > <Python file name> << 'EOF'
import os
from mem0 import MemoryClient
import json
import time
import requests

def check_job_status(host: str, api_key: str, event_id: str):
# 轮询任务状态直到 SUCCEEDED
job_status = "PENDING"
print(f"开始轮询任务状态，event_id: {event_id}")
while job_status != "SUCCEEDED":
try:
# 发送请求查询任务状态
response = requests.get(
f"{host}/v1/job/{event_id}/",
headers={"Authorization": f"Token {api_key}"}
)
if response.status_code == 200:
job_info = response.json()
job_status = job_info["status"]
print(f"当前任务状态: {job_status}")
# 如果任务已经完成，打印完整信息
if job_status == "SUCCEEDED":
print(f"任务已完成，完整信息: {json.dumps(job_info, ensure_ascii=False)}")
else:
print(f"查询任务状态失败，状态码: {response.status_code}")
# 每秒轮询一次
time.sleep(1)
except Exception as e:
print(f"轮询过程中发生错误: {str(e)}")
time.sleep(1)
print(f"任务{event_id}处理完成")

# 记忆库 Mem0 配置信息
api_key="<Your API Key value>"
host="<Your Mem0 connection address>:8000"
m = MemoryClient(host=host,api_key=api_key)

user_id="xiaochen1"
messages = [
{"role": "assistant", "content": "你好呀！能先告诉我你叫什么名字吗？"},
{"role": "user", "content": "我叫李明。今年30岁了。"},
{"role": "assistant", "content": "很高兴认识你，李明！看您资料是来自北京对吧？为了能更好地为您服务，想了解一下您的饮食偏好，比如您吃辣吗？"},
{"role": "user", "content": "是休闲旅游。我穿衣嘛，平时基本都穿黑色或灰色的衣服，比较喜欢简约休闲的风格，舒服最重要。"}
]

print("===================异步添加记忆===============================")
ret = m.add(messages, user_id=user_id, async_mode=True)
print(json.dumps(ret, ensure_ascii=False))
check_job_status(host, api_key, event_id=ret["results"][0]["event_id"])

time.sleep(5)

print("===================获取全部记忆===============================")
all_memories = m.get_all(user_id=user_id)
print(json.dumps(all_memories, ensure_ascii=False))

print("===================根据 query 召回记忆===============================")
query = "用户的工作信息"
relevant_memories = m.search(query, user_id=user_id)
print(json.dumps(relevant_memories, ensure_ascii=False))

print("===================获取每条记忆详细信息===============================")
for item in relevant_memories["results"]:
memory_id = item["id"]
response = m.get(memory_id)
print(json.dumps(response, ensure_ascii=False))

print("===================获取每条记忆历史记录===============================")
memory_id=0
for item in relevant_memories["results"]:
memory_id = item["id"]
response = m.history(memory_id)
print(json.dumps(response, ensure_ascii=False))

print("===================更新单条记忆===============================")
response = m.update(memory_id, "没有结婚，是单身")
print(json.dumps(response, ensure_ascii=False))

print("===================删除单条记忆===============================")
for item in relevant_memories["results"]:
memory_id = item["id"]
response = m.delete(memory_id)
print(json.dumps(response, ensure_ascii=False))

print("===================删除所有记忆===============================")
time.sleep(5)
response = m.delete_all(user_id=user_id)
print(json.dumps(response, ensure_ascii=False))

print("===================Demo执行完成===============================")
EOF

执行 Python 脚本文件

在命令行窗口，使用如下命令执行 Python 脚本。

python <Python file name>

命令示例如下。

python test_mem0.py

最近更新时间：2026.01.13 14:33:37

这个页面对您有帮助吗？

有用

无用