Corrective RAG: The Self-Correcting System for Perfect Answers

Naive RAG has a problem everyone knows but few solve: it generates answers even when retrieved documents don't contain the right information. Corrective RAG (CRAG) and Self-RAG introduce a self-evaluation layer that lets the system detect bad retrievals and correct itself automatically. The result: a 40 to 60% reduction in hallucinations.

TL;DR

• CRAG (Yan et al., 2024) evaluates retrieved document relevance and decides on strategy: use, supplement, or replace
• Self-RAG (Asai et al., 2024) goes further with reflection tokens to evaluate each step
• Three decisions: Correct (use docs), Ambiguous (supplement via web search), Incorrect (full fallback)
• -40-60% hallucinations compared to naive RAG
• Latency overhead: +1-3 seconds (relevance evaluation)
• Implementable with LangChain/LangGraph in under 100 lines

The Problem with Naive RAG

Naive RAG blindly trusts retrieved documents. If the top-k contains irrelevant documents, the LLM will still use them to generate an answer, often hallucinating in the process.

Naive RAG Pipeline:
┌──────────┐    ┌──────────────┐    ┌─────────────┐    ┌──────────┐
│ Question │───▶│  Retrieval   │───▶│  Documents  │───▶│   LLM    │
│          │    │  (top-k=5)   │    │ (relevant   │    │ Generates│
│          │    │              │    │   or not!)   │    │ anyway.. │
│          │    │              │    │              │    │          │
└──────────┘    └──────────────┘    └─────────────┘    └──────────┘
                                          ❌
                                  No relevance check!

Typical Failure Cases

Corrective RAG (CRAG) Architecture

The CRAG paper (Yan et al., 2024) introduces a lightweight relevance evaluator that acts as a filter between retrieval and generation.

The Decision Flow

┌──────────────────────────────────────────────────────────────────┐
│                    Corrective RAG Pipeline                        │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│  Question ──▶ Retrieval (top-k) ──▶ Relevance Evaluation         │
│                                           │                       │
│                          ┌────────────────┼────────────────┐      │
│                          │                │                │      │
│                          ▼                ▼                ▼      │
│                     ┌─────────┐    ┌───────────┐    ┌──────────┐ │
│                     │ CORRECT │    │ AMBIGUOUS │    │ INCORRECT│ │
│                     │ Score   │    │  Score    │    │  Score   │ │
│                     │ > 0.7   │    │ 0.3-0.7  │    │  < 0.3   │ │
│                     └────┬────┘    └─────┬─────┘    └────┬─────┘ │
│                          │               │               │       │
│                          ▼               ▼               ▼       │
│                     Use the        Supplement with   Full web     │
│                     retrieved      web search        search      │
│                     documents      + docs            (replace)   │
│                          │               │               │       │
│                          └───────────────┼───────────────┘       │
│                                          ▼                       │
│                                   Knowledge                      │
│                                   Refinement                     │
│                                          │                       │
│                                          ▼                       │
│                                     Generation                   │
│                                     (final LLM)                  │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

Implementation with 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. Relevance evaluator
def evaluate_relevance(state: CRAGState) -> CRAGState:
    """Evaluate the relevance of retrieved documents."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

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

    scores = []
    for doc in documents:
        prompt = f"""Evaluate the relevance of this document for answering the question.

Question: {question}
Document: {doc.page_content[:500]}

Relevance score (0.0 to 1.0):
- 1.0 = perfectly relevant, directly answers
- 0.5 = partially relevant
- 0.0 = not relevant at all

Reply with ONLY a decimal number."""

        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:
    """Web search when documents are insufficient."""
    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:
    """Refine and merge knowledge sources."""
    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", [])

    # Deduplication and scoring
    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. Final generation
def generate_response(state: CRAGState) -> CRAGState:
    """Generate the final response with refined documents."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

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

    prompt = f"""Answer the question based ONLY on the provided context.
If the context doesn't contain enough information, say so clearly.

Context:
{context}

Question: {state["question"]}

Answer:"""

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

