Le SDK Python xAI : opérations asynchrones — Collections, RAG & fichiers

Leçon corrigée le 2 novembre 2026 d’après le code du SDK xai-sdk 1.21.0 et l’exemple officiel de xAI : les appels await exigent le client asynchrone AsyncClient (le Client est synchrone), les objets renvoyés exposent collection_id, collections, matches et chunk_content, le statut d’indexation se compare à collections_pb2.DOCUMENT_STATUS_PROCESSED, et upload_document téléverse et rattache le fichier en un seul appel (l’ancien exemple le téléversait deux fois, dont une par une méthode inexistante). Sources : Collections via l’API · Collections Search · xai-sdk-python.

Automatiser vos workflows RAG en Python

Tout ce que vous avez appelé en curl depuis le début de ce cours existe en Python : upload de documents, gestion de collections, recherche, génération. Le SDK xAI propose deux clients, l’un synchrone (Client), l’autre asynchrone (AsyncClient), aux méthodes identiques. Le second est le bon choix pour un workflow RAG, qui passe le plus clair de son temps à attendre — un upload qui monte, un document qui s’indexe, une recherche qui revient — : l’asynchrone fait tenir ces attentes en parallèle plutôt qu’à la file.

Installation et configuration

pip install xai-sdk

Le client réclame vos deux clés : celle de l’API standard pour la recherche et la génération, celle de la Management API pour les collections.

import os
from xai_sdk import AsyncClient

client = AsyncClient(
    api_key=os.getenv("XAI_API_KEY"),
    management_api_key=os.getenv("XAI_MANAGEMENT_API_KEY"),
)

Chaque requête dispose par défaut de 27 minutes avant d’expirer, ce qui couvre l’upload d’un PDF volumineux. Le paramètre timeout, en secondes, permet de resserrer cette borne pour une application interactive.

Les opérations sur les collections

Trois appels couvrent le cycle de vie complet. La création vous rend l’identifiant que vous réutiliserez partout ensuite :

collection = await client.collections.create(name="base-documentaire-2026")
collection_id = collection.collection_id
print(f"Collection créée : {collection_id}")

Le listing sert surtout à retrouver un identifiant que vous n’avez pas persisté, ou à vérifier l’état d’un compte partagé entre plusieurs projets :

reponse = await client.collections.list()
for col in reponse.collections:
    print(f"{col.collection_id} — {col.collection_name} ({col.documents_count} documents)")

Et la suppression, irréversible, emporte les documents et leurs index :

await client.collections.delete(collection_id)

Ajouter un document et attendre son indexation

Un seul appel, upload_document, téléverse le fichier par la Files API et le rattache à la collection :

from pathlib import Path


async def ajouter_document(client, collection_id, chemin_fichier):
    chemin = Path(chemin_fichier)
    document = await client.collections.upload_document(
        collection_id=collection_id,
        name=chemin.name,
        data=chemin.read_bytes(),
    )
    return document.file_metadata.file_id

Si le fichier se trouve déjà dans la Files API, add_existing_document(collection_id, file_id) le rattache sans le téléverser de nouveau.

Un document ajouté n’est pas pour autant interrogeable. L’indexation prend un temps variable selon la taille du fichier, et une recherche lancée trop tôt renverra simplement moins de résultats que prévu — un bug particulièrement pénible à diagnostiquer, puisqu’il ne lève aucune erreur. D’où cette boucle d’attente, qui interroge le statut jusqu’à DOCUMENT_STATUS_PROCESSED :

import asyncio

from xai_sdk.proto import collections_pb2


async def attendre_indexation(client, file_id, collection_id):
    while True:
        doc = await client.collections.get_document(file_id, collection_id)
        if doc.status == collections_pb2.DOCUMENT_STATUS_PROCESSED:
            print("Document prêt")
            return
        if doc.status == collections_pb2.DOCUMENT_STATUS_FAILED:
            raise RuntimeError(f"Indexation échouée : {doc.error_message}")
        await asyncio.sleep(3)

Le statut est une valeur d’énumération, pas une chaîne : le comparer au texte "DOCUMENT_STATUS_PROCESSED" ne serait jamais vrai, et la boucle tournerait sans fin. Pour un cas simple, upload_document(..., wait_for_indexing=True) fait cette attente à votre place.

Rechercher depuis le SDK

La recherche renvoie les mêmes fragments scorés que l’endpoint HTTP, dans la liste matches :

