Annotations en batch
Mis à jour le 29 juillet 2026
Passer à l’échelle
Traiter un document à la fois convient pour le prototypage. En production, la réalité est différente : un service comptable reçoit des centaines de factures par jour, un cabinet juridique numérise des milliers de pièces lors d’une due diligence. Le code change alors de nature — il ne s’agit plus d’obtenir une réponse, mais de garantir qu’aucun document ne se perd en route et que vous savez lesquels ont échoué.
Le traitement séquentiel, d’abord
Commencez simple. La boucle ci-dessous parcourt un dossier de PDF, encode chaque fichier en base64, l’envoie à l’OCR avec le schéma d’annotation, puis accumule les résultats. L’essentiel se joue dans le try/except : une facture illisible ou un appel qui échoue ne doit jamais interrompre le lot. On enregistre l’erreur avec le nom du fichier et on continue, ce qui vous laisse à la fin une liste où chaque entrée porte un statut. Le compteur affiché à chaque itération n’est pas un détail de confort : sur trois cents fichiers, savoir où l’on en est évite de relancer un traitement déjà à moitié terminé.
import os
import json
import time
from pathlib import Path
from mistralai 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="Nom du fournisseur")
numero: str = Field(..., description="Numéro de facture")
montant_ttc: float = Field(..., description="Montant TTC")
date: str = Field(..., description="Date (YYYY-MM-DD)")
def traiter_dossier_factures(dossier: str) -> list[dict]:
"""Traite toutes les factures d'un dossier."""
resultats = []
chemin = Path(dossier)
fichiers = list(chemin.glob("*.pdf"))
print(f"Traitement de {len(fichiers)} fichiers...")
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_base64",
"document_base64": doc_b64
},
document_annotation_format=response_format_from_pydantic_model(
ExtractionFacture
),
table_format="markdown"
)
for page in response.pages:
if page.document_annotation:
facture = ExtractionFacture.model_validate_json(
page.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 resultatsParalléliser sans saturer l’API
Le séquentiel passe l’essentiel de son temps à attendre le réseau. Les appels asynchrones règlent ce gaspillage, à condition de brider la concurrence : lancer trois cents requêtes d’un coup vous vaudra des erreurs de limitation de débit et un lot à moitié perdu. Le sémaphore d’asyncio sert exactement à cela — max_concurrent=5 signifie cinq documents en vol au maximum, quel que soit le nombre de fichiers en attente. La fonction unitaire conserve la même discipline que la version séquentielle : elle rattrape l’exception et renvoie un dictionnaire de statut plutôt que de faire tomber gather.
import os
import asyncio
import base64
from pathlib import Path
from mistralai 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:
"""Traite un document de manière asynchrone."""
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_base64",
"document_base64": doc_b64
},
document_annotation_format=response_format_from_pydantic_model(
schema
)
)
annotations = []
for page in response.pages:
if page.document_annotation:
annotations.append(page.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]:
"""Traite un dossier avec parallélisme contrôlé."""
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)
# Exécution
# résultats = asyncio.run(traiter_batch("./factures/", ExtractionFacture))Quand le volume justifie le Batch Inference
Au-delà d’un certain seuil, la question n’est plus la vitesse mais le coût. Mistral propose un service de Batch Inference dédié, plus économique que les appels individuels, auquel vous soumettez un fichier JSONL contenant une requête par ligne. Le champ custom_id — ici le nom du fichier sans extension — vous permettra de rattacher chaque réponse à son document d’origine lorsque le lot reviendra.
import json
def preparer_batch_jsonl(dossier: str, sortie: str) -> int:
"""Prépare un fichier JSONL pour le 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_base64",
"document_base64": doc_b64
}
}
f.write(json.dumps(requete) + "\n")
compte += 1
print(f"{compte} requêtes préparées dans {sortie}")
return compteSavoir ce qui s’est passé
Un traitement de masse sans mesure est une boîte noire. La petite dataclasse qui suit compte les succès et les erreurs, calcule la durée écoulée et en déduit une vitesse en documents par seconde. C’est cette vitesse qui vous dira si votre max_concurrent est bien réglé, et la liste des fichiers en erreur qui alimentera votre reprise sur incident.
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"Traitement terminé en {self.duree:.1f}s\n"
f" Succès : {self.succes}/{self.total}\n"
f" Erreurs : {self.erreurs}\n"
f" Vitesse : {self.vitesse:.2f} docs/s\n"
f" Fichiers en erreur : {self.fichiers_erreur}"
)Reste à sortir les résultats du script. Le CSV ira vers le tableur du service comptable, le JSON vers votre application ; dans les deux cas, pensez à l’encodage UTF-8 et à ensure_ascii=False, faute de quoi vos accents ressortiront en séquences illisibles chez le destinataire.
import csv
import json
def exporter_csv(resultats: list[dict], chemin: str) -> None:
"""Exporte les résultats en 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"Export CSV : {chemin}")
def exporter_json(resultats: list[dict], chemin: str) -> None:
"""Exporte les résultats en JSON."""
with open(chemin, "w", encoding="utf-8") as f:
json.dump(resultats, f, ensure_ascii=False, indent=2)
print(f"Export JSON : {chemin}")Points clés à retenir
- Le traitement séquentiel convient pour les petits volumes avec une bonne gestion d’erreurs
- Le traitement parallèle avec
asyncioet sémaphore accélère les gros volumes - Le service Batch Inference de Mistral est la solution la plus économique pour les très gros volumes
- Ajoutez du monitoring et du reporting pour suivre la progression et identifier les erreurs
- Exportez les résultats en CSV ou JSON pour l’intégration avec vos systèmes