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

Votre fichier JSONL est préparé et uploadé, vous disposez d’un file_id. Reste à transformer ce fichier dormant en résultats exploitables : créer le job batch, suivre sa progression pendant les minutes ou les heures que dure le traitement, puis récupérer et raccrocher les réponses. Cette leçon parcourt le code Python de bout en bout, dans l’ordre où vous l’écrirez.

Créer un job batch

La création tient en un seul appel. Vous désignez le fichier source, le modèle 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 méritent chacun une remarque :

• 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.

La contrainte d’un seul modèle par job structure votre découpage : si vous voulez comparer deux modèles sur le même corpus, vous soumettez deux jobs distincts pointant vers le même fichier. Quant aux métadonnées, elles paraissent accessoires jusqu’au jour où vous avez trente jobs en historique ; le couple projet/version affiché ci-dessus suffit alors à retrouver en trois secondes le batch qui a produit les données douteuses.

Les statuts d’un job batch

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

Quatre de ces sept états sont terminaux — SUCCESS, FAILED, TIMEOUT_EXCEEDED et CANCELLED — et c’est cette distinction qui commande la boucle de suivi : tant que le statut n’appartient pas à cet ensemble, le job vit encore et il faut attendre. CANCELLATION_REQUESTED, en particulier, n’est pas un état d’arrêt mais un état de transition, et un code qui l’interpréterait comme terminal sortirait de la boucle trop tôt.

Suivre la progression

Le pattern de polling ci-dessous encapsule exactement cette logique, avec un garde-fou de durée maximale pour éviter qu’un script oublié tourne indéfiniment :

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}")

L’intervalle de dix secondes convient à un batch de quelques minutes ; pour un traitement nocturne de plusieurs centaines de milliers de lignes, montez-le à une minute ou davantage, la précision du suivi n’apporte rien et les appels s’accumulent inutilement.

Télécharger les résultats

Une fois le statut SUCCESS atteint, le job expose un output_file que vous téléchargez et parsez ligne à ligne :

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, et chaque ligne porte le custom_id d’origine à côté de la réponse complète du modèle. C’est ici que l’effort de nommage de la leçon précédente est récompensé : l’indexation par custom_id dans un dictionnaire suffit à réconcilier les réponses avec vos documents sources, quel que soit l’ordre dans lequel elles reviennent. L’affichage des trois premières entrées tronquées à deux cents caractères est un réflexe de contrôle utile avant de lancer l’écriture en base.

Lister et gérer vos jobs

L’inventaire des jobs se filtre par statut, ce qui permet de savoir à tout moment ce qui tourne réellement sur votre compte :

# 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}")

Et lorsqu’un job s’éternise ou qu’une erreur de prompt a été repérée après soumission, l’annulation évite de payer un traitement inutile :

# 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. Ne considérez donc pas la ressource comme libérée dès le retour de l’appel.

Workflow complet de bout en bout

Assemblées, les quatre étapes forment un script court que vous pouvez placer tel quel dans un ordonnanceur :

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}")

Reprenez ce squelette sur votre propre corpus avec une dizaine de lignes seulement : vous validerez l’enchaînement complet en quelques minutes, avant d’engager le volume réel.

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