第16章 · LangGraph 实操与工作流编排

版本基线：langgraph 1.2.10 / langgraph-checkpoint-sqlite 3.1.1。全部 lab 用确定性 mock model，只证明框架合同，不证明 live planner 质量。

先修：第13章的手写 agent harness（显式状态机 s_{t+1}=δ(s_t,e_t,validate(a_t))、typed tools、幂等、HITL、checkpoint）+ 第14章的真实 LLM API 路径。

本章目标

• 把第13章手写状态机的每个部件逐件映射到 LangGraph：State(TypedDict)+reducer、节点+条件边、@tool+ToolNode、checkpointer+thread_id、interrupt+Command(resume)。
• 真实跑通 langgraph：图编译与调用、reducer 追加语义、条件边路由、SqliteSaver 跨实例恢复、interrupt/resume。
• 说清 interrupt 的"节点从头重跑"语义及三条工程规则，并用一次故障注入亲眼看到非幂等副作用被重复执行。
• 画出 Anthropic《Building effective agents》五种 workflow 模式的图形态，说清 workflow（预定义代码路径）与 agent（模型自主循环）的决策边界。
• 知道 LangChain v1 的分工：create_agent + middleware 是生产快捷方式，create_react_agent 已废弃，LCEL 退居 langchain-classic（本课不教）。

概念映射表（手写 → LangGraph）

本章的一切都是这张表的展开。先把它读懂，再动手。

第13章手写件	LangGraph 对应	关键差异
状态 s_t（AgentState dataclass）	State（TypedDict），字段用 Annotated[list, operator.add] 声明 reducer	节点返回部分更新，reducer 决定合并方式（追加 vs 覆盖）；手写版是节点内直接 mutate
转移函数 δ / while 主循环	StateGraph 的节点 + 边；循环 = 条件边回指上游节点	循环不再是 while 关键字，而是图拓扑；终止是路由到 END
validate(a_t)（ToolSpec input_schema）	@tool 装饰器从函数签名/类型标注自动生成 JSON Schema + ToolNode 执行与错误回传	schema 是"免费"的，但风险分级、policy、幂等键仍要你自己做——框架不替代第13章的 gate
HITL 批准（waiting_approval + approval token）	节点内 interrupt(payload) + Command(resume=value) 恢复	恢复时整个节点从头重跑（详见第 4 节三条规则）
CheckpointStore.save/load	compile(checkpointer=...) + config={"configurable": {"thread_id": ...}}；get_state_history = time-travel	同一 thread_id 即恢复；InMemorySaver 进程内，SqliteSaver 落盘跨进程
幂等约束（idempotency key）	interrupt 重跑语义同样要求幂等：interrupt 前的副作用必须幂等或移后	同一条纪律，框架不会替你执行

依赖安装（可选依赖，不进课程基线）

.venv/bin/pip install -r requirements-agent.txt   # 追加项：langgraph、langgraph-checkpoint-sqlite

langgraph 会带入 langchain-core（@tool、消息类型来自这里）。测试在依赖缺失时整模块 skip（诚实降级），不会把"没装"伪装成"通过"。

从零实践（五节）

参考实现是 python/labs/langgraph_labs.py，测试是 python/tests/test_langgraph_labs.py（13 项，全部真实运行 langgraph，不打桩）。所有 lab 用确定性 mock model / 假决策函数——与第13章 FixtureModel 同一边界。

第 1 节 · 从手写状态机到 StateGraph

手写 loop 的骨架：while running: action = planner(state); state.step += 1; ...。LangGraph 版（build_loop_graph）：

class LoopState(TypedDict):
    step: int
    done: bool
    trace: Annotated[list[str], operator.add]  # reducer：追加而非覆盖

graph = StateGraph(LoopState)
graph.add_node("decide", decide)   # = planner
graph.add_node("act", act)         # = executor
graph.add_edge(START, "decide")
graph.add_conditional_edges("decide", route, {"act": "act", "end": END})
graph.add_edge("act", "decide")    # 循环 = 边回指，不是 while
app = graph.compile()
result = app.invoke({"step": 0, "done": False, "trace": []})

动手：跑 lab1_stategraph_loop(max_steps=3)，数 trace 长度（实测 7 = 3 轮 decide+act + 1 次 done 判定）。然后把 reducer 从 operator.add 改成默认（无 Annotated），观察 trace 只剩最后一个节点的返回值——这就是 reducer 的存在意义：多个节点的部分更新如何合并进同一份 State。

第 2 节 · Typed tools 与 ToolNode

第13章的两个 typed tools 框架化（lookup_fact / append_note）：

@tool
def append_note(note: str, count: int = 1) -> str:
    """Append a note `count` times to the scratchpad."""
    if count < 1 or count > 5:
        raise ValueError("count must be within [1, 5]")
    return f"noted:{note} x{count}"

