Gestion des erreurs et mise en production

Preparer votre application pour la production

Le passage en production d’un systeme de generation video necessite une gestion robuste des erreurs, une strategie de retry, et une architecture adaptee au caractere asynchrone de l’API. Cette lecon couvre les bonnes pratiques pour deployer un service de generation video fiable.

Taxonomie des erreurs

Erreurs de requete (synchrones)

Ces erreurs surviennent immediatement lors de l’appel POST :

Erreurs de generation (asynchrones)

Ces erreurs apparaissent lors du polling :

Strategie de retry robuste

Backoff exponentiel

import asyncio
import random

async def appel_avec_retry(func, max_retries=3, base_delay=2):
    """Execute 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)

Retry specifique par type d’erreur

Toutes les erreurs ne meritent pas un retry :

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 delai 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)

Architecture de production

File d’attente de generations

Pour un systeme qui recoit des demandes de generation video, implementez une file d’attente :

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):
        """Execute une generation avec controle 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)
                }

Sauvegarde systematique

Les URLs de video sont temporaires. Implementez un telechargement automatique des que la video est prete :

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

Verifier l’info de votre cle API

L’endpoint GET /v1/api-key vous permet de verifier l’etat de votre cle en production :

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("Cle API desactivee ou bloquee")

Cela vous permet de detecter proactivement un probleme d’authentification avant de lancer des generations.

Monitoring et observabilite

Metriques a suivre

• Taux de succes : pourcentage de generations terminee en done
• Temps moyen de generation : duree entre la soumission et le statut done
• Taux d’erreur par type : failed, expired, timeout, rate limit
• Cout cumule : somme des durees generees x $0.05
• Utilisation du quota : requetes par minute vs limite de 60 RPM

Logging structure

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
    }))

Checklist de mise en production

• Implementer le retry avec backoff exponentiel et jitter
• Limiter la concurrence (semaphore) pour respecter les 60 RPM
• Telecharger automatiquement les videos des qu’elles sont pretes
• Logger chaque generation avec le request_id, le statut et le cout
• Surveiller le taux d’erreur et le temps moyen de generation
• Verifier l’etat de la cle API au demarrage du service
• Prevoir un budget et des alertes de depassement de cout

Points cles a retenir

• Les erreurs se divisent en synchrones (requete) et asynchrones (generation)
• Toutes les erreurs ne meritent pas un retry : la moderation est definitive
• Utilisez un semaphore pour limiter les generations concurrentes
• Les URLs temporaires imposent un telechargement immediat et systematique
• Suivez les metriques de cout, de succes et de temps de generation pour piloter votre service