10 Erreurs Courantes dans Active Directory : Guide

Chunks trop grands (> 1000 tokens) : À l'inverse, des chunks trop volumineux introduisent du bruit et diluent l'information pertinente. Un chunk de 2000 tokens contenant un article entier rend difficile l'identification de l'information spécifique recherchée. De plus, vous payez des coûts d'embedding et de génération pour du contenu non pertinent. Découvrez les erreurs les plus fréquentes dans le chunking de documents pour le RAG et comment les éviter. Exemples concrets et solutions éprouvées.

Conséquences

Chunks trop petits :

• Perte de contexte sémantique (pronoms sans référents, concepts incomplets)
• Augmentation du nombre de chunks = plus d'embeddings = coûts multipliés
• Nécessité de récupérer plus de chunks pour reconstituer le contexte
• Réponses vagues ou incorrectes du LLM par manque d'information

Chunks trop grands :

Avez-vous évalué les risques d'injection de prompt sur vos systèmes d'IA en production ?

• Information pertinente noyée dans du contenu non pertinent (ratio signal/bruit faible)
• Coûts API élevés (embedding + génération sur tokens inutiles)
• Latence accrue (plus de tokens à traiter)
• Dépassement des limites de contexte des LLM (4K, 8K, 16K tokens)
• Moins de granularité dans la recherche vectorielle

La solution

Adoptez une approche empirique basée sur votre cas d'usage :

Testez systématiquement plusieurs tailles sur un échantillon représentatif de 100-200 requêtes réelles.

Exemple concret

# ❌ MAUVAIS : Taille fixe arbitraire sans contexte
from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
 chunk_size=100, # Trop petit !
 chunk_overlap=0
)
chunks = splitter.split_text(document)

# ✅ BON : Taille adaptée avec overlap et mesure réelle
from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken

# Calculer la taille optimale pour votre modèle
encoding = tiktoken.encoding_for_model("text-embedding-3-small")

def get_optimal_chunk_size(documents, target_tokens=512):
 """Analyse un échantillon pour calibrer la taille"""
 avg_chars_per_token = sum(
 len(doc) / len(encoding.encode(doc))
 for doc in documents[:50]
 ) / 50
 return int(target_tokens * avg_chars_per_token)

optimal_size = get_optimal_chunk_size(sample_docs, target_tokens=512)

splitter = RecursiveCharacterTextSplitter(
 chunk_size=optimal_size,
 chunk_overlap=int(optimal_size * 0.15), # 15% overlap
 length_function=lambda x: len(encoding.encode(x)),
 separators=["\n\n", "\n", ". ", " ", ""]
)

chunks = splitter.split_text(document)

# Validation : vérifier la distribution réelle
tokens_per_chunk = [len(encoding.encode(chunk)) for chunk in chunks]
print(f"Moyenne: {sum(tokens_per_chunk)/len(tokens_per_chunk):.0f} tokens")
print(f"Min: {min(tokens_per_chunk)}, Max: {max(tokens_per_chunk)}")

Erreur 2 : Ignorer la structure du document

Le problème

Utiliser un découpage aveugle par nombre de caractères sans tenir compte de la structure du document (titres, paragraphes, sections, listes) détruit la cohérence sémantique. Vous obtenez des chunks qui commencent au milieu d'une phrase ou coupent une liste en deux.

Exemple typique : un chunk qui se termine par "Les avantages sont :" et un autre qui commence par "1. Performance accrue". Le contexte est cassé, les embeddings sont moins pertinents, et la récupération devient aléatoire.

Conséquences

• Perte de cohérence sémantique : Les phrases coupées n'ont plus de sens
• Embeddings de mauvaise qualité : Un fragment incomplet génère un vecteur peu représentatif
• Duplication d'information : Besoin de récupérer plusieurs chunks pour reconstituer une idée
• Listes et tableaux fragmentés : Impossible de comprendre la structure complète
• Titres séparés du contenu : Le titre d'une section dans un chunk, le contenu dans un autre

La solution

Utilisez des splitters hiérarchiques qui respectent la structure naturelle du document :

1. Préserver les frontières naturelles : paragraphes, sections, éléments de liste
2. Conserver les titres avec leur contenu : inclure le titre de section dans chaque chunk de cette section
3. Respecter la hiérarchie : H1 > H2 > H3 > paragraphe
4. Garder les blocs complets : code, tableaux, listes comme unités indivisibles si possible

Exemple concret

# ❌ MAUVAIS : Découpage aveugle par caractères
chunks = [text[i:i+500] for i in range(0, len(text), 500)]
# Résultat : "Les fonctionnalités incluent : 1. Scal"

# ✅ BON : Découpage structuré et hiérarchique
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Définir une hiérarchie de séparateurs logiques
splitter = RecursiveCharacterTextSplitter(
 chunk_size=512,
 chunk_overlap=50,
 separators=[
 "\n\n\n", # Séparations de sections majeures
 "\n\n", # Paragraphes
 "\n", # Lignes
 ". ", # Phrases
 ", ", # Clauses
 " ", # Mots
 "" # Caractères (dernier recours)
 ],
 keep_separator=True
)

# Ou mieux : parser la structure Markdown/HTML d'abord
from langchain.document_loaders import UnstructuredMarkdownLoader
from langchain.text_splitter import MarkdownHeaderTextSplitter

# Splitter qui comprend Markdown
markdown_splitter = MarkdownHeaderTextSplitter(
 headers_to_split_on=[
 ("#", "Header 1"),
 ("##", "Header 2"),
 ("###", "Header 3"),
 ]
)

md_header_splits = markdown_splitter.split_text(markdown_document)

# Puis découper finement tout en gardant les headers comme métadonnées
final_splits = []
for doc in md_header_splits:
 # Ajouter le contexte hiérarchique à chaque chunk
 header_context = " > ".join([
 doc.metadata.get("Header 1", ""),
 doc.metadata.get("Header 2", ""),
 doc.metadata.get("Header 3", "")
 ]).strip(" > ")

 chunks = splitter.split_text(doc.page_content)
 for chunk in chunks:
 # Préfixer avec le contexte hiérarchique
 final_splits.append({
 "content": f"{header_context}\n\n{chunk}",
 "metadata": doc.metadata
 })

print(f"Créé {len(final_splits)} chunks structurés avec contexte")

Erreur 3 : Absence d'overlapping

Le problème

Sans chevauchement (overlap) entre chunks, vous créez des frontières artificielles qui coupent des concepts. Une information cruciale qui se trouve à cheval sur deux chunks risque de ne jamais être récupérée correctement.

Exemple : "Notre service est disponible 24/7 avec une garantie de disponibilité de 99.9%. [FRONTIÈRE DE CHUNK] Le support technique répond en moins de 2 heures." Une recherche sur "temps de réponse du support" ne trouvera pas le contexte complet.

Conséquences

• Perte d'information contextuellement liée : Les concepts qui s'étendent sur plusieurs chunks sont fragmentés
• Baisse de la qualité du retrieval : L'embedding d'un chunk incomplet est moins pertinent
• Incohérence dans les réponses : Le LLM reçoit des informations partielles et contradictoires
• Nécessité de récupérer plus de chunks : top_k plus élevé pour compenser, donc latence et coûts accrus

La solution

Implémentez un overlap stratégique :

Vos pipelines de données d'entraînement sont-ils protégés contre l'empoisonnement ?

• 10-15% pour les documents structurés : Articles, documentation technique
• 15-20% pour les documents denses : Contrats, manuels, recherche scientifique
• 20-25% pour les transcriptions : Conversations, interviews (contexte fluide)

Le overlap doit être suffisant pour capturer une idée complète, généralement 1-2 phrases ou 50-100 tokens.

Exemple concret

# ❌ MAUVAIS : Pas d'overlap
from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
 chunk_size=500,
 chunk_overlap=0 # ❌ Frontières dures
)
chunks = splitter.split_text(text)

# ✅ BON : Overlap calculé et adaptatif
import tiktoken

encoding = tiktoken.encoding_for_model("gpt-4")

