Advanced RAG: Corrective RAG (CRAG) with LangGraph

Retrieval-Augmented Generation (RAG) is designed to mitigate hallucinations by grounding Large Language Models (LLMs) in external or proprietary data. However, standard RAG architectures operate under an unchecked assumption: they blindly trust that the vector database always returns relevant and sufficient context.

When internal document retrieval yields low-quality, out-of-domain, or irrelevant chunks, traditional RAG models still force the LLM to generate an answer. This leads directly to hallucinations or catastrophic failures in enterprise environments. Corrective RAG (CRAG) introduces a self-correcting evaluation layer, granular sentence-level knowledge refinement, and adaptive web search fallbacks to ensure document quality before text generation occurs.

1. The Core Limitation of Traditional RAG

The Standard RAG Workflow

In a standard RAG pipeline, the system progresses through three discrete phases:

The "Blind Faith" Failure Mode

A vector database performs geometric proximity matching (e.g., Cosine Similarity, Euclidean Distance). It calculates distances between vector embeddings and must return the $k$ nearest chunks, regardless of whether those chunks actually answer the prompt.

Real-World Business Analogy: The HR Policy Dilemma

The Enterprise Risk: Imagine an employee querying an internal HR bot: "What is the parental leave policy for contracted staff across Europe?"

If the internal database only contains leave policies for full-time US employees, standard RAG will still pull US full-time documents (the nearest vectors) and present them to the LLM. The LLM then confidently provides US-specific full-time guidelines to an EU contractor. The employee acts on false assurances, leading to legal and compliance disputes.

2. Corrective RAG (CRAG) Mental Model & Architecture

CRAG addresses this architectural flaw by decoupling retrieval from generation through a Retrieval Evaluator. Instead of immediately routing retrieved chunks into the prompt context, CRAG inspects, scores, and branches the execution path.

The Three Confidence States

The Retrieval Evaluator grades chunks against the user query, calculating an evaluation score $S \in [0, 1]$ for each chunk. The overall retrieval state is classified using an Upper Threshold ($\tau_U$) and a Lower Threshold ($\tau_L$):

1. CORRECT ($\exists \, S_i \ge \tau_U$): At least one retrieved chunk strongly answers the query. The system isolates the relevant chunks, filters out extraneous sentences, and generates an answer using internal knowledge.
2. INCORRECT ($\forall \, S_i < \tau_L$): Not a single retrieved chunk clears the minimum relevance bar. The internal documents are discarded, the query is rewritten for public search engines, and real-time external search results supply the context.
3. AMBIGUOUS (Fallback/Middle Ground): Chunks provide partial context ($S_i \ge \tau_L$) but lack sufficient clarity or depth to clear $\tau_U$. The system refines the usable internal chunks, runs an external search for missing components, and merges both contexts.

3. Comparative Architectural Overview

4. Feature 1: Granular Knowledge Refinement

Fixed-size chunking strategies (e.g., recursive character splitting at 900 characters with 150-character overlaps) inherently capture multiple distinct topics within a single chunk. Even when a chunk contains the correct answer, it often contains sentences that discuss unrelated topics.

Passing raw chunks into generation pollutes the context window, distracts attention mechanisms, and degrades response clarity. CRAG resolves this with a three-step Decompose $\rightarrow$ Filter $\rightarrow$ Recombine sequence.

Note on Paper vs. Production Implementation: The original CRAG research paper fine-tuned an ultra-lightweight Google T5-large (770M parameters) transformer model to perform sentence-level classification. Because this specific checkpoint is proprietary/unreleased, modern implementations use lightweight instruction-tuned LLMs (e.g., gpt-4o-mini, Claude 3.5 Haiku) paired with structured outputs.

5. Feature 2: Retrieval Evaluation & Dynamic Routing

The retrieval evaluation stage prevents bad documents from leaking downstream. It processes every retrieved chunk through an LLM configured with a strict scoring prompt and Pydantic validation:

Mathematical Routing Invariants

• Let $D = \{d_1, d_2, \dots, d_n\}$ be the set of retrieved chunks.
• Let $S(d_i) \in [0.0, 1.0]$ be the relevance score assigned to $d_i$.
• Let $D_{good} = \{d_i \in D \mid S(d_i) \ge \tau_L\}$.

