Créer un pipeline ETL de production pour RAG à l’aide de LangChain et FAISS.
Concevoir le découpage, l’incorporation et l’indexation vectorielle pour 40 000 PDFs de conformité, en intégrant des tentatives répétées, des files d’attente de messages non traités, des métriques Prometheus et des benchmarks mesurés.
Sources: Building a Production-Grade ETL Pipeline for RAG: A Low-Level Design Deep Dive
Imaginez-vous en ingénieur en GenAI au sein d’une fintech de taille moyenne. L’équipe chargée du respect des réglementations possède plus de 40 000 PDF, comprenant des dossiers réglementaires, des circulaires de banques centrales, des directives des autorités de régulation des valeurs mobilières ainsi que des politiques internes, dispersés sur SharePoint, S3 et des disques locaux. Les analystes y passent quatre à six heures par jour pour effectuer des recherches manuelles. La demande du CTO est un chatbot capable de répondre à n’importe quelle question liée au respect des réglementations issues de cette bibliothèque en moins de trois secondes.
C’est évidemment un travail de génération augmentée par la récupération d’informations. Ce qui est facile à négliger, c’est le fait que la qualité des réponses est déterminée bien avant l’arrivée d’une question, dans le pipeline ETL qui prépare la base de connaissances. Un découpage inadéquat fragmente le contexte, des embeddings naïfs laissent le sens s’éloigner et un indexation insuffisante ralentit la récupération des informations. Une règle empirique indique que près de 80 % des problèmes de qualité liés aux méthodes RAG sont en réalité des problèmes d’ETL. Le modèle n’est généralement pas le goulot d’étranglement, c’est plutôt le pipeline.
La technologie utilisée ici est Python, LangChain, FAISS, Pinecone ainsi que des embeddings d’OpenAI.
L’échelle concernée
- Environ 40 000 documents dans des formats variés : PDF, DOCX, HTML et texte brut.
- Aproximativement 120 fragments par document après un découpage récursif à 512 tokens.
- Un index vectoriel contenant 4,8 millions d’embeddings dans Pinecone, avec 1536 dimensions.
Architecture du pipeline
Il s’agit de la méthode classique extraire, transformer, charger. Chaque étape cache des complexités spécifiques au domaine qui sont facilement sous-estimées.
┌────────────────────────────────────────────────────────────────────────────┐
│ RAG ETL Pipeline │
│ │
│ [EXTRACT] [TRANSFORM 1] [TRANSFORM 2] [LOAD] │
│ Doc Loader → Chunker → Embedder → VectorDB → RAG API │
│ PDF·DOCX·HTML Semantic splits Batch encode Pinecone/FAISS │
└────────────────────────────────────────────────────────────────────────────┘
Chaque étape s’exécute en tant que processus idempotent indépendant et écrit dans un stockage de préparation avant que la suivante ne commence. Cela assure une tolérance aux pannes. Si l’encodage échoue au document 30 000, rien ne force une extraction à nouveau de l’ensemble des 40 000 documents.
L’orchestrateur est un exécuteur d’étapes géré par des dataclasses :
# pipeline/orchestrator.py
# Top-level pipeline orchestration using a dataclass-driven stage runner
from dataclasses import dataclass, field
from typing import List, Optional
import logginglogger = logging.getLogger(__name__)@dataclass
class PipelineConfig:
source_bucket: str
chunk_size: int = 512 # tokens, not characters
chunk_overlap: int = 64 # token overlap between chunks
embedding_model: str = "text-embedding-3-small"
embedding_batch_size: int = 256
vector_store: str = "pinecone"
index_name: str = "compliance-docs"
namespace: str = "v2"
max_workers: int = 8 # parallel PDF extraction workersclass ETLPipeline:
def __init__(self, config: PipelineConfig):
self.config = config
self.loader = DocumentLoader(config)
self.chunker = AdaptiveChunker(config)
self.embedder = BatchEmbedder(config)
self.indexer = VectorIndexer(config) def run(self, doc_ids: Optional[List[str]] = None) -> dict:
"""Full ETL run. Pass doc_ids for incremental updates."""
stats = {} # Stage 1: Extract (parallelised — 2-5s per PDF, 8 workers)
raw_docs = self.loader.load(doc_ids=doc_ids)
stats["extracted"] = len(raw_docs)
logger.info(f"Extracted {len(raw_docs)} documents") # Stage 2a: Chunk
chunks = self.chunker.chunk_batch(raw_docs)
stats["chunks"] = len(chunks) # Stage 2b: Embed
embedded = self.embedder.embed_batch(chunks) # Stage 3: Index
upserted = self.indexer.upsert(embedded)
stats["upserted"] = upserted
return stats
Le diagramme de flux comme source de vérité
Un diagramme de flux en ASCII représente chaque point de décision, depuis la réception des documents bruts, en passant par les tentatives de traitement et le regroupement, jusqu’à l’écriture finale dans le stockage vectoriel.
┌──────────────────────┐
│ START: Doc Queue │
└──────────┬───────────┘
│
┌──────────▼───────────┐
│ Detect Doc Type │ ← PDF · DOCX · HTML · MD · TXT
└──────────┬───────────┘
│
┌──────────▼───────────┐
│ Parseable? (OCR) │
└────┬──────────┬──────┘
YES │ │ NO
│ ▼
│ ┌──────────────────┐
│ │ Dead-Letter Queue │ ← PagerDuty alert
│ └──────────────────┘
┌────────────▼────────────────────┐
EXTRACT │ Text Extraction + Cleaning │
│ PyMuPDF · python-docx · BS4 │
│ 2–5s / PDF · Parallel workers │
└────────────┬────────────────────┘
│
┌────────────▼────────────────────┐
│ Select Chunking Strategy │
└──────┬────────────┬─────────────┘
│ │ │
Code/Tables Default (prose) Long-form
│ │ │
┌────────▼──┐ ┌──────▼──────┐ ┌───▼─────────┐
TRANSFORM │ Fixed-Size│ │ Recursive │ │ Semantic │
(CHUNK) │ char split│ │ Char Split │ │ (NLTK+emb) │
└────────┬──┘ └──────┬──────┘ └───┬─────────┘
└────────────┼─────────────┘
│ (converge)
┌───────────▼─────────────┐
│ Attach Metadata Per Chunk│
│ doc_id·page·source·auth │
└───────────┬─────────────┘
│
┌───────────▼─────────────┐
│ Token Length Validation │
│ Assert chunk ≤ 8191 tok │
└───────────┬─────────────┘
│
┌───────────▼─────────────┐
│ Group Chunks → Batches │
│ batch_size=256 │
└───────────┬─────────────┘
│
┌───────────▼─────────────┐
TRANSFORM │ Call Embedding API │ ◄─────────┐
(EMBED) │ async · retry · rate │ │
└───────────┬─────────────┘ │
│ │
┌───────────▼─────────────┐ ┌─────┴────── ┐
│ API Call Success? │ NO │ Exp Backoff│
└──────┬──────────┬────────┘ │ 2^n+jitter │
YES │ └────────────►└─────────────┘
│
┌──────▼──────────────────┐
│ Normalize Embeddings │
│ L2 norm · float32 │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
│ Write to Redis Cache │
│ hash(text)→vec · TTL7d │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
LOAD │ Upsert to Vector Store │
(INDEX) │ Pinecone · batch=100 │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
│ Index Updated? │ NO ──► Retry DLQ (max 5)
└──────┬──────────────────┘
YES │
┌──────▼──────────────────┐
│ Emit Pipeline Metrics │
│ Prometheus · Grafana │
└──────┬──────────────────┘
│
┌──────▼──────────────────┐
│ Pipeline Complete ✓ │
└─────────────────────────┘
Chaque diamant de ce diagramme correspond à une branche réelle if/else dans le code. Le dessiner en premier permet de faire du diagramme de flux une référence pour l’architecture avant même toute mise en œuvre, et non simplement de la documentation.
Chunking
Le chunking est peut-être la décision la plus influente dans tout le processus. Un chunk représente l’unité de récupération des données. S’il est trop grand, on gaspille de l’espace mémoire pour du texte irrelevant ; s’il est trop petit, le sens des informations se perd. Aucune taille de chunk n’est universellement appropriée.
Départage à taille fixe
L’approche la plus simple consiste à diviser le texte tous les N caractères ou tokens, éventuellement avec chevauchement. Elle convient aux données uniformes comme les journaux d’activité et les tableaux, mais elle perturbe les limites des phrases et des paragraphes dans le texte narratif.
# chunking/fixed_size.py
from langchain.text_splitter import CharacterTextSplitter
import tiktokenclass FixedSizeChunker:
"""
Use for: Code files, CSV-like data, log files.
Avoid for: Prose documents, legal text, RBI circulars.
"""
def __init__(self, chunk_size: int = 512, overlap: int = 50):
self.enc = tiktoken.get_encoding("cl100k_base")
self.splitter = CharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=overlap,
separator="", # hard cut — no respect for sentences
length_function=lambda t: len(self.enc.encode(t)),
) def chunk(self, text: str, metadata: dict) -> list[dict]:
chunks = self.splitter.split_text(text)
return [
{
"text": c,
"token_count": len(self.enc.encode(c)),
"chunk_idx": i,
**metadata,
}
for i, c in enumerate(chunks)
]
Séparation récursive des caractères (par défaut)
C’est la méthode principale. Elle tente d’abord le séparateur le plus significatif (ligne vide, saut de ligne, point, espace, caractère unique) et n’a recours à une séparation plus stricte que si c’est absolument nécessaire. RecursiveCharacterTextSplitter de LangChain l’implémente. La version adaptative ci-dessous compte les tokens avec tiktoken, de sorte qu’aucun groupe ne dépasse la limite de 8192 tokens du modèle d’incorporation.
# chunking/recursive_splitter.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken
from typing import Listclass AdaptiveChunker:
"""
Production-grade chunker for compliance PDFs.
Uses tiktoken to count tokens (not characters), ensuring
we never exceed the 8192 embedding model context limit.
""" # Separator hierarchy: try broader splits first
SEPARATORS = [
"\n\n\n", # Section breaks
"\n\n", # Paragraph breaks
".\n", # Sentence at line end
". ", # Sentence in paragraph
"\n", # Line break
" ", # Word boundary (last resort)
] def __init__(self, config):
self.enc = tiktoken.encoding_for_model("text-embedding-3-small")
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=config.chunk_size, # 512 tokens
chunk_overlap=config.chunk_overlap, # 64 tokens
separators=self.SEPARATORS,
length_function=self._token_len,
is_separator_regex=False,
keep_separator=True, # preserve paragraph structure
) def _token_len(self, text: str) -> int:
return len(self.enc.encode(text, disallowed_special=())) def chunk_document(self, raw_doc: dict) -> List[dict]:
text = raw_doc["text"]
splits = self.splitter.split_text(text) chunks = []
for idx, chunk_text in enumerate(splits):
token_count = self._token_len(chunk_text) # Safety guard — should not happen, but log if it does
if token_count > 8191:
raise ValueError(
f"Chunk {idx} has {token_count} tokens — exceeds model limit. "
f"doc_id={raw_doc['doc_id']}"
) chunks.append({
"chunk_id": f"{raw_doc['doc_id']}_chunk_{idx}",
"text": chunk_text,
"token_count": token_count,
"chunk_index": idx,
"total_chunks": len(splits), # for reconstruction context
# Metadata propagated to vector store
"doc_id": raw_doc["doc_id"],
"source": raw_doc["source"],
"page_number": raw_doc.get("page_number", 0),
"doc_type": raw_doc["doc_type"],
"created_date": raw_doc["created_date"],
"authority": raw_doc.get("authority", "unknown"), # RBI, SEBI, etc.
}) return chunks def chunk_batch(self, raw_docs: List[dict]) -> List[dict]:
all_chunks = []
for doc in raw_docs:
try:
all_chunks.extend(self.chunk_document(doc))
except Exception as e:
logger.error(f"Chunking failed for {doc['doc_id']}: {e}")
return all_chunks
Chunking sémantique
Ici, chaque phrase est intégrée, et le texte est divisé là où la similarité cosinus entre les éléments voisins diminue fortement. Les blocs ainsi formés restent cohérents, mais le processus est 10 à 15 fois plus lent et nécessite des appels d’incorporation uniquement pour le chunking, ce qui double en pratique l’utilisation de l’API. Cette méthode n’a de sens que pour des documents narratifs à forte valeur ajoutée tels que de longs avis juridiques, des rapports annuels ou des documents politiques.
# chunking/semantic_chunker.py
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddingsclass SemanticChunkerWrapper:
"""
Use ONLY for: Long-form legal opinions, annual reports,
policy white papers where paragraph coherence is critical. Cost warning: This calls the embedding API TWICE per document
(once for chunking, once for indexing). Cache aggressively.
"""
def __init__(self, breakpoint_threshold: float = 0.85):
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.splitter = SemanticChunker(
embeddings=embeddings,
# Splits when cosine similarity between sentences drops below 0.85
breakpoint_threshold_type="cosine",
breakpoint_threshold_amount=breakpoint_threshold,
buffer_size=1, # include 1 sentence of context across boundaries
) def chunk(self, text: str) -> list[str]:
return self.splitter.split_text(text)
Comparaison :
- Taille fixe : idéal pour le code, les journaux et les données tabulaires. Rapide et peu coûteux, mais avec une faible cohérence.
- Récursif (par défaut) : le mieux adapté aux PDF, DOCX et HTML dans la plupart des cas. Rapide et peu coûteux, avec une cohérence moyenne.
- Sémantique : optimal pour les longs documents juridiques et réglementaires. Haute cohérence, mais 10 à 15 fois plus lent et nécessitant environ deux fois plus d’appels API.
Génération d’embeddings
Les embeddings transforment un fragment de texte en un vecteur dense qui en capture le sens. Les questions à se poser concernent le modèle à utiliser, la manière de gérer les lots, l’opportunité de mettre en cache et la façon de traiter les limites de vitesse de manière asynchrone.
Les calculs liés à la limitation de débit sont alarmants. Le modèle text-embedding-3-small d’OpenAI permet un million de tokens par minute au niveau 2. Avec 4,8 millions de fragments représentant en moyenne 300 tokens chacun, le total atteint 1,44 milliard de tokens. L’incorporation séquentielle à cette vitesse nécessiterait environ 24 heures, tandis que le traitement par lots asynchrone avec des travailleurs en parallèle réduit ce temps à environ deux heures et demie.
# embedding/batch_embedder.py
import asyncio
import hashlib
import json
import time
from typing import List
import openai
import redis
import numpy as np
from tenacity import retry, stop_after_attempt, wait_exponentialclass BatchEmbedder:
"""
Async batch embedder with:
- Redis-backed cache (avoid re-embedding unchanged chunks)
- Exponential backoff on rate limit errors
- Token budget management to avoid 429s
"""
TOKEN_BUDGET_PER_MINUTE = 900_000 # Stay under 1M limit with headroom def __init__(self, config):
self.client = openai.AsyncOpenAI()
self.model = config.embedding_model
self.batch_size = config.embedding_batch_size # 256
self.redis = redis.Redis(host="localhost", decode_responses=True)
self.tokens_this_minute = 0
self.minute_start = time.monotonic() def _cache_key(self, text: str) -> str:
return f"emb:{hashlib.sha256(text.encode()).hexdigest()[:16]}" def _from_cache(self, text: str):
val = self.redis.get(self._cache_key(text))
return json.loads(val) if val else None def _to_cache(self, text: str, vector: list, ttl: int = 604800): # 7 days
self.redis.setex(self._cache_key(text), ttl, json.dumps(vector)) @retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=60),
)
async def _call_api(self, texts: List[str]) -> List[List[float]]:
response = await self.client.embeddings.create(
model=self.model,
input=texts,
encoding_format="float",
)
return [item.embedding for item in response.data] async def _embed_batch(self, chunks: List[dict]) -> List[dict]:
texts = [c["text"] for c in chunks]
cache_hits, uncached_texts, uncached_idxs = [], [], [] # Separate cache hits from misses
for i, text in enumerate(texts):
cached = self._from_cache(text)
if cached:
cache_hits.append((i, cached))
else:
uncached_texts.append(text)
uncached_idxs.append(i) # Token budget enforcement
elapsed = time.monotonic() - self.minute_start
if elapsed > 60:
self.tokens_this_minute = 0
self.minute_start = time.monotonic() estimated_tokens = sum(len(t.split()) * 1.3 for t in uncached_texts)
if self.tokens_this_minute + estimated_tokens > self.TOKEN_BUDGET_PER_MINUTE:
sleep_time = 60 - elapsed + 2
logger.info(f"Token budget hit, sleeping {sleep_time:.1f}s")
await asyncio.sleep(sleep_time)
self.tokens_this_minute = 0
self.minute_start = time.monotonic() # Embed uncached chunks
new_vectors = []
if uncached_texts:
new_vectors = await self._call_api(uncached_texts)
self.tokens_this_minute += estimated_tokens
for text, vec in zip(uncached_texts, new_vectors):
self._to_cache(text, vec) # Assemble results in original order
all_vectors = [None] * len(chunks)
for i, vec in cache_hits:
all_vectors[i] = vec
for i, vec in zip(uncached_idxs, new_vectors):
all_vectors[i] = vec # L2 normalize for cosine similarity in FAISS / Pinecone
for chunk, vector in zip(chunks, all_vectors):
arr = np.array(vector, dtype=np.float32)
arr /= np.linalg.norm(arr) + 1e-9
chunk["embedding"] = arr.tolist() return chunks async def embed_batch_async(self, all_chunks: List[dict]) -> List[dict]:
batches = [
all_chunks[i : i + self.batch_size]
for i in range(0, len(all_chunks), self.batch_size)
]
tasks = [self._embed_batch(batch) for batch in batches]
results = await asyncio.gather(*tasks, return_exceptions=True) embedded_chunks = []
for r in results:
if isinstance(r, Exception):
logger.error(f"Batch embedding failed: {r}")
else:
embedded_chunks.extend(r)
return embedded_chunks def embed_batch(self, chunks: List[dict]) -> List[dict]:
return asyncio.run(self.embed_batch_async(chunks))
La règle à appliquer est de ne jamais intégrer deux fois le même morceau de texte. Un cache basé sur Redis réduit les exécutions incrémentielles de 70 à 90 pour cent pour les collections qui changent rarement.
Indexation
L’indexation est le processus par lequel les morceaux intégrés sont stockés dans un système de recherche. Deux décisions priment ici : le schéma des métadonnées (pour une récupération filtrée) et le type d’index (plat, HNSW ou IVF).
Les métadonnées distinguent un système RAG simplifié d’un système destiné à un usage en production. Prenons l’exemple de la question concernant les déclarations de l’autorité de régulation des valeurs mobilières sur le trading d’informations privilégiées en 2023 : le système doit d’abord filtrer par authority=SEBI et year=2023 avant d’effectuer toute recherche de voisin le plus proche, ce qui réduit l’espace de recherche et améliore la précision.
# indexing/vector_indexer.py
from pinecone import Pinecone, ServerlessSpec
from typing import List
import uuid# ── Pinecone Metadata Schema ──────────────────────────────────────────
# These fields are indexed for filtered retrieval.
# Only add fields you will actually filter on — excess metadata costs money.
METADATA_SCHEMA = {
"doc_id": str, # for document-level retrieval
"chunk_index": int, # for ordered reconstruction
"total_chunks": int, # for context window sizing
"source": str, # S3 path or SharePoint URL
"doc_type": str, # "circular", "guideline", "policy"
"authority": str, # "RBI", "SEBI", "IRDAI", "internal"
"year": int, # publication year (for temporal filters)
"page_number": int, # for citation in UI
"token_count": int, # for context window management at query time
}class VectorIndexer:
def __init__(self, config):
self.pc = Pinecone(api_key=config.pinecone_api_key)
self.index_name = config.index_name
self.namespace = config.namespace
self._ensure_index() def _ensure_index(self):
existing = [i.name for i in self.pc.list_indexes()]
if self.index_name not in existing:
self.pc.create_index(
name=self.index_name,
dimension=1536, # text-embedding-3-small output dim
metric="cosine",
spec=ServerlessSpec(cloud="aws", region="us-east-1")
)
logger.info(f"Created Pinecone index: {self.index_name}")
self.index = self.pc.Index(self.index_name) def _build_vectors(self, embedded_chunks: List[dict]) -> list:
"""
Convert embedded chunk dicts to Pinecone upsert format.
Each vector = (id, embedding, metadata).
"""
vectors = []
for chunk in embedded_chunks:
vector_id = chunk.get("chunk_id") or str(uuid.uuid4())
metadata = {
k: chunk[k]
for k in METADATA_SCHEMA.keys()
if k in chunk
}
# Store a snippet for quick UI display without fetching full doc
metadata["text_preview"] = chunk["text"][:200] vectors.append({
"id": vector_id,
"values": chunk["embedding"],
"metadata": metadata,
})
return vectors def upsert(self, embedded_chunks: List[dict]) -> int:
vectors = self._build_vectors(embedded_chunks)
total_upserted = 0 # Pinecone upsert in batches of 100
for i in range(0, len(vectors), 100):
batch = vectors[i : i + 100]
self.index.upsert(vectors=batch, namespace=self.namespace)
total_upserted += len(batch) logger.info(f"Upserted {total_upserted} vectors to '{self.namespace}'")
return total_upserted def query(
self,
query_vector: list,
top_k: int = 10,
filters: dict = None
) -> list:
"""
Filtered ANN search. Example:
filters = {"authority": {"$in": ["SEBI", "RBI"]}, "year": {"$gte": 2022}}
"""
result = self.index.query(
vector=query_vector,
top_k=top_k,
namespace=self.namespace,
filter=filters,
include_metadata=True,
include_values=False, # we don't need the vectors back
)
return result.matches
FAISS pour les installations sur site
Lorsque les données ne peuvent pas être transférées vers une base de données vectorielle cloud, ce qui est fréquent dans la banque, la finance et le secteur gouvernemental, FAISS constitue la solution standard. Le choix du type d’index est très important.
# indexing/faiss_indexer.py
import faiss
import numpy as npclass FAISSIndexer:
def __init__(self, dim: int = 1536, n_vectors: int = 5_000_000):
self.dim = dim
# Index selection logic based on corpus size:
# < 1M vectors → IndexFlatIP (exact, fast enough)
# 1M–10M vectors → IndexHNSWFlat (M=32, efConstruction=64)
# > 10M vectors → IndexIVFPQ (nlist=4096, m=32)
if n_vectors < 1_000_000:
self.index = faiss.IndexFlatIP(dim)
elif n_vectors < 10_000_000:
self.index = faiss.IndexHNSWFlat(dim, 32) # M=32 edges per node
self.index.hnsw.efConstruction = 64
self.index.hnsw.efSearch = 200 # higher = more accurate, slower
else:
quantizer = faiss.IndexFlatIP(dim)
self.index = faiss.IndexIVFPQ(quantizer, dim, 4096, 32, 8) # GPU offload for faster build (comment out if no GPU)
res = faiss.StandardGpuResources()
self.index = faiss.index_cpu_to_gpu(res, 0, self.index) def add(self, vectors: np.ndarray):
# Vectors must be float32, L2 normalized for cosine similarity
faiss.normalize_L2(vectors)
self.index.add(vectors) def search(self, query: np.ndarray, k: int = 10):
faiss.normalize_L2(query)
distances, indices = self.index.search(query, k)
return distances, indices def save(self, path: str):
cpu_index = faiss.index_gpu_to_cpu(self.index)
faiss.write_index(cpu_index, path) def load(self, path: str):
self.index = faiss.read_index(path)
IndexFlatIPconvient aux corpus comptant moins d’un million de vecteurs, offre un taux de rappel exact (100 %), fonctionne à vitesse moyenne et consomme beaucoup de mémoire.IndexHNSWFlatconvient aux corpus de 1 à 10 millions de vecteurs, atteint un taux de rappel d’environ 98,5 % à 10 requêtes, permet des recherches rapides et utilise une quantité moyenne de mémoire.IndexIVFPQconvient aux corpus comptant plus de 10 millions de vecteurs, atteint un taux de rappel de 94 à 96 %, est très rapide et consomme peu de mémoire grâce à une compression de 4 à 8 bits.
Gestion des pannes et tentatives de répétition
Toute étape d’un pipeline de production connaîtra parfois des pannes. Ce qui est important, c’est de gérer ces pannes de manière propre : conserver les documents défectueux sans interrompre l’exécution, et laisser une trace d’audit claire pour un examen manuel. Une file de messages échoués remplit cette fonction, en annotant chaque panne avec l’étape où elle s’est produite.
# pipeline/fault_handler.py
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetimeclass FailureStage(Enum):
EXTRACTION = "extraction"
CHUNKING = "chunking"
EMBEDDING = "embedding"
INDEXING = "indexing"@dataclass
class PipelineFailure:
doc_id: str
stage: FailureStage
error_type: str
error_message: str
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
retry_count: int = 0
is_retriable: bool = Trueclass DeadLetterQueue:
"""Stores failed items for later inspection and retry."""
def __init__(self, redis_client, key_prefix="dlq"):
self.r = redis_client
self.prefix = key_prefix def push(self, failure: PipelineFailure):
key = f"{self.prefix}:{failure.stage.value}"
self.r.lpush(key, json.dumps(failure.__dict__))
logger.warning(
f"DLQ push: doc={failure.doc_id} stage={failure.stage.value} "
f"err={failure.error_type}: {failure.error_message[:80]}"
) def pop_batch(self, stage: FailureStage, n: int = 50) -> list:
key = f"{self.prefix}:{stage.value}"
items = self.r.lrange(key, 0, n - 1)
self.r.ltrim(key, n, -1)
return [json.loads(i) for i in items]# ── Categorize errors for smart retry decisions ────────────────────
RETRIABLE_ERRORS = {
"openai.RateLimitError",
"openai.APITimeoutError",
"pinecone.exceptions.PineconeApiException",
"ConnectionError",
}NON_RETRIABLE_ERRORS = {
"UnicodeDecodeError", # Corrupted doc — send to DLQ permanently
"ValueError", # Bad chunking logic — needs code fix
"pdfminer.pdfpage.PDFTextExtractionNotAllowed", # Encrypted PDF
}def is_retriable(exc: Exception) -> bool:
return type(exc).__name__ in RETRIABLE_ERRORS
Surveillance
Un pipeline sans métriques est une boîte noire. Suivez au moins le débit, la latence, le taux d’erreurs et le coût par document, en les exposant à Prometheus et en les affichant dans Grafana.
# monitoring/metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server# ── Define metrics ─────────────────────────────────────────────────
docs_processed = Counter(
"etl_docs_processed_total",
"Total documents processed",
["status"], # labels: "success" or "failed"
)
chunks_created = Counter(
"etl_chunks_created_total",
"Total chunks created"
)
embedding_latency = Histogram(
"etl_embedding_seconds",
"Embedding API call latency",
buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)
index_queue_depth = Gauge(
"etl_index_queue_depth",
"Vectors waiting to be indexed"
)
embedding_cache_hit_rate = Gauge(
"etl_cache_hit_rate",
"Redis cache hit rate for embeddings"
)
dlq_depth = Gauge(
"etl_dlq_depth",
"Dead-letter queue depth",
["stage"]
)# Usage in BatchEmbedder:
with embedding_latency.time(): # measures duration automatically
vectors = await self._call_api(uncached_texts)# Start Prometheus metrics endpoint on port 8001
start_http_server(8001)
Panneaux de tableau de bord utiles :
- Dossiers par minute au fil du temps.
- Un histogramme de la latence d’incorporation P95.
- Taux de réussite du cache Redis, visant plus de 80 %.
- Niveau de la file des messages non traités par étape, avec alerte si elle n’est pas nulle pendant plus de cinq minutes.
- Cost par document, soit le nombre de tokens multiplié par le prix par token.
- Latence des requêtes FAISS ou Pinecone au moment du retrait (P50, P95, P99).
Résultats mesurés
Ces chiffres proviennent du système de conformité après ajustements, sur un AWS c5.4xlarge (16 cœurs vCPU, 32 Go de RAM) avec des travailleurs d’incorporation asynchrone.
Extraction de PDF
PyMuPDF sans OCR met entre 2 et 5 secondes par document, en fonction du nombre de pages et du format, et le travail s’exécute de manière bien parallélisée. Avec 8 travailleurs simultanés, l’extraction de 40 000 PDF prend environ :
40,000 docs × avg 3.5s / 8 workers = ~4.9 hours wall-clock
(scales linearly — 16 workers → ~2.4 hours)
Par type de fichier : un document circulaire simple de 1 à 5 pages met 2 à 3 secondes, car la couche de texte est simple. Un guide de 20 à 50 pages avec plusieurs colonnes et des tableaux intégrés nécessite de 3 à 4 secondes. Un rapport annuel de plus de 100 pages prend de 4 à 5 secondes, car il faut ignorer des mises en page denses et de nombreuses images. Les PDF chiffrés ou scannés sont envoyés dans la file d’attente des messages non traités, car ils nécessitent du OCR.
L’OCR constitue un processus distinct. Les fichiers scannés qui requièrent Tesseract ou AWS Textract prennent de 15 à 45 secondes chacun et doivent être traités dans un groupe de travail isolé, afin de ne pas ralentir le processus rapide.
Étape par étape : méthode naïve versus méthode optimisée
- Extraction du PDF : 2 à 5 secondes par document dans les deux cas, mais avec 8 travailleurs en parallèle via un
ThreadPoolExecutorplutôt qu’une boucle séquentielle. Le temps de traitement est limité par les opérations sur le disque et le réseau, et non par la CPU.
Extraction parallèle
Le chargeur répartit les documents dans un pool de threads :
# loader/parallel_pdf_loader.py
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Optional
import fitz # PyMuPDFclass DocumentLoader:
"""
Parallel PDF extraction using ThreadPoolExecutor.
PDF parsing is I/O-bound (disk reads + text decode),
so threads outperform processes here. Benchmarks (c5.4xlarge, non-OCR PDFs):
- 1 worker: ~2–5s per doc, sequential
- 8 workers: ~2–5s per doc, 8x throughput
- 16 workers: diminishing returns past disk I/O saturation
""" def __init__(self, config):
self.max_workers = config.max_workers # default: 8 def _extract_single(self, doc_path: str) -> dict:
"""Extract text from a single non-OCR PDF via PyMuPDF."""
doc = fitz.open(doc_path)
pages_text = []
for page in doc:
pages_text.append(page.get_text("text"))
doc.close() full_text = "\n\n".join(pages_text)
return {
"doc_id": doc_path,
"text": full_text,
"page_count": len(pages_text),
"source": doc_path,
"doc_type": "pdf",
} def load(
self,
doc_paths: List[str],
doc_ids: Optional[List[str]] = None
) -> List[dict]:
raw_docs, failed = [], [] with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_path = {
executor.submit(self._extract_single, path): path
for path in doc_paths
}
for future in as_completed(future_to_path):
path = future_to_path[future]
try:
raw_docs.append(future.result())
except Exception as e:
logger.error(f"Extraction failed: {path} → {e}")
failed.append(path) logger.info(
f"Extracted {len(raw_docs)} docs, "
f"{len(failed)} failures routed to DLQ"
)
return raw_docs
Leçons tirées de son déploiement à grande échelle
La création d’un RAG en production relève principalement de l’ingénierie ETL, et quelques leçons se détachent.
- La stratégie de segmentation influence la qualité de la récupération plus que le choix du modèle. Étudiez vos types de documents avant de sélectionner un outil de segmentation.
- Mesurez la longueur en tokens, et non en caractères. Utilisez
tiktoken. Le nombre de caractères n’a aucune signification pour un modèle d’incorporation. - Concevez le schéma des métadonnées avant d’écrire du code ETL. Il est impossible d’ajouter des filtres à un corpus déjà indexé sans le ré-ingérer.
- Mémorisez activement les embeddings dans le cache. Avec Redis, les exécutions incrémentielles sur des documents mis à jour coûtent de 10 à 15 pour cent du temps nécessaire pour une première exécution complète.
- Utilisez une file de messages défectueux. Un PDF défectueux ne doit pas faire échouer l’exécution. Isolez-le, enregistrez-le, envoyez une alerte et traitez-le plus tard.
IndexFlatIP à IndexHNSWFlat implique une reconstruction complète, il faut donc se décider tôt en fonction de la taille prévue du corpus.