How to Build a Corrective RAG Agent with LangGraph: A Step‑by‑Step Guide

LangGraph, introduced in LangChain 0.1, represents single‑ or multi‑agent systems as graphs, allowing developers to define explicit task nodes and state that flows between them. Its main goal is to solve the complex “loop” problem in LLM applications and make agent logic less of a black box.

Corrective RAG (C‑RAG) Overview

RAG (Retrieval‑Augmented Generation) performance heavily depends on the relevance of retrieved documents. C‑RAG adds a lightweight evaluator (often an LLM) that classifies each document as relevant , suspect , or irrelevant . Based on the classification, the pipeline either proceeds to answer generation or triggers additional steps such as query rewriting and web search.

Graph and State Design for C‑RAG

The graph consists of five core nodes:

retrieve : fetches semantically similar documents from a vector store.

grade_documents : runs the evaluator and returns only the relevant documents while setting a flag run_web_search to "Yes" when none are relevant.

transform_query : rewrites the original question using an LLM when the evaluator signals low relevance.

web_search : calls a Bing search wrapper to supplement knowledge.

generate : feeds the (possibly rewritten) question and the final set of documents to an LLM to produce the answer.

The shared State (implemented as a TypedDict) stores: question: original or rewritten query. documents: list of retrieved and/or web‑search results. run_web_search: "Yes" or "No" flag from the evaluator. generation: final answer generated by the LLM.

class GraphState(TypedDict):
    keys: Dict[str, any]

Node Implementations

retrieve

def retrieve(state):
    """Extract the question, run the retriever, and store documents in state."""
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = retriever.get_relevant_documents(question)
    return {"keys": {"documents": documents, "question": question}}

grade_documents

def grade_documents(state):
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]
    # Build a prompt and invoke an LLM‑with‑tool chain to get a yes/no per doc
    filtered_docs = []
    search = "No"
    for d in documents:
        score = chain.invoke({"question": question, "context": d.page_content})
        if score[0].binary_score == "yes":
            filtered_docs.append(d)
        else:
            search = "Yes"
    return {"keys": {"documents": filtered_docs, "question": question, "run_web_search": search}}

generate

def generate(state):
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]
    prompt = hub.pull("rlm/rag-prompt")
    llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0, streaming=True)
    rag_chain = prompt | llm | StrOutputParser()
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"keys": {"documents": documents, "question": question, "generation": generation}}

transform_query

def transform_query(state):
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]
    prompt = PromptTemplate(template="""你需要生成针对检索优化的问题。请根据输入内容，尝试推理其中的语义意图/含义。这是初始问题：
{question}
请提出一个改进的问题：""",
                            input_variables=["question"])
    model = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-1106", streaming=True)
    chain = prompt | model | StrOutputParser()
    better_question = chain.invoke({"question": question})
    return {"keys": {"documents": documents, "question": better_question}}

web_search

def web_search(state):
    state_dict = state["keys"]
    question = state_dict["question"]
    documents = state_dict["documents"]
    tool = BingSearchAPIWrapper(k=3)
    docs = tool.run(question)
    web_results = Document(page_content="
".join(docs))
    documents.append(web_results)
    return {"keys": {"documents": documents, "question": question}}

decide_to_generate (conditional edge helper)

def decide_to_generate(state):
    state_dict = state["keys"]
    search = state_dict["run_web_search"]
    if search == "Yes":
        return "transform_query"
    else:
        return "generate"

Assembling the Graph

# Define the workflow
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search", web_search)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {"transform_query": "transform_query", "generate": "generate"},
)
workflow.add_edge("transform_query", "web_search")
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)
app = workflow.compile()

Testing the Application

Two scenarios are demonstrated:

Relevant documents : The pipeline follows retrieve → grade_documents → generate and returns an answer without invoking web_search.

Irrelevant documents : After grading, the evaluator triggers transform_query and web_search, resulting in the path

retrieve → grade_documents → transform_query → web_search → generate

, which enriches the knowledge base before answering.

Logs printed during execution show which nodes were visited and the final generated answer stored in state["generation"].

Conclusion

The example demonstrates how LangGraph can be used to build a more robust RAG system by adding document relevance assessment, query rewriting, and fallback web search. Because the workflow is expressed as a graph, complex conditional logic and multi‑agent coordination become straightforward, showcasing LangGraph’s power for constructing advanced AI agents.