def smart_chunking_with_overlap(text, chunk_tokens=512, overlap_percent=0.15):
 """
 Chunking intelligent avec overlap proportionnel.

 Args:
 text: Le texte à découper
 chunk_tokens: Taille cible en tokens
 overlap_percent: Pourcentage d'overlap (0.10 à 0.25)
 """
 from langchain.text_splitter import RecursiveCharacterTextSplitter

 # Calculer les tailles réelles
 overlap_tokens = int(chunk_tokens * overlap_percent)

 # Convertir en caractères approximatifs (ajustez selon votre corpus)
 chars_per_token = 4 # Moyenne pour l'anglais
 chunk_size = chunk_tokens * chars_per_token
 overlap_size = overlap_tokens * chars_per_token

 splitter = RecursiveCharacterTextSplitter(
 chunk_size=chunk_size,
 chunk_overlap=overlap_size,
 length_function=lambda x: len(encoding.encode(x)),
 separators=["\n\n", "\n", ". ", "! ", "? ", "; ", ": ", " ", ""]
 )

 chunks = splitter.split_text(text)

 # Validation : vérifier l'overlap réel
 overlaps = []
 for i in range(len(chunks) - 1):
 # Vérifier combien de texte est partagé entre chunks consécutifs
 chunk1_end = chunks[i][-200:] # Derniers 200 chars
 chunk2_start = chunks[i+1][:200] # Premiers 200 chars

 # Trouver la plus longue sous-chaîne commune
 overlap = longest_common_substring(chunk1_end, chunk2_start)
 overlaps.append(len(encoding.encode(overlap)))

 avg_overlap = sum(overlaps) / len(overlaps) if overlaps else 0
 print(f"Overlap moyen réel : {avg_overlap:.1f} tokens ({avg_overlap/chunk_tokens*100:.1f}%)")

 return chunks

def longest_common_substring(s1, s2):
 """Trouve la plus longue sous-chaîne commune."""
 m = [[0] * (1 + len(s2)) for _ in range(1 + len(s1))]
 longest, x_longest = 0, 0
 for x in range(1, 1 + len(s1)):
 for y in range(1, 1 + len(s2)):
 if s1[x - 1] == s2[y - 1]:
 m[x][y] = m[x - 1][y - 1] + 1
 if m[x][y] > longest:
 longest = m[x][y]
 x_longest = x
 return s1[x_longest - longest: x_longest]

# Usage
chunks = smart_chunking_with_overlap(
 document,
 chunk_tokens=512,
 overlap_percent=0.15
)

Erreur 4 : Perdre le contexte et les métadonnées

Le problème

Extraire le contenu brut sans conserver les métadonnées essentielles (source, auteur, date, section, tags, version) rend impossible la traçabilité et le filtrage contextuel. Un chunk isolé "Le taux est passé à 3.5%" n'a aucune valeur sans savoir : de quel taux, quelle date, quelle source ?

De même, ne pas inclure le contexte hiérarchique (titre du document, section, sous-section) dans le chunk le rend difficile à interpréter pour le LLM.

Conséquences

• Impossible de citer les sources : Non-conformité réglementaire, manque de crédibilité
• Pas de filtrage temporel : Impossible de restreindre à "documents après 2023"
• Confusion entre versions : Ancien contenu mélangé avec nouveau
• Pas de filtrage par domaine : Impossible de restreindre à "uniquement documentation technique"
• Perte du contexte hiérarchique : Le LLM ne sait pas dans quelle partie du document il se trouve
• Débogage impossible : Pas moyen de tracer pourquoi un chunk a été récupéré

La solution

Implémentez un système de métadonnées riche attaché à chaque chunk :

Exemple concret

# ❌ MAUVAIS : Chunks sans contexte ni métadonnées
chunks = text_splitter.split_text(document)
for chunk in chunks:
 vector = embed(chunk) # Perte totale de contexte
 vector_db.add(vector)

# ✅ BON : Enrichissement systématique avec métadonnées
import hashlib
from datetime import datetime
from pathlib import Path

def create_enriched_chunks(document, metadata):
 """
 Crée des chunks enrichis avec contexte et métadonnées complètes.
 """
 from langchain.text_splitter import MarkdownHeaderTextSplitter

 # 1. Parser la structure
 markdown_splitter = MarkdownHeaderTextSplitter(
 headers_to_split_on=[
 ("#", "h1"),
 ("##", "h2"),
 ("###", "h3"),
 ]
 )
 structured_chunks = markdown_splitter.split_text(document)

 enriched_chunks = []

 for idx, chunk in enumerate(structured_chunks):
 # 2. Construire le contexte hiérarchique
 hierarchy = " > ".join([
 chunk.metadata.get("h1", ""),
 chunk.metadata.get("h2", ""),
 chunk.metadata.get("h3", "")
 ]).strip(" > ")

 # 3. Préparer le contenu avec contexte injecté
 contextualized_content = f"""Document: {metadata['title']}
Section: {hierarchy}

{chunk.page_content}"""

 # 4. Générer un ID unique et stable
 chunk_id = hashlib.sha256(
 f"{metadata['source']}_{idx}_{chunk.page_content[:100]}".encode()
 ).hexdigest()[:16]

 # 5. Assembler toutes les métadonnées
 full_metadata = {
 # Identification
 "chunk_id": chunk_id,
 "doc_id": metadata.get("doc_id"),
 "chunk_index": idx,
 "total_chunks": len(structured_chunks),

 # Source et traçabilité
 "source": metadata.get("source"),
 "source_type": metadata.get("source_type", "unknown"), # pdf, html, markdown
 "url": metadata.get("url"),

 # Temporalité
 "created_at": metadata.get("created_at"),
 "modified_at": metadata.get("modified_at"),
 "indexed_at": datetime.now().isoformat(),
 "version": metadata.get("version", "1.0"),

 # Structure et contexte
 "title": metadata.get("title"),
 "h1": chunk.metadata.get("h1", ""),
 "h2": chunk.metadata.get("h2", ""),
 "h3": chunk.metadata.get("h3", ""),
 "hierarchy": hierarchy,
 "page_number": metadata.get("page_number"),

 # Classification
 "category": metadata.get("category"),
 "tags": metadata.get("tags", []),
 "language": metadata.get("language", "fr"),
 "author": metadata.get("author"),

 # Métriques
 "char_count": len(chunk.page_content),
 "word_count": len(chunk.page_content.split()),
 "token_count": len(encoding.encode(chunk.page_content)),

 # Technique
 "chunking_method": "markdown_hierarchical",
 "embedding_model": "text-embedding-3-small",
 }

 enriched_chunks.append({
 "content": contextualized_content,
 "raw_content": chunk.page_content, # Conserver aussi le brut
 "metadata": full_metadata
 })

 return enriched_chunks

# Usage avec filtrage contextuel
document_metadata = {
 "doc_id": "doc_12345",
 "source": "technical_manual_v2.pdf",
 "source_type": "pdf",
 "title": "Guide d'installation serveur",
 "created_at": "2024-11-15",
 "modified_at": "2024-12-20",
 "version": "2.1",
 "category": "technical_documentation",
 "tags": ["installation", "server", "linux"],
 "language": "fr",
 "author": "Équipe DevOps"
}

chunks = create_enriched_chunks(document, document_metadata)

# Indexation avec métadonnées
for chunk in chunks:
 vector = embed(chunk["content"])
 vector_db.add(
 vector=vector,
 text=chunk["content"],
 metadata=chunk["metadata"]
 )

# Recherche avec filtrage contextuel
results = vector_db.search(
 query_vector=embed(query),
 filter={
 "category": "technical_documentation",
 "modified_at": {"$gte": "2024-01-01"}, # Docs récents uniquement
 "language": "fr"
 },
 top_k=5
)

# Citation avec traçabilité complète
for result in results:
 print(f"Source: {result.metadata['source']}")
 print(f"Section: {result.metadata['hierarchy']}")
 print(f"Version: {result.metadata['version']}")
 print(f"Dernière mise à jour: {result.metadata['modified_at']}")

Erreur 5 : Ne pas tester différentes stratégies

Le problème

Choisir une stratégie de chunking (taille, overlap, méthode) sans tests empiriques sur vos données réelles. Chaque corpus est unique : ce qui fonctionne pour des articles de blog ne fonctionnera pas pour des contrats légaux ou du code source.

