La Batch API de Gemini : un fichier JSONL, moitié prix

Synchrone ou en lot ?

Jusqu'ici, classer.py envoie les requêtes en direct (mode synchrone) : on obtient la réponse tout de suite, mais on gère soi-même concurrence, débit et réessais. La Batch API inverse la logique : vous envoyez toutes les requêtes d'un coup dans un fichier, et Google se charge de l'ordonnancement.

sequenceDiagram
  participant S as Script
  participant F as API Files
  participant B as Batch API
  S->>S: construire requetes_batch.jsonl
  S->>F: files.upload
  F-->>S: files/xyz
  S->>B: batches.create(model, src="files/xyz")
  B-->>S: batches/abc (PENDING)
  loop toutes les 30 s
    S->>B: batches.get
    B-->>S: RUNNING puis SUCCEEDED
  end
  S->>F: files.download(dest.file_name)
  S->>S: écrire resultats_batch.jsonl

	Synchrone (`classer.py`)	Batch API
Délai de réponse	secondes	différé : objectif de 24 h, souvent bien moins
Prix	tarif standard	la moitié du tarif standard (page Tarifs)
Concurrence, débit, réessais	à votre charge	gérés par Google
Taille	limitée par vos quotas	fichier d'entrée jusqu'à 2 Go
Bon pour	interactif, petits volumes, besoin immédiat	gros volumes non urgents : nuit, historique, réindexation

Le fichier JSONL

Chaque ligne contient une clé choisie par vous et une requête complète au format REST de generateContent :

files/batch_api.py

def requete(lot: list[dict]) -> dict:
    return {
        "contents": [{"role": "user", "parts": [{"text": prompt_lot(lot)}]}],
        "systemInstruction": {"parts": [{"text": SYSTEM}]},
        "generationConfig": {"temperature": 0, "responseMimeType": "application/json",
                             "responseJsonSchema": SCHEMA},
    }

ligne = {"key": f"lot-{i}", "request": requete(lot)}

On réutilise le même SCHEMA, la même instruction et le même prompt_lot que le mode synchrone : les deux chemins produisent des résultats comparables.

requetes_batch.jsonl (début de la 1re ligne)

{"key": "lot-0", "request": {"contents": [{"role": "user", "parts": [{"text": "[A0001] Génial, mon grille-pain Nimbus est arrivé avec trois semaines de retard. Vraiment bravo.\n[A0002] ...

Les réponses d'un job de la Batch API ne reviennent pas forcément dans l'ordre des requêtes. Comment retrouver à quel lot correspond chaque réponse ?

Envoyer, attendre, récupérer

files/batch_api.py

fichier = client.files.upload(file="requetes_batch.jsonl",
                              config=types.UploadFileConfig(display_name="avis-lot", mime_type="jsonl"))
job = client.batches.create(model=os.environ.get("MODEL", "gemini-3.5-flash-lite"), src=fichier.name,
                            config={"display_name": "lab-batch-ia-avis"})
job = attendre(client, job.name, max_s)        # batches.get toutes les 30 s

contenu = client.files.download(file=job.dest.file_name).decode("utf-8")
for ligne in contenu.splitlines():
    r = json.loads(ligne)
    lot = lots[int(r["key"].split("-")[1])]  # la clé relie chaque réponse à sa requête
    if "response" in r:
        ecrire_lot(sortie, lire_reponse(lot, r["response"]))

Les états possibles d'un job : JOB_STATE_PENDING, JOB_STATE_RUNNING, puis l'un des états finaux JOB_STATE_SUCCEEDED, JOB_STATE_FAILED, JOB_STATE_CANCELLED, JOB_STATE_EXPIRED (un job resté plus de 48 heures en attente ou en cours expire).

Nettoyer

Le fichier d'entrée est supprimé par le script (client.files.delete) dès que le job est terminé. Le job lui-même peut être supprimé avec client.batches.delete(name=...) ; les fichiers de l'API Files sont de toute façon supprimés automatiquement après un délai (voir la documentation).

Vous devez classer 2 millions d'avis historiques d'ici la semaine prochaine. Quelle approche choisir en priorité ?

Exercice : reprendre un job

Votre script a été coupé alors que le job batches/abc123 tournait. Écrivez les quelques lignes qui récupèrent ses résultats sans relancer de job.

Voir la solution

from google import genai
import json
from batch_api import lire_reponse
from classer import ecrire_lot, lire_avis

client = genai.Client()
job = client.batches.get(name="batches/abc123")
print(job.state.name)
if job.state.name == "JOB_STATE_SUCCEEDED":
    avis = lire_avis("avis.csv")[:20]
    lots = [avis[i:i + 10] for i in range(0, len(avis), 10)]   # la même découpe qu'à l'envoi
    for ligne in client.files.download(file=job.dest.file_name).decode("utf-8").splitlines():
        r = json.loads(ligne)
        if "response" in r:
            ecrire_lot("resultats_batch.jsonl", lire_reponse(lots[int(r["key"].split("-")[1])], r["response"]))

Point clé : il faut reconstruire exactement la même découpe en lots (mêmes --n et --lot) pour que les clés correspondent. Une alternative plus robuste : mettre directement les ids dans la clé ("key": "A0001-A0010").

Résultat obtenu

Exécution réelle du 2026-10-15 : 20 avis, 2 requêtes de 10, gemini-3.5-flash-lite.

$ python batch_api.py gemini --n 20
Job créé : batches/o4d9fd5c…
  16:40:57 JOB_STATE_RUNNING
  16:41:28 JOB_STATE_RUNNING
  ...
  16:43:29 JOB_STATE_SUCCEEDED
$ python rapport.py --resultats resultats_batch.jsonl
Avis classés : 20/200  (manquants : 180 ['A0021', 'A0022', ...])
Accord avec les réponses attendues : sentiment 95%, catégorie 100% ...
Jetons : 694 en entrée, 941 en sortie

Environ 3 minutes pour ce tout petit job (un délai indicatif, pas une garantie). Le job a ensuite été supprimé (batches.delete, vérifié par un batches.get qui renvoie 404).

Je sais faire

Choisir entre traitement synchrone et Batch API selon le volume et l'urgence
Construire un fichier JSONL de requêtes avec des clés
Envoyer, suivre et récupérer un job avec files.upload, batches.create, batches.get
Gérer les erreurs ligne par ligne et nettoyer fichiers et job