本节进入 LangGraph 进阶用 State、Node、Edge、Reducer 与条件路由构建可循环、可暂停、可恢复的 AI 工作流并以“文章生成—审核—修改”为例串起核心能力。1. LangGraph 的核心抽象抽象作用State工作流当前快照保存节点共享的数据Node读取 State、执行任务、返回部分更新Edge指定固定的下一节点Conditional Edge根据 State 动态选择下一节点Reducer定义新旧状态如何合并Checkpointer保存各步骤状态用于恢复、记忆与人工介入可以把 LangGraph 理解为节点负责做事边负责决定接下来做什么状态负责携带全过程数据。2. 定义 StatefromoperatorimportaddfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictclassArticleState(TypedDict,totalFalse):topic:stroutline:strdraft:strfeedback:strapproved:boolrevision_count:inthistory:Annotated[list[str],add]没有声明 Reducer 的字段默认采用覆盖策略。例如新节点返回{draft: 新内容}旧草稿会被替换。history使用operator.add作为 Reducer每个节点返回的新列表会追加到旧列表return{history:[已完成初稿]}节点只需要返回发生变化的字段不必复制完整 State。3. 编写节点defbuild_outline(state:ArticleState)-dict:responsemodel.invoke(f为主题《{state[topic]}》生成一份五段式技术文章大纲。)return{outline:response.content,revision_count:0,history:[outline_created],}defwrite_draft(state:ArticleState)-dict:responsemodel.invoke(请根据以下大纲撰写技术文章包含解释、代码与小结\nf{state[outline]})return{draft:response.content,history:[draft_created],}defreview_draft(state:ArticleState)-dict:responsemodel.invoke(请审核下面的文章。若可发布第一行输出 PASS否则第一行输出 REVISE并列出最多三条具体修改意见。\n\nf{state[draft]})textresponse.content approvedtext.lstrip().startswith(PASS)return{approved:approved,feedback:text,history:[review_passedifapprovedelsereview_failed],}defrevise_draft(state:ArticleState)-dict:responsemodel.invoke(根据审核意见修改文章。只输出修改后的完整正文。\n\nf原文\n{state[draft]}\n\nf审核意见\n{state[feedback]})return{draft:response.content,revision_count:state.get(revision_count,0)1,history:[draft_revised],}每个节点保持单一职责便于单元测试、重试和替换模型。4. 条件路由与循环fromtypingimportLiteraldefroute_after_review(state:ArticleState,)-Literal[revise,finish]:ifstate.get(approved,False):returnfinishifstate.get(revision_count,0)2:returnfinishreturnreviserevision_count是循环保险。任何循环图都应设置业务上限或recursion_limit否则模型可能在“审核—修改”之间无限运行。5. 组装与编译 Graphfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportInMemorySaver builderStateGraph(ArticleState)builder.add_node(outline,build_outline)builder.add_node(draft,write_draft)builder.add_node(review,review_draft)builder.add_node(revise,revise_draft)builder.add_edge(START,outline)builder.add_edge(outline,draft)builder.add_edge(draft,review)builder.add_conditional_edges(review,route_after_review,{revise:revise,finish:END,},)builder.add_edge(revise,review)graphbuilder.compile(checkpointerInMemorySaver(),)Graph 必须先compile()才能执行。编译阶段会检查部分结构问题并注入 Checkpointer 等运行能力。6. 执行工作流config{configurable:{thread_id:article-2026-001},recursion_limit:20,}resultgraph.invoke({topic:LangGraph 状态图入门,history:[],},configconfig,)print(result[draft])print(result[history])同一thread_id下的状态可以被 Checkpointer 恢复。生产中应使用数据库型 Checkpointer而不是只存在进程内存中的实现。7. 流式观察节点执行forupdateingraph.stream({topic:LangGraph 状态图入门,history:[]},configconfig,stream_modeupdates,):print(update)updates每步只返回节点产生的字段更新values每步返回累计后的完整 State消息流模式适合逐 Token 展示模型输出。调试复杂图时updates往往最容易看出是哪个节点修改了哪个字段。8. 使用Command同时更新状态和路由当节点本身已经知道下一步去哪可以返回CommandfromtypingimportLiteralfromlanggraph.typesimportCommanddefclassify(state:ArticleState,)-Command[Literal[draft,finish]]:ifnotstate.get(outline):returnCommand(update{history:[missing_outline]},gotofinish,)returnCommand(update{history:[outline_ready]},gotodraft,)Command适合节点在一次判断中同时完成“更新状态”和“选择目的地”。条件边适合把路由逻辑独立成函数两种方式按可读性选择。9. 人工介入暂停与恢复fromlanggraph.typesimportinterrupt,Commanddefhuman_review(state:ArticleState)-dict:decisioninterrupt({message:请审核文章,draft:state[draft],requested_fields:[approved,feedback],})return{approved:bool(decision[approved]),feedback:decision.get(feedback,),history:[human_reviewed],}图运行到interrupt()时会暂停并把状态保存到 Checkpointer。人工完成审核后使用相同的thread_id恢复resumedgraph.invoke(Command(resume{approved:False,feedback:代码示例需要补充异常处理。,}),configconfig,)恢复时节点会从头重新执行到interrupt()因此中断之前的副作用必须幂等或移动到中断之后。10. 状态设计原则保存原始数据而不是把所有内容提前格式化成 Prompt每个字段含义明确避免多个节点随意复用同一个字符串列表累计必须声明 Reducer大文件与数据库连接不要直接写入 State只保存引用会跨中断恢复的数据必须可序列化对包含副作用的节点增加幂等键。11. LangChain Agent 与 LangGraph 的关系create_agent()适合标准“模型—工具—模型”循环当流程包含多个角色、确定性步骤、复杂路由、并行分支或人工节点时再使用 LangGraph 显式编排。两者不是二选一一个完整的 Agent 可以作为 LangGraph 的节点或子图Agent 中的 Middleware 仍然继续生效。12. 常见问题12.1 状态列表被覆盖为需要累计的字段声明 Reducer否则每次返回的新值会覆盖旧值。12.2 图无法结束检查条件路由是否存在到END的分支并为循环增加最大次数与recursion_limit。12.3 恢复后重复发信或写库节点可能因恢复或重试再次执行。副作用操作应使用幂等键并记录执行状态。12.4 中断后找不到状态确认编译时提供了 Checkpointer恢复时使用原来的thread_id并且持久化后端仍可访问。13. 本节小结LangGraph 的价值在于把隐式的 Agent 推理过程变成显式、可观察、可暂停、可恢复的状态机。复杂工作流的稳定性主要取决于清晰的 State、单一职责节点、确定的终止条件和幂等副作用。