Optimiser le Chunking de Documents pour le RAG

Exemple : 10 000 documents (100 pages chacun)

• Chunks 256 tokens : ~4M chunks, 15 GB embeddings (Ada-002), coût indexation $320
• Chunks 512 tokens : ~2M chunks, 7.5 GB embeddings, coût indexation $160
• Chunks 1024 tokens : ~1M chunks, 3.8 GB embeddings, coût indexation $80

# Exemple : découpage en tokens
import tiktoken

encoder = tiktoken.encoding_for_model("gpt-4")
text = "Votre document à découper..."
tokens = encoder.encode(text)
print(f"Nombre de tokens : {len(tokens)}")  # Ex: 1247 tokens

# Validation de taille avant embedding
MAX_TOKENS = 512  # Pour BGE-large

for chunk in chunks:
    token_count = len(encoder.encode(chunk))
    if token_count > MAX_TOKENS:
        logger.warning(f"Chunk trop large : {token_count} tokens (max {MAX_TOKENS})")
        # Option 1 : Re-chunker
        # Option 2 : Tronquer avec warning

# Fixed-size chunking avec LangChain
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=100,
    length_function=len,  # Caractères (utiliser tiktoken pour tokens)
    separators=["\n\n", "\n", ". ", " ", ""]  # Hiérarchie de séparateurs
)

chunks = splitter.split_text(document_text)

# Semantic chunking avec embeddings
import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
sentences = text.split('. ')
embeddings = model.encode(sentences)

chunks = []
current_chunk = [sentences[0]]

for i in range(1, len(sentences)):
    similarity = np.dot(embeddings[i], embeddings[i-1])

    if similarity > 0.75:  # Seuil de cohérence
        current_chunk.append(sentences[i])
    else:
        chunks.append('. '.join(current_chunk))
        current_chunk = [sentences[i]]

chunks.append('. '.join(current_chunk))

# Structure-based chunking pour Markdown
import re

def chunk_by_markdown_sections(markdown_text):
    # Découpage par headers de niveau 2
    sections = re.split(r'\n## ', markdown_text)
    chunks = []

    for section in sections:
        # Si section trop grande, sous-découper par paragraphes
        if len(section) > 1500:
            paragraphs = section.split('\n\n')
            chunks.extend([p for p in paragraphs if len(p) > 100])
        else:
            chunks.append(section)

    return chunks

# Recursive chunking (implémentation LangChain)
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=[
        "\n\n\n",  # Sections
        "\n\n",    # Paragraphes
        "\n",      # Lignes
        ". ",      # Phrases
        " ",       # Mots
        ""         # Caractères (fallback)
    ]
)

# Essaie de découper par sections, si trop grand essaie paragraphes, etc.
chunks = splitter.split_text(document)

# Sentence-window avec LlamaIndex
from llama_index.node_parser import SentenceWindowNodeParser

parser = SentenceWindowNodeParser(
    window_size=3,  # 3 phrases avant/après
    window_metadata_key="window",
    original_text_metadata_key="original_sentence"
)

nodes = parser.get_nodes_from_documents(documents)

# Au retrieval : recherche sur phrase unique, retourne contexte étendu

Chunk 1 : tokens [0-500]
Chunk 2 : tokens [400-900]    ← 100 tokens en commun avec Chunk 1
Chunk 3 : tokens [800-1300]   ← 100 tokens en commun avec Chunk 2

def deduplicate_overlapping_chunks(chunks, overlap_threshold=0.5):
    """Retire les chunks qui se chevauchent trop."""
    deduplicated = [chunks[0]]  # Garde le plus pertinent (rank 1)

    for chunk in chunks[1:]:
        # Vérifie si chevauchement avec chunks déjà sélectionnés
        is_duplicate = False
        for selected in deduplicated:
            overlap_ratio = compute_text_overlap(chunk, selected)
            if overlap_ratio > overlap_threshold:
                is_duplicate = True
                break

        if not is_duplicate:
            deduplicated.append(chunk)

    return deduplicated[:10]  # Garde top-10 uniques

