記憶儲存 Python SDK 使用介紹 - Tablestore

Table Store 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())