Vollständige RAG-Architektur

Lektion aktualisiert am 3. November 2026 basierend auf der OpenAI-Dokumentation: temperature und top_p werden nur mit einem Reasoning-Effort von none akzeptiert, während GPT-5.6-Modelle standardmäßig nachdenken (medium). Aufrufe, die diese Parameter festlegen, müssen daher reasoning={"effort": "none"} (API Responses) oder reasoning_effort="none" (Chat Completions) angeben. Quellen: GPT-6 verwenden · GPT-5.6 Terra.

Lernziele

• Die RAG-Architektur von Anfang bis Ende verstehen
• Eine funktionierende RAG-Pipeline implementieren
• Komponenten identifizieren, die optimiert werden können

Was ist RAG?

RAG (Retrieval-Augmented Generation) kombiniert die Dokumentensuche mit der Textgenerierung durch ein LLM. Anstatt sich ausschließlich auf sein internes Wissen zu verlassen, erhält das Modell relevante Dokumente als Kontext und formuliert daraufhin seine Antwort. Alles, was Sie in diesem Kurs bisher aufgebaut haben – Embeddings, Indexierung, Suche, Evaluation – bildet hier die vorgelagerte Hälfte eines vollständigen Systems.

Warum RAG?

Vier Gründe rechtfertigen den Einsatz, und sie verstärken sich gegenseitig. Erstens eigenes Datenmaterial: Ihre Remote-Arbeitsrichtlinien, interne Verfahren und Produktkataloge waren nie in den Trainingsdaten des Modells enthalten, und kein Prompt wird dies ändern. Zweitens aktuelle Informationen: Alles, was nach dem Training hinzugekommen ist, ist für das Modell unsichtbar, wohingegen ein Index sich täglich aktualisiert. Überprüfbarkeit ist ebenso wichtig, da jede Antwort in identifizierbaren Quellen verankert ist, die der Nutzer öffnen und prüfen kann – dies verwandelt die Art des Vertrauens, das in das System gesetzt wird. Schließlich wirkt der Kostenfaktor zu Ihren Gunsten: Das Hinzufügen von Wissen über RAG kostet weit weniger als das Feinabstimmen (Fine-Tuning), und Korrekturen erfolgen durch das Löschen eines Dokuments statt durch das Neutrainieren eines Modells.

Pipeline-Stufen

Lesen Sie RAG in zwei klar getrennte Phasen, die Sie nicht verwechseln dürfen. Die Ingestion erfolgt offline, einmal und für alle Zeiten oder whenever das Korpus aktualisiert wird:

Dokumente → Chunking → Embeddings → Vektorspeicher

Die Abfrage erfolgt online für jede gestellte Frage und muss schnell bleiben:

Frage → Embedding → Suche → Kontext + Frage → LLM → Antwort

Diese Trennung hat eine praktische Konsequenz: Die Kosten für die Ingestion werden einmalig bezahlt und können hoch sein, während die Kosten für die Abfrage pro Aufruf anfallen und die Benutzererfahrung bestimmen. Sie können sich daher aufwändige Ingestionsverarbeitung leisten, müssen aber auf der Abfrage-Seite schlank bleiben.

Vollständige Implementierung

Die folgende Klasse materialisiert die drei Phasen in drei separaten Methoden sowie einer vierten, die sie zusammenführt. Diese Trennung ist nicht kosmetisch: Sie ermöglicht es Ihnen, die Suche zu testen, ohne Generierungstokens zu verbrauchen, und die Generierung auf einem festen Kontext wiederzugeben, um zwei Prompts zu vergleichen.

from openai import OpenAI
import numpy as np

client = OpenAI()

class PipelineRAG:
    def __init__(
        self,
        model_embedding: str = "text-embedding-3-large",
        model_generation: str = "gpt-5.6-terra"
    ):
        self.client = OpenAI()
        self.model_embedding = model_embedding
        self.model_generation = model_generation
        self.chunks: list[dict] = []
        self.embeddings: np.ndarray | None = None

    # --- Ingestion Phase ---

    def ingest(self, documents: list[dict]):
        """Importiert Dokumente {id, text, metadata}."""
        self.chunks = documents
        texts = [d["text"] for d in documents]

        # Generiert Embeddings in Batches
        all_emb = []
        for i in range(0, len(texts), 100):
            batch = texts[i:i + 100]
            resp = self.client.embeddings.create(
                input=batch, model=self.model_embedding
            )
            batch_emb = sorted(resp.data, key=lambda x: x.index)
            all_emb.extend([e.embedding for e in batch_emb])

        self.embeddings = np.array(all_emb, dtype=np.float32)
        print(f"Ingestion abgeschlossen: {len(self.chunks)} Chunkes indexiert")

    # --- Retrieval Phase ---

    def search(self, question: str, k: int = 5) -> list[dict]:
        """Findet die k relevantesten Chunkes."""
        query_emb = self.client.embeddings.create(
            input=question, model=self.model_embedding
        ).data[0].embedding

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

        return [
            {**self.chunks[i], "score": float(scores[i])}
            for i in top_k
        ]

    # --- Generation Phase ---

    def generate(self, question: str, contexts: list[dict]) -> str:
        """Generiert eine Antwort basierend auf abgerufenen Kontexten."""
        # Baut den Prompt auf
        formatted_context = "\n\n---\n\n".join(
            f"[Quelle: {c.get('id', 'unknown')}]\n{c['text']}"
            for c in contexts
        )

        messages = [
            {
                "role": "system",
                "content": (
                    "Sie sind ein Assistent, der Fragen ausschließlich "
                    "auf Basis der bereitgestellten Dokumente beantwortet. "
                    "Wenn die Informationen nicht in den Dokumenten enthalten sind, "
                    "stellen Sie dies klar fest. Zitieren Sie Ihre Quellen."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Referenzdokumente:\n\n{formatted_context}\n\n"
                    f"---\n\nFrage: {question}"
                )
            }
        ]

        response = self.client.chat.completions.create(
            model=self.model_generation,
            reasoning_effort="none",  # temperature erfordert effort none
            messages=messages,
            temperature=0.2
        )

        return response.choices[0].message.content

    # --- Complete Pipeline ---

    def answer(self, question: str, k: int = 5) -> dict:
        """Vollständige RAG-Pipeline: Suche → Generierung."""
        contexts = self.search(question, k=k)
        response = self.generate(question, contexts)

        return {
            "question": question,
            "response": response,
            "sources": [
                {"id": c["id"], "score": c["score"],
                 "excerpt": c["text"][:150]}
                for c in contexts
            ]
        }