Erreur typique : copier-coller une configuration trouvée sur un blog technique sans la valider sur votre cas d'usage spécifique.

Conséquences

• Performance sous-optimale : Vous laissez 20-40% de qualité potentielle sur la table
• Coûts excessifs : Configuration inefficace = plus de tokens = facture API multipliée
• Latence inutile : Chunking mal calibré = plus de temps de traitement
• Mauvaise expérience utilisateur : Réponses imprécises ou incomplètes
• Découverte tardive des problèmes : En production, avec de vrais utilisateurs frustrés

La solution

Adoptez une approche scientifique avec A/B testing systématique :

1. Définir des métriques objectives (précision, rappel, latence, coût)
2. Créer un dataset de test représentatif (100-200 requêtes réelles)
3. Tester plusieurs configurations en parallèle
4. Mesurer et comparer quantitativement
5. Itérer sur les meilleures variantes

Framework de tests A/B

import pandas as pd
from typing import List, Dict, Any
import time
from dataclasses import dataclass

@dataclass
class ChunkingConfig:
 """Configuration de chunking à tester."""
 name: str
 chunk_size: int
 chunk_overlap: int
 method: str
 separators: List[str]

@dataclass
class TestResult:
 """Résultats d'un test."""
 config_name: str
 precision: float
 recall: float
 f1_score: float
 avg_latency_ms: float
 total_chunks: int
 avg_chunk_tokens: float
 total_cost_usd: float

def evaluate_chunking_strategy(
 documents: List[str],
 test_queries: List[Dict[str, Any]], # {"query": "...", "expected_docs": [...]}
 config: ChunkingConfig
) -> TestResult:
 """
 Évalue une stratégie de chunking sur un dataset de test.

 Args:
 documents: Liste des documents à indexer
 test_queries: Requêtes de test avec réponses attendues
 config: Configuration de chunking à tester

 Returns:
 Métriques de performance
 """
 from langchain.text_splitter import RecursiveCharacterTextSplitter

 print(f"\nTest de configuration: {config.name}")

 # 1. Chunking
 splitter = RecursiveCharacterTextSplitter(
 chunk_size=config.chunk_size,
 chunk_overlap=config.chunk_overlap,
 separators=config.separators
 )

 start_time = time.time()
 all_chunks = []
 for doc in documents:
 chunks = splitter.split_text(doc)
 all_chunks.extend(chunks)
 chunking_time = time.time() - start_time

 # 2. Embedding (simulé ici)
 encoding = tiktoken.encoding_for_model("text-embedding-3-small")
 chunk_tokens = [len(encoding.encode(chunk)) for chunk in all_chunks]
 avg_tokens = sum(chunk_tokens) / len(chunk_tokens)

 # Coût estimé (exemple : $0.02 / 1M tokens pour text-embedding-3-small)
 total_tokens = sum(chunk_tokens)
 embedding_cost = (total_tokens / 1_000_000) * 0.02

 # 3. Indexation dans vector DB (simulé)
 # vectors = [embed_function(chunk) for chunk in all_chunks]
 # vector_db.add(vectors, all_chunks)

 # 4. Évaluation sur requêtes de test
 latencies = []
 true_positives = 0
 false_positives = 0
 false_negatives = 0

 for test_query in test_queries:
 query = test_query["query"]
 expected_docs = set(test_query["expected_docs"])

 # Mesure de latence
 start = time.time()
 # results = vector_db.search(embed_function(query), top_k=5)
 # Pour la démo, simulation
 results = []
 latency = (time.time() - start) * 1000 # en ms
 latencies.append(latency)

 # Calcul précision/rappel
 retrieved_docs = set([r["doc_id"] for r in results])
 tp = len(retrieved_docs & expected_docs)
 fp = len(retrieved_docs - expected_docs)
 fn = len(expected_docs - retrieved_docs)

 true_positives += tp
 false_positives += fp
 false_negatives += fn

 # 5. Calcul des métriques finales
 precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
 recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
 f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0

 return TestResult(
 config_name=config.name,
 precision=precision,
 recall=recall,
 f1_score=f1,
 avg_latency_ms=sum(latencies) / len(latencies),
 total_chunks=len(all_chunks),
 avg_chunk_tokens=avg_tokens,
 total_cost_usd=embedding_cost
 )

# Définir les configurations à tester
configs = [
 ChunkingConfig(
 name="Small_NoOverlap",
 chunk_size=256,
 chunk_overlap=0,
 method="recursive",
 separators=["\n\n", "\n", ". ", " ", ""]
 ),
 ChunkingConfig(
 name="Medium_15pOverlap",
 chunk_size=512,
 chunk_overlap=75,
 method="recursive",
 separators=["\n\n", "\n", ". ", " ", ""]
 ),
 ChunkingConfig(
 name="Large_20pOverlap",
 chunk_size=1024,
 chunk_overlap=200,
 method="recursive",
 separators=["\n\n", "\n", ". ", " ", ""]
 ),
 ChunkingConfig(
 name="Semantic_Adaptive",
 chunk_size=512,
 chunk_overlap=100,
 method="semantic",
 separators=["\n\n\n", "\n\n", "\n", ". ", " "]
 ),
]

# Exécuter les tests
results = []
for config in configs:
 result = evaluate_chunking_strategy(
 documents=sample_documents,
 test_queries=test_queries,
 config=config
 )
 results.append(result)

# Comparer les résultats
df = pd.DataFrame([vars(r) for r in results])
df = df.sort_values('f1_score', ascending=False)

print("\n=== COMPARAISON DES STRATÉGIES ===")
print(df.to_string(index=False))

# Identifier le meilleur compromis
best_config = df.iloc[0]
print(f"\n✅ Meilleure configuration: {best_config['config_name']}")
print(f" F1-Score: {best_config['f1_score']:.2%}")
print(f" Précision: {best_config['precision']:.2%}")
print(f" Rappel: {best_config['recall']:.2%}")
print(f" Latence: {best_config['avg_latency_ms']:.1f}ms")
print(f" Coût: ${best_config['total_cost_usd']:.4f}")

Erreur 6 : Chunking identique pour tous types de documents

Le problème

Appliquer la même stratégie de chunking à tous vos documents, qu'il s'agisse de code source, de contrats PDF, de transcriptions audio ou d'articles de blog. Chaque type de document a une structure, une densité d'information et des caractéristiques uniques qui nécessitent une approche adaptée.

Exemple : découper du code Python avec la même stratégie qu'un contrat légal produit des résultats catastrophiques (fonctions coupées, contexte perdu). Pour approfondir, consultez Agentic AI 2026 : Autonomie en Entreprise.

Conséquences

• Code source fragmenté : Fonctions coupées en deux, perte de la logique
• Tableaux cassés : Lignes séparées des headers, données incompréhensibles
• Contrats légaux maltraités : Clauses divisées, numérotation perdue
• Emails mal parsés : Headers séparés du corps, threads cassés
• Performance inégale : Excellente sur certains types, désastreuse sur d'autres

La solution

Implémentez un système de chunking polymorphe qui détecte le type de document et applique la stratégie appropriée.

Stratégies par type de document

from typing import List, Dict
from enum import Enum
import re

class DocumentType(Enum):
 CODE = "code"
 MARKDOWN = "markdown"
 PDF_CONTRACT = "pdf_contract"
 TRANSCRIPT = "transcript"
 EMAIL = "email"
 CSV = "csv"
 API_DOC = "api_doc"
 GENERIC = "generic"

def detect_document_type(content: str, filename: str) -> DocumentType:
 """Détecte automatiquement le type de document."""
 # Par extension
 if filename.endswith(('.py', '.js', '.java', '.cpp', '.go')):
 return DocumentType.CODE
 elif filename.endswith(('.md', '.markdown')):
 return DocumentType.MARKDOWN
 elif filename.endswith('.csv'):
 return DocumentType.CSV

 # Par contenu
 if re.search(r'^(def|class|function|import)\s', content, re.MULTILINE):
 return DocumentType.CODE
 elif re.search(r'^#{1,6}\s', content, re.MULTILINE):
 return DocumentType.MARKDOWN
 elif re.search(r'\[\d{2}:\d{2}:\d{2}\]', content):
 return DocumentType.TRANSCRIPT
 elif 'From:' in content and 'Subject:' in content:
 return DocumentType.EMAIL

 return DocumentType.GENERIC

