LangGraph 状态机:把 Agent 流程拆成可恢复、可路由的步骤
LangGraph 里的“状态机”不是学术概念炫技,而是把一个复杂 Agent 任务拆成节点、状态和路由规则,让你知道系统现在在哪、下一步去哪。
01.LangGraph 里的“状态机”到底是什么
很多人第一次听到状态机,会想到一张很理论化的转换图。但在 LangGraph 里,它更接近一种工程组织方式:
- •节点表示一个离散步骤
- •状态表示步骤之间共享的数据
- •边或
Command表示从当前步骤如何流向下一步
所以 LangGraph 的状态机思路,本质上是在回答三个问题:
- •当前任务走到哪一步了
- •下一步该去哪里
- •中途停下来之后,怎么继续
只要这三个问题是清楚的,复杂 Agent 流程就不容易变成一团“写在循环里的隐式逻辑”。
02.先别急着画图,先把流程拆成步骤
官方 “Thinking in LangGraph” 文档给了一个很实用的顺序:先把要自动化的流程拆成离散步骤,再决定状态和路由。
这个顺序非常重要,因为很多人会先想节点名,再拼流程,结果是图很完整,职责却很模糊。
以一个客服邮件 Agent 为例,比较合理的步骤拆分可能是:
- •读取用户请求
- •分类问题类型
- •搜索文档或查询工单
- •生成回复草稿
- •人工审核
- •发送回复
每个节点只做一件事,后面才好定义状态和错误处理。
03.状态里应该放什么,不该放什么
LangGraph 的状态不是“什么都往里塞的全局变量”,而是步骤之间需要共享、并且值得持久化的数据。
建议放进状态的内容
- •原始输入
- •分类结果
- •检索结果
- •草稿输出
- •执行元数据
尽量不要放进状态的内容
- •已经格式化好的长 prompt
- •可以现算出来的临时字符串
- •只在单个节点内部用一次的局部变量
官方文档里也专门强调了一个原则:状态尽量保留原始数据,prompt 在节点内部按需格式化。这样做的好处是:
- •不同节点可以用同一份数据生成不同上下文
- •更容易调试每一步到底拿到了什么
- •状态结构更稳定,不会因为 prompt 改版而频繁变化
04.一个更贴近当前文档的状态机写法
下面用一个简化的客服流转示例说明:
from typing import Literal
from typing_extensions import TypedDict
from langgraph.types import Command
class SupportState(TypedDict):
email_content: str
classification: dict | None
search_results: list[str]
draft_response: str这个状态已经能支撑一条多步骤流程:
- •
email_content是原始输入 - •
classification供后续路由判断 - •
search_results给生成回复节点用 - •
draft_response会在审核和发送阶段继续使用
重点不是字段多少,而是字段是否真的要跨节点存在。
05.什么时候用条件边,什么时候直接返回 Command
LangGraph 里有两种很常见的路由方式:
- •条件边
- •节点返回
Command
条件边更适合什么
当你只需要“根据状态决定下一跳”,而不需要同时更新状态时,条件边最直观。
from langgraph.graph import END
def route_after_classify(state: SupportState) -> Literal["search_docs", "draft_reply", END]:
intent = (state.get("classification") or {}).get("intent")
if intent == "question":
return "search_docs"
if intent == "bug":
return "draft_reply"
return END它的优点是简单,图也比较清晰。
Command 更适合什么
如果一个节点既要更新状态,又要决定下一步,那直接返回 Command 会更自然。当前官方 Graph API 文档也把它作为核心控制原语之一。
from langgraph.types import Command
def classify_intent(state: SupportState) -> Command[Literal["search_docs", "human_review"]]:
classification = {
"intent": "question",
"urgency": "medium",
}
goto = "search_docs"
if classification["urgency"] == "high":
goto = "human_review"
return Command(
update={"classification": classification},
goto=goto,
)这个模式的好处是:
- •路由和状态更新在一个地方完成
- •节点职责更完整
- •不用再额外写一个“只负责跳转”的函数
经验上,如果节点本来就承担“判断并路由”的职责,Command 往往会比条件边更顺手。
06.人工中断为什么必须和持久化一起考虑
LangGraph 很重要的一个能力是 human-in-the-loop,也就是在关键步骤停下来,让人来决定下一步。
这类场景通常要配合 interrupt() 使用:
from langgraph.types import Command, interrupt
from langgraph.graph import END
def human_review(state: SupportState) -> Command[Literal["send_reply", END]]:
decision = interrupt(
{
"email": state["email_content"],
"draft_response": state["draft_response"],
"action": "请审核是否发送",
}
)
if decision.get("approved"):
return Command(
update={"draft_response": decision.get("edited_response", state["draft_response"])},
goto="send_reply",
)
return Command(update={}, goto=END)但这里有个容易忽略的前提:如果没有 checkpointer,图就没有稳定的暂停和恢复基础。
所以人工中断从来不是单独的功能点,它和状态持久化是绑在一起的。
07.重试和错误处理,也应该是状态机的一部分
官方 “Thinking in LangGraph” 文档把错误大致分成几类:
- •瞬时错误,例如网络抖动、限流
- •LLM 可恢复错误,例如工具失败后可重新规划
- •需要用户或人工修复的错误
- •真正的程序错误
这类分法很实用,因为不同错误对应不同处理方式。
例如,对搜索节点做自动重试:
from langgraph.graph import StateGraph, START
from langgraph.types import RetryPolicy
def search_docs(state: SupportState) -> dict:
return {"search_results": ["文档 A", "文档 B"]}
builder = StateGraph(SupportState)
builder.add_node(
"search_docs",
search_docs,
retry_policy=RetryPolicy(max_attempts=3),
)
builder.add_edge(START, "search_docs")这样做的思路很对:把易失败的外部操作当成“图中的一个节点”,而不是躲在某个大函数里默默重试。
08.一个完整状态机的最小骨架
把上面的部分拼起来,LangGraph 状态机大概会形成这样的结构:
read_email
-> classify_intent
-> search_docs
-> draft_reply
-> human_review
-> send_reply
-> END这个结构的关键不是图好不好看,而是每个节点都满足:
- •职责单一
- •输入来自状态
- •输出是状态更新或带路由的 Command
- •中断、重试和恢复都有明确位置
只要这几点做到了,Agent 流程即使继续增长,也不会失去可读性。
09.最常见的三个误区
1. 节点太大
把分类、检索、生成、发送都塞进一个节点,表面上节点少了,实际上图已经失去意义,错误也更难定位。
2. 状态太脏
如果状态里同时堆满原始输入、格式化 prompt、日志字符串、临时变量,后面几乎一定会改不动。
3. 忘了恢复路径
很多人会先做“正常流程”,但真实系统里更常见的问题是:
- •中断后怎么继续
- •人工审核后怎么恢复
- •节点失败后从哪一步重试
状态机最有价值的地方,恰恰就在这些“不顺利”的路径里。
10.总结
LangGraph 状态机不是为了让流程看起来更高级,而是为了让复杂 Agent 的执行路径真正可解释、可恢复、可治理。最值得记住的几个原则是:
- •先拆步骤,再设计状态
- •状态存原始数据,不存格式化 prompt
- •简单路由用条件边,更新加路由时优先考虑
Command - •人工中断、持久化和恢复应该一起设计
理解了这些,再去扩展多 Agent、子图、长期记忆,系统会稳很多。