Ingestion de données RAG pour les banques : PDF, OCR, tableaux et connecteurs
Comment intégrer des documents bancaires pour RAG : classer les PDF, exécuter l’OCR avec des indicateurs de révision, conserver la structure Word, gérer les tableaux et les images, et créer des connecteurs résilients.
Sources: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
Le système RAG d’une banque échoue en silence
Imaginez une banque de taille moyenne qui met en place un système de génération améliorée par la recherche sur sa base de connaissances interne. Des milliers de documents sont concernés : circulaires réglementaires du RBI, manuels de politique, directives de Basel III, procédures de traitement des prêts, dossiers KYC et guides opérationnels. Le projet semble bien avancé. Un ingénieur met en place la base de données vectorielle, un autre développe le pipeline de recherche, et un troisième connecte le LLM et crée l’interface. Puis le système est mis en service.
Après trois semaines, les utilisateurs commencent à se plaindre de réponses incorrectes, et la cause n’est pas une hallucination. Le système de recherche trouve le document adéquat, et le modèle répond fidèlement en se basant sur ce qui lui a été fourni. Les problèmes ont commencé bien plus tôt. Un circulaire de la RBI concernant les normes de liquidité des NBFC n’existait que sous forme de PDF scanné, et lors du traitement OCR, le chiffre « 10 % » est apparu sous la forme « lO % », avec le chiffre 1 remplacé par une minuscule L. Le moteur de parsing l’a accepté sans vérification. Le modèle d’encodage a stocké ce texte endommagé, le système de stockage vectoriel l’a enregistré, le système de recherche l’a classé parmi les premiers résultats, et le LLM a affirmé avec assurance à un responsable de conformité que le ratio requis était « lO percent ».
La récupération n’a pas échoué, ni la génération. Chaque composant a fait ce pour quoi il avait été conçu. Le défaut résidait dans la première étape du pipeline, l’ingestion des données, qui compte parmi les points aveugles les plus courants et coûteux dans les systèmes RAG en production. Les équipes passent des semaines à ajuster les embeddings, le découpage en blocs, les outils de réclassification et les prompts, tout en considérant l’ingestion comme une simple étape de prétraitement. Pourtant, la qualité de la récupération est déterminée bien avant l’arrivée d’une quelconque requête. Si la source est analysée incorrectement, aucun modèle d’embedding, aucun index ni aucune LLM ne pourra restituer les faits originaux.
Ce qui suit porte sur l’ingestion dans les systèmes RAG d’entreprise : gestion des sources structurées, semi-structurées et non structurées, traitement des documents multimodaux, conception pour une grande échelle, détection des erreurs d’analyse et intégration de tout cela à l’aide de Python fonctionnel.
Quel est le rôle de l’ingestion
Dans un pipeline RAG, l’ingestion prend des documents sources bruts, quel que soit leur format ou leur origine, et les transforme en texte propre et structuré, prêt à être fragmenté et intégré.
Le traitement a lieu entièrement dans le pipeline hors ligne, avant même que n’intervienne toute question de l’utilisateur. Ce que ce pipeline produit – du texte analysé et nettoyé accompagné de métadonnées – sert de données d’entrée pour chaque étape suivante. Trois tâches y sont effectuées. Le chargement récupère le document brut depuis son emplacement, qui peut être un système de fichiers, un bucket S3, un site SharePoint, une API REST ou une base de données. L’analyse extrait le texte propre à partir du format brut, que ce soit en décodant un PDF, en supprimant les balises HTML, en désérialisant du JSON ou en appliquant une reconnaissance optique de caractères sur une image. L’extraction initiale des métadonnées recueille l’origine du document, la date de création, son type ainsi que tous les détails spécifiques au format visibles au moment de l’analyse. Le nettoyage, le découpage en parties et l’intégration des métadonnées interviennent ensuite. La seule fonction de l’ingestion est d’obtenir du texte propre à partir des données brutes.
La réalité complexe des documents bancaires
La base de connaissances d’une banque ne se présente pas sous forme de Markdown ordonné. Il s’agit de formats accumulés au fil des décennies : PDF scannés datant des années 1990, fichiers Word avec suivi des modifications, pages HTML provenant d’un portail interne de conformité, exports JSON issus d’un système de gestion des politiques, e-mails contenant des pièces jointes ainsi que sites SharePoint dotés de structures de dossiers complexes. Chaque format nécessite son propre analyseur, et chaque analyseur échoue à sa manière. Pire encore, l’échec de l’analyse reste silencieux : un analyseur qui génère du texte corrompu ne lance aucune exception. Il renvoie simplement ce texte corrompu et continue son travail.
PDF : trois types, trois méthodes de traitement
Le PDF domine les documents réglementaires et liés à la conformité, et c’est aussi le format le plus difficile à analyser. Trois variantes exigent des traitements différents :
- PDF à couche de texte proviennent de logiciels de traitement de texte ou de fonctions d’impression vers PDF. Le texte se trouve dans le flux de contenu et peut être extrait directement.
- PDF scannés sont des fichiers d’image créés à partir de papier physique. Il n’y a absolument pas de couche de texte, ce qui rend l’OCR indispensable. Les circulaires anciennes et les politiques historiques relèvent pour la plupart de cette catégorie.
- PDF hybrides sont des scans qui ont reçu une couche de texte OCR de mauvaise qualité. Ce sont ceux qui présentent le plus de risques, car un analyseur naïf utilise cette couche pleine d’erreurs et suppose que tout fonctionne correctement.
Distinguer ces deux types est la première étape ; la fonction de classification ci-dessous examine un PDF et indique de quel type il s’agit.
import fitz # PyMuPDF
from pathlib import Path
def classify_pdf(pdf_path: str) -> str:
"""
Classify a PDF as text-native, scanned, or hybrid.
Returns: 'text', 'scanned', or 'hybrid'
"""
doc = fitz.open(pdf_path)
total_pages = len(doc)
pages_with_text = 0
total_text_length = 0
for page in doc:
text = page.get_text("text").strip()
if text:
pages_with_text += 1
total_text_length += len(text)
doc.close()
# No text layer at all: scanned PDF
if pages_with_text == 0:
return "scanned"
# Text exists but average is suspiciously short: likely hybrid with poor OCR
avg_text_per_page = total_text_length / total_pages
if avg_text_per_page < 100 and pages_with_text < total_pages * 0.5:
return "hybrid"
return "text"
def load_text_pdf(pdf_path: str) -> list[dict]:
"""
Extract text from a text-native PDF using PyMuPDF.
Returns a list of page dicts with text and metadata.
"""
doc = fitz.open(pdf_path)
pages = []
for page_num, page in enumerate(doc, start=1):
text = page.get_text("text")
pages.append({
"page_number": page_num,
"text": text.strip(),
"source": pdf_path,
"total_pages": len(doc),
})
doc.close()
return pages
Les fichiers numérisés nécessitent un processus d’OCR, et le graphique compare les trois options de production réalistes.
Pour une banque qui doit se conformer aux circulaires de la RBI, Azure Document Intelligence est le choix pratique. Il gère les scans de qualité variable, comprend la structure des tableaux présents dans les pages numérisées et évalue le degré de fiabilité de chaque mot, ce qui permet d’identifier les extractions peu fiables à examiner manuellement. Le chargeur ci-dessous l’appelle et applique cette logique.
from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential
def load_scanned_pdf_azure(
pdf_path: str,
endpoint: str,
api_key: str,
) -> list[dict]:
"""
Extract text from a scanned PDF using Azure Document Intelligence.
Returns pages with text, confidence scores, and table data.
"""
client = DocumentAnalysisClient(
endpoint=endpoint,
credential=AzureKeyCredential(api_key),
)
with open(pdf_path, "rb") as f:
poller = client.begin_analyze_document("prebuilt-read", f)
result = poller.result()
pages = []
for page in result.pages:
page_text = ""
low_confidence_words = []
for line in page.lines:
page_text += line.content + "\n"
# Flag words with confidence below threshold
for word in page.words:
if word.confidence < 0.85:
low_confidence_words.append({
"word": word.content,
"confidence": word.confidence,
})
pages.append({
"page_number": page.page_number,
"text": page_text.strip(),
"low_confidence_words": low_confidence_words,
"needs_review": len(low_confidence_words) > 5,
"source": pdf_path,
})
return pages
Le drapeau needs_review est important dans un environnement réglementé. Une page contenant plus de cinq mots à faible fiabilité ne devrait pas être transmise sans être remarquée. Elle doit figurer dans la file d’attente des évaluateurs humains, ou au moins porter un drapeau dans ses métadonnées afin que son évaluation ultérieure puisse en tenir compte.
Documents Word, HTML et sources structurées
Les fichiers Word sont utilisés principalement pour rédiger les politiques et procédures internes. Ils contiennent des titres, des tableaux et des listes qui facilitent leur récupération ultérieure, mais disparaissent avec un extracteur de texte simpliste. Le chargeur prend en compte cette structure lors du traitement du document.
from docx import Document as DocxDocument
from docx.oxml.ns import qn
def load_docx(docx_path: str) -> list[dict]:
"""
Load a DOCX file preserving heading structure and tables.
Returns a list of content blocks with type and text.
"""
doc = DocxDocument(docx_path)
blocks = []
current_section = "Introduction"
for element in doc.element.body:
# Paragraphs
if element.tag == qn("w:p"):
para = element
style = para.style.name if hasattr(para, "style") else ""
text = para.text.strip()
if not text:
continue
if style.startswith("Heading"):
level = int(style.replace("Heading ", "")) if style != "Heading" else 1
current_section = text
blocks.append({
"type": "heading",
"level": level,
"text": text,
"section": current_section,
"source": docx_path,
})
else:
blocks.append({
"type": "paragraph",
"text": text,
"section": current_section,
"source": docx_path,
})
# Tables
elif element.tag == qn("w:tbl"):
table_data = []
for row in element.findall(f".//{qn('w:tr')}"):
row_data = []
for cell in row.findall(f".//{qn('w:tc')}"):
cell_text = " ".join(
p.text for p in cell.findall(f".//{qn('w:t')}")
if p.text
)
row_data.append(cell_text.strip())
table_data.append(row_data)
blocks.append({
"type": "table",
"data": table_data,
"section": current_section,
"source": docx_path,
})
return blocks
Le suivi par sections est très utile. Dans une procédure KYC composée de quarante paragraphes, savoir qu’une partie provient de la « Section 3 : Due Diligence renforcée » et non de la « Section 1 : KYC standard » constitue des métadonnées qui améliorent considérablement la précision du retrait des informations.
Les portails de conformité sont presque toujours des applications web, il est donc nécessaire que le chargement HTML passe directement au contenu principal, en ignorant la navigation, les publicités, les bannières de cookies et autres éléments génériques.
import httpx
from bs4 import BeautifulSoup
def load_html_page(url: str, headers: dict | None = None) -> dict:
"""
Load an HTML page and extract main content, stripping boilerplate.
"""
response = httpx.get(url, headers=headers or {}, follow_redirects=True)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
# Remove boilerplate elements common in compliance portals
for tag in soup(["nav", "header", "footer", "script", "style",
"aside", "advertisement", ".cookie-banner",
".sidebar", ".breadcrumb"]):
tag.decompose()
# Find main content area (adjust selectors for your portal)
main_content = (
soup.find("main")
or soup.find("article")
or soup.find(id="content")
or soup.find(class_="content")
or soup.find("body")
)
text = main_content.get_text(separator="\n", strip=True) if main_content else ""
return {
"url": url,
"title": soup.title.string if soup.title else "",
"text": text,
"source": url,
}
Les sources structurées telles que les systèmes de gestion des politiques, les bases de données de cas et les journaux d’opérations exigent une approche différente. Leur texte est déjà structuré, il s’agit donc de déterminer quels champs intégrer et comment afficher les lignes sous forme de texte consultable.
import json
import psycopg2
from typing import Any
def load_json_policies(json_path: str, text_fields: list[str]) -> list[dict]:
"""
Load a JSON array of policy documents, extracting specified text fields.
Banking example: a JSON export from a policy management system.
"""
with open(json_path, "r", encoding="utf-8") as f:
policies = json.load(f)
documents = []
for policy in policies:
# Combine relevant text fields into a single document
text_parts = []
for field in text_fields:
if field in policy and policy[field]:
text_parts.append(f"{field.replace('_', ' ').title()}: {policy[field]}")
documents.append({
"text": "\n\n".join(text_parts),
"policy_id": policy.get("id"),
"policy_type": policy.get("type"),
"effective_date": policy.get("effective_date"),
"jurisdiction": policy.get("jurisdiction"),
"source": json_path,
})
return documents
def load_from_postgres(
connection_string: str,
query: str,
text_column: str,
metadata_columns: list[str],
) -> list[dict]:
"""
Load documents from a PostgreSQL database.
Banking example: regulatory requirement records from a compliance database.
"""
conn = psycopg2.connect(connection_string)
cursor = conn.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
cursor.close()
conn.close()
documents = []
for row in rows:
row_dict = dict(zip(columns, row))
documents.append({
"text": str(row_dict.get(text_column, "")),
**{col: row_dict.get(col) for col in metadata_columns},
"source": "postgresql",
})
return documents
Les banques conservent la plupart de leurs fichiers de travail dans SharePoint ou Google Workspace, et pour y accéder, il faut OAuth ainsi que les API Graph ou Drive correspondantes.
import httpx
from typing import Generator
def load_sharepoint_library(
tenant_id: str,
client_id: str,
client_secret: str,
site_id: str,
drive_id: str,
) -> Generator[dict, None, None]:
"""
Iterate over all DOCX and PDF files in a SharePoint document library.
Yields file metadata dicts; caller is responsible for downloading and parsing.
"""
# Get access token
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_response = httpx.post(token_url, data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": "https://graph.microsoft.com/.default",
})
access_token = token_response.json()["access_token"]
headers = {"Authorization": f"Bearer {access_token}"}
base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
# List files recursively
next_link = f"{base_url}/root/children"
while next_link:
response = httpx.get(next_link, headers=headers)
data = response.json()
for item in data.get("value", []):
if item.get("file") and item["name"].endswith((".pdf", ".docx")):
yield {
"name": item["name"],
"download_url": item.get("@microsoft.graph.downloadUrl"),
"created": item.get("createdDateTime"),
"modified": item.get("lastModifiedDateTime"),
"size": item.get("size"),
"web_url": item.get("webUrl"),
}
next_link = data.get("@odata.nextLink")
En lot ou en flux continu
Toute conception d’ingestion commence par une décision : traiter les documents dès leur arrivée, ou en lots planifiés ?
- L’ingestion en lot s’exécute selon un calendrier, généralement chaque nuit ou chaque semaine. Elle convient lorsque la fraîcheur des données n’est pas cruciale, que le volume est élevé et que les coûts comptent. Une banque qui ingère des circulaires mensuelles de la RBI n’a pas besoin d’une transmission en temps réel, et un traitement nocturne est moins coûteux.
- L’ingestion en flux continu fonctionne via une file d’événements au fur et à mesure que les documents arrivent. Elle est adaptée aux cas où la fraîcheur des données est essentielle, comme dans un système de conformité qui doit afficher une nouvelle circulaire quelques minutes seulement après sa publication.
Les connecteurs externes nécessitent une protection contre les limites de fréquence, et la classe ci-dessous enveloppe les requêtes avec des tentatives répétées et des mécanismes de recul temporel.
import time
import httpx
from pathlib import Path
class RateLimitedConnector:
"""
A base connector with retry logic and rate limiting.
Use this pattern for any external API that imposes rate limits.
Banking example: connecting to an RBI document portal or a Bloomberg data feed.
"""
def __init__(
self,
base_url: str,
api_key: str,
requests_per_minute: int = 60,
max_retries: int = 3,
backoff_factor: float = 2.0,
):
self.base_url = base_url
self.api_key = api_key
self.min_interval = 60.0 / requests_per_minute
self.last_request_time = 0.0
self.max_retries = max_retries
self.backoff_factor = backoff_factor
def _wait_for_rate_limit(self):
elapsed = time.monotonic() - self.last_request_time
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request_time = time.monotonic()
def fetch(self, endpoint: str, params: dict | None = None) -> dict:
url = f"{self.base_url}/{endpoint}"
headers = {"Authorization": f"Bearer {self.api_key}"}
for attempt in range(self.max_retries):
self._wait_for_rate_limit()
try:
response = httpx.get(url, headers=headers, params=params or {})
if response.status_code == 429:
# Rate limited: back off and retry
wait_time = self.backoff_factor ** attempt
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
if attempt == self.max_retries - 1:
raise
time.sleep(self.backoff_factor ** attempt)
raise RuntimeError(f"Failed to fetch {url} after {self.max_retries} attempts")
Cette configuration n’est pas facultative. La plupart des API externes, y compris les fournisseurs de données financières, bloqueront une IP ou annuleront une clé dès que la limite est dépassée. Le backoff_factor permet un recul exponentiel, avec des temps d’attente de 2, 4 et 8 secondes lors des tentatives successives, ce qui constitue la politique de tentative minimale requise pour un connecteur en production.
Tableaux, graphiques et images
Le changement en 2025 dans les systèmes RAG en production a consisté à passer de pipelines uniquement textuels à des versions multimodales. Les documents bancaires ne sont pas purement textuels : ils contiennent des tableaux de ratios d’adéquation des capitaux, des graphiques montrant les tendances des NPA, des diagrammes des processus KYC ainsi que des images d’accords signés. Ignorer le contenu non textuel revient à jeter une grande partie des informations contenues dans chaque document, et selon la réglementation, cette partie est souvent la plus précieuse. Le tableau présent dans une circulaire de l’RBI, qui indique des seuils numériques exacts, est précisément ce dont a besoin une équipe de conformité.
Une table peut être traitée de deux manières : sérialisée en tant que texte, ou extraite sous forme de données structurées. Le choix approprié dépend de l’utilisation ultérieure. La fonction d’ extraction ci-dessous permet d’extraire les tables à partir d’un PDF.
import fitz
import pdfplumber
def extract_tables_from_pdf(pdf_path: str) -> list[dict]:
"""
Extract tables from a PDF using pdfplumber.
Returns tables as both structured data and serialised text.
Banking example: extracting capital ratio tables from Basel III compliance reports.
"""
tables_output = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages, start=1):
tables = page.extract_tables()
for table_idx, table in enumerate(tables):
if not table or len(table) < 2:
continue
# First row as headers
headers = [str(h).strip() if h else f"col_{i}"
for i, h in enumerate(table[0])]
rows = table[1:]
# Structured data representation
structured = []
for row in rows:
row_dict = {}
for header, cell in zip(headers, row):
row_dict[header] = str(cell).strip() if cell else ""
structured.append(row_dict)
# Text serialisation (markdown table format for LLM consumption)
header_row = " | ".join(headers)
separator = " | ".join(["---"] * len(headers))
data_rows = [
" | ".join(str(cell or "").strip() for cell in row)
for row in rows
]
serialised = "\n".join([header_row, separator] + data_rows)
tables_output.append({
"page_number": page_num,
"table_index": table_idx,
"headers": headers,
"structured_data": structured,
"serialised_text": serialised,
"source": pdf_path,
})
return tables_output
Le Markdown sérialisé convient aux tableaux que les modèles LLM peuvent lire et analyser, comme par exemple pour la question « Quel est le ratio CET1 requis par Basel III ? ». Le JSON structuré convient aux valeurs qui seront interrogées dans du code ou comparées numériquement.
Les graphiques, diagrammes et infographies ne peuvent pas non plus être ignorés. Un graphique montrant l’évolution des NPA par trimestre, ou un organigramme du processus d’approbation des transactions FEMA, contient des informations absentes dans le texte environnant. Le schéma courant pour 2025 consiste à envoyer chaque image à un modèle langage-vision et à stocker la description générée afin de pouvoir l’incorporer ou la récupérer ultérieurement.
import fitz
import base64
import httpx
from io import BytesIO
def extract_and_describe_images(
pdf_path: str,
openai_api_key: str,
min_image_size: int = 5000, # bytes, to skip tiny decorative images
) -> list[dict]:
"""
Extract images from a PDF and generate text descriptions using GPT-4V.
Banking example: describing NPA trend charts and regulatory workflow diagrams.
"""
doc = fitz.open(pdf_path)
image_docs = []
for page_num in range(len(doc)):
page = doc[page_num]
image_list = page.get_images(full=True)
for img_idx, img_info in enumerate(image_list):
xref = img_info[0]
base_image = doc.extract_image(xref)
image_bytes = base_image["image"]
# Skip tiny decorative images
if len(image_bytes) < min_image_size:
continue
# Encode to base64 for the vision API
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
image_ext = base_image["ext"]
# Generate description with GPT-4V
response = httpx.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {openai_api_key}"},
json={
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": (
"This image is from a banking regulatory document. "
"Describe its content precisely, including any numerical "
"values, trends, labels, axes, or process steps shown. "
"If it is a chart, describe what metric is shown, the "
"time period, and the key data points. If it is a "
"diagram, describe the process or relationship shown."
),
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/{image_ext};base64,{image_b64}"
},
},
],
}
],
"max_tokens": 400,
},
)
description = response.json()["choices"][0]["message"]["content"]
image_docs.append({
"type": "image_description",
"page_number": page_num + 1,
"image_index": img_idx,
"description": description,
"source": pdf_path,
})
doc.close()
return image_docs
Le texte d’instruction est crucial ici. Une demande générique du type « décrivez cette image » produit un résultat générique. Une instruction spécifique à un domaine, demandant des valeurs numériques, des tendances, des étiquettes ou des étapes de processus, génère des descriptions exploitables. Un ingénieur qui demande des informations sur les lignes de tendance NPA dans le rapport du troisième trimestre a besoin d’une description contenant les termes « NPA », « tendance » et le trimestre concerné, et non pas d’une description relative à un graphique en barres bleues.
Audio et vidéo
L’audio et la vidéo représentent actuellement la frontière du RAG dans le secteur bancaire. Les enregistrements des réunions du conseil d’administration, les transcriptions des appels avec les analystes, les conférences de presse de la RBI ainsi que les vidéos de formation contiennent des informations auxquelles les systèmes RAG ne peuvent pas encore accéder.
import httpx
import time
def transcribe_audio(
audio_path: str,
openai_api_key: str,
language: str = "en",
) -> dict:
"""
Transcribe an audio file using OpenAI Whisper.
Banking example: transcribing an RBI monetary policy press conference recording.
"""
with open(audio_path, "rb") as audio_file:
response = httpx.post(
"https://api.openai.com/v1/audio/transcriptions",
headers={"Authorization": f"Bearer {openai_api_key}"},
data={
"model": "whisper-1",
"language": language,
"response_format": "verbose_json", # includes timestamps per segment
"timestamp_granularities[]": "segment",
},
files={"file": (audio_path, audio_file, "audio/mpeg")},
)
result = response.json()
return {
"transcript": result.get("text", ""),
"segments": [
{
"start": seg["start"],
"end": seg["end"],
"text": seg["text"],
}
for seg in result.get("segments", [])
],
"language": result.get("language", language),
"source": audio_path,
}
Les timestamps des segments dans la sortie sont importants. Lorsqu’une question est répondue par une déclaration faite à 34 minutes dans le cadre d’une réunion, le système doit indiquer cette position précise et non se contenter de citer une enregistrement de la réunion.
Un routeur pour chaque format
Des chargeurs séparés ne suffisent pas. Un pipeline de production doit envoyer chaque document au bon analyseur, gérer les pannes de manière appropriée, suivre l’état du traitement et générer un format de sortie cohérent, quel que soit l’entrée utilisée.
from pathlib import Path
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class DocumentType(Enum):
PDF_TEXT = "pdf_text"
PDF_SCANNED = "pdf_scanned"
DOCX = "docx"
HTML = "html"
JSON = "json"
AUDIO = "audio"
UNKNOWN = "unknown"
def detect_document_type(file_path: str) -> DocumentType:
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".pdf":
classification = classify_pdf(file_path)
return DocumentType.PDF_SCANNED if classification == "scanned" else DocumentType.PDF_TEXT
elif suffix == ".docx":
return DocumentType.DOCX
elif suffix in (".html", ".htm"):
return DocumentType.HTML
elif suffix == ".json":
return DocumentType.JSON
elif suffix in (".mp3", ".mp4", ".wav", ".m4a"):
return DocumentType.AUDIO
return DocumentType.UNKNOWN
def ingest_document(
file_path: str,
azure_endpoint: str | None = None,
azure_api_key: str | None = None,
openai_api_key: str | None = None,
) -> list[dict]:
"""
Route a document to the correct parser and return a list of content blocks
with a consistent schema regardless of input format.
Every output block contains:
- text: the extracted text content
- source: origin path or URL
- doc_type: the detected document type
- page_number: where applicable
- needs_review: flag for human review queue
"""
doc_type = detect_document_type(file_path)
results = []
try:
if doc_type == DocumentType.PDF_TEXT:
pages = load_text_pdf(file_path)
for page in pages:
page["doc_type"] = "pdf_text"
page["needs_review"] = False
results.append(page)
elif doc_type == DocumentType.PDF_SCANNED:
if azure_endpoint and azure_api_key:
pages = load_scanned_pdf_azure(file_path, azure_endpoint, azure_api_key)
for page in pages:
page["doc_type"] = "pdf_scanned"
results.append(page)
else:
logger.warning(
f"Scanned PDF detected but no OCR credentials provided: {file_path}"
)
results.append({
"text": "",
"source": file_path,
"doc_type": "pdf_scanned",
"needs_review": True,
"error": "No OCR credentials available",
})
elif doc_type == DocumentType.DOCX:
blocks = load_docx(file_path)
for block in blocks:
block["doc_type"] = "docx"
block["needs_review"] = False
results.append(block)
elif doc_type == DocumentType.AUDIO:
if openai_api_key:
transcript = transcribe_audio(file_path, openai_api_key)
for segment in transcript["segments"]:
results.append({
"text": segment["text"],
"source": file_path,
"doc_type": "audio_transcript",
"start_seconds": segment["start"],
"end_seconds": segment["end"],
"needs_review": False,
})
else:
logger.warning(f"Unknown document type, skipping: {file_path}")
except Exception as e:
logger.error(f"Ingestion failed for {file_path}: {e}")
results.append({
"text": "",
"source": file_path,
"doc_type": str(doc_type.value),
"needs_review": True,
"error": str(e),
})
return results
Le schéma de sortie cohérent est le choix de conception le plus important. Le découpeur, le générateur d’embeddings et l’enrichisseur de métadonnées reçoivent tous la même structure de dictionnaire, que l’entrée soit une image circulaire scannée, une politique de prêt au format DOCX ou un transcript audio provenant d’un cours de conformité. Le flag needs_review envoie les documents problématiques dans une file d’attente humaine et empêche la propagation des erreurs.
Trois causes principales des problèmes
La plupart des problèmes liés à la qualité de la récupération dans les systèmes RAG bancaires trouvent leur origine dans trois erreurs d’ingestion.
Ignorance du type de document au moment de la parsing. Un extracteur de texte exécuté sur un PDF scanné ne renvoie rien, ou seulement quelques caractères isolés. Ce fichier se transforme en fragments vides, en vecteurs proches de zéro et en entrées d’index que la recherche restitue ensuite. Le LLM, ne voyant aucun contexte, invente une réponse. La fonction classify_pdf empêche cela grâce à quelques lignes de code qui épargnent des semaines de débogage.
Traitement des tableaux comme du texte principal. Un tableau de notation de risque transformé en prose perd ses lignes et ses colonnes, et une chaîne de caractères telle que "AAA 0.01% AA 0.05% A 0.10%" ne peut pas être récupérée de manière utile. En revanche, lorsqu’il est sérialisé sous forme de tableau Markdown avec des en-têtes, cela devient possible. Il convient d’extraire les tableaux séparément afin de conserver leur structure.
Perte de tentatives de réessai sur les connecteurs. Lorsqu’un connecteur SharePoint rencontre un temps d’attente réseau sans réagir, le corpus présente des lacunes et personne ne sait quels documents ont disparu. Le modèle RateLimitedConnector déclenche une alerte lorsque les tentatives de réessai sont épuisées et enregistre chaque tentative, ce qui permet de visualiser ces absences.
Compromis entre les outils
- PyMuPDF, pdfplumber et PyPDF2. PyMuPDF est l’extracteur le plus rapide pour les PDF contenant uniquement du texte. pdfplumber est plus lent mais bien meilleur pour les tableaux. PyPDF2 étant le plus ancien et le moins précis, il convient d’éviter son utilisation dans de nouveaux pipelines. Utilisez PyMuPDF pour le texte et pdfplumber pour les tableaux provenant du même fichier.
- OCR auto-hébergé ou en cloud. Tesseract est gratuit, fonctionne localement et gère bien les scans de bonne qualité. Pour des milliers de circulaires historiques de qualité variable, l’avantage en termes de précision d’Azure Document Intelligence justifie son prix par page. Une mauvaise interprétation des seuils réglementaires n’est pas un coût dont on doive se priver.
- Coût du streaming. Le streaming de chaque téléchargement en temps réel représente l’architecture la plus coûteuse, et une banque qui ajoute deux ou trois documents par jour n’en a pas besoin. Préservez cette solution pour les cas où la fraîcheur des données influence directement les résultats commerciaux, comme lorsqu’un service de trading a besoin d’accéder immédiatement aux nouvelles circulaires SEBI. Une bibliothèque de politiques RH ne relève pas de ce cas de figure.
- Coste multimodal. L’envoi de chaque image contenue dans chaque document via GPT-4o entraîne des dépenses réelles à grande échelle. Il convient de filtrer les images de petite taille, comme celles décoratives inférieures à 5 KB, d’éliminer celles dont le rapport d’aspect indique qu’il s’agit de logos ou d’icônes, et de mettre en cache les descriptions afin qu’une image réutilisée dans plusieurs documents ne soit décrite qu’une seule fois.
Le texte brut n’est pas un texte net
Ce que l’ingestion fournit, c’est du texte brut et non du texte net. Une étape de nettoyage distincte s’occupe de la suppression des bruits, de la déduplication, de la correction des erreurs d’OCR, de la normalisation multilingue et du retrait des éléments génériques, située entre l’ingestion et le découpage en blocs. Ce découpage est important car les problèmes diffèrent : les échecs d’ingestion produisent du contenu vide ou corrompu, tandis que les échecs de nettoyage génèrent du contenu qui semble correct mais contient des bruits subtils nuisant à la qualité de l’embedding, et qui ne se manifestent que lors d’un test de récupération.
Pour une utilisation en production, les exemples de code nécessitent un système d’journalisation, une gestion des secrets (conserver les clés API dans des variables d’environnement ou un gestionnaire de secrets, jamais en code brut) ainsi que des outils d’analyse de la latence à chaque étape.