RAG Pipeline Architecture

Eine RAG-Pipeline besteht aus zwei Hauptteilen: der Ingestion Pipeline (Daten aufbereiten und indexieren) und der Query Pipeline (Fragen beantworten). Beide müssen sorgfältig designed werden.

Ingestion Pipeline

Die Ingestion Pipeline verarbeitet Rohdaten und macht sie für Retrieval verfügbar:

Rohdaten → Loader → Splitter → Enricher → Embedder → Vector Store

Schritt-für-Schritt

# 1. Dokumente laden
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("company_handbook.pdf")
documents = loader.load()

# 2. Chunks erstellen
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=150)
chunks = splitter.split_documents(documents)

# 3. Metadata anreichern
for chunk in chunks:
    chunk.metadata["category"] = classify_content(chunk.page_content)
    chunk.metadata["summary"] = generate_summary(chunk.page_content)

# 4. Embeddings erstellen und speichern
from langchain_community.vectorstores import Chroma
vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings())

Query Pipeline

Die Query Pipeline beantwortet Fragen basierend auf den indexierten Daten:

Frage → Query Transformation → Retrieval → Re-Ranking → Context → LLM → Antwort

Query Transformation

Vor dem Retrieval kann die Frage optimiert werden:

Re-Ranking

Nach dem Retrieval werden die Ergebnisse neu sortiert für höhere Relevanz:

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

reranker = CohereRerank(model="rerank-v3.5", top_n=3)
retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)
# Holt 20 Ergebnisse, re-ranked auf Top 3

Warum Re-Ranking?

Contextual Compression

Chunks enthalten oft irrelevante Teile. Contextual Compression extrahiert nur die relevanten Passagen:

from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)
retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

Multi-Hop Retrieval

Manche Fragen erfordern mehrere Retrieval-Schritte:

Frage: "Wie hat sich der Umsatz des Produkts mit dem höchsten Kundenzufriedenheitsscore entwickelt?"

Hop 1: "Welches Produkt hat den höchsten Kundenzufriedenheitsscore?"
        → Produkt X

Hop 2: "Wie hat sich der Umsatz von Produkt X entwickelt?"
        → Umsatzdaten

Implementation

# Iteratives Retrieval
def multi_hop_retrieve(question: str, max_hops: int = 3):
    context = []
    current_query = question

    for hop in range(max_hops):
        results = retriever.invoke(current_query)
        context.extend(results)

        # Prüfe ob genug Kontext vorhanden
        if has_sufficient_context(context, question):
            break

        # Generiere Follow-up-Query basierend auf bisherigem Kontext
        current_query = generate_followup(question, context)

    return context

Pipeline-Architektur

                    ┌─── Query Rewrite ───┐
                    │                     │
User Query ──▶ Router ──▶ Retriever ──▶ Re-Ranker ──▶ LLM ──▶ Answer
                    │                     │
                    └─── Metadata Filter ─┘

Praxis-Tipp: Starten Sie mit einer einfachen Pipeline: Retriever + LLM. Fügen Sie Re-Ranking hinzu, wenn die Relevanz nicht ausreicht. Query Transformation lohnt sich, wenn Nutzer umgangssprachlich fragen. Multi-Hop ist nur für komplexe Fragen nötig — messen Sie, bevor Sie optimieren.