Memory Store Python SDK - Tablestore - Alibaba Cloud Documentation Center

The Tablestore Python SDK provides OTSClient for synchronous calls and AsyncOTSClient for asynchronous calls to create memory stores, write and search memories, and query audit logs.

Install

Install the Python SDK with pip. Version 6.4.7 or later is required.

pip install "tablestore>=6.4.7"

Initialize a client

Initialize OTSClient with your instance endpoint, AccessKey pair, and instance name.

from tablestore import OTSClient

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

Parameters:

Endpoint: The access endpoint of your instance, in the format https://<instance>.<region>.ots.aliyuncs.com.
AccessKey ID: The AccessKey ID of your Alibaba Cloud account or RAM user.
AccessKey Secret: The AccessKey Secret of your Alibaba Cloud account or RAM user.
Instance name: The name of your Tablestore instance.

Create a memory store

Call create_memory_store to create a memory store for agent long-term memories.

client.create_memory_store({
    "memoryStoreName": "agent_memory",
    "description": "Agent long-term memory store",
})

After creation, the secondary index initializes asynchronously. Wait for initialization to complete before writing or searching.

Write memories

Pass a list of conversation messages (messages) or preprocessed text (text). Both require appId, tenantId, agentId, and runId.

Write conversation messages

Pass multi-turn conversations between the user and the agent in the messages field. The service extracts long-term memories in the background.

client.add_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "messages": [
        {"role": "user", "content": "I like coffee."},
        {"role": "assistant", "content": "Got it. I've noted that."},
    ],
    "metadata": {
        "source": "chat"
    },
    "sync": True,
})

Parameters:

memoryStoreName: The name of the target memory store.
scope: The scope that the memory belongs to. All four fields are required.
messages: A list of conversation messages. Each message contains role and content.
metadata: Custom metadata for business tagging.
sync: Whether to extract long-term memories synchronously. Default: False. When False, the API returns immediately and extracts long-term memories in the background.

Write text

Add preprocessed text as a memory.

client.add_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "text": "The user likes coffee and prefers concise responses.",
})

Search long-term memories

Call search_memories to run a vector search within a specified scope. Use the wildcard * in the agentId and runId fields to retrieve memories across agents or sessions.

result = client.search_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "*",
        "runId": "*",
    },
    "query": "What beverages does the user like?",
    "topK": 5,
    "enableRerank": True,
})

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

Parameters:

query: The search text.
topK: The maximum number of results to return. Default: 10. Valid values: 1 to 50.
enableRerank: Whether to enable reranking. Default: True.

Each result contains a matched memory unit (unit) and a relevance score (score). Fields in unit use snake_case naming. For the full field reference, see Memory Storage API reference.

Manage long-term memories

List, get, update, and delete long-term memories. To get, update, or delete a single memory, all four scope fields are required and wildcards (*) are not allowed.

List memories

List long-term memories that match the specified scope. The agentId and runId fields support the wildcard *.

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

Get a memory

Get a single long-term memory by memoryId. All four scope fields are required.

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

Update a memory

Update the text or metadata of a single long-term memory. All four scope fields are required.

client.update_memory({
    "memoryStoreName": "agent_memory",
    "memoryId": "mem-001",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "text": "The user prefers coffee and concise answers.",
    "metadata": {
        "source": "manual"
    },
})

Delete a memory

Delete a single long-term memory. All four scope fields are required.

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

Query short-term memories

Short-term memories are the raw conversation messages. To query short-term memories, all four scope fields (appId, tenantId, agentId, runId) are required. Wildcards (*) are not allowed.

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

Query request audit logs

Call list_memory_store_requests to query audit logs for a memory store. Filter by scope and operation type (operation) to troubleshoot issues or monitor usage.

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

Parameters:

operation: The operation type, such as AddMemories and SearchMemories.
limit: The maximum number of records to return.

List scopes

Call list_memory_store_scopes to list the scope combinations that already exist in a memory store. agentId and runId support the wildcard *.

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

Track async tasks

Write operations run asynchronously by default. Use the requestId returned by add_memories to query the status and output of a single task with get_memory_task, or list tasks in bulk by scope and status with list_memory_tasks.

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

Memory consolidation (Dream)

Dream deduplicates, merges, and refines existing memories, and can extract skills and profiles from history. The typical workflow is: create the task, poll for status, review action proposals, and apply the selected actions.

Step 1: Call create_memory_dream_task to create a consolidation task. The response contains a 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"]

Step 2: Poll get_memory_dream_task until the task status is completed.

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

Step 3: Call list_memory_dream_actions to review the action proposals generated by the task.

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

Step 4: Call apply_memory_dream_actions to apply the selected actions.

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

In addition, you can call list_memory_dream_tasks to list consolidation tasks in a memory store, and cancel_memory_dream_task to cancel tasks that have not yet completed.

Async client

AsyncOTSClient provides asynchronous APIs with the same method signatures and parameters as OTSClient. Use await for all calls. Recommended for high-concurrency scenarios such as web services and agent applications.

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": "The user prefers concise responses.",
        })

        result = await client.search_memories({
            "memoryStoreName": "agent_memory",
            "scope": {
                "appId": "app-001",
                "tenantId": "user-001",
                "agentId": "*",
                "runId": "*",
            },
            "query": "What response style does the user prefer?",
            "topK": 5,
        })
        print(result)

asyncio.run(main())