Critical Context Filtering Rule

Even when the verdict is declared CORRECT, the system does not forward all retrieved documents $D$ to the refinement stage. Only the subset $D_{good}$ ($S_i \ge \tau_L$) is passed forward. Chunks falling below $\tau_L$ are dropped immediately.

6. Feature 3: Query Rewriting & External Search Fallback

Direct conversational queries sent by users are poorly formatted for search engine indexing APIs. Consider the differences:

• Raw User Input: "What happened in AI last month?"
• Search Engine Limitations: Unclear temporal anchors, lacking technical keywords, vague phrasing.
• Rewritten Search Query: "AI developments breakthroughs January 2026 news summary"

When the evaluator triggers an external retrieval branch (INCORRECT or AMBIGUOUS), the query must be rewritten using an instruction-tuned LLM before calling a real-time web search tool (such as the Tavily Search API).

7. Step-by-Step Evolution to Full CRAG

We implement the complete CRAG system incrementally across four foundational phases:

8. Complete LangGraph Implementation

Below is the complete, self-contained implementation of Corrective RAG built on LangGraph.

Setup and Environment Configuration

import os
import re
from typing import List, Literal, TypedDict
from pydantic import BaseModel, Field

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, START, END

# Configure environment variables
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
os.environ["TAVILY_API_KEY"] = "your-tavily-api-key"

Graph State Architecture

class GraphState(TypedDict):
    question: str
    documents: List[Document]
    good_docs: List[Document]
    web_docs: List[Document]
    web_query: str
    verdict: Literal["CORRECT", "INCORRECT", "AMBIGUOUS"]
    reason: str
    strips: List[str]
    kept_strips: List[str]
    refined_context: str
    answer: str

Ingestion and Retriever Pipeline

# 1. Document Ingestion
loader = PyPDFLoader("data/hands_on_machine_learning.pdf")
raw_docs = loader.load()

# 2. Text Splitting with Unicode Cleanup
text_splitter = RecursiveCharacterTextSplitter(chunk_size=900, chunk_overlap=150)
docs = text_splitter.split_documents(raw_docs)

for doc in docs:
    doc.page_content = re.sub(r'[^\x00-\x7F]+', ' ', doc.page_content)

# 3. Vector Database & Base Retriever
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# LLM Backbones
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

Schema Definitions & System Prompts

# Evaluator Structured Output Schema
class ChunkEvaluation(BaseModel):
    score: float = Field(
        description="Relevance score between 0.0 and 1.0 indicating if the document answers the query."
    )
    reason: str = Field(description="Concise rationale for the assigned score.")

# Sentence Filter Schema
class SentenceFilter(BaseModel):
    keep: bool = Field(description="True if the sentence directly helps answer the query, else False.")

# Query Rewriter Schema
class RewrittenQuery(BaseModel):
    query: str = Field(description="Engine-optimized keyword search query.")

Graph Nodes

1. Retrieval Node

def retrieve_node(state: GraphState) -> dict:
    query = state["question"]
    retrieved_documents = retriever.invoke(query)
    return {"documents": retrieved_documents}

2. Evaluation Node

UPPER_THRESHOLD = 0.7
LOWER_THRESHOLD = 0.3

def evaluate_node(state: GraphState) -> dict:
    query = state["question"]
    docs = state["documents"]
    
    evaluator_prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an uncompromising retrieval evaluator for a RAG system. "
                   "Grade the relevance of the retrieved document chunk to the user question. "
                   "Return a score between 0.0 and 1.0 and a brief reason. Be conservative."),
        ("human", "Question: {question}\n\nDocument Chunk:\n{context}")
    ])
    eval_chain = evaluator_prompt | llm.with_structured_output(ChunkEvaluation)
    
    good_documents = []
    scores = []
    reasons = []
    
    for doc in docs:
        result: ChunkEvaluation = eval_chain.invoke({
            "question": query,
            "context": doc.page_content
        })
        scores.append(result.score)
        reasons.append(result.reason)
        if result.score >= LOWER_THRESHOLD:
            good_documents.append(doc)
            
    # CRAG Evaluation Branching Logic
    if any(s >= UPPER_THRESHOLD for s in scores):
        verdict = "CORRECT"
        summary_reason = f"At least one document scored >= {UPPER_THRESHOLD}."
    elif all(s < LOWER_THRESHOLD for s in scores):
        verdict = "INCORRECT"
        summary_reason = f"All documents scored < {LOWER_THRESHOLD}."
    else:
        verdict = "AMBIGUOUS"
        summary_reason = "Retrieved documents provide partial context without a clear high-confidence match."
        
    return {
        "good_docs": good_documents,
        "verdict": verdict,
        "reason": summary_reason
    }

