Batching und Chunking

Die Herausforderung großer Datenmengen

Wenn Sie mit Tausenden von Dokumenten arbeiten, können Sie nicht alles in einer einzigen API-Anfrage kodieren. Sie müssen Batching (Verarbeitung in Gruppen) und Chunking (Aufteilen von Text) verwalten. Diese beiden Techniken sind unerlässlich für den Aufbau von semantischen Suchsystemen im großen Maßstab.

Die beiden Konzepte klingen ähnlich, lösen aber entgegengesetzte Probleme. Das Chunking behandelt ein einzelnes Dokument, das zu groß ist, um effektiv als Ganzes kodiert zu werden; das Batching behandelt zu viele Dokumente, um in einer einzigen Anfrage Platz zu finden. Ein dreihundertseitiges technisches Handbuch fällt in die erste Kategorie, ein Export von zwanzigtausend Tickets in die zweite, und ein reales Projekt beinhaltet fast immer beides.

Batching: Verarbeitung in Gruppen

Die mistral-embed-API akzeptiert eine Liste von Eingaben, aber es bestehen praktische Grenzen (Anfragedateigröße, Token-Anzahl). Die Lösung besteht darin, Ihren Korpus in Batches aufzuteilen und diese zu durchlaufen. Die folgende Funktion tut genau das und zeigt den Fortschritt an – ein entscheidender Detailaspekt, wenn das Kodieren eines vollständigen Korpus mehrere Minuten dauert und Sie wissen müssen, ob der Prozess voranschreitet oder stecken geblieben ist.

import numpy as np
from mistralai.client import Mistral

client = Mistral(api_key=api_key)

def get_embeddings_by_chunks(data, chunk_size=25):
    """Kodiert eine große Liste von Texten in Batches."""
    chunks = [data[x:x + chunk_size] for x in range(0, len(data), chunk_size)]
    embeddings = []

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

    return embeddings

# Beispiel mit 1000 Dokumenten
documents = [f"Dokument Nummer {i} mit varied content." for i in range(1000)]
all_embeddings = get_embeddings_by_chunks(documents, chunk_size=50)

print(f"Insgesamt: {len(all_embeddings)} Embeddings mit {len(all_embeddings[0])} Dimensionen")

Die Verwendung von extend statt append stellt sicher, dass die Reihenfolge der Vektoren exakt der Reihenfolge des Eingabetextes entspricht. Dies ermöglicht es Ihnen, das ursprüngliche Dokument später über einen einfachen Index wiederherzustellen.

Optimierung der Batch-Größe

Die optimale Batch-Größe hängt von der durchschnittlichen Länge Ihrer Texte ab: fünfzig Artikelüberschriften und fünfzig vollständige Berichte wiegen in Tokens nicht gleich. Anstatt eine Konstante willkürlich zu wählen, sollten Sie sie aus dem Korpus selbst ableiten.

def optimal_batch_size(documents, max_tokens_per_batch=16000):
    """Schätzt die optimale Batch-Größe."""
    # Grobe Schätzung: 1 Token ~ 4 Zeichen in Englisch
    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)  # Begrenzung auf 100

# Anpassung basierend auf Ihrem Korpus
batch_size = optimal_batch_size(documents)
print(f"Empfohlene Batch-Größe: {batch_size}")

Die Begrenzung auf 100 ist nicht kosmetisch: Darüber hinaus führt ein einzelner fehlgeschlagener Batch zu erheblichem Arbeitsverlust, und der von der Antwort verbrauchte Speicher wird beträchtlich.

Chunking: Aufteilen von Text

Lange Texte müssen vor dem Kodieren in Chunks aufgeteilt werden. Es gibt mehrere Strategien, jede mit ihren eigenen Vorteilen, und die Wahl wirkt sich direkt auf die Antwortqualität aus: Ein Chunk, das mitten in einem Satz beginnt, erzeugt einen unscharfen Vektor, was zu unscharfen Suchergebnissen führt.

Aufteilen nach Zeichen (Einfach)

Die direkteste Methode schneidet alle N Zeichen ab, unabhängig vom Inhalt. Ihr Verdienst liegt in der Vorhersagbarkeit und Geschwindigkeit, was sie für einen Prototypen oder für Text ohne klare Struktur akzeptabel macht.

