Python SDK-表格存储 Tablestore(Tablestore)-阿里云帮助中心

Tablestore Agent Storage SDK 面向 AI Agent 存储场景，统一封装记忆库、文件记忆和知识库接口，提供 Python 与 TypeScript 版本，并支持通过 API Key 简化服务接口认证。本文介绍 Python SDK 的认证与各能力使用示例。

前提条件

已开通表格存储服务并创建实例，获取实例的访问地址（endpoint）与实例名。
若使用 API Key 认证：已在目标实例上创建 API Key，并为 API Key 绑定的 RAM 子账号授予 ots:CallWithBearerToken 权限。API Key 的接口权限继承该子账号的 RAM 策略。
若使用知识库能力：已开通阿里云对象存储 OSS 并创建 Bucket，用于存储原始文档。上传文档时 SDK 需要 OSS 访问凭证（AccessKey），初始化客户端时需额外传入 oss_endpoint 和 oss_bucket 参数。

初始化客户端

推荐使用 API Key 认证；如需知识库上传或其他表格存储接口，需同时提供 AccessKey 与 OSS 配置。

使用 API Key 访问（推荐）

通过 api_key 参数和实例信息初始化客户端，端点必须为 HTTPS。

from tablestore_agent_storage import AgentStorageClient

client = AgentStorageClient(
    api_key="<your-api-key>",
    ots_endpoint="https://<endpoint>.ots.aliyuncs.com",
    ots_instance_name="<instance-name>",
)

resp = client.list_memory_stores({})
for store in resp["stores"]:
    print(store["memoryStoreName"])

使用 AccessKey 访问

访问表格存储完整接口或上传知识库文档（涉及 OSS）时，使用 AccessKey 认证。

from tablestore_agent_storage import AgentStorageClient

client = AgentStorageClient(
    access_key_id="<AccessKey ID>",
    access_key_secret="<AccessKey Secret>",
    ots_endpoint="https://<endpoint>.ots.aliyuncs.com",
    ots_instance_name="<instance-name>",
    # 若使用知识库能力，需同时配置 OSS
    oss_endpoint="https://oss-<region>.aliyuncs.com",
    oss_bucket_name="<bucket-name>",
)

知识库使用示例

知识库（Knowledge Base）支持文档自动切片、向量化与混合检索，适合企业问答、RAG 等场景。文档源文件存储在 OSS，切片索引与检索由表格存储完成，使用前需同时配置 OSS 与表格存储。

创建知识库时如果不显式指定向量模型，SDK 使用阿里云百炼 text-embedding-v4 模型（1024 维），默认检索方式为向量检索与全文检索的混合检索。

from tablestore_agent_storage import AgentStorageClient

client = AgentStorageClient(
    access_key_id="<AccessKey ID>",
    access_key_secret="<AccessKey Secret>",
    oss_endpoint="https://oss-<region>.aliyuncs.com",
    oss_bucket_name="<bucket-name>",
    ots_endpoint="https://<endpoint>.ots.aliyuncs.com",
    ots_instance_name="<instance-name>",
)

# 1. 创建知识库（默认使用百炼 text-embedding-v4 向量模型，
#    默认检索方式为向量检索 + 全文检索的混合检索）
client.create_knowledge_base({
    "knowledgeBaseName": "product_kb",
    "description": "产品知识库",
})

# 2. 上传本地文档（SDK 会先将文件上传至 OSS，再触发切片与向量化）
client.upload_documents({
    "knowledgeBaseName": "product_kb",
    "documents": [
        {
            "documentId": "doc-guide-001",
            "filePath": "./docs/product-overview.txt",
        },
        {
            "documentId": "doc-guide-002",
            "filePath": "./docs/sdk-intro.txt",
        },
    ],
})

# 3. 列出知识库中的文档
resp = client.list_documents({
    "knowledgeBaseName": "product_kb",
    "maxResults": 10,
})
for doc in resp["data"]["documentDetails"]:
    print(doc.get("documentId"), doc.get("ossKey"), doc.get("status"))

# 4. 检索
from tablestore_agent_storage.models import RetrieveRequest, RetrievalQuery

req = RetrieveRequest(
    knowledge_base_name="product_kb",
    retrieval_query=RetrievalQuery(text="Agent Storage SDK 有什么用"),
)
result = client.retrieve(req)
for hit in result["data"]["retrievalResults"]:
    print(hit)

# 5. 删除文档
client.delete_documents({
    "knowledgeBaseName": "product_kb",
    "documents": [{"documentId": "doc-guide-001"}],
})

# 6. 删除知识库
client.delete_knowledge_base({"knowledgeBaseName": "product_kb"})

上传文档后，切片与向量化异步执行，通常数十秒到数分钟完成。可通过 list_documents查询文档 status 字段确认索引进度。

记忆库使用示例

