Corrective RAG: Das selbstkorrigierende System für perfekte Antworten

Naives RAG hat ein Problem, das jeder kennt, aber wenige loesen: Es generiert Antworten, selbst wenn die abgerufenen Dokumente nicht die richtigen Informationen enthalten. Corrective RAG (CRAG) und Self-RAG fuehren eine Selbstbewertungsschicht ein, die es dem System ermoeglicht, schlechte Abrufe zu erkennen und sich automatisch zu korrigieren. Das Ergebnis: eine Reduktion der Halluzinationen um 40 bis 60%.

TL;DR

• CRAG (Yan et al., 2024) bewertet die Relevanz abgerufener Dokumente und entscheidet ueber die Strategie: verwenden, ergaenzen oder ersetzen
• Self-RAG (Asai et al., 2024) geht weiter mit Reflexions-Tokens zur Bewertung jedes Schritts
• Drei Entscheidungen: Korrekt (Dokumente verwenden), Mehrdeutig (per Websuche ergaenzen), Falsch (vollstaendiger Fallback)
• -40-60% Halluzinationen im Vergleich zu naivem RAG
• Latenz-Overhead: +1-3 Sekunden (Relevanzbewertung)
• Implementierbar mit LangChain/LangGraph in unter 100 Zeilen

Das Problem mit naivem RAG

Naives RAG vertraut blind den abgerufenen Dokumenten. Wenn die Top-k irrelevante Dokumente enthalten, verwendet das LLM sie trotzdem zur Antwortgenerierung, wobei es oft halluziniert.

Naive RAG Pipeline:
┌──────────┐    ┌──────────────┐    ┌─────────────┐    ┌──────────┐
│  Frage   │───▶│  Retrieval   │───▶│  Dokumente  │───▶│   LLM    │
│          │    │  (top-k=5)   │    │ (relevant   │    │ Generiert│
│          │    │              │    │  oder nicht!)│    │ trotzdem │
│          │    │              │    │              │    │          │
└──────────┘    └──────────────┘    └─────────────┘    └──────────┘
                                          ❌
                                 Keine Relevanzpruefung!

Typische Fehlerfaelle

Corrective RAG (CRAG) Architektur

Das CRAG-Paper (Yan et al., 2024) fuehrt einen leichtgewichtigen Relevanzbewerter ein, der als Filter zwischen Retrieval und Generierung fungiert.

Der Entscheidungsfluss

┌──────────────────────────────────────────────────────────────────┐
│                    Corrective RAG Pipeline                        │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│  Frage ──▶ Retrieval (top-k) ──▶ Relevanzbewertung              │
│                                           │                       │
│                          ┌────────────────┼────────────────┐      │
│                          │                │                │      │
│                          ▼                ▼                ▼      │
│                     ┌─────────┐    ┌───────────┐    ┌──────────┐ │
│                     │ KORREKT │    │ MEHRDEUTIG│    │ FALSCH   │ │
│                     │ Score   │    │  Score    │    │  Score   │ │
│                     │ > 0.7   │    │ 0.3-0.7  │    │  < 0.3   │ │
│                     └────┬────┘    └─────┬─────┘    └────┬─────┘ │
│                          │               │               │       │
│                          ▼               ▼               ▼       │
│                     Abgerufene     Ergaenzen durch  Vollstaendige│
│                     Dokumente      Websuche         Websuche     │
│                     verwenden      + Dokumente      (ersetzen)   │
│                          │               │               │       │
│                          └───────────────┼───────────────┘       │
│                                          ▼                       │
│                                   Knowledge                      │
│                                   Refinement                     │
│                                          │                       │
│                                          ▼                       │
│                                     Generierung                  │
│                                     (finales LLM)                │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

Implementierung mit LangGraph

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

class CRAGState(TypedDict):
    question: str
    documents: List[Document]
    relevance_decision: Literal["correct", "ambiguous", "incorrect"]
    web_results: List[Document]
    refined_documents: List[Document]
    generation: str