def chunk_by_chars(text, chunk_size=2048, overlap=200):
    """Teilt Text in Chunks mit Überlappung auf."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap  # Überlappung für Kontext
    return chunks

long_text = "Ein sehr langes Dokument..." * 500
chunks = chunk_by_chars(long_text, chunk_size=2048, overlap=200)
print(f"{len(chunks)} Chunks erstellt")

Aufteilen nach Absätzen (Empfohlen)

Das Aufteilen nach Absätzen respektiert die vom Autor intendierte Einheit der Bedeutung. In Produktdokumentationen behandelt jeder Absatz im Allgemeinen einen bestimmten Punkt; indem er intakt bleibt, ergibt sich ein Chunk, das eine Frage eigenständig beantworten kann. Die Funktion sammelt Absätze, bis die maximale Größe überschritten wird, und schließt dann das aktuelle Chunk ab.

def chunk_by_paragraphs(text, max_chunk_size=2048):
    """Teilt, während Absatzgrenzen respektiert werden."""
    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

Aufteilen nach Sätzen

Wenn Text keine nutzbaren Absätze aufweist – etwa ein Anrufprotokoll oder ein laufender Protokoll – wird der Satz zur kleinsten verfügbaren Einheit der Bedeutung. Das Aufteilen folgt dann derselben Akkumulationslogik und stützt sich auf Satzende-Punktuierung.

import re

def chunk_by_sentences(text, max_chunk_size=1500):
    """Teilt, während Satzgrenzen respektiert werden."""
    # Aufteilung am Ende des Satzes
    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

Überlappung

Eine Überlappung zwischen Chunks verhindert, dass eine Idee in der Mitte abgeschnitten wird. Durch Hinzufügen einer Überlappung von 10 % bis 20 % der Chunk-Größe bewahren Sie den Kontext zwischen benachbarten Stücken. Stellen Sie sich ein Verfahren vor, bei dem Schritt 4 bei Zeichen 2040 endet und die Konsequenz unmittelbar danach erklärt wird: Ohne Überlappung enthält kein Chunk beide Hälften der Argumentation, und die Suche wird die vollständige Antwort niemals abrufen können.

Die folgenden Werte dienen als bewährte Ausgangspunkte je nach Inhaltstyp:

# Empfohlene Parameter nach Inhaltstyp
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},
}

Gespräche sinken auf 500 Zeichen, da ein Sprachbeitrag kurz ist und ein größeres Chunk mehrere Themen vermischen würde; Code steigt auf 2000 mit großzügiger Überlappung, da eine Funktion ihre Bedeutung verliert, wenn ihre Signatur vom Körper getrennt wird.

Kompletter Pipeline: Chunk + Batch + Embed

Hier ist, wie Sie Chunking und Batching in einer Produktionspipeline kombinieren. Das Element, das Sie nicht vernachlässigen dürfen, ist chunk_metadata: Ohne es erhalten Sie verwaiste Vektoren, die nicht angeben können, aus welchem Dokument sie stammen, und Sie werden Ihre Quellen in der endgültigen Antwort niemals zitieren können.

def pipeline_embed(documents, chunk_size=1500, overlap=200, batch_size=50):
    """Komplette Pipeline: Chunk, Batch, Embed."""
    # 1. Chunk alle Dokumente
    all_chunks = []
    chunk_metadata = []  # Herkunftsdokument verfolgen

    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)} Dokumente -> {len(all_chunks)} Chunks")

    # 2. Kodieren in Batches
    embeddings = get_embeddings_by_chunks(all_chunks, chunk_size=batch_size)

    # 3. Strukturierte Ergebnisse zurückgeben
    return {
        "chunks": all_chunks,
        "embeddings": np.array(embeddings),
        "metadata": chunk_metadata,
    }

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

Führen Sie diese Pipeline auf einem Dutzend Ihrer tatsächlichen Dokumente aus und inspizieren Sie all_chunks visuell, bevor Sie im großen Maßstab etwas kodieren. Eine schlechte Aufteilung ist beim Lesen sofort sichtbar, während sie in der Matrixform niemals auftritt.

Kernaussagen

• Batching ermöglicht das Kodieren von Tausenden von Dokumenten, ohne API-Grenzen zu überschreiten
• Chunking teilt lange Texte für eine bessere semantische Granularität auf
• Eine Überlappung von 10–20 % zwischen Chunks bewahrt den Kontext
• Das Aufteilen nach Absätzen oder Sätzen ist vorzuziehen gegenüber dem Aufteilen nach Zeichen
• Bewahren Sie Metadaten (Herkunftsdokument, Position) für jedes Chunk auf