记忆模块 SDK 接入-云监控(CMS)-阿里云帮助中心

在 Python 应用中直接集成 AgentLoop 记忆库，通过 SDK 实现记忆的添加、检索和管理。

前置条件

拥有阿里云账号，并已开通 AgentLoop 记忆库服务。
已在控制台创建 Workspace 和记忆库。
获取阿里云账号 AccessKey ID / Secret。

步骤一：安装 SDK

pip install agentloop-memory

步骤二：配置环境变量

export ALIBABA_CLOUD_ACCESS_KEY_ID=your_access_key_id
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=your_access_key_secret
export CMS_ENDPOINT=cms.cn-hangzhou.aliyuncs.com
export CMS_WORKSPACE=your_workspace_name
export CMS_MEMORY_STORE=your_memory_store_name

步骤三：初始化 Client

import os
from agentloop_memory import Config
from agentloop_memory.client import AgentLoopMemoryClient

config = Config(
    access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
    access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
    endpoint=os.getenv("CMS_ENDPOINT"),
)

client = AgentLoopMemoryClient(
    config,
    workspace=os.getenv("CMS_WORKSPACE"),
    memory_store=os.getenv("CMS_MEMORY_STORE"),
)

步骤四：核心操作

添加记忆

# 添加单条文本
result = client.add(
    messages="我住在杭州，喜欢 Python 和云原生技术",
    user_id="user123",
)

# 添加多轮对话
result = client.add(
    messages=[
        {"role": "user", "content": "我的项目使用 FastAPI 框架"},
        {"role": "assistant", "content": "好的，我记住了您使用 FastAPI"}
    ],
    user_id="user123",
)

add方法定义

def add(
    self,
    messages: Union[str, Dict[str, str], List[Dict[str, str]]],
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    app_id: Optional[str] = None,
    run_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    infer: bool = True,
    async_mode: bool = True,
)

messages: 被添加到记忆库的对话记录。

str: 我住在杭州，喜欢杭州的风景，经常会去西湖玩
单条对话: {"role": "user", "content": "我住在杭州，喜欢杭州的风景，经常会去西湖玩"}
多轮对话：[{"role": "user", "content": "我住在杭州，喜欢杭州的风景，经常会去西湖玩"},
         {"role": "assistant", "content": "杭州是个好地方..."}]

user_id, agent_id, app_id, run_id: 单条对话的标识。在 search, get_all, delete_all 中可以用来进行过滤。最常见使用 user_id 来区分不同用户的对话记忆。
metadata: 用户自定义标签。
infer: 默认为 True。如果为 False，将不进行记忆提取，而是直接将文本写入到记忆中。
async_mode: 后台异步提取记忆，当前只支持异步提取。

语义检索

result = client.search(
    query="用户的技术栈偏好",
    user_id="user123",
    top_k=3,
    rerank=True,  # 启用 Rerank 精排
)
print(result)
# {'results': [{'id': '...', 'memory': 'The user likes Python...', 'score': 0.89}]}

search方法定义

def search(
    self,
    query: str,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    app_id: Optional[str] = None,
    run_id: Optional[str] = None,
    top_k: Optional[int] = None,
    rerank: bool = False,
)

query: 被用于记忆检索的文本。
user_id, agent_id, app_id, run_id: 至少设置一个，限定记忆检索范围。
top_k: 返回前 k 个最相似的文本。
rerank: 是否调用 rerank 模型进行重排序。

返回结果

{'results': [{'hash': 'b1dde187c55624254f1ab8eca53755d6', 'id': '019c269b-3969-7232-bf18-052e8cb29433', 'memory': 'The user lives in Hangzhou.', 'score': 0.5933572760058075}]}

获取所有记忆

result = client.get_all(user_id="user123")
print(result)
# {'results': [{'id': '...', 'memory': '...'}, ...]}

get_all方法定义

def get_all(
    self,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    app_id: Optional[str] = None,
    run_id: Optional[str] = None,
)

user_id, agent_id, app_id, run_id: 至少设置一个，限定记忆检索范围。

返回结果

{'results': [{'hash': 'b1dde187c55624254f1ab8eca53755d6', 'id': '019c269b-3969-7232-bf18-052e8cb29433', 'memory': 'The user lives in Hangzhou.'}, {'hash': '94fe0a7513144efac1c31a84cdafb544', 'id': '019c269b-3ac0-7472-8adb-c41fc9b29a10', 'memory': 'The user likes the scenery of Hangzhou.'}, {'hash': 'da45e1d91bbabe7e3870683d29492c3e', 'id': '019c269b-3bbe-7f31-81f6-e5d7198125a3', 'memory': 'The user frequently visits West Lake.'}]}

删除记忆

# 删除单条
client.delete(memory_id="019c269b-3969-7232-bf18-052e8cb29433")

# 删除用户所有记忆
client.delete_all(user_id="user123")

方法定义

delete

def delete(self, memory_id: str)

memory_id: 被删除的记忆 id。

delete_all

def delete_all(
    self,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    app_id: Optional[str] = None,
    run_id: Optional[str] = None
)

user_id, agent_id, app_id, run_id: 至少设置一个，限定记忆删除范围。

完整示例

import os
from agentloop_memory import Config
from agentloop_memory.client import AgentLoopMemoryClient

# 初始化
config = Config(
    access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
    access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
    endpoint=os.getenv("CMS_ENDPOINT", "cms.cn-hangzhou.aliyuncs.com"),
)
client = AgentLoopMemoryClient(
    config,
    workspace=os.getenv("CMS_WORKSPACE"),
    memory_store=os.getenv("CMS_MEMORY_STORE"),
)

# 添加记忆
client.add(
    messages="我是一名后端开发，主要使用 Python 和 Go",
    user_id="demo_user",
)

# 等待异步处理完成（生产环境无需等待）
import time
time.sleep(3)

# 检索记忆
result = client.search(
    query="用户的技术背景",
    user_id="demo_user",
    top_k=3,
)
print("检索结果:", result)

# 查看所有记忆
all_memories = client.get_all(user_id="demo_user")
print("全部记忆:", all_memories)

验证

运行上述代码后，看到类似输出表示成功：

检索结果: {'results': [{'id': '...', 'memory': 'The user is a backend developer who mainly uses Python and Go.', 'score': 0.85}]}
全部记忆: {'results': [{'id': '...', 'memory': 'The user is a backend developer who mainly uses Python and Go.'}]}