第 7 课：LangGraph 图编排 —— 把零件装成机器

本节目标：理解图编排解决什么问题、掌握 RagState 设计、10 个节点 + 2 个条件边的拓扑、修正循环的安全阀、以及三模式兼容的设计技巧。

学完本课后，推荐阅读 附07：Agent Tool 调用机制深度 —— @Tool 注解到 JSON Schema 全链路、ReAct 循环全景、Tool 设计四原则、Self-RAG 反思机制。

前 6 课认识了所有"零件"：rewrite、retrieve、generate、verify、verifier、revise。这一课用 LangGraph 把它们连成一张图——数据自动流转、分支自动决策、修正自动循环。

1. 为什么需要"图"

如果把所有逻辑写成一个超长函数：分支嵌套 3 层、看不出执行到哪一步、想跳过 verify 只跑 generate 要复制粘贴。

LangGraph 解法：每个步骤 = 节点，分支逻辑 = 条件边，循环 = 回边。图的形状就是业务逻辑，一目了然。同一张图靠 state 开关支持三种模式，零代码修改。

2. RagState 设计

python

class RagState(TypedDict, total=False):
    # 输入
    question: str
    top_k: int
    retrieval_mode: Literal["vector", "keyword", "hybrid"]
    enable_verify: bool
    enable_verifier: bool
    max_revisions: int

    # 中间产物
    rewritten_query: str
    hits: list[RetrievedChunkDict]
    context: str

    # 输出
    answer: str
    citations: list[CitationDict]
    finish_reason: Literal[...]

    # 修正循环
    revision_count: int
    revision_hints: str
    verifier: dict[str, Any]

    # 元信息（自定义合并函数，浅合并而非覆盖）
    meta: Annotated[dict[str, Any], _merge_meta]

三个设计要点：

total=False→ 所有字段可选。每个节点只返回自己写的字段，LangGraph 自动 merge 进 state。接力赛模式——每人只跑自己那一棒
meta 用自定义合并器→ rewrite_node 写 rewrite_elapsed，retrieve_node 写 retrieve_elapsed → 合并后完整保留所有计时。命名空间防冲突
不存大对象→ State 会被序列化传递，塞原始 response 进内存爆炸

3. 图拓扑：10 节点 + 2 条件边

Architecture Flow

START → rewrite → retrieve → judge
                                │
              ┌─────────────────┼──────────────────┐
              ▼                 ▼                  ▼
           [miss]          [generate]          [verify]
              │                 │                  │
              ▼                 ▼                  ▼
          fallback          generate          verify_node
              │                 │                  │
              ▼                 ▼                  ▼
           format            format          run_verifier
              │                 │                  │
              ▼                 ▼         ┌────────┼────────┬──────────┐
             END               END       ▼        ▼        ▼          ▼
                                      [pass]  [revise] [insuff.] [conflict]
                                        │        │        │          │
                                        ▼        ▼        ▼          ▼
                                     format  prepare_  fallback   conflict
                                        │    revise       │          │
                                        ▼       │         ▼          ▼
                                       END      │      format      format
                                                ▼        │          │
                                              verify     ▼          ▼
                                            (循环)      END        END

构建代码 30 行：

Code

def build_rag_graph():
    builder = StateGraph(RagState)

    # 10 个节点
    for name in ["rewrite", "retrieve", "judge", "generate", "verify",
                  "run_verifier", "prepare_revise", "conflict", "fallback", "format"]:
        builder.add_node(name, globals()[f"{name}_node"])

    # 主链
    builder.add_edge(START, "rewrite")
    builder.add_edge("rewrite", "retrieve")
    builder.add_edge("retrieve", "judge")
    builder.add_conditional_edges("judge", route_after_judge,
        {"generate": "generate", "verify": "verify", "miss": "fallback"})

    # 验证链 + 修正回边
    builder.add_edge("verify", "run_verifier")
    builder.add_conditional_edges("run_verifier", route_after_verifier,
        {"pass": "format", "revise": "prepare_revise",
         "insufficient": "fallback", "conflict": "conflict"})
    builder.add_edge("prepare_revise", "verify")  # ← 回边：形成循环

    # 所有路径汇聚 → format → END
    for n in ["generate", "conflict", "fallback"]:
        builder.add_edge(n, "format")
    builder.add_edge("format", END)

    return builder.compile()

三类边：固定边 add_edge(A, B)、条件边 add_conditional_edges(A, router, mapping)、回边 prepare_revise → verify。

4. 节点详解

rewrite_node：多轮指代消解

python

async def rewrite_node(state):
    if not state.get("enable_rewrite") or not state.get("conv_id"):
        return {"rewritten_query": state["question"]}  # 跳过
    rewritten = await conversation_service.rewrite_query_with_history(...)
    return {"rewritten_query": rewritten}

跳过条件：单轮对话或开关关闭时直接透传原 query。

retrieve_node：调用第5课 hybrid search

python

async def retrieve_node(state):
    query = state.get("rewritten_query") or state["question"]
    hits = hybrid_retrieval_service.retrieve(query, top_k=..., mode=...)
    return {"hits": hits}

judge_node + 条件边：过滤 + 分流

python

async def judge_node(state):
    filtered = [h for h in state["hits"] if h["score"] >= state.get("min_score", 0.3)]
    return {"hits": filtered}

def route_after_judge(state):
    if not state.get("hits"):   return "miss"
    if state.get("enable_verify"): return "verify"
    return "generate"

judge 和路由分开 = 单一职责。judge 改过滤逻辑，路由改分支策略，互不影响。

verify_node vs generate_node

	generate_node	verify_node