def merge_overlapping_chunks(chunks):
    """Fusionne les chunks overlapés en un seul contexte continu."""
    if not chunks:
        return []

    # Trier par position dans document source
    chunks = sorted(chunks, key=lambda c: c['start_pos'])

    merged = [chunks[0]['text']]
    last_end = chunks[0]['end_pos']

    for chunk in chunks[1:]:
        if chunk['start_pos'] < last_end:  # Overlap détecté
            # Ajoute uniquement la partie non overlapée
            non_overlap_start = last_end - chunk['start_pos']
            merged.append(chunk['text'][non_overlap_start:])
        else:
            merged.append(chunk['text'])

        last_end = max(last_end, chunk['end_pos'])

    return ' '.join(merged)

# Chunking de code Python par fonctions
import ast

def chunk_python_code(source_code):
    tree = ast.parse(source_code)
    chunks = []

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
            # Extraire le code de la fonction/classe
            start_line = node.lineno
            end_line = node.end_lineno
            func_code = '\n'.join(source_code.split('\n')[start_line-1:end_line])

            chunks.append({
                'type': type(node).__name__,
                'name': node.name,
                'code': func_code,
                'docstring': ast.get_docstring(node) or ""
            })

    return chunks

# Chunking de tableaux avec header répété
import pandas as pd

def chunk_table_with_headers(df, rows_per_chunk=50):
    chunks = []
    header = df.columns.tolist()

    for i in range(0, len(df), rows_per_chunk):
        chunk_df = df.iloc[i:i+rows_per_chunk]

        # Formater en texte lisible
        chunk_text = f"Tableau (lignes {i+1}-{i+len(chunk_df)}) :\n"
        chunk_text += f"Colonnes : {', '.join(header)}\n\n"

        for idx, row in chunk_df.iterrows():
            row_text = ' | '.join([f"{col}={row[col]}" for col in header])
            chunk_text += row_text + "\n"

        chunks.append(chunk_text)

    return chunks

# Extraction PDF avec unstructured
from unstructured.partition.pdf import partition_pdf
from unstructured.chunking.title import chunk_by_title

# Extraction avec détection de layout
elements = partition_pdf(
    filename="document.pdf",
    strategy="hi_res",  # OCR si nécessaire
    infer_table_structure=True
)

# Chunking par sections (détectées automatiquement)
chunks = chunk_by_title(
    elements,
    max_characters=1000,
    combine_text_under_n_chars=200
)

# Chunking multilingue avec détection
from langdetect import detect_langs

def chunk_multilingual(text, chunk_size_by_lang={'en': 600, 'fr': 512, 'es': 520}):
    paragraphs = text.split('\n\n')
    chunks = []
    current_chunk = []
    current_lang = None
    current_size = 0

    for para in paragraphs:
        # Détecter langue du paragraphe
        try:
            lang = detect_langs(para)[0].lang
        except:
            lang = current_lang or 'en'

        para_tokens = len(encoder.encode(para))
        max_size = chunk_size_by_lang.get(lang, 512)

        # Nouveau chunk si changement langue ou dépassement taille
        if (current_lang and lang != current_lang) or (current_size + para_tokens > max_size):
            chunks.append({
                'text': '\n\n'.join(current_chunk),
                'language': current_lang
            })
            current_chunk = [para]
            current_size = para_tokens
            current_lang = lang
        else:
            current_chunk.append(para)
            current_size += para_tokens
            current_lang = lang

    if current_chunk:
        chunks.append({'text': '\n\n'.join(current_chunk), 'language': current_lang})

    return chunks

# Exemple structure parent-child
parent_chunk = {
    'id': 'doc1_chapter3',
    'text': "Chapitre 3 : Sécurité des données\n\n[Contenu complet du chapitre - 1024 tokens]",
    'metadata': {
        'level': 'chapter',
        'title': 'Sécurité des données',
        'chapter_num': 3
    }
}

