LangGraph 编排实战
把 Agent 建模成显式状态图：节点、条件边、checkpoint 持久化、interrupt 中断、多 Agent 通信与失败兜底、防死循环。
LangGraph 编排实战
LangGraph 把 Agent 流程建模成显式状态图——节点和边都可见、可调试、可回放，天然契合「节点固定、流程声明化」的编排诉求。
为什么是 LangGraph（三连对比）
| 对比 | 结论 |
|---|---|
| vs LangChain Chain/LCEL | Chain 是线性固定流水线，无全局状态、无分支循环、不可持久化；LangGraph 是 State 状态机驱动，支持条件分支、循环、断点续跑、状态快照 |
| vs LangChain Agent | Agent 是「循环式思考→行动」，路径由模型自由发挥不可控、黑盒难调；LangGraph 把流程变成显式状态图——可预测、可调试、可回放 |
| vs CrewAI | CrewAI 角色任务导向、固定协作模板、开箱即用但自定义路由/循环/状态管控弱；LangGraph 底层状态机编排、任意自定义，量产复杂系统选 LangGraph |
金句：简单标准化选 CrewAI/Chain，复杂、量产、高可控必须 LangGraph。
核心概念
graph TD
START((start)) --> N1[研究节点]
N1 --> N2[撰写节点]
N2 --> N3{审核条件边}
N3 -->|passed| N4[发布节点]
N3 -->|未过 & 轮次<3| N2
N3 -->|轮次=3| N5[转人工]
N4 -->|interrupt| HITL[人工确认]
HITL -->|resume| END((end))
1. StateGraph vs MessageGraph
- MessageGraph：只维护消息列表，所有节点输出合并为消息，无法字段隔离，只适合简单对话
- StateGraph：支持自定义结构化 State、字段级读写隔离、精准数据管控——生产多 Agent 强制 StateGraph
字段级隔离：State Schema 里定义每个字段 + 哪些节点可写哪些字段（配 reducer/权限）。并行写同一字段防丢 → 自定义 Reducer 追加合并（messages 用 operator.add），默认覆盖模式并行必丢数据。
2. 条件边（conditional edge）
= 路由函数 + 返回值到下游节点的映射表：
graph.add_conditional_edges(source, routing_fn, {"call_tool": "tool_node", "finish": END})
执行顺序：source 跑完 → 运行时拿 state 调 routing_fn → 按返回字符串查映射表跳转。路由函数是普通 Python 函数，不强制跑 LLM。
⚠️ 映射表里没写的返回串 = 运行期崩溃（不是编译期报错）→ 加
"default": END。
3. checkpoint 持久化
状态持久化到 MySQL/PG，进程重启后图能从 checkpoint 恢复。编译时挂 checkpointer：
graph.compile(checkpointer=PostgresSaver.from_conn_string(...))
按 thread_id 存状态快照。临时 checkpoint 可用 Redis，永久业务状态落 PG。
4. interrupt + HITL
发布节点执行前 interrupt 挂起，生成待审批记录；审批通过 resume 恢复执行。因为状态已持久化，即使进程重启也能恢复。
# interrupt() 挂起 → 人工审批 → Command(resume=...) 恢复
多 Agent 通信与数据流转
核心是全局 State 共享 + 边路由调度——Agent 之间不点对点直连，全部读写统一全局状态；前序 Agent 把结果/日志/状态写进 State，经普通边/条件边流转，下游读最新 State 执行再回写。收益：解耦、可追溯、可持久化。
顺带澄清一个高频误区：如果只是固定 Router 分发、Agent 无自主决策/互相调用，业界定义这叫 Workflow 不叫 Multi-Agent（Anthropic《Building Effective Agents》定调）。被问就诚实承认 + 说清「我需要的是可控流程而非自由协作」。
子 Agent 失败与兜底（分层策略）
1. 重试层
最多 N 次（2-3 次），指数退避（1s/2s/4s），只对可重试错误重试（超时/5xx），4xx/参数错不重试。
2. 降级层
重试仍失败 → 返回结构化错误或降级结果（如「检索失败，跳过此部分」），不能让单个子任务失败拖垮整个任务。
并行降级正确写法：
results = await asyncio.gather(*tasks, return_exceptions=True)
# 逐个检查 Exception 实例分别处理；同步 invoke 放进 gather 不会真并发，要用 ainvoke
3. 流程层兜底
子 Agent 降级后 Supervisor 要感知（读 agent_results 里的 status 字段），决定跳过该模块、用默认值还是转人工。
防死循环 / 无限重试（必考工程题）
迭代次数硬上限 + 超时阈值双兜底（LangGraph: recursion_limit；自研：max_rounds）。单靠「模型判断完成」不可靠。
四个高频线上 BUG + 根治：
- 状态覆盖污染（并行同字段）→ Reducer 追加合并 + 字段隔离
- 死循环雪崩（迭代无终止）→ 最大迭代次数 + 超时双兜底
- 任务重复抢占（分布式）→ 状态任务锁 + 执行标记
- 调度分配错误（路由规则模糊）→ 结构化 Prompt + 量化评分规则
记忆口诀：丢数加合并、循环加阈值、抢活加锁、错配加规则。
关键决策：依赖注入
LangGraph 图全部依赖注入（retriever/generator/classifier 参数化），契约测试不碰向量库与模型——这是「编排与检索解耦」的关键。模型分级（fast/strong 档位）也让不同节点用不同档位模型成为可能。
测试方法：单测（每个 node 函数）+ 图级测试（完整跑通、路由正确、状态按预期更新）+ 关键路径专项测试（如审核回写循环）；加分项是可重放（replay）调试——把 trace 存下来，出问题用同一输入回放。