# 1. Relevanzbewerter
def evaluate_relevance(state: CRAGState) -> CRAGState:
    """Bewertet die Relevanz der abgerufenen Dokumente."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    question = state["question"]
    documents = state["documents"]

    scores = []
    for doc in documents:
        prompt = f"""Bewerte die Relevanz dieses Dokuments fuer die Beantwortung der Frage.

Frage: {question}
Dokument: {doc.page_content[:500]}

Relevanz-Score (0.0 bis 1.0):
- 1.0 = perfekt relevant, beantwortet direkt
- 0.5 = teilweise relevant
- 0.0 = ueberhaupt nicht relevant

Antworte NUR mit einer Dezimalzahl."""

        score = float(llm.invoke(prompt).content.strip())
        scores.append(score)

    avg_score = sum(scores) / len(scores) if scores else 0

    if avg_score > 0.7:
        decision = "correct"
    elif avg_score > 0.3:
        decision = "ambiguous"
    else:
        decision = "incorrect"

    return {
        **state,
        "relevance_decision": decision
    }

# 2. Web-Search-Fallback
async def web_search_fallback(state: CRAGState) -> CRAGState:
    """Websuche wenn Dokumente unzureichend sind."""
    from langchain_community.tools import TavilySearchResults

    search = TavilySearchResults(max_results=5)
    results = await search.ainvoke(state["question"])

    web_docs = [
        Document(
            page_content=r["content"],
            metadata={"source": r["url"], "type": "web"}
        )
        for r in results
    ]

    return {**state, "web_results": web_docs}

# 3. Knowledge Refinement
def refine_knowledge(state: CRAGState) -> CRAGState:
    """Verfeinert und fuehrt Wissensquellen zusammen."""
    decision = state["relevance_decision"]

    if decision == "correct":
        refined = state["documents"]
    elif decision == "ambiguous":
        refined = state["documents"] + state.get("web_results", [])
    else:  # incorrect
        refined = state.get("web_results", [])

    # Deduplizierung und Bewertung
    seen = set()
    unique_docs = []
    for doc in refined:
        content_hash = hash(doc.page_content[:100])
        if content_hash not in seen:
            seen.add(content_hash)
            unique_docs.append(doc)

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

# 4. Finale Generierung
def generate_response(state: CRAGState) -> CRAGState:
    """Generiert die finale Antwort mit verfeinerten Dokumenten."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

    context = "\n\n".join([
        doc.page_content for doc in state["refined_documents"]
    ])

    prompt = f"""Beantworte die Frage NUR basierend auf dem bereitgestellten Kontext.
Wenn der Kontext nicht genuegend Informationen enthaelt, sage das deutlich.

Kontext:
{context}

Frage: {state["question"]}

Antwort:"""

    response = llm.invoke(prompt)
    return {**state, "generation": response.content}

# LangGraph-Workflow aufbauen
def route_by_relevance(state: CRAGState) -> str:
    """Leitet zum richtigen Knoten basierend auf der Entscheidung."""
    if state["relevance_decision"] == "correct":
        return "refine"
    else:
        return "web_search"

workflow = StateGraph(CRAGState)

workflow.add_node("evaluate", evaluate_relevance)
workflow.add_node("web_search", web_search_fallback)
workflow.add_node("refine", refine_knowledge)
workflow.add_node("generate", generate_response)

workflow.set_entry_point("evaluate")
workflow.add_conditional_edges(
    "evaluate",
    route_by_relevance,
    {"refine": "refine", "web_search": "web_search"}
)
workflow.add_edge("web_search", "refine")
workflow.add_edge("refine", "generate")
workflow.add_edge("generate", END)

crag_chain = workflow.compile()

Self-RAG: Selbstreflexion auf die Spitze getrieben

