Corrective RAG (CRAG)

What is Corrective RAG?

Corrective RAG (CRAG) is a framework that adds a retrieval evaluation step to the RAG pipeline. Before generating an answer, CRAG evaluates whether the retrieved documents are relevant enough to answer the question. Based on this evaluation, it takes one of three actions: 1. Correct — Retrieved documents are relevant → proceed to generate 2. Incorrect — Retrieved documents are irrelevant → try alternative retrieval (e.g., web search) or refine the query 3. Ambiguous — Some documents are relevant, some aren't → filter to keep only relevant ones, optionally supplement with additional retrieval Example Scenario: A support assistant retrieves troubleshooting documents for a customer's question about a specific error. The evaluator determines that the retrieved documents discuss a different error entirely. Instead of generating a misleading answer, CRAG triggers an alternative search strategy to find more relevant documents.

Retrieval Quality Evaluation

The core innovation in CRAG is the retrieval evaluator — a component that assesses whether retrieved documents contain information relevant to the query. Evaluation Approaches: LLM-Based Evaluation: Use a language model to score each retrieved document's relevance to the query. The LLM reads the query and document and outputs a relevance judgment (e.g., "relevant", "partially relevant", or "irrelevant"). Threshold-Based: Set a minimum similarity score threshold. Documents below the threshold are considered irrelevant. Cross-Encoder Scoring: Use a cross-encoder model to compute precise relevance scores for each query-document pair. The evaluation result determines the corrective action: • All documents relevant → proceed with generation (Correct) • No documents relevant → trigger alternative retrieval (Incorrect) • Mixed relevance → filter to relevant documents only, optionally search for more (Ambiguous)

retrieval_evaluation.pypython

from langchain_openai import ChatOpenAI

evaluator_llm = ChatOpenAI(model="gpt-4", temperature=0)

def evaluate_document_relevance(question: str, document: str) -> str:
    """Evaluate if a document is relevant to the question.
    Returns: 'relevant', 'partially_relevant', or 'irrelevant'
    """
    response = evaluator_llm.invoke(
        f"""Evaluate whether the following document contains information 
        relevant to answering the question.

        Question: {question}
        Document: {document}

        Respond with exactly one word: 'relevant', 'partially_relevant', or 'irrelevant'."""
    )
    return response.content.strip().lower()

def grade_retrieval(question: str, documents: list) -> str:
    """Grade the overall retrieval quality."""
    grades = [evaluate_document_relevance(question, doc.page_content) 
              for doc in documents]
    
    relevant_count = grades.count('relevant') + grades.count('partially_relevant')
    
    if relevant_count == len(documents):
        return "correct"
    elif relevant_count == 0:
        return "incorrect"
    else:
        return "ambiguous"

Corrective Actions

Based on the evaluation result, CRAG takes different corrective actions: When Correct (All Relevant): • Proceed directly to answer generation • Use all retrieved documents as context • This is the same as basic RAG for this particular query When Incorrect (None Relevant): • The internal knowledge base doesn't contain the needed information • Option 1: Query an alternative source (e.g., web search) • Option 2: Rewrite the query and retry internal retrieval • Option 3: Acknowledge that the question cannot be answered from available sources • Note: External search is optional and should only be used when configured When Ambiguous (Mixed Relevance): • Filter out irrelevant documents • Optionally supplement with additional retrieval using a refined query • Proceed with only the relevant subset Key Distinction from Agentic RAG: While CRAG and Agentic RAG both evaluate retrieval quality, CRAG specifically formalizes the three-way classification (correct/incorrect/ambiguous) and defines structured corrective actions for each case. Agentic RAG is a broader concept where an agent has general decision-making capabilities.

External Search is Optional

The original CRAG paper proposes web search as a fallback, but this is not always appropriate. For enterprise systems handling sensitive data, external search may be disabled. The corrective mechanism should be configurable.

CRAG Implementation

Here is a complete CRAG implementation using LangGraph that demonstrates the corrective workflow:

corrective_rag.pypython

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
from langchain_core.documents import Document

class CRAGState(TypedDict):
    question: str
    documents: List[Document]
    filtered_documents: List[Document]
    web_results: List[Document]
    grade: str
    generation: str

def retrieve(state: CRAGState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"documents": docs}

def evaluate(state: CRAGState) -> dict:
    grade = grade_retrieval(state["question"], state["documents"])
    return {"grade": grade}

def route_by_grade(state: CRAGState) -> str:
    if state["grade"] == "correct":
        return "generate"
    elif state["grade"] == "incorrect":
        return "web_search"
    else:  # ambiguous
        return "filter_and_supplement"

def filter_documents(state: CRAGState) -> dict:
    filtered = [doc for doc in state["documents"]
                if evaluate_document_relevance(state["question"], doc.page_content) 
                in ("relevant", "partially_relevant")]
    return {"filtered_documents": filtered}

def web_search(state: CRAGState) -> dict:
    # Optional: search the web for additional context
    # Only if external search is configured
    web_docs = web_search_tool.invoke(state["question"])
    return {"web_results": web_docs}

def generate(state: CRAGState) -> dict:
    # Use filtered docs if available, otherwise original docs
    docs = state.get("filtered_documents") or state["documents"]
    web = state.get("web_results", [])
    all_docs = docs + web
    
    context = "\n\n".join(d.page_content for d in all_docs)
    answer = llm.invoke(
        f"Context: {context}\n\nQuestion: {state['question']}\n\nAnswer:"
    ).content
    return {"generation": answer}

# Build CRAG workflow
workflow = StateGraph(CRAGState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("evaluate", evaluate)
workflow.add_node("filter_and_supplement", filter_documents)
workflow.add_node("web_search", web_search)
workflow.add_node("generate", generate)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "evaluate")
workflow.add_conditional_edges("evaluate", route_by_grade, {
    "generate": "generate",
    "web_search": "web_search",
    "filter_and_supplement": "filter_and_supplement",
})
workflow.add_edge("filter_and_supplement", "generate")
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

crag_app = workflow.compile()

Learning Objectives

What is Corrective RAG?

Retrieval Quality Evaluation

Corrective Actions

CRAG Implementation

Key Takeaways