@tool 从签名生成 JSON Schema（append_note.args_schema.model_json_schema() 可验证：note: string 必填、count: integer）。真实系统里模型节点是 model.bind_tools(tools) 的 chat model；本章用 MockToolModel（按脚本发 AIMessage(tool_calls=[...])）保持确定性。ToolNode 执行工具并把结果作为 ToolMessage 回传——这正是手写版 observe → plan 的那条边。

动手：跑通 lab2_tool_loop（2 次工具调用、6 条消息、最终文本由 tool 结果拼成）。再跑 lab2_tool_error_path：工具抛 ValueError 时图不崩，错误作为 status="error" 的 ToolMessage 回传给模型。注意一个实测坑：ToolNode 默认只捕获部分异常类型，ValueError 会被重新抛出炸掉图——需要显式 ToolNode(..., handle_tool_errors=True) 才把所有异常转成 error 消息。回答：为什么工具错误应该作为 data 回传给模型而不是直接让图失败？（模型可以观察错误并修正参数重试——对应第13章的"schema 失败提示模型重写参数"。）

第 3 节 · Checkpoint 与恢复

from langgraph.checkpoint.sqlite import SqliteSaver
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
app = graph.compile(checkpointer=SqliteSaver(conn))
config = {"configurable": {"thread_id": "w10c-lab3"}}
app.invoke(input, config=config)          # 进程 A
# ... 进程 A 退出；进程 B 新建连接、重新 compile，同一 thread_id ...
app_b.get_state(config)                   # 完整恢复，零节点重跑
list(app_b.get_state_history(config))     # time-travel：逐步 checkpoint 历史

对照手写版：CheckpointStore.save(state) → checkpointer 在每个节点边界自动落盘；resume(checkpoint_hash) → 同一 thread_id 继续。InMemorySaver 适合开发与测试，SqliteSaver 才能跨进程。

动手：跑 lab3_sqlite_recovery——第一个实例跑完两步图后 conn.close()（模拟进程退出），第二个全新实例用同一 thread_id 读回状态。实测：恢复状态与首次运行完全一致、第二个实例零节点重跑、get_state_history 返回 4 个 checkpoint。进阶：lab3_interrupt_resume_sqlite 演示中断中的执行也能跨实例 resume（Command(resume=True) 在全新进程里继续跑完 before → gate → after）。

第 4 节 · HITL = interrupt

第13章的批准门在 LangGraph 里是一个插在敏感工具前的节点：

from langgraph.types import interrupt, Command

def approval_node(state):
    answer = interrupt({"action": "delete_records", "risk": "high"})  # 挂起，payload 须 JSON 可序列化
    side_effects["delete_records"] += 1   # 副作用放 interrupt 之后
    return {"decision": str(answer)}

first = app.invoke(input, config=config)       # 返回 {"__interrupt__": [...]}
resumed = app.invoke(Command(resume="approved"), config=config)  # 人工批准后恢复

关键语义：恢复时整个节点从头重跑，不是从 interrupt 行继续。 由此三条规则：

1. interrupt 之前的副作用必须幂等（或移到 interrupt 之后）；
2. 不要用裸 try/except 包住 interrupt——重跑时异常路径会再次触发；
3. 一个节点内多个 interrupt 按索引匹配 resume 值。

官方不推荐用静态 interrupt_before 做 HITL（它适合调试断点），批准语义应该用节点内 interrupt 表达。

故障题（必做）：build_hitl_graph(idempotent=False) 故意把 side_effects["delete_records"] += 1 放在 interrupt 之前。第一次 invoke 挂起时副作用已执行 1 次；Command(resume=...) 恢复后节点从头重跑，副作用执行第 2 次——实测计数 1→2。修正版把同一行移到 interrupt 之后，实测恢复后仍只执行 1 次。这与第13章幂等键解决的是同一类问题：任何"恢复"机制都会重放，副作用必须能安全重放。

第 5 节 · Workflow 模式谱系 + 何时不用 agent

Anthropic《Building effective agents》把 LLM 系统分两档：workflow（LLM 走预定义代码路径编排）与 agent（模型自主决定过程与工具使用）。决策边界一句话：从最简单的方案起步，仅在任务需要模型级自主决策时才增加自主性——每多一分自主性，就多一分延迟、成本与不可预测性，要用可观测的收益去换。

五种 workflow 模式的图形态（全部能用本章的 StateGraph 直接画出来）：

模式	图形态	适用
prompt chaining	线性链 A → B → C，节点间可有程序化的 gate	任务可分解为固定子步骤
routing	一个分类节点 + 条件边分发到专用分支	输入类别决定处理路径
parallelization	扇出并行节点 + 聚合节点（sectioning / voting）	子任务独立，或需多视角投票
orchestrator-workers	orchestrator 动态分解 → worker 并行 → synthesizer 合并	子任务无法预先穷举
evaluator-optimizer	生成 → 评估 → 条件边回环，达标或轮次用尽退出	有明确可计算的评估标准

