Corrective RAG with LangChain and LangGraph: How to Self-Heal Retrieval Failures

Corrective RAG with LangChain and LangGraph: How to Self-Heal Retrieval Failures

Imagine you ask a question and get a wrong or incomplete answer. It’s frustrating, right? This happens sometimes even with smart AI systems called RAG applications. RAG stands for Retrieval Augmented Generation.

RAG systems work by finding information in your documents first, then using that information to create an answer. But what if the first search doesn’t find the right stuff? That’s where “Corrective RAG” comes in to save the day.

We’re going to explore how to build these smart, self-healing RAG systems using LangChain corrective RAG and LangGraph. You’ll learn how your AI can fix its own mistakes when looking for information. This makes your AI assistant much more reliable and helpful.

The Problem with Basic RAG: When Retrieval Fails

Think of a basic RAG system like a student looking up answers in a textbook. You ask a question, and the student quickly finds some pages that seem related. Then, they try to answer your question using only those pages.

Sometimes, the student might pick the wrong pages, or the pages might not have enough information. This means the answer you get might be wrong or unclear. This is a common problem in RAG, often called “retrieval failure.”

Retrieval failure can happen for many reasons. Maybe your question wasn’t clear enough, or the documents are too technical. Sometimes, the initial search simply doesn’t pull up the most relevant pieces of text. This leads to the AI making up answers, which we call “hallucinations,” or giving you incomplete information.

Enter Corrective RAG: The Self-Healing AI

Corrective RAG, or CRAG, is like giving that student a supervisor. If the student picks some pages and the supervisor thinks they’re not good enough, the supervisor tells them to try again. The supervisor might even give them new ideas on how to search better.

In the world of AI, corrective retrieval means your RAG system doesn’t just retrieve once and answer. It first checks if the retrieved information is good. If it’s not, the system tries different ways to find better information before giving you a final answer. This “self-correction” makes the AI much smarter.

It helps to ensure you get more accurate and complete answers. Your AI can then adapt and improve its search strategy on the fly. This way, you get reliable information every time.

LangChain: Your Foundation for Smart RAG

LangChain is a powerful toolkit that helps you build AI applications like RAG. It provides many building blocks that make it easy to connect different parts of your AI system. You can load documents, split them into smaller pieces, and store them in a special database called a vector store.

LangChain also helps you talk to large language models (LLMs), which are the “brains” of your AI. It makes it simple to send your question and the retrieved documents to the LLM to get an answer. Building a RAG application with LangChain and a vector store is a common starting point for many AI projects. For a deeper dive into setting up your RAG, you might want to read Build RAG Applications with LangChain & Vector Store in 2026.

These building blocks are essential for even the most basic RAG. They handle the initial steps of finding and preparing your information. With LangChain, you have a solid foundation to start adding more advanced features.

LangGraph: Giving Your RAG System a Brain

While LangChain helps you build the individual parts, LangGraph helps you connect them in a smart way. Think of LangGraph as the conductor of an orchestra. It tells each instrument (or AI component) when to play and what to do based on what’s happening.

LangGraph allows you to create complex “state machines” or “graphs” for your AI. This means your AI can follow different paths depending on the situation. For example, if the first search fails, LangGraph can direct it to try a different search method.

This is perfect for LangGraph RAG because it enables the self-correction loop. Your AI can decide, “Should I answer now, or do I need to re-retrieve or try a web search fallback?” You can learn more about building these smart multi-step agents in LangGraph StateGraph: Building Multi-Step AI Agents. LangGraph is what truly gives your RAG system the ability to think and adapt.

How Corrective RAG Works: A Step-by-Step Journey

Let’s break down how a corrective retrieval system makes decisions. It’s like a smart detective following clues and changing tactics if needed. This process involves several key stages.

1. Initial Query & Retrieval

You ask your question. The RAG system takes your question and tries to find related documents in its knowledge base. This is the standard first step of any RAG application.

