AI 助理
文档备案控制台
官方文档
首页

Python SDK 使用介绍

更新时间：
复制 MD 格式
产品详情
我的收藏

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

安装

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

pip install "tablestore>=6.4.7"

初始化客户端

使用实例 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：返回条数上限。

列出 Scope

通过 list_memory_store_scopes 列出记忆库中已存在的 Scope 组合，agentId 和 runId 支持通配符 *。

scopes = client.list_memory_store_scopes({
    "memoryStoreName": "agent_memory",
    "scope": {"appId": "app-001", "tenantId": "*", "agentId": "*", "runId": "*"},
    "limit": 100,
})

跟踪异步任务

写入操作默认异步执行。可使用 add_memories 返回的 requestId，通过 get_memory_task 查询单个任务的状态与产出，或通过 list_memory_tasks 按 Scope 与状态批量列出任务。

task = client.get_memory_task({
    "memoryStoreName": "agent_memory",
    "requestId": "<requestId>",
})

tasks = client.list_memory_tasks({
    "memoryStoreName": "agent_memory",
    "scope": {"appId": "app-001", "tenantId": "user-001", "agentId": "assistant", "runId": "session-001"},
    "status": "completed",
    "limit": 50,
})

记忆整理（Dream）

记忆整理（Dream）对已有记忆进行去重、合并与精炼，并可基于历史提取技能与画像。典型流程为：创建任务、轮询状态、查看动作提案、应用选定动作。

步骤 1：调用 create_memory_dream_task 创建整理任务，返回 dreamId。

created = client.create_memory_dream_task({
    "memoryStoreName": "agent_memory",
    "scopes": [{"appId": "app-001", "tenantId": "user-001"}],
    "taskType": "memory",
    "applyMode": "proposal",
})
dream_id = created["dreamId"]

步骤 2：轮询 get_memory_dream_task，直到任务状态为 completed。

task = client.get_memory_dream_task({
    "memoryStoreName": "agent_memory",
    "dreamId": dream_id,
})

步骤 3：调用 list_memory_dream_actions 查看整理任务生成的动作提案。

actions = client.list_memory_dream_actions({
    "memoryStoreName": "agent_memory",
    "dreamId": dream_id,
    "limit": 20,
})

步骤 4：调用 apply_memory_dream_actions 应用选定的动作。

client.apply_memory_dream_actions({
    "memoryStoreName": "agent_memory",
    "dreamId": dream_id,
    "actionIds": [a["actionId"] for a in actions["actions"]],
})

此外，可通过 list_memory_dream_tasks 列出记忆库下的整理任务，通过 cancel_memory_dream_task 取消尚未完成的任务。

异步客户端

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())
上一篇：SDK 参考下一篇：Node.js SDK 使用介绍