def chunk_document_adaptive(
 content: str,
 filename: str,
 metadata: Dict
) -> List[Dict]:
 """
 Chunking adaptatif selon le type de document.
 """
 doc_type = detect_document_type(content, filename)
 print(f"Type détecté: {doc_type.value}")

 if doc_type == DocumentType.CODE:
 return chunk_code(content, metadata)
 elif doc_type == DocumentType.MARKDOWN:
 return chunk_markdown(content, metadata)
 elif doc_type == DocumentType.PDF_CONTRACT:
 return chunk_contract(content, metadata)
 elif doc_type == DocumentType.TRANSCRIPT:
 return chunk_transcript(content, metadata)
 elif doc_type == DocumentType.EMAIL:
 return chunk_email(content, metadata)
 elif doc_type == DocumentType.CSV:
 return chunk_csv(content, metadata)
 elif doc_type == DocumentType.API_DOC:
 return chunk_api_doc(content, metadata)
 else:
 return chunk_generic(content, metadata)

def chunk_code(content: str, metadata: Dict) -> List[Dict]:
 """
 Chunking spécialisé pour code source.
 Découpe par fonction/classe en utilisant l'AST.
 """
 import ast
 from langchain.text_splitter import PythonCodeTextSplitter

 splitter = PythonCodeTextSplitter(
 chunk_size=256,
 chunk_overlap=30
 )
 chunks = splitter.split_text(content)

 enriched = []
 for idx, chunk in enumerate(chunks):
 # Extraire le nom de la fonction/classe
 try:
 tree = ast.parse(chunk)
 names = [node.name for node in ast.walk(tree)
 if isinstance(node, (ast.FunctionDef, ast.ClassDef))]
 entity_name = names[0] if names else "code_block"
 except:
 entity_name = "code_fragment"

 enriched.append({
 "content": chunk,
 "metadata": {
 **metadata,
 "chunk_type": "code",
 "entity_name": entity_name,
 "chunk_index": idx
 }
 })
 return enriched

def chunk_markdown(content: str, metadata: Dict) -> List[Dict]:
 """
 Chunking pour Markdown avec respect de la structure.
 """
 from langchain.text_splitter import MarkdownHeaderTextSplitter

 markdown_splitter = MarkdownHeaderTextSplitter(
 headers_to_split_on=[
 ("#", "h1"),
 ("##", "h2"),
 ("###", "h3"),
 ]
 )
 splits = markdown_splitter.split_text(content)

 enriched = []
 for idx, doc in enumerate(splits):
 hierarchy = " > ".join([
 doc.metadata.get("h1", ""),
 doc.metadata.get("h2", ""),
 doc.metadata.get("h3", "")
 ]).strip(" > ")

 enriched.append({
 "content": f"Section: {hierarchy}\n\n{doc.page_content}",
 "metadata": {
 **metadata,
 **doc.metadata,
 "chunk_type": "markdown",
 "hierarchy": hierarchy,
 "chunk_index": idx
 }
 })
 return enriched

def chunk_csv(content: str, metadata: Dict) -> List[Dict]:
 """
 Chunking pour CSV : grouper lignes avec header répété.
 """
 import csv
 from io import StringIO

 reader = csv.reader(StringIO(content))
 rows = list(reader)
 header = rows[0]
 data_rows = rows[1:]

 chunks = []
 chunk_size = 50 # lignes par chunk

 for i in range(0, len(data_rows), chunk_size):
 batch = data_rows[i:i + chunk_size]
 # Reconstituer avec header
 chunk_content = "\n".join([
 ",".join(header),
 *[",".join(row) for row in batch]
 ])

 chunks.append({
 "content": chunk_content,
 "metadata": {
 **metadata,
 "chunk_type": "csv",
 "row_start": i + 1,
 "row_end": min(i + chunk_size, len(data_rows)),
 "total_rows": len(data_rows)
 }
 })
 return chunks

def chunk_generic(content: str, metadata: Dict) -> List[Dict]:
 """
 Fallback : chunking générique standard.
 """
 from langchain.text_splitter import RecursiveCharacterTextSplitter

 splitter = RecursiveCharacterTextSplitter(
 chunk_size=512,
 chunk_overlap=75,
 separators=["\n\n", "\n", ". ", " ", ""]
 )
 chunks = splitter.split_text(content)

 return [{
 "content": chunk,
 "metadata": {
 **metadata,
 "chunk_type": "generic",
 "chunk_index": idx
 }
 } for idx, chunk in enumerate(chunks)]

# Usage
documents = [
 {"content": python_code, "filename": "app.py", "metadata": {...}},
 {"content": markdown_doc, "filename": "README.md", "metadata": {...}},
 {"content": csv_data, "filename": "data.csv", "metadata": {...}},
]

all_chunks = []
for doc in documents:
 chunks = chunk_document_adaptive(
 content=doc["content"],
 filename=doc["filename"],
 metadata=doc["metadata"]
 )
 all_chunks.extend(chunks)

print(f"Créé {len(all_chunks)} chunks adaptés")

Erreur 7 : Couper au milieu d'éléments critiques

Le problème

Diviser un document sans détecter et protéger les éléments atomiques critiques qui doivent rester intègres : blocs de code, formules mathématiques, tableaux, listes numérotées, citations longues, équations chimiques, etc.

Exemple catastrophique : Une formule chimique coupée en deux ("C6H12O" dans un chunk, "6" dans le suivant) devient incompréhensible et potentiellement dangereuse dans un contexte médical.

Conséquences

• Perte totale de sens : Formules, équations, code inutilisables
• Erreurs critiques : Dans des domaines sensibles (médical, légal, scientifique)
• Tableaux illisibles : Headers séparés des données
• Listes brisées : "1. Item A, 2. Item B" séparé de "3. Item C"
• Code non exécutable : Fonctions incomplètes, syntaxe cassée
• Citations tronquées : Perte du contexte source

La solution

Implémentez une détection et protection automatique des éléments critiques avant le chunking.

Éléments à protéger

import re
from typing import List, Tuple

def detect_atomic_blocks(text: str) -> List[Tuple[int, int, str]]:
 """
 Détecte les blocs atomiques qui ne doivent pas être coupés.

 Returns:
 Liste de (start_pos, end_pos, block_type)
 """
 atomic_blocks = []

 # 1. Blocs de code (```...```)
 for match in re.finditer(r'```[\s\S]*?```', text):
 atomic_blocks.append((match.start(), match.end(), "code_block"))

 # 2. Formules LaTeX ($...$ ou $$...$$)
 for match in re.finditer(r'\$\$[\s\S]*?\$\$|\$[^\$]+\$', text):
 atomic_blocks.append((match.start(), match.end(), "latex_formula"))

 # 3. Tableaux Markdown
 table_pattern = r'(\|[^\n]+\|\n)(\|[-:\s]+\|\n)((?:\|[^\n]+\|\n)+)'
 for match in re.finditer(table_pattern, text):
 atomic_blocks.append((match.start(), match.end(), "markdown_table"))

 # 4. Listes ordonnées complètes
 list_pattern = r'((?:^\d+\.\s+.+$\n?)+)'
 for match in re.finditer(list_pattern, text, re.MULTILINE):
 if len(match.group(0).split('\n')) >= 2: # Au moins 2 items
 atomic_blocks.append((match.start(), match.end(), "ordered_list"))

 # 5. Blocs de citation (>...)
 quote_pattern = r'((?:^>\s*.+$\n?)+)'
 for match in re.finditer(quote_pattern, text, re.MULTILINE):
 atomic_blocks.append((match.start(), match.end(), "quote_block"))

 # 6. URLs complètes
 url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+'
 for match in re.finditer(url_pattern, text):
 atomic_blocks.append((match.start(), match.end(), "url"))

 # 7. Équations chimiques (simplifié)
 chemical_pattern = r'\b[A-Z][a-z]?\d*(?:[A-Z][a-z]?\d*)*\b'
 for match in re.finditer(chemical_pattern, text):
 # Vérifier que c'est vraiment une formule (longueur, présence de chiffres)
 if re.search(r'\d', match.group()) and len(match.group()) > 3:
 atomic_blocks.append((match.start(), match.end(), "chemical_formula"))

 # Trier par position
 atomic_blocks.sort(key=lambda x: x[0])
 return atomic_blocks