child_chunks = [
    {
        'id': 'doc1_chapter3_section1',
        'text': "3.1 Chiffrement\n\nLe chiffrement des données...",
        'parent_id': 'doc1_chapter3',
        'metadata': {
            'level': 'section',
            'title': 'Chiffrement',
            'breadcrumb': 'Chapitre 3 > 3.1 Chiffrement'
        }
    },
    # ... autres sections
]

# Exemple avec expansion de contexte
retrieved_chunk = vector_db.search(query, top_k=1)[0]

# Stratégie 1 : Expansion linéaire (contexte avant/après)
context_chunks = [
    vector_db.get_by_id(retrieved_chunk['metadata']['prev_chunk_id']),
    retrieved_chunk,
    vector_db.get_by_id(retrieved_chunk['metadata']['next_chunk_id'])
]

# Stratégie 2 : Expansion sémantique
related_ids = retrieved_chunk['metadata']['related_chunks']
context_chunks = [vector_db.get_by_id(id) for id in related_ids[:3]]

full_context = '\n\n---\n\n'.join([c['text'] for c in context_chunks])

# Exemple avec Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range

client = QdrantClient(url="http://localhost:6333")

# Cas 1 : Filtering par type de document
results = client.search(
    collection_name="docs",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="document_type", match=MatchValue(value="legal"))
        ]
    ),
    limit=10
)

# Cas 2 : Filtering temporel (documents récents)
results = client.search(
    collection_name="docs",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="date",
                range=Range(gte="2024-01-01")  # Depuis janvier 2024
            )
        ]
    ),
    limit=10
)

# Cas 3 : Filtering multi-critères
results = client.search(
    collection_name="docs",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="language", match=MatchValue(value="fr")),
            FieldCondition(key="tags", match=MatchValue(any=["RGPD", "sécurité"]))
        ],
        must_not=[
            FieldCondition(key="status", match=MatchValue(value="archived"))
        ]
    ),
    limit=10
)

# Calcul de métriques avec dataset de test
import numpy as np

def calculate_retrieval_metrics(queries, ground_truth, retrieval_function, k=10):
    precisions, recalls, mrr_scores = [], [], []

    for query, relevant_ids in zip(queries, ground_truth):
        # Récupérer top-k chunks
        retrieved = retrieval_function(query, k=k)
        retrieved_ids = [chunk['id'] for chunk in retrieved]

        # Precision@k
        relevant_retrieved = set(retrieved_ids) & set(relevant_ids)
        precision = len(relevant_retrieved) / k
        precisions.append(precision)

        # Recall@k
        recall = len(relevant_retrieved) / len(relevant_ids) if relevant_ids else 0
        recalls.append(recall)

        # MRR
        for rank, chunk_id in enumerate(retrieved_ids, 1):
            if chunk_id in relevant_ids:
                mrr_scores.append(1 / rank)
                break
        else:
            mrr_scores.append(0)

    return {
        'precision@k': np.mean(precisions),
        'recall@k': np.mean(recalls),
        'mrr': np.mean(mrr_scores)
    }

# Exemple d'utilisation
metrics = calculate_retrieval_metrics(
    queries=test_queries,
    ground_truth=test_relevant_chunks,
    retrieval_function=my_rag_retrieval,
    k=10
)

print(f"Precision@10: {metrics['precision@k']:.2%}")
print(f"Recall@10: {metrics['recall@k']:.2%}")
print(f"MRR: {metrics['mrr']:.3f}")

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

model = SentenceTransformer('all-MiniLM-L6-v2')

def calculate_chunk_coherence(chunk_text):
    """Mesure la cohérence sémantique interne d'un chunk."""
    sentences = chunk_text.split('. ')
    if len(sentences) < 2:
        return 1.0

    # Embeddings des phrases
    embeddings = model.encode(sentences)

    # Similarité moyenne entre toutes paires de phrases
    similarities = []
    for i in range(len(embeddings)):
        for j in range(i+1, len(embeddings)):
            sim = cosine_similarity([embeddings[i]], [embeddings[j]])[0][0]
            similarities.append(sim)

    return np.mean(similarities)

