Workflow asynchrone et polling

Comprendre le cycle de vie d’une vidéo

Toutes les opérations vidéo de l’API xAI sont asynchrones, et cette caractéristique conditionne l’architecture de votre application bien plus qu’on ne l’imagine au départ. Un appel de chat ou une génération d’image vous rendent un résultat dans la foulée ; la génération vidéo, elle, demande un temps de traitement significatif et vous rend seulement un accusé de réception. Il vous revient donc d’implémenter le suivi : interroger périodiquement l’état de la tâche, détecter sa terminaison, et récupérer le fichier. Le cycle complet tient en quatre moments.

Interroger l’état d’une génération

Le suivi passe par un simple GET sur l’endpoint de statut, avec l’identifiant reçu à la soumission.

curl "https://api.x.ai/v1/videos/vid_abc123def456" \
  -H "Authorization: Bearer $XAI_API_KEY"

La réponse expose quatre informations utiles :

• status : l’état courant de la génération
• progress : un entier de 0 à 100 indiquant l’avancement
• video_url : l’URL de la vidéo (uniquement quand le statut est done)
• video.respect_moderation : un booléen indiquant si le contenu respecte la politique de modération

Le champ progress mérite une remarque : il ne sert pas votre logique de contrôle, qui doit toujours s’appuyer sur status, mais il vaut de l’or dans une interface utilisateur. Afficher « 62 % » plutôt qu’un spinner immobile change complètement la perception d’une attente qui peut durer plusieurs minutes.

Quatre états, quatre conduites à tenir

Tant que le statut vaut pending, la génération suit son cours : vous continuez simplement d’interroger à intervalles réguliers, en relayant progress vers votre interface. Quand il passe à done, la vidéo est prête, et la seule chose à faire est de lire video_url et de télécharger immédiatement le fichier — ces URL sont temporaires et expirent après un délai qui n’est pas spécifié, ce qui vous interdit de les stocker en base pour un usage ultérieur.

Le statut expired signale précisément ce cas de figure : le résultat a expiré avant que vous ne le récupériez, et il ne reste qu’à relancer la génération, en la repayant. Le statut failed, enfin, indique un échec de production dont les causes possibles sont un prompt non conforme à la politique de contenu, un incident technique, ou une image ou une vidéo source invalide. La distinction compte pour la suite : expired justifie une nouvelle tentative à l’identique, failed demande d’abord de comprendre pourquoi.

Implémenter le polling

La version synchrone tient en une fonction. Elle boucle jusqu’au timeout, coupe court sur les deux statuts terminaux d’échec, et retourne l’URL dès que la vidéo est prête.

import httpx
import time

def attendre_video(request_id, api_key, timeout=600, intervalle=5):
    """Attend la completion d'une generation video."""
    headers = {"Authorization": f"Bearer {api_key}"}
    debut = time.time()

    while time.time() - debut < timeout:
        resp = httpx.get(
            f"https://api.x.ai/v1/videos/{request_id}",
            headers=headers
        ).json()

        status = resp.get("status")
        progress = resp.get("progress", 0)

        if status == "done":
            return resp["video_url"]
        elif status == "failed":
            raise Exception("Generation echouee")
        elif status == "expired":
            raise Exception("Generation expiree")

        print(f"En cours... {progress}%")
        time.sleep(intervalle)

    raise TimeoutError(f"Timeout apres {timeout} secondes")

Dans un service qui traite plusieurs demandes de front, cette version bloquante devient un goulet : chaque attente immobilise un thread pendant des minutes. La transposition asynchrone libère la boucle d’événements entre deux interrogations et vous permet de suivre des dizaines de générations en parallèle.

import httpx
import asyncio

async def attendre_video_async(request_id, api_key, timeout=600, intervalle=5):
    """Version asynchrone du polling."""
    headers = {"Authorization": f"Bearer {api_key}"}
    async with httpx.AsyncClient() as client:
        debut = asyncio.get_event_loop().time()

        while asyncio.get_event_loop().time() - debut < timeout:
            resp = await client.get(
                f"https://api.x.ai/v1/videos/{request_id}",
                headers=headers
            )
            data = resp.json()

            if data["status"] == "done":
                return data["video_url"]
            elif data["status"] in ("failed", "expired"):
                raise Exception(f"Statut : {data['status']}")

            await asyncio.sleep(intervalle)

    raise TimeoutError("Timeout")

Régler l’intervalle d’interrogation

Le choix de l’intervalle arbitre entre réactivité et gaspillage de requêtes. Cinq secondes constituent un bon compromis dans la plupart des cas. Sur des générations longues — dix à quinze secondes de vidéo —, dix à quinze secondes d’intervalle suffisent amplement et allègent le trafic. À l’inverse, une application interactive où l’utilisateur regarde une barre de progression justifie de descendre à deux ou trois secondes. En dessous d’une seconde, en revanche, vous n’accélérez rien du tout : la génération avance à son rythme, et vous ne faites que consommer votre quota de 60 requêtes par minute pour rien.

Points clés à retenir

• Toutes les opérations vidéo sont asynchrones : POST pour soumettre, GET pour récupérer
• Quatre statuts possibles : pending, done, expired, failed
• Le champ progress (0-100) permet de suivre l’avancement
• Implémentez un timeout et un intervalle de polling adapté à votre cas d’usage
• Les URLs de vidéo sont temporaires : téléchargez immédiatement