Pipeline de production

De la preuve de concept à la production

Vous avez appris à utiliser l’OCR, les annotations et le QnA de Document AI. Il reste l’écart le plus coûteux à franchir : celui qui sépare un script qui marche sur votre poste d’un service dont vos collègues dépendent tous les matins. Cette dernière leçon rassemble les briques précédentes dans un pipeline robuste, avec reprise sur erreur, monitoring et maîtrise des coûts.

Les six composants d’un pipeline complet

1. File d’attente d’ingestion : réception et priorisation des documents
2. Pré-traitement : validation, conversion de format, amélioration d’image
3. OCR + Annotations : extraction du texte et des données structurées
4. Post-traitement : validation, nettoyage, enrichissement
5. Stockage : base de données, vector store, archivage
6. Interface de requête : API QnA, dashboard, intégrations

Le code de cette leçon couvre le cœur de la chaîne — les étapes 3 et 4 — parce que c’est là que se concentrent les difficultés d’exploitation.

Une classe qui encapsule le traitement

La classe PipelineDocumentAI fixe une configuration une fois pour toutes — le schéma d’annotation, le format des tableaux, le nombre de tentatives — puis l’applique à chaque document. Trois mécanismes méritent votre attention. Le retry avec backoff exponentiel (time.sleep(2 ** tentative)) espace les tentatives d’une, puis deux, puis quatre secondes, ce qui laisse le temps à une saturation passagère de l’API de se résorber au lieu de l’aggraver. Le logging trace chaque succès avec son nombre de pages et sa durée, et chaque échec avec son numéro de tentative : sans ces lignes, vous découvrirez les incidents par un appel du service comptable. Le dictionnaire stats, enfin, alimente le rapport de fin de lot.

Notez que la méthode renvoie toujours un ResultatTraitement, même après épuisement des tentatives : l’appelant n’a jamais à gérer d’exception, il lit le champ erreur.

import os
import time
import base64
import logging
from pathlib import Path
from dataclasses import dataclass, field
from mistralai import Mistral
from mistralai.extra import response_format_from_pydantic_model
from pydantic import BaseModel, Field
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("document_ai_pipeline")

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


@dataclass
class ResultatTraitement:
    fichier: str
    pages: int = 0
    texte: str = ""
    annotations: Optional[dict] = None
    duree_secondes: float = 0.0
    erreur: Optional[str] = None


class PipelineDocumentAI:
    def __init__(
        self,
        schema_annotation=None,
        table_format: str = "markdown",
        max_retries: int = 3,
    ):
        self.schema_annotation = schema_annotation
        self.table_format = table_format
        self.max_retries = max_retries
        self.stats = {"total": 0, "succes": 0, "erreurs": 0}

    def traiter_document(self, chemin: str) -> ResultatTraitement:
        """Traite un document avec retry et monitoring."""
        debut = time.time()
        self.stats["total"] += 1

        for tentative in range(self.max_retries):
            try:
                # Lecture et encodage
                with open(chemin, "rb") as f:
                    doc_b64 = base64.b64encode(f.read()).decode("utf-8")

                # Paramètres de l'appel
                params = {
                    "model": "mistral-ocr-latest",
                    "document": {
                        "type": "document_base64",
                        "document_base64": doc_b64
                    },
                    "table_format": self.table_format,
                }

                if self.schema_annotation:
                    params["document_annotation_format"] = (
                        response_format_from_pydantic_model(
                            self.schema_annotation
                        )
                    )

                # Appel OCR
                response = client.ocr.process(**params)

                # Extraction du texte
                texte = "\n\n".join(
                    p.markdown for p in response.pages
                )

                # Extraction des annotations
                annotations = None
                if self.schema_annotation:
                    for page in response.pages:
                        if page.document_annotation:
                            annotations = page.document_annotation
                            break

                duree = time.time() - debut
                self.stats["succes"] += 1

                logger.info(
                    f"OK: {chemin} ({len(response.pages)} pages, "
                    f"{duree:.1f}s)"
                )

                return ResultatTraitement(
                    fichier=chemin,
                    pages=len(response.pages),
                    texte=texte,
                    annotations=annotations,
                    duree_secondes=duree,
                )

            except Exception as e:
                logger.warning(
                    f"Tentative {tentative + 1}/{self.max_retries} "
                    f"échouée pour {chemin}: {e}"
                )
                if tentative < self.max_retries - 1:
                    time.sleep(2 ** tentative)  # Backoff exponentiel

        # Toutes les tentatives échouées
        self.stats["erreurs"] += 1
        return ResultatTraitement(
            fichier=chemin,
            duree_secondes=time.time() - debut,
            erreur="Échec après toutes les tentatives",
        )

    def traiter_dossier(self, dossier: str) -> list[ResultatTraitement]:
        """Traite tous les PDF d'un dossier."""
        chemin = Path(dossier)
        fichiers = sorted(chemin.glob("*.pdf"))
        resultats = []

        for fichier in fichiers:
            resultat = self.traiter_document(str(fichier))
            resultats.append(resultat)

        return resultats

    def rapport(self) -> str:
        """Génère un rapport de traitement."""
        return (
            f"Documents traités : {self.stats['total']}\n"
            f"Succès : {self.stats['succes']}\n"
            f"Erreurs : {self.stats['erreurs']}\n"
            f"Taux de succès : "
            f"{self.stats['succes']/max(self.stats['total'],1)*100:.1f}%"
        )

