Batch Annotations
On this page
Updated November 2, 2026 — OCR. OCR 4.0 (
mistral-ocr-4-0) was retired on September 30, 2026. Use OCR 4.1 (mistral-ocr-4-1), generally available since August 31 at the same price, or the aliasmistral-ocr-latest, which has pointed to it since July 16. OCR 4.1 adds confidence scores at the block level. Addendum November 3, 2026: A local PDF is transmitted in base64 as{"type": "document_url", "document_url": "data:application/pdf;base64,…"}; thedocument_base64type from older examples exists neither in the documentation nor in themistralaiSDK, and those examples would have failed. Sources: OCR Processor · Document QnA. Corrected November 3, 2026: The code previously readpage.document_annotationpage by page; in SDKmistralai3.7.0, document annotation is a single field in the response (response.document_annotation), and image annotation is read fromimage.image_annotation.
Scaling Up
Processing one document at a time works for prototyping. In production, the reality is different: an accounting department receives hundreds of invoices daily, and a law firm digitizes thousands of documents during due diligence. The code must change accordingly—it is no longer just about getting an answer, but ensuring no document gets lost along the way and that you know exactly which ones failed.
Start with Sequential Processing
Keep it simple at first. The loop below iterates through a folder of PDFs, encodes each file in base64, sends it to the OCR engine with the annotation schema, and accumulates the results. The critical part lies in the try/except block: an unreadable invoice or a failed API call must never interrupt the entire batch. We log the error with the filename and continue, leaving you with a final list where every entry has a status. The counter displayed at each iteration isn’t just a nice-to-have; when processing three hundred files, knowing your progress prevents you from restarting a job that is already halfway done.
import os
import json
import time
from pathlib import Path
from mistralai.client import Mistral
from mistralai.extra import response_format_from_pydantic_model
from pydantic import BaseModel, Field
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
class ExtractionFacture(BaseModel):
fournisseur: str = Field(..., description="Supplier name")
numero: str = Field(..., description="Invoice number")
montant_ttc: float = Field(..., description="Total amount including tax")
date: str = Field(..., description="Date (YYYY-MM-DD)")
def traiter_dossier_factures(dossier: str) -> list[dict]:
"""Processes all invoices in a folder."""
resultats = []
chemin = Path(dossier)
fichiers = list(chemin.glob("*.pdf"))
print(f"Processing {len(fichiers)} files...")
for i, fichier in enumerate(fichiers):
print(f" [{i+1}/{len(fichiers)}] {fichier.name}")
try:
import base64
with open(fichier, "rb") as f:
doc_b64 = base64.b64encode(f.read()).decode("utf-8")
response = client.ocr.process(
model="mistral-ocr-latest",
document={
"type": "document_url",
"document_url": f"data:application/pdf;base64,{doc_b64}"
},
document_annotation_format=response_format_from_pydantic_model(
ExtractionFacture
),
table_format="markdown"
)
if response.document_annotation:
facture = ExtractionFacture.model_validate_json(
response.document_annotation
)
resultats.append({
"fichier": fichier.name,
"fournisseur": facture.fournisseur,
"numero": facture.numero,
"montant_ttc": facture.montant_ttc,
"date": facture.date,
"statut": "ok"
})
except Exception as e:
resultats.append({
"fichier": fichier.name,
"statut": "erreur",
"erreur": str(e)
})
return resultatsParallelize Without Saturating the API
Sequential processing spends most of its time waiting on the network. Asynchronous calls eliminate this waste, provided you throttle concurrency: launching three hundred requests at once will trigger rate-limit errors and leave you with a half-finished batch. The asyncio semaphore handles exactly this—max_concurrent=5 means a maximum of five documents in flight at any time, regardless of how many files are queued. The individual function maintains the same discipline as the sequential version: it catches exceptions and returns a status dictionary rather than letting gather crash.
import os
import asyncio
import base64
from pathlib import Path
from mistralai.client import Mistral
from mistralai.extra import response_format_from_pydantic_model
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
async def traiter_document(chemin_fichier: Path, schema) -> dict:
"""Processes a document asynchronously."""
try:
with open(chemin_fichier, "rb") as f:
doc_b64 = base64.b64encode(f.read()).decode("utf-8")
response = client.ocr.process(
model="mistral-ocr-latest",
document={
"type": "document_url",
"document_url": f"data:application/pdf;base64,{doc_b64}"
},
document_annotation_format=response_format_from_pydantic_model(
schema
)
)
annotations = []
if response.document_annotation:
annotations.append(response.document_annotation)
return {
"fichier": chemin_fichier.name,
"annotations": annotations,
"statut": "ok"
}
except Exception as e:
return {
"fichier": chemin_fichier.name,
"statut": "erreur",
"erreur": str(e)
}
async def traiter_batch(
dossier: str,
schema,
max_concurrent: int = 5
) -> list[dict]:
"""Processes a folder with controlled parallelism."""
chemin = Path(dossier)
fichiers = list(chemin.glob("*.pdf"))
semaphore = asyncio.Semaphore(max_concurrent)
async def traiter_avec_limite(fichier):
async with semaphore:
return await traiter_document(fichier, schema)
taches = [traiter_avec_limite(f) for f in fichiers]
resultats = await asyncio.gather(*taches)
return list(resultats)
# Execution
# résultats = asyncio.run(traiter_batch("./factures/", ExtractionFacture))When Volume Justifies Batch Inference
Beyond a certain threshold, the question shifts from speed to cost. Mistral offers a dedicated Batch Inference service that is more economical than individual calls. You submit a JSONL file containing one request per line. The custom_id field—here, the filename without extension—allows you to map each response back to its original document when the batch completes.
import json
def preparer_batch_jsonl(dossier: str, sortie: str) -> int:
"""Prepares a JSONL file for batch processing."""
chemin = Path(dossier)
fichiers = list(chemin.glob("*.pdf"))
compte = 0
with open(sortie, "w") as f:
for fichier in fichiers:
with open(fichier, "rb") as pdf:
doc_b64 = base64.b64encode(pdf.read()).decode("utf-8")
requete = {
"custom_id": fichier.stem,
"model": "mistral-ocr-latest",
"document": {
"type": "document_url",
"document_url": f"data:application/pdf;base64,{doc_b64}"
}
}
f.write(json.dumps(requete) + "\n")
compte += 1
print(f"{compte} requests prepared in {sortie}")
return compteTracking What Happened
Mass processing without metrics is a black box. The small dataclass below counts successes and errors, calculates elapsed time, and derives a speed in documents per second. This speed tells you whether your max_concurrent setting is optimal, while the list of failed files feeds your incident recovery process.
import time
from dataclasses import dataclass, field
@dataclass
class StatsBatch:
total: int = 0
succes: int = 0
erreurs: int = 0
temps_debut: float = field(default_factory=time.time)
fichiers_erreur: list = field(default_factory=list)
@property
def duree(self) -> float:
return time.time() - self.temps_debut
@property
def vitesse(self) -> float:
if self.duree > 0:
return self.succes / self.duree
return 0.0
def rapport(self) -> str:
return (
f"Processing completed in {self.duree:.1f}s\n"
f" Successes: {self.succes}/{self.total}\n"
f" Errors: {self.erreurs}\n"
f" Speed: {self.vitesse:.2f} docs/s\n"
f" Failed files: {self.fichiers_erreur}"
)Finally, export the script results. CSV goes to the accounting department’s spreadsheet, JSON to your application; in both cases, ensure UTF-8 encoding and ensure_ascii=False, otherwise your accented characters will appear as unreadable sequences to the recipient.
import csv
import json
def exporter_csv(resultats: list[dict], chemin: str) -> None:
"""Exports results to CSV."""
if not resultats:
return
cles = resultats[0].keys()
with open(chemin, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=cles)
writer.writeheader()
writer.writerows(resultats)
print(f"CSV export: {chemin}")
def exporter_json(resultats: list[dict], chemin: str) -> None:
"""Exports results to JSON."""
with open(chemin, "w", encoding="utf-8") as f:
json.dump(resultats, f, ensure_ascii=False, indent=2)
print(f"JSON export: {chemin}")Key Takeaways
- Sequential processing suits small volumes with robust error handling
- Parallel processing with
asyncioand semaphores accelerates large volumes - Mistral’s Batch Inference service is the most cost-effective solution for very large volumes
- Add monitoring and reporting to track progress and identify errors
- Export results to CSV or JSON for integration with your systems