It pulls out what it thinks are the most relevant pieces of information, called “chunks.” These chunks are then sent to the next stage for review.

2. Critique Stage

This is where the “self-healing” truly begins. An LLM (the AI brain) acts as a critic. It looks at your original question and the documents that were just retrieved. The LLM’s job is to decide if those documents are actually useful enough to answer your question.

It might ask questions like: “Do these documents directly address the user’s query?” or “Is there enough information here to form a complete answer?” If the critique determines the documents are not good enough, the system moves to the correction stage.

3. Correction Strategies

If the critique says, “Not good enough!” the system doesn’t give up. Instead, it tries different ways to find better information. These are its fallback retrieval strategies.

• Query Rewriting: The AI might try to rephrase your original question. Sometimes, a slightly different way of asking can lead to much better search results.
• Re-ranking: It might have retrieved many documents, some good, some less so. The system can then use another method to sort them again, pushing the truly relevant ones to the top.
• Web Search Fallback: If its internal documents simply don’t have the answer, the system can decide to use a web search tool. This is like looking beyond your textbook when you can’t find the answer there.

4. Re-retrieval

After applying a correction strategy, the system performs another search. If it rewrote the query, it searches with the new query. If it decided to use web search, it performs a web search.

The goal is to get a fresh set of potentially better documents. These new documents then go back through the critique stage.

5. Final Answer Generation

Once the critique stage confirms that the retrieved documents are good enough, the system stops searching. It then takes all the good information it has found and uses the LLM to create a clear and helpful answer to your original question. This multi-step process ensures higher quality responses.

Building Corrective RAG with LangChain and LangGraph: Practical Steps

Let’s see how you can actually build a LangChain corrective RAG system that can self-heal. We’ll combine LangChain for the individual steps and LangGraph for the smart decision-making.

1. Setting Up Your Basic RAG (LangChain)

First, you need a basic RAG setup. This involves loading your documents, splitting them, creating embeddings, and storing them in a vector database. We’ll use a simple in-memory vector store for this example.

You’ll need langchain and langgraph installed. Let’s start with a mock document loader and vector store.

pip install -qU langchain langchain-community langchain-openai chromadb

Now, the Python code:

from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# --- 1. Define Dummy Documents ---
docs_content = [
    "The capital of France is Paris. Paris is famous for its Eiffel Tower.",
    "Berlin is the capital of Germany, known for its Brandenburg Gate.",
    "What is the largest animal on Earth? The blue whale is the largest animal.",
    "The Amazon rainforest is the largest tropical rainforest in the world.",
    "The deepest ocean trench is the Mariana Trench in the Pacific Ocean."
]

# Save to a temporary file for TextLoader
with open("temp_docs.txt", "w") as f:
    f.write("\n".join(docs_content))

# --- 2. Load and Split Documents ---
loader = TextLoader("temp_docs.txt")
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=20)
splits = text_splitter.split_documents(documents)

# --- 3. Create Embeddings and Vector Store ---
embeddings_model = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents=splits, embedding=embeddings_model)
retriever = vectorstore.as_retriever()

# --- 4. Basic RAG Chain ---
llm = ChatOpenAI(model="gpt-4o", temperature=0) # Use a good LLM
rag_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are an assistant for question-answering tasks. Use the following retrieved context to answer the question. If you don't know the answer, just say that you don't know."),
        ("user", "Question: {question}\nContext: {context}")
    ]
)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

basic_rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

print("--- Basic RAG Example ---")
# Example query that might lead to bad retrieval for correction later
query_basic = "What is the capital of Japan?" # This data is NOT in our docs
print(f"Query: {query_basic}")
print(f"Basic RAG Answer: {basic_rag_chain.invoke(query_basic)}")

query_good = "What is the capital of France?" # This data IS in our docs
print(f"\nQuery: {query_good}")
print(f"Basic RAG Answer: {basic_rag_chain.invoke(query_good)}")

