Self-RAG 与 CRAG：让 RAG 系统自我纠错

[进阶选读] 本篇适合遇到以下问题的读者：检索出来的文档质量参差不齐，模型有时候会用不相关的内容编造答案。Self-RAG 和 CRAG 让系统能主动判断"这次检索结果是否可用"并在不够时自动纠正。需要有 LangGraph 基础。

1.1 基础 RAG 的根本局限

Self-RAG 自适应流程——通过特殊 Token 实现检索决策、相关性评估与质量自检

标准 RAG 系统存在一个根本性的设计缺陷：对检索结果照单全收。无论向量数据库返回的文档质量如何——相关或不相关，足够或不足，准确或过时——系统都会将其塞入 Prompt 让模型生成答案。

这导致两类典型问题：

检索结果不相关时： 模型要么根据无关文档产生幻觉，要么给出"根据文档无法回答"的无效回复，而实际上正确答案可能存在于未被检索到的文档中，或者可以通过其他途径获取（如网络搜索）。

问题本身不需要检索时： 对于"1+1等于几"这类常识性问题，系统仍然执行检索流程，浪费延迟和 token。

Self-RAG 和 CRAG 是两篇独立论文提出的解决方案，分别从不同角度让 RAG 系统具备"自我判断"和"自我纠错"的能力。

LangGraph 是 LangChain 框架下的一个组件，专门用于构建有状态的 Agent 工作流，通过"节点"（执行步骤）和"边"（流转逻辑）描述复杂的决策流程。下面的代码示例都基于 LangGraph 实现。

1.2 Self-RAG 架构

Self-RAG（Self-Reflective Retrieval-Augmented Generation，自反思检索增强生成，Asai et al. 2023）的核心思想是：训练模型生成特殊的反思 token（特殊标记词），让模型自主判断何时需要检索，以及检索结果的质量。

1.2.1 四种反思 token

Self-RAG 在模型的生成过程中引入四种特殊标记：

Retrieve token： 判断当前是否需要检索。模型在生成过程中自主决定是否调用检索器，而不是每次都强制检索。

ISREL token（Is Relevant）： 判断检索到的文档是否与问题相关。每个检索到的文档片段都会获得一个相关性评分。

ISSUP token（Is Supported）： 判断生成的答案片段是否有文档支撑。这个 token 在生成每个句子后出现，评估该句子是否基于检索文档。

ISUSE token（Is Useful）： 判断整体回答是否对用户有用。作为最终的质量控制。

1.2.2 用 LangGraph 实现 Self-RAG 流程

原版 Self-RAG 需要专门微调的模型。以下用 LangGraph 实现一个基于 Prompt Engineering 的近似版本，适合生产使用：

python

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.messages import HumanMessage, SystemMessage

class SelfRAGState(TypedDict):
    question: str
    needs_retrieval: bool
    retrieved_docs: list[str]
    docs_relevance: list[str]      # "relevant" 或 "not_relevant"
    answer_draft: str
    is_supported: bool
    is_useful: bool
    final_answer: str
    iteration: int

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
vectorstore = Chroma(
    collection_name="knowledge_base",
    embedding_function=OpenAIEmbeddings(),
)

# 节点1：判断是否需要检索
def assess_retrieval_need(state: SelfRAGState) -> Command:
    response = llm.invoke([
        SystemMessage(content="""判断以下问题是否需要从知识库检索信息才能准确回答。
对于常识性问题、数学计算、明确的事实（如"中国首都是哪里"）回答：no_retrieval
对于需要专业知识、最新信息、或具体领域知识的问题回答：needs_retrieval
只输出 no_retrieval 或 needs_retrieval，不加任何解释。"""),
        HumanMessage(content=state["question"])
    ])

    decision = response.content.strip().lower()
    needs = "needs_retrieval" in decision

    if needs:
        return Command(goto="retrieve", update={"needs_retrieval": True})
    else:
        return Command(goto="generate_direct", update={"needs_retrieval": False})

# 节点2a：直接回答（不需要检索）
def generate_direct(state: SelfRAGState) -> Command:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return Command(
        goto="assess_usefulness",
        update={"answer_draft": response.content, "is_supported": True}
    )

# 节点2b：检索文档
def retrieve(state: SelfRAGState) -> Command:
    results = vectorstore.similarity_search(state["question"], k=4)
    docs = [doc.page_content for doc in results]
    return Command(goto="assess_relevance", update={"retrieved_docs": docs})