记忆库（Memory Store）用于持久化 AI Agent 的多轮对话记忆，并支持基于向量的语义检索。SDK 方法与 REST API 一一对应，Python 使用 snake_case，入参与接口一致。

# 1. 创建记忆库
client.create_memory_store({
    "memoryStoreName": "agent_memory",
    "description": "Agent 长期记忆库",
    "extractInstructions": "重点关注用户的饮食偏好与出行习惯",
})

# 2. 写入记忆
client.add_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "messages": [
        {"role": "user", "content": "我喜欢喝美式咖啡"},
    ],
    "sync": True,
})

# 3. 检索记忆
result = client.search_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "*",
        "runId": "*",
    },
    "query": "用户喜欢什么饮品",
    "topK": 5,
    "includeEvidence": True,
    "minSimilarity": 0.3,
})
for hit in result["data"]["memories"]:
    print(hit["content"], hit["similarity"])

# 4. 列出记忆库中的 Scope
scopes = client.list_memory_store_scopes({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "*",
        "agentId": "*",
        "runId": "*",
    },
})
for scope in scopes["scopes"]:
    print(scope)

# 5. 列出记忆库
resp = client.list_memory_stores({})
for store in resp["stores"]:
    print(store["memoryStoreName"])

除上述示例外，SDK 还提供更新和删除记忆库、获取和更新单条记忆、查询异步任务以及记忆整理（Dream）等方法。完整接口清单与参数说明详见记忆存储 API 接口介绍。

文件记忆使用示例

文件记忆使用文件路径组织长期上下文，适合保存用户画像、会话摘要和 Agent 工作文件。SDK 会自动补充 Item 接口所需的 memoryfile 类型，调用时只需指定记忆库、Scope、路径和内容。

能力	方法
新增文件	`add_item`
列出文件	`list_items`
读取文件或元数据	`get_item`
更新内容或重命名	`update_item`
删除文件	`delete_item`
列出历史版本	`list_item_versions`
读取历史版本	`get_item_version`
脱敏历史版本	`redact_item_version`

所有文件操作都要求传入完整的四级 Scope，即 appId、tenantId、agentId 和 runId，且不支持通配符 *。

警告

历史版本脱敏不可逆，只清除指定历史版本的路径、内容、摘要和大小，不修改当前文件。

以下示例假设已完成客户端初始化。

scope = {
    "appId": "app-001",
    "tenantId": "user-001",
    "agentId": "assistant",
    "runId": "session-001",
}

client.create_memory_store({
    "memoryStoreName": "agent_files",
    "storageMode": "filemem",
})

created = client.add_item({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "path": "/profile/preferences.md",
    "content": "# 用户偏好\n\n- 喜欢美式咖啡\n",
})

page = client.list_items({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "pathPrefix": "/profile/",
})

current = client.get_item({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "path": "/profile/preferences.md",
    "includeContent": False,
})

updated = client.update_item({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "path": "/profile/preferences.md",
    "content": "# 用户偏好\n\n- 喜欢拿铁\n",
    "expectedSha256": current["contentSha256"],
})

renamed = client.update_item({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "path": "/profile/preferences.md",
    "newPath": "/profile/user-preferences.md",
    "expectedSha256": updated["contentSha256"],
})

versions = client.list_item_versions({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "itemId": created["itemId"],
    "limit": 20,
})
version = versions["versions"][0]

snapshot = client.get_item_version({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "itemId": version["itemId"],
    "versionId": version["versionId"],
    "versionSeq": version["versionSeq"],
})

client.redact_item_version({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "itemId": version["itemId"],
    "versionId": version["versionId"],
    "versionSeq": version["versionSeq"],
    "sessionId": "privacy-job-001",
})

client.delete_item({
    "memoryStoreName": "agent_files",
    "scope": scope,
    "path": "/profile/user-preferences.md",
    "expectedSha256": renamed["contentSha256"],
})

print(page["items"], snapshot.get("content"))

expectedSha256 是可选的并发保护参数。多个调用方可能修改同一路径时，建议传入最近一次读取到的 contentSha256；摘要不匹配时，应重新读取文件，再决定是否重试。

文件视图使用示例

创建 file+ots 模式的记忆库后，可以继续通过结构化记忆接口写入、检索和更新记忆，并通过 Item 接口读取服务生成的文件视图。

client.create_memory_store({
    "memoryStoreName": "agent_memory",
    "storageMode": "file+ots",
})

使用完整 Scope 列出文件，并按路径读取文件内容。

page = client.list_items({
    "memoryStoreName": "agent_memory",
    "scope": scope,
})

for entry in page["items"]:
    item = client.get_item({
        "memoryStoreName": "agent_memory",
        "scope": scope,
        "path": entry["path"],
    })
    print(item["path"], item["content"])

重要

当 list_items 返回 readOnly: true 时，文件视图为只读。应通过 add_memories、update_memory 和 delete_memory 等结构化记忆接口维护源记忆；对文件视图调用写入方法会返回 READ_ONLY_STORE。