# Build LangGraph workflow
def route_by_relevance(state: CRAGState) -> str:
    """Route to the right node based on the decision."""
    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: Self-Reflection Taken to the Extreme

Self-RAG (Asai et al., 2024) goes further than CRAG by integrating reflection tokens directly into the generation process. The model learns to evaluate every step of its own reasoning.

The Four Reflection Tokens

Self-RAG Flow

Question: "What are the advantages of RAG over fine-tuning?"

Step 1: [Retrieve] = yes → Launch retrieval
Step 2: Doc 1 retrieved → [IsRel] = relevant ✅
        Doc 2 retrieved → [IsRel] = irrelevant ❌ (filtered)
Step 3: Generation with Doc 1
        → [IsSup] = fully supported ✅
        → [IsUse] = 4/5 ✅
Step 4: Final answer validated

Self-RAG Implementation

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  # Your retriever here

    def should_retrieve(self, question: str) -> RetrieveDecision:
        """[Retrieve] token: should we search for documents?"""
        prompt = f"""Evaluate if this question requires document retrieval.

Question: {question}

Reply with ONE word only:
- "yes": the question requires specific facts
- "no": the question can be answered with general knowledge
- "continue": the current generation is sufficient"""

        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: is each document relevant?"""
        evaluated = []
        for doc in documents:
            prompt = f"""Is the following document relevant to the question?

Question: {question}
Document: {doc.page_content[:500]}

Reply "relevant" or "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: is the response supported?"""
        context = "\n".join([d.page_content for d in documents])
        prompt = f"""Is the response supported by the documents?

Documents: {context[:2000]}
Response: {response}

Reply with:
- "fully_supported": every claim is in the documents
- "partially_supported": some claims are supported
- "no_support": the response is not in the documents"""

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

    def rate_usefulness(self, question: str, response: str) -> int:
        """[IsUse] token: is the response useful?"""
        prompt = f"""Rate the usefulness of this response (1 to 5).

Question: {question}
Response: {response}

5 = Perfectly useful and complete
4 = Very useful but could be more complete
3 = Moderately useful
2 = Not very useful
1 = Useless

Reply with ONE number only."""

        return int(self.critic_llm.invoke(prompt).content.strip())

    async def query(self, question: str) -> str:
        """Complete Self-RAG pipeline."""
        # Step 1: Should we retrieve?
        retrieve_decision = self.should_retrieve(question)

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

        # Step 2: Retrieve and evaluate relevance
        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 "I couldn't find relevant information for this question."

        # Step 3: Generate with relevant documents
        context = "\n\n".join([d.page_content for d in relevant_docs])
        response = self.gen_llm.invoke(
            f"Context:\n{context}\n\nQuestion: {question}"
        ).content

        # Step 4: Check support and usefulness
        support = self.check_support(response, relevant_docs)
        usefulness = self.rate_usefulness(question, response)

        if support == SupportScore.NONE or usefulness < 3:
            # Regenerate with a stricter prompt
            response = self.gen_llm.invoke(
                f"""Answer STRICTLY from the context.
Make no assumptions. If the info is missing, say so.

Context:\n{context}\n\nQuestion: {question}"""
            ).content

        return response

Comparison of RAG Variants

Benchmarks on Standard Datasets

Practical Optimizations

1. Fast Relevance Evaluator

Instead of using an LLM to evaluate each document, train a small classifier.

DEVELOPERpython
from sentence_transformers import CrossEncoder

class FastRelevanceEvaluator:
    """Fast evaluator based on a 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]:
        """Evaluate all documents in a single 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:
        """Return relevant docs and the decision."""
        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. Relevance Decision Cache

DEVELOPERpython
import hashlib
from functools import lru_cache

