Gestion des erreurs et mise en production

Leçon corrigée le 2 novembre 2026 d’après la documentation vidéo de xAI et le code du SDK xai-sdk 1.21.0 : client.video.generate (et non client.videos), résultat dans url, échecs signalés par VideoGenerationError avec un code (invalid_argument, service_unavailable…), limite de débit reçue en grpc.RpcError de code RESOURCE_EXHAUSTED — la classe RateLimitError et le statut expired n’existent pas. Complément du 3 novembre 2026 : la limite de débit se compte en requêtes par seconde (10 au palier d’entrée), et non en « 60 RPM », et le coût réel de chaque génération se lit dans usage.cost_in_usd_ticks. Sources : Video Generation · Rate limits.

Préparer votre application pour la production

Un script qui génère une vidéo sur votre machine et un service qui en produit des centaines par jour pour des utilisateurs réels n’ont presque rien en commun. Le second doit encaisser les refus de modération, les dépassements de quota, les incidents serveur et les URL qui expirent, sans jamais perdre silencieusement une génération facturée. Cette leçon rassemble les mécanismes qui font cette différence : classification des erreurs, stratégie de reprise, architecture de file d’attente et observabilité.

Deux familles d’erreurs, deux moments

La première famille apparaît immédiatement, à l’instant de l’appel POST. Ce sont des erreurs de requête au sens strict, et le code HTTP suffit à décider quoi faire.

La seconde famille se manifeste bien plus tard, pendant le polling, alors que la requête a été acceptée et que la génération a démarré.

Cette distinction n’est pas académique : un 400 doit remonter instantanément vers l’appelant parce que le code est fautif, tandis qu’un failed survient après plusieurs minutes d’attente et doit être traité dans le contexte du job, pas de la requête HTTP qui l’a déclenché.

Réessayer intelligemment

Le socle commun est le backoff exponentiel assorti d’un peu d’aléatoire. Le doublement du délai laisse le temps à un incident passager de se résorber, et le jitter évite que tous vos processus ne retentent au même instant après un pic — phénomène qui transforme une microcoupure en panne durable.

import asyncio
import random

async def appel_avec_retry(func, max_retries=3, base_delay=2):
    """Exécute une fonction avec retry et backoff exponentiel."""
    for tentative in range(max_retries + 1):
        try:
            return await func()
        except Exception as e:
            if tentative == max_retries:
                raise

            # Backoff exponentiel avec jitter
            delai = base_delay * (2 ** tentative) + random.uniform(0, 1)
            print(f"Tentative {tentative + 1} echouee : {e}. Retry dans {delai:.1f}s")
            await asyncio.sleep(delai)

Ce schéma générique ne suffit toutefois pas, car toutes les erreurs ne méritent pas la même patience. Un dépassement de quota se résout en attendant une minute entière ; un refus de modération, lui, ne se résoudra jamais, et le retenter revient à payer plusieurs fois pour le même échec. La fonction suivante différencie les cas.

import asyncio
from datetime import timedelta

import grpc
from xai_sdk.video import VideoGenerationError

# Codes d'échec qu'une nouvelle tentative ne corrigera pas
A_CORRIGER = {"invalid_argument", "permission_denied", "failed_precondition"}


async def generer_avec_retry(client, params, max_retries=3):
    """Génération vidéo avec retry ciblé (client : xai_sdk.AsyncClient)."""
    for tentative in range(max_retries + 1):
        derniere = tentative == max_retries
        try:
            return await client.video.generate(**params)
        except VideoGenerationError as e:
            if e.code.lower() in A_CORRIGER or derniere:
                raise  # requête ou contenu refusés : corriger avant de relancer
            await asyncio.sleep(5 * (tentative + 1))  # service surchargé, incident
        except grpc.RpcError as e:
            if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED or derniere:
                raise
            await asyncio.sleep(2 ** tentative)  # limite de débit : backoff
        except TimeoutError:
            if derniere:
                raise
            # Laisser plus de temps à la tentative suivante
            params["timeout"] = params.get("timeout", timedelta(minutes=10)) + timedelta(minutes=2)

Une file d’attente pour tenir le rythme

Dès que votre service reçoit des demandes de plusieurs utilisateurs, il faut découpler la réception des requêtes de leur exécution, et surtout borner la concurrence pour rester sous la limite de débit de votre palier. Une file associée à un sémaphore répond aux deux besoins : les jobs s’empilent librement, mais seuls dix au maximum s’exécutent en même temps.

import asyncio
from collections import deque

class FileGenerationVideo:
    def __init__(self, client, max_concurrent=10):
        self.client = client
        self.file = deque()
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.resultats = {}

    async def ajouter(self, job_id, params):
        """Ajoute une generation a la file."""
        self.file.append((job_id, params))

    async def traiter(self):
        """Traite toutes les generations en attente."""
        taches = []
        while self.file:
            job_id, params = self.file.popleft()
            tache = asyncio.create_task(
                self._executer(job_id, params)
            )
            taches.append(tache)
        await asyncio.gather(*taches, return_exceptions=True)

    async def _executer(self, job_id, params):
        """Exécute une generation avec contrôle de concurrence."""
        async with self.semaphore:
            try:
                result = await generer_avec_retry(
                    self.client, params
                )
                self.resultats[job_id] = {
                    "status": "done",
                    "url": result.url
                }
            except Exception as e:
                self.resultats[job_id] = {
                    "status": "failed",
                    "error": str(e)
                }

Notez que chaque job conserve son issue dans resultats, succès comme échec : une génération qui disparaît sans trace est un incident que personne ne pourra diagnostiquer.