def smart_split_respecting_blocks(
 text: str,
 target_chunk_size: int = 512,
 overlap: int = 50
) -> List[str]:
 """
 Découpe le texte en respectant les blocs atomiques.
 """
 import tiktoken
 encoding = tiktoken.encoding_for_model("gpt-4")

 # 1. Détecter tous les blocs atomiques
 atomic_blocks = detect_atomic_blocks(text)
 print(f"Détecté {len(atomic_blocks)} blocs atomiques")

 # 2. Créer des zones interdites de coupure
 forbidden_ranges = set()
 for start, end, block_type in atomic_blocks:
 forbidden_ranges.update(range(start, end))

 # 3. Découpage intelligent
 chunks = []
 current_chunk = ""
 current_pos = 0

 # Séparateurs naturels par ordre de préférence
 separators = ["\n\n", "\n", ". ", "! ", "? ", "; ", ", ", " "]

 while current_pos < len(text):
 # Calculer l'espace disponible
 remaining_space = target_chunk_size - len(encoding.encode(current_chunk))

 if remaining_space <= 0:
 # Chunk plein, chercher point de coupure valide
 cut_pos = find_safe_cut_point(
 text,
 current_pos,
 forbidden_ranges,
 separators
 )

 if cut_pos:
 chunks.append(current_chunk)
 # Démarrer nouveau chunk avec overlap
 overlap_text = current_chunk[-overlap:] if len(current_chunk) > overlap else current_chunk
 current_chunk = overlap_text + text[current_pos:cut_pos]
 current_pos = cut_pos
 else:
 # Aucun point de coupure sûr trouvé, forcer l'inclusion du bloc
 current_chunk += text[current_pos]
 current_pos += 1
 else:
 # Ajouter caractère par caractère
 current_chunk += text[current_pos]
 current_pos += 1

 # Ajouter le dernier chunk
 if current_chunk:
 chunks.append(current_chunk)

 return chunks

def find_safe_cut_point(
 text: str,
 start_pos: int,
 forbidden_ranges: set,
 separators: List[str]
) -> int:
 """
 Trouve un point de coupure sûr qui ne tombe pas dans un bloc atomique.
 """
 # Chercher le prochain séparateur
 for separator in separators:
 pos = text.find(separator, start_pos)
 if pos != -1:
 # Vérifier que ce n'est pas dans une zone interdite
 if pos not in forbidden_ranges:
 return pos + len(separator)

 # Aucun point sûr trouvé
 return None

# Usage
text = """
# Documentation API

Voici un exemple de requête :

```python
import requests

def fetch_data():
 response = requests.get("https://api.huggingface.co/data")
 return response.json()
```

La formule de calcul est : $E = mc^2$

| Paramètre | Type | Description |
|-----------|------|-------------|
| user_id | int | ID utilisateur |
| token | str | Token d'auth |

Liste des étapes :
1. Authentification
2. Récupération des données
3. Traitement
4. Retour du résultat
"""

chunks = smart_split_respecting_blocks(text, target_chunk_size=200)

for i, chunk in enumerate(chunks):
 print(f"\n=== CHUNK {i+1} ===")
 print(chunk)
 print(f"Tokens: {len(encoding.encode(chunk))}")

Erreur 8 : Négliger le preprocessing

Le problème

Appliquer le chunking directement sur le contenu brut non nettoyé : HTML avec balises, PDFs avec artefacts OCR, encodages cassés, espaces multiples, sauts de ligne incohérents, caractères spéciaux mal encodés. Le garbage in, garbage out s'applique pleinement.

Exemple : Un PDF scanné avec OCR imparfait contenant "L e p r o d u i t" (espaces insérés) sera indexé ainsi, rendant la recherche "produit" inefficace.

Conséquences

• Embeddings de mauvaise qualité : Le bruit pollue les vecteurs
• Recherche inefficace : Les requêtes ne matchent pas à cause des artefacts
• Tokens gaspillés : Balises HTML, metadata invisible prennent de la place
• Duplication invisible : Espaces/sauts de ligne différents = chunks considérés distincts
• Problèmes d'encodage : Caractères étranges (é au lieu de é) cassent la compréhension
• Performance dégradée : Plus de chunks inutiles = base plus lourde

La solution

Implémentez un pipeline de preprocessing robuste et systématique avant le chunking.

Pipeline de preprocessing recommandé

import re
import unicodedata
from bs4 import BeautifulSoup
from typing import Optional
import ftfy # pip install ftfy

class DocumentPreprocessor:
 """
 Pipeline de preprocessing complet pour nettoyer les documents.
 """

 def __init__(self, source_type: str):
 """
 Args:
 source_type: 'html', 'pdf', 'text', 'markdown'
 """
 self.source_type = source_type

 def preprocess(self, text: str) -> str:
 """
 Pipeline complet de nettoyage.
 """
 # 1. Correction d'encodage
 text = self._fix_encoding(text)

 # 2. Nettoyage spécifique au type
 if self.source_type == 'html':
 text = self._clean_html(text)
 elif self.source_type == 'pdf':
 text = self._clean_pdf_artifacts(text)

 # 3. Normalisation Unicode
 text = self._normalize_unicode(text)

 # 4. Nettoyage des espaces
 text = self._normalize_whitespace(text)

 # 5. Correction de ponctuation
 text = self._fix_punctuation(text)

 # 6. Suppression du contenu non informatif
 text = self._remove_boilerplate(text)

 # 7. Normalisation des sauts de ligne
 text = self._normalize_line_breaks(text)

 # 8. Validation finale
 text = self._final_validation(text)

 return text.strip()

 def _fix_encoding(self, text: str) -> str:
 """Corrige les problèmes d'encodage (mojibake)."""
 # ftfy répare automatiquement les encodages cassés
 return ftfy.fix_text(text)

 def _clean_html(self, text: str) -> str:
 """Supprime balises HTML et scripts."""
 soup = BeautifulSoup(text, 'html.parser')

 # Supprimer scripts, styles, metadata
 for élément in soup(['script', 'style', 'meta', 'link', 'noscript']):
 élément.decompose()

 # Extraire texte propre
 text = soup.get_text(separator=' ', strip=True)

 # Nettoyer les entités HTML restantes
 import html
 text = html.unescape(text)

 return text

 def _clean_pdf_artifacts(self, text: str) -> str:
 """Nettoie les artefacts typiques de l'OCR PDF."""

 # Supprimer les numéros de page isolés
 text = re.sub(r'^\s*\d+\s*$', '', text, flags=re.MULTILINE)

 # Corriger les espaces insérés entre lettres (artefact OCR)
 # "L e p r o d u i t" -> "Le produit"
 text = re.sub(r'\b(\w)\s+(\w)\s+(\w)', r'\1\2\3', text)

 # Supprimer les tirets de césure en fin de ligne
 # "connais-\nsance" -> "connaissance"
 text = re.sub(r'(\w+)-\s*\n\s*(\w+)', r'\1\2', text)

 # Supprimer headers/footers répétitifs
 # (Détecter lignes qui apparaissent plus de 3 fois)
 lines = text.split('\n')
 line_counts = {}
 for line in lines:
 stripped = line.strip()
 if len(stripped) > 10 and len(stripped) < 100:
 line_counts[stripped] = line_counts.get(stripped, 0) + 1

 # Supprimer lignes répétitives (headers/footers)
 repeated = {line for line, count in line_counts.items() if count > 3}
 text = '\n'.join([line for line in lines if line.strip() not in repeated])

 return text

 def _normalize_unicode(self, text: str) -> str:
 """Normalise les caractères Unicode."""
 # NFC : forme canonique composée
 text = unicodedata.normalize('NFC', text)

 # Supprimer caractères de contrôle invisibles (sauf \n, \t)
 text = ''.join(
 char for char in text
 if unicodedata.category(char)[0] != 'C' or char in '\n\t'
 )

 return text

 def _normalize_whitespace(self, text: str) -> str:
 """Normalise les espaces et tabulations."""
 # Remplacer tabs par espaces
 text = text.replace('\t', ' ')

 # Supprimer espaces multiples (sauf dans code blocks)
 text = re.sub(r' +', ' ', text)

 # Supprimer espaces en début/fin de ligne
 text = '\n'.join([line.strip() for line in text.split('\n')])

 return text

 def _fix_punctuation(self, text: str) -> str:
 """Corrige la ponctuation."""
 # Espace avant ponctuation (erreur fréquente OCR)
 text = re.sub(r'\s+([.,;:!?])', r'\1', text)

 # Espace après ponctuation manquant
 text = re.sub(r'([.,;:!?])([A-ZÀ-Ü])', r'\1 \2', text)

 # Points de suspension multiples
 text = re.sub(r'\.{4,}', '...', text)

 # Guillemets droits vs courbes (uniformiser)
 text = text.replace('‘', "'").replace('’', "'")
 text = text.replace('“', '"').replace('”', '"')

 return text

 def _remove_boilerplate(self, text: str) -> str:
 """Supprime le contenu répétitif non informatif."""
 # Patterns communs de boilerplate
 boilerplate_patterns = [
 r'(?i)copyright \u00a9? \d{4}.*',
 r'(?i)all rights reserved.*',
 r'(?i)confidential.*not to be distributed.*',
 r'(?i)printed on \d{1,2}/\d{1,2}/\d{2,4}',
 r'(?i)page \d+ of \d+',
 ]

 for pattern in boilerplate_patterns:
 text = re.sub(pattern, '', text)

 return text

 def _normalize_line_breaks(self, text: str) -> str:
 """Normalise les sauts de ligne."""
 # Supprimer plus de 3 sauts de ligne consécutifs
 text = re.sub(r'\n{4,}', '\n\n\n', text)

 # Séparer paragraphes clairement
 text = re.sub(r'\n\n+', '\n\n', text)

 return text

 def _final_validation(self, text: str) -> str:
 """Validation et nettoyage final."""
 # Supprimer lignes trop courtes (artefacts)
 lines = text.split('\n')
 cleaned_lines = [
 line for line in lines
 if len(line.strip()) == 0 or len(line.strip()) > 3
 ]

 text = '\n'.join(cleaned_lines)

 # Vérifier qu'il reste du contenu
 if len(text.strip()) < 50:
 raise ValueError("Document trop court après preprocessing (< 50 chars)")

 return text

