Tablestore:Python

Prérequis

• Activez Tablestore et créez une instance. Récupérez le point de terminaison et le nom de l'instance.

• Pour l'authentification par clé API : créez une clé API sur l'instance cible et accordez l'autorisation ots:CallWithBearerToken au sous-compte RAM associé à cette clé. La clé API hérite des stratégies RAM du sous-compte lié.

• Pour utiliser la fonctionnalité de base de connaissances : activez Alibaba Cloud Object Storage (OSS) et créez un bucket pour stocker les documents sources. Lors du téléchargement des documents, le SDK requiert les identifiants d'accès OSS (AccessKey). Spécifiez les paramètres oss_endpoint et oss_bucket lors de l'initialisation du client.

Initialiser le client

L'authentification par clé API est recommandée. Si vous devez télécharger des documents dans une base de connaissances ou appeler d'autres interfaces Tablestore, fournissez à la fois les configurations AccessKey et OSS.

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>",
    # Also configure OSS when using knowledge base capabilities
    oss_endpoint="https://oss-<region>.aliyuncs.com",
    oss_bucket_name="<bucket-name>",
)

Exemples d'utilisation de la base de connaissances

Une base de connaissances prend en charge le découpage automatique des documents, leur vectorisation et la récupération hybride. Elle convient parfaitement aux scénarios de questions-réponses en entreprise, aux architectures RAG et autres cas similaires. Les documents sources sont stockés dans OSS, tandis que le découpage, l'indexation et la récupération sont gérés par Tablestore. Configurez à la fois OSS et Tablestore avant d'utiliser une base de connaissances.

Si vous ne spécifiez pas explicitement de modèle d'embedding lors de la création d'une base de connaissances, le SDK utilise par défaut le modèle Alibaba Cloud Bailian text-embedding-v4 (1 024 dimensions). La méthode de récupération par défaut est la récupération hybride, qui combine la recherche vectorielle et la recherche en texte intégral.

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. Create a knowledge base (uses the Bailian text-embedding-v4 embedding model by default,
#    with hybrid retrieval that combines vector and full-text retrieval)
client.create_knowledge_base({
    "knowledgeBaseName": "product_kb",
    "description": "Knowledge base for product information",
})

# 2. Upload local documents (the SDK uploads files to OSS, then triggers chunking and vectorization)
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. List documents in the knowledge base
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. Retrieve information
from tablestore_agent_storage.models import RetrieveRequest, RetrievalQuery

req = RetrieveRequest(
    knowledge_base_name="product_kb",
    retrieval_query=RetrievalQuery(text="What is the Agent Storage SDK used for?"),
)
result = client.retrieve(req)
for hit in result["data"]["retrievalResults"]:
    print(hit)

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

# 6. Delete the knowledge base
client.delete_knowledge_base({"knowledgeBaseName": "product_kb"})

Après le téléchargement des documents, le découpage et la vectorisation s'exécutent de manière asynchrone et se terminent généralement en quelques secondes ou minutes. Appelez la méthode list_documents pour vérifier le champ status d'un document et confirmer la progression de l'indexation.

Exemples d'utilisation du magasin de mémoire

Un magasin de mémoire persiste la mémoire des conversations multi-tours pour les agents IA et prend en charge la récupération sémantique basée sur les vecteurs. Les méthodes du SDK correspondent directement aux API REST. En Python, la convention snake_case est utilisée et les paramètres correspondent à ceux de l'API.

# 1. Create a memory store
client.create_memory_store({
    "memoryStoreName": "agent_memory",
    "description": "Long-term memory store for an agent",
    "extractInstructions": "Focus on the user's dietary preferences and travel habits",
})

# 2. Write memories
client.add_memories({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "user-001",
        "agentId": "assistant",
        "runId": "session-001",
    },
    "messages": [
        {"role": "user", "content": "I like drinking Americano coffee"},
    ],
    "sync": True,
})

# 3. Search memories
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,
    "includeEvidence": True,
    "minSimilarity": 0.3,
})
for hit in result["data"]["memories"]:
    print(hit["content"], hit["similarity"])

# 4. List scopes in the memory store
scopes = client.list_memory_store_scopes({
    "memoryStoreName": "agent_memory",
    "scope": {
        "appId": "app-001",
        "tenantId": "*",
        "agentId": "*",
        "runId": "*",
    },
})
for scope in scopes["scopes"]:
    print(scope)

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

Outre les exemples ci-dessus, le SDK propose également des méthodes pour mettre à jour et supprimer des magasins de mémoire, récupérer et modifier des souvenirs individuels, interroger des tâches asynchrones et consolider la mémoire (Dream). Pour la liste complète des API et les détails des paramètres, consultez Aperçu des API de stockage de mémoire.

Exemples d'utilisation de la mémoire basée sur les fichiers

La mémoire basée sur les fichiers organise le contexte à long terme par chemin de fichier. Elle est adaptée au stockage des profils utilisateurs, des résumés de session et des fichiers de travail des agents. Le SDK fournit automatiquement le type memoryfile requis par l'interface Item. Il vous suffit de spécifier le magasin de mémoire, la portée, le chemin et le contenu.

Toutes les opérations sur les fichiers nécessitent une portée complète à quatre niveaux : appId, tenantId, agentId et runId. Les caractères génériques (*) ne sont pas pris en charge.

Les exemples suivants supposent que le client est déjà initialisé.

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": "# User preferences\n\n- Prefers Americano coffee\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": "# User preferences\n\n- Prefers latte\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"))

Le paramètre expectedSha256 est un paramètre optionnel de contrôle de concurrence. Lorsque plusieurs appelants peuvent modifier le même chemin, transmettez la valeur contentSha256 issue de la lecture la plus récente. Si le condensé ne correspond pas, relisez le fichier avant de décider de retenter l'opération.

Exemples d'utilisation de la vue fichier

Après avoir créé un magasin de mémoire en mode file+ots, vous pouvez continuer à écrire, récupérer et mettre à jour des souvenirs via l'interface de mémoire structurée, et lire les vues de fichiers générées par le serveur via l'interface Item.

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

Listez les fichiers avec une portée complète et lisez le contenu des fichiers par chemin.

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