Production Pipeline — Document AI: OCR and Document Intelligence

Update 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), which has been generally available since August 31 at the same price, or the alias mistral-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,…"}; the document_base64 type from older examples exists neither in the documentation nor in the mistralai SDK, and those examples would have failed. Sources: OCR Processor · Document QnA. Corrected November 3, 2026: The code previously read page.document_annotation page by page; in mistralai SDK 3.7.0, the document annotation is a single field in the response (response.document_annotation), and an image’s annotation is read from image.image_annotation.

From Proof of Concept to Production

You have learned how to use OCR, annotations, and Document AI QnA. One costly gap remains: the distance between a script that works on your laptop and a service your colleagues rely on every morning. This final lesson assembles the previous building blocks into a robust pipeline, complete with error recovery, monitoring, and cost control.

The Six Components of a Complete Pipeline

Ingestion Queue: Receives and prioritizes documents.
Pre-processing: Validates, converts formats, and enhances images.
OCR + Annotations: Extracts text and structured data.
Post-processing: Validates, cleans, and enriches data.
Storage: Databases, vector stores, and archiving.
Query Interface: QnA API, dashboards, and integrations.

The code in this lesson covers the core of the chain—steps 3 and 4—because that is where operational difficulties concentrate.

A Class That Encapsulates Processing

The PipelineDocumentAI class sets a configuration once and for all—the annotation schema, table format, and number of retries—then applies it to every document. Three mechanisms deserve your attention. First, retry with exponential backoff (time.sleep(2 ** tentative)) spaces out attempts by one, then two, then four seconds. This gives transient API saturation time to resolve rather than worsening it. Second, logging tracks every success with its page count and duration, and every failure with its attempt number; without these lines, you will only discover incidents when the accounting department calls. Finally, the stats dictionary feeds the end-of-batch report.

Note that the method always returns a ResultatTraitement, even after exhausting retries: the caller never has to handle an exception, it simply reads the erreur field.

import os
import time
import base64
import logging
from pathlib import Path
from dataclasses import dataclass, field
from mistralai.client 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:
        """Processes a document with retry and monitoring."""
        debut = time.time()
        self.stats["total"] += 1

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

                # Call parameters
                params = {
                    "model": "mistral-ocr-latest",
                    "document": {
                        "type": "document_url",
                        "document_url": f"data:application/pdf;base64,{doc_b64}"
                    },
                    "table_format": self.table_format,
                }

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

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

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

                # Extract annotations
                annotations = None
                if self.schema_annotation:
                    if response.document_annotation:
                        annotations = response.document_annotation

                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)  # Exponential backoff

        # All attempts failed
        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]:
        """Processes all PDFs in a folder."""
        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:
        """Generates a processing report."""
        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}%"
        )

In practice, the same class covers both needs: instantiated without a schema, it simply extracts text; instantiated with a Pydantic model, it also produces structured data.

# Simple pipeline (OCR only)
pipeline = PipelineDocumentAI(table_format="markdown")
resultats = pipeline.traiter_dossier("./factures/")
print(pipeline.rapport())

# Pipeline with 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/")

Anticipating the Bill

Cost control is essential in production, and it starts with understanding what you are billed for.

Item	Billing Basis
OCR	Per page processed
Annotations	Included in the OCR call (no extra cost)
QnA	Input/output tokens based on the chat model
Embeddings	Input tokens

The fact that annotations incur no extra cost changes the design: extract metadata during the OCR pass rather than running a separate LLM call afterward. The estimation function below lets you price a project before launching it; its rates are order-of-magnitude figures to adjust, with Mistral’s published reference table remaining the authority.

def estimer_cout(
    nombre_pages: int,
    avec_qna: bool = False,
    modele_qna: str = "mistral-small-latest"
) -> dict:
    """Estimates processing cost."""

    # Approximate rates (check current rates)
    cout_ocr_par_page = 0.001  # Adjust based on 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:
        # Estimate: ~500 tokens per page on average
        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 Service Health

A pipeline degrades slowly and silently: a new vendor sends poor-quality scans, the error rate climbs from 2% to 15%, and no one notices until the monthly close. Four metrics are enough to detect drift—volume processed, average duration, error rate, and estimated cost—provided you emit them systematically and send them to your usual tool, whether Datadog, Prometheus, or another.

import json
from datetime import datetime

def log_metrique(nom: str, valeur: float, tags: dict = None) -> None:
    """Records a monitoring metric."""
    metrique = {
        "timestamp": datetime.utcnow().isoformat(),
        "nom": nom,
        "valeur": valeur,
        "tags": tags or {}
    }
    # In production: send to Datadog, Prometheus, etc.
    logger.info(f"METRIQUE: {json.dumps(metrique)}")

# Example metrics to track
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)

Where to Go Next

You now have the basics to build a complete Document AI system, and several extensions open up depending on your constraints. Very high volumes justify an on-premise deployment of the OCR model on a dedicated GPU; documents highly specific to your business may motivate fine-tuning, a topic to address directly with Mistral. On the integration side, the challenge is connecting the pipeline to your existing systems—ERP, CRM—and building a dashboard so your teams can query documents themselves without going through you. The final step is removing manual triggering: the pipeline runs automatically when new documents arrive.

Course Summary

These sixteen lessons have taken you through the entire cycle. You started with the fundamentals—what Document AI is, the OCR 4 model, and use cases—before moving into advanced OCR with the API, table extraction, headers/footers, and optimization. Annotations then gave you structured extraction, batch processing, and compliance control, followed by Document QnA, which opened up natural language querying, from isolated documents to corpora indexed in RAG. You now have the necessary building blocks to equip your organization’s document management with Mistral Document AI.

Key Takeaways

A production pipeline includes retry, logging, monitoring, and cost management.
Exponential backoff protects against transient API errors.
Estimate your costs before scaling (OCR per page, QnA per token).
Continuous monitoring allows you to detect performance degradation.
Document AI covers the entire document lifecycle, from ingestion to operation.

Test Your Knowledge

OCR, annotations, QnA: validate the complete document chain.

1. What does Mistral's OCR API produce from a document?

Answer: The structured content of the document—text, tables, layout—ready for programmatic use: OCR 4 reads the document as a data source, not an image.

2. What is the purpose of the Annotations API?

Answer: To extract structured metadata according to your schema (invoice fields, contract clauses)—either individually or in batch for high volumes.

3. What does Document QnA enable?

Answer: Asking questions in natural language about one or more documents and getting answers based on their content—the conversational layer on top of extraction.

4. How do you optimize OCR accuracy?

Answer: Through input quality (resolution, sharpness, orientation) and handling difficult areas—headers, footers, pagination—which the course teaches you to manage explicitly.

5. Why combine OCR and RAG?

Answer: OCR transforms scanned documents into indexable text, while RAG makes them queryable at scale: together, they enable semantic search across entire archives.

Extraction, annotation, question-and-answer, indexing: the production pipeline above chains all four together—your paperwork becomes a database.