When Retrieval Goes Wrong, Fix It: Build a Self-Correcting RAG Pipeline With LangGraph

Chris Harper

4 min read

Jul 13, 2026 · 20:11 UTC

Tutorial

RAG

Best Practices

Agents

CRAG adds an LLM-graded document check after each retrieval — if scores are low it rewrites the query and searches again, so your pipeline catches its own bad retrieval before hallucinating on poor context.

What you'll be able to do after this:

Understand the retrieve → grade → correct loop that distinguishes CRAG from basic RAG
Implement a LangGraph state machine with document-grader, query-rewriter, and web-fallback nodes
Know when CRAG is worth the extra latency vs simpler retrieval patterns

The silent failure mode in basic RAG

Standard RAG has a fundamental blind spot: if the vectorstore returns stale, irrelevant, or off-topic chunks, the LLM doesn't refuse — it hallucinates a confident-sounding answer from bad context. You only discover the failure downstream, through wrong outputs or user complaints.

CRAG (Corrective RAG) closes this loop by inserting a grading step between retrieval and generation.

The CRAG loop

After each retrieval, an LLM grades every document as Correct (clearly relevant), Incorrect (clearly irrelevant), or Ambiguous:

All Correct → generate immediately from the retrieved docs
All Incorrect → rewrite the query, fall back to web search (Tavily/DuckDuckGo), generate from fresh context
Mixed/Ambiguous → refine: strip each ambiguous document into short "knowledge strips," re-grade each strip individually, keep only the relevant ones, supplement with web results if needed

LangGraph models this as a state machine with conditional edges — the graph decides at runtime whether to proceed directly to generation or loop back through rewrite and web search.

Runnable implementation

from langgraph.graph import StateGraph, END
from typing import List, TypedDict
from langchain_core.documents import Document
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults

class RAGState(TypedDict):
    question: str
    documents: List[Document]
    generation: str
    web_search_needed: bool

llm = ChatAnthropic(model="claude-sonnet-5", temperature=0)

# Step 1: grade each retrieved document
def grade_documents(state: RAGState) -> RAGState:
    grader = llm.with_structured_output({
        "type": "object",
        "properties": {"relevant": {"type": "boolean"}},
        "required": ["relevant"]
    })
    filtered, web_needed = [], False
    for doc in state["documents"]:
        result = grader.invoke(
            f"Is this passage relevant to the question '{state['question']}'?\n\n"
            f"{doc.page_content[:600]}"
        )
        if result["relevant"]:
            filtered.append(doc)
        else:
            web_needed = True
    return {"documents": filtered, "web_search_needed": web_needed or len(filtered) == 0}

# Step 2: rewrite query when docs were insufficient
def rewrite_query(state: RAGState) -> RAGState:
    rewritten = llm.invoke(
        f"Rewrite this question for a better web search: {state['question']}"
    ).content
    return {"question": rewritten}

# Step 3: web search fallback
def web_search(state: RAGState) -> RAGState:
    results = TavilySearchResults(max_results=3).invoke(state["question"])
    web_docs = [Document(page_content=r["content"]) for r in results]
    return {"documents": state["documents"] + web_docs}

# Step 4: generate from final context
def generate(state: RAGState) -> RAGState:
    context = "\n\n".join(d.page_content for d in state["documents"])
    answer = llm.invoke(
        f"Answer based only on this context:\n{context}\n\nQuestion: {state['question']}"
    ).content
    return {"generation": answer}

# Conditional edge: go to rewrite or straight to generate
def route_after_grading(state: RAGState) -> str:
    return "rewrite" if state["web_search_needed"] else "generate"

workflow = StateGraph(RAGState)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite", rewrite_query)
workflow.add_node("web_search", web_search)
workflow.add_node("generate", generate)

workflow.set_entry_point("grade_documents")
workflow.add_conditional_edges("grade_documents", route_after_grading,
    {"rewrite": "rewrite", "generate": "generate"})
workflow.add_edge("rewrite", "web_search")
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

app = workflow.compile()
result = app.invoke({
    "question": "What changed in vLLM v0.9?",
    "documents": retriever.invoke("vLLM v0.9"),  # your vectorstore retrieval here
    "generation": "",
    "web_search_needed": False
})
print(result["generation"])

When to use CRAG

High-stakes Q&A where a wrong answer has real consequences
Knowledge gaps: your vectorstore is a snapshot; questions may need fresher information
Ambiguous queries that can semantically match unrelated documents

When to skip it

Latency-sensitive paths: grading adds 1–2 extra LLM calls per query
Closed-domain corpora where you fully control document quality and freshness
Simple lookup tasks where keyword or metadata filters already ensure precision

CRAG pairs naturally with the cross-encoder reranker from an earlier post — run the reranker first to surface the best candidates, then CRAG to verify relevance and trigger correction if needed.