Créer un Job Batch, Tracker le Statut et Télécharger les Résultats

Du fichier JSONL au résultat : le cycle de vie complet

Vous avez préparé et uploadé votre fichier JSONL. Il est temps de créer un job batch, de suivre sa progression, et de récupérer les résultats. Cette leçon vous guide à travers le code Python complet pour orchestrer ce workflow.

Créer un job batch

La création d’un job batch se fait en un seul appel API. Vous spécifiez le fichier source, le modèle à utiliser, et l’endpoint cible :

from mistralai import Mistral
import time

client = Mistral(api_key="votre-clé-api")

# Créer le job batch
job = client.batch.jobs.create(
    input_files=[batch_file.id],   # ID du fichier uploadé
    model="mistral-small-latest",
    endpoint="/v1/chat/completions",
    metadata={"project": "analyse-documents", "version": "1.0"}
)

print(f"Job créé : {job.id}")
print(f"Statut initial : {job.status}")

Les paramètres de création

• input_files — Liste d’identifiants de fichiers uploadés. Vous pouvez passer plusieurs fichiers pour un même job.
• model — Le modèle Mistral à utiliser. Un seul modèle par job batch.
• endpoint — L’endpoint API cible (doit correspondre au format du body dans votre JSONL).
• metadata — Dictionnaire optionnel de métadonnées custom pour organiser et retrouver vos jobs.

Les statuts d’un job batch

Un job batch traverse plusieurs états au cours de son cycle de vie :

Suivre la progression

Voici un pattern robuste pour suivre la progression d’un job jusqu’à sa complétion :

def wait_for_batch(client, job_id, poll_interval=10, max_wait=3600):
    """Attend la fin d'un job batch avec polling."""
    elapsed = 0
    terminal_states = {"SUCCESS", "FAILED", "TIMEOUT_EXCEEDED", "CANCELLED"}

    while elapsed < max_wait:
        job = client.batch.jobs.get(job_id=job_id)
        print(f"[{elapsed}s] Statut : {job.status}")

        if job.status in terminal_states:
            return job

        time.sleep(poll_interval)
        elapsed += poll_interval

    raise TimeoutError(f"Job {job_id} toujours en cours après {max_wait}s")

# Utilisation
completed_job = wait_for_batch(client, job.id)
print(f"Résultat final : {completed_job.status}")

Télécharger les résultats

Une fois le job terminé avec le statut SUCCESS, vous pouvez télécharger les résultats :

import json

if completed_job.status == "SUCCESS":
    # Récupérer l'ID du fichier de résultats
    output_file_id = completed_job.output_file

    # Télécharger le contenu
    result_data = client.files.download(file_id=output_file_id)

    # Parser les résultats JSONL
    results = {}
    for line in result_data.decode("utf-8").strip().split("\n"):
        entry = json.loads(line)
        custom_id = entry["custom_id"]
        response = entry["response"]
        results[custom_id] = response

    # Afficher un exemple
    for cid, resp in list(results.items())[:3]:
        content = resp["body"]["choices"][0]["message"]["content"]
        print(f"\n--- {cid} ---")
        print(content[:200])

Le fichier de résultats est lui aussi au format JSONL. Chaque ligne contient le custom_id d’origine et la réponse complète du modèle, ce qui vous permet de raccrocher facilement chaque résultat à votre requête initiale.

Lister et gérer vos jobs

Lister les jobs en cours

# Lister tous les jobs en cours d'exécution
running_jobs = client.batch.jobs.list(status="RUNNING")
for j in running_jobs.data:
    print(f"Job {j.id} - Modèle: {j.model} - Créé: {j.created_at}")

Annuler un job

# Annuler un job qui prend trop de temps
client.batch.jobs.cancel(job_id="job-abc123")

L’annulation est asynchrone : le statut passe d’abord à CANCELLATION_REQUESTED, puis à CANCELLED une fois l’arrêt effectif.

Workflow complet de bout en bout

Voici le script complet qui enchaîne toutes les étapes :

from mistralai import Mistral
import json
import time

client = Mistral(api_key="votre-clé-api")

# 1. Upload du fichier
with open("batch_requests.jsonl", "rb") as f:
    batch_file = client.files.upload(
        file=("batch_requests.jsonl", f),
        purpose="batch"
    )

# 2. Créer le job
job = client.batch.jobs.create(
    input_files=[batch_file.id],
    model="mistral-small-latest",
    endpoint="/v1/chat/completions"
)

# 3. Attendre la complétion
completed = wait_for_batch(client, job.id)

# 4. Télécharger et parser les résultats
if completed.status == "SUCCESS":
    data = client.files.download(file_id=completed.output_file)
    for line in data.decode("utf-8").strip().split("\n"):
        entry = json.loads(line)
        print(f"{entry['custom_id']}: OK")
else:
    print(f"Échec du batch : {completed.status}")

Points clés à retenir

• Un job batch se crée avec client.batch.jobs.create() en spécifiant fichier, modèle et endpoint
• Sept statuts possibles, dont trois terminaux : SUCCESS, FAILED, CANCELLED
• Le polling avec intervalle est le pattern standard pour suivre la progression
• Les résultats sont au format JSONL, raccordés aux requêtes via custom_id
• Vous pouvez lister, inspecter et annuler vos jobs à tout moment