Batching et Chunking

Le défi des gros volumes

Quand vous travaillez avec des milliers de documents, vous ne pouvez pas tout encoder en un seul appel API. Vous devez gérer le batching (traitement par lots) et le chunking (découpage du texte). Ces deux techniques sont essentielles pour construire des systèmes de recherche sémantique à l’échelle.

Batching : traitement par lots

L’API mistral-embed accepte une liste d’inputs, mais il y a des limites pratiques (taille de la requête, nombre de tokens). La solution : découper votre corpus en lots.

import numpy as np
from mistralai import Mistral

client = Mistral(api_key=api_key)

def get_embeddings_by_chunks(data, chunk_size=25):
    """Encoder une grande liste de textes par lots."""
    chunks = [data[x:x + chunk_size] for x in range(0, len(data), chunk_size)]
    embeddings = []

    for i, chunk in enumerate(chunks):
        print(f"Lot {i+1}/{len(chunks)} ({len(chunk)} textes)...")
        response = client.embeddings.create(
            model="mistral-embed",
            inputs=chunk,
        )
        embeddings.extend([d.embedding for d in response.data])

    return embeddings

# Exemple avec 1000 documents
documents = [f"Document numéro {i} avec du contenu varié." for i in range(1000)]
all_embeddings = get_embeddings_by_chunks(documents, chunk_size=50)

print(f"Total: {len(all_embeddings)} embeddings de {len(all_embeddings[0])} dimensions")

Optimiser la taille des lots

La taille optimale du lot dépend de la longueur moyenne de vos textes :

def optimal_batch_size(documents, max_tokens_per_batch=16000):
    """Estimer la taille de lot optimale."""
    # Estimation grossière : 1 token ~ 4 caractères en français
    avg_chars = sum(len(d) for d in documents) / len(documents)
    avg_tokens = avg_chars / 4

    batch_size = max(1, int(max_tokens_per_batch / avg_tokens))
    return min(batch_size, 100)  # Plafonner à 100

# Adapter selon votre corpus
batch_size = optimal_batch_size(documents)
print(f"Taille de lot recommandée: {batch_size}")

Chunking : découpage du texte

Les textes longs doivent être découpés en chunks (morceaux) avant l’encodage. Plusieurs stratégies existent, chacune avec ses avantages.

Découpage par caractères (simple)

def chunk_by_chars(text, chunk_size=2048, overlap=200):
    """Découper un texte en chunks avec chevauchement."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap  # Chevauchement pour le contexte
    return chunks

long_text = "Un très long document..." * 500
chunks = chunk_by_chars(long_text, chunk_size=2048, overlap=200)
print(f"{len(chunks)} chunks créés")

Découpage par paragraphes (recommandé)

def chunk_by_paragraphs(text, max_chunk_size=2048):
    """Découper en respectant les limites de paragraphes."""
    paragraphs = text.split("\n\n")
    chunks = []
    current_chunk = ""

    for para in paragraphs:
        if len(current_chunk) + len(para) + 2 > max_chunk_size:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = para
        else:
            current_chunk += "\n\n" + para if current_chunk else para

    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks

Découpage par phrases

import re

def chunk_by_sentences(text, max_chunk_size=1500):
    """Découper en respectant les limites de phrases."""
    # Séparer sur les ponctuations de fin de phrase
    sentence_pattern = r'(?<=[.!?])\s+'
    sentences = re.split(sentence_pattern, text)
    chunks = []
    current_chunk = ""

    for sentence in sentences:
        if len(current_chunk) + len(sentence) + 1 > max_chunk_size:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = sentence
        else:
            current_chunk += " " + sentence if current_chunk else sentence

    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks

Chevauchement (overlap)

Le chevauchement entre les chunks évite de couper une idée en plein milieu. En ajoutant un overlap de 10 à 20 % de la taille du chunk, vous préservez le contexte entre morceaux adjacents.

# Paramètres recommandés selon le type de contenu
PARAMS = {
    "articles":     {"chunk_size": 1500, "overlap": 200},
    "documentation": {"chunk_size": 1000, "overlap": 150},
    "conversations": {"chunk_size": 500,  "overlap": 50},
    "code":          {"chunk_size": 2000, "overlap": 300},
}

Pipeline complet : chunk + batch + embed

Voici comment combiner chunking et batching dans un pipeline de production :

def pipeline_embed(documents, chunk_size=1500, overlap=200, batch_size=50):
    """Pipeline complet : chunk, batch, embed."""
    # 1. Chunker tous les documents
    all_chunks = []
    chunk_metadata = []  # Garder la trace du document source

    for doc_idx, doc in enumerate(documents):
        chunks = chunk_by_paragraphs(doc, max_chunk_size=chunk_size)
        for chunk_idx, chunk in enumerate(chunks):
            all_chunks.append(chunk)
            chunk_metadata.append({
                "doc_index": doc_idx,
                "chunk_index": chunk_idx,
            })

    print(f"{len(documents)} documents -> {len(all_chunks)} chunks")

    # 2. Encoder par lots
    embeddings = get_embeddings_by_chunks(all_chunks, chunk_size=batch_size)

    # 3. Retourner les résultats structurés
    return {
        "chunks": all_chunks,
        "embeddings": np.array(embeddings),
        "metadata": chunk_metadata,
    }

# Utilisation
result = pipeline_embed(documents)
print(f"Embeddings: {result['embeddings'].shape}")

Points clés à retenir

• Le batching permet d’encoder des milliers de documents sans dépasser les limites API
• Le chunking découpe les textes longs pour une meilleure granularité sémantique
• Un chevauchement de 10-20 % entre les chunks préserve le contexte
• Le découpage par paragraphes ou phrases est préférable au découpage par caractères
• Gardez les métadonnées (document source, position) pour chaque chunk