Batch API : traitement asynchrone massif

Échanger du délai contre de l’argent

La Batch API repose sur un marché très simple : vous acceptez que vos requêtes soient traitées de manière asynchrone, dans une fenêtre pouvant aller jusqu’à vingt-quatre heures, et vous payez cinquante pour cent moins cher. Tout ce qui n’a pas besoin d’être servi à un humain qui attend devant son écran relève de ce régime. Sur un volume important, ce n’est pas une micro-optimisation : c’est la moitié de la ligne budgétaire.

Le réflexe à installer dans une équipe consiste à se demander, pour chaque nouveau traitement, si quelqu’un attend le résultat. L’analyse d’un stock de contrats, la génération de fiches produit pour un catalogue, l’exécution d’un benchmark sur des milliers de cas de test, la classification d’un backlog de tickets accumulés depuis six mois, l’extraction de champs sur un lot de factures : dans tous ces cas, la réponse arrive dans un tableau de bord ou une base de données, pas dans une interface conversationnelle. Le délai est invisible, l’économie ne l’est pas.

Construire et lancer un lot

Le format d’entrée est un fichier JSONL : une ligne par requête, chaque ligne étant un objet complet et indépendant. Le champ custom_id est celui qui vous permettra de raccrocher chaque réponse à sa requête d’origine, puisque rien ne garantit l’ordre du fichier de sortie. Choisissez-le stable et signifiant — un identifiant de document en base plutôt qu’un numéro d’itération — sinon la phase de réconciliation deviendra un casse-tête.

import json

requetes = [
    {
        "custom_id": f"doc-{i}",
        "method": "POST",
        "url": "/v1/responses",
        "body": {
            "model": "gpt-5.6-luna",
            "input": f"Résumez ce document en 3 points clés : {doc}",
            "max_output_tokens": 500,
        },
    }
    for i, doc in enumerate(liste_documents)
]

# Écrire le fichier JSONL
with open("batch_input.jsonl", "w") as f:
    for requete in requetes:
        f.write(json.dumps(requete) + "\n")

Le fichier est ensuite téléversé, puis référencé à la création du batch. Les métadonnées ne servent à rien techniquement, mais elles vous sauveront le jour où trois lots tourneront en parallèle et où il faudra retrouver lequel correspond à quelle version de prompt.

import openai

client = openai.OpenAI()

# Upload du fichier
fichier = client.files.create(
    file=open("batch_input.jsonl", "rb"),
    purpose="batch",
)

# Création du batch
batch = client.batches.create(
    input_file_id=fichier.id,
    endpoint="/v1/responses",
    completion_window="24h",
    metadata={"projet": "analyse-contrats", "version": "2.1"},
)

print(f"Batch créé : {batch.id}")
print(f"Statut : {batch.status}")

Le suivi consiste à interroger le batch jusqu’à ce qu’il quitte l’état en cours, puis à télécharger le fichier de sortie, lui aussi au format JSONL. La boucle ci-dessous distingue trois issues terminales — failed, expired, cancelled — et lève une exception dans chacune : un batch expiré est un batch dont la fenêtre de vingt-quatre heures s’est écoulée sans que tout soit traité, et le traiter comme un succès partiel silencieux serait la pire des options.

import time

def attendre_batch(batch_id: str, intervalle: int = 60) -> dict:
    """Attend la fin d'un batch et retourne les résultats."""
    while True:
        batch = client.batches.retrieve(batch_id)
        print(
            f"Statut: {batch.status} | "
            f"Terminés: {batch.request_counts.completed}/"
            f"{batch.request_counts.total} | "
            f"Échoués: {batch.request_counts.failed}"
        )

        if batch.status == "completed":
            break
        elif batch.status in ("failed", "expired", "cancelled"):
            raise RuntimeError(f"Batch échoué : {batch.status}")

        time.sleep(intervalle)

    # Télécharger les résultats
    contenu = client.files.content(batch.output_file_id)
    resultats = []
    for ligne in contenu.text.strip().split("\n"):
        resultats.append(json.loads(ligne))

    return resultats


resultats = attendre_batch(batch.id)

# Traiter les résultats
for resultat in resultats:
    custom_id = resultat["custom_id"]
    if resultat["response"]["status_code"] == 200:
        texte = resultat["response"]["body"]["output"][0]["content"][0]["text"]
        print(f"{custom_id}: {texte[:100]}...")
    else:
        print(f"{custom_id}: ERREUR {resultat[response][status_code]}")

Prévoir les échecs partiels

Sur un lot volumineux, une partie des requêtes échouera : un document tronqué, un dépassement de contexte, une erreur transitoire côté serveur. Un lot n’est donc jamais simplement « réussi » ou « raté » — il est réussi en partie, et c’est la part manquante qu’il faut savoir isoler. Le pattern à adopter consiste à séparer immédiatement les succès des échecs et à régénérer un fichier JSONL prêt à être relancé, plutôt que d’aller relire les logs trois jours plus tard pour reconstituer ce qui manque.

def traiter_erreurs_batch(resultats: list[dict]) -> tuple[list, list]:
    """Sépare les succès des échecs pour retraitement."""
    succes = []
    echecs = []

    for r in resultats:
        if r["response"]["status_code"] == 200:
            succes.append(r)
        else:
            echecs.append(r)

    if echecs:
        print(f"Attention : {len(echecs)} requêtes échouées sur {len(resultats)}")
        # Réécrire un fichier JSONL avec les échecs pour retry
        with open("batch_retry.jsonl", "w") as f:
            for echec in echecs:
                requete_originale = {
                    "custom_id": echec["custom_id"],
                    "method": "POST",
                    "url": "/v1/responses",
                    "body": echec["request"]["body"],
                }
                f.write(json.dumps(requete_originale) + "\n")

    return succes, echecs

Cumuler les économies

Sur un traitement de dix mille documents avec GPT-5.6 Luna, l’API synchrone représente le coût de référence, la Batch API en représente la moitié, et le prompt caching s’ajoute par-dessus puisque les tokens mis en cache bénéficient eux aussi de leur propre réduction. Les deux mécanismes ne se concurrencent pas : si vos dix mille requêtes partagent le même prompt système long, structurez-le en préfixe stable et vous cumulerez les deux effets.

Vient enfin la question de la découpe. Un lot ne doit pas être infini, et il est plus confortable d’en surveiller plusieurs de taille maîtrisée qu’un seul monolithe dont l’échec coûte une journée entière. La fonction suivante tranche une liste de requêtes en lots de taille bornée.

def creer_batches_optimises(
    requetes: list[dict],
    taille_max: int = 50_000,
) -> list[list[dict]]:
    """Découpe en batches de taille optimale."""
    batches = []
    batch_courant = []

    for requete in requetes:
        batch_courant.append(requete)
        if len(batch_courant) >= taille_max:
            batches.append(batch_courant)
            batch_courant = []

    if batch_courant:
        batches.append(batch_courant)

    return batches

Points clés à retenir

• La Batch API offre 50 % de réduction pour les traitements non temps réel
• Préparez vos requêtes en fichier JSONL avec un custom_id unique par requête
• Prévoyez un mécanisme de retry pour les requêtes échouées
• Combinez avec le prompt caching pour maximiser les économies