[!NOTE] TL;DR Pour l'extraction structurée de PDF multipages, un résultat Textract incomplet peut sembler parfaitement plausible. CAP relie chaque réponse à son bloc, à sa page et à son score, puis ferme le dossier si une preuve manque. Le seuil de confiance oriente la revue, il ne délivre jamais une autorisation métier.
Le rapport amputé
Le défaut tenait dans un seul NextToken. Pour étudier l'extraction structurée de PDF multipages, j'ai posé un scénario de recette : un rapport fictif de maintenance électrique de 12 pages. Mon classeur imaginaire avait déjà signé la remise en service. L'API, elle, n'avait livré que le premier lot de blocs.
Je ne présente pas ce scénario comme un incident client ni comme un benchmark. Il sert à isoler un défaut d'architecture reproductible à partir du contrat public de Textract : une réponse de GetDocumentAnalysis peut contenir NextToken. Le compte rendu doit conserver une date de contrôle et une valeur d'isolement, avec leurs pages d'origine. Sans ces preuves, aucun automate ne doit conclure que l'équipement est conforme.
La question documentaire
Ma question principale était la suivante : comment empêcher une décision métier à partir d'un PDF dont l'extraction semble complète, mais ne l'est pas ? Pour un dossier énergétique, le texte seul n'est pas la réponse. Il faut aussi pouvoir désigner l'endroit exact d'où provient chaque valeur.
J'ai nommé CAP, pour Contrôle des Ancrages de Preuve, le petit contrat qui relie chaque champ à un Block.Id, une page et un score. Trois objectifs structurent CAP : 1) parcourir tous les lots ; 2) attacher chaque réponse à sa question ; 3) empêcher qu'une preuve absente soit remplacée par une supposition. Le troisième point est une règle de décision, pas une promesse de l'OCR.
L'illusion du premier lot
On imagine spontanément que le premier retour contient le document entier. La référence AWS de GetDocumentAnalysis dit autre chose : MaxResults plafonne à 1,000 blocs, et NextToken indique qu'il reste des résultats. Un bloc n'est pas une page. LINE, WORD, QUERY et QUERY_RESULT occupent chacun une place dans cette pagination.
Un appel qui retourne SUCCEEDED ne prouve donc pas que ton propre collecteur a fini son travail. Plus gênant, l'API connaît aussi PARTIAL_SUCCESS et expose des Warnings. Je refuse ces états dans CAP, même si quelques réponses exploitables sont déjà présentes. Une réponse plausible n'est pas un dossier complet.
Il existe un second piège, plus discret. Selon les bonnes pratiques AWS pour Queries, une question sans Pages cible la page 1 par défaut. Sur un PDF de 12 pages, cela ne constitue pas une recherche dans 12 pages. J'indique donc explicitement Pages=["*"] au lancement pour les champs dont la position varie, puis je vérifie la page renvoyée. Une requête QUERIES ne remplace pas TABLES pour extraire un tableau entier.
Bien que Queries fasse gagner du temps sur les champs ciblés, elle ne reconstruit pas à elle seule les lignes d'un tableau de mesures. J'active TABLES si la décision dépend de plusieurs cellules, puis je traite leurs relations séparément. C'est moins flatteur qu'une démonstration sur un PDF propre. C'est aussi plus proche du terrain.
L'architecture CAP
Voici la place du contrôle dans la chaîne. Le PDF réside dans S3 ; StartDocumentAnalysis reçoit FeatureTypes=["QUERIES", "TABLES"], deux questions nommées par Alias, Pages=["*"], un ClientRequestToken stable et un NotificationChannel. La référence AWS de StartDocumentAnalysis documente ces paramètres et le JobId renvoyé. Je lie le token à la version S3 et à la version des questions : réutiliser un token avec d'autres paramètres provoque une erreur, pas une nouvelle analyse.
flowchart LR
A[PDF versionné dans S3] --> B[StartDocumentAnalysis]
B --> C[Textract QUERIES et TABLES]
C --> D[SNS fin de traitement]
D --> E[SQS]
E --> F[Collecteur CAP]
F --> G[GetDocumentAnalysis tous les lots]
G --> H[Preuves et statut en base]
H --> I[Revue humaine]
I --> J[Décision métier]
Le dossier ressemble à un meuble à tiroirs. Lire le premier tiroir donne des pièces authentiques, mais ne dit rien des pièces rangées derrière ; CAP garde l'étiquette du tiroir avec chaque pièce. Cette image explique pourquoi je stocke le JobId, la version S3, l'alias de question et l'identifiant du bloc avant toute revue.
La documentation AWS des opérations asynchrones recommande la notification SNS, éventuellement reçue via SQS, plutôt qu'un polling répété de Get. Sans OutputConfig, AWS conserve les résultats dans son espace pendant 7 jours. Je prévois donc une conservation maîtrisée des preuves et une politique de reprise avant cette échéance ; S3 et la base métier n'héritent pas automatiquement de cette durée.
L'assemblage des preuves
J'ai écrit le collecteur ci-dessous pour l'étape après notification SUCCEEDED. Il s'exécute avec Python 3.11, boto3 et Pydantic 2 ; passe-lui un JobId valide dans un compte AWS configuré. Le démarrage Textract, l'abonnement SNS/SQS et la persistance sont des composants d'infrastructure distincts, pas des lignes cachées dans l'exemple. Les modèles Pydantic valident la frontière d'entrée, conformément à la documentation des validateurs Pydantic.
from __future__ import annotations
import sys
import boto3
from pydantic import BaseModel, ConfigDict, Field, computed_field, field_validator
class Relation(BaseModel):
model_config = ConfigDict(strict=True, extra="ignore")
type: str = Field(alias="Type", description="Nature de la relation Textract.")
ids: list[str] = Field(alias="Ids", description="Identifiants des blocs liés.")
class Question(BaseModel):
model_config = ConfigDict(strict=True, extra="ignore")
alias: str = Field(alias="Alias", description="Nom stable du champ demandé.")
class Bloc(BaseModel):
model_config = ConfigDict(strict=True, extra="ignore")
id: str = Field(alias="Id", description="Identifiant du bloc Textract.")
kind: str = Field(alias="BlockType", description="Type du bloc.")
page: int | None = Field(default=None, alias="Page", description="Page source.")
text: str | None = Field(default=None, alias="Text", description="Texte extrait.")
confidence: float | None = Field(default=None, alias="Confidence", description="Score sur 100.")
query: Question | None = Field(default=None, alias="Query", description="Question d'origine.")
relations: list[Relation] = Field(default_factory=list, alias="Relationships", description="Liens vers les réponses.")
class Avertissement(BaseModel):
model_config = ConfigDict(strict=True, extra="ignore")
code: str = Field(alias="ErrorCode", description="Code de l'avertissement Textract.")
class Lot(BaseModel):
model_config = ConfigDict(strict=True, extra="ignore")
status: str = Field(alias="JobStatus", description="État du traitement.")
blocks: list[Bloc] = Field(alias="Blocks", description="Blocs de ce lot.")
next_token: str | None = Field(default=None, alias="NextToken", description="Curseur du lot suivant.")
warnings: list[Avertissement] = Field(default_factory=list, alias="Warnings", description="Avertissements éventuels.")
class Preuve(BaseModel):
model_config = ConfigDict(strict=True)
alias: str = Field(description="Champ demandé.")
text: str = Field(description="Valeur extraite.")
page: int = Field(ge=1, description="Page source dans le PDF.")
block_id: str = Field(description="Identifiant du bloc QUERY_RESULT.")
confidence: float = Field(ge=0, le=100, description="Confiance Textract sur 100.")
@field_validator("text")
@classmethod
def texte_present(cls, value: str) -> str:
if not value.strip():
raise ValueError("Une preuve vide n'est pas une preuve")
return value
class Dossier(BaseModel):
model_config = ConfigDict(strict=True)
job_id: str = Field(description="Identifiant du traitement Textract.")
preuves: list[Preuve] = Field(description="Réponses avec provenance.")
@computed_field
@property
def file_de_revue(self) -> str:
attendus = {"date_controle", "valeur_isolement"}
if {p.alias for p in self.preuves} != attendus:
return "REVUE_MANQUANT"
if any(p.confidence < 95.0 for p in self.preuves):
return "REVUE_INCERTAINE"
return "CONTROLE_METIER"
def recuperer(job_id: str) -> list[Bloc]:
client = boto3.client("textract")
token: str | None = None
blocs: list[Bloc] = []
while True:
brut = client.get_document_analysis(JobId=job_id, MaxResults=1000, NextToken=token) if token else client.get_document_analysis(JobId=job_id, MaxResults=1000)
lot = Lot.model_validate(brut)
if lot.status != "SUCCEEDED" or lot.warnings:
raise ValueError("Traitement incomplet ou avertissement Textract")
blocs.extend(lot.blocks)
token = lot.next_token
if token is None:
return blocs
def assembler(job_id: str) -> Dossier:
blocs = recuperer(job_id)
index: dict[str, Bloc] = {bloc.id: bloc for bloc in blocs}
preuves: list[Preuve] = []
for question in (bloc for bloc in blocs if bloc.kind == "QUERY"):
if question.query is None:
raise ValueError("Question sans alias")
reponses = [index[identifiant] for lien in question.relations if lien.type == "ANSWER" for identifiant in lien.ids if identifiant in index and index[identifiant].kind == "QUERY_RESULT"]
if len(reponses) != 1:
continue # Champ absent ou ambigu : la file de revue le signalera.
reponse = reponses[0]
if reponse.page is None or reponse.text is None or reponse.confidence is None:
continue # Aucune provenance complète, aucune preuve exploitable.
preuves.append(Preuve(alias=question.query.alias, text=reponse.text, page=reponse.page, block_id=reponse.id, confidence=reponse.confidence))
if len(preuves) != len({preuve.alias for preuve in preuves}):
raise ValueError("Alias dupliqué : vérification humaine requise")
return Dossier(job_id=job_id, preuves=preuves)
if __name__ == "__main__":
if len(sys.argv) != 2:
raise SystemExit("Usage : python cap.py JOB_ID après notification SUCCEEDED")
print(assembler(sys.argv[1]).model_dump_json(indent=2))
J'ai laissé une limite visible dans ce code : la route CONTROLE_METIER conduit encore à un humain. Elle ne signifie jamais « machine autorisée ». Le seuil de 95 % n'est ici qu'une règle de tri proposée pour la recette, sans calibration sur ton corpus. En production, je n'imprimerais pas les valeurs extraites dans les logs ; cet affichage sert uniquement à inspecter un jeu de test non sensible.
Résultats de complétude
Je préfère montrer un calcul falsifiable plutôt qu'un faux gain de productivité. Le tableau applique la limite documentée de 1,000 blocs par appel à des tailles de réponse hypothétiques. La colonne « blocs omis » décrit uniquement un collecteur fautif qui s'arrête après le premier appel ; elle ne décrit ni la précision de Textract ni une mesure de latence.
| Blocs dans la réponse hypothétique | Appels nécessaires au plus | Blocs omis si arrêt au premier lot |
|---|---|---|
| 1,000 | 1 | 0 |
| 1,001 | 2 | 1 |
| 2,500 | 3 | 1,500 |
| 4,000 | 4 | 3,000 |
Sur le cas à 2,500 blocs, corriger le parcours de NextToken ramène le nombre de blocs omis par ce défaut précis de 1,500 à 0. C'est un invariant algorithmique, pas un score d'extraction. Néanmoins, ce correctif ne sauve ni une question mal formulée ni une cellule fusionnée mal reconnue. Il enlève un mode de panne avant de mesurer les autres.
J'appelle hypothèse de la page fantôme l'idée suivante : un dossier peut paraître cohérent simplement parce que sa partie contradictoire n'a jamais atteint le collecteur. Ce n'est pas une propriété démontrée de tous les rapports. C'est une raison testable de comparer chaque décision à une annotation humaine portant sur le PDF entier.
Limites et tests à conduire
CAP ne calibre ni l'exactitude des dates ni les unités électriques. Une valeur « 17 » sans « MΩ » resterait dangereusement ambiguë ; j'ajouterais un validateur métier avant toute décision, et je mesurerais séparément les faux positifs. Les bonnes pratiques Textract d'AWS situent les scores entre 0 et 100 et rappellent que le bon seuil dépend du risque. Elles recommandent aussi des images d'au moins 150 DPI pour une meilleure lecture, sans garantir le résultat sur un scan dégradé.
Ce protocole pourrait facilement être testé sur un échantillon annoté par type de document : proportion de champs absents, accord sur la page source, taux de revue humaine et délai du PDF à la décision. Je comparerais les distributions avant et après changement de version des questions. Ici, je n'ai ni corpus annoté ni relevé de production ; aucune précision, aucun coût et aucune latence ne sont donc annoncés.
Un PDF de santé appelle des contrôles supplémentaires : accès limité aux preuves, conservation définie et validation humaine avant usage clinique. Je n'étends pas le scénario électrique au diagnostic. Même pour l'énergie, la bonne sortie d'un OCR reste une proposition sourcée, jamais une signature réglementaire.
Le terrain comme critère
Plus généralement, je juge un pipeline documentaire à la décision qu'il permet de contester, pas à la beauté de son JSON. Si tu veux éprouver ce contrat sur tes propres rapports, consulte mes solutions de traitement documentaire ou parlons de ton pipeline. Mon classeur imaginaire réclame désormais un NextToken avant d'apposer son tampon !