Indexierung eines Dokumentenkorpus

Lernziele

• Aufbau eines vollständigen semantischen Suchindex
• Verarbeitung verschiedener Dokumentformate (Text, PDF, Markdown)
• Inkrementelle Aktualisierung des Index

Architektur eines semantischen Index

Ein semantischer Index stützt sich auf drei Elemente, die synchron bleiben müssen: die ursprünglichen Dokumente mit ihren Metadaten, die aus dem Text berechneten Embeddings und die Zuordnung, die jeden Vektor mit seinem Ursprungsdokument verknüpft. Dieser dritte Aspekt ist der fragilste – hier korruptieren sich Indizes fast immer, wenn beispielsweise eine Löschung auf einer Seite die Indizes verschiebt, auf der anderen Seite jedoch nicht.

Die untenstehende Document-Klasse fügt eine automatisch bei der Erstellung berechnete MD5-Hash-Zeichenfolge hinzu. Wir werden diese später verwenden, um zu erkennen, welche Inhalte bereits indiziert wurden.

from dataclasses import dataclass, field
from openai import OpenAI
import numpy as np
import json
import hashlib

@dataclass
class Document:
    id: str
    texte: str
    metadata: dict = field(default_factory=dict)
    hash: str = ""

    def __post_init__(self):
        self.hash = hashlib.md5(self.texte.encode()).hexdigest()

Indexierungspipeline

Schritt 1: Laden von Dokumenten

Das Laden hängt vom Format ab, und das Format bestimmt die Granularität. Eine reine Textdatei wird zu einem einzigen Dokument; eine Markdown-Datei enthält jedoch oft mehrere distincte Themen, und die Indizierung als ein einziger Block würde jedes Thema im Durchschnitt der anderen untergehen. Der zweite Lader teilt daher auf Ebene-2-Überschriften auf und injiziert den Abschnittstitel erneut in den indizierten Text – dies ist die Enrichment-Technik, die in Lektion 3 behandelt wird.

from pathlib import Path

def charger_fichiers_texte(dossier: str) -> list[Document]:
    """Lädt alle Textdateien aus einem Ordner."""
    documents = []
    for fichier in Path(dossier).rglob("*.txt"):
        contenu = fichier.read_text(encoding="utf-8").strip()
        if contenu:
            documents.append(Document(
                id=fichier.stem,
                texte=contenu,
                metadata={
                    "fichier": str(fichier),
                    "type": "txt",
                    "taille": len(contenu)
                }
            ))
    return documents

def charger_fichiers_markdown(dossier: str) -> list[Document]:
    """Lädt Markdown-Dateien und teilt sie nach Abschnitten auf."""
    documents = []
    for fichier in Path(dossier).rglob("*.md"):
        contenu = fichier.read_text(encoding="utf-8")
        sections = contenu.split("\n## ")

        for i, section in enumerate(sections):
            if not section.strip():
                continue
            titre = section.split("\n")[0].strip("# ")
            corps = "\n".join(section.split("\n")[1:]).strip()
            if corps:
                documents.append(Document(
                    id=f"{fichier.stem}_section_{i}",
                    texte=f"{titre}\n\n{corps}",
                    metadata={
                        "fichier": str(fichier),
                        "section": titre,
                        "type": "markdown"
                    }
                ))
    return documents

Schritt 2: Aufbau des Indexes

Die IndexSemantique-Klasse vereint den Batch-Aufbau, die Suche mit Schwellenwert und die Persistenz auf Festplatte. Der Parameter seuil (Schwellenwert) in der rechercher-Methode verdient besondere Aufmerksamkeit: Ohne ihn liefert eine themenfremde Abfrage immer noch die fünf am wenigsten schlechten Dokumente, mit Scores um 0,15, die ein naiver Aufrufer als gültige Antworten behandeln würde. Das Filtern von allem unter einem Mindestscore ermöglicht es Ihnen, ehrlich zu antworten: „Ich habe nichts gefunden.”

class IndexSemantique:
    def __init__(self, model: str = "text-embedding-3-large"):
        self.client = OpenAI()
        self.model = model
        self.documents: list[Document] = []
        self.embeddings: np.ndarray | None = None

    def construire(self, documents: list[Document], batch_size: int = 100):
        """Erstellt den Index aus einer Liste von Dokumenten."""
        self.documents = documents
        textes = [d.texte for d in documents]
        tous_embeddings = []

        for i in range(0, len(textes), batch_size):
            batch = textes[i:i + batch_size]
            response = self.client.embeddings.create(
                input=batch,
                model=self.model
            )
            batch_sorted = sorted(response.data, key=lambda x: x.index)
            tous_embeddings.extend([e.embedding for e in batch_sorted])
            print(f"  Batch {i // batch_size + 1} verarbeitet "
                  f"({min(i + batch_size, len(textes))}/{len(textes)})")

        self.embeddings = np.array(tous_embeddings, dtype=np.float32)
        print(f"Index erstellt: {len(self.documents)} Dokumente, "
              f"{self.embeddings.shape[1]} Dimensionen")

    def rechercher(self, requete: str, k: int = 5,
                   seuil: float = 0.0) -> list[dict]:
        """Sucht nach den k relevantesten Dokumenten."""
        query_emb = self.client.embeddings.create(
            input=requete,
            model=self.model
        ).data[0].embedding

        scores = self.embeddings @ np.array(query_emb)
        top_indices = np.argsort(scores)[-k:][::-1]

        resultats = []
        for idx in top_indices:
            score = float(scores[idx])
            if score >= seuil:
                doc = self.documents[idx]
                resultats.append({
                    "id": doc.id,
                    "texte": doc.texte[:200],
                    "score": score,
                    "metadata": doc.metadata
                })
        return resultats

    def sauvegarder(self, chemin: str):
        """Speichert den Index auf der Festplatte."""
        np.save(f"{chemin}_embeddings.npy", self.embeddings)
        docs_serialisables = [
            {"id": d.id, "texte": d.texte,
             "metadata": d.metadata, "hash": d.hash}
            for d in self.documents
        ]
        with open(f"{chemin}_documents.json", "w") as f:
            json.dump(docs_serialisables, f, ensure_ascii=False)

    def charger(self, chemin: str):
        """Lädt einen Index von der Festplatte."""
        self.embeddings = np.load(f"{chemin}_embeddings.npy")
        with open(f"{chemin}_documents.json") as f:
            docs = json.load(f)
        self.documents = [
            Document(id=d["id"], texte=d["texte"],
                     metadata=d["metadata"])
            for d in docs
        ]

