用 LangGraph 编排 RAG 动态检索工作流

传统 RAG 管道在处理“先查A再查B”的多跳问题时，检索路径固化导致中间环节上下文缺失，命中率难以提升。LangGraph 将检索、判断、生成建模为状态机节点，通过条件路由动态决策下一步动作。例如，当首轮检索结果相关性低于阈值时，系统可自动触发查询改写或扩大检索范围，而非直接生成低质答案。这种机制在金融风控、法律尽调等复杂问答场景中，可将多跳问题检索命中率提升 20-35%。

一、状态机核心抽象

LangGraph 基于 StateGraph 构建工作流，状态（State）是所有节点共享的字典结构。定义状态时，需明确关键字段：用户问题、历史检索结果、当前推理步骤、下一步动作标识。状态即数据容器，节点即逻辑单元，边即流转条件。生产环境中，状态设计需考虑扩展性，例如预留 trace_id 字段用于全链路追踪，或 error_count 字段记录节点失败次数以触发降级策略。

# 定义 RAG 状态模式
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, START, END

class RAGState(TypedDict):
    question: str                    # 用户原始问题
    retrieved_docs: List[str]        # 检索到的文档文本列表
    reasoning: str                   # LLM 推理过程记录
    next_action: str                 # 路由函数返回的下一节点名
    query_rewrite: str               # 改写后的查询语句（可选）
    docs_score: List[float]          # 向量相似度分数

状态可继承 TypedDict 实现类型检查，生产环境中建议扩展 docs_score、query_rewrite 等字段以支持重排序与查询扩展。若使用 LangGraph 的 MessageGraph 变体，状态还可携带消息历史，便于调试多轮对话场景下的上下文传递问题。

二、节点职责划分

将 RAG 流程拆解为四个核心节点：查询改写、向量检索、质量评估、答案生成。每个节点接收状态，执行单一职责，输出更新后的状态。节点间通过状态传递数据，避免硬编码调用链。例如，向量检索节点不应直接调用答案生成逻辑，而是将结果写入状态后由路由函数决定流转方向。这种解耦设计使得单个节点可独立测试与替换，如将向量数据库从 Milvus 切换为 Weaviate 时，仅需修改检索节点内部实现，其余节点无需改动。

# 示例：向量检索节点
def vector_search(state: RAGState) -> RAGState:
    # 优先使用改写后的查询，否则回退到原始问题
    query = state.get("query_rewrite") or state["question"]
    try:
        # 调用向量数据库，返回 Top-K 文档及相似度分数
        docs = vector_db.search(query, top_k=5)
        return {
            "retrieved_docs": [doc.text for doc in docs],
            "docs_score": [doc.score for doc in docs]
        }
    except Exception as e:
        # 检索失败时降级为关键词搜索，避免整条工作流中断
        fallback_docs = keyword_search(query, top_k=3)
        return {
            "retrieved_docs": [doc.text for doc in fallback_docs],
            "docs_score": [0.5] * len(fallback_docs),  # 降级标记
            "reasoning": f"向量检索失败，降级为关键词搜索: {str(e)}"
        }

节点函数必须返回状态字典的子集，LangGraph 自动合并至全局状态。注意避免副作用，保持节点幂等性以便重试机制生效。例如，查询改写节点不应修改传入的 state 对象，而应返回新字段值。若节点内部调用外部 LLM，建议使用带超时的异步客户端，避免单个节点阻塞整条工作流。

三、条件路由实现

3.1 动态分支判断

路由函数接收状态，返回下一个节点名称字符串。LangGraph 根据返回值决定流转方向，支持动态选择检索策略。路由逻辑需配合 add_conditional_edges 方法配置，该方法的参数为起点节点、路由函数、以及字典形式的映射关系（键为路由返回值，值为目标节点名或 END）。若路由返回的节点名未在映射中声明，工作流将抛出异常，因此需在开发阶段覆盖所有可能的返回值。

# 示例：质量评估路由
def evaluate_quality(state: RAGState) -> str:
    docs = state["retrieved_docs"]
    scores = state.get("docs_score", [])
    if len(docs) < 3:
        return "query_rewrite"   # 结果不足，重新改写查询
    if scores and max(scores) < 0.6:
        return "vector_search"   # 相关性低，重新检索
    if len(docs) > 0:
        return "answer_gen"      # 质量达标，生成答案
    return "query_rewrite"       # 无结果，触发改写

# 配置条件边
graph.add_conditional_edges(
    "quality_eval",           # 起点节点
    evaluate_quality,        # 路由函数
    {
        "query_rewrite": "query_rewrite",
        "vector_search": "vector_search",
        "answer_gen": "answer_gen"
    }
)

3.2 循环与终止

