Quickstart: run a batch job
In this quickstart, you write two chat requests to a file, run them as one batch job, wait for the job to finish, and read the answers. Batch jobs cost less than the same requests sent one by one, and they suit work that doesn't need an answer right away. Writing the scripts takes about 5 minutes; the job itself usually finishes within minutes, but can take longer when the platform is busy. For limits and pricing, see Batch processing.
Before you begin
- A Mistral API key. To create one, see Create your first API key.
- Python 3.10 or later and uv.
Set up your environment
Create a project, install the Mistral SDK, and export your API key:
uv init batch-job --python 3.12 && cd batch-job
uv add mistralai
export MISTRAL_API_KEY="your-api-key"1. Create the job
Create create_job.py. It writes one request per line to requests.jsonl, uploads the file, and creates the 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}")Run it, then export the job ID it prints:
uv run create_job.py
export JOB_ID="<job-id>"2. Wait and read the answers
Create read_results.py. It checks the job every 10 seconds, then downloads the output file and prints one answer per request:
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}")Run it:
uv run read_results.pyYou should see
The status while the job runs, then one answer per custom_id. The wording varies between runs:
Status: RUNNING, waiting...
Status: RUNNING, waiting...
Status: SUCCESS
q1: Bonjour
q2: GraciasIf the status ends as FAILED, check the job's error_file in the same way as the output file.
Clean up
The input and output files stay in your workspace after the job ends. To delete them, create 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.")Run it:
uv run clean_up.py