自纠正 RAG (CRAG)

自适应 RAG 解决了"要不要检索"和"文档相不相关"的问题，但遇到"检索结果整体不行"时只会改写问题再检索一遍，手段比较单一。本篇介绍 Corrective RAG（CRAG）——它把检索结果分三档评估，对差的检索果断切换到 web 搜索兜底，对好的文档做"知识精炼"去噪，是一套更工程化的纠错方案。

一、CRAG 论文核心思想

CRAG 出自 2024 年论文 Corrective Retrieval Augmented Generation，核心是在"检索"和"生成"之间插一个检索评估器（Retrieval Evaluator），对检索结果打分，然后分三档处理：

档位	含义	处理方式
Correct	检索结果可靠、相关	做知识精炼（去噪压缩）后送生成
Incorrect	检索结果基本没用	丢弃，转向 web 搜索兜底
Ambiguous	不确定 / 部分相关	既精炼本地文档，也补 web 搜索，合并

关键点：

三档而非二档：相比自适应 RAG 的"相关/不相关"二分，CRAG 多了"模糊"档，更贴近现实——很多检索就是"沾边但不全"。
Web 搜索兜底：本地知识库覆盖不到时，主动换数据源，而不是死磕。
知识精炼：即使是"正确"文档，也先去噪（剔除无关句子）再喂给 LLM，提升信号密度。

对 Java 同学：这就像一个带"降级策略"的服务——主库数据好用就精炼后用，主库不行就切备用库（web），不确定就两边都取再合并。

二、整体流程图

mermaid

flowchart TD
    S([START]) --> R[retrieve 检索本地库]
    R --> EV[grade_retrieval 检索评估器]
    EV -- correct --> REF[refine_knowledge 知识精炼]
    EV -- ambiguous --> REF
    EV -- ambiguous --> WEB[web_search web搜索]
    EV -- incorrect --> WEB
    REF --> G[generate 生成]
    WEB --> G
    G --> E([END])

评估器输出三档，路由到 refine_knowledge 和/或 web_search，最后合并送生成。

三、状态设计

CRAG 状态在自适应 RAG 基础上多了"评估档位""web 结果""精炼后知识"。

python

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

class CRAGState(TypedDict):
    question: str
    documents: List[Document]        # 本地检索结果
    retrieval_grade: Literal["correct", "ambiguous", "incorrect"]  # 评估档位
    web_results: List[Document]      # web 搜索兜底结果
    refined_knowledge: str           # 精炼后的最终知识文本
    answer: str
    retry_count: int

四、检索评估器

评估器是 CRAG 的灵魂。它对整体检索结果打一个三档分（而不是逐条文档打分，这点和自适应 RAG 不同）。

python

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class RetrievalGrade(BaseModel):
    grade: str = Field(description="检索结果档位：correct / ambiguous / incorrect")

grade_llm = llm.with_structured_output(RetrievalGrade)

def grade_retrieval(state: CRAGState) -> dict:
    """对整体检索结果打三档分"""
    question = state["question"]
    docs = state["documents"]
    context = "\n".join(d.page_content for d in docs)
    prompt = (
        "判断下面检索结果对回答问题的整体帮助程度，给出三档评分：\n"
        "- correct：资料能直接回答问题\n"
        "- ambiguous：资料部分相关，不能完整回答\n"
        "- incorrect：资料与问题基本无关\n\n"
        f"问题：{question}\n\n检索资料：\n{context}\n\n只输出档位词。"
    )
    result = grade_llm.invoke(prompt)
    grade = result.grade.strip().lower()
    if grade not in ("correct", "ambiguous", "incorrect"):
        grade = "ambiguous"   # 兜底：模型乱答时按模糊处理
    return {"retrieval_grade": grade}

五、知识精炼与 Web 搜索

1. 知识精炼节点

对 correct/ambiguous 档的本地文档去噪：把每段拆成句子，只保留与问题相关的句子。

python

