单 Agent 会遇到天花板工具一多就乱选、长任务一断就从头再来。LangGraph 用「把流程画成状态机」的方式解决这些问题这也是它成为 2026 年生产级 Agent 首选的原因。附完整可运行代码。文章目录一、为什么不是 LangChain 而是 LangGraph二、环境与最小示例三、条件边让流程学会「拐弯」四、多智能体三种协作拓扑怎么选五、断点续跑与人工介入生产刚需5.1 Checkpoint进程崩了也能续5.2 Interrupt关键操作前停下来等人批5.3 节点级重试让偶发故障自愈5.4 流式输出用户体验的分水岭5.5 什么时候不要用多 Agent六、生产化检查清单一、为什么不是 LangChain 而是 LangGraph一句话说清关系LangChain 是组件库LangGraph 是把组件串成可控流程的编排层。它把 Agent 的工作流建模成图GraphNode节点一个函数干一件事调模型、跑工具、查库Edge边节点之间的流转支持条件分支State状态全局共享的数据所有节点读写。好处在生产环境才显出来能力裸写 Agent 循环LangGraph流程可控模型自己决定下一步黑盒图里画死白盒可审计断点续跑断了重来token 白烧checkpoint 自动落盘从断点恢复人工介入基本没法做原生 interrupt人审后继续出错定位翻日志猜每个节点独立可测二、环境与最小示例pipinstalllanggraph langchain-openaiexportOPENAI_API_KEYsk-xxx# 或用 Ollama把 ChatOpenAI 换成 base_url 指向本地先看一个「三个节点串成审查流水线」的骨架# minimal_graph.pyfromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassState(TypedDict):question:strdraft:strreview:strfinal:strdefdraft_node(state:State)-dict:# 实际项目在这里调 LLM示例用占位return{draft:f关于「{state[question]}」的初稿}defreview_node(state:State)-dict:return{review:f初稿审查通过{state[draft]}}defpolish_node(state:State)-dict:return{final:f定稿{state[draft]}已按审查意见修改}builderStateGraph(State)builder.add_node(draft,draft_node)builder.add_node(review,review_node)builder.add_node(polish,polish_node)builder.add_edge(START,draft)builder.add_edge(draft,review)builder.add_edge(review,polish)builder.add_edge(polish,END)graphbuilder.compile()resultgraph.invoke({question:AI Agent 的三大组件})print(result[final])状态就是 TypedDict节点返回「要更新的字段」框架负责合并——这就是 LangGraph 的全部心智模型。三、条件边让流程学会「拐弯」审查不通过要打回重写用条件边实现fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,LiteralclassState(TypedDict):draft:strscore:intattempts:intdefreview_node(state:State)-dict:return{score:60ifstate[attempts]2else85}# 模拟评分defroute(state:State)-Literal[rewrite,accept]:returnrewriteifstate[score]80elseacceptbuilderStateGraph(State)builder.add_node(review,review_node)builder.add_conditional_edges(review,route,{rewrite:review,accept:END})builder.add_edge(START,review)# 真实项目记得加最大重试次数防止无限循环烧钱defroute_safe(state:State)-Literal[rewrite,accept]:ifstate[attempts]3:returnaccept# 超限降级接受并标记人工处理returnrewriteifstate[score]80elseaccept踩坑所有循环边必须有退出条件。没设重试上限的审查循环模型可能永远打回重写一晚上烧掉一个月预算。四、多智能体三种协作拓扑怎么选拓扑结构适用注意Supervisor主管式一个协调节点分派给专家节点大多数业务场景推荐默认协调节点逻辑要简单明确流水线节点串行各自加工文档处理、内容生产上游输出质量决定一切辩论/群聊多个 Agent 互评方案评审、提高结论可靠性token 消耗 ×N慎用Supervisor 模式骨架生产最常用# supervisor.py核心逻辑节选fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,LiteralclassTeamState(TypedDict):task:strnext_agent:strresearch:strcode:strreport:strdefsupervisor(state:TeamState)-dict:# 实际项目用 LLM 结构化输出决定下一个 agent# prompt 要求只输出 researcher / coder / writer / FINISHifnotstate.get(research):return{next_agent:researcher}ifnotstate.get(code):return{next_agent:coder}ifnotstate.get(report):return{next_agent:writer}return{next_agent:FINISH}defroute(state:TeamState)-Literal[researcher,coder,writer,__end__]:return{FINISH:__end__}.get(state[next_agent],state[next_agent])builderStateGraph(TeamState)builder.add_node(supervisor,supervisor)builder.add_node(researcher,lambdas:{research:调研结果...})builder.add_node(coder,lambdas:{code:print(hello)})builder.add_node(writer,lambdas:{report:最终报告...})builder.add_edge(START,supervisor)builder.add_conditional_edges(supervisor,route)fornodein[researcher,coder,writer]:builder.add_edge(node,supervisor)# 干完活回主管由主管决定下一步注意专家节点干完活必须回到 supervisor由它统一调度——让专家之间直接对话是多 Agent 系统失控的头号原因。五、断点续跑与人工介入生产刚需5.1 Checkpoint进程崩了也能续fromlanggraph.checkpoint.sqliteimportSqliteSaver checkpointerSqliteSaver.from_conn_string(runs.db)graphbuilder.compile(checkpointercheckpointer)# 每次调用带 thread_id同一个 id 的执行历史自动持久化config{configurable:{thread_id:order-20260928-001}}graph.invoke({task:处理这批退货工单},config)# 进程重启后同 thread_id 再 invoke从最近的 checkpoint 继续5.2 Interrupt关键操作前停下来等人批fromlanggraph.typesimportinterrupt,Commanddefrefund_node(state:TeamState)-dict:# 流程会在这里暂停把内容抛给人类approvalinterrupt({question:是否同意退款,detail:state[task]})ifapproval!yes:return{report:已取消人工拒绝了退款}return{report:退款已执行}人工放行只需对同一thread_id再发一次指令graph.invoke(Command(resumeyes),config)凡是动钱的、发出去就收不回的操作退款、发邮件、调用生产 API前面一律加 interrupt。这一条能避开 90% 的 Agent 事故。5.3 节点级重试让偶发故障自愈调外部 API 的节点模型、搜索、数据库总有抖动LangGraph 内置了重试策略不用自己包 try/exceptfromlanggraph.pregelimportRetryPolicy builder.add_node(researcher,research_fn,retryRetryPolicy(max_attempts3,initial_interval1.0,backoff_factor2.0),)注意只给会抖动的节点加重试。LLM 调用、HTTP 请求值得加纯本地计算解析、格式化加了反而掩盖真 bug。5.4 流式输出用户体验的分水岭多 Agent 链路长不流式的话用户盯着白屏几十秒必走人。LangGraph 两种流式都支持# 1) token 级流式模型边生成边吐字forchunkingraph.stream({task:总结这批工单},config,stream_modemessages):ifchunk.content:print(chunk.content,end,flushTrue)# 2) 节点级流式前端展示「正在执行调研 → 写码 → 出报告」forupdateingraph.stream(input,config,stream_modeupdates):fornode_name,state_deltainupdate.items():print(f[步骤]{node_name}完成)生产建议两个都开节点级做进度条token 级做正文输出。5.5 什么时候不要用多 Agent泼一盆必要的冷水——多 Agent 不是银弹出现以下信号时留在单 Agent任务流程固定、步骤少写成普通函数链更稳、更便宜、更好测单模型 好的 Prompt 能达标多 Agent 拓扑本身会引入新的失败模式调度错、状态污染token 预算紧张Supervisor 每次调度都是额外 LLM 调用成本轻松翻倍。判断标准当「一个 Prompt 装不下所有工具和规则」或者「不同环节需要不同模型/温度/权限」时才值得拆多 Agent。拆分时优先按权限拆能碰钱的和不能碰钱的分开其次按领域拆。六、生产化检查清单每个节点单一职责可独立单测LangGraph 节点就是普通函数所有循环有最大次数限制thread_id有业务含义订单号/会话 id方便追溯危险操作前有 interrupt接 LangSmith 或自建日志每个节点的输入输出留痕状态里的字段做减法——状态越大每次 checkpoint 越贵。