幻觉，是大模型与生俱来的“阿喀琉斯之踵”。

在 GPT-4 等通用大模型横扫各行各业的今天，企业级落地却面临着一个尴尬的现实：模型宁愿编造一个看似合理的谎言，也不愿承认“我不知道”。

这种“一本正经地胡说八道”在严肃场景（如法律、医疗、金融）是致命的。传统的 RAG（检索增强生成）虽然通过外挂知识库缓解了这一问题，但仍存在严重缺陷：如果检索到的文档本身就是垃圾，或者与问题毫不相关，模型依然会基于这些错误上下文进行强行总结。

我们需要一种机制，让 AI 在生成回答前先进行一次“自我审视”：这文档到底有没有用？没用就别瞎编！

这就是 Corrective RAG (CRAG) 的核心思想。今天，我们将利用 LangChain 最新的LangGraph框架，从零手搓一套生产级 CRAG 系统。这不仅是代码实战，更是对 Agentic RAG（代理式 RAG）架构的一次深度思维升级。

---

一、 深度洞察：从 Naive RAG 到 Corrective RAG 的进化论

在动手之前，必须先厘清技术演进的脉络。为什么传统的 RAG 不够用？

1. 传统 RAG 的“信任危机”

在标准的 Naive RAG 流程中：用户提问 -> 向量检索 -> 拼接上下文 -> LLM 生成。这里存在一个默认假设：检索到的 Top-K 个文档一定包含答案。

然而，现实往往是残酷的。根据行业数据，当检索内容的精准度低于阈值时，LLM 极易产生“误读”或“幻觉”。

2. CRAG：引入“自省开关”

CRAG（Corrective Retrieval-Augmented Generation）由 IBM Research 等机构提出，其核心在于引入了一个轻量级检索评估器。它不再盲目信任检索结果，而是增加了以下决策逻辑：

• Correct（正确）：文档与问题相关 -> 走生成流程。
• Incorrect（错误）：文档不相关 -> 拒绝回答 或 触发 Web Search 补救。
• Ambiguous（模糊）：不确定 -> 这一环节在工程中通常被合并到 Correct 或 Incorrect 以简化逻辑。

3. 技术选型对比：LangChain (LCEL) vs LangGraph

为什么我们选择 LangGraph 而不是传统的 LangChain Chain？因为 CRAG 是一个包含循环和条件分支的图结构，而不是一条直线。

特性维度	Traditional Chain (LCEL)	LangGraph (State Machine)
架构模式	线性流 (DAG)	徤环图
流程控制	顺序执行，难以中途跳转	支持条件边、循环
状态管理	隐式传递	显式 State 对象，易于持久化与中断
可控性	黑盒，难以插入人工审核	支持人在回路
适用场景	简单的 Q&A	复杂 Agent、多轮反思、CRAG

---

二、 架构可视化：CRAG 的逻辑蓝图

在编写代码前，我们先透视 LangGraph 的核心架构。可以看到，最关键的变化在于 Grade Documents 这一节点，它决定了流程的走向。

流程解析：

1. Retrieve: 从向量库捞数据。
2. Grade (自省开关): 使用小模型（如 GPT-3.5 或 Llama Guard）快速判断文档是否包含答案。
3. Decision:
   • 如果 YES：直接生成。
   • 如果 NO：不走生成，而是去“查重写”或者“搜网络”，甚至直接抛出“抱歉，知识库未涵盖此问题”。

---

三、 硬核手搓：LangGraph 实战详解

现在，让我们打开 IDE，用代码构建这个“自省开关”。我们将使用 Python 和 langgraph 库。

3.1 环境与状态定义

LangGraph 的核心是 State（状态）。它像是一个全局的“内存条”，在各个节点间传递和更新。

from typing import List, TypedDict, Optional
from langchain_core.documents import Document

# 定义图的状态
class CRAGState(TypedDict):
    """
    代表了在图中流转的所有信息
    """
    question: str               # 用户原始问题
    generation: str             # LLM 生成的回答
    web_fallback: bool          # 是否触发网络搜索兜底
    documents: List[Document]   # 检索到的文档列表
    num_steps: int              # 记录走了几步 (用于防止死循环)

3.2 节点 1：检索器

这是最基础的 RAG 能力，这里假设你已经配置好了 retriever（如 ChromaDB 或 Pinecone）。

from langchain_core.messages import HumanMessage

def retrieve_node(state: CRAGState):
    print("--- [Node] RETRIEVE ---")
    documents = retriever.invoke(state["question"])
    return {"documents": documents, "num_steps": state["num_steps"] + 1}

3.3 节点 2：文档相关性评估

这是 “自省开关” 的核心。我们不直接生成，而是先让 LLM 打个分：yes or no。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 定义输出格式
class GradeDocuments(BaseModel):
    """Binary score for relevance check on retrieved documents."""
    binary_score: str = Field(description="Documents are relevant to the question, 'yes' or 'no'")

# LLM 初始化
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)

# Prompt 设计 (关键！)
system_prompt = """
You are a grader assessing relevance of a retrieved document to a user question.
If the document contains keyword(s) or semantic meaning related to the question, grade it as relevant.
It does not need to be a stringent test. The goal is to filter out erroneous retrievals.
Give a binary score 'yes' or 'no' to indicate whether the document is relevant to the question.
"""

grade_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "Retrieved document: \n\n {document} \n\n User question: {question}"),
])

retrieval_grader = grade_prompt | structured_llm_grader