Le second réflexe d’architecture concerne le stockage. Les URL de vidéo étant temporaires, le téléchargement doit être automatique et immédiat, avec un nommage horodaté qui garantit l’absence de collision entre deux générations simultanées.

import httpx
from pathlib import Path
from datetime import datetime

async def sauvegarder_video(video_url, dossier_sortie="./videos"):
    """Telecharge et sauvegarde une video generee."""
    Path(dossier_sortie).mkdir(parents=True, exist_ok=True)
    horodatage = datetime.now().strftime("%Y%m%d_%H%M%S")
    fichier = f"{dossier_sortie}/video_{horodatage}.mp4"

    async with httpx.AsyncClient() as client:
        resp = await client.get(video_url)
        resp.raise_for_status()
        Path(fichier).write_bytes(resp.content)

    return fichier

Vérifier sa clé avant de dépendre d’elle

L’endpoint GET /v1/api-key renseigne sur l’état de votre clé. L’appeler au démarrage du service vous évite de découvrir une clé bloquée au bout de vingt requêtes en échec, alors que vos utilisateurs attendent.

resp = httpx.get(
    "https://api.x.ai/v1/api-key",
    headers={"Authorization": f"Bearer {api_key}"}
).json()

if resp.get("api_key_blocked") or resp.get("api_key_disabled") or resp.get("team_blocked"):
    raise RuntimeError("Clé API ou équipe bloquée, ou clé désactivée")

Voir ce qui se passe en production

Cinq indicateurs suffisent à piloter un service de génération vidéo. Le taux de succès, c’est-à-dire la part des générations qui se terminent en done, mesure la santé globale. Le temps moyen de génération, mesuré entre la soumission et le statut done, sert à calibrer vos timeouts et à informer vos utilisateurs. Le taux d’erreur ventilé par type — failed selon son code d’erreur, timeout, rate limit — vous dit où porter l’effort : une hausse des timeouts sans hausse des failed signale par exemple des générations plus lentes que votre délai d’attente, pas des échecs. Le coût cumulé, obtenu en additionnant le champ usage.cost_in_usd_ticks de chaque résultat (un dollar vaut 10 milliards de ticks), ou à défaut en multipliant les durées générées par le prix à la seconde du modèle, doit être surveillé quotidiennement. Enfin, l’utilisation du quota, en requêtes par seconde rapportées à la limite de votre palier (10 au palier d’entrée pour les modèles vidéo), anticipe la saturation avant qu’elle ne se traduise en 429.

Ces métriques se construisent naturellement à partir d’un journal structuré, où chaque génération laisse une ligne exploitable par votre outil d’analyse.

import logging
import json

logger = logging.getLogger("video-generation")

def log_generation(request_id, status, duration_s, cout, temps_ms):
    logger.info(json.dumps({
        "event": "video_generation",
        "request_id": request_id,
        "status": status,
        "duration_seconds": duration_s,
        "cost_usd": cout,
        "processing_time_ms": temps_ms
    }))

Avant d’ouvrir votre service au public, reprenez l’ensemble depuis le début. Le retry avec backoff exponentiel et jitter doit être en place, et la concurrence bornée par un sémaphore pour rester sous la limite de débit. Le téléchargement des vidéos doit se déclencher automatiquement dès qu’elles sont prêtes, sans dépendre d’une action humaine. Chaque génération doit laisser une ligne de journal portant son request_id, son statut et son coût, faute de quoi le taux d’erreur et le temps moyen de génération resteront invérifiables. Vérifiez l’état de la clé API au démarrage plutôt qu’au premier échec. Et fixez un budget assorti d’alertes de dépassement : à $0.05 la seconde, une boucle de retry mal bornée consomme plus vite qu’on ne le croit.

Tarifs vérifiés le 3 novembre 2026 — les prix évoluent régulièrement : avant tout calcul de budget, vérifiez la grille en vigueur sur la page officielle des modèles et tarifs xAI.

Points clés à retenir

• Les erreurs se divisent en synchrones (requête) et asynchrones (génération)
• Toutes les erreurs ne méritent pas un retry : la modération est définitive
• Utilisez un sémaphore pour limiter les générations concurrentes
• Les URLs temporaires imposent un téléchargement immédiat et systématique
• Suivez les métriques de coût, de succès et de temps de génération pour piloter votre service

Testez vos connaissances

La vidéo générative en production : validez le workflow complet.

1. Quelles sont les particularités d'un prompt vidéo ?

Réponse : Décrire l’action et le mouvement, pas seulement la scène : sujet, déplacement, ambiance, cadrage — la dimension temporelle est ce qui distingue le prompt vidéo du prompt image.

2. Que permet l'image-to-video ?

Réponse : Animer une image existante, ou faire apparaître dans une scène nouvelle des personnes et des produits donnés grâce aux images de référence — et, sur grok-imagine-video-1.5, épingler la première, la dernière image ou des images clés : votre identité visuelle devient un point de départ, pas un hasard.

3. Édition et extension : quels usages ?

Réponse : /videos/edits modifie une vidéo générée, /videos/extensions la prolonge — et les deux se combinent pour construire des séquences longues par itérations.

4. Pourquoi le workflow est-il asynchrone ?

Réponse : La génération vidéo prend du temps : on soumet un job, on interroge son statut (polling) puis on récupère le résultat — le SDK xAI automatise ce polling.

5. Que prévoir pour la production ?

Réponse : La gestion des erreurs et des jobs échoués, le téléchargement immédiat des sorties, et le suivi des coûts (résolution, durée, itérations) — la vidéo est la modalité la plus chère : elle se budgète.

Prompt temporel, référence visuelle, asynchrone maîtrisé : la vidéo générative devient un pipeline — pas une loterie.