Le SDK Python xAI : opérations asynchrones

Automatiser vos workflows RAG en Python

Tout ce que vous avez appelé en curl depuis le début de ce cours existe en Python natif : upload de fichiers, gestion de collections, recherche, génération. Le SDK xAI est construit autour de async/await, et ce n’est pas un détail de style. Un workflow RAG passe le plus clair de son temps à attendre — un upload qui monte, un document qui s’indexe, une recherche qui revient — et l’asynchrone permet de faire 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 Client

client = Client(
    api_key=os.getenv("XAI_API_KEY"),
    management_api_key=os.getenv("XAI_MANAGEMENT_API_KEY"),
    timeout=3600
)

Le timeout, exprimé en secondes, mérite votre attention : la valeur par défaut suffit pour une recherche mais pas pour l’upload d’un PDF volumineux ni pour un polling d’indexation qui s’étire. Une heure vous met à l’abri des coupures en plein traitement.

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.id
print(f"Collection creee : {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 :

collections = await client.collections.list()
for col in collections:
    print(f"{col.id} — {col.name}")

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

await client.collections.delete(collection_id)

Ajouter un document et attendre son indexation

L’ajout se fait en deux temps, l’upload du fichier puis son rattachement à la collection :

import asyncio

async def ajouter_document(client, collection_id, chemin_fichier):
    # Upload du fichier
    with open(chemin_fichier, "rb") as f:
        upload = await client.files.create(
            file=f,
            purpose="assistants"
        )
    file_id = upload.id

    # Ajout a la collection
    await client.collections.upload_document(
        collection_id=collection_id,
        name=chemin_fichier.split("/")[-1],
        data=open(chemin_fichier, "rb").read()
    )

    return file_id

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 :

async def attendre_indexation(client, file_id, collection_id):
    while True:
        doc = await client.collections.get_document(
            file_id, collection_id
        )
        if doc.status == "DOCUMENT_STATUS_PROCESSED":
            print("Document pret")
            return
        print(f"Statut : {doc.status}, attente...")
        await asyncio.sleep(3)

Rechercher depuis le SDK

La recherche renvoie les mêmes fragments scorés que l’endpoint HTTP, sous forme d’objets Python :

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

for result in results:
    print(f"Score: {result.score:.2f}")
    print(f"Contenu: {result.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 :

import asyncio
import os
from xai_sdk import Client

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

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

    # 2. Ajouter des documents en parallèle
    fichiers = [
        "docs/architecture.pdf",
        "docs/api-reference.pdf",
        "docs/guide-deploiement.md"
    ]

    tasks = [
        ajouter_document(client, collection.id, f)
        for f in fichiers
    ]
    file_ids = await asyncio.gather(*tasks)

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

    # 4. Rechercher
    results = await client.collections.search(
        query="Comment déployer en production ?",
        collection_ids=[collection.id]
    )

    for r in results:
        print(f"[{r.score:.2f}] {r.content[:150]}")

    # 5. Générer une réponse avec RAG
    response = await client.responses.create(
        model="grok-4.5",
        input="Explique la procédure de deploiement.",
        tools=[{
            "type": "collections_search",
            "collection_ids": [collection.id]
        }]
    )

    print(response.output_text)

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. Sur dix fichiers, ce simple changement de structure divise le temps total par cinq ou davantage face à une boucle séquentielle — 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, où il continuera de servir les autres requêtes pendant que vos documents s’indexent.

Points clés à retenir

• Le SDK Python xAI fournit une interface async native pour toutes les opérations Collections et RAG
• Les deux clés (API + Management) se configurent dans le client
• asyncio.gather() permet d’uploader et d’attendre l’indexation en parallèle
• Le pipeline complet va de la création de collection à la génération de réponse avec RAG
• L’architecture async s’intègre naturellement avec les frameworks web modernes (FastAPI, etc.)

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 à l'API Responses ?

Réponse : Par collections_search et file_search : le modèle interroge vos documents pendant la génération et cite ses sources — le pattern RAG hybride 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.