10 Erreurs Courantes dans - Guide Pratique Cybersecurite

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 ?

Analyse approfondie et recommandations

• 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.

Outils et ressources complementaires

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": "Equipe 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 :

Mise en pratique et recommandations

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.

Mise en pratique et recommandations

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.

Mise en pratique et recommandations

É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 element in soup(['script', 'style', 'meta', 'link', 'noscript']):
            element.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")

Aspects techniques complementaires

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.

Mise en pratique et recommandations

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 specifiques

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.

Mise en pratique et recommandations

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