In this setup, if you ask “What is the capital of Japan?”, the basic RAG will likely say it doesn’t know. Our documents don’t have that information. This is a retrieval failure we want to self-heal.

2. The “Critique” Module (LangChain)

Now, let’s build the part that judges the retrieved documents. This module will take the user’s question and the documents found by the retriever. It then uses an LLM to decide if these documents are relevant and sufficient.

from typing import List, Dict, Any

# --- Critique Prompt ---
critique_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant that evaluates the relevance of retrieved documents to a user's question. "
                   "Your task is to determine if the provided documents are sufficient and relevant to answer the question. "
                   "Respond with 'yes' if they are sufficient and relevant, and 'no' otherwise. "
                   "Also, provide a brief reason for your 'yes' or 'no' answer."),
        ("user", "Question: {question}\n\nRetrieved Documents:\n{context}\n\nAre these documents sufficient and relevant to answer the question? (yes/no):")
    ]
)

critique_chain = (
    critique_prompt
    | llm
    | StrOutputParser()
)

# Function to parse critique output
def parse_critique(critique_response: str) -> Dict[str, Any]:
    response = critique_response.lower().strip()
    if response.startswith("yes"):
        return {"decision": "yes", "reason": response[3:].strip()}
    elif response.startswith("no"):
        return {"decision": "no", "reason": response[2:].strip()}
    return {"decision": "no", "reason": "Could not parse critique response."}

print("\n--- Critique Module Example ---")
# Test with good retrieval
good_context = "Paris is the capital of France."
critique_result_good = critique_chain.invoke({"question": "What is the capital of France?", "context": good_context})
print(f"Good Retrieval Critique: {parse_critique(critique_result_good)}")

# Test with bad retrieval
bad_context = "The largest animal is the blue whale."
critique_result_bad = critique_chain.invoke({"question": "What is the capital of Japan?", "context": bad_context})
print(f"Bad Retrieval Critique: {parse_critique(critique_result_bad)}")

This critique_chain will output a “yes” or “no” decision. This decision is crucial for LangGraph to know whether to proceed to answer or to try a correction.

3. Implementing Correction Strategies

If the critique says “no,” our system needs to try different strategies. We’ll implement two common ones: query rewriting and web search fallback.

a. Query Rewriting

Sometimes, the original query might be ambiguous or not perfectly aligned with the terms in your documents. An LLM can rephrase the query to improve retrieval.

# --- Query Rewriting Prompt ---
query_rewrite_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a query rewriting assistant. Given a user's original question and the fact that initial retrieval was poor, "
                   "rewrite the question to be more effective for searching a document database or the web. "
                   "Focus on clarity and keywords that might yield better results. "
                   "Original question: {original_question}"),
        ("user", "Rewrite the question: {original_question}")
    ]
)

query_rewrite_chain = (
    query_rewrite_prompt
    | llm
    | StrOutputParser()
)

print("\n--- Query Rewriting Example ---")
original_q = "Tell me about the biggest whale."
rewritten_q = query_rewrite_chain.invoke({"original_question": original_q})
print(f"Original Query: {original_q}")
print(f"Rewritten Query: {rewritten_q}")

This rewritten query can then be used with our existing retriever to try and get better documents.

b. Web Search Fallback

If internal documents consistently fail, a web search can be a powerful fallback. We’ll simulate a web search tool. For a real web search, you’d integrate with tools like Google Search API, Brave Search, etc., often using LangChain’s tool integrations. LangChain’s ability to use custom tools is powerful, as discussed in LangChain Google Gemini Function Calling Agent & Custom Tools.

For this example, we’ll mock a web search result.

# --- Mock Web Search Tool ---
def web_search_tool(query: str) -> str:
    print(f"--- Performing Mock Web Search for: '{query}' ---")
    if "capital of japan" in query.lower():
        return "Tokyo is the capital of Japan. It is located on the island of Honshu."
    if "largest animal" in query.lower():
        return "The blue whale (Balaenoptera musculus) is a marine mammal and is the largest animal known to have ever lived."
    return "No relevant web search results found for this query."