# Analyse d'un corpus de chunks
coherences = [calculate_chunk_coherence(chunk) for chunk in chunks]
print(f"Cohérence moyenne : {np.mean(coherences):.3f}")
print(f"Chunks faible cohérence (<0.5) : {sum(c < 0.5 for c in coherences)}")

# Grid search pour hyperparams chunking
import itertools
from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness

# Définir la grille de recherche
chunk_sizes = [256, 512, 768, 1024]
overlaps = [0.0, 0.1, 0.2, 0.3]
strategies = ['fixed', 'recursive', 'semantic']

results = []

for size, overlap, strategy in itertools.product(chunk_sizes, overlaps, strategies):
    print(f"Testing: size={size}, overlap={overlap}, strategy={strategy}")

    # Recréer chunks avec params
    chunks = create_chunks(
        documents=test_documents,
        chunk_size=size,
        chunk_overlap=int(size * overlap),
        strategy=strategy
    )

    # Ré-indexer base vectorielle
    vector_db.delete_all()
    vector_db.index(chunks)

    # Évaluer sur dataset de test
    rag_results = run_rag_evaluation(test_queries, vector_db)

    metrics = evaluate(
        dataset=rag_results,
        metrics=[context_precision, context_recall, faithfulness]
    )

    results.append({
        'chunk_size': size,
        'overlap': overlap,
        'strategy': strategy,
        **metrics
    })

# Trouver la meilleure config
best = max(results, key=lambda x: x['context_precision'])
print(f"\nBest config: {best}")

# Exemple évaluation avec RAGAS
from ragas import evaluate
from ragas.metrics import (
    context_precision,
    context_recall,
    faithfulness,
    answer_relevancy
)
from datasets import Dataset

# Préparer dataset d'évaluation
eval_data = {
    'question': ["Quel est le montant du contrat ?", ...],
    'contexts': [[chunk1, chunk2], ...],  # Chunks récupérés
    'answer': ["Le montant est 150K€", ...],  # Réponse générée
    'ground_truth': ["Le contrat est de 150 000€ sur 3 ans", ...]  # Référence
}

dataset = Dataset.from_dict(eval_data)

# Évaluer
results = evaluate(
    dataset=dataset,
    metrics=[context_precision, context_recall, faithfulness, answer_relevancy]
)

print(results)
# Output:
# {'context_precision': 0.82, 'context_recall': 0.89,
#  'faithfulness': 0.94, 'answer_relevancy': 0.87}

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    MarkdownHeaderTextSplitter,
    PythonCodeTextSplitter
)

# 1. Recursive (recommandé par défaut)
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=100,
    length_function=len,
    separators=["\n\n", "\n", ". ", " ", ""]
)

# 2. Markdown avec préservation structure
markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
)

# 3. Code Python spécialisé
code_splitter = PythonCodeTextSplitter(
    chunk_size=800,
    chunk_overlap=50
)

chunks = splitter.split_text(document_text)

from llama_index.node_parser import (
    SimpleNodeParser,
    SentenceSplitter,
    SemanticSplitterNodeParser,
    HierarchicalNodeParser
)
from llama_index.embeddings import OpenAIEmbedding

# 1. Sentence-based avec window
parser = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=100
)

# 2. Semantic chunking
embed_model = OpenAIEmbedding()
semantic_parser = SemanticSplitterNodeParser(
    buffer_size=1,
    embed_model=embed_model,
    breakpoint_percentile_threshold=95  # Seuil de coupure
)

# 3. Hiérarchique (parent-child)
hierarchical_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128]  # 3 niveaux
)

nodes = parser.get_nodes_from_documents(documents)

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant
import tiktoken

# 1. Charger documents
loader = PyPDFLoader("contract.pdf")
documents = loader.load()

# 2. Configurer splitter avec tokenizer
encoder = tiktoken.encoding_for_model("gpt-4")

def tiktoken_len(text):
    return len(encoder.encode(text))

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=100,
    length_function=tiktoken_len,  # Mesure en tokens GPT-4
    separators=["\n\n", "\n", ". ", ", ", " ", ""]
)

