记忆存储 Python SDK 使用介绍 - 表格存储

表格存储 Python SDK 已集成记忆存储服务接口，通过 OTSClient 调用同步接口，通过 AsyncOTSClient 调用异步接口。

安装

通过 pip 安装 Python SDK，要求版本不低于 6.4.5。

pip install "tablestore>=6.4.5"

初始化客户端

使用实例 Endpoint、AccessKey 和实例名初始化同步客户端 OTSClient。

from tablestore import OTSClient

client = OTSClient(
    "https://<instance>.cn-beijing.ots.aliyuncs.com",
    "<AccessKey ID>",
    "<AccessKey Secret>",
    "<instance-name>",
)

参数说明：

Endpoint：实例的访问地址，格式为 https://<instance>.<region>.ots.aliyuncs.com。
AccessKey ID：阿里云账号或 RAM 用户的 AccessKey ID。
AccessKey Secret：阿里云账号或 RAM 用户的 AccessKey Secret。
实例名：Tablestore 实例的名称。

创建记忆库

调用 create_memory_store 创建一个用于存储 Agent 长期记忆的记忆库。

client.create_memory_store({
    "memoryStoreName": "agent_memory",
    "description": "Agent 长期记忆库",
})

记忆库创建后会异步初始化二级索引，需等待约 1 分钟索引就绪后才能正常写入和检索。

写入记忆

写入记忆支持两种入参形式：传入对话消息列表（messages），或传入预加工的文本（text）。两种形式均需指定完整 Scope，包含 appId、tenantId、agentId、runId 四段。

写入对话消息

通过 messages 字段传入用户与 Agent 的多轮对话，服务端在后台抽取长期记忆。

client.add_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "messages": [
        {"role": "user", "content": "我喜欢喝咖啡"},
        {"role": "assistant", "content": "好的，我记住了"},
    ],
    "metadata": {
        "source": "chat"
    },
    "sync": True,
})

参数说明：

memoryStoreName：目标记忆库名称。
scope：记忆归属范围，4 段全部必填。
messages：对话消息列表，每条消息包含 role 与 content。
metadata：自定义元数据，用于业务标记。
sync：是否同步抽取长期记忆。默认值为 False，异步写入时接口先保存原始消息并立即返回，长期记忆抽取在后台完成。

写入文本

将一段文本添加为记忆。

client.add_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "text": "用户喜欢喝咖啡，偏好简洁的回答风格",
})

检索长期记忆

通过 search_memories 在指定 Scope 范围内进行向量检索。检索时可在 agentId 与 runId 两段使用通配符 *，跨 Agent、跨会话召回相关记忆。

result = client.search_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "*",
        "runId": "*",
    },
    "query": "用户喜欢什么饮品",
    "topK": 5,
    "enableRerank": True,
})

for item in result.get("results", []):
    unit = item["unit"]
    print(unit["id"], unit["text"], item["score"])

参数说明：

query：检索文本。
topK：返回结果数量上限。默认值为 10，取值范围为 1~50。
enableRerank：是否启用 Rerank 重排序。默认值为 True。

响应中 results 列表的每个元素包含命中的记忆单元 unit 与得分 score。unit 内部字段使用 snake_case 命名，详见 API 接口介绍。

管理长期记忆

支持列出、获取、更新和删除长期记忆。获取、更新、删除 单条记忆时，scope 4 段全部必填，不允许使用通配符 *。

列出记忆

按 Scope 列出符合条件的长期记忆。agentId 与 runId 支持通配符 *。

result = client.list_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "*",
        "agentId": "*",
        "runId": "*",
    },
    "limit": 20,
})

获取记忆

按 memoryId 获取单条长期记忆，必须传入完整 Scope（4 段全部必填）。

memory = client.get_memory({
    "memoryStoreName": "agent_memory",
    "memoryId": "mem-001",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
})

更新记忆

更新单条长期记忆的文本或元数据，必须传入完整 Scope（4 段全部必填）。

client.update_memory({
    "memoryStoreName": "agent_memory",
    "memoryId": "mem-001",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "text": "用户偏好咖啡，且喜欢简洁回答",
    "metadata": {
        "source": "manual"
    },
})

删除记忆

删除单条长期记忆，必须传入完整 Scope（4 段全部必填）。

client.delete_memory({
    "memoryStoreName": "agent_memory",
    "memoryId": "mem-001",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
})

查询短期记忆

短期记忆即原始会话消息。查询短期记忆需要传入 appId、tenantId、agentId、runId 四段，全部必填且不允许使用通配符 *。

messages = client.list_memory_store_messages({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "limit": 100,
})

查询请求审计

通过 list_memory_store_requests 查询记忆库的接口调用审计日志，可按 Scope、操作类型（operation）等维度过滤，便于排查问题与监控调用量。

requests = client.list_memory_store_requests({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "*",
        "agentId": "*",
        "runId": "*",
    },
    "operation": "AddMemories",
    "limit": 50,
})

参数说明：

operation：操作类型，例如 AddMemories、SearchMemories 等。
limit：返回条数上限。

异步客户端

AsyncOTSClient 提供异步接口，方法签名与入参与 OTSClient 一致，所有调用通过 await 执行。建议在 Web 服务、Agent 应用等高并发场景中使用。

import asyncio
from tablestore import AsyncOTSClient

async def main():
    async with AsyncOTSClient(
        "https://<instance>.cn-beijing.ots.aliyuncs.com",
        "<AccessKey ID>",
        "<AccessKey Secret>",
        "<instance-name>",
    ) as client:
        await client.add_memories({
            "memoryStoreName": "agent_memory",
            "scope": {
                "appId": "app-001",
                "tenantId": "user-001",
                "agentId": "assistant",
                "runId": "session-001",
            },
            "text": "用户偏好简洁的回答风格",
        })

        result = await client.search_memories({
            "memoryStoreName": "agent_memory",
            "scope": {
                "appId": "app-001",
                "tenantId": "user-001",
                "agentId": "*",
                "runId": "*",
            },
            "query": "用户偏好什么回答风格",
            "topK": 5,
        })
        print(result)

asyncio.run(main())