Aller au contenu principal

Timeout et robustesse en production

Préparer votre application au monde réel

En production, les requêtes à la Responses API peuvent prendre du temps — surtout avec les modèles de raisonnement qui réfléchissent longuement avant de répondre. Une mauvaise gestion des timeouts peut provoquer des erreurs silencieuses ou une expérience utilisateur dégradée.

Le timeout de 3600 secondes

xAI recommande un timeout minimum de 3600 secondes (1 heure) pour les requêtes aux modèles de raisonnement. C’est considérablement plus long que les APIs classiques, et cela a des implications sur votre architecture.

Pourquoi si long ?

Les modèles grok-4 de raisonnement peuvent utiliser des centaines de milliers de tokens de réflexion interne avant de produire une réponse. Sur des problèmes complexes (mathématiques avancées, analyse de code, raisonnement multi-étapes), le modèle peut réfléchir pendant plusieurs minutes.

Configuration côté client

import httpx
from openai import OpenAI

# SDK OpenAI avec timeout étendu
client = OpenAI(
    api_key="xai-...",
    base_url="https://api.x.ai/v1",
    timeout=httpx.Timeout(3600.0, connect=10.0)
)

En curl :

curl --max-time 3600 https://api.x.ai/v1/responses \
  -H "Authorization: Bearer $XAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{...}'

Gestion des erreurs

Erreurs courantes

Code HTTPCauseAction
401Clé API invalideVérifier la clé
429Rate limit atteintRetry avec backoff exponentiel
500Erreur serveur xAIRetry après quelques secondes
503Service indisponibleRetry après 30 secondes
TimeoutRequête trop longueAugmenter le timeout ou simplifier le prompt

Pattern de retry avec backoff

import time

def call_with_retry(client, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return client.responses.create(**kwargs)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # 1s, 2s, 4s
            print(f"Erreur : {e}. Retry dans {wait}s...")
            time.sleep(wait)

Rate limiting

xAI impose des limites de débit selon votre plan. Implémentez un rate limiter côté client :

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_requests_per_minute=60):
        self.max_rpm = max_requests_per_minute
        self.timestamps = deque()

    def wait_if_needed(self):
        now = time.time()
        # Supprimer les timestamps de plus d'une minute
        while self.timestamps and self.timestamps[0] < now - 60:
            self.timestamps.popleft()
        if len(self.timestamps) >= self.max_rpm:
            sleep_time = 60 - (now - self.timestamps[0])
            time.sleep(max(0, sleep_time))
        self.timestamps.append(time.time())

Architecture asynchrone

Pour les applications web, ne bloquez pas le thread principal pendant 3600 secondes. Utilisez une architecture asynchrone :

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key="xai-...",
    base_url="https://api.x.ai/v1"
)

async def process_request(user_input):
    response = await client.responses.create(
        model="grok-4.20-reasoning",
        input=user_input,
        stream=True
    )
    async for event in response:
        if event.type == "response.output_text.delta":
            yield event.delta

Le streaming est la meilleure approche pour les applications interactives : l’utilisateur voit les premiers tokens en quelques secondes, même si la génération complète prend plus de temps.

Circuit breaker

Pour protéger votre application contre les pannes de l’API xAI :

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure = 0
        self.is_open = False

    def call(self, func, *args, **kwargs):
        if self.is_open:
            if time.time() - self.last_failure > self.reset_timeout:
                self.is_open = False
                self.failures = 0
            else:
                raise Exception("Circuit ouvert — API indisponible")

        try:
            result = func(*args, **kwargs)
            self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.threshold:
                self.is_open = True
            raise

Points clés à retenir

  • Configurez un timeout de 3600 secondes minimum pour les modèles de raisonnement
  • Implémentez un retry avec backoff exponentiel pour les erreurs 429 et 500
  • Utilisez le streaming pour les applications interactives — les utilisateurs voient les tokens immédiatement
  • Gérez le rate limiting côté client pour éviter les 429
  • Envisagez un circuit breaker pour protéger votre application contre les pannes API