# 3. Chunker avec métadonnées
chunks = []
for doc in documents:
    splits = splitter.split_text(doc.page_content)

    for i, split in enumerate(splits):
        chunk = {
            'text': split,
            'metadata': {
                'source': doc.metadata['source'],
                'page': doc.metadata.get('page', 0),
                'chunk_index': i,
                'total_chunks': len(splits),
                'token_count': tiktoken_len(split)
            }
        }
        chunks.append(chunk)

print(f"Created {len(chunks)} chunks from {len(documents)} documents")

# 4. Indexer dans base vectorielle
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = Qdrant.from_texts(
    texts=[c['text'] for c in chunks],
    embedding=embeddings,
    metadatas=[c['metadata'] for c in chunks],
    url="http://localhost:6333",
    collection_name="contracts"
)

print("Indexing complete!")

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import spacy

class SemanticChunker:
    def __init__(
        self,
        model_name='all-MiniLM-L6-v2',
        similarity_threshold=0.7,
        max_chunk_size=800,
        min_chunk_size=200
    ):
        self.model = SentenceTransformer(model_name)
        self.nlp = spacy.load('fr_core_news_sm')  # Pour segmentation phrases
        self.similarity_threshold = similarity_threshold
        self.max_chunk_size = max_chunk_size
        self.min_chunk_size = min_chunk_size

    def split(self, text):
        # 1. Segmenter en phrases
        doc = self.nlp(text)
        sentences = [sent.text for sent in doc.sents]

        if len(sentences) == 0:
            return []

        # 2. Calculer embeddings des phrases
        embeddings = self.model.encode(sentences)

        # 3. Grouper par cohérence sémantique
        chunks = []
        current_chunk = [sentences[0]]
        current_size = len(sentences[0])

        for i in range(1, len(sentences)):
            sentence = sentences[i]
            sentence_size = len(sentence)

            # Calculer similarité avec chunk courant
            chunk_embedding = np.mean([embeddings[j] for j in range(i-len(current_chunk), i)], axis=0)
            sentence_embedding = embeddings[i]
            similarity = cosine_similarity([chunk_embedding], [sentence_embedding])[0][0]

            # Décision : ajouter au chunk courant ou créer nouveau chunk
            if (
                similarity > self.similarity_threshold
                and current_size + sentence_size < self.max_chunk_size
            ):
                # Ajouter au chunk courant
                current_chunk.append(sentence)
                current_size += sentence_size
            else:
                # Créer nouveau chunk
                if current_size >= self.min_chunk_size:
                    chunks.append(' '.join(current_chunk))
                    current_chunk = [sentence]
                    current_size = sentence_size
                else:
                    # Chunk trop petit, forcer ajout
                    current_chunk.append(sentence)
                    current_size += sentence_size

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

        return chunks

# Utilisation
chunker = SemanticChunker(
    similarity_threshold=0.75,
    max_chunk_size=600,
    min_chunk_size=200
)

text = """[Votre long document...]"""
chunks = chunker.split(text)

for i, chunk in enumerate(chunks, 1):
    print(f"\n=== Chunk {i} ({len(chunk)} chars) ===")
    print(chunk[:200] + "...")

import logging
from typing import List, Dict
from dataclasses import dataclass
from pathlib import Path

@dataclass
class Chunk:
    text: str
    metadata: Dict
    token_count: int
    coherence_score: float

