RAG in reinem Python: Chunking, TF-IDF, Kosinusähnlichkeit und Prompting

In einem typischen RAG-Tutorial muss man fünf Bibliotheken installieren und .from_documents() aufrufen. Zehn Minuten später funktioniert eine Demo – doch man versteht davon nichts. Wenn die Informationsabrufung Unsinn liefert oder das Modell Halluzinationen erzeugt, steckt man in der Klemme, weil die Bibliothek das Denken bereits für einen erledigt hat.

Hier wird der Pipeline-Prozess auf die umgekehrte Weise konstruiert, rein in Python, unter Verwendung nur des math-Moduls sowie eines einzigen Aufrufs einer reinen-Text-LLM am Ende. Kein Embedding-Modell, kein FAISS, keine Pinecone. Wenn man die Mechanismen kennt, versteht man später, was bei LangChain oder einer echten Vektordatenbank vor sich geht. Am Ende steht eine funktionierende Pipeline – und noch wichtiger: ein Verständnis für jede Zahl, die sie erzeugt.

Das Problem, das RAG löst

Das Wissen eines Sprachmodells wird beim Ende der Trainingsphase festgelegt und existiert ausschließlich in seinen Gewichten. Um es dazu zu bringen, Fragen zu internen Unternehmensdokumenten, einer Codebasis oder den Nachrichten vom heutigen Morgen zu beantworten, enttäuschen beide offensichtlichen Lösungsansätze:

1. Das Modell erneut trainieren. Das ist langsam und kostspielig, und morgige Änderungen erfordern eine weitere Trainingsrunde.
2. Alles in den Prompt einfügen. Dies ist nicht skalierbar. Selbst ein Fenster mit einer Million Token kann keine 50.000 Dokumente aufnehmen, und Forschungen zu Modellen mit langen Kontexten weisen immer wieder auf ein „Verlorengehen in der Mitte“-Phänomen hin, bei dem Informationen, die tief im riesigen Prompt versteckt sind, trotz ihrer Verfügbarkeit schlechter genutzt werden. Zudem führen zusätzliche Token zu höherer Latenz und höheren Kosten pro Aufruf.

RAGs eigentlicher Trick ist enger, als es klingt. Die Dokumente bleiben im externen Speicher, und zur Beantwortung einer Frage werden nur die wenigen für diese Frage relevanten Teile abgerufen und an das Modell übergeben. Das LLM enthält niemals Ihre Daten – es erhält jedes Mal einen frischen, relevanten Teil.

Daher besteht jedes RAG-System aus zwei miteinander verbundenen Systemen: der Abrufphase, die für eine gegebene Anfrage den richtigen Text in einer weitaus größeren Sammlung findet, und der Generierungsphase, die diesen Text zusammen mit der Anfrage an ein LLM weiterleitet, um eine Antwort zu erhalten. Der Rest dieses Artikels erläutert, wie beide Komponenten aus grundlegenden Elementen aufgebaut werden.

Schritt 1: Chunking

Wenn ein ganzer 50-seitiger PDF in einen einzigen Vektor umgewandelt wird, gehen fast alle nützlichen Informationen verloren, da die Details, die eine Frage beantworten könnten, zu nichts zusammengefasst werden. Die erste Aufgabe besteht darin, Dokumente in kleine, gezielte Teile zu zerlegen.

Die einfachste Methode ist ein fest dimensionierter Schiebepuffer, der in Wörtern gezählt wird, wobei zwischen den benachbarten Blöcken Überschneidungen vorhanden sind:

