Batch API et checklist production

L’alternative Batch API : -50% sur les tokens

Quand le temps réel n’est pas une contrainte, la Batch API offre une réduction de 50% sur tous les tokens texte. Au lieu d’envoyer des requêtes une par une et de gérer la concurrence, vous soumettez un fichier JSONL contenant toutes vos requêtes, et vous récupérez les résultats quand le traitement est terminé.

Quand utiliser la Batch API

• Traitement de catalogues produits (milliers de descriptions)
• Classification de tickets support en masse
• Génération de résumés pour une base documentaire
• Toute tâche ou un délai de quelques heures est acceptable

Quand rester en asynchrone classique

• Chatbot interactif (réponse en temps réel)
• Pipeline où chaque étape dépend de la précédente
• Monitoring et alertes en temps réel
• Tout cas où le délai de la Batch API est inacceptable

Comparatif des approches

Checklist de mise en production

Avant de déployer votre système de requêtes asynchrones en production, vérifiez chaque point de cette liste :

Configuration

• Le timeout est configuré à 3600 secondes pour les modèles de raisonnement
• Le sémaphore est dimensionne en fonction de vos rate limits réels (pas des estimations)
• Le backoff exponentiel est implémenté avec jitter pour les erreurs 429
• Les clés API sont chargées depuis des variables d’environnement, jamais en dur dans le code
• Le header x-grok-conv-id est utilisé pour les conversations multi-tour

Gestion des erreurs

• return_exceptions=True est utilisé dans asyncio.gather pour isoler les erreurs
• Les erreurs 429, timeout et connexion déclenchent un retry
• Les erreurs de requête invalide (BadRequestError) ne déclenchent pas de retry
• Les requêtes échouées sont identifiées et peuvent être retraitees
• Les logs incluent le type d’erreur, le numéro de tentative et le délai de retry

Monitoring

• Le nombre de tokens consommés est suivi via response.usage
• Le taux de cache est mesure pour valider l’utilisation de x-grok-conv-id
• Les coûts sont estimés en temps réel avec cost_in_usd_ticks
• Les rate limits restants sont surveilles pour ajuster le sémaphore dynamiquement

Performances

• Le chunking est utilisé pour les lots de plus de 100 requêtes
• Les résultats intermédiaires sont sauvegardés entre les sous-lots
• La progression est affichée pour les traitements longs
• Le sémaphore est ajuste en fonction du modèle utilise (reasoning = plus bas)

Sécurité

• Les clés API ne sont pas exposées dans les logs ou les messages d’erreur
• Les réponses contenant des données sensibles sont traitées en mémoire, pas écrites sur disque
• Le stockage distant (store_messages) est désactivé pour les données confidentielles
• use_encrypted_content est active quand la confidentialité du raisonnement est requise

Architecture recommandée

import os
import asyncio
import uuid
from asyncio import Semaphore
from openai import AsyncOpenAI, RateLimitError, APITimeoutError
import httpx

class GrokAsyncProcessor:
    def __init__(self, max_concurrent: int = 5):
        self.client = AsyncOpenAI(
            api_key=os.getenv("XAI_API_KEY"),
            base_url="https://api.x.ai/v1",
            timeout=httpx.Timeout(3600.0)
        )
        self.sem = Semaphore(max_concurrent)
        self.conv_id = str(uuid.uuid4())

    async def request(self, prompt: str, max_retries: int = 5) -> str:
        async with self.sem:
            for attempt in range(max_retries):
                try:
                    response = await self.client.chat.completions.create(
                        model="grok-4.5",
                        messages=[{"role": "user", "content": prompt}],
                        extra_headers={"x-grok-conv-id": self.conv_id}
                    )
                    return response.choices[0].message.content
                except RateLimitError:
                    if attempt == max_retries - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)
                except APITimeoutError:
                    if attempt == max_retries - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)

    async def batch(self, prompts: list[str]) -> list[str | Exception]:
        tasks = [self.request(p) for p in prompts]
        return await asyncio.gather(*tasks, return_exceptions=True)

Cette classe encapsule tous les patterns vus dans ce cours : client asynchrone, sémaphore, backoff, cache, et gestion des erreurs. Elle constitue un point de départ solide pour vos projets en production.

Points clés à retenir

• La Batch API offre -50% mais impose un délai de traitement
• Choisissez entre async temps réel et batch selon vos contraintes de latence
• La checklist production couvre configuration, erreurs, monitoring, performances et sécurité
• La classe GrokAsyncProcessor encapsule tous les patterns du cours en une seule abstraction
• Testez toujours votre configuration avec un petit lot avant de lancer un traitement massif

Testez vos connaissances

Async, limites, cache : les patterns de production en cinq questions.

1. Quand l'async s'impose-t-il face au sync ?

Réponse : Dès qu’on traite plusieurs requêtes indépendantes : l’AsyncClient (ou AsyncOpenAI) superpose les attentes réseau — le débit se multiplie sans multiplier les threads.

2. Comment traiter un lot sans se faire limiter ?

Réponse : asyncio.gather pour le parallélisme, un sémaphore pour plafonner les requêtes simultanées, et le backoff exponentiel sur HTTP 429 — la politesse envers les RPM/TPM de la console.

3. Comment gère-t-on le contexte multi-tour à l'échelle ?

Réponse : Par le stockage serveur des réponses et le raisonnement chiffré réutilisé — l’état vit côté API, votre code n’a que des identifiants à suivre.

4. Que faut-il savoir du cache et des timeouts ?

Réponse : Le cache par conversation (x-grok-conv-id) accélère les fils suivis, et les requêtes avec raisonnement exigent des timeouts adaptés — deux réglages qui changent la production.

5. Que contient la checklist de production du cours ?

Réponse : Async maîtrisé, limites respectées (sémaphore + backoff), batch pour le volume froid, timeouts et retries configurés, coûts suivis — la robustesse en liste de contrôle.

Concurrence bornée, échecs prévus, état côté serveur : les patterns async transforment vos scripts en services — la checklist finale les verrouille.