def refine_knowledge(state: CRAGState) -> dict:
    """对本地文档做知识精炼：剔除无关句子"""
    question = state["question"]
    docs = state["documents"]
    keep_sentences = []
    for d in docs:
        # 简单按句号切，实际可用更智能的分句器
        sentences = [s.strip() for s in d.page_content.replace("。", ".\n").split("\n") if s.strip()]
        for s in sentences:
            # 用 LLM 快速判断句子是否相关
            verdict = llm.invoke(
                f"这句话与问题相关吗？只回 yes 或 no。\n问题：{question}\n句子：{s}"
            ).content.strip().lower()
            if verdict.startswith("yes"):
                keep_sentences.append(s)
    # 合并已有 web 结果（如果有），凑成最终知识
    web_text = "\n".join(d.page_content for d in state.get("web_results", []))
    refined = "\n".join(keep_sentences) + ("\n" + web_text if web_text else "")
    return {"refined_knowledge": refined.strip()}

提示：这里为演示清晰用了逐句 LLM 判断，生产里开销大。可以用 embedding 相似度筛句子，或一次性把所有句子喂给 LLM 让它返回相关索引。

2. Web 搜索节点（Mock 版）

真实场景接 Tavily/Serper/Bing API。这里用一个 Mock 函数模拟，保证示例能离线跑。

python

def web_search(state: CRAGState) -> dict:
    """web 搜索兜底（此处 Mock，真实项目替换为搜索 API）"""
    question = state["question"]
    # ===== Mock：返回两条与问题相关的伪结果 =====
    mock_results = [
        Document(page_content=f"[web] 关于「{question}」的权威说明：通常需要结合具体上下文判断。"),
        Document(page_content=f"[web] 网络资料补充：{question} 涉及的常见结论可参考相关标准文档。"),
    ]
    # 真实写法示例（需 pip install langchain-tavily 并配置 API key）：
    # from langchain_tavily import TavilySearch
    # results = TavilySearch(max_results=3).invoke({"query": question})
    # mock_results = [Document(page_content=r["content"]) for r in results["results"]]
    return {"web_results": mock_results}

六、条件边路由

评估器之后是 CRAG 最有意思的分叉：三档路由到不同节点组合。LangGraph 的条件边只能返回"下一个节点名"，所以 ambiguous 档要"同时"走精炼和 web 搜索，有两种实现：

方案 A（串行）：ambiguous → web_search → refine_knowledge，让 web 结果也参与精炼。
方案 B（并行）：用 Send API 并行触发两节点，再合并。

为降低新手门槛，本篇用方案 A 串行，效果稳、好理解。Send 并行方案见进阶特性·动态图与Send。

python

def route_after_grade(state: CRAGState) -> str:
    g = state["retrieval_grade"]
    if g == "correct":
        return "refine_knowledge"
    elif g == "incorrect":
        return "web_search"
    else:  # ambiguous：先 web 兜底，再一起精炼
        return "web_search"

ambiguous 和 incorrect 都先走 web_search，但 ambiguous 接着还会进 refine_knowledge（精炼本地 + web），而 incorrect 因为本地文档没用，精炼时基本只剩 web 内容——靠 refine_knowledge 里的 web_text 合并自然实现差异。

七、组装图

python

from langgraph.graph import StateGraph, START, END

gb = StateGraph(CRAGState)
gb.add_node("retrieve", retrieve)               # 复用基础 RAG 检索
gb.add_node("grade_retrieval", grade_retrieval)
gb.add_node("refine_knowledge", refine_knowledge)
gb.add_node("web_search", web_search)
gb.add_node("generate", generate)

gb.add_edge(START, "retrieve")
gb.add_edge("retrieve", "grade_retrieval")
gb.add_conditional_edges("grade_retrieval", route_after_grade, {
    "refine_knowledge": "refine_knowledge",
    "web_search": "web_search",
})
# web 搜完进精炼（合并本地+web）
gb.add_edge("web_search", "refine_knowledge")
gb.add_edge("refine_knowledge", "generate")
gb.add_edge("generate", END)