动手：build_evaluator_optimizer 把手写 loop 改造成 evaluator-optimizer——mock generator 每轮产出 draft-v{n}，mock evaluator 给确定性评分，条件边在 score >= target 或 rounds >= max_rounds 时退出，否则回环到 generator。实测 target=3 时 3 轮收敛；max_rounds=2 时不收敛但按预算退出。把它和第 1 节的纯循环对比：同样的条件边回指，多了一个"评分驱动"的退出条件——workflow 模式的差别全在图拓扑与路由函数里。

收尾（知道即可，不展开）：LangChain v1 的高阶入口 create_agent 就跑在 LangGraph 上，配合 middleware（Summarization、HumanInTheLoop 等预置件）是生产快捷方式；旧的 langgraph.prebuilt.create_react_agent 已废弃；LCEL 退居 langchain-classic，本课不教。判断不变：快捷方式不替代你在第13章学过的 schema/policy/幂等/批准 gate。

故障注入与预期信号（权威表）

注入	预期失败信号	修复后证据
State 字段不写 reducer（无 Annotated[..., operator.add]）	后一个节点的返回值覆盖前面的累积，trace/messages 只剩最后一项	声明 reducer 后 trace 按节点顺序完整追加（lab1 实测 7 项）
ToolNode 默认错误处理 + 工具抛 ValueError	异常穿透节点，整个 invoke 抛错	handle_tool_errors=True 后错误作为 status="error" 的 ToolMessage 回传，模型可见并继续
interrupt 之前放非幂等副作用	Command(resume=...) 恢复后副作用执行第 2 次（实测计数 1→2）	副作用移到 interrupt 之后（或加幂等键），恢复后仍只执行 1 次
裸 try/except 包住 interrupt	重跑时异常路径再次触发，状态不可预测	interrupt 不做异常包装；恢复值校验放在 interrupt 之后
换了进程/实例但 thread_id 写错	get_state 读不到历史，从头空跑	同一 thread_id 跨实例恢复，零节点重跑（lab3 实测）
无 checkpointer 就用 interrupt	图无法挂起/恢复，直接报错	compile(checkpointer=InMemorySaver()) 起步，落盘换 SqliteSaver
evaluator-optimizer 只设 target 不设 max_rounds	评分永远不达标时图无限回环	双退出条件（达标 OR 轮次预算），超预算如实报"未收敛"

规范与延伸

• LangGraph 官方文档（以执行日为准）。
• Anthropic: Building effective agents——五模式与 workflow/agent 边界的原始出处。
• LangChain v1 agents（create_agent + middleware）。

前端/Agent 迁移

StateGraph 就是你写过的 store + reducer + 路由：Annotated[list, operator.add] 等价于 Redux 里 case APPEND: [...state, action.payload]；条件边等价于路由守卫；checkpointer 等价于把 store 快照持久化后按 session id 恢复；interrupt/resume 等价于一个会序列化自身等待用户操作的弹窗流程。真正的迁移判断在别处：图的拓扑是开发者定义的代码路径（workflow），还是模型运行时自主选择的循环（agent）——这个边界决定你的测试策略（前者可确定性回放，后者必须加预算与 gate）。

口述与自测（不看资料，5–10 分钟）

• 口述概念映射表全表：手写 loop 的每个部件在 LangGraph 里是什么，差异在哪。
• 解释 reducer：为什么 messages 字段不写 add_messages 会被覆盖？State 的更新机制到底是"返回全量"还是"返回增量"？
• 对比 conditional edge 与 Command(goto=...)：前者是静态声明的路由（图结构可见、可渲染），后者是节点内动态跳转（可带状态更新）；什么场景该用哪个？
• 背出 interrupt 的三条规则，并解释"节点从头重跑"为什么推出"副作用必须幂等"——用 lab4 的实测计数 1→2 作证。
• 说清 checkpointer / thread_id / store 的分工：checkpointer 管"执行到哪儿"（thread 内状态快照），thread_id 标识一段会话/任务，store 管跨 thread 的长期记忆（本课未展开）。
• 解释 LangChain 与 LangGraph 的关系（v1 答案）：LangGraph 是底层有状态编排引擎，LangChain v1 的 create_agent 是跑在其上的高阶封装 + middleware；create_react_agent 已废弃。
• 说出 Chain / Agent / Tool 三个抽象各自的定义；加分题：LCEL 现状（退居 langchain-classic，不再是新项目主线）。
• 用"从最简单方案起步"原则，给三个具体任务分别判定该用 workflow 还是 agent，并说出理由。

能不查资料写出 StateGraph 骨架（State+reducer+条件边回环）