class DocumentPreprocessingPipeline:
    def __init__(self, config):
        self.config = config
        self.logger = logging.getLogger(__name__)

    def process(self, file_path: Path) -> List[Chunk]:
        """Pipeline complet de preprocessing."""

        # 1. Extraction
        self.logger.info(f"Extracting {file_path}")
        raw_text = self.extract(file_path)

        # 2. Nettoyage
        self.logger.info("Cleaning text")
        cleaned_text = self.clean(raw_text)

        # 3. Détection langue
        language = self.detect_language(cleaned_text)

        # 4. Chunking
        self.logger.info(f"Chunking (lang={language})")
        chunks = self.chunk(cleaned_text, language)

        # 5. Post-processing
        self.logger.info("Post-processing chunks")
        chunks = self.postprocess(chunks, file_path, language)

        # 6. Validation
        self.logger.info("Validating chunks")
        valid_chunks = [c for c in chunks if self.validate_chunk(c)]

        self.logger.info(f"Pipeline complete: {len(valid_chunks)} valid chunks")
        return valid_chunks

    def extract(self, file_path: Path) -> str:
        """Extraire texte selon type de fichier."""
        suffix = file_path.suffix.lower()

        if suffix == '.pdf':
            from unstructured.partition.pdf import partition_pdf
            elements = partition_pdf(filename=str(file_path))
            return '\n\n'.join([e.text for e in elements])

        elif suffix == '.docx':
            from docx import Document
            doc = Document(file_path)
            return '\n\n'.join([p.text for p in doc.paragraphs])

        elif suffix in ['.txt', '.md']:
            return file_path.read_text(encoding='utf-8')

        else:
            raise ValueError(f"Unsupported file type: {suffix}")

    def clean(self, text: str) -> str:
        """Nettoyer le texte."""
        import re

        # Retirer caractères de contrôle
        text = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f]', '', text)

        # Normaliser espaces
        text = re.sub(r' +', ' ', text)
        text = re.sub(r'\n{3,}', '\n\n', text)

        # Retirer URLs (optionnel)
        # text = re.sub(r'http\S+', '', text)

        return text.strip()

    def detect_language(self, text: str) -> str:
        """Détecter langue du document."""
        from langdetect import detect
        try:
            return detect(text[:1000])  # Échantillon
        except:
            return 'en'  # Défaut

    def chunk(self, text: str, language: str) -> List[str]:
        """Chunking adapté à la langue."""
        from langchain.text_splitter import RecursiveCharacterTextSplitter
        import tiktoken

        encoder = tiktoken.encoding_for_model("gpt-4")
        chunk_size = self.config['chunk_size'].get(language, 512)

        splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=int(chunk_size * 0.2),
            length_function=lambda t: len(encoder.encode(t)),
            separators=["\n\n", "\n", ". ", " ", ""]
        )

        return splitter.split_text(text)

    def postprocess(self, chunks: List[str], file_path: Path, language: str) -> List[Chunk]:
        """Enrichir chunks avec métadonnées."""
        import tiktoken
        encoder = tiktoken.encoding_for_model("gpt-4")

        processed = []
        for i, chunk_text in enumerate(chunks):
            chunk = Chunk(
                text=chunk_text,
                metadata={
                    'source': str(file_path),
                    'filename': file_path.name,
                    'language': language,
                    'chunk_index': i,
                    'total_chunks': len(chunks),
                    'prev_chunk_id': f"{file_path.stem}_{i-1}" if i > 0 else None,
                    'next_chunk_id': f"{file_path.stem}_{i+1}" if i < len(chunks)-1 else None
                },
                token_count=len(encoder.encode(chunk_text)),
                coherence_score=self.calculate_coherence(chunk_text)
            )
            processed.append(chunk)

        return processed

    def calculate_coherence(self, text: str) -> float:
        """Calculer score de cohérence sémantique."""
        # Implémentation simplifiée
        sentences = text.split('. ')
        return min(1.0, len(sentences) / 10)  # Proxy simple

    def validate_chunk(self, chunk: Chunk) -> bool:
        """Valider qu'un chunk respecte les contraintes."""
        # Vérifications
        if chunk.token_count < 50:  # Trop petit
            self.logger.warning(f"Chunk trop petit : {chunk.token_count} tokens")
            return False

        if chunk.token_count > 1500:  # Trop grand
            self.logger.warning(f"Chunk trop grand : {chunk.token_count} tokens")
            return False

        if chunk.coherence_score < 0.3:  # Cohérence insuffisante
            self.logger.warning(f"Cohérence faible : {chunk.coherence_score}")
            return False

        return True