Self-RAG (Asai et al., 2024) geht weiter als CRAG, indem es Reflexions-Tokens direkt in den Generierungsprozess integriert. Das Modell lernt, jeden Schritt seiner eigenen Argumentation zu bewerten.

Die vier Reflexions-Tokens

Self-RAG-Fluss

Frage: "Was sind die Vorteile von RAG gegenueber Fine-Tuning?"

Schritt 1: [Retrieve] = yes → Retrieval starten
Schritt 2: Dok 1 abgerufen → [IsRel] = relevant ✅
           Dok 2 abgerufen → [IsRel] = irrelevant ❌ (gefiltert)
Schritt 3: Generierung mit Dok 1
           → [IsSup] = fully supported ✅
           → [IsUse] = 4/5 ✅
Schritt 4: Finale Antwort validiert

Self-RAG Implementierung

DEVELOPERpython
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from typing import List, Tuple
from dataclasses import dataclass
from enum import Enum

class RetrieveDecision(Enum):
    YES = "yes"
    NO = "no"
    CONTINUE = "continue"

class RelevanceScore(Enum):
    RELEVANT = "relevant"
    IRRELEVANT = "irrelevant"

class SupportScore(Enum):
    FULLY = "fully_supported"
    PARTIALLY = "partially_supported"
    NONE = "no_support"

@dataclass
class ReflectionResult:
    retrieve: RetrieveDecision
    relevance: dict  # doc_id -> RelevanceScore
    support: SupportScore
    usefulness: int  # 1-5

class SelfRAG:
    def __init__(self, llm_model: str = "gpt-4o"):
        self.critic_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.gen_llm = ChatOpenAI(model=llm_model, temperature=0.1)
        self.retriever = None  # Ihr Retriever hier

    def should_retrieve(self, question: str) -> RetrieveDecision:
        """[Retrieve]-Token: Sollen Dokumente gesucht werden?"""
        prompt = f"""Bewerte ob diese Frage einen Dokumentenabruf erfordert.

Frage: {question}

Antworte mit NUR einem Wort:
- "yes": Die Frage erfordert spezifische Fakten
- "no": Die Frage kann mit allgemeinem Wissen beantwortet werden
- "continue": Die aktuelle Generierung ist ausreichend"""

        result = self.critic_llm.invoke(prompt).content.strip().lower()
        return RetrieveDecision(result)

    def evaluate_relevance(
        self, question: str, documents: List[Document]
    ) -> List[Tuple[Document, RelevanceScore]]:
        """[IsRel]-Token: Ist jedes Dokument relevant?"""
        evaluated = []
        for doc in documents:
            prompt = f"""Ist das folgende Dokument fuer die Frage relevant?

Frage: {question}
Dokument: {doc.page_content[:500]}

Antworte "relevant" oder "irrelevant"."""

            score = self.critic_llm.invoke(prompt).content.strip().lower()
            evaluated.append((
                doc,
                RelevanceScore(score)
            ))
        return evaluated

    def check_support(
        self, response: str, documents: List[Document]
    ) -> SupportScore:
        """[IsSup]-Token: Wird die Antwort gestuetzt?"""
        context = "\n".join([d.page_content for d in documents])
        prompt = f"""Wird die Antwort durch die Dokumente gestuetzt?

Dokumente: {context[:2000]}
Antwort: {response}

Antworte mit:
- "fully_supported": Jede Behauptung ist in den Dokumenten
- "partially_supported": Einige Behauptungen werden gestuetzt
- "no_support": Die Antwort ist nicht in den Dokumenten"""

        result = self.critic_llm.invoke(prompt).content.strip().lower()
        return SupportScore(result)

    async def query(self, question: str) -> str:
        """Vollstaendige Self-RAG-Pipeline."""
        # Schritt 1: Soll abgerufen werden?
        retrieve_decision = self.should_retrieve(question)

        if retrieve_decision == RetrieveDecision.NO:
            return self.gen_llm.invoke(question).content

        # Schritt 2: Abrufen und Relevanz bewerten
        documents = await self.retriever.aget_relevant_documents(question)
        evaluated = self.evaluate_relevance(question, documents)

        relevant_docs = [
            doc for doc, score in evaluated
            if score == RelevanceScore.RELEVANT
        ]

        if not relevant_docs:
            return "Ich konnte keine relevanten Informationen finden."

        # Schritt 3: Mit relevanten Dokumenten generieren
        context = "\n\n".join([d.page_content for d in relevant_docs])
        response = self.gen_llm.invoke(
            f"Kontext:\n{context}\n\nFrage: {question}"
        ).content

        # Schritt 4: Stuetzung und Nuetzlichkeit pruefen
        support = self.check_support(response, relevant_docs)

        if support == SupportScore.NONE:
            response = self.gen_llm.invoke(
                f"""Beantworte STRIKT basierend auf dem Kontext.
Mache keine Annahmen. Wenn Informationen fehlen, sage das.

Kontext:\n{context}\n\nFrage: {question}"""
            ).content

        return response