Die Methoden sauvegarder und charger schreiben zwei Dateien mit demselben Präfix. Behandeln Sie sie als untrennbare Einheit: Das Kopieren einer ohne die andere oder die Wiederherstellung eines unvollständigen Backups erzeugt einen Index, bei dem die Vektoren den Dokumenten nicht mehr entsprechen – ein stilles Versagen, das sich erst als absurde Ergebnisse zeigt.

Inkrementelle Aktualisierungen

Ein Korpus entwickelt sich: drei Dokumente wurden am Montag geändert, zehn wurden am Mittwoch hinzugefügt. Die Neuberechnung von fünfzehntausend Vektoren für jede Änderung verschwendet Zeit und Geld. Hier kommt die MD5-Hash-Zeichenfolge ins Spiel: Sie identifiziert den Inhalt, nicht die Datei, sodass ein umbenanntes, unverändertes Dokument nicht erneut indiziert wird, während ein geändertes Dokument eine andere Hash-Zeichenfolge erzeugt.

def mettre_a_jour(self, nouveaux_docs: list[Document]):
    """Fügt Dokumente zum bestehenden Index hinzu."""
    # Filtert bereits indizierte Dokumente heraus
    hashes_existants = {d.hash for d in self.documents}
    docs_a_ajouter = [
        d for d in nouveaux_docs if d.hash not in hashes_existants
    ]

    if not docs_a_ajouter:
        print("Keine neuen Dokumente zum Indizieren")
        return

    textes = [d.texte for d in docs_a_ajouter]
    response = self.client.embeddings.create(
        input=textes,
        model=self.model
    )

    nouveaux_vecteurs = np.array(
        [e.embedding for e in sorted(response.data, key=lambda x: x.index)],
        dtype=np.float32
    )

    self.documents.extend(docs_a_ajouter)
    self.embeddings = np.vstack([self.embeddings, nouveaux_vecteurs])
    print(f"{len(docs_a_ajouter)} Dokumente hinzugefügt. "
          f"Insgesamt: {len(self.documents)}")

np.vstack stapelt die neuen Vektoren unter den alten in derselben Reihenfolge wie das extend auf der Dokumentliste: Diese Symmetrie bewahrt die Zuordnung. Beachten Sie jedoch die Einschränkung dieser Implementierung – sie fügt hinzu, löscht oder ersetzt aber nicht. Ein geändertes Dokument befindet sich daher doppelt, in seiner alten und neuen Version. Das Behandeln von Löschungen erfordert das Entfernen der entsprechenden Zeile aus der Matrix, eine aufwändige Operation in NumPy, und dies ist einer der Gründe, warum Teams zu einer echten Vektordatabase tendieren.

End-to-End-Nutzung

End-to-End dauert der Aufbau eines Indexes über Dokumentation nur wenige Zeilen.

# Erstellen und Belegen des Indexes
index = IndexSemantique()

docs = charger_fichiers_markdown("./documentation/")
print(f"{len(docs)} Abschnitte geladen")

index.construire(docs)
index.sauvegarder("mon_index")

# Suche
resultats = index.rechercher("wie authentifizierung konfigurieren", k=3)
for r in resultats:
    print(f"[{r['score']:.3f}] {r['id']}")
    print(f"  {r['texte'][:100]}...")

Führen Sie dies auf Ihrer eigenen Dokumentation aus und stellen Sie drei Fragen, deren Antworten Sie kennen. Wenn das richtige Dokument nicht an erster Stelle rangiert, ist der Übeltäter fast immer das Chunking – Abschnitte zu lang oder Titel nicht erneut injiziert – und nicht das Embedding-Modell.

Zusammenfassung

• Ein semantischer Index verknüpft Dokumente, Embeddings und Metadaten
• Laden von Dokumenten aus verschiedenen Formaten (Text, Markdown usw.)
• Aufbau des Indexes in Batches, um große Korpora zu verarbeiten
• Speichern und inkrementelles Aktualisieren, um alles neu zu berechnen zu vermeiden