Vollständige RAG-Pipeline

Zusammenfassung aller Komponenten

In den vorherigen Lektionen wurden die einzelnen Glieder der Kette getrennt behandelt: Laden, Aufteilen, Kodierung, Indizierung, Abruf und Generierung. Jetzt ist es an der Zeit, sie zu einem einzigen, vollständigen, end-to-end-Programm zu verschweißen, das Sie sofort ausführen und dann an Ihre eigenen Dokumente anpassen können. Dieser Skript passt in eine einzige Datei und erfordert nur vier Abhängigkeiten.

pip install mistralai faiss-cpu numpy requests

Die vollständige Pipeline

Zwei Designentscheidungen strukturieren diesen Code. Erstens gruppieren wir alle Konfigurationen am Anfang der Datei: Modelle, Chunk-Größe, Batch-Größe und TOP_K sind benannte Konstanten. Dies spart Ihnen die Suche nach in Funktionen versteckten, fest codierten Werten, wenn Sie das Verhalten anpassen müssen. Zweitens teilen wir die Logik in sechs unabhängige Funktionen auf, die jeweils einem Schritt entsprechen und ihren Fortschritt als [n/6] ausgeben. Diese Spur ist nicht dekorativ: Wenn die Pipeline fehlschlägt oder eine seltsame Antwort zurückgibt, zeigt sie Ihnen sofort, ob das Problem beim Laden, beim Aufteilen oder beim Abruf liegt. Jede Funktion ist auch isoliert austauschbar – Sie können beispielsweise einen PDF-Lader für load_data einfügen, ohne den Rest des Codes zu berühren.

Die main-Funktion verknüpft die Schritte, fragt dann den Korpus mit drei Fragen zur als Demo-Datensatz verwendeten Essay von Paul Graham ab. Beachten Sie, dass Abruf und Generierung innerhalb der Schleife ausgeführt werden, während die Indizierung nur einmal im Voraus erfolgt: Das erneute Kodieren des gesamten Korpus für jede Frage wäre in Bezug auf Kosten und Latenz absurd.

import os
import json
import numpy as np
import faiss
import requests
from mistralai.client import Mistral

# ============================================================
# Konfiguration
# ============================================================
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
EMBED_MODEL = "mistral-embed"
CHAT_MODEL = "mistral-large-latest"
CHUNK_SIZE = 2048
BATCH_SIZE = 25
TOP_K = 3

client = Mistral(api_key=MISTRAL_API_KEY)


# ============================================================
# Schritt 1: Daten laden
# ============================================================
def load_data(url):
    """Lädt ein Textdokument von einer URL."""
    print("[1/6] Daten werden geladen...")
    response = requests.get(url)
    response.raise_for_status()
    text = response.text
    print(f"  -> {len(text)} Zeichen geladen")
    return text


# ============================================================
# Schritt 2: Chunks erstellen
# ============================================================
def create_chunks(text, chunk_size=CHUNK_SIZE):
    """Teilt Text in Chunks auf."""
    print("[2/6] Chunks werden erstellt...")
    chunks = []
    paragraphs = text.split("\n\n")
    current = ""

    for para in paragraphs:
        para = para.strip()
        if not para:
            continue
        if len(current) + len(para) + 2 <= chunk_size:
            current = current + "\n\n" + para if current else para
        else:
            if current:
                chunks.append(current)
            current = para

    if current:
        chunks.append(current)

    print(f"  -> {len(chunks)} Chunks erstellt (durchschnittliche Größe: {sum(len(c) for c in chunks) // len(chunks)} Zeichen)")
    return chunks


# ============================================================
# Schritt 3: Embeddings generieren
# ============================================================
def create_embeddings(chunks, batch_size=BATCH_SIZE):
    """Kodiert Chunks in Embeddings via mistral-embed."""
    print("[3/6] Embeddings werden generiert...")
    all_embeddings = []

    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        response = client.embeddings.create(
            model=EMBED_MODEL,
            inputs=batch,
        )
        all_embeddings.extend([d.embedding for d in response.data])
        print(f"  -> Batch {i // batch_size + 1}/{(len(chunks) - 1) // batch_size + 1} kodiert")

    embeddings = np.array(all_embeddings, dtype=np.float32)
    print(f"  -> Matrix: {embeddings.shape}")
    return embeddings


# ============================================================
# Schritt 4: FAISS-Index erstellen
# ============================================================
def create_index(embeddings):
    """Erstellt und füllt einen FAISS-Index."""
    print("[4/6] FAISS-Index wird erstellt...")
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatL2(dimension)
    index.add(embeddings)
    print(f"  -> Index: {index.ntotal} Vektoren, Dimension {dimension}")
    return index


# ============================================================
# Schritt 5: Relevante Chunks abrufen
# ============================================================
def retrieve(question, index, chunks, k=TOP_K):
    """Ruft die relevantesten Chunks ab."""
    print(f"[5/6] Suche nach: '{question[:50]}...'")
    response = client.embeddings.create(
        model=EMBED_MODEL,
        inputs=[question],
    )
    query_vector = np.array([response.data[0].embedding], dtype=np.float32)

    distances, indices = index.search(query_vector, k)

    results = []
    for dist, idx in zip(distances[0], indices[0]):
        if idx >= 0:
            results.append({
                "text": chunks[idx],
                "distance": float(dist),
            })

    print(f"  -> {len(results)} Chunks abgerufen")
    return results