Vergleich der RAG-Varianten

Benchmarks auf Standard-Datensaetzen

Praktische Optimierungen

1. Schneller Relevanzbewerter

Anstatt ein LLM zur Bewertung jedes Dokuments zu verwenden, trainieren Sie einen kleinen Klassifikator.

DEVELOPERpython
from sentence_transformers import CrossEncoder

class FastRelevanceEvaluator:
    """Schneller Bewerter basierend auf einem Cross-Encoder."""

    def __init__(self, model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model = CrossEncoder(model_name)

    def evaluate_batch(
        self, question: str, documents: List[Document]
    ) -> List[float]:
        """Bewertet alle Dokumente in einem einzigen Batch."""
        pairs = [(question, doc.page_content) for doc in documents]
        scores = self.model.predict(pairs)
        return scores.tolist()

    def filter_relevant(
        self,
        question: str,
        documents: List[Document],
        threshold: float = 0.5
    ) -> tuple:
        """Gibt relevante Dokumente und die Entscheidung zurueck."""
        scores = self.evaluate_batch(question, documents)

        relevant = [
            doc for doc, score in zip(documents, scores)
            if score > threshold
        ]

        avg = sum(scores) / len(scores) if scores else 0
        if avg > 0.7:
            decision = "correct"
        elif avg > 0.3:
            decision = "ambiguous"
        else:
            decision = "incorrect"

        return relevant, decision, scores

2. Domaenenadaptive Schwellenwerte

DEVELOPERpython
DOMAIN_THRESHOLDS = {
    "medical": {
        "correct_threshold": 0.85,   # Strenger
        "ambiguous_threshold": 0.5,
        "min_relevant_docs": 2       # Erfordert 2+ uebereinstimmende Dokumente
    },
    "legal": {
        "correct_threshold": 0.80,
        "ambiguous_threshold": 0.45,
        "min_relevant_docs": 2
    },
    "general": {
        "correct_threshold": 0.7,
        "ambiguous_threshold": 0.3,
        "min_relevant_docs": 1
    },
    "ecommerce": {
        "correct_threshold": 0.6,    # Toleranter
        "ambiguous_threshold": 0.25,
        "min_relevant_docs": 1
    }
}

Bewertungsmetriken

Um die Wirksamkeit Ihrer CRAG-Implementierung zu messen, verfolgen Sie diese Metriken.

FAQ

Ist CRAG mit allen Retrievern kompatibel?

Ja. CRAG ist unabhaengig vom verwendeten Retriever. Es funktioniert mit reiner Vektorsuche, hybrider Suche, BM25 oder sogar einem wissensbasierten Graph-Retriever. Die Relevanzbewertung wird auf den zurueckgegebenen Dokumenten durchgefuehrt, unabhaengig von der Abrufmethode.

Wie geht man mit dem Fall um, dass die Websuche auch nichts Relevantes liefert?

In diesem Fall sollte das System ehrlich zugeben, dass es die Information nicht hat. Implementieren Sie einen zweiten Bewerter fuer die Webergebnisse. Wenn die Webergebnisse ebenfalls irrelevant sind, geben Sie eine explizite Nachricht zurueck: "Ich konnte keine zuverlaessigen Informationen finden, um diese Frage zu beantworten." Das ist immer besser als eine Halluzination.

Was sind die zusaetzlichen Kosten von CRAG im Vergleich zu naivem RAG?

Die Relevanzbewertung mit einem Cross-Encoder ist nahezu kostenlos (< 50ms). Wenn Sie ein LLM zur Bewertung verwenden, rechnen Sie mit etwa 500 zusaetzlichen Tokens pro bewertetem Dokument, ungefaehr $0,002-0,005 pro Abfrage mit GPT-4o-mini. Der Web-Search-Fallback ueber Tavily kostet etwa $0,005 pro Suche. Insgesamt fuegt CRAG $0,005-0,01 pro Abfrage hinzu.

Erfordert Self-RAG ein feinabgestimmtes Modell?

Das urspruengliche Self-RAG-Paper verwendet ein feinabgestimmtes Llama-Modell mit Reflexions-Tokens. In der Praxis koennen Sie das Verhalten mit strukturierten Prompts auf GPT-4o oder Claude simulieren, wie in den obigen Codebeispielen gezeigt. Die Leistung ist etwas geringer, bleibt aber deutlich besser als bei naivem RAG.

Funktioniert CRAG in Echtzeit fuer einen Chatbot?

Ja, mit Optimierungen. Verwenden Sie einen schnellen Cross-Encoder (MiniLM) anstelle eines LLM fuer die Relevanzbewertung. Berechnen Sie Schwellenwerte fuer Ihre Domaene vor. Implementieren Sie Bewertungs-Caching. Mit diesen Optimierungen fuegt CRAG 500ms bis 1,5s Latenz hinzu, was fuer einen Chatbot akzeptabel bleibt.

Fazit

Corrective RAG und Self-RAG stellen eine bedeutende Weiterentwicklung der Zuverlaessigkeit von RAG-Systemen dar. Durch Hinzufuegen einer Selbstbewertungsschicht verwandeln diese Ansaetze RAG von einer "Fire-and-Forget"-Pipeline in ein intelligentes System, das seine eigenen Ergebnisse hinterfragen kann.

Wichtige Erkenntnisse:

1. CRAG fuegt eine Relevanzbewertung zwischen Retrieval und Generierung hinzu
2. Self-RAG integriert die Reflexion direkt in den Generierungsprozess
3. Der Web-Search-Fallback ist essentiell fuer Fragen ausserhalb des Bereichs
4. Ein schneller Cross-Encoder ermoeglicht Echtzeit-Bewertung
5. Domaenenadaptive Schwellenwerte optimieren das Praezisions-/Recall-Verhaeltnis

Ailog integriert Selbstkorrekturmechanismen, die von CRAG inspiriert sind, um die Zuverlaessigkeit seiner Chatbot-Antworten zu garantieren. Erstellen Sie Ihr kostenloses Konto, um ein RAG zu entdecken, das nicht nur antwortet, sondern vor dem Sprechen prueft.

Ressourcen

• CRAG Paper - "Corrective Retrieval Augmented Generation" (Yan et al., 2024)
• Self-RAG Paper - "Self-RAG: Learning to Retrieve, Generate, and Critique" (Asai et al., 2024)
• LangGraph Dokumentation - Framework fuer RAG-Workflows
• Retrieval-Grundlagen - Retrieval-Grundlagen
• Halluzinationserkennung - Ergaenzend zu CRAG
• Agentic RAG - Der naechste Schritt: autonome Agenten