# Usage complet
def process_document_safely(raw_content: str, source_type: str) -> str:
 """
 Traite un document avec gestion d'erreurs.
 """
 try:
 preprocessor = DocumentPreprocessor(source_type)
 cleaned = preprocessor.preprocess(raw_content)

 # Log des stats
 original_size = len(raw_content)
 cleaned_size = len(cleaned)
 reduction = (1 - cleaned_size / original_size) * 100

 print(f"Preprocessing : {original_size} -> {cleaned_size} chars (-{reduction:.1f}%)")

 return cleaned

 except Exception as e:
 print(f"Erreur preprocessing : {e}")
 # Fallback : nettoyage minimal
 return raw_content.strip()

# Exemple d'intégration dans pipeline complet
raw_pdf = """
P a g e 1

L e p r o d u i t e s t d i s p o n i b l e .

[Artefact OCR bizarre]

Copyright © 2024 Company Inc. All rights reserved.

----------
Page 2

Voici les caractéristiques :élément 1élément 2
"""

cleaned = process_document_safely(raw_pdf, source_type='pdf')
print("\n=== TEXTE NETTOYÉ ===")
print(cleaned)

# Puis chunking sur texte propre
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
 chunk_size=512,
 chunk_overlap=75
)
chunks = splitter.split_text(cleaned)
print(f"\nCréé {len(chunks)} chunks propres")

Erreur 9 : Ignorer les coûts et la qualité du retrieval

Le problème

Optimiser uniquement pour la vitesse d'indexation sans mesurer la qualité du retrieval ni les coûts réels (tokens API, stockage vector DB, latence utilisateur). Un chunking rapide qui produit des résultats médiocres est inutile.

Erreur classique : "Mon indexation prend 2 minutes au lieu de 10, c'est parfait !" mais les utilisateurs ne trouvent plus rien et la facture API a doublé.

Conséquences

• Fausse optimisation : Rapide mais inefficace = temps perdu
• Coûts cachés : Plus de chunks = plus d'embeddings = facture multipliée
• Expérience dégradée : Résultats non pertinents = utilisateurs frustrés
• Scalabilité compromise : Coûts exponentiels avec la croissance
• Décisions basées sur mauvaises métriques : Optimiser le mauvais indicateur

La solution

Implémentez un système de métriques holistique couvrant qualité, coûts et performance. Pour approfondir, consultez Human-AI Collaboration 2026 : Travailler avec des Agents.

Métriques de qualité essentielles

from dataclasses import dataclass
from typing import List, Dict
import time
import numpy as np

@dataclass
class ChunkingMetrics:
 """Métriques complètes d'une stratégie de chunking."""

 # Qualité
 precision_at_3: float
 precision_at_5: float
 recall_at_5: float
 mrr: float # Mean Reciprocal Rank
 hit_rate: float

 # Coûts
 embedding_cost_per_doc: float # USD
 storage_cost_per_month: float # USD
 generation_cost_per_query: float # USD

 # Performance
 avg_indexing_latency_ms: float
 avg_search_latency_ms: float
 p95_latency_ms: float

 # Efficacité
 avg_chunks_per_doc: float
 avg_tokens_per_chunk: float
 total_chunks: int
 total_tokens: int

