Adaptives RAG: KI die automatisch die beste Strategie waehlt

Warum einen Hammer fuer jede Schraube verwenden? Festes RAG wendet die gleiche Retrieval-Strategie auf jede Abfrage an, ob einfach oder komplex. Adaptives RAG analysiert zuerst die Frage und waehlt dann automatisch den besten Ansatz: kein Retrieval, einfaches Retrieval oder mehrstufiges Retrieval. Ergebnis: eine Genauigkeit nahe an einer systematischen Multi-Step-Strategie, aber mit deutlich weniger Suchaufrufen (durchschnittlich 1,08 statt 4,69 Schritte) und geringeren Kosten bei einfachen Fragen.

TL;DR

• Adaptives RAG routet dynamisch jede Abfrage zur optimalen Strategie
• Drei Routen: No-Retrieval (direktes LLM), Single-Step-Retrieval, Multi-Step-Retrieval
• Klassifikator: Leichtgewichtiges LLM oder trainiertes Modell, das die Abfragekomplexitaet analysiert
• Genauigkeit nahe am systematischen Multi-Step-Ansatz bei ~4x weniger Retrieval-Schritten — durchschnittlich 1,08 statt 4,69 (Adaptive-RAG, Jeong et al., 2024)
• Weniger Retrieval-Aufrufe durch Vermeidung unnoetiger Suche bei einfachen Fragen
• Kompatibel mit CRAG, Self-RAG und anderen fortgeschrittenen Techniken

Das Problem mit festem RAG

Festes RAG behandelt alle Abfragen gleich. Das ist Verschwendung.

Festes RAG (Einheitsgroesse fuer alle):

"Hallo!" ──────────────────▶ Retrieval + Generierung  ❌ Unnoetig
"Wie spaet ist es?" ───────▶ Retrieval + Generierung  ❌ Unnoetig
"Ihre Rueckgaberichtlinie?" ▶ Retrieval + Generierung  ✅ Relevant
"Vergleichen Sie Plan A und B
 und empfehlen Sie basierend
 auf meinem KMU-Profil" ──▶ Retrieval + Generierung  ⚠️ Unzureichend
                                                        (braucht Multi-Step)

Die drei Abfragetypen

*Typische Verteilung bei E-Commerce- und Support-Chatbots.

Architektur des adaptiven RAG

┌──────────────────────────────────────────────────────────────────────┐
│                     Adaptive RAG Pipeline                             │
├──────────────────────────────────────────────────────────────────────┤
│                                                                       │
│  Benutzerabfrage                                                      │
│         │                                                             │
│         ▼                                                             │
│  ┌─────────────────┐                                                 │
│  │  Abfrage-        │  Analyse: Komplexitaet, faktischer Bedarf,     │
│  │  Klassifikator   │  Domaene, erwarteter Antworttyp                │
│  └────────┬────────┘                                                 │
│           │                                                           │
│     ┌─────┼─────────────────┐                                        │
│     │     │                 │                                        │
│     ▼     ▼                 ▼                                        │
│  ┌──────┐ ┌──────────┐ ┌────────────────┐                           │
│  │ Kein │ │ Single   │ │ Multi-Step     │                           │
│  │Retr. │ │ Step     │ │ Retrieval      │                           │
│  │      │ │ Retrieval│ │                │                           │
│  │ LLM  │ │ Abfrage→ │ │ Zerlegung      │                           │
│  │direkt│ │ Abrufen  │ │ → Unterfragen  │                           │
│  │      │ │ → Generi.│ │ → Multi-Abruf  │                           │
│  │      │ │          │ │ → Synthese     │                           │
│  └──┬───┘ └────┬─────┘ └───────┬────────┘                           │
│     │          │               │                                     │
│     └──────────┴───────────────┘                                     │
│                    │                                                  │
│                    ▼                                                  │
│             Finale Antwort                                           │
│                                                                       │
└──────────────────────────────────────────────────────────────────────┘

Der Abfrage-Klassifikator

Das Herzstuck des adaptiven RAG ist der Klassifikator, der die optimale Strategie fuer jede Abfrage bestimmt.

Ansatz 1: LLM-Klassifikator (Zero-Shot)

Die einfachste Methode: Ein leichtgewichtiges LLM zur Klassifizierung verwenden.

DEVELOPERpython
from langchain_openai import ChatOpenAI
from typing import Literal
from pydantic import BaseModel, Field