模板	`rag_qa`（回答+标 [N]）	`verified_qa`（回答+拆 claims+绑 evidence）
输出	answer + citations	answer + claims + evidences
后续	→ format → END	→ verifier → 4 分支
修正提示	无	读 `revision_hints` 注入 prompt

verifier_node（空操作模式）

python

async def verifier_node(state):
    if not state.get("enable_verifier"):
        return {"verifier": {"verdict": "pass", "score": 1.0}}  # 跳过
    # 正常走 RuleVerifier + LLMJudgeVerifier

enable_verifier=False → 直接写 pass → 路由走 pass 分支 → 跳过整个验证链路。

prepare_revise_node：修正循环的钥匙

Code

async def prepare_revise_node(state):
    hints = VerifierResult(**state["verifier"]).to_hints()   # issues+suggestions → 文字
    return {"revision_count": state.get("revision_count", 0) + 1, "revision_hints": hints}

然后 prepare_revise → verify 回边 → verify_node 读 hints → LLM 看到修正提示 → 重新生成 → 再过 verifier。

route_after_verifier：防死循环安全阀

python

def route_after_verifier(state):
    verdict = state["verifier"]["verdict"]
    if verdict == "revise":
        if state["revision_count"] >= state.get("max_revisions", 2):
            return "insufficient"  # ← 强制退出
    return verdict

没有计数器 = verifier 次次判 REVISE → 无限循环 → 无限 LLM 调用 → 费用爆炸。

format_node：漏斗口

所有路径最终汇聚到 format → 落库（多轮对话记录）+ 汇总耗时 total_elapsed。

5. 一张图，三种模式

靠 state 初始值控制，零代码修改：

模式	enable_verify	enable_verifier	路径
简单 RAG	未覆盖	-	judge → generate → format
带验证 RAG	已覆盖	未覆盖	judge → verify → verifier(空) → pass → format
完整修正循环	已覆盖		judge → verify → verifier → PASS/REVISE/...

6. 成本分析：最坏情况多少次 LLM 调用

场景	LLM 调用	延迟	费用
最好（规则 PASS，无修正）	1 次	~0.5s	$0.003
典型（1 轮修正）	3-4 次	~2s	$0.012
最坏（3 轮修正全 LLM）	9 次	~4.5s	$0.027

优化建议：max_revisions 设 2（第 3 轮成功率极低）、HybridVerifier 规则层多承担判定、监控 revision_count 平均值判断 prompt 质量。

7. 工程教训

节点只返回自己写的字段→ 返回整个 state 会覆盖别人写的 → 难追踪"谁改了什么"
路由函数必须是纯函数→ 只读 state 不修改，不调外部服务。判断逻辑放节点里
回边必须有计数器→ revision_count >= max → 强制退出。没有就是费用黑洞
format_node 是漏斗口→ 落库、计时统一收编，不分散在每个分支末尾
lru_cache 缓存编译结果→ 图编译一次性操作，全局单例零重复开销

8. 自测

问题 1：新增 translate_node 在 format 前把答案翻译成英文。要改哪些地方？

问题 2：修正上限到了降级为 insufficient 一刀切——score=0.6 的答案也被判"无法回答"。怎么改进？

问题 3：_merge_meta 浅合并下两个节点写同名 key 会怎样？怎么防御？

问题 4：verify_node 和 generate_node 逻辑高度相似但模板不同。如何设计才不违反单一职责？

问题 5（最重要）：max_revisions=3 时最坏情况多少次 LLM 调用？列出每轮每个调用的用途。

答案与解析

问题 1：新增翻译节点

3 处改动：State 新增 enable_translate/target_lang/original_answer + 新建 translate_node（空操作模式向后兼容）+ 改图拓扑（所有到 format 的边改为到 translate → translate → format）。

translate 放在 format 之前——落库的应该是翻译后的最终答案。保留 original_answer 备查。

问题 2：分级降级

python

if cur >= limit:
    if score >= 0.5:
        return "pass"      # 答案还行，凑合用
    else:
        return "insufficient"  # 真的不行

score ≥ 0.5 降为 PASS 但标记 degraded_pass: true → 前端提示"答案可能不完整"。

"无法回答"是用户体验的核弹——只有真的没办法才用。能凑合给个答案（带免责声明）比完全拒绝好。

问题 3：命名空间防冲突

两个节点都写 meta.elapsed → 后写覆盖先写 → 数据丢失。防御：每个节点用带前缀的 key——rewrite_elapsed、retrieve_elapsed。当前代码已采用此方案。

问题 4：提取公共子步骤

不是把两个不同东西塞进一个函数。真正的复用：提取 _prepare_context 公共子步骤，generate_node 和 verify_node 各自调用，但模板渲染和输出解析各写各的。

问题 5：最坏成本

max_revisions=3 全 LLM 模式：初始轮 rewrite(1) + verify(1) + verifier_LLM(1) = 3 次；3 轮修正 × (verify(1) + verifier_LLM(1)) = 6 次。总计 9 次。费用约 $0.027，延迟约 4.5 秒。

图里每个"回边"都是成本放大器。上线前用最坏情况估算，确认业务能承受。

本节要点

图编排把业务流程变可视化——节点=步骤、条件边=分支、回边=循环。节点只写自己的字段，图自动 merge state
一张图支持三种模式——靠 state 开关，零代码修改
回边必须有安全阀——revision_count >= max 强制退出，防止无限循环烧钱
format_node 是所有路径的漏斗口——落库、计时统一收编
最坏情况下 9 次 LLM 调用——每个回边都是成本放大器，必须监控