class CachedCRAG:
    """CRAG with relevance evaluation caching."""

    def __init__(self):
        self.evaluator = FastRelevanceEvaluator()
        self.cache = {}

    def _cache_key(self, question: str, doc_content: str) -> str:
        combined = f"{question}||{doc_content[:200]}"
        return hashlib.md5(combined.encode()).hexdigest()

    def evaluate_with_cache(
        self, question: str, documents: List[Document]
    ) -> List[float]:
        scores = []
        uncached_indices = []
        uncached_docs = []

        for i, doc in enumerate(documents):
            key = self._cache_key(question, doc.page_content)
            if key in self.cache:
                scores.append(self.cache[key])
            else:
                scores.append(None)
                uncached_indices.append(i)
                uncached_docs.append(doc)

        if uncached_docs:
            new_scores = self.evaluator.evaluate_batch(
                question, uncached_docs
            )
            for idx, score in zip(uncached_indices, new_scores):
                key = self._cache_key(
                    question, documents[idx].page_content
                )
                self.cache[key] = score
                scores[idx] = score

        return scores

3. Domain-Adaptive Thresholds

DEVELOPERpython
DOMAIN_THRESHOLDS = {
    "medical": {
        "correct_threshold": 0.85,   # Stricter
        "ambiguous_threshold": 0.5,
        "min_relevant_docs": 2       # Requires 2+ concordant docs
    },
    "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,    # More tolerant
        "ambiguous_threshold": 0.25,
        "min_relevant_docs": 1
    }
}

Evaluation Metrics

To measure the effectiveness of your CRAG implementation, track these metrics.

FAQ

Is CRAG compatible with all retrievers?

Yes. CRAG is independent of the retriever used upstream. It works with pure vector search, hybrid search, BM25, or even a knowledge graph-based retriever. The relevance evaluation is performed on the returned documents, regardless of the retrieval method.

How do you handle the case where web search returns nothing relevant either?

In this case, the system should honestly acknowledge that it doesn't have the information. Implement a second evaluator on web results. If web results are also irrelevant, return an explicit message: "I couldn't find reliable information to answer this question." This is always better than a hallucination.

What is the additional cost of CRAG compared to naive RAG?

Relevance evaluation with a cross-encoder is nearly free (< 50ms). If you use an LLM for evaluation, expect about 500 additional tokens per evaluated document, approximately $0.002-0.005 per query with GPT-4o-mini. The web search fallback via Tavily costs about $0.005 per search. In total, CRAG adds $0.005-0.01 per query.

Does Self-RAG require a fine-tuned model?

The original Self-RAG paper uses a fine-tuned Llama model with reflection tokens. In practice, you can simulate the behavior with structured prompts on GPT-4o or Claude, as shown in the code examples above. Performance is slightly lower but remains significantly better than naive RAG.

Does CRAG work in real-time for a chatbot?

Yes, with optimizations. Use a fast cross-encoder (MiniLM) rather than an LLM for relevance evaluation. Pre-compute thresholds for your domain. Implement evaluation caching. With these optimizations, CRAG adds 500ms to 1.5s of latency, which remains acceptable for a chatbot.

Conclusion

Corrective RAG and Self-RAG represent a major evolution in RAG system reliability. By adding a self-evaluation layer, these approaches transform RAG from a "fire and forget" pipeline into an intelligent system capable of questioning its own results.

Key takeaways:

1. CRAG adds relevance evaluation between retrieval and generation
2. Self-RAG integrates reflection directly into the generation process
3. Web search fallback is essential for out-of-scope questions
4. A fast cross-encoder enables real-time evaluation
5. Domain-adaptive thresholds optimize the precision/recall balance

Ailog integrates self-correction mechanisms inspired by CRAG to guarantee the reliability of its chatbot responses. Create your free account to discover a RAG that doesn't just answer but verifies before speaking.

Resources

• 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 Documentation - Framework for RAG workflows
• Retrieval Fundamentals Guide - Retrieval basics
• Hallucination Detection - Complementary to CRAG
• Agentic RAG - The next step: autonomous agents