如何利用 LangGraph 构建具有自我修正能力的 RAG 工作流

传统的 RAG 只要检索结果偏差，后面的生成环节就彻底跑偏，这就是典型的“垃圾进，垃圾出”。我最近在尝试用 LangGraph 把 RAG 改造为一个带循环的图结构，核心逻辑就是给 LLM 增加一个“审核员”角色，让它在输出前先自我检查检索内容的相关性和回答的准确性。

最关键的配置在于定义 State 和 Conditional Edges。不要把整个流程写成线性的 Chain，要把它定义成一个状态机。

首先，定义状态对象，确保它能记录当前的检索文档和评分结果：

from typing import TypedDict, List

class GraphState(TypedDict):
    question: str
    generation: str
    documents: List[str]
    iteration_count: int

在构建图的时候，我设置了一个 grade_documents 节点。这个节点不负责回答问题，只负责打分。如果检索到的文档与问题无关，它会触发一个 rewrite_query 节点，利用 LLM 把原始问题重新表述，再次进入检索环节。

这里有个踩坑点：如果不设置 iteration_count 的上限，一旦 LLM 陷入“检索-不相关-重写-再检索”的死循环，你的 Token 费用会瞬间爆炸。建议在条件边里强制限制 3 次迭代。

实现自我修正的条件路由逻辑大致如下：

def decide_to_generate(state):
    # 检查文档评分，如果相关文档数少于阈值且未达到最大迭代次数，则重写查询
    if len([d for d in state['documents'] if d.score > 0.7]) > 0:
        return "generate"
    elif state['iteration_count'] < 3:
        return "rewrite_query"
    else:
        return "generate" # 强制输出，避免死循环

为了提升效率，我在提示词里采用了“少样本（Few-Shot）”策略，明确告诉审核节点什么样的文档叫“不相关”。比如，如果用户问的是具体数值而文档里只有概括性描述，直接判定为 irrelevant。

具体的图拓扑配置是：Start -> Retrieve -> Grade -> (相关则 Generate / 不相关则 Rewrite -> Retrieve)。

这种架构比简单的 RAG 强在它具备了“反思”能力。实测下来，针对一些长尾问题的回答准确率提升了约 30%，因为它不再盲目信任向量数据库的一次性检索结果，而是通过多次尝试直到找到匹配信息。