À l’usage, la même classe couvre les deux besoins : instanciée sans schéma, elle se contente d’extraire le texte ; instanciée avec un modèle Pydantic, elle produit en plus les données structurées.

# Pipeline simple (OCR seul)
pipeline = PipelineDocumentAI(table_format="markdown")
resultats = pipeline.traiter_dossier("./factures/")
print(pipeline.rapport())

# Pipeline avec annotations
from pydantic import BaseModel, Field

class ExtractionFacture(BaseModel):
    fournisseur: str = Field(..., description="Nom du fournisseur")
    montant_ttc: float = Field(..., description="Montant TTC")
    date: str = Field(..., description="Date (YYYY-MM-DD)")

pipeline_annote = PipelineDocumentAI(
    schema_annotation=ExtractionFacture,
    table_format="markdown"
)
resultats = pipeline_annote.traiter_dossier("./factures/")

Anticiper la facture

La maîtrise des coûts est essentielle en production, et elle commence par comprendre ce qui est facturé.

Le fait que les annotations n’entraînent aucun surcoût change la conception : autant extraire les métadonnées dès le passage OCR plutôt que de repasser un LLM derrière. La fonction d’estimation ci-dessous vous permet de chiffrer un projet avant de le lancer ; ses tarifs sont des ordres de grandeur à ajuster, la table de référence restant celle publiée par Mistral.

def estimer_cout(
    nombre_pages: int,
    avec_qna: bool = False,
    modele_qna: str = "mistral-small-latest"
) -> dict:
    """Estime le coût de traitement."""

    # Tarifs approximatifs (vérifier les tarifs actuels)
    cout_ocr_par_page = 0.001  # Ajuster selon le tier
    cout_qna_par_1k_tokens = {
        "mistral-small-latest": 0.001,
        "mistral-large-latest": 0.008,
    }

    cout_ocr = nombre_pages * cout_ocr_par_page
    cout_qna = 0.0

    if avec_qna:
        # Estimation : ~500 tokens par page en moyenne
        tokens_estimes = nombre_pages * 500
        cout_qna = (
            tokens_estimes / 1000
            * cout_qna_par_1k_tokens.get(modele_qna, 0.001)
        )

    return {
        "cout_ocr": round(cout_ocr, 4),
        "cout_qna": round(cout_qna, 4),
        "cout_total": round(cout_ocr + cout_qna, 4),
    }

Surveiller la santé du service

Un pipeline se dégrade lentement et silencieusement : un nouveau fournisseur envoie des scans de mauvaise qualité, le taux d’erreur monte de 2 % à 15 %, personne ne s’en aperçoit avant la clôture mensuelle. Quatre métriques suffisent à détecter la dérive — volume traité, durée moyenne, taux d’erreur, coût estimé — à condition de les émettre systématiquement et de les envoyer vers votre outil habituel, Datadog, Prometheus ou autre.