crag_app = gb.compile()

完整流程图（含三档路由）：

mermaid

flowchart TD
    S([START]) --> retrieve
    retrieve --> grade_retrieval
    grade_retrieval -- correct --> refine_knowledge
    grade_retrieval -- ambiguous --> web_search
    grade_retrieval -- incorrect --> web_search
    web_search --> refine_knowledge
    refine_knowledge --> generate
    generate --> E([END])

八、完整可运行示例

python

import os
from typing import TypedDict, List, Literal
from pydantic import BaseModel, Field

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.graph import StateGraph, START, END

# ---- 知识库：故意只放少量内容，便于触发 incorrect 走 web ----
docs = [Document(page_content=t) for t in [
    "LangGraph 用图建模流程，节点是函数。",
    "条件边支持根据状态做分支。",
]]
vs = InMemoryVectorStore(OpenAIEmbeddings(model="text-embedding-3-small"))
vs.add_documents(docs)
retriever = vs.as_retriever(search_kwargs={"k": 2})

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class RG(BaseModel):
    grade: str

# ---- 状态 ----
class S(TypedDict):
    question: str
    documents: List[Document]
    retrieval_grade: str
    web_results: List[Document]
    refined_knowledge: str
    answer: str

# ---- 节点 ----
def retrieve(state):
    return {"documents": retriever.invoke(state["question"])}

def grade_retrieval(state):
    ctx = "\n".join(d.page_content for d in state["documents"])
    g = llm.with_structured_output(RG).invoke(
        "给检索结果三档评分(correct/ambiguous/incorrect)：\n"
        f"问题：{state['question']}\n资料：{ctx}\n只回档位词。")
    grade = g.grade.strip().lower()
    if grade not in ("correct","ambiguous","incorrect"):
        grade = "ambiguous"
    print(f"[eval] 档位={grade}")
    return {"retrieval_grade": grade}

def web_search(state):
    # Mock web 搜索
    return {"web_results": [
        Document(page_content=f"[web] {state['question']}：来自网络的权威解释。"),
    ]}

def refine_knowledge(state):
    keep = []
    for d in state["documents"]:
        for s in d.page_content.replace("。",".\n").split("\n"):
            s = s.strip()
            if not s: continue
            v = llm.invoke(f"句子与问题相关吗只回yes/no：\n问题：{state['question']}\n句子：{s}").content.lower()
            if v.startswith("yes"): keep.append(s)
    web = "\n".join(d.page_content for d in state.get("web_results", []))
    return {"refined_knowledge": "\n".join(keep) + ("\n"+web if web else "")}

def generate(state):
    a = llm.invoke(f"只根据资料回答，不足就说不知道。\n资料：{state['refined_knowledge']}\n问题：{state['question']}")
    return {"answer": a.content}

# ---- 路由 ----
def route(state):
    g = state["retrieval_grade"]
    if g == "correct": return "refine_knowledge"
    return "web_search"   # ambiguous / incorrect 都先 web

# ---- 建图 ----
gb = StateGraph(S)
for n,f in [("retrieve",retrieve),("grade_retrieval",grade_retrieval),
            ("web_search",web_search),("refine_knowledge",refine_knowledge),
            ("generate",generate)]:
    gb.add_node(n,f)
gb.add_edge(START,"retrieve")
gb.add_edge("retrieve","grade_retrieval")
gb.add_conditional_edges("grade_retrieval",route,{
    "refine_knowledge":"refine_knowledge","web_search":"web_search"})
gb.add_edge("web_search","refine_knowledge")
gb.add_edge("refine_knowledge","generate")
gb.add_edge("generate",END)
app = gb.compile()

if __name__ == "__main__":
    # 第一问命中本地 → correct；第二问本地答不了 → incorrect → web 兜底
    for q in ["条件边是什么？", "2024年诺贝尔文学奖得主是谁？"]:
        r = app.invoke({"question": q})
        print(q, "=>", r["answer"], "| 档位:", r["retrieval_grade"])