# ============================================================
# Schritt 6: Antwort generieren
# ============================================================
def generate(question, context_chunks):
    """Generiert die Antwort unter Verwendung des abgerufenen Kontexts."""
    print("[6/6] Antwort wird generiert...")

    context = "\n\n---\n\n".join([c["text"] for c in context_chunks])

    prompt = f"""Informativer Kontext:

{context}

---

Basierend AUF DEM OBEN STEHENDEN KONTEXT beantworten Sie die folgende Frage präzise und vollständig.

Frage: {question}"""

    response = client.chat.complete(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": "Sie sind ein Assistent, der basierend auf dem bereitgestellten Kontext antwortet."},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
    )

    answer = response.choices[0].message.content
    print(f"  -> Antwort generiert ({len(answer)} Zeichen)")
    return answer


# ============================================================
# Haupt-Pipeline
# ============================================================
def main():
    """Führt die vollständige RAG-Pipeline aus."""
    # 1. Laden
    url = "https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt"
    text = load_data(url)

    # 2. Aufteilen
    chunks = create_chunks(text)

    # 3. Kodieren
    embeddings = create_embeddings(chunks)

    # 4. Indizieren
    index = create_index(embeddings)

    # 5+6. Frage -> Antwort
    questions = [
        "Was waren die beiden Hauptprojekte, an denen der Autor vor dem College gearbeitet hat?",
        "Was hat der Autor während seiner Zeit bei Y Combinator getan?",
        "Welche Programmiersprache hat der Autor zuerst gelernt?",
    ]

    for question in questions:
        print(f"\n{'='*60}")
        context = retrieve(question, index, chunks)
        answer = generate(question, context)
        print(f"\nF: {question}")
        print(f"A: {answer}")


if __name__ == "__main__":
    main()

Ausführung der Pipeline

Speichern Sie diesen Code als rag_pipeline.py, exportieren Sie Ihren API-Schlüssel in der Umgebung – codieren Sie ihn niemals direkt in der Datei, da diese sich schließlich in einem Repository befinden wird – und führen Sie dann das Skript aus. Die Ausgabe durchläuft die sechs Schritte und verarbeitet anschließend die drei Fragen mit ihren Antworten. Wenn eine Antwort vom Thema abweicht, vergleichen Sie sie mit den in Schritt 5 angezeigten Chunks, bevor Sie das Generierungsmodell beschuldigen.

export MISTRAL_API_KEY="your-api-key"
python rag_pipeline.py

Fünf Richtungen für weitere Verbesserungen

Diese Pipeline ist ein Fundament, kein fertiges Produkt, und jede ihrer Einschränkungen weist auf einen nächsten Schritt hin. Das unmittelbarste Problem ist die Persistenz: In der aktuellen Form kodiert das Skript den gesamten Korpus bei jeder Ausführung erneut, wohingegen das Speichern des FAISS-Indexes und der Chunks nachfolgende Ausführungen nahezu sofort erledigen würde. Das Re-Ranking ist der nächste Schritt nach oben, bei dem FAISS-Ergebnisse durch einen Cross-Encoder weitergegeben werden, der die Relevanz viel feiner bewertet als nur der Vektorabstand. Die Metadaten-Filterung adressiert einen weiteren Bedarf: Die Suche wird durch Datum, Kategorie oder Autor zusätzlich zur Ähnlichkeit eingeschränkt. Zwei weitere Verbesserungen konzentrieren sich auf die Benutzererfahrung: Streaming, das die Antwort anzeigt, während sie generiert wird, statt nach mehreren Sekunden Stille, und die Gesprächsgeschichte, die es Benutzern ermöglicht, mit „was ist mit Version 2?” nachzufragen, ohne ihre gesamte Frage erneut zu stellen.

# 1. Index speichern, um Neuberechnung zu vermeiden
faiss.write_index(index, "my_rag.index")
np.save("my_chunks.npy", np.array(chunks, dtype=object))

# 2. Re-Ranking hinzufügen
# Nach dem FAISS-Abruf einen Cross-Encoder-Modell verwenden,
# um Ergebnisse nach feiner abgestufter Relevanz neu zu ordnen

# 3. Metadaten-Filterung
# Filter (Datum, Kategorie, Autor) zusätzlich
# zur Vektorähnlichkeit hinzufügen

# 4. Antwort-Streaming
# client.chat.stream() verwenden, um die Antwort
# während der Generierung anzuzeigen

# 5. Gesprächsgeschichte
# Vorherige Austausche dem Prompt hinzufügen
# für einen kontextuellen RAG-Chatbot

Kernaussagen

• Die vollständige RAG-Pipeline gliedert sich in 6 Schritte: laden, aufteilen, kodieren, indizieren, abrufen, generieren
• Jeder Schritt ist eine unabhängige Funktion, einfach zu testen und auszutauschen
• Verwenden Sie temperature=0.1 für faktische Antworten
• Speichern Sie den FAISS-Index und die Embeddings, um bei jeder Ausführung Neuberechnungen zu vermeiden
• Diese Pipeline ist ein solider Ausgangspunkt, um sie nach Ihren Bedürfnissen zu erweitern