class ChunkingEvaluator:
 """
 Évalue compréhensivement une stratégie de chunking.
 """

 def __init__(
 self,
 embedding_cost_per_1k_tokens: float = 0.00002, # text-embedding-3-small
 storage_cost_per_1m_vectors: float = 0.25, # Pinecone
 generation_cost_per_1k_tokens: float = 0.0005 # GPT-4o-mini
 ):
 self.embedding_cost = embedding_cost_per_1k_tokens
 self.storage_cost = storage_cost_per_1m_vectors
 self.generation_cost = generation_cost_per_1k_tokens

 def evaluate(
 self,
 chunks: List[str],
 test_queries: List[Dict], # {"query": str, "relevant_chunks": List[int]}
 retrieval_results: List[List[int]] # Chunks retrieved per query
 ) -> ChunkingMetrics:
 """
 Évalue toutes les métriques.
 """
 import tiktoken
 encoding = tiktoken.encoding_for_model("gpt-4")

 # 1. Métriques de qualité
 precisions_3 = []
 precisions_5 = []
 recalls_5 = []
 reciprocal_ranks = []
 hits = 0

 for i, query_data in enumerate(test_queries):
 relevant = set(query_data["relevant_chunks"])
 retrieved = retrieval_results[i]

 # Précision@K
 top_3 = set(retrieved[:3])
 top_5 = set(retrieved[:5])
 precisions_3.append(len(top_3 & relevant) / 3 if top_3 else 0)
 precisions_5.append(len(top_5 & relevant) / 5 if top_5 else 0)

 # Rappel@5
 recalls_5.append(len(top_5 & relevant) / len(relevant) if relevant else 0)

 # MRR : position du premier résultat pertinent
 rank = None
 for pos, chunk_id in enumerate(retrieved, 1):
 if chunk_id in relevant:
 rank = pos
 hits += 1
 break
 reciprocal_ranks.append(1 / rank if rank else 0)

 precision_at_3 = np.mean(precisions_3)
 precision_at_5 = np.mean(precisions_5)
 recall_at_5 = np.mean(recalls_5)
 mrr = np.mean(reciprocal_ranks)
 hit_rate = hits / len(test_queries)

 # 2. Métriques de coûts
 total_tokens = sum(len(encoding.encode(chunk)) for chunk in chunks)
 avg_tokens_per_chunk = total_tokens / len(chunks)

 # Coût embedding (une fois à l'indexation)
 embedding_cost_total = (total_tokens / 1000) * self.embedding_cost
 embedding_cost_per_doc = embedding_cost_total # Si 1 doc

 # Coût stockage mensuel
 num_vectors = len(chunks)
 storage_cost_month = (num_vectors / 1_000_000) * self.storage_cost

 # Coût génération par requête (en moyenne 5 chunks * avg_tokens)
 avg_context_tokens = 5 * avg_tokens_per_chunk
 generation_cost_query = (avg_context_tokens / 1000) * self.generation_cost

 # 3. Simulation de performance (remplacer par mesures réelles)
 # Ici on simule, en production vous mesurez réellement
 avg_indexing_latency = 50 # ms
 avg_search_latency = 120 # ms
 p95_latency = 180 # ms

 # 4. Métriques d'efficacité
 avg_chunks_per_doc = len(chunks) # Si 1 doc

 return ChunkingMetrics(
 # Qualité
 precision_at_3=precision_at_3,
 precision_at_5=precision_at_5,
 recall_at_5=recall_at_5,
 mrr=mrr,
 hit_rate=hit_rate,
 # Coûts
 embedding_cost_per_doc=embedding_cost_per_doc,
 storage_cost_per_month=storage_cost_month,
 generation_cost_per_query=generation_cost_query,
 # Performance
 avg_indexing_latency_ms=avg_indexing_latency,
 avg_search_latency_ms=avg_search_latency,
 p95_latency_ms=p95_latency,
 # Efficacité
 avg_chunks_per_doc=avg_chunks_per_doc,
 avg_tokens_per_chunk=avg_tokens_per_chunk,
 total_chunks=len(chunks),
 total_tokens=total_tokens
 )

 def print_report(self, metrics: ChunkingMetrics):
 """Affiche un rapport lisible."""
 print("\n" + "="*60)
 print(" RAPPORT D'ÉVALUATION CHUNKING")
 print("="*60)

 print("\n🎯 QUALITÉ DU RETRIEVAL")
 print(f" Précision@3: {metrics.precision_at_3:.1%}")
 print(f" Précision@5: {metrics.precision_at_5:.1%}")
 print(f" Rappel@5: {metrics.recall_at_5:.1%}")
 print(f" MRR: {metrics.mrr:.3f}")
 print(f" Hit Rate: {metrics.hit_rate:.1%}")

 print("\n💰 COÛTS")
 print(f" Embedding/doc: ${metrics.embedding_cost_per_doc:.4f}")
 print(f" Stockage/mois: ${metrics.storage_cost_per_month:.2f}")
 print(f" Génération/query: ${metrics.generation_cost_per_query:.4f}")

 # Projection 1000 docs, 10K requêtes/mois
 total_monthly = (
 metrics.embedding_cost_per_doc * 1000 + # 1000 docs
 metrics.storage_cost_per_month +
 metrics.generation_cost_per_query * 10000 # 10K queries
 )
 print(f" TOTAL (1K docs, 10K queries/mois): ${total_monthly:.2f}")

 print("\n⏱️ PERFORMANCE")
 print(f" Latence indexation: {metrics.avg_indexing_latency_ms:.0f}ms")
 print(f" Latence recherche: {metrics.avg_search_latency_ms:.0f}ms")
 print(f" P95 latence: {metrics.p95_latency_ms:.0f}ms")

 print("\n📊 EFFICACITÉ")
 print(f" Chunks/document: {metrics.avg_chunks_per_doc:.1f}")
 print(f" Tokens/chunk: {metrics.avg_tokens_per_chunk:.0f}")
 print(f" Total chunks: {metrics.total_chunks:,}")
 print(f" Total tokens: {metrics.total_tokens:,}")

 # Score global (pondéré)
 quality_score = (metrics.precision_at_5 + metrics.recall_at_5) / 2
 cost_score = 1 - min(total_monthly / 1000, 1) # Normaliser
 perf_score = 1 - min(metrics.avg_search_latency_ms / 1000, 1)

 global_score = (quality_score * 0.5 + cost_score * 0.3 + perf_score * 0.2)

 print(f"\n⭐ SCORE GLOBAL: {global_score:.1%}")
 print("="*60 + "\n")

# Usage
evaluator = ChunkingEvaluator()

# Exemple de résultats
test_queries = [
 {"query": "comment installer", "relevant_chunks": [0, 5, 12]},
 {"query": "coûts abonnement", "relevant_chunks": [23, 24]},
 # ...
]

retrieval_results = [
 [0, 5, 3, 12, 8], # Résultats pour query 1
 [23, 15, 24, 7, 9], # Résultats pour query 2
 # ...
]

metrics = evaluator.evaluate(chunks, test_queries, retrieval_results)
evaluator.print_report(metrics)

Points d'attention spécifiques

Erreur 10 : Configuration statique sans monitoring

Le problème

Définir une stratégie de chunking une fois au lancement puis ne jamais la monitorer ni l'ajuster. Vos données évoluent, les patterns de requêtes changent, de nouveaux types de documents arrivent, mais votre chunking reste figé dans le temps.

Exemple réel : Une entreprise avait configuré son chunking pour des articles courts. 6 mois plus tard, 40% des documents étaient des rapports techniques longs. Personne ne s'est rendu compte que la qualité avait chuté de 30%.

Conséquences

• Dégradation silencieuse : La qualité baisse progressivement sans alarme
• Inadaptation aux évolutions : Nouveaux types de docs mal gérés
• Coûts incontrôlés : Explosion des coûts sans explication
• Pas de détection d'anomalies : Bugs ou régressions non repérés
• Optimisations manquées : Opportunités d'amélioration invisibles
• Incidents en production : Problèmes découverts par les utilisateurs frustrés

La solution

Implémentez un système de monitoring continu avec alertes automatiques et ajustements adaptatifs.

Dashboard de monitoring recommandé

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json

@dataclass
class ChunkingHealthMetrics:
 """Métriques de santé du système de chunking."""
 timestamp: datetime

 # Distribution
 avg_chunk_size: float
 median_chunk_size: float
 p95_chunk_size: float
 std_chunk_size: float

 # Qualité
 avg_retrieval_score: float # 0-1
 user_satisfaction_rate: float # %
 failed_searches_rate: float # %

 # Volume
 total_chunks: int
 new_chunks_24h: int
 deleted_chunks_24h: int

 # Coûts
 daily_embedding_cost: float
 daily_storage_cost: float
 daily_generation_cost: float

 # Performance
 avg_indexing_time_ms: float
 avg_search_time_ms: float
 error_rate: float # %