3. Query Rewriter Node

def rewrite_query_node(state: GraphState) -> dict:
    query = state["question"]
    
    rewrite_prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert search engine query optimizer. Convert the user input "
                   "into a keyword-dense web search query. If the input implies recency, append temporal "
                   "constraints like 'last 30 days'. Do not answer the question; return only the query."),
        ("human", "User query: {question}")
    ])
    rewrite_chain = rewrite_prompt | llm.with_structured_output(RewrittenQuery)
    result = rewrite_chain.invoke({"question": query})
    return {"web_query": result.query}

4. Web Search Node

def web_search_node(state: GraphState) -> dict:
    query = state["web_query"]
    tavily_tool = TavilySearchResults(max_results=3)
    search_results = tavily_tool.invoke({"query": query})
    
    external_docs = []
    for item in search_results:
        external_docs.append(Document(
            page_content=item["content"],
            metadata={"source": item["url"]}
        ))
    return {"web_docs": external_docs}

5. Knowledge Refinement Node

def decompose_into_sentences(text: str) -> List[str]:
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    return [s.strip() for s in sentences if len(s.strip()) > 5]

def refine_node(state: GraphState) -> dict:
    query = state["question"]
    verdict = state["verdict"]
    
    # Adaptive context selection depending on verdict
    if verdict == "CORRECT":
        active_docs = state["good_docs"]
    elif verdict == "INCORRECT":
        active_docs = state["web_docs"]
    else:  # AMBIGUOUS uses both internal and external context
        active_docs = state.get("good_docs", []) + state.get("web_docs", [])
        
    full_text = " ".join([d.page_content for d in active_docs])
    strips = decompose_into_sentences(full_text)
    
    filter_prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a strict relevance filter. Set keep=True ONLY if the sentence directly "
                   "helps answer the user question. Rely strictly on the text provided."),
        ("human", "Question: {question}\n\nSentence: {sentence}")
    ])
    filter_chain = filter_prompt | llm.with_structured_output(SentenceFilter)
    
    kept_strips = []
    for strip in strips:
        res: SentenceFilter = filter_chain.invoke({"question": query, "sentence": strip})
        if res.keep:
            kept_strips.append(strip)
            
    refined_context = " ".join(kept_strips)
    return {
        "strips": strips,
        "kept_strips": kept_strips,
        "refined_context": refined_context
    }

6. Generation Node

def generate_node(state: GraphState) -> dict:
    query = state["question"]
    context = state["refined_context"]
    
    gen_prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert technical assistant. Answer the user prompt using ONLY the provided context. "
                   "If the context does not contain sufficient facts to answer accurately, declare that you do not know. "
                   "Do not fabricate or rely on external assumptions.\n\nContext:\n{context}"),
        ("human", "{question}")
    ])
    gen_chain = gen_prompt | llm
    response = gen_chain.invoke({"question": query, "context": context})
    return {"answer": response.content}

Dynamic Routing Function

def route_after_evaluation(state: GraphState) -> Literal["refine_node", "rewrite_query_node"]:
    verdict = state["verdict"]
    if verdict == "CORRECT":
        return "refine_node"
    else:  # Handles both INCORRECT and AMBIGUOUS
        return "rewrite_query_node"

Graph Assembly and Compilation

workflow = StateGraph(GraphState)