# --- Web Search Fallback Chain ---
web_search_chain = (
    RunnablePassthrough.assign(web_context=lambda x: web_search_tool(x["question"]))
    | ChatPromptTemplate.from_messages(
        [
            ("system", "You are an assistant that answers questions using the provided web search results. If the web search results are not sufficient, say you don't know."),
            ("user", "Question: {question}\nWeb Search Results: {web_context}")
        ]
    )
    | llm
    | StrOutputParser()
)

print("\n--- Web Search Fallback Example ---")
web_answer = web_search_chain.invoke({"question": "What is the capital of Japan?"})
print(f"Web Search Answer: {web_answer}")

4. Orchestrating with LangGraph

Now, we bring everything together using LangGraph to define the flow and decision-making logic. We’ll define states (what information we have) and nodes (what actions we can take).

First, define the GraphState that will carry information through our system.

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
import operator

# --- Define Graph State ---
class GraphState(TypedDict):
    """
    Represents the state of our graph.

    Attributes:
        question: The user's question.
        documents: List of retrieved documents.
        critique: The critique result (e.g., {"decision": "no", "reason": "..."}).
        rewritten_query: A potentially rewritten query.
        web_search_performed: Whether a web search has been attempted.
        answer: The final answer.
    """
    question: str
    documents: Annotated[List[str], operator.add] # Use operator.add to append documents
    critique: dict
    rewritten_query: str
    web_search_performed: bool
    answer: str

# --- Define Nodes (Functions for each step) ---

def retrieve(state: GraphState) -> GraphState:
    """
    Retrieve documents based on the current question or rewritten query.
    """
    print("---NODE: RETRIEVE---")
    question = state["rewritten_query"] if state["rewritten_query"] else state["question"]
    docs = retriever.invoke(question)
    return {"documents": docs}

def critique_retrieval(state: GraphState) -> GraphState:
    """
    Critique the retrieved documents for relevance and sufficiency.
    """
    print("---NODE: CRITIQUE RETRIEVAL---")
    question = state["question"]
    context = format_docs(state["documents"])
    critique_response = critique_chain.invoke({"question": question, "context": context})
    critique = parse_critique(critique_response)
    return {"critique": critique}

def rewrite_query(state: GraphState) -> GraphState:
    """
    Rewrite the original query if retrieval was poor.
    """
    print("---NODE: REWRITE QUERY---")
    original_question = state["question"]
    rewritten_q = query_rewrite_chain.invoke({"original_question": original_question})
    return {"rewritten_query": rewritten_q}

def web_search_fallback_node(state: GraphState) -> GraphState:
    """
    Perform a web search if other retrieval methods fail.
    """
    print("---NODE: WEB SEARCH FALLBACK---")
    question = state["rewritten_query"] if state["rewritten_query"] else state["question"]
    web_context = web_search_tool(question)
    # Here we append web context as a document for the final answer
    # A more sophisticated approach might merge or rank internal and external docs
    web_doc_object = [doc for doc in text_splitter.split_documents([TextLoader("dummy.txt").load()[0]])] # Create a dummy doc object
    web_doc_object[0].page_content = web_context
    return {"documents": [web_doc_object[0]], "web_search_performed": True}


def generate_answer(state: GraphState) -> GraphState:
    """
    Generate the final answer using all available documents.
    """
    print("---NODE: GENERATE ANSWER---")
    question = state["question"]
    context = format_docs(state["documents"])

    if state["web_search_performed"]:
        # If web search was performed, use the web_search_chain which expects specific formatting
        answer = web_search_chain.invoke({"question": question, "web_context": context})
    else:
        # Otherwise, use the standard RAG chain
        answer = basic_rag_chain.invoke({"context": context, "question": question})

    return {"answer": answer}