# Utilisation
config = {
    'chunk_size': {
        'en': 600,
        'fr': 512,
        'es': 520
    }
}

pipeline = DocumentPreprocessingPipeline(config)
chunks = pipeline.process(Path("document.pdf"))

print(f"Processed {len(chunks)} valid chunks")

import prometheus_client as prom
from dataclasses import dataclass
from datetime import datetime

# Métriques Prometheus
chunk_size_histogram = prom.Histogram(
    'chunking_size_tokens',
    'Distribution des tailles de chunks (tokens)',
    buckets=[100, 256, 512, 768, 1024, 1536, 2048]
)

chunk_coherence_gauge = prom.Gauge(
    'chunking_coherence_score',
    'Score de cohérence moyen des chunks'
)

processing_time_histogram = prom.Histogram(
    'chunking_processing_seconds',
    'Temps de traitement chunking'
)

@dataclass
class ChunkingMetrics:
    total_chunks: int
    avg_token_count: float
    avg_coherence: float
    processing_time: float
    error_rate: float

class MonitoredChunker:
    def __init__(self, base_chunker):
        self.chunker = base_chunker

    def chunk_with_monitoring(self, text: str) -> List[Chunk]:
        start_time = datetime.now()

        try:
            chunks = self.chunker.split(text)

            # Enregistrer métriques
            for chunk in chunks:
                chunk_size_histogram.observe(chunk.token_count)

            avg_coherence = sum(c.coherence_score for c in chunks) / len(chunks)
            chunk_coherence_gauge.set(avg_coherence)

            processing_time = (datetime.now() - start_time).total_seconds()
            processing_time_histogram.observe(processing_time)

            return chunks

        except Exception as e:
            logging.error(f"Chunking failed: {e}")
            # Incrémenter compteur erreurs
            raise

import hashlib
import redis
import pickle

class CachedChunker:
    def __init__(self, base_chunker, redis_client):
        self.chunker = base_chunker
        self.redis = redis_client
        self.cache_ttl = 86400  # 24h

    def chunk(self, text: str, cache_key: str = None) -> List[Chunk]:
        # Générer clé de cache
        if cache_key is None:
            text_hash = hashlib.sha256(text.encode()).hexdigest()
            config_hash = hashlib.sha256(
                str(self.chunker.config).encode()
            ).hexdigest()
            cache_key = f"chunks:{text_hash}:{config_hash}"

        # Vérifier cache
        cached = self.redis.get(cache_key)
        if cached:
            return pickle.loads(cached)

        # Chunker et mettre en cache
        chunks = self.chunker.split(text)
        self.redis.setex(
            cache_key,
            self.cache_ttl,
            pickle.dumps(chunks)
        )

        return chunks

class QualityMonitor:
    def __init__(self, alert_threshold=0.15):
        self.baseline_metrics = None
        self.alert_threshold = alert_threshold

    def set_baseline(self, metrics: ChunkingMetrics):
        """Définir baseline de référence."""
        self.baseline_metrics = metrics

    def check_degradation(self, current_metrics: ChunkingMetrics):
        """Détecter dégradation significative."""
        if not self.baseline_metrics:
            return

        # Comparer cohérence
        coherence_drop = (
            self.baseline_metrics.avg_coherence - current_metrics.avg_coherence
        ) / self.baseline_metrics.avg_coherence

        if coherence_drop > self.alert_threshold:
            self.send_alert(
                f"Dégradation cohérence : {coherence_drop:.1%} "
                f"(baseline={self.baseline_metrics.avg_coherence:.2f}, "
                f"current={current_metrics.avg_coherence:.2f})"
            )

        # Comparer error rate
        if current_metrics.error_rate > 0.05:  # 5%
            self.send_alert(
                f"Taux d'erreur élevé : {current_metrics.error_rate:.1%}"
            )

    def send_alert(self, message: str):
        """Envoyer alerte (Slack, PagerDuty, etc.)."""
        logging.error(f"ALERT: {message}")
        # Implémenter intégration Slack/PagerDuty