Erstellen Sie eine lokale RAG-Pipeline in Python ohne API-Kosten
Ein praktischer Tutorial, der Retrieval-augmented Generation auf CPU mit Sentence-Transformers, NumPy und Flan-T5 erstellt, einschließlich Chunking, Suche und einem A/B-Test.
Quellen: Build a RAG System From Scratch — Hands-On, No API Costs
Ein vollständiger Retrieval-Augmented Generation-Pipeline kann ausschließlich auf einem Personalcomputer erstellt und ausgeführt werden – nur mit CPU, kostenlosen Open-Source-Modellen, ohne OpenAI-Schlüssel, ohne Kreditkarte und ohne Geschwindigkeitsbeschränkungen. Es geht darum, jeden Bestandteil zu verstehen, anstatt eine Blackbox zu kopieren.
Der beste Weg ist die praktische Anwendung: Halten Sie ein Terminal sowie eine Python-Datei mit dem Namen rag.py geöffnet, führen Sie jeden Codeblock aus und prognostizieren Sie die Ausgabe, bevor Sie ihn tatsächlich ausführen. Nach den meisten Schritten folgen kurze Übungen, in denen der größte Teil des Lernens stattfindet.
Das Problem, das RAG löst
Ein Sprachmodell kennt nur das, was in seinen Trainingsdaten enthalten ist. Private Notizen, eine Unternehmens-Wiki, die Besprechung von letzter Woche oder jedes nicht gesehene Dokument sind für es unsichtbar. Wenn danach gefragt wird, gibt es entweder zu, nichts zu wissen, oder – schlimmer noch – erfindet etwas mit Überzeugung.
RAG teilt die Aufgabe in zwei Teile auf:
- Auswählen: Finden Sie die Abschnitte Ihrer eigenen Dokumente, die zur Frage passen. Das ist ein Suchproblem.
- Erzeugen: Geben Sie diese Abschnitte dem Modell zusammen mit der Anweisung, daraufhin eine Antwort zu geben.
Das Modell hört auf zu raten und beginnt zu lesen. Alles Weitere sind Mechanismen, um diese beiden Schritte gut auszuführen.
question ──► [embed] ──► [search your docs] ──► top chunks ──┐
▼
[LLM: "answer using this context"] ──► answer
Das kostenlose Toolset
Drei Komponenten bilden das gesamte System. sentence-transformers wandelt Text in Vektoren um, wobei ein kleines Modell verwendet wird, das auf der CPU läuft und nur einmal heruntergeladen werden muss. numpy kümmert sich um die Mathematik zur Berechnung der Ähnlichkeit, und die gesamte „Vektordatenbank“ passt in etwa fünf Zeilen – dadurch bleiben die Abläufe übersichtlich. transformers mit Flan-T5 schreibt die Antwort, da dieses Modell auf Anweisungen trainiert ist, auf der CPU läuft und keine API benötigt.
Installieren Sie alles einmal:
pip install sentence-transformers transformers torch numpy
Zum ersten Mal beim Laden holt jedes Modell seine Gewichte von Hugging Face. Der Embedder ist etwa 80 MB groß und flan-t5-base rund 1 GB. Danach werden sie im Cache gespeichert und ermöglichen den Betrieb offline. Bei begrenztem Speicherplatz oder RAM kann flan-t5-small (etwa 300 MB) das Basismodell vollständig ersetzen – zwar in geringerer Qualität, aber mit deutlich höherer Geschwindigkeit.
Schritt 1: ein Korpus, das das Modell nicht kennt
RAG lässt sich am besten mit Dokumenten verstehen, die kein Modell je gesehen haben; daher beginnt das Skript mit einer kleinen fiktiven Unternehmens-Wiki. Erstellen Sie rag.py und fügen Sie die Dokumente ein:
# rag.py
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
Die Wiki enthält weitere Einträge:
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""", """Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""", """The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""", """Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""", """Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]
Vor dem Bau irgendetwas sollte man zunächst nachweisen, dass das Problem tatsächlich existiert. In einer Python-Shell kann man ein einfaches Modell nach Nimbus fragen:
from transformers import pipeline
gen = pipeline("text2text-generation", model="google/flan-t5-base")
print(gen("How much does Nimbus Pro cost?", max_new_tokens=50)[0]["generated_text"])
Es wird etwas erzeugen, und dieses Etwas wird falsch oder leer sein, weil es sich um eine Schätzung handelt. Behalten Sie diese Antwort im Hinterkopf, da der Rest des Tutorials sie behebt.
Schritt 2: Chunking
Ganze Dokumente werden nicht an das Modell übergeben. Sie werden aus zwei Gründen in kleinere Chunks aufgeteilt: Die Abfrage wird genauer, da das Modell den Absatz findet, der die Frage beantwortet, und nicht ein ganzes Handbuch, außerdem ist das Kontextfenster des Modells begrenzt.
Fügen Sie eine Chunking-Funktion hinzu:
def chunk_text(text, chunk_size=60, overlap=15):
"""Split text into overlapping chunks of `chunk_size` words."""
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 by `overlap` so context isn't cut
return chunks
Dann erstellen Sie die Chunk-Liste, wobei Sie festhalten, aus welchem Dokument jeder Chunk stammt:
# Build our chunk list, remembering which doc each chunk came from
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})print(f"{len(DOCUMENTS)} documents -> {len(chunks)} chunks")
for c in chunks[:3]:
print("-", c["text"][:70], "...")
Überschneidungen entstehen, weil eine Tatsache eine Grenze überschreiten kann, wie beispielsweise eine Rückerstattungspolitik, die zwischen „innerhalb von 5“ und „Bürotagen“ angesetzt ist. Ohne Überschneidung enthält keines der Teilstücke die gesamte Tatsache. Durch das Wiederholen einiger Wörter an jeder Schnittstelle können Tatsachen den Schnitt überleben.
Übung: Führen Sie den Vorgang erneut mit chunk_size=20 durch, um mehr kleinere Teilstücke zu erhalten, und versuchen Sie anschließend chunk_size=500, wobei aus diesen kurzen Dokumenten jeweils ein einzelnes Teilstück entsteht. Bevor Sie starten, raten Sie, welche Einstellung präziser ist, aber das Risiko birgt, den umliegenden Kontext zu verlieren. Kleine Teilstücke sind präzise, aber kontextarm – und diese Spannung liegt im Kern der Teilung in Teilstücke. Es gibt keine universell richtige Größe.
Schritt 3: Embeddings
Ein Embedding ist ein Vektor aus Zahlen, der die Bedeutung eines Textes kodiert. Textabschnitte, die ähnliche Bedeutungen haben, ergeben Vektoren, die in dieselbe Richtung zeigen – selbst wenn sie keine gemeinsamen Wörter enthalten – wodurch „How much is it?“ neben „What is the price?“ landet.
Importieren Sie das Embedding-Modell:
from sentence_transformers import SentenceTransformer
Dann fügen Sie jeden Teil ein:
embedder = SentenceTransformer("all-MiniLM-L6-v2")# Embed every chunk. normalize_embeddings=True makes the vectors unit-length,
# which lets us measure similarity with a simple dot product later.
chunk_texts = [c["text"] for c in chunks]
chunk_vectors = embedder.encode(chunk_texts, normalize_embeddings=True)print("vector shape:", chunk_vectors.shape) # (num_chunks, 384)
Jeder Datenblock entspricht nun einem 384-dimensionalen Vektor, wobei 384 die Anzahl der Zahlen ist, die dieses spezielle Modell zur Darstellung von Bedeutung verwendet.
Übung: Beobachten Sie, wie Bedeutung in einem kleinen Experiment zum Tragen kommt:
import numpy as np
pairs = embedder.encode(
["the price of the pro plan", "how much does it cost", "the weather in Paris"],
normalize_embeddings=True,
)
print("price vs cost :", round(float(pairs[0] @ pairs[1]), 3)) # should be HIGH
print("price vs weather:", round(float(pairs[0] @ pairs[2]), 3)) # should be LOW
Die Ausdrucksform a @ b bezeichnet ein Skalarprodukt, das für Einheitsvektoren der Kosinusähnlichkeit entspricht. Ein Wert von 1,0 bedeutet dieselbe Richtung, während ein Wert von 0 auf keine Beziehung hindeutet. Das erste Paar sollte einen deutlich höheren Wert aufweisen als das zweite, obwohl die Phrasen keine gemeinsamen Wörter enthalten – das ist semantische Suche. Versuchen Sie, beide Werte vor dem Ausführen auf unter 0,2 zu schätzen.
Schritt 4: Abruf
Zur Suchschritt werden die Frage in Vektoren umgewandelt, diese mit jedem Chunk-Vektor verglichen und die besten Ergebnisse ausgewählt. Da die Vektoren normalisiert sind, reduziert sich das „Vergleichen“ auf eine Matrixmultiplikation.
import numpy as np
Die Abruffunktion:
def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec # cosine similarity to every chunk
top_idx = np.argsort(scores)[::-1][:k] # indices of the k highest scores
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]for text, score in retrieve("How much does Nimbus Pro cost?"):
print(f"[{score:.3f}] {text[:80]}...")
Der Preis-Abschnitt sollte mit der besten Bewertung an die Spitze rücken, obwohl in der Frage „Kosten“ stand und im Dokument „8 Dollar pro Monat Kosten“ angegeben waren. Es geht um Bedeutungsebene statt um Schlüsselwortübereinstimmung.
Übung: Probieren Sie drei Fragen aus und sehen Sie, was angezeigt wird:
retrieve("Kann ich mein Geld zurückbekommen?")sollte den Rückerstattungs-Abschnitt liefern, obwohl das Wort „Rückerstattung“ in der Frage nie vorkam.retrieve("Sind meine Daten sicher?")sollte den Verschlüsselungs-Abschnitt anzeigen.retrieve("Funktioniert es auch ohne Internetverbindung?")soll den Offline-Sync-Abschnitt zurückgeben.
Viele Menschen unterschätzen diese Phase. Wenn die Abrufung falsche Datenblöcke zurückgibt, kann kein cleverer Prompt die richtige Antwort liefern. Was hineingefüttert wird, ist das, was herauskommt – daher muss hier zuerst an der Fehlersuche gearbeitet werden.
Schritt 5: Generierung
Nun treffen die beiden Teile aufeinander. Erstellen Sie einen Prompt, der die abgerufenen Datenblöcke an Flan-T5 weiterleitet und ihm vorschreibt, ausschließlich darauf zu antworten.
from transformers import pipeline
Die Generierungsfunktion:
generator = pipeline("text2text-generation", model="google/flan-t5-base")def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved) prompt = f"""Answer the question using only the context below.
If the answer is not in the context, say you don't know.Context:
{context}Question: {question}
Answer:""" out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip(), retrievedanswer, sources = rag_answer("How much does Nimbus Pro cost?")
print("ANSWER:", answer)
print("\nBased on:")
for text, score in sources:
print(f" [{score:.3f}] {text[:70]}...")
Vergleichen Sie das Ergebnis mit der falschen oder leeren Antwort aus Schritt 1. Das Modell und die Frage sind identisch, der einzige Unterschied besteht darin, dass nun relevanter Text zum Lesen bereitgestellt wird. Dieser Unterschied stellt den gesamten Wert von RAG dar.
Übung: Stellen Sie eine Frage, die in den Dokumenten nicht behandelt wird – beispielsweise zu einem Linux-Desktop-Klient für Nimbus – und rufen Sie dazu rag_answer auf. Eine ordnungsgemäß konfigurierte Lösung gibt zu, dass sie die Antwort nicht kennt. Wenn Ihre Lösung eine Antwort erfindet, handelt es sich um einen echten Fehlerzustand, der als Halluzination bezeichnet wird, obwohl Daten abgerufen wurden; dazu wird in einem späteren Abschnitt eingegangen.
Das fertige Skript
Hier ist der gesamte rag.py als ein einziges ausführbares Programm mit einer interaktiven Schleife. Es beginnt mit den Importen:
# rag.py — a complete, local, no-API RAG system
import numpy as np
from sentence_transformers import SentenceTransformer
from transformers import pipeline
Der vollständige Text folgt:
DOCUMENTS = [
"""Nimbus is a fictional note-taking app launched in 2023. The free plan,
called Nimbus Lite, allows up to 50 notes and 1 GB of storage. There are no
collaboration features on the free plan.""",
"""Nimbus Pro costs 8 dollars per month billed annually, or 10 dollars billed
monthly. Pro removes the note limit, gives 50 GB of storage, and unlocks
real-time collaboration with up to 5 people per note.""",
"""Nimbus stores all notes encrypted at rest using AES-256. End-to-end
encryption is only available on the Pro plan and must be enabled manually in
Settings > Security. Once enabled it cannot be turned off for that note.""",
"""The Nimbus mobile app supports offline editing. Changes made offline are
queued and sync automatically the next time the device is online. If two
devices edit the same note offline, Nimbus keeps both versions and flags a
conflict for the user to resolve.""",
"""Nimbus offers a 30-day refund policy on all paid plans, no questions asked.
Refunds are processed to the original payment method within 5 business days.
Annual plans cancelled after 30 days are not refundable but stay active until
the end of the billing period.""",
"""Nimbus support is available via email at help@nimbus.example and live chat.
Live chat is only staffed for Pro customers, Monday to Friday, 9am to 6pm UTC.
Free-plan users receive email support with a typical 48-hour response time.""",
]def chunk_text(text, chunk_size=60, overlap=15):
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
return chunksprint("Loading models (first run downloads them)...")
embedder = SentenceTransformer("all-MiniLM-L6-v2")
generator = pipeline("text2text-generation", model="google/flan-t5-base")# Index the documents once at startup
chunks = []
for doc_id, doc in enumerate(DOCUMENTS):
for c in chunk_text(doc):
chunks.append({"doc_id": doc_id, "text": c})
chunk_vectors = embedder.encode(
[c["text"] for c in chunks], normalize_embeddings=True
)def retrieve(question, k=3):
q_vec = embedder.encode([question], normalize_embeddings=True)[0]
scores = chunk_vectors @ q_vec
top_idx = np.argsort(scores)[::-1][:k]
return [(chunks[i]["text"], float(scores[i])) for i in top_idx]def rag_answer(question, k=3):
retrieved = retrieve(question, k=k)
context = "\n".join(text for text, _ in retrieved)
prompt = (
"Answer the question using only the context below. "
"If the answer is not in the context, say you don't know.\n\n"
f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
)
out = generator(prompt, max_new_tokens=80)[0]["generated_text"]
return out.strip()if __name__ == "__main__":
print("RAG ready. Ask about Nimbus (or type 'quit').\n")
while True:
q = input("You: ").strip()
if q.lower() in {"quit", "exit", ""}:
break
print("Nimbus bot:", rag_answer(q), "\n")
Fangen Sie damit an:
python rag.py
Probieren Sie dann Anfragen wie zum Beispiel, wie lang das Rückgabefenster ist, ob kostenlose Nutzer einen Live-Chat erhalten oder wie groß die kostenlose Speicherkapazität ist.
Den Wert durch einen A/B-Test nachweisen
Ein Experiment macht die Sache klar, indem es dieselbe Frage mit und ohne abgerufenen Kontext stellt. Zunächst eine Basismethode:
def no_rag(question):
out = generator(f"Question: {question}\nAnswer:", max_new_tokens=80)
return out[0]["generated_text"].strip()
Dann der Vergleich:
q = "Can free-plan Nimbus users use live chat support?"
print("WITHOUT context:", no_rag(q))
print("WITH context :", rag_answer(q))
Ohne Kontext ist die Antwort nur eine Vermutung, während eine Antwort im Kontext klarstellen sollte, dass der Live-Chat nur für Premium-Nutzer verfügbar ist und kostenlose Nutzer per E-Mail kontaktiert werden. Probieren Sie drei verschiedene Fragen aus. Dieses A/B-Verfahren ist die überzeugendste RAG-Demo, die es gibt – nutzen Sie es daher, um die Idee anderen zu erklären.
Upgrades nach Nutzen
Sobald man ein funktionierendes System hat, sind diese Upgrades am wichtigsten, ungefähr in Reihenfolge des Nutzens:
- Quellen anzeigen. Die Funktion gibt bereits
retrievedzurück, daher sollten die jeweiligen Abschnitte unter jeder Antwort ausgegeben werden. Echte RAG-Produkte bestehen oder scheitern an diesem Punkt. Eine zitierte Antwort ist vertrauenswürdig, während eine unzitierte Antwort lediglich als Vermutung in einer schönen Schrift dargestellt wird.
k auf 5 und legen Sie eine Mindestpunktzahl fest, wodurch alles unter etwa 0,25 weggeworfen wird. Wenn kein Datenblock bestehen bleibt, antworten Sie mit „Ich weiß es nicht“ und überspringen Sie den gesamten Modellaufruf. Dadurch werden die meisten Halluzinationen, wie sie im Übungsschritt 5 auftreten, beseitigt.chunk_vectors @ q_vec durchsucht jeden Datenblock, was für Hunderte in Ordnung ist, aber bei Millionen langsam wird. Chroma (pip install chromadb) indiziert Vektoren für eine schnelle Suche und speichert sie auf der Festplatte, sodass beim Starten nichts erneut eingebettet werden muss. Die Konzepte des Einbettens, Speicherns und der Kosinus-Suche bleiben unverändert; Chroma führt sie lediglich in großem Maßstab aus.generator(...) durch eine Anfrage an dessen lokale API ersetzen. Ein kostenloser API-Plan von Google’s Gemini oder Groq funktioniert ebenfalls, indem lediglich die Generierungszeile in rag_answer ausgetauscht wird. Nur eine Funktion ändert sich, während die Informationsabruflogik unverändert bleibt.rag_answer und überprüfen Sie die Richtigkeit. Das ist zwar nicht besonders ansprechend, doch genau das unterscheidet ein Prototyp von einem echten Produkt – denn was nicht gemessen wird, kann auch nicht verbessert werden.Probleme diagnostizieren
RAG ist eine Suchfunktion in Kombination mit einem Schreibassistenten; fast jedes Problem lässt sich auf eines dieser beiden Elemente zurückführen. Eine falsche Antwort bei korrektem Abruf eines Fragments ist ein Problem mit der Generierung oder dem Prompt. Eine falsche Antwort, bei der das richtige Fragment nie abgerufen wurde, ist ein Problem mit dem Abrufprozess im Zusammenhang mit Fragmentierung, Embeddings oder k. Bestimmen Sie zunächst, welches Element fehlerhaft ist, bevor Sie etwas ändern. retrieve() und rag_answer() als separate Funktionen zu halten, dient genau dazu, dass jedes Element einzeln überprüft werden kann – und diese Trennung ist hier die wertvollste Gewohnheit.
Fehlerbehebung
- Schlechter oder fehlgeschlagener Download: Die Modelle werden beim ersten Start von Hugging Face heruntergeladen und im Verzeichnis
~/.cache/huggingfacegespeichert. Sobald sie gespeichert sind, funktionieren sie auch ohne Internetverbindung.
google/flan-t5-small mit einer chunk_size von 40. Das Embedding-Modell ist klein genug, um keine Verzögerungen zu verursachen.pip install torch --index-url https://download.pytorch.org/whl/cpu.max_new_tokens, da Flan-T5 dazu neigt, frühzeitig aufzuhören. Ein größeres Generierungsmodell wie das von Ollama hilft mehr als jede Anpassung.Durch das Bearbeiten der Übungen entwickelt man ein tieferes Verständnis für RAG als die meisten Menschen durch die Verwendung von Bibliothekskapseln.