Tablestore の Python SDK では、同期呼び出し用の OTSClient と、非同期呼び出し用の AsyncOTSClient を提供します。これらを使用して、メモリストアの作成、記憶の書き込みと検索、監査ログのクエリを実行できます。
インストール
pip を使用して Python SDK をインストールします。バージョン 6.4.7 以降が必要です。
pip install "tablestore>=6.4.7"クライアントの初期化
インスタンスのエンドポイント、アクセスキーペア、インスタンス名を使用して 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：Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID です。
AccessKey Secret：Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret です。
Instance name：Tablestore インスタンスの名前です。
メモリストアの作成
create_memory_store を呼び出して、エージェントの長期記憶用のメモリストアを作成します。
client.create_memory_store({
"memoryStoreName": "agent_memory",
"description": "Agent long-term memory store",
})作成後、セカンダリインデックスは非同期で初期化されます。書き込みまたは検索を行う前に、初期化が完了するまで待機してください。
記憶の書き込み
会話メッセージのリスト (messages) または前処理済みテキスト (text) を渡します。いずれも appId、tenantId、agentId、runId が必要です。
会話メッセージの書き込み
messages フィールドに、ユーザーとエージェント間のマルチターン会話を渡します。サービスはバックグラウンドで長期記憶を抽出します。
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,
})パラメータ：
memoryStoreName：対象メモリストアの名前です。scope：記憶が属するスコープです。4 つのフィールドはすべて必須です。messages：会話メッセージのリストです。各メッセージにはroleとcontentが含まれます。metadata：業務上のタグ付けに使用するカスタムメタデータです。sync：長期記憶を同期的に抽出するかどうか。デフォルトはFalseです。Falseの場合、API は即座にレスポンスを返し、バックグラウンドで長期記憶を抽出します。
テキストの書き込み
前処理済みテキストを記憶として追加します。
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_memories を呼び出して、指定したスコープ内でベクター検索を実行します。agentId と runId フィールドにワイルドカード * を使用すると、エージェントまたはセッションをまたいで記憶を取得できます。
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"])パラメータ：
query：検索テキストです。topK：返す結果の最大数です。デフォルトは10です。有効な値：1 ～ 50。enableRerank：再ランキングを有効にするかどうか。デフォルトはTrueです。
各結果には、一致した記憶ユニット (unit) と関連性スコア (score) が含まれます。unit 内のフィールドはスネークケース命名を使用します。フィールドの完全なリファレンスについては、Memory Storage API reference をご参照ください。
長期記憶の管理
長期記憶の一覧表示、取得、更新、削除を行います。単一の記憶を取得、更新、または削除するには、4 つの scope フィールドがすべて必須で、ワイルドカード (*) は使用できません。
記憶の一覧
指定したスコープに一致する長期記憶を一覧表示します。ワイルドカード * は、tenantId、agentId、および runId フィールドでサポートされています。
result = client.list_memories({
"memoryStoreName": "agent_memory",
"scope": {
"appId": "app-001",
"tenantId": "*",
"agentId": "*",
"runId": "*",
},
"limit": 20,
})記憶の取得
memoryId により単一の長期記憶を取得します。4 つの scope フィールドはすべて必須です。
memory = client.get_memory({
"memoryStoreName": "agent_memory",
"memoryId": "mem-001",
"scope": {
"appId": "app-001",
"tenantId": "user-001",
"agentId": "assistant",
"runId": "session-001",
},
})記憶の更新
単一の長期記憶のテキストまたはメタデータを更新します。4 つの scope フィールドはすべて必須です。
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"
},
})記憶の削除
単一の長期記憶を削除します。4 つの scope フィールドはすべて必須です。
client.delete_memory({
"memoryStoreName": "agent_memory",
"memoryId": "mem-001",
"scope": {
"appId": "app-001",
"tenantId": "user-001",
"agentId": "assistant",
"runId": "session-001",
},
})短期記憶のクエリ
短期記憶は、生の会話メッセージです。短期記憶をクエリするには、4 つの scope フィールド (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 を呼び出して、メモリストアの監査ログをクエリします。スコープと操作タイプ (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：返すレコードの最大数です。
スコープの一覧
list_memory_store_scopes を呼び出して、メモリストア内にすでに存在するスコープの組み合わせを一覧表示します。ワイルドカード * は、tenantId、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 を使用して、スコープとステータスでタスクを一括で一覧表示できます。
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： タスクのステータスが completed になるまで、get_memory_dream_task をポーリングします。
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 と同一のメソッドシグネチャおよびパラメータを持つ非同期 API を提供します。すべての呼び出しで await を使用してください。Web サービスやエージェントアプリケーションなどの高同時実行シナリオに推奨します。
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())