# 节点3：评估检索文档的相关性
def assess_relevance(state: SelfRAGState) -> Command:
    relevance_list = []
    for doc in state["retrieved_docs"]:
        response = llm.invoke([
            SystemMessage(content="""判断以下文档片段是否与给定问题相关。
输出 relevant 或 not_relevant，不加任何解释。"""),
            HumanMessage(content=f"问题：{state['question']}\n\n文档片段：{doc[:300]}")
        ])
        relevance_list.append(response.content.strip().lower())

    # 过滤：只保留相关文档
    relevant_docs = [
        doc for doc, rel in zip(state["retrieved_docs"], relevance_list)
        if "relevant" in rel and "not" not in rel
    ]

    if not relevant_docs:
        # 没有相关文档：直接生成，标记不支持
        return Command(
            goto="generate_direct",
            update={"docs_relevance": relevance_list, "retrieved_docs": []}
        )

    return Command(
        goto="generate_with_docs",
        update={"docs_relevance": relevance_list, "retrieved_docs": relevant_docs}
    )

# 节点4：基于相关文档生成答案
def generate_with_docs(state: SelfRAGState) -> Command:
    context = "\n\n".join(state["retrieved_docs"])
    response = llm.invoke([
        SystemMessage(content="严格基于提供的文档内容回答问题，不添加文档中没有的信息。"),
        HumanMessage(content=f"文档：\n{context}\n\n问题：{state['question']}")
    ])
    return Command(goto="assess_support", update={"answer_draft": response.content})

# 节点5：评估答案是否有文档支撑
def assess_support(state: SelfRAGState) -> Command:
    if not state["retrieved_docs"]:
        return Command(goto="assess_usefulness", update={"is_supported": False})

    context = "\n\n".join(state["retrieved_docs"])
    response = llm.invoke([
        SystemMessage(content="""判断给定答案中的主要陈述是否都能从文档中找到支撑。
如果答案中存在文档未提及的关键陈述，输出 not_supported。
否则输出 supported。
只输出这两个词之一。"""),
        HumanMessage(content=f"文档：\n{context}\n\n答案：{state['answer_draft']}")
    ])

    supported = "not_supported" not in response.content.lower()
    return Command(goto="assess_usefulness", update={"is_supported": supported})

# 节点6：评估答案是否有用
def assess_usefulness(state: SelfRAGState) -> Command:
    iteration = state.get("iteration", 0)

    response = llm.invoke([
        SystemMessage(content="""判断以下答案是否直接、有效地回答了问题。
如果答案有用，输出 useful。
如果答案无法回答问题或没有实质内容，输出 not_useful。
只输出这两个词之一。"""),
        HumanMessage(content=f"问题：{state['question']}\n\n答案：{state['answer_draft']}")
    ])

    is_useful = "not_useful" not in response.content.lower()

    # 如果无用且还有重试次数，尝试重新生成
    if not is_useful and iteration < 2:
        return Command(
            goto="retrieve",
            update={"is_useful": False, "iteration": iteration + 1}
        )

    return Command(
        goto=END,
        update={"is_useful": is_useful, "final_answer": state["answer_draft"]}
    )

# 构建图
builder = StateGraph(SelfRAGState)
for name, fn in [
    ("assess_retrieval_need", assess_retrieval_need),
    ("generate_direct", generate_direct),
    ("retrieve", retrieve),
    ("assess_relevance", assess_relevance),
    ("generate_with_docs", generate_with_docs),
    ("assess_support", assess_support),
    ("assess_usefulness", assess_usefulness),
]:
    builder.add_node(name, fn)

builder.add_edge(START, "assess_retrieval_need")
self_rag_graph = builder.compile()

1.3 Corrective RAG（CRAG）架构

CRAG（Corrective RAG，纠正式检索增强生成，Yan et al. 2024）的关注点与 Self-RAG 不同：它不修改模型训练，而是在检索和生成之间加入一个评估-纠正环节，根据检索质量动态选择处理策略。

1.3.1 三种处理策略

CRAG 对每个检索结果评分，然后根据得分选择策略：

策略一：直接使用（Correct） — 检索质量高，直接用检索结果生成答案。

策略二：补充搜索（Ambiguous） — 检索质量中等，结合向量检索结果和网络搜索结果共同生成。

策略三：完全替换（Incorrect） — 检索质量差，放弃向量检索结果，完全使用网络搜索。

1.3.2 用 LangGraph 实现 CRAG

python

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.messages import HumanMessage, SystemMessage

class CRAGState(TypedDict):
    question: str
    retrieved_docs: list[str]
    retrieval_grade: Literal["correct", "ambiguous", "incorrect"]
    web_results: list[str]
    final_context: list[str]
    answer: str

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
vectorstore = Chroma(
    collection_name="knowledge_base",
    embedding_function=OpenAIEmbeddings(),
)
web_search = TavilySearchResults(max_results=3)  # 需要 TAVILY_API_KEY

# 节点1：从向量库检索
def retrieve_from_vectorstore(state: CRAGState) -> dict:
    results = vectorstore.similarity_search(state["question"], k=4)
    docs = [doc.page_content for doc in results]
    return {"retrieved_docs": docs}