通过 add_conditional_edges 建立条件边，当路由返回特定节点名时跳转。需设置最大迭代次数防止无限循环，生产建议限制在 5-8 次内。可在状态中维护 loop_count 字段，路由函数检查该值，超过阈值时强制返回 "answer_gen" 并标记为低置信度结果。循环终止条件还应包括用户取消请求、超时中断等外部信号，确保工作流可被安全终止。

路由逻辑可嵌入 LLM 判断，如使用小模型评估检索结果是否包含关键实体，比纯阈值规则更适应复杂语义场景。参考 LangGraph 实战：用状态图编排多智能体协作 中的智能体通信模式，可将质量评估节点扩展为多智能体投票机制，提升判断鲁棒性。

四、ReAct 循环优化

ReAct 模式将推理（Reasoning）与行动（Acting）交替执行。在 RAG 场景中，每轮先让 LLM 分析当前检索结果，决定是否需要补充检索或调整查询，再执行行动节点。循环直到 LLM 判断信息充足或达到最大迭代次数。ReAct 循环适合多跳问题，但会增加 LLM 调用次数，建议为每轮推理设置 token 上限，防止长上下文导致成本失控。

# 示例：ReAct 推理节点
def react_reason(state: RAGState) -> RAGState:
    docs_summary = "\n".join(state["retrieved_docs"][:3])  # 仅取前3篇避免上下文过长
    prompt = f"""问题: {state['question']}
已检索结果摘要:
{docs_summary}

请分析当前信息是否足够回答问题。若不足，请指出缺失的子问题并建议改写方向。
输出格式：需要补充检索|需要生成答案"""
    llm_response = llm.invoke(prompt)
    # 解析 LLM 输出，决定 next_action
    if "需要生成答案" in llm_response:
        next_action = "answer_gen"
    else:
        next_action = "query_rewrite"
    return {
        "reasoning": llm_response,
        "next_action": next_action,
        "query_rewrite": extract_query_rewrite(llm_response)  # 提取改写建议
    }

ReAct 循环会增加 LLM 调用次数，建议结合 One API 实战：自建大模型 API 网关聚合多模型 管理多模型路由，将推理任务分配给低成本小模型，仅将最终生成任务交给大模型，降低推理成本 30% 以上。

五、检索质量评估机制

评估维度包括：相关性（向量相似度）、完整性（是否覆盖子问题）、冗余度（文档重复率）。使用专用评估模型对 Top-K 文档打分，分数低于阈值则触发查询改写或扩大检索范围。评估节点应在答案生成前执行，作为质量门禁，避免将低质结果直接送入 LLM 导致幻觉放大。生产环境中，评估模型可选用 bge-reranker 或开源 LLM 零样本判断，前者速度快但泛化性弱，后者效果好但成本高。

评估结果存入状态字段，供路由函数读取。避免在生成节点中直接信任检索结果，必须经过质量门禁。若评估模型本身出现故障，应回退到纯阈值规则，确保工作流不因评估环节崩溃而中断。评估节点的日志需记录原始文档 ID、分数明细，便于事后 bad case 分析。

六、生产部署要点

工作流编译后生成 CompiledGraph，支持 checkpoint 持久化，断点续跑。设置节点级超时与重试策略，关键节点（如向量检索）建议设置 10 秒超时，失败后降级为关键词搜索。checkpoint 可存储于 Redis 或 Postgres，确保服务重启后能恢复未完成的请求。对于长耗时 ReAct 循环，建议将中间状态定期写入 checkpoint，避免 LLM 调用超时导致整条链路丢失。

监控指标包括：平均循环次数、各节点耗时、检索命中率、降级触发率。使用 OpenTelemetry 埋点，对接 Open WebUI 实战：自建本地大模型可视化平台 中的日志面板，实现全链路追踪。在 Gradio 演示界面中，可实时展示状态流转过程，帮助用户理解系统内部决策逻辑，提升可信度。

七、常见反模式

避免在节点中修改全局变量，状态是唯一数据传递通道。全局变量导致测试困难且并发不安全。避免循环内调用外部 LLM 而无缓存，重复查询浪费资源；建议使用语义缓存，对相似查询复用已有 LLM 输出。避免路由逻辑过于复杂，单一路由函数分支不超过 4 个，超出时应拆分为多个路由节点。避免在状态中存储大型文档全文，应仅存文档 ID 与摘要，通过 ID 回查原文，减少 checkpoint 存储压力。

八、小结

LangGraph 将 RAG 从线性管道升级为状态机，条件路由与 ReAct 循环使系统能自适应复杂查询。关键在于状态设计清晰、节点职责单一、质量评估前置。生产环境中结合 checkpoint 与监控，可将多跳问题检索命中率提升 20-35%。配合 本地大模型量化部署：GGUF 与 llama.cpp 调优实战 方案，可在边缘侧运行轻量评估模型，降低延迟至 200ms 以内，适用于离线或隐私敏感场景。使用 Aider 等结对编程工具可快速迭代节点实现，加速开发周期。