resultats = await client.collections.search(
    query="Procédure de validation des commandes",
    collection_ids=[collection_id],
)

for m in resultats.matches:
    print(f"Score : {m.score:.2f}")
    print(f"Contenu : {m.chunk_content[:200]}")
    print("---")

Un pipeline complet de bout en bout

Les briques précédentes s’assemblent en un script qui crée la collection, y verse des documents, attend leur indexation, cherche, puis génère une réponse avec l’outil collections_search :

import asyncio
import os

from xai_sdk import AsyncClient
from xai_sdk.chat import user
from xai_sdk.tools import collections_search


async def pipeline_rag():
    client = AsyncClient(
        api_key=os.getenv("XAI_API_KEY"),
        management_api_key=os.getenv("XAI_MANAGEMENT_API_KEY"),
    )

    # 1. Créer la collection
    collection = await client.collections.create(name="documentation-projet")
    collection_id = collection.collection_id

    # 2. Ajouter des documents en parallèle
    fichiers = [
        "docs/architecture.pdf",
        "docs/api-reference.pdf",
        "docs/guide-deploiement.md",
    ]
    file_ids = await asyncio.gather(*[
        ajouter_document(client, collection_id, f) for f in fichiers
    ])

    # 3. Attendre l'indexation de tous les documents
    await asyncio.gather(*[
        attendre_indexation(client, fid, collection_id) for fid in file_ids
    ])

    # 4. Rechercher
    resultats = await client.collections.search(
        query="Comment déployer en production ?",
        collection_ids=[collection_id],
    )
    for m in resultats.matches:
        print(f"[{m.score:.2f}] {m.chunk_content[:150]}")

    # 5. Générer une réponse qui s'appuie sur la collection
    chat = client.chat.create(
        model="grok-4.7",
        tools=[collections_search(collection_ids=[collection_id])],
    )
    chat.append(user("Explique la procédure de déploiement et cite tes sources."))
    reponse = await chat.sample()
    print(reponse.content)
    print(reponse.citations)


asyncio.run(pipeline_rag())

Les deux asyncio.gather() font tout le travail. Le premier lance les trois uploads ensemble au lieu d’attendre la fin de chacun, le second surveille les trois indexations en même temps : le temps total se rapproche de celui du document le plus long, au lieu de la somme de tous. Et comme l’attente ne bloque pas la boucle d’événements, le même code s’insère tel quel dans une application FastAPI ou aiohttp, qui continue de servir les autres requêtes pendant que vos documents s’indexent.

Points clés à retenir

Le SDK Python xAI existe en version synchrone (Client) et asynchrone (AsyncClient) ; await exige la seconde
Les deux clés (API et Management) se configurent dans le client
upload_document téléverse et rattache en un appel ; l’indexation se surveille par get_document ou wait_for_indexing=True
asyncio.gather() permet d’uploader et d’attendre l’indexation en parallèle
La génération s’appuie sur la collection avec l’outil collections_search, réponse et citations à l’appui

Testez vos connaissances

Files, collections, recherche : le RAG géré par xAI.

1. Que permet la Files API ?

Réponse : Uploader et gérer des fichiers (lister, consulter, supprimer) dans les formats supportés — la matière première des collections et des requêtes avec documents.

2. Qu'est-ce qu'une collection ?

Réponse : Un ensemble de documents géré côté xAI, avec métadonnées et permissions — l’index prêt à l’emploi qu’on interroge sans construire son propre pipeline vectoriel.

3. Quels modes de recherche sont disponibles ?

Réponse : Keyword (termes exacts), semantic (sens) et hybrid (les deux) — le choix suit la nature des questions, l’hybride couvrant la plupart des cas.

4. Comment le RAG s'intègre-t-il à la génération ?

Réponse : Par l’outil de recherche dans les collections : collections_search dans le SDK xAI, file_search avec les identifiants de collections dans l’API Responses. Le modèle interroge vos documents pendant la génération et cite ses sources — le pattern RAG sans infrastructure à maintenir.

5. Quels points d'attention pour la production ?

Réponse : La confidentialité des données uploadées (permissions, cycle de vie), la pertinence à optimiser (métadonnées, découpage) — et le SDK asynchrone pour les traitements en volume.

Un RAG géré : vous apportez les documents et les questions, xAI porte l’index — gardez la main sur la confidentialité et la pertinence.