Batch API et checklist production

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

Quand le temps reel n’est pas une contrainte, la Batch API offre une reduction de 50% sur tous les tokens texte. Au lieu d’envoyer des requetes une par une et de gerer la concurrence, vous soumettez un fichier JSONL contenant toutes vos requetes, et vous recuperez les resultats quand le traitement est termine.

Quand utiliser la Batch API

• Traitement de catalogues produits (milliers de descriptions)
• Classification de tickets support en masse
• Generation de resumes pour une base documentaire
• Toute tache ou un delai de quelques heures est acceptable

Quand rester en asynchrone classique

• Chatbot interactif (reponse en temps reel)
• Pipeline ou chaque etape depend de la precedente
• Monitoring et alertes en temps reel
• Tout cas ou le delai de la Batch API est inacceptable

Comparatif des approches

Checklist de mise en production

Avant de deployer votre systeme de requetes asynchrones en production, verifiez chaque point de cette liste :

Configuration

• Le timeout est configure a 3600 secondes pour les modeles de raisonnement
• Le semaphore est dimensionne en fonction de vos rate limits reels (pas des estimations)
• Le backoff exponentiel est implemente avec jitter pour les erreurs 429
• Les cles API sont chargees depuis des variables d’environnement, jamais en dur dans le code
• Le header x-grok-conv-id est utilise pour les conversations multi-tour

Gestion des erreurs

• return_exceptions=True est utilise dans asyncio.gather pour isoler les erreurs
• Les erreurs 429, timeout et connexion declenchent un retry
• Les erreurs de requete invalide (BadRequestError) ne declenchent pas de retry
• Les requetes echouees sont identifiees et peuvent etre retraitees
• Les logs incluent le type d’erreur, le numero de tentative et le delai de retry

Monitoring

• Le nombre de tokens consommes est suivi via response.usage
• Le taux de cache est mesure pour valider l’utilisation de x-grok-conv-id
• Les couts sont estimes en temps reel avec cost_in_usd_ticks
• Les rate limits restants sont surveilles pour ajuster le semaphore dynamiquement

Performances

• Le chunking est utilise pour les lots de plus de 100 requetes
• Les resultats intermediaires sont sauvegardes entre les sous-lots
• La progression est affichee pour les traitements longs
• Le semaphore est ajuste en fonction du modele utilise (reasoning = plus bas)

Securite

• Les cles API ne sont pas exposees dans les logs ou les messages d’erreur
• Les reponses contenant des donnees sensibles sont traitees en memoire, pas ecrites sur disque
• Le stockage distant (store_messages) est desactive pour les donnees confidentielles
• use_encrypted_content est active quand la confidentialite du raisonnement est requise

Architecture recommandee

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",
                        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, semaphore, backoff, cache, et gestion des erreurs. Elle constitue un point de depart solide pour vos projets en production.

Points cles a retenir

• La Batch API offre -50% mais impose un delai de traitement
• Choisissez entre async temps reel et batch selon vos contraintes de latence
• La checklist production couvre configuration, erreurs, monitoring, performances et securite
• 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