Batching et Chunking — Embeddings, RAG et Recherche Sémantique

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.

Les deux notions se ressemblent mais répondent à des problèmes opposés. Le chunking traite le cas d’un document trop gros pour être encodé utilement d’un seul tenant ; le batching traite le cas de documents trop nombreux pour tenir dans une seule requête. Un manuel technique de trois cents pages relève du premier, un export de vingt mille tickets relève du second, et un projet réel relève presque toujours des deux à la fois.

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 consiste à découper votre corpus en lots et à boucler. La fonction ci-dessous fait exactement cela, en affichant sa progression — détail qui compte quand l’encodage d’un corpus complet occupe plusieurs minutes et que vous devez savoir si le processus avance ou s’il est bloqué.

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")

L’usage de extend plutôt que append garantit que l’ordre des vecteurs suit exactement l’ordre des textes en entrée. C’est ce qui vous permettra, plus tard, de retrouver le document d’origine à partir d’un simple indice.

Optimiser la taille des lots

La taille optimale du lot dépend de la longueur moyenne de vos textes : cinquante titres d’articles et cinquante rapports complets ne pèsent pas du tout le même nombre de tokens. Plutôt que de choisir une constante au hasard, estimez-la à partir du corpus lui-même.

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}")

Le plafond à 100 n’est pas cosmétique : au-delà, un seul lot en échec vous fait perdre beaucoup de travail, et la mémoire consommée par la réponse devient significative.

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, et le choix a un effet direct sur la qualité des réponses : un chunk qui commence au milieu d’une phrase produit un vecteur flou, donc un résultat de recherche flou.

Découpage par caractères (simple)

La méthode la plus directe coupe tous les N caractères, sans se soucier du contenu. Elle a le mérite d’être prévisible et rapide, ce qui la rend acceptable pour un prototype ou pour du texte sans structure claire.

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é)

Le découpage par paragraphes respecte l’unité de sens voulue par l’auteur. Sur une documentation produit, chaque paragraphe traite en général d’un point précis ; en le gardant entier, vous obtenez un chunk qui répond à lui seul à une question. La fonction accumule les paragraphes tant que la taille maximale n’est pas dépassée, puis ferme le chunk courant.

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

Quand le texte n’a pas de paragraphes exploitables — une transcription d’appel, un compte rendu au fil de l’eau — la phrase devient la plus petite unité de sens disponible. Le découpage suit alors la même logique d’accumulation, en s’appuyant sur les ponctuations de fin de phrase.

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. Imaginez une procédure dont l’étape 4 se termine au caractère 2040 et dont la conséquence est expliquée juste après : sans chevauchement, aucun chunk ne contient les deux moitiés du raisonnement, et la recherche ne remontera jamais la réponse complète.

Les valeurs ci-dessous constituent des points de départ éprouvés selon la nature du contenu :

# 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},
}

Les conversations descendent à 500 caractères parce qu’un tour de parole y est court et qu’un chunk plus large mélangerait plusieurs sujets ; le code monte à 2000 avec un chevauchement généreux parce qu’une fonction perd son sens si sa signature est séparée de son corps.

Pipeline complet : chunk + batch + embed

Voici comment combiner chunking et batching dans un pipeline de production. L’élément à ne pas négliger est chunk_metadata : sans lui, vous obtenez des vecteurs orphelins, incapables d’indiquer de quel document ils proviennent, et vous ne pourrez jamais citer vos sources dans la réponse finale.

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}")

Faites tourner ce pipeline sur une dizaine de vos vrais documents et inspectez all_chunks à l’oeil nu avant d’encoder quoi que ce soit à grande échelle. Un découpage raté se voit immédiatement à la lecture, alors qu’il ne se voit jamais dans la forme de la matrice.

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