第二个问题本地知识库答不了，评估器判 incorrect，自动走 web 搜索兜底，最终回答里会带上 [web] 来源的内容。

九、与普通 RAG / 自适应 RAG 对比

维度	基础 RAG	自适应 RAG	CRAG
评估粒度	不评估	逐条文档相关/不相关	整体三档
失败兜底	无	改写问题重检索	切 web 搜索
文档处理	原样塞 prompt	仅按相关性筛选	额外做知识精炼去噪
复杂度	低	中	高
适用场景	demo / 知识库质量高	通用问答	严肃问答、知识库覆盖不全

简单选择：先基础 RAG 跑通 → 加查询分析+文档评分变自适应 → 再加 web 兜底+知识精炼变 CRAG。不要一上来就上 CRAG，调试成本很高。

十、常见踩坑

1. 评估器阈值难调

三档边界很模糊，"ambiguous"和"correct"之间尤其难。建议：

在 prompt 里给每档配一个示例，减少模型主观飘移。
把 grade 落库统计分布，如果 90% 都判 correct，说明阈值太松，CRAG 等于白加。
实在难分时，倾向于判 ambiguous（两边都取），比误判 incorrect 丢掉本地好文档更安全。

2. Web 搜索依赖

CRAG 的命门是 web 搜索质量。坑点：

API key 与配额：Tavily/Serper 都要付费，高频调用烧钱。生产里要加缓存和限流。
国内可访问性：部分搜索 API 在国内需代理。可换成支持国内的搜索源或自建爬虫。
结果质量参差：web 内容可能比本地库还脏，建议对 web 结果也做一遍相关性过滤再进精炼。

3. 知识精炼把信息精炼没了

逐句过滤可能误删关键转折句（比如"但是"开头的句子）。缓解：

不要让 LLM 只看单句，给它前后相邻句作为上下文。
设置"宁留勿删"：不确定就保留，靠生成节点的 prompt 抑制幻觉。

4. 状态里 web_results 与 documents 容易混

精炼节点要同时读 documents（本地）和 web_results（网络），新手常只读一个导致另一边丢失。建议在 refine_knowledge 开头 print 一遍两边长度，调试期就能发现问题。

十一、小结

CRAG = 检索评估器（三档）+ 知识精炼 + web 搜索兜底，是一套比自适应 RAG 更工程化的纠错方案。
三档路由：correct 走精炼，incorrect 切 web，ambiguous 两边都取。
知识精炼提升信号密度，web 搜索解决本地库覆盖不足。
复杂度高，建议在基础 RAG / 自适应 RAG 跑稳后再引入，并重点调评估器阈值。

下一篇检索增强智能体换个思路：不再固定流程，而是把检索作为"工具"交给智能体自己决定怎么用。

自纠正 RAG (CRAG) ​

一、CRAG 论文核心思想 ​

二、整体流程图 ​

三、状态设计 ​

四、检索评估器 ​

五、知识精炼与 Web 搜索 ​

1. 知识精炼节点 ​

2. Web 搜索节点（Mock 版） ​

六、条件边路由 ​

七、组装图 ​

八、完整可运行示例 ​

九、与普通 RAG / 自适应 RAG 对比 ​

十、常见踩坑 ​

1. 评估器阈值难调 ​

2. Web 搜索依赖 ​

3. 知识精炼把信息精炼没了 ​

4. 状态里 web_results 与 documents 容易混 ​

十一、小结 ​

自纠正 RAG (CRAG)

一、CRAG 论文核心思想

二、整体流程图

三、状态设计

四、检索评估器

五、知识精炼与 Web 搜索

1. 知识精炼节点

2. Web 搜索节点（Mock 版）

六、条件边路由

七、组装图

八、完整可运行示例

九、与普通 RAG / 自适应 RAG 对比

十、常见踩坑

1. 评估器阈值难调

2. Web 搜索依赖

3. 知识精炼把信息精炼没了

4. 状态里 web_results 与 documents 容易混

十一、小结