import json
from datetime import datetime

def log_metrique(nom: str, valeur: float, tags: dict = None) -> None:
    """Enregistre une métrique de monitoring."""
    metrique = {
        "timestamp": datetime.utcnow().isoformat(),
        "nom": nom,
        "valeur": valeur,
        "tags": tags or {}
    }
    # En production : envoyez vers Datadog, Prometheus, etc.
    logger.info(f"METRIQUE: {json.dumps(metrique)}")

# Exemples de métriques à suivre
log_metrique("ocr.pages_traitees", 150, {"batch": "factures_mars"})
log_metrique("ocr.duree_moyenne_secondes", 2.3)
log_metrique("ocr.taux_erreur", 0.02)
log_metrique("ocr.cout_estime_usd", 0.15)

Où aller ensuite

Vous disposez maintenant des bases pour construire un système Document AI complet, et plusieurs prolongements s’ouvrent selon vos contraintes. Les très gros volumes justifient un déploiement on-premise du modèle OCR sur GPU dédié ; des documents très spécifiques à votre métier peuvent motiver un fine-tuning, sujet à aborder directement avec Mistral. Côté intégration, l’enjeu est de connecter le pipeline à vos systèmes existants — ERP, CRM — et de construire un dashboard pour que vos équipes interrogent elles-mêmes les documents sans passer par vous. L’étape finale consiste à supprimer le déclenchement manuel : le pipeline part tout seul à l’arrivée de nouveaux documents.

Récapitulatif du cours

Ces seize leçons vous ont fait parcourir tout le cycle. Vous êtes parti des fondamentaux — ce qu’est Document AI, le modèle OCR 4, les cas d’usage — avant d’entrer dans l’OCR avancé avec l’API, l’extraction de tableaux, les headers/footers et l’optimisation. Les annotations vous ont ensuite donné l’extraction structurée, le traitement par lots et le contrôle de conformité, puis le QnA documentaire a ouvert l’interrogation en langage naturel, du document isolé au corpus indexé en RAG. Vous disposez désormais des briques nécessaires pour outiller la gestion documentaire de votre organisation avec Mistral Document AI.

Points clés à retenir

• Un pipeline de production inclut retry, logging, monitoring et gestion des coûts
• Le backoff exponentiel protège contre les erreurs transitoires de l’API
• Estimez vos coûts avant de passer à l’échelle (OCR par page, QnA par token)
• Le monitoring en continu permet de détecter les dégradations de performance
• Document AI couvre l’ensemble du cycle de vie documentaire, de l’ingestion à l’exploitation

Testez vos connaissances

OCR, annotations, QnA : validez la chaîne documentaire complète.

1. Que produit l'API OCR de Mistral à partir d'un document ?

Réponse : Le contenu structuré du document — texte, tableaux, mise en page — exploitable par programme : OCR 4 lit le document comme une source de données, pas comme une image.

2. À quoi sert l'API Annotations ?

Réponse : À extraire des métadonnées structurées selon votre schéma (champs d’une facture, clauses d’un contrat) — en unitaire ou en batch pour les gros volumes.

3. Que permet Document QnA ?

Réponse : Poser des questions en langage naturel sur un ou plusieurs documents et obtenir des réponses fondées sur leur contenu — la couche conversationnelle au-dessus de l’extraction.

4. Comment optimise-t-on la précision de l'OCR ?

Réponse : Par la qualité d’entrée (résolution, netteté, orientation) et le traitement des zones difficiles — headers, footers, pagination — que le cours apprend à gérer explicitement.

5. Pourquoi combiner OCR et RAG ?

Réponse : L’OCR transforme les documents scannés en texte indexable, le RAG les rend interrogeables à l’échelle : ensemble, ils ouvrent la recherche sémantique sur des archives entières.

Extraction, annotation, question-réponse, indexation : le pipeline de production ci-dessus enchaîne les quatre — votre paperasse devient une base de données.