Drei Details in der Generierungsphase treiben die meisten Qualitätsaspekte voran. Die Systemnachricht verlangt, dass ausschließlich auf den bereitgestellten Dokumenten geantwortet wird und dass zugegeben wird, wenn Informationen fehlen: Ohne diese Anweisung füllt das Modell Lücken mit seinem allgemeinen Wissen, was zu einer plausiblen, aber erfundenen Antwort führt. Der Präfix [Quelle: id] vor jedem Auszug gibt dem Modell etwas Spezifisches, das es zitieren kann, statt einer vagen „laut den Dokumenten”. Schließlich reduziert eine temperature von 0,2 die Variabilität: Bei faktischen Fragen gegen eine Dokumentenbasis wünschen Sie Konsistenz, nicht Kreativität.

Nutzung

Testen wir die Pipeline an einem minimalen Geschäftsfall – drei interne Richtlinien und eine Frage, die das Wortlaut eines ihrer Texte nicht kopiert.

# Vorbereiten von Dokumenten
documents = [
    {
        "id": "urlaubspolitik",
        "text": "Mitarbeiter erhalten 25 Tage bezahlten Urlaub "
                 "pro Jahr. Urlaub muss mindestens 2 Wochen "
                 "im Voraus über das HR-Portal beantragt werden.",
        "metadata": {"type": "hr", "datum": "2026-01"}
    },
    {
        "id": "remote-work-policy",
        "text": "Remote-Arbeit ist bis zu 3 Tage pro Woche genehmigt. "
                 "Dienstag und Donnerstag sind verpflichtende Bürotage. "
                 "Eine schriftliche Vereinbarung mit dem Vorgesetzten ist erforderlich.",
        "metadata": {"type": "hr", "datum": "2026-01"}
    },
    {
        "id": "erstattungsverfahren",
        "text": "Aufwandsnachweise müssen innerhalb von 30 Tagen "
                 "nach der Ausgabe eingereicht werden. Fügen Sie den "
                 "Originalquittung bei. Validierung durch den direkten "
                 "Vorgesetzten erforderlich für Beträge über 200 Euro.",
        "metadata": {"type": "finance", "datum": "2026-01"}
    },
]

# Erstellen der Pipeline
rag = PipelineRAG()
rag.ingest(documents)

# Eine Frage stellen
result = rag.answer("Wie viele Tage Remote-Arbeit pro Woche?")
print(result["response"])
print("\nVerwendete Quellen:")
for s in result["sources"]:
    print(f"  [{s['score']:.3f}] {s['id']}: {s['excerpt']}")

Treiben Sie die Übung weiter, indem Sie eine Frage stellen, die kein Dokument beantwortet, wie zum Beispiel eine zum Firmen-Gesundheitsplan. Die Pipeline wird dennoch drei Quellen zurückgeben – die Suche gibt immer ihre k besten Kandidaten zurück – und es ist die Systemnachricht, die verhindern muss, dass das Modell eine Antwort erfindet. Diese Verhaltensweise vor der Produktion zu überprüfen, ist besser als sie durch eine Beschwerde zu entdecken.

Optimierungspunkte

Jeder Schritt der Pipeline kann unabhängig verbessert werden, was Fortschritte durch kleine, messbare Anpassungen ermöglicht.

Die folgenden Lektionen erläutern jede dieser Optimierungen. Bearbeiten Sie sie in dieser Reihenfolge: Das Chunking bestimmt, was abgerufen werden kann, und kein Reranking wird ein schlecht chunkiertes Dokument beheben.

Zusammenfassung

• RAG = Suche + Kontext + LLM-Generierung
• Die Pipeline teilt sich in Ingestion (offline) und Abfrage (online)
• Jede Komponente (Chunking, Embedding, Suche, Prompt, Generierung) kann unabhängig optimiert werden
• RAG ist flexibler und kostengünstiger als das Feinabstimmen zum Hinzufügen von Wissen