Pipeline de production
De la preuve de concept à la production
Vous avez appris à utiliser l’OCR, les annotations et le QnA de Document AI. Cette dernière leçon rassemble tous ces éléments dans un pipeline de production robuste, avec monitoring, gestion des coûts et bonnes pratiques opérationnelles.
Architecture du pipeline complet
Un pipeline Document AI en production comprend les composants suivants :
- File d’attente d’ingestion : réception et priorisation des documents
- Pré-traitement : validation, conversion de format, amélioration d’image
- OCR + Annotations : extraction du texte et des données structurées
- Post-traitement : validation, nettoyage, enrichissement
- Stockage : base de données, vector store, archivage
- Interface de requête : API QnA, dashboard, intégrations
Implémentation du pipeline
Classe Pipeline
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}%"
)Utilisation du pipeline
# 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/")Estimation des coûts
La maîtrise des coûts est essentielle en production. Voici les facteurs principaux :
- OCR : facturé à la page traitée
- Annotations : incluses dans l’appel OCR (pas de surcoût)
- QnA : facturé en tokens d’entrée/sortie selon le modèle de chat
- Embeddings : facturé en tokens d’entrée
Calcul du coût estimé
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),
}Monitoring en production
Ajoutez des métriques pour suivre la santé de votre pipeline :
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)Prochaines étapes
Vous avez maintenant les bases pour construire un système Document AI complet. Voici les pistes pour aller plus loin :
- Déploiement sur GPU dédié : pour les très gros volumes, envisagez un déploiement on-premise du modèle OCR
- Fine-tuning : contactez Mistral pour un modèle adapté à vos documents spécifiques
- Intégration ERP/CRM : connectez votre pipeline à vos systèmes métier existants
- Interface utilisateur : construisez un dashboard pour que vos équipes interrogent les documents
- Automatisation complète : déclenchez le pipeline automatiquement à la réception de nouveaux documents
Récapitulatif du cours
Au fil de ces 16 leçons, vous avez appris :
- Les fondamentaux : ce qu’est Document AI, le modèle OCR 3, les cas d’usage
- L’OCR avancé : l’API, l’extraction de tableaux, les headers/footers, l’optimisation
- Les annotations : l’extraction structurée, le batch processing, la conformité
- Le QnA documentaire : l’interrogation de documents, le multi-documents, le RAG
Vous disposez désormais de tous les outils pour transformer 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