def grade_documents_node(state: CRAGState):
    print("--- [Node] GRADE DOCUMENTS ---")
    question = state["question"]
    documents = state["documents"]
    
    # 默认假设：文档不相关，触发 fallback
    relevant_docs = []
    fallback = True
    
    for doc in documents:
        score = retrieval_grader.invoke({"question": question, "document": doc.page_content})
        if score.binary_score == "yes":
            print(f"  ---> Document is RELEVANT")
            relevant_docs.append(doc)
            fallback = False # 只要有一个相关，就尝试回答
    
    # 如果没有任何相关文档，保留空列表，触发后续的网络搜索或重试
    return {"documents": relevant_docs, "web_fallback": fallback, "num_steps": state["num_steps"] + 1}

3.4 节点 3：兜底策略

如果自省发现“文档全是垃圾”，我们选择查询重写+ 网络搜索（模拟）。

def transform_query_node(state: CRAGState):
    print("--- [Node] TRANSFORM QUERY ---")
    question = state["question"]
    
    # Prompt: 重写问题，使其更适合搜索引擎
    rewrite_prompt = ChatPromptTemplate.from_messages([
        ("system", "You a question re-writer that converts an input question to a better version that is optimized for web search. Only output the new question."),
        ("human", "Here is the initial question: \n\n {question} \n Formulate an improved question."),
    ])
    
    question_rewriter = rewrite_prompt | llm | StrOutputParser()
    better_question = question_rewriter.invoke({"question": question})
    
    # 模拟 Web Search (实际场景中可接入 Tavily 或 Google Search API)
    # 这里简单打印，实际应返回 web_search 的结果作为 documents
    print(f"  ---> Rewritten question: {better_question}")
    web_docs = [{"page_content": "Web search result placeholder...", "metadata": {}}] # 伪代码
    
    return {"documents": web_docs, "question": better_question, "num_steps": state["num_steps"] + 1}

3.5 节点 4：生成回答

最后是生成节点。

def generate_node(state: CRAGState):
    print("--- [Node] GENERATE ---")
    question = state["question"]
    documents = state["documents"]
    
    # RAG Prompt
    prompt = ChatPromptTemplate.from_template(
        "Answer the question based only on the following context:\n{context}\n\nQuestion: {question}"
    )
    
    rag_chain = prompt | llm | StrOutputParser()
    
    context = "\n".join(doc.page_content for doc in documents)
    generation = rag_chain.invoke({"context": context, "question": question})
    
    return {"generation": generation, "num_steps": state["num_steps"] + 1}

3.6 灵魂注入：构建 Graph 与 路由逻辑

这是很多教程缺失的关键部分。有了节点，还必须把它们连起来。我们需要定义一个 Conditional Edge（条件边）。

from langgraph.graph import StateGraph, END

# 1. 初始化图
workflow = StateGraph(CRAGState)

# 2. 添加节点
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("grade_documents", grade_documents_node) # 自省节点
workflow.add_node("generate", generate_node)
workflow.add_node("transform_query", transform_query_node)

# 3. 构建边

# 起点
workflow.set_entry_point("retrieve")

# Retrieve -> Grade
workflow.add_edge("retrieve", "grade_documents")

# Grade -> Decision (条件路由)
def decide_to_generate(state: CRAGState):
    """
    根据 grade_documents 的结果决定下一步
    """
    if state["web_fallback"]:
        print("--- [Decision] All documents are NOT relevant -> Transform Query ---")
        return "transform_query"
    else:
        print("--- [Decision] Documents are relevant -> Generate ---")
        return "generate"

# 添加条件边：从 grade_documents 出发，根据 decide_to_generate 函数的返回值决定走向
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate"
    }
)

# Transform -> Generate (兜底后生成)
workflow.add_edge("transform_query", "generate")

# Generate -> END
workflow.add_edge("generate", END)

# 4. 编译图
app = workflow.compile()

---

四、 运行效果演示：有与无的区别

场景 A：知识库中不存在答案（瞎编测试）

用户提问：“请告诉我 2024 年火星殖民地的平均房价是多少？”

• Without CRAG (Naive RAG):
  • 检索到可能相关的文档（比如关于地球房产或火星探测器的）。
  • LLM 强行总结：“根据文档，2024年火星殖民地平均房价约为…” (一本正经胡说)
• With CRAG (LangGraph):
  • [Retrieve]: 找到一些不相关文档。
  • [Grade]: LLM 判定文档不包含“火星房价”信息 -> web_fallback=True。
  • [Decision]: 路由到 transform_query。
  • [Transform]: 重写问题，尝试搜索或触发“无法回答”逻辑。
  • 最终输出: “抱歉，根据现有知识库无法回答该问题，建议查阅最新资讯。” (或者是基于 Web Search 的真实结果)

场景 B：知识库中有准确答案

• With CRAG:
  • [Retrieve]: 检索到文档。
  • [Grade]: 判定相关 -> web_fallback=False。
  • [Generate]: 精准回答。

---

五、 总结与资源

从 Naive RAG 到 Corrective RAG，本质上是从“盲目相信”到“批判性思考”的跨越。通过 LangGraph，我们不再是在写死板的代码，而是在设计一个具有神经回路的 AI 大脑。

核心收获：

1. 自省机制：Grade 节点是解决幻觉的第一道防线。
2. 图的灵活性：LangGraph 的 StateGraph 让复杂逻辑（循环、分支）变得清晰可维护。
3. Fallback 兜底：查不到就闭嘴或去搜，绝不能瞎编。

开源资源与引用：

• LangGraph 官方文档: https://langchain-ai.github.io/langgraph/
• CRAG 论文: Corrective Retrieval Augmented Generation (Yan et al., 2024) https://arxiv.org/abs/2401.15884
• LangChain Github: https://github.com/langchain-ai/langchain

这篇文章仅是 Agentic RAG 的冰山一角。在后续的文章中，我们将探讨如何结合 Self-RAG 进一步提升推理能力。敬请期待。