能解释 reducer 追加语义并演示覆盖故障

能说清 thread_id 与 checkpointer 的恢复机制

能背出 interrupt 三条规则并举实测故障

能画出五种 workflow 模式并说 workflow vs agent 边界

能讲 LangChain/LangGraph v1 分工与 LCEL 现状

动手实验

五节全部真实运行 langgraph，确定性 mock model，无网络、无 API key。

环境准备

cd <仓库根>
export PYTHONPATH="$PWD/python"
.venv/bin/pip install -r requirements-agent.txt   # langgraph + langgraph-checkpoint-sqlite

命令与预期输出

# 1) 全部 13 项测试（真实跑 langgraph）
.venv/bin/python -m pytest python/tests/test_langgraph_labs.py -q

# 2) 五节 lab 汇总指标（evidence 数据源）
.venv/bin/python python/labs/langgraph_labs.py

.............                                                            [100%]
13 passed in 1.90s

lab1: steps=3, trace_length=7（decide/act 交替追加，末位 decide:done@3）
lab2: 2 次工具调用，6 条消息，final 文本由 tool 结果拼成
lab2_error: ToolNode 把 ValueError 回传为 status="error" 的 ToolMessage
lab3: 跨实例恢复 recovered_matches=true，节点重跑 0 次，history 4 个 checkpoint
lab3_hitl: 中断中的执行跨实例 resume 后跑完 ["before","gate","after"]
lab4_faulty: interrupt 前的副作用恢复后执行第 2 次（1→2，故障复现）
lab4_fixed: 副作用移到 interrupt 后，恢复后仍只执行 1 次（0→1，修复验证）
lab5: evaluator-optimizer 3 轮收敛（target=3），超预算时如实报未收敛

判定信号：
  reducer 追加、条件边路由、ToolNode 错误回传全部通过
  SqliteSaver 同一 thread_id 跨实例恢复零重跑
  interrupt/resume 节点重跑语义被故障题实证

概念图

资源 / 成本 / 隐私

全部 lab 本地运行、无网络、无真实模型调用，gross cost 为 0。langgraph / langgraph-checkpoint-sqlite 为 MIT 许可的可选依赖，不进入课程核心基线；依赖缺失时测试诚实 skip。trace 与 checkpoint 内容遵循第13章的 redact 纪律：真实凭据、cookie、他人数据一律不进 State、不进 checkpoint 文件——sqlite checkpoint 文件本身按敏感产物处理，不进 git。

Evidence

仓库当前机器证据（只读快照）

evidence/16-langgraph-v1.json 是当前 checkout 的脱敏机器运行记录：test_langgraph_labs.py 13 项全部通过、reducer/条件边/SqliteSaver 跨实例恢复/interrupt 重跑语义的实测指标（含故障版副作用 1→2 与修复版 0→1 的对照计数）。全部 lab 用确定性 mock model，只证明框架合同，不证明 live planner 质量；本模块已登记进 evidence/module-manifest-v1.json。

学习者提交模板（待填写，不是当前机器证据）

复制下面模板并填写自己的真实运行结果。所有 <...> 都是未填写状态；actual 和 artifacts 尤其不能被当作已运行或已通过。artifacts 必须替换为本次提交中真实存在的仓库相对路径。

schema: learn-llm.evidence.v1
module: 16-langgraph
commit: <learner-commit-sha>
verified_at: <iso-date>
environment: <sanitized-python-device>
seed: 10
commands:
  - PYTHONPATH=python python -m pytest python/tests/test_langgraph_labs.py -q
  - PYTHONPATH=python python python/labs/langgraph_labs.py
metrics:
  - name: langgraph_tests_passed
    expected: 13
    actual: <recorded-value>
  - name: lab3_recovered_state_matches
    expected: true
    actual: <recorded-value>
  - name: lab3_node_reruns_on_recovery
    expected: 0
    actual: <recorded-value>
  - name: lab4_faulty_side_effects_after_resume
    expected: 2
    actual: <recorded-value>
  - name: lab4_fixed_side_effects_after_resume
    expected: 1
    actual: <recorded-value>
  - name: lab5_rounds_to_converge
    expected: 3
    actual: <recorded-value>
artifacts:
  - <learner-repo-relative-artifact-path>
cost:
  gross_usd: 0
  credit_usd: 0
licenses:
  - source: langgraph
    version: <installed-version>
    license: MIT
    attribution: LangChain Inc.
    redistribution: permitted-by-license
known_failures:
  - <sanitized-failure-or-none>

只有 invoke 成功截图、没有跨实例恢复与 interrupt 重跑故障对照计数时，本章保持 gate。

下一步

• 第17章 · 全栈流式交付（页面由后续工作流创建）：把第14章的 streaming 解析接到前端——SSE 协议、fetch+ReadableStream 消费、增量渲染。