# Add Nodes
workflow.add_node("retrieve_node", retrieve_node)
workflow.add_node("evaluate_node", evaluate_node)
workflow.add_node("rewrite_query_node", rewrite_query_node)
workflow.add_node("web_search_node", web_search_node)
workflow.add_node("refine_node", refine_node)
workflow.add_node("generate_node", generate_node)

# Connect Edges
workflow.add_edge(START, "retrieve_node")
workflow.add_edge("retrieve_node", "evaluate_node")

# Conditional Branching Edge
workflow.add_conditional_edges(
    "evaluate_node",
    route_after_evaluation,
    {
        "refine_node": "refine_node",
        "rewrite_query_node": "rewrite_query_node"
    }
)

workflow.add_edge("rewrite_query_node", "web_search_node")
workflow.add_edge("web_search_node", "refine_node")
workflow.add_edge("refine_node", "generate_node")
workflow.add_edge("generate_node", END)

crag_app = workflow.compile()

9. Full Flow Execution Walkthrough

The following traces illustrate how the compiled state graph handles distinct runtime conditions:

10. Real-World Practical Pitfalls & Debugging

When deploying CRAG in production environments, several operational issues can emerge:

1. High Latency Overhead: CRAG runs an evaluation step and a refinement step using LLM calls. If evaluated sequentially over 4 chunks containing 5 sentences each, you could trigger up to $1 + 4 + 20 = 25$ LLM calls per request.

   • Fix: Batch sentence evaluations into a single JSON-array structured output call, or deploy a localized small language model (SLM) like Llama-3-8B-Instruct or fine-tuned T5 for filtering.

2. Overly Conservative Evaluator Scoring: If the prompt for the evaluator instructs it to be "overly strict," valid chunks may receive scores of $0.29$, mistakenly routing the system into the INCORRECT path and invoking unnecessary web searches.

   • Fix: Tune $\tau_U$ and $\tau_L$ empirically based on a domain-specific evaluation test set. Set baseline defaults ($\tau_U = 0.7$, $\tau_L = 0.3$) and adjust using precision-recall metrics against golden datasets.

3. External Search Contamination: Web results returned by tools like Tavily often include cookie banners, navigation menus, and sponsored content.

   • Fix: Ensure external results pass through the same sentence-level decomposition and filtering steps applied to internal documents. Never forward raw web scrapes directly to the generator.

4. Web API Rate Limits & Timeouts: Network calls to external search engines can fail or hang under heavy traffic.

   • Fix: Wrap the web_search_node with circuit breakers, timeouts (e.g., maximum 3.0 seconds), and automatic fallbacks to internal parametric responses with explicit disclaimers.

Summary and Key Takeaways

• The Fundamental Flaw in Base RAG: Standard RAG trusts the vector database blindly. When retrieved documents are irrelevant, the system forces generation anyway, producing hallucinations or factually incorrect claims.
• The Core Role of the Evaluator: CRAG inserts an evaluation node between retrieval and generation to classify context into three distinct states: CORRECT, INCORRECT, or AMBIGUOUS.
• Confidence-Driven Routing: The evaluation path is governed by an upper threshold ($\tau_U$) and a lower threshold ($\tau_L$). Chunks clearing $\tau_U$ trigger internal refinement, chunks falling below $\tau_L$ trigger external web searches, and middle-ground chunks merge both sources.
• Granular Knowledge Refinement: Chunking documents by raw character length introduces off-topic noise. CRAG breaks chunks down into individual sentences (strips), filters out unhelpful lines, and recombines the remaining text.
• Low-Quality Chunk Discarding: Even in CORRECT retrievals, only chunks that clear the lower threshold ($\ge \tau_L$) are retained for answer synthesis; irrelevant neighbors returned by the vector search are dropped.
• Search Query Adaptation: Raw user queries perform poorly on search engines. CRAG rewrites user prompts into keyword-rich queries with temporal boundaries before sending them to external tools like Tavily.
• State-Driven Orchestration via LangGraph: By maintaining unified state attributes (good_docs, web_docs, verdict), the execution pipeline routes both INCORRECT and AMBIGUOUS flows cleanly through a shared query-rewriting and refinement pipeline without duplicating graph nodes.