def chunk_text(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
    if overlap >= chunk_size:
        raise ValueError("overlap must be smaller than chunk_size")

    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        if end >= len(words):
            break
        start = end - overlap  # step back so chunks share context
    return chunks

Diese Schutzklausel ist keine Dekoration. Wenn overlap den Wert von chunk_size erreicht oder überschreitet, hört die Vorwärtsbewegung der Startposition auf und der Schleifenlauf endet nie. Eine Anpassung von overlap nach oben „für mehr Kontext“ führt leicht zu diesem Fehler, weshalb die Funktion statt still abzustürzen einfach nicht ausgeführt wird.

Die Überschneidung ist wichtiger, als es scheint. Ohne sie könnte ein Satz, der eine Frage beantwortet, eine Grenze zwischen Blöcken überschreiten und dadurch geteilt werden, sodass keiner der Blöcke eine vollständige Idee enthält, die abgerufen werden könnte. Mit Überschneidung bleibt das Grenzgebiet zumindest in einem Block vollständig erhalten.

Ein Sonderfall, den man kennen sollte: Fest definierte Wortfenster ignorieren die Struktur von Sätzen und Abschnitten, wodurch ein Satz oder eine Tabellenzeile in zwei Teile geteilt werden kann. Eine nach Sätzen oder Abschnitten ausgerichtete Aufteilung vermeidet das, führt aber zu ungleichen Blöckengrößen, da ein 10-Wort-Satz genauso wie ein 150-Wort-Satz zu einem Block wird. Dadurch wird das Tokenbudget verfälscht sowie die Stärke, mit der ein einzelner Block die späteren Suchalgorithmen beeinflusst. Keiner der Aufteilungsmethoden ist ohne Kompromisse, und die Wahl hängt letztlich davon ab, welches Problem man lieber beheben möchte.

Schritt 2: Umwandlung von Text in Zahlen mit TF-IDF

Ein Computer kann die Ähnlichkeit zweier Zeichenketten nicht direkt vergleichen. Jeder folgende Schritt benötigt Vektoren, also Listen von Zahlen, mit denen arithmetische Operationen durchgeführt werden können. Die klassische, abhängigkeitsfreie Methode zur Erstellung solcher Vektoren ist TF-IDF – die Multiplikation der Termhäufigkeit mit der umgekehrten Dokumenthäufigkeit.

• TF gibt an, wie oft ein Wort in diesem bestimmten Abschnitt vorkommt. Eine hohe Anzahl für „Niere“ deutet darauf hin, dass der Abschnitt sich auf Nieren bezieht.
• IDF misst, wie selten ein Wort in der gesamten Sammlung vorkommt. Wörter wie „der“ und „ist“ kommen überall vor und sagen nichts darüber, welchen Abschnitt man gerade liest, weshalb sie einen Wert nahe null erhalten. Ein seltenes Wort in einem Abschnitt ist ein viel stärkeres Signal.

Der Code erstellt ein Wörterbuch aus allen Abschnitten, berechnet für jeden Begriff den IDF-Wert und wandelt anschließend jeden Abschnitt in einen TF-IDF-Vektor um:

import math
from collections import Counter

def build_vocabulary(chunks: list[str]) -> list[str]:
    vocab = set()
    for chunk in chunks:
        vocab.update(chunk.lower().split())
    return sorted(vocab)

def compute_idf(chunks: list[str], vocab: list[str]) -> dict[str, float]:
    num_chunks = len(chunks)
    idf = {}
    for word in vocab:
        containing = sum(1 for c in chunks if word in c.lower().split())
        idf[word] = math.log(num_chunks / (1 + containing))
    return idf

def compute_tf(chunk: str, vocab: list[str]) -> dict[str, float]:
    words = chunk.lower().split()
    counts = Counter(words)
    total = len(words)
    return {word: counts.get(word, 0) / total for word in vocab}

def vectorize(chunk: str, vocab: list[str], idf: dict[str, float]) -> list[float]:
    tf = compute_tf(chunk, vocab)
    return [tf[word] * idf[word] for word in vocab]

Probieren Sie es mit drei Textabschnitten aus – dann wird etwas klar. Ein Wort, das in fast jedem Abschnitt vorkommt, wie zum Beispiel „the“, erhält einen IDF-Wert nahe null; daher fügt seine Wiederholung nichts zum Vektor hinzu. Ein seltenes, themenspezifisches Wort hingegen erhält einen deutlich positiven IDF-Wert. TF-IDF ist im Grunde eine Formel, um herauszufinden, ob ein Wort etwas Besonderes über einen Textabschnitt verrät.

Der Nachteil liegt in der Skalierbarkeit. Bei 100.000 Dokumenten kann das Vokabular mehr als 500.000 einzigartige Wörter enthalten; daher hat jeder Abschnittsvektor, selbst bei einem dreiwortigen Abschnitt, eine Länge von 500.000 Zahlen, wobei die meisten davon Nullen sind. Solche spärlichen Vektoren sind aufwendig zu speichern und zu vergleichen. Das ist der Hauptgrund, warum Produktionsysteme auf dichte neuronale Embeddings mit 384 bis 1536 Dimensionen ohne nahezu Nullen umsteigen.

Schritt 3: Messung der Relevanz mit Kosinusähnlichkeit

Wenn Blöcke als Vektoren verwendet werden, stellt sich die nächste Frage: Wie relevant ist ein Block für eine Abfrage? Die reine Distanz ist nicht das richtige Werkzeug, denn bei Texten spielt der Winkel zwischen den Vektoren eine größere Rolle als ihre Länge. Ein Block, der „Katze“ einmal erwähnt, und einer, der es zwanzigmal wiederholt, beziehen sich immer noch auf dasselbe Thema, und die Kosinusähnlichkeit behandelt sie gleich, ohne zu berücksichtigen, welcher Satz lauter ist.

Die Implementierung benötigt ein Skalarprodukt, die Länge der Vektoren sowie das Verhältnis zwischen diesen beiden Werten:

def dot_product(a, b):
    return sum(x * y for x, y in zip(a, b))

def magnitude(v):
    return math.sqrt(sum(x * x for x in v))

def cosine_similarity(a, b):
    denom = magnitude(a) * magnitude(b)
    return 0.0 if denom == 0 else dot_product(a, b) / denom

Die Kosinusähnlichkeit liegt im Prinzip zwischen -1 und 1, und bei dieser TF-IDF-Einstellung bleibt sie fast immer in dem Bereich von 0 bis 1:

• 1,0 bedeutet, dass die Vektoren genau in derselben Richtung zeigen – maximale Ähnlichkeit.
• 0,0 bedeutet, dass sie orthogonal zueinander sind und keine gemeinsamen signifikanten Dimensionen aufweisen; im einfachen Wort bedeutet das, dass die Abfrage und der Textabschnitt keine gemeinsamen, nicht-trivialen Wörter haben.

Eine Wertung von 0 ist kein Sonderfall. Es bedeutet lediglich, dass das System korrekt feststellt, dass der Textabschnitt nichts mit der Abfrage zu tun hat.

Ein Punkt zur Genauigkeit: Die geglättete IDF-Formel kann einem Wort, das in jedem Textabschnitt vorkommt, einen leicht negativen Wert geben, sodass die Kosinusähnlichkeit gelegentlich etwas unter Null fallen kann. Bei echtem Text spielt das kaum eine Rolle, aber zu behaupten, sie sei niemals negativ, wäre übertrieben.

Der Blindpunkt von TF-IDF in Kombination mit der Kosinusähnlichkeit sind Synonyme. Eine Abfrage, die „Automobile“ enthält, erhält gegenüber einem Dokument, das „Car“ sagt, eine nahe Null als Bewertung, da nur ein wörtlicher Wortschatzüberschneidungen berücksichtigt werden und es keine Bedeutungsebene gibt. Dichte Embedding-Modelle, die Synonyme in der Nähe voneinander platzieren, dienen genau dazu, diese Lücke zu schließen.

Schritt 4: Der Retriever

Sobald Vektorisierung und Ähnlichkeitsbetrachtung bereitstehen, bedeutet das Abrufen, jeden Textabschnitt einmal zu vektorisieren, die eingehende Abfrage ebenfalls zu vektorisieren, jeden Abschnitt gegen diese Abfrage zu bewerten, sortieren und die besten k Ergebnisse beizubehalten:

class VectorStore:
    def __init__(self):
        self.chunks, self.vocab, self.idf, self.vectors = [], [], {}, []

    def index(self, documents, chunk_size=50, overlap=10):
        all_chunks = []
        for doc in documents:
            all_chunks.extend(chunk_text(doc, chunk_size, overlap))
        self.chunks = all_chunks
        self.vocab = build_vocabulary(all_chunks)
        self.idf = compute_idf(all_chunks, self.vocab)
        self.vectors = [vectorize(c, self.vocab, self.idf) for c in all_chunks]

    def search(self, query, top_k=3):
        query_vector = vectorize(query, self.vocab, self.idf)
        scored = [(c, cosine_similarity(query_vector, v))
                  for c, v in zip(self.chunks, self.vectors)]
        scored.sort(key=lambda pair: pair[1], reverse=True)
        return scored[:top_k]

Ein einziger Detail kann alles stillschweigend ruinieren. Die Abfrage muss mithilfe des aus den Dokumenten berechneten Wortschatzes sowie der IDF-Werte umgewandelt werden – ein Wortschatz, der ausschließlich aus der Abfrage selbst abgeleitet wird, ist ein Fehler. Die Kosinusähnlichkeit setzt Vektoren gleicher Länge voraus, wobei die Position i für dasselbe Wort in jedem Vektor steht. Wenn der Abfrage ein eigener Wortschatz gegeben wird, kann die Position 5 in einem Vektor „heart“ und im anderen „and“ sein, was dazu führt, dass der Ähnlichkeitswert strukturell sinnlos wird, anstatt nur leicht abweichend zu sein.

Schritt 5: Erweiterung des Prompts

Bei diesem Schritt kommt es nicht auf Mathematik an, sondern allein auf Disziplin. Sie kombinieren die abgerufenen Textabschnitte mit der Frage des Benutzers zu einem Prompt, der dem Modell genau vorgibt, wie es sich verhalten soll:

def build_augmented_prompt(query: str, retrieved_chunks: list[str]) -> str:
    context = "\n\n".join(
        f"[Source {i+1}]: {chunk}" for i, chunk in enumerate(retrieved_chunks)
    )
    return f"""Answer the question using ONLY the context below.
If the answer is not contained in the context, say "I don't know based on the provided context."
Cite which source number(s) you used.

Context:
{context}

Question: {query}

Answer:"""

Zwei Aspekte hier sind nicht rein kosmetischer Natur.

„Verwenden Sie NUR den folgenden Kontext.“ Ohne diese klare Regel mischt das Modell die abgerufenen Textabschnitte mit dem, was es aus der Trainingsphase in Erinnerung hat, sodass man nicht mehr erkennen kann, welcher Teil einer Antwort aus zuverlässigen Dokumenten stammt und welcher nur eine Schätzung ist. Eine solche Mischung, die selbstbewusst präsentiert wird, schadet mehr als das vollständige Vermeiden von RAG, da es den Anschein hat, eine Quelle zu haben, obwohl dies nicht der Fall ist.

Numerierte Quellbezeichnungen und Anweisungen zur Zitierung. Dadurch sind Antworten nachvollziehbar. Wenn eine Antwort falsch ist – was vorkommen wird – zeigen die beschrifteten Quellen an, welcher Textabschnitt das Modell in die Irre geführt hat, ob der Abschnitt selbst veraltet oder fehlerhaft war oder ob das Modell einen korrekten Abschnitt falsch interpretiert hat. Ohne solche Bezeichnungen ist es größtenteils Spekulation, die Ursache einer schlechten Antwort herauszufinden.

Schritt 6: Zusammenführung

Der gesamte Ablauf umfasst zunächst die Abfrage, danach die Erweiterung und schließlich einen Aufruf des LLM. Die entscheidende Gestaltungsentscheidung besteht darin, Indexierung und Abfragen als zwei getrennte Lebenszyklen zu betrachten anstelle davon, dass eine einzige Funktion beides erledigt:

class RagPipeline:
    def __init__(self, client, model="openai/gpt-oss-20b"):
        self.store = VectorStore()
        self.client = client
        self.model = model

    def build_index(self, documents: list[str]):
        """Run once, whenever documents are loaded or updated."""
        self.store.index(documents, chunk_size=50, overlap=10)

    def answer(self, query: str, top_k: int = 3) -> str:
        """Run on every user query — never re-indexes."""
        results = self.store.search(query, top_k=top_k)
        chunks = [c for c, _ in results]
        prompt = build_augmented_prompt(query, chunks)

        response = self.client.chat.completions.create(
            model=self.model,
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.choices[0].message.content

Würde die Indizierung innerhalb der pro Abfrage laufenden Funktion stattfinden, würde jede Abfrage den Text erneut aufteilen, das Vokabular neu erstellen und alle TF-IDF-Vektoren für die gesamte Sammlung neu berechnen – obwohl sich zwischen den Abfragen nichts ändert. Bei 10.000 Dokumenten und 1.000 Abfragen bedeutet das 999 identische Neuerrichtungen des Index. Der Index einmal zu erstellen und anschließend mehrfach abzufragen, ist die Vorgehensweise, die FAISS, Chroma und Pinecone ebenfalls empfehlen. Es handelt sich dabei um die richtige Architektur von Anfang an und nicht um eine vorzeitige Optimierung.

Was ist mit diesem System nicht in Ordnung?

Die Einschränkungen verdienen Ehrlichkeit, denn es handelt sich dabei nicht um Fehler. Sie folgen direkt aus den oben genannten Designentscheidungen.

• Kein semantisches Verständnis. „Auto“ und „Kraftfahrzeug“ stimmen niemals überein, da TF-IDF nur eine wörtliche Überschneidung im Vokabular erkennt.

Sparse, sehr breite Vektoren. Die Größe des Wortschatzes folgt dem Korpus, wodurch sich jeder Vektor entsprechend erweitert. Ab einigen Tausend Dokumenten wird dies unhandlich.

Im Code gibt es nichts, das überprüft, ob die Suche etwas Nützliches gefunden hat. Selbst wenn alle top-k Ergebnisse nahe bei null liegen, leitet der Prozess sie dennoch an das Modell weiter. Die Aufforderung im Prompt, Unwissenheit einzugestehen, setzt vollständig auf die Kooperation des LLM – daher kann auch eine schwache Suche ungeprüft in die Generierung übergehen.