Le SDK Python xAI : operations asynchrones

Automatiser vos workflows RAG en Python

Le SDK Python de xAI offre une interface native pour toutes les operations vues dans ce cours : upload de fichiers, gestion de collections, recherche et generation. Son architecture asynchrone (async/await) permet de gerer efficacement les operations paralleles et les temps d’attente.

Installation et configuration

pip install xai-sdk

La configuration du client necessite vos deux cles API :

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 parametre timeout (en secondes) est important pour les operations longues comme l’upload de gros fichiers ou le polling du traitement des documents.

Operations sur les collections

Creer une collection

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

Lister les collections

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

Supprimer une collection

await client.collections.delete(collection_id)

Upload et ajout de documents

Le SDK simplifie le processus en deux etapes :

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

Polling du traitement

Attendez que le document soit indexe avant de lancer des recherches :

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)

Recherche dans les collections

results = await client.collections.search(
    query="Procedure 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("---")

Pipeline complet async

Voici un pipeline complet qui cree une collection, y ajoute des documents et lance une recherche :

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. Creer la collection
    collection = await client.collections.create(
        name="documentation-projet"
    )

    # 2. Ajouter des documents en parallele
    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 deployer en production ?",
        collection_ids=[collection.id]
    )

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

    # 5. Generer une reponse avec RAG
    response = await client.responses.create(
        model="grok-3",
        input="Explique la procedure de deploiement.",
        tools=[{
            "type": "collections_search",
            "collection_ids": [collection.id]
        }]
    )

    print(response.output_text)

asyncio.run(pipeline_rag())

L’avantage de l’asynchrone

L’architecture async du SDK permet de :

• Uploader plusieurs fichiers en parallele avec asyncio.gather()
• Attendre le traitement sans bloquer grace a await asyncio.sleep()
• Lancer des recherches concurrentes dans plusieurs collections
• Integrer dans des frameworks web async (FastAPI, aiohttp)

Pour un workflow avec 10 fichiers, l’upload parallele peut diviser le temps total par 5 ou plus, compare a un traitement sequentiel.

Points cles a retenir

• Le SDK Python xAI fournit une interface async native pour toutes les operations Collections et RAG
• Les deux cles (API + Management) se configurent dans le client
• asyncio.gather() permet d’uploader et d’attendre l’indexation en parallele
• Le pipeline complet va de la creation de collection a la generation de reponse avec RAG
• L’architecture async s’integre naturellement avec les frameworks web modernes (FastAPI, etc.)