class ChunkingMonitor:
 """
 Système de monitoring continu du chunking.
 """

 def __init__(self, alert_thresholds: Dict):
 self.thresholds = alert_thresholds
 self.metrics_history: List[ChunkingHealthMetrics] = []

 def collect_metrics(self) -> ChunkingHealthMetrics:
 """
 Collecte les métriques actuelles du système.
 (Implémentez selon votre infra : Prometheus, CloudWatch, etc.)
 """
 # Exemple simulé - remplacer par vraies métriques
 return ChunkingHealthMetrics(
 timestamp=datetime.now(),
 avg_chunk_size=487.3,
 median_chunk_size=512.0,
 p95_chunk_size=745.2,
 std_chunk_size=123.5,
 avg_retrieval_score=0.78,
 user_satisfaction_rate=0.82,
 failed_searches_rate=0.05,
 total_chunks=125_487,
 new_chunks_24h=1_243,
 deleted_chunks_24h=87,
 daily_embedding_cost=12.45,
 daily_storage_cost=3.21,
 daily_generation_cost=18.73,
 avg_indexing_time_ms=52.3,
 avg_search_time_ms=118.7,
 error_rate=0.02
 )

 def check_health(self, metrics: ChunkingHealthMetrics) -> List[str]:
 """
 Vérifie la santé et retourne les alertes si nécessaire.
 """
 alerts = []

 # 1. Vérifier distribution de taille
 if metrics.std_chunk_size > self.thresholds['max_std_deviation']:
 alerts.append(
 f"⚠️ ALERTE : Variance de taille trop élevée ({metrics.std_chunk_size:.1f}). "
 f"Risque de chunks trop petits ou trop grands."
 )

 # 2. Vérifier qualité du retrieval
 if metrics.avg_retrieval_score < self.thresholds['min_retrieval_score']:
 alerts.append(
 f"🔴 CRITIQUE : Score de retrieval bas ({metrics.avg_retrieval_score:.1%}). "
 f"Seuil minimum : {self.thresholds['min_retrieval_score']:.1%}"
 )

 # 3. Vérifier satisfaction utilisateur
 if metrics.user_satisfaction_rate < self.thresholds['min_satisfaction']:
 alerts.append(
 f"🟠 ATTENTION : Satisfaction utilisateur en baisse ({metrics.user_satisfaction_rate:.1%})"
 )

 # 4. Vérifier taux d'échec
 if metrics.failed_searches_rate > self.thresholds['max_failure_rate']:
 alerts.append(
 f"🔴 CRITIQUE : Trop de recherches échouées ({metrics.failed_searches_rate:.1%})"
 )

 # 5. Vérifier coûts
 daily_total = metrics.daily_embedding_cost + metrics.daily_storage_cost + metrics.daily_generation_cost
 if daily_total > self.thresholds['max_daily_cost']:
 alerts.append(
 f"💰 ALERTE COÛT : Dépassement du budget (${daily_total:.2f}/jour)"
 )

 # 6. Vérifier performance
 if metrics.avg_search_time_ms > self.thresholds['max_search_latency_ms']:
 alerts.append(
 f"⏱️ PERFORMANCE : Latence de recherche élevée ({metrics.avg_search_time_ms:.0f}ms)"
 )

 # 7. Détecter anomalies (comparaison avec historique)
 if len(self.metrics_history) > 7:
 anomalies = self.detect_anomalies(metrics)
 alerts.extend(anomalies)

 return alerts

 def detect_anomalies(self, current: ChunkingHealthMetrics) -> List[str]:
 """
 Détecte les anomalies par rapport à l'historique.
 """
 alerts = []

 # Calculer moyennes sur 7 derniers jours
 recent = self.metrics_history[-7:]
 avg_chunk_size_7d = sum(m.avg_chunk_size for m in recent) / len(recent)
 avg_cost_7d = sum(
 m.daily_embedding_cost + m.daily_storage_cost + m.daily_generation_cost
 for m in recent
 ) / len(recent)

 # Détecter variations anormales (> 20%)
 chunk_size_change = abs(current.avg_chunk_size - avg_chunk_size_7d) / avg_chunk_size_7d
 if chunk_size_change > 0.20:
 alerts.append(
 f"📈 ANOMALIE : Taille moyenne chunks a varié de {chunk_size_change:.1%} "
 f"({avg_chunk_size_7d:.0f} -> {current.avg_chunk_size:.0f} tokens)"
 )

 current_daily_cost = current.daily_embedding_cost + current.daily_storage_cost + current.daily_generation_cost
 cost_change = abs(current_daily_cost - avg_cost_7d) / avg_cost_7d
 if cost_change > 0.25:
 alerts.append(
 f"💸 ANOMALIE COÛT : Coûts ont varié de {cost_change:.1%} "
 f"(${avg_cost_7d:.2f} -> ${current_daily_cost:.2f})"
 )

 return alerts

 def auto_tune_recommendations(self, metrics: ChunkingHealthMetrics) -> List[str]:
 """
 Génère des recommandations d'optimisation automatiques.
 """
 recommendations = []

 # 1. Si chunks trop variables
 if metrics.std_chunk_size > 150:
 recommendations.append(
 "Considérer un chunking plus uniforme ou augmenter le preprocessing"
 )

 # 2. Si qualité en baisse mais coûts OK
 if metrics.avg_retrieval_score < 0.75 and metrics.daily_embedding_cost < 20:
 recommendations.append(
 "Augmenter la taille des chunks de 10-15% pour plus de contexte"
 )

 # 3. Si coûts élevés mais qualité OK
 if metrics.daily_embedding_cost > 30 and metrics.avg_retrieval_score > 0.85:
 recommendations.append(
 "Réduire la taille des chunks ou l'overlap pour optimiser les coûts"
 )

 # 4. Si latence élevée
 if metrics.avg_search_time_ms > 200:
 recommendations.append(
 "Optimiser l'indexation vectorielle ou réduire top_k"
 )

 # 5. Si taux d'échec élevé
 if metrics.failed_searches_rate > 0.10:
 recommendations.append(
 "Auditer les requêtes échouées et ajuster la stratégie de chunking"
 )

 return recommendations

 def generate_report(self, metrics: ChunkingHealthMetrics) -> str:
 """
 Génère un rapport complet.
 """
 alerts = self.check_health(metrics)
 recommendations = self.auto_tune_recommendations(metrics)

 report = f"""
┌────────────────────────────────────────────────────────────┐
│ RAPPORT DE MONITORING CHUNKING - {metrics.timestamp.strftime('%Y-%m-%d %H:%M')} │
└────────────────────────────────────────────────────────────┘

📊 DISTRIBUTION DES CHUNKS
 Taille moyenne: {metrics.avg_chunk_size:.0f} tokens
 Médiane: {metrics.median_chunk_size:.0f} tokens
 95e percentile: {metrics.p95_chunk_size:.0f} tokens
 Écart-type: {metrics.std_chunk_size:.0f} tokens

🎯 QUALITÉ
 Score retrieval: {metrics.avg_retrieval_score:.1%} {' ✅' if metrics.avg_retrieval_score > 0.75 else ' ⚠️'}
 Satisfaction: {metrics.user_satisfaction_rate:.1%} {' ✅' if metrics.user_satisfaction_rate > 0.80 else ' ⚠️'}
 Taux d'échec: {metrics.failed_searches_rate:.1%} {' ✅' if metrics.failed_searches_rate < 0.05 else ' ⚠️'}

💾 VOLUME
 Total chunks: {metrics.total_chunks:,}
 Nouveaux (24h): {metrics.new_chunks_24h:,}
 Supprimés (24h): {metrics.deleted_chunks_24h:,}

💰 COÛTS (quotidien)
 Embedding: ${metrics.daily_embedding_cost:.2f}
 Stockage: ${metrics.daily_storage_cost:.2f}
 Génération: ${metrics.daily_generation_cost:.2f}
 TOTAL: ${metrics.daily_embedding_cost + metrics.daily_storage_cost + metrics.daily_generation_cost:.2f}

⏱️ PERFORMANCE
 Indexation: {metrics.avg_indexing_time_ms:.0f}ms {' ✅' if metrics.avg_indexing_time_ms < 100 else ' ⚠️'}
 Recherche: {metrics.avg_search_time_ms:.0f}ms {' ✅' if metrics.avg_search_time_ms < 150 else ' ⚠️'}
 Taux d'erreur: {metrics.error_rate:.2%} {' ✅' if metrics.error_rate < 0.05 else ' ⚠️'}
"""

 if alerts:
 report += "\n🚨 ALERTES\n"
 for alert in alerts:
 report += f" {alert}\n"

 if recommendations:
 report += "\n💡 RECOMMANDATIONS\n"
 for i, rec in enumerate(recommendations, 1):
 report += f" {i}. {rec}\n"

 if not alerts:
 report += "\n✅ Tous les indicateurs sont dans les normes\n"

 return report

# Configuration des seuils
thresholds = {
 'max_std_deviation': 150,
 'min_retrieval_score': 0.70,
 'min_satisfaction': 0.75,
 'max_failure_rate': 0.08,
 'max_daily_cost': 50.0,
 'max_search_latency_ms': 200
}

# Initialisation du monitoring
monitor = ChunkingMonitor(alert_thresholds=thresholds)

# Exécution périodique (cron job quotidien)
metrics = monitor.collect_metrics()
monitor.metrics_history.append(metrics)

report = monitor.generate_report(metrics)
print(report)

# Envoyer alertes (email, Slack, PagerDuty)
alerts = monitor.check_health(metrics)
if alerts:
 # send_to_slack(report) # Votre intégration
 # send_email_alert(alerts) # Votre intégration
 pass

# Sauvegarder historique
with open('chunking_metrics_history.jsonl', 'a') as f:
 f.write(json.dumps({
 'timestamp': metrics.timestamp.isoformat(),
 'metrics': vars(metrics)
 }) + '\n')

Monitoring et alertes en production

Checklist de validation avant production

Avant de déployer votre système de chunking en production, validez systématiquement ces points critiques.

Questions à se poser

Tests obligatoires

Documentation requise