SDK xAI et polling automatique

Simplifier le polling avec les SDK

La leçon précédente vous a fait écrire une boucle de polling à la main. C’est un exercice utile pour comprendre le cycle de vie d’une génération, mais c’est aussi une trentaine de lignes à maintenir dans chaque projet, avec sa gestion de timeout, ses cas d’échec et ses conversions de statut. Les SDK officiels de xAI encapsulent tout cela : vous soumettez la requête, ils s’occupent de l’attente. Voyons ce que cela change concrètement avec le SDK Python.

Un appel unique au lieu d’une boucle

La méthode videos.generate regroupe la soumission, le suivi et la récupération. Elle ne rend la main qu’une fois la vidéo prête, ou lève une exception si quelque chose s’est mal passé.

from xai_sdk import Client

client = Client(api_key="votre_cle")

# Generation text-to-video avec polling automatique
result = await client.videos.generate(
    model="grok-imagine-video",
    prompt="Un lever de soleil sur les Alpes, time-lapse accélère",
    duration=5,
    aspect_ratio="16:9",
    resolution="720p"
)

print(result.video_url)

Derrière ces quelques lignes, le SDK effectue exactement ce que vous codiez précédemment : l’envoi de la requête POST, le polling à intervalles réguliers — environ 100 ms par défaut —, la détection du statut final (done, failed ou expired) et, selon le cas, le retour du résultat ou la levée d’une exception.

Un réglage mérite votre attention dès la mise en service : par défaut, le SDK patiente jusqu’à 10 minutes avant de déclarer un timeout. C’est confortable pour un script de production nocturne, beaucoup trop long pour une requête web à laquelle un utilisateur attend une réponse. Le paramètre timeout vous laisse resserrer cette borne.

result = await client.videos.generate(
    model="grok-imagine-video",
    prompt="Votre description",
    duration=10,
    timeout=300  # 5 minutes max
)

Traiter les erreurs à la source

Le SDK expose des exceptions typées, ce qui vous évite d’analyser des chaînes de caractères pour savoir ce qui a échoué. Un bloc try structuré vous permet ainsi de distinguer un prompt refusé d’un simple dépassement de délai.

from xai_sdk.exceptions import VideoGenerationError, TimeoutError

try:
    result = await client.videos.generate(
        model="grok-imagine-video",
        prompt="Votre description",
        duration=5
    )
    print(f"Video prete : {result.video_url}")
except VideoGenerationError as e:
    print(f"Erreur de generation - Code: {e.code}, Message: {e.message}")
except TimeoutError:
    print("La generation a pris trop de temps")
except Exception as e:
    print(f"Erreur inattendue : {e}")

Chaque exception appelle une réaction différente, et confondre les quatre conduit à des boucles de retry inutiles ou, pire, à des générations facturées en pure perte.

Édition et extension, mêmes facilités

Les deux autres endpoints disposent de leurs méthodes dédiées, avec le même polling intégré. L’édition applique une modification à une vidéo existante :

result = await client.videos.edit(
    model="grok-imagine-video",
    prompt="Transforme en style anime japonais",
    video_url="https://exemple.com/video.mp4"
)

et l’extension prolonge une séquence sur la durée que vous précisez :

result = await client.videos.extend(
    model="grok-imagine-video",
    prompt="Continue la scène avec un panoramique vers la droite",
    video_url="https://exemple.com/video.mp4",
    duration=5
)

Ne jamais laisser une vidéo dans le cloud

Le SDK ne change rien au caractère temporaire des URL de résultat. Si votre code se contente d’afficher result.video_url, vous avez payé une génération dont il ne restera bientôt plus rien. La bonne pratique consiste à enchaîner systématiquement génération et téléchargement dans la même fonction, de sorte qu’il devienne impossible d’oublier l’étape.

import httpx
from pathlib import Path

async def generer_et_sauvegarder(client, prompt, fichier_sortie, **kwargs):
    """Génère une video et la sauvegarde localement."""
    result = await client.videos.generate(
        model="grok-imagine-video",
        prompt=prompt,
        **kwargs
    )

    # Telechargement immediat
    async with httpx.AsyncClient() as http:
        resp = await http.get(result.video_url)
        Path(fichier_sortie).write_bytes(resp.content)

    print(f"Video sauvegardee : {fichier_sortie}")
    return fichier_sortie

Générer plusieurs vidéos de front

Comme chaque appel est une coroutine, asyncio.gather permet de lancer un lot entier en parallèle. L’option return_exceptions=True est ici décisive : sans elle, une seule requête refusée ferait échouer l’ensemble du lot et vous perdriez les résultats déjà obtenus.

import asyncio

async def batch_generation(client, requetes):
    """Génère plusieurs videos en parallele."""
    taches = []
    for req in requetes:
        tache = client.videos.generate(
            model="grok-imagine-video",
            prompt=req["prompt"],
            duration=req.get("duration", 5),
            aspect_ratio=req.get("aspect_ratio", "16:9")
        )
        taches.append(tache)

    resultats = await asyncio.gather(*taches, return_exceptions=True)

    for i, res in enumerate(resultats):
        if isinstance(res, Exception):
            print(f"Requête {i+1} echouee : {res}")
        else:
            print(f"Requête {i+1} terminee : {res.video_url}")

    return resultats

Deux garde-fous encadrent cette parallélisation. Respectez la limite de 60 RPM, et ne dépassez pas dix à quinze générations simultanées. Rappelez-vous aussi ce que le parallélisme fait et ne fait pas : il n’accélère aucune génération individuelle, il réduit seulement le temps total du lot.

Faut-il abandonner le polling manuel ?

Pas systématiquement. La comparaison suivante montre où chaque approche prend l’avantage.

Le polling manuel garde sa pertinence dans deux situations : quand vous avez besoin d’un contrôle fin, par exemple pour relayer la progression vers une interface temps réel ou pour l’intégrer à un système de files existant, et quand votre contexte de déploiement vous interdit d’ajouter une dépendance supplémentaire.

Points clés à retenir

• Le SDK gère automatiquement le polling avec un timeout de 10 minutes par défaut
• Les exceptions spécifiques facilitent le traitement des erreurs
• Téléchargez toujours la vidéo immédiatement après la génération (URLs temporaires)
• Le parallélisme via asyncio.gather accélère les batches de générations
• Le polling manuel reste pertinent pour des besoins de contrôle avancés