本节进入 LangGraph 进阶：用 State、Node、Edge、Reducer 与条件路由构建可循环、可暂停、可恢复的 AI 工作流，并以“文章生成—审核—修改”为例串起核心能力。
1. LangGraph 的核心抽象
| 抽象 | 作用 |
|---|---|
| State | 工作流当前快照，保存节点共享的数据 |
| Node | 读取 State、执行任务、返回部分更新 |
| Edge | 指定固定的下一节点 |
| Conditional Edge | 根据 State 动态选择下一节点 |
| Reducer | 定义新旧状态如何合并 |
| Checkpointer | 保存各步骤状态，用于恢复、记忆与人工介入 |
可以把 LangGraph 理解为：节点负责做事，边负责决定接下来做什么，状态负责携带全过程数据。
2. 定义 State
fromoperatorimportaddfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictclassArticleState(TypedDict,total=False):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:response=model.invoke(f"为主题《{state['topic']}》生成一份五段式技术文章大纲。")return{"outline":response.content,"revision_count":0,"history":["outline_created"],}defwrite_draft(state:ArticleState)->dict:response=model.invoke("请根据以下大纲撰写技术文章，包含解释、代码与小结：\n"f"{state['outline']}")return{"draft":response.content,"history":["draft_created"],}defreview_draft(state:ArticleState)->dict:response=model.invoke("请审核下面的文章。若可发布，第一行输出 PASS；""否则第一行输出 REVISE，并列出最多三条具体修改意见。\n\n"f"{state['draft']}")text=response.content approved=text.lstrip().startswith("PASS")return{"approved":approved,"feedback":text,"history":["review_passed"ifapprovedelse"review_failed"],}defrevise_draft(state:ArticleState)->dict:response=model.invoke("根据审核意见修改文章。只输出修改后的完整正文。\n\n"f"原文：\n{state['draft']}\n\n"f"审核意见：\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):return"finish"ifstate.get("revision_count",0)>=2:return"finish"return"revise"revision_count是循环保险。任何循环图都应设置业务上限或recursion_limit，否则模型可能在“审核—修改”之间无限运行。
5. 组装与编译 Graph
fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportInMemorySaver builder=StateGraph(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")graph=builder.compile(checkpointer=InMemorySaver(),)Graph 必须先compile()才能执行。编译阶段会检查部分结构问题，并注入 Checkpointer 等运行能力。
6. 执行工作流
config={"configurable":{"thread_id":"article-2026-001"},"recursion_limit":20,}result=graph.invoke({"topic":"LangGraph 状态图入门","history":[],},config=config,)print(result["draft"])print(result["history"])同一thread_id下的状态可以被 Checkpointer 恢复。生产中应使用数据库型 Checkpointer，而不是只存在进程内存中的实现。
7. 流式观察节点执行
forupdateingraph.stream({"topic":"LangGraph 状态图入门","history":[]},config=config,stream_mode="updates",):print(update)updates：每步只返回节点产生的字段更新；values：每步返回累计后的完整 State；- 消息流模式：适合逐 Token 展示模型输出。
调试复杂图时，updates往往最容易看出是哪个节点修改了哪个字段。
8. 使用Command同时更新状态和路由
当节点本身已经知道下一步去哪，可以返回Command：
fromtypingimportLiteralfromlanggraph.typesimportCommanddefclassify(state:ArticleState,)->Command[Literal["draft","finish"]]:ifnotstate.get("outline"):returnCommand(update={"history":["missing_outline"]},goto="finish",)returnCommand(update={"history":["outline_ready"]},goto="draft",)Command适合节点在一次判断中同时完成“更新状态”和“选择目的地”。条件边适合把路由逻辑独立成函数，两种方式按可读性选择。
9. 人工介入：暂停与恢复
fromlanggraph.typesimportinterrupt,Commanddefhuman_review(state:ArticleState)->dict:decision=interrupt({"message":"请审核文章","draft":state["draft"],"requested_fields":["approved","feedback"],})return{"approved":bool(decision["approved"]),"feedback":decision.get("feedback",""),"history":["human_reviewed"],}图运行到interrupt()时会暂停并把状态保存到 Checkpointer。人工完成审核后，使用相同的thread_id恢复：
resumed=graph.invoke(Command(resume={"approved":False,"feedback":"代码示例需要补充异常处理。",}),config=config,)恢复时节点会从头重新执行到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、单一职责节点、确定的终止条件和幂等副作用。
