Gestion des erreurs et mise en production

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.

async def generer_avec_retry(client, params, max_retries=3):
    """Generation video avec retry intelligent."""
    for tentative in range(max_retries + 1):
        try:
            result = await client.videos.generate(**params)
            return result
        except RateLimitError:
            # Retry avec délai plus long
            await asyncio.sleep(60)
        except VideoGenerationError as e:
            if "moderation" in str(e).lower():
                # Ne pas retry : le prompt est refuse
                raise
            # Erreur technique : retry
            await asyncio.sleep(5 * (tentative + 1))
        except TimeoutError:
            # Retry avec timeout plus long
            params["timeout"] = params.get("timeout", 600) + 120
        except Exception:
            if tentative == max_retries:
                raise
            await asyncio.sleep(5)

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 les 60 RPM. 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.video_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"):
    raise Exception("Clé API désactivée ou bloqué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, expired, timeout, rate limit — vous dit où porter l’effort, un pic d’expired signalant par exemple un problème de téléchargement plutôt que de génération. Le coût cumulé, obtenu en multipliant la somme des durées générées par $0.05, doit être surveillé quotidiennement. Enfin, l’utilisation du quota, en requêtes par minute rapportées à la limite de 60 RPM, 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 les 60 RPM. 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 relevés le 5 août 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 — avec images de référence et guide de style pour garder la cohérence visuelle : votre identité graphique 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.