# --- Define Conditional Edges (Decision Points) ---
def decide_to_re_retrieve_or_answer(state: GraphState) -> str:
    """
    Decides whether to re-retrieve (with a rewritten query), perform web search, or generate an answer.
    """
    print(f"---DECISION: CRITIQUE IS: {state['critique']['decision']}---")
    if state["critique"]["decision"] == "yes":
        print("---DECISION: Retrieval is good, GENERATE ANSWER---")
        return "generate_answer"
    else:
        # If critique is 'no', and we haven't tried web search yet
        if not state["web_search_performed"]:
            print("---DECISION: Retrieval is bad. Trying REWRITE QUERY and re-retrieve, or WEB SEARCH if too many retries.---")
            # For simplicity, let's always try rewrite first if not good.
            # In a real system, you might count retries or have more complex logic
            if not state["rewritten_query"]: # Only rewrite once
                return "rewrite_query"
            else: # If query already rewritten, go to web search
                return "web_search_fallback"
        else:
            # If critique is 'no' and web search was already performed,
            # we've done all we can, generate answer with what we have.
            print("---DECISION: Retrieval bad, web search done. GENERATE ANSWER with best available.---")
            return "generate_answer"

Now, we build the LangGraph graph:

# --- Build the LangGraph Graph ---
workflow = StateGraph(GraphState)

# Add Nodes
workflow.add_node("retrieve", retrieve)
workflow.add_node("critique_retrieval", critique_retrieval)
workflow.add_node("rewrite_query", rewrite_query)
workflow.add_node("web_search_fallback", web_search_fallback_node)
workflow.add_node("generate_answer", generate_answer)

# Set Entry Point
workflow.set_entry_point("retrieve")

# Add Edges
workflow.add_edge("retrieve", "critique_retrieval")
workflow.add_conditional_edges(
    "critique_retrieval",
    decide_to_re_retrieve_or_answer,
    {
        "generate_answer": "generate_answer",
        "rewrite_query": "rewrite_query",
        "web_search_fallback": "web_search_fallback"
    }
)
workflow.add_edge("rewrite_query", "retrieve") # After rewriting, re-retrieve
workflow.add_edge("web_search_fallback", "generate_answer") # After web search, generate answer with new context
workflow.add_edge("generate_answer", END) # End of the graph

# Compile the graph
app = workflow.compile()

# --- Run the Corrective RAG System ---
print("\n--- Running Corrective RAG System (Question 1: Known in docs) ---")
inputs1 = {"question": "What is the capital of France?", "documents": [], "critique": {}, "rewritten_query": "", "web_search_performed": False, "answer": ""}
for s in app.stream(inputs1):
    print(s)
    print("---")
final_state1 = app.invoke(inputs1)
print(f"\nFinal Answer for '{inputs1['question']}': {final_state1['answer']}")


print("\n--- Running Corrective RAG System (Question 2: Not in docs, needs web search) ---")
inputs2 = {"question": "What is the capital of Japan?", "documents": [], "critique": {}, "rewritten_query": "", "web_search_performed": False, "answer": ""}
for s in app.stream(inputs2):
    print(s)
    print("---")
final_state2 = app.invoke(inputs2)
print(f"\nFinal Answer for '{inputs2['question']}': {final_state2['answer']}")


print("\n--- Running Corrective RAG System (Question 3: Ambiguous, needs rewrite, then web search) ---")
# Let's assume a question that might first retrieve irrelevant docs, then rewrite, then web search
inputs3 = {"question": "Tell me about the biggest animal that swims.", "documents": [], "critique": {}, "rewritten_query": "", "web_search_performed": False, "answer": ""}
for s in app.stream(inputs3):
    print(s)
    print("---")
final_state3 = app.invoke(inputs3)
print(f"\nFinal Answer for '{inputs3['question']}': {final_state3['answer']}")

In this LangGraph RAG example:

1. Retrieve: Gets documents based on the current question.
2. Critique Retrieval: Checks if documents are good enough.
3. Decide to Re-retrieve or Answer (Conditional Edge):
   • If documents are good (yes), go to generate_answer.
   • If documents are bad (no):
     • If no rewritten_query exists, go to rewrite_query.
     • If rewritten_query exists (meaning we already tried rewriting and it didn’t help enough), go to web_search_fallback.
4. Rewrite Query: Rewrites the query.
5. Web Search Fallback: Performs a mock web search.
6. Generate Answer: Uses the best available documents (either from internal retrieval or web search) to create the final answer.

This flow effectively demonstrates how your RAG system can adapt and self-correct its retrieval strategy.

Benefits of Corrective RAG

Implementing a LangChain corrective RAG system offers significant advantages:

1. Improved Accuracy and Reliability

Your AI is less likely to give wrong or incomplete answers. By verifying retrieval quality and trying again, the system ensures it has the best possible context. This directly leads to more trustworthy output.

2. Reduced Hallucinations

When the AI has poor information, it tends to “hallucinate” or make things up. Corrective RAG actively prevents this by ensuring only relevant and sufficient information is used. It acts as a guardrail against misinformation.

3. Enhanced User Experience

Users get better, more consistent answers without needing to rephrase their questions repeatedly. A self-healing system feels more intelligent and capable. This leads to higher user satisfaction and engagement.

4. Robustness to Vague Queries

Corrective retrieval makes your system more forgiving of unclear or ambiguous questions. If the initial interpretation doesn’t yield good results, the system can autonomously refine the query. This means users don’t have to be perfect in their phrasing.

5. Better Resource Utilization

By strategically using web search only when internal documents fail, you optimize resource usage. It avoids unnecessary external calls when answers are readily available internally. This approach saves time and computational cost.

Best Practices for Your Self-Healing RAG

Building a robust corrective RAG system requires attention to detail. Here are some tips to make your self-healing AI even better:

1. Careful Prompt Engineering for Critique

The critique module is the heart of corrective retrieval. Design your critique prompts very carefully. They need to clearly guide the LLM on how to judge relevance and sufficiency. For custom output parsing, you might find LangChain Custom Output Parser Tutorial helpful.

2. Implement Multiple Correction Strategies

Don’t just rely on one fallback retrieval method. Combine query rewriting, document re-ranking, and web search fallback. The more tools your AI has to fix problems, the more resilient it will be.

3. Define Clear State Transitions in LangGraph

Your LangGraph flow should be logical and cover all possible scenarios. Clearly define when to move from one step to another, based on the critique or other conditions. A well-structured StateGraph is key for a reliable system.

4. Test Thoroughly with Edge Cases

Test your corrective RAG with questions that are known to cause retrieval failures. Include queries that require rewriting, those that need web search, and those with ambiguous phrasing. This ensures your system handles real-world complexity.

5. Monitor Performance

Continuously monitor how often your system needs to self-correct and which strategies are most effective. This data can help you refine your prompts and retrieval methods over time. Performance monitoring is crucial for long-term success.

6. Optimize Document Chunking

The way you split your documents into chunks can significantly impact retrieval quality. Smaller, more focused chunks can sometimes be better, but too small can lose context. Explore techniques like semantic text splitting, which chunks by meaning, as discussed in LangChain Semantic Text Splitter: Chunk by Meaning.

Conclusion

You’ve learned how to build a smart RAG system that can find and fix its own retrieval mistakes. By using LangChain corrective RAG with the intelligent orchestration of LangGraph, you can create AI applications that are much more reliable and accurate. This self-healing capability is a game-changer for building truly intelligent assistants.

Moving forward, you can explore more advanced correction methods or even combine different vector stores, as shown in LangChain Weaviate Hybrid Search: Scalable RAG. The power of corrective retrieval ensures your AI delivers the best possible information, every single time. Your AI applications will be more robust and provide a much better experience for users.