Annotations en batch

Passer à l’échelle

Traiter un document à la fois convient pour le prototypage. En production, vous devez traiter des centaines ou des milliers de documents par jour. Cette leçon couvre les stratégies de traitement par lots avec les annotations Document AI.

Traitement séquentiel avec gestion d’erreurs

La première approche, simple et robuste :

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 resultats

Traitement parallèle avec asyncio

Pour accélérer le traitement, utilisez les appels asynchrones :

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
# resultats = asyncio.run(traiter_batch("./factures/", ExtractionFacture))

Service Batch Inference de Mistral

Pour les très gros volumes, Mistral propose un service de Batch Inference dédié, plus économique que les appels individuels :

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 compte

Suivi et reporting

Pour un traitement en production, ajoutez du monitoring :

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

Export des résultats

Exportez vos résultats vers les formats courants :

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 asyncio et 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