class QueryClassification(BaseModel):
    """Abfrageklassifizierung fuer RAG-Routing."""
    complexity: Literal["simple", "factual", "complex"] = Field(
        description="Komplexitaetsstufe der Abfrage"
    )
    reasoning: str = Field(
        description="Erklaerung der Klassifizierung"
    )
    needs_retrieval: bool = Field(
        description="Benoetigt die Abfrage Dokumente?"
    )
    estimated_steps: int = Field(
        description="Anzahl benoetigter Retrieval-Schritte",
        ge=0, le=5
    )

class QueryClassifier:
    def __init__(self):
        self.llm = ChatOpenAI(
            model="gpt-4o-mini",
            temperature=0
        ).with_structured_output(QueryClassification)

    def classify(self, query: str) -> QueryClassification:
        prompt = f"""Klassifiziere diese Abfrage, um die optimale Retrieval-Strategie zu bestimmen.

Abfrage: "{query}"

Klassifizierungsregeln:
- "simple": Begruessung, allgemeine Fragen ohne Bedarf an spezifischen Dokumenten
  → needs_retrieval = false, estimated_steps = 0
- "factual": Direkte Frage mit einer praezisen faktischen Antwort
  → needs_retrieval = true, estimated_steps = 1
- "complex": Vergleich, Multi-Aspekt-Analyse, erfordert mehrere Quellen
  → needs_retrieval = true, estimated_steps = 2-5"""

        return self.llm.invoke(prompt)

Ansatz 2: Trainierter Klassifikator (schneller)

Fuer die Produktion ist ein kleines trainiertes Modell schneller und kostenguenstiger.

DEVELOPERpython
from sentence_transformers import SentenceTransformer
from sklearn.ensemble import GradientBoostingClassifier
import numpy as np
import joblib

class TrainedQueryClassifier:
    """Trainierter Klassifikator fuer RAG-Routing."""

    def __init__(self, model_path: str = None):
        self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
        if model_path:
            self.classifier = joblib.load(model_path)
        else:
            self.classifier = GradientBoostingClassifier(
                n_estimators=100, max_depth=5
            )

    def train(self, queries: list, labels: list):
        """Trainiert den Klassifikator mit gelabelten Beispielen."""
        embeddings = self.encoder.encode(queries)

        features = []
        for q, emb in zip(queries, embeddings):
            extra = [
                len(q),
                q.count("?"),
                len(q.split()),
                int(any(w in q.lower() for w in
                    ["vergleiche", "versus", "unterschied", "vorteil"])),
                int(any(w in q.lower() for w in
                    ["dann", "auch", "zusaetzlich", "ausserdem"])),
            ]
            features.append(np.concatenate([emb, extra]))

        X = np.array(features)
        self.classifier.fit(X, labels)

    def predict(self, query: str) -> str:
        """Sagt die Kategorie einer Abfrage vorher."""
        emb = self.encoder.encode([query])[0]
        extra = [
            len(query),
            query.count("?"),
            len(query.split()),
            int(any(w in query.lower() for w in
                ["vergleiche", "versus", "unterschied", "vorteil"])),
            int(any(w in query.lower() for w in
                ["dann", "auch", "zusaetzlich", "ausserdem"])),
        ]
        X = np.array([np.concatenate([emb, extra])])
        return self.classifier.predict(X)[0]

Der Adaptive Router

Der Router verbindet den Klassifikator mit den verschiedenen Retrieval-Strategien.

DEVELOPERpython
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from typing import TypedDict, List, Optional

class AdaptiveRAGState(TypedDict):
    question: str
    classification: str
    sub_questions: Optional[List[str]]
    documents: List[Document]
    generation: str

