Tablestore:Python

Prerequisites

• Tablestore is activated and an instance is created. Obtain the instance endpoint and instance name.

• If you authenticate with an API Key: Create an API Key on the target instance and grant the ots:CallWithBearerToken permission to the RAM sub-account bound to the API Key. The API Key inherits the RAM policies of the bound sub-account.

• If you use the knowledge base capability: Activate Alibaba Cloud Object Storage (OSS) and create a bucket to store source documents. When you upload documents, the SDK requires OSS access credentials (AccessKey). Pass the oss_endpoint and oss_bucket parameters when you initialize the client.

Initialize the client

API Key authentication is recommended. If you need to upload documents to a knowledge base or call other Tablestore interfaces, provide both AccessKey and OSS configurations.

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"])

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 usage examples

A knowledge base supports automatic document chunking, vectorization, and hybrid retrieval, making it suitable for enterprise Q&A, RAG, and similar scenarios. Source documents are stored in OSS, while chunking, indexing, and retrieval are handled by Tablestore. Configure both OSS and Tablestore before you use a knowledge base.

If you do not explicitly specify an embedding model when you create a knowledge base, the SDK uses the Alibaba Cloud Bailian text-embedding-v4 model (1024 dimensions). The default retrieval method is hybrid retrieval, which combines vector retrieval and full-text retrieval.

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"})

After you upload documents, chunking and vectorization run asynchronously and typically complete within seconds to minutes. Query the list_documents method to check the status field of a document and confirm indexing progress.

Memory store usage examples

A memory store persists multi-turn conversation memory for AI agents and supports vector-based semantic retrieval. SDK methods correspond one-to-one with the REST API. Python uses snake_case, and the parameters match the API.

# 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"])

In addition to the examples above, the SDK also provides methods for updating and deleting memory stores, retrieving and updating individual memories, querying asynchronous tasks, and memory consolidation (Dream). For the complete API list and parameter details, see Memory Storage API overview.

File memory usage examples

File memory organizes long-term context by file path, making it suitable for storing user profiles, session summaries, and agent working files. The SDK automatically supplies the memoryfile type required by the Item interface. You only need to specify the memory store, scope, path, and content.

All file operations require a complete four-level scope: appId, tenantId, agentId, and runId. Wildcards (*) are not supported.

The following examples assume the client has been initialized.

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 is an optional concurrency control parameter. When multiple callers may modify the same path, pass the contentSha256 from the most recent read. If the digest does not match, re-read the file before you decide whether to retry.

File view usage examples

After you create a memory store in file+ots mode, you can continue to write, retrieve, and update memories through the structured memory interface, and read the server-generated file views through the Item interface.

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

List files with a complete scope and read file content by path.

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"])