Guide de démarrage : exécuter un job batch

Dans ce guide de démarrage, vous écrivez deux requêtes de chat dans un fichier, les exécutez en un seul job batch, attendez la fin du job et lisez les réponses. Les jobs batch coûtent moins cher que les mêmes requêtes envoyées une par une, et ils conviennent aux tâches qui n'ont pas besoin d'une réponse immédiate. L'écriture des scripts prend environ 5 minutes ; le job lui-même se termine généralement en quelques minutes, mais peut prendre plus de temps lorsque la plateforme est occupée. Pour connaître les limites et les tarifs, consultez Traitement batch.

Avant de commencer

Une clé API Mistral. Pour en créer une, consultez Créer votre première clé API.
Python 3.10 ou version ultérieure et uv.

Configurer votre environnement

Créez un projet, installez le SDK Mistral et exportez votre clé API :

uv init batch-job --python 3.12 && cd batch-job
uv add mistralai
export MISTRAL_API_KEY="your-api-key"

uv init batch-job --python 3.12 && cd batch-job
uv add mistralai
export MISTRAL_API_KEY="your-api-key"

1. Créer le job

Créez create_job.py. Il écrit une requête par ligne dans requests.jsonl, télécharge le fichier et crée le job :

import json
import os

from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

questions = {
    "q1": "Translate 'good morning' into French. Reply with the translation only.",
    "q2": "Translate 'thank you' into Spanish. Reply with the translation only.",
}
with open("requests.jsonl", "w") as f:
    for custom_id, question in questions.items():
        request = {
            "custom_id": custom_id,
            "body": {"max_tokens": 50, "messages": [{"role": "user", "content": question}]},
        }
        f.write(json.dumps(request) + "\n")

with open("requests.jsonl", "rb") as f:
    batch_file = client.files.upload(
        file={"file_name": "requests.jsonl", "content": f}, purpose="batch"
    )

job = client.batch.jobs.create(
    input_files=[batch_file.id],
    model="mistral-small-latest",
    endpoint="/v1/chat/completions",
)
print(f"Job ID: {job.id}, status: {job.status}")

import json
import os

from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

questions = {
    "q1": "Translate 'good morning' into French. Reply with the translation only.",
    "q2": "Translate 'thank you' into Spanish. Reply with the translation only.",
}
with open("requests.jsonl", "w") as f:
    for custom_id, question in questions.items():
        request = {
            "custom_id": custom_id,
            "body": {"max_tokens": 50, "messages": [{"role": "user", "content": question}]},
        }
        f.write(json.dumps(request) + "\n")

with open("requests.jsonl", "rb") as f:
    batch_file = client.files.upload(
        file={"file_name": "requests.jsonl", "content": f}, purpose="batch"
    )

job = client.batch.jobs.create(
    input_files=[batch_file.id],
    model="mistral-small-latest",
    endpoint="/v1/chat/completions",
)
print(f"Job ID: {job.id}, status: {job.status}")

Exécutez-le, puis exportez l'ID du job qu'il affiche :

uv run create_job.py
export JOB_ID="<job-id>"

uv run create_job.py
export JOB_ID="<job-id>"

2. Attendre et lire les réponses

Créez read_results.py. Il vérifie le statut du job toutes les 10 secondes, puis télécharge le fichier de sortie et affiche une réponse par requête :

import json
import os
import time

from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

job = client.batch.jobs.get(job_id=os.environ["JOB_ID"])
while job.status in ("QUEUED", "RUNNING"):
    print(f"Status: {job.status}, waiting...")
    time.sleep(10)
    job = client.batch.jobs.get(job_id=job.id)

print(f"Status: {job.status}")
output = client.files.download(file_id=job.output_file)
for line in output.read().decode().splitlines():
    result = json.loads(line)
    answer = result["response"]["body"]["choices"][0]["message"]["content"]
    print(f"{result['custom_id']}: {answer}")

import json
import os
import time

from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

job = client.batch.jobs.get(job_id=os.environ["JOB_ID"])
while job.status in ("QUEUED", "RUNNING"):
    print(f"Status: {job.status}, waiting...")
    time.sleep(10)
    job = client.batch.jobs.get(job_id=job.id)

print(f"Status: {job.status}")
output = client.files.download(file_id=job.output_file)
for line in output.read().decode().splitlines():
    result = json.loads(line)
    answer = result["response"]["body"]["choices"][0]["message"]["content"]
    print(f"{result['custom_id']}: {answer}")

Exécutez-le :

uv run read_results.py

uv run read_results.py

Résultat attendu

Le statut pendant l'exécution du job, puis une réponse par custom_id. La formulation varie d'une exécution à l'autre :

Status: RUNNING, waiting...
Status: RUNNING, waiting...
Status: SUCCESS
q1: Bonjour
q2: Gracias

Status: RUNNING, waiting...
Status: RUNNING, waiting...
Status: SUCCESS
q1: Bonjour
q2: Gracias

Si le statut final est FAILED, consultez le fichier error_file du job de la même manière que le fichier de sortie.

Nettoyage

Les fichiers d'entrée et de sortie restent dans votre espace de travail après la fin du job. Pour les supprimer, créez clean_up.py :

import os

from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

job = client.batch.jobs.get(job_id=os.environ["JOB_ID"])
for file_id in [*job.input_files, job.output_file]:
    client.files.delete(file_id=file_id)
print("Deleted the input and output files.")

import os

from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

job = client.batch.jobs.get(job_id=os.environ["JOB_ID"])
for file_id in [*job.input_files, job.output_file]:
    client.files.delete(file_id=file_id)
print("Deleted the input and output files.")

Exécutez-le :

uv run clean_up.py

uv run clean_up.py

Aller plus loin

Traitement batch

Guide de démarrage : obtenir une sortie structurée

Priority Tier