# Route 1: Kein Retrieval (direktes LLM)
def direct_llm_response(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Antwortet direkt ohne Retrieval."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    response = llm.invoke(state["question"]).content
    return {**state, "generation": response}

# Route 2: Single-Step-Retrieval
async def single_step_retrieval(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Standard-Retrieval in einem Schritt."""
    docs = await retriever.aget_relevant_documents(state["question"])

    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    context = "\n\n".join([d.page_content for d in docs])

    response = llm.invoke(
        f"Kontext:\n{context}\n\nFrage: {state['question']}"
    ).content

    return {**state, "documents": docs, "generation": response}

# Route 3: Multi-Step-Retrieval
async def decompose_question(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Zerlegt eine komplexe Frage in Unterfragen."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    prompt = f"""Zerlege diese komplexe Frage in einfache,
unabhaengige Unterfragen, die nach Beantwortung die
Hauptfrage beantworten lassen.

Frage: {state["question"]}

Gib eine JSON-Liste von Unterfragen zurueck (max 4).
Beispiel: ["Unterfrage 1?", "Unterfrage 2?"]"""

    result = llm.invoke(prompt).content
    import json
    sub_questions = json.loads(result)

    return {**state, "sub_questions": sub_questions}

async def multi_step_retrieval(state: AdaptiveRAGState) -> AdaptiveRAGState:
    """Ruft Dokumente fuer jede Unterfrage ab."""
    all_docs = []

    for sub_q in state["sub_questions"]:
        docs = await retriever.aget_relevant_documents(sub_q)
        all_docs.extend(docs)

    # Deduplizierung
    seen = set()
    unique_docs = []
    for doc in all_docs:
        key = hash(doc.page_content[:100])
        if key not in seen:
            seen.add(key)
            unique_docs.append(doc)

    return {**state, "documents": unique_docs}

# Adaptiven Workflow aufbauen
def classify_and_route(state: AdaptiveRAGState) -> str:
    """Klassifiziert und routet zur richtigen Strategie."""
    classifier = QueryClassifier()
    result = classifier.classify(state["question"])

    if result.complexity == "simple":
        return "direct"
    elif result.complexity == "factual":
        return "single_step"
    else:
        return "decompose"

workflow = StateGraph(AdaptiveRAGState)

workflow.add_node("direct", direct_llm_response)
workflow.add_node("single_step", single_step_retrieval)
workflow.add_node("decompose", decompose_question)
workflow.add_node("multi_retrieve", multi_step_retrieval)
workflow.add_node("synthesize", synthesize_response)

workflow.set_conditional_entry_point(
    classify_and_route,
    {"direct": "direct", "single_step": "single_step", "decompose": "decompose"}
)

workflow.add_edge("direct", END)
workflow.add_edge("single_step", END)
workflow.add_edge("decompose", "multi_retrieve")
workflow.add_edge("multi_retrieve", "synthesize")
workflow.add_edge("synthesize", END)

adaptive_rag = workflow.compile()

Vergleich: Festes RAG vs Adaptives RAG

Benchmarks (Jeong et al., 2024)

Ueber die 6 im Paper evaluierten QA-Datensaetze (SQuAD, Natural Questions, TriviaQA, MuSiQue, HotpotQA, 2WikiMultiHopQA) mit FLAN-T5-XL (3B) ergeben sich folgende Durchschnittswerte:

Zentrale Erkenntnis: Adaptive-RAG erreicht nahezu die Genauigkeit des systematischen Multi-Step-Ansatzes (46,9 vs 48,9 F1), benoetigt dabei aber im Durchschnitt nur 1,08 Retrieval-Schritte statt 4,69 — was die Suchkosten um mehr als das Vierfache reduziert. Es uebertrifft Single-Step deutlich (44,3 F1) und verdoppelt nahezu die Genauigkeit von No-Retrieval (21,1 F1).

Detaillierter Entscheidungsbaum

Eingehende Abfrage
│
├── Enthaelt Begrüssungs-/Hoeflichkeitswoerter?
│   ├── Ja → Ist es NUR eine Begrüssung?
│   │   ├── Ja → Route: KEIN RETRIEVAL
│   │   └── Nein → Analyse fortsetzen ↓
│   └── Nein → Analyse fortsetzen ↓
│
├── Betrifft die Abfrage einen spezifischen Fakt?
│   ├── Ja → Nur ein Fakt angefragt?
│   │   ├── Ja → Route: SINGLE-STEP
│   │   └── Nein → Mehrere zusammenhaengende Fakten?
│   │       ├── Ja → Route: MULTI-STEP (2 Schritte)
│   │       └── Nein → Route: MULTI-STEP (3+ Schritte)
│   └── Nein → Analyse fortsetzen ↓
│
├── Fragt die Abfrage nach einem Vergleich?
│   ├── Ja → Route: MULTI-STEP (Zerlegung + Synthese)
│   └── Nein → Analyse fortsetzen ↓
│
├── Fragt die Abfrage nach Analyse oder Empfehlung?
│   ├── Ja → Route: MULTI-STEP (kontextuelle Analyse)
│   └── Nein → Route: SINGLE-STEP (Standard)
│
└── Fallback → Route: SINGLE-STEP

Erweiterte Optimierungen

Kombination mit CRAG

Adaptives RAG kann mit Corrective RAG fuer eine doppelte Qualitaetsschicht kombiniert werden.

DEVELOPERpython
async def adaptive_corrective_pipeline(question: str) -> str:
    """Pipeline, die adaptives und korrektives RAG kombiniert."""
    # Schritt 1: Klassifizierung
    classification = classifier.classify(question)

    if classification.complexity == "simple":
        return direct_llm(question)

    # Schritt 2: Adaptives Retrieval
    if classification.complexity == "factual":
        docs = await single_step_retrieve(question)
    else:
        sub_questions = decompose(question)
        docs = await multi_step_retrieve(sub_questions)

    # Schritt 3: Korrektive Bewertung (CRAG)
    relevance = evaluate_relevance(question, docs)

    if relevance == "incorrect":
        docs = await web_search_fallback(question)
    elif relevance == "ambiguous":
        web_docs = await web_search_fallback(question)
        docs = docs + web_docs

    # Schritt 4: Generierung
    return generate_response(question, docs)

Ueberwachungsmetriken

FAQ

Wie bestimmt man die Schwellenwerte des Klassifikators?

Beginnen Sie mit Standardschwellenwerten (einfach < 0,3, faktisch 0,3-0,7, komplex > 0,7) und passen Sie dann basierend auf Ihren Daten an. Analysieren Sie 500 Abfragen manuell, labeln Sie sie und messen Sie die Klassifikator-Genauigkeit. Passen Sie die Schwellenwerte an, um die kostspieligsten Fehler zu minimieren (eine komplexe Frage, die als "einfach" geroutet wird, ist schlimmer als umgekehrt).

Fuegt adaptives RAG Latenz hinzu?

Der Klassifikator selbst fuegt 100-300ms (LLM) oder 10-50ms (trainiertes Modell) hinzu. Dies wird durch Einsparungen bei einfachen Abfragen, die Retrieval vermeiden, mehr als ausgeglichen. Im Durchschnitt ist adaptives RAG schneller als festes RAG, da 20-35% der Abfragen ohne Retrieval verarbeitet werden.

Wie geht man mit Abfragen um, die zwischen zwei Kategorien mehrdeutig sind?

Zwei Strategien: (1) Im Zweifelsfall die reichhaltigere Route waehlen ("faktisch" statt "einfach", "komplex" statt "faktisch"), um unvollstaendige Antworten zu vermeiden; (2) einen Konfidenzschwellenwert implementieren und zur hoeheren Kategorie routen, wenn die Konfidenz unter 80% liegt.

Funktioniert adaptives RAG fuer Multi-Turn-Chatbots?

Ja, und es ist sogar noch relevanter. Im Multi-Turn-Chat sind Folgenachrichten ("Und fuer den Enterprise-Plan?") oft einfach und benoetigen nur eine Aenderung des vorherigen Kontexts, kein vollstaendig neues Retrieval. Der Klassifikator kann den Gespraechsverlauf beruecksichtigen, um das Routing zu optimieren.

Kann man mehr als drei Routen hinzufuegen?

Absolut. Haeufige zusaetzliche Routen umfassen: "SQL-Abfrage" (fuer strukturierte Datenabfragen), "API-Aufruf" (fuer Echtzeitaktionen), "menschliche Eskalation" (fuer sensible Themen). Das Prinzip bleibt gleich: Klassifizieren und dann zum optimalen Handler routen.

Fazit

Adaptives RAG ist eine wesentliche Optimierung fuer jedes RAG-System in der Produktion. Durch Anpassung der Retrieval-Strategie an die Komplexitaet jeder Abfrage bietet es gleichzeitig bessere Leistung und reduzierte Kosten.

Wichtige Erkenntnisse:

1. Klassifizieren vor dem Abrufen spart Zeit und Geld
2. Drei Basisrouten decken die Mehrheit der Anwendungsfaelle ab
3. Ein trainierter Klassifikator ist schneller als ein LLM fuer das Routing
4. Kombination mit CRAG bietet eine doppelte Qualitaetsschicht
5. Die Feedback-Schleife verbessert kontinuierlich die Routing-Genauigkeit

Ailog verwendet adaptives Routing, um jede Chatbot-Interaktion zu optimieren. Testen Sie es kostenlos und sehen Sie den Unterschied zwischen einem RAG, das denkt, und einem RAG, das blind ausfuehrt.

Ressourcen

• Adaptive RAG Paper - "Adaptive-RAG" (Jeong et al., 2024)
• LangGraph Dokumentation - Routing und Workflows
• Query Routing RAG - Ergaenzender Routing-Leitfaden
• Corrective RAG - Selbstkorrektur der Ergebnisse
• Retrieval-Grundlagen - Retrieval-Grundlagen