# 节点2：评估检索质量（核心节点）
def grade_retrieval(state: CRAGState) -> Command:
    docs_text = "\n\n---\n\n".join(state["retrieved_docs"])

    response = llm.invoke([
        SystemMessage(content="""评估检索到的文档与问题的相关性。
根据以下标准给出评分：
- correct：文档高度相关，包含回答问题所需的完整信息
- ambiguous：文档部分相关，可以提供一些信息但不够完整，需要补充
- incorrect：文档与问题不相关，无法用于回答问题

只输出 correct、ambiguous 或 incorrect 三者之一。"""),
        HumanMessage(content=f"问题：{state['question']}\n\n检索文档：\n{docs_text[:2000]}")
    ])

    grade = response.content.strip().lower()
    if grade not in ["correct", "ambiguous", "incorrect"]:
        grade = "ambiguous"  # 默认保守策略

    if grade == "correct":
        # 直接进入生成，不需要网络搜索
        return Command(
            goto="generate_answer",
            update={
                "retrieval_grade": "correct",
                "final_context": state["retrieved_docs"]
            }
        )
    elif grade == "ambiguous":
        # 补充网络搜索
        return Command(
            goto="web_search",
            update={"retrieval_grade": "ambiguous"}
        )
    else:
        # 完全使用网络搜索
        return Command(
            goto="web_search",
            update={"retrieval_grade": "incorrect", "retrieved_docs": []}
        )

# 节点3：网络搜索（Fallback）
def web_search_node(state: CRAGState) -> Command:
    try:
        results = web_search.invoke(state["question"])
        web_docs = [r["content"] for r in results if isinstance(r, dict) and "content" in r]
    except Exception as e:
        web_docs = [f"网络搜索失败：{str(e)}"]

    if state["retrieval_grade"] == "ambiguous":
        # 合并：向量检索 + 网络搜索
        combined = state["retrieved_docs"] + web_docs
    else:
        # 只用网络搜索
        combined = web_docs

    return Command(
        goto="generate_answer",
        update={"web_results": web_docs, "final_context": combined}
    )

# 节点4：生成最终答案
def generate_answer(state: CRAGState) -> dict:
    context = "\n\n".join(state["final_context"])
    response = llm.invoke([
        SystemMessage(content="基于提供的文档内容准确回答问题，如果文档不足以回答，请如实说明。"),
        HumanMessage(content=f"文档：\n{context}\n\n问题：{state['question']}")
    ])
    return {"answer": response.content}

# 构建图
builder = StateGraph(CRAGState)
builder.add_node("retrieve_from_vectorstore", retrieve_from_vectorstore)
builder.add_node("grade_retrieval", grade_retrieval)
builder.add_node("web_search_node", web_search_node)
builder.add_node("generate_answer", generate_answer)

builder.add_edge(START, "retrieve_from_vectorstore")
builder.add_edge("retrieve_from_vectorstore", "grade_retrieval")
builder.add_edge("generate_answer", END)

crag_graph = builder.compile()

# 使用示例
result = crag_graph.invoke({
    "question": "LangGraph 0.2 版本有哪些新特性？",
    "retrieved_docs": [],
    "retrieval_grade": "correct",
    "web_results": [],
    "final_context": [],
    "answer": "",
})
print(result["answer"])

1.3.3 CRAG 决策流程图

1.4 Self-RAG 与 CRAG 的对比

对比维度	Self-RAG	CRAG
核心机制	模型内生成反思 token，自主判断	外部评估器评估检索质量
模型要求	原版需要微调；近似实现用 Prompt	任何 LLM 均可使用
是否需要检索	模型自主判断（Retrieve token）	总是先检索，评估后决定
检索失败处理	重试检索	切换到网络搜索
外部工具依赖	无	需要 Web Search API（如 Tavily）
答案质量控制	ISSUP + ISUSE 双重检验	通过上下文质量间接控制
实现复杂度	较高（多个反思环节）	中等（评估 + 路由）
适用场景	对答案忠实度要求极高；不希望依赖外部搜索	知识库覆盖不完整；允许联网补充
延迟影响	多次 LLM 调用，延迟较高	额外一次评估调用 + 可能的网络搜索

选型建议：

选 CRAG 的场景：知识库可能覆盖不全，或者需要回答最新信息（新闻、文档更新）；对实现复杂度敏感；不需要严格控制是否联网。

选 Self-RAG（近似实现）的场景：私有环境，不能访问互联网；对答案忠实度要求极高；需要系统能够主动判断"我不知道"而不是给出错误答案。

两者也可以组合：先用 CRAG 的评估机制判断检索质量，再用 Self-RAG 的 ISSUP 机制验证生成答案的支撑度。