前面几篇介绍了 LangGraph 的 State、Node、Edge 和状态更新机制。到这里,Graph 已经可以按预先定义的路径运行。
但 Agent 的执行路径通常不是固定的。
模型返回 Tool Call 时,需要进入 Tool 节点;没有 Tool Call 时,可以直接结束。检索结果不满足要求时,可能重新检索,也可能进入其他处理流程。某一步执行失败后,也可能选择重试、降级或终止。
这些场景最终都要解决一个问题:当前 Node 执行完成后,下一个 Node 怎么确定。
LangGraph 中常用的方式有三种:
- 固定路径使用普通 Edge
- 根据 State 选择路径时使用 Conditional Edge
- Node 需要同时更新 State 和决定下一步时使用
Command
本文重点介绍后两种。
一、Graph中的执行路径
最简单的 Graph 只有固定顺序:
load
↓
process
↓
save
直接使用普通 Edge:
builder.add_edge("load", "process")
builder.add_edge("process", "save")
builder.add_edge("save", END)
add_edge(start_key, end_key) 用来连接两个 Node。
前一个 Node 执行完成后,Graph 直接进入后一个 Node,不需要额外判断。
但 Agent 中更常见的是分支:
analyze
│
├── 条件满足 ──→ process
│
└── 条件不满足 → fallback
或者:
model
│
├── 有 Tool Call ──→ tools ──→ model
│
└── 无 Tool Call ──→ END
这些路径无法在构建 Graph 时完全确定,需要等 Node 执行结束后,根据当前 State 再决定下一步。
如果需要根据 State 判断决定下一步执行哪个节点,就要使用动态路由。
二、Conditional Edge
Conditional Edge,也就是条件边,用于根据 State 动态选择后续节点。
对应的方法是:
StateGraph.add_conditional_edges(
source,
path,
path_map=None,
)
三个参数分别是:
source:从哪个 Node 开始分支path:路由函数path_map:可选,把路由结果映射到具体 Node
例如:
builder.add_conditional_edges(
"check",
route_after_check,
)
这里的 route_after_check 是一个普通 Python 函数:
from typing import Literal
def route_after_check(
state: State,
) -> Literal["process", "fallback"]:
if state["score"] >= 60:
return "process"
return "fallback"
它读取当前 State,然后返回下一个 Node 的名称。
Literal 用来限制函数可能返回的值:
Literal["process", "fallback"]
表示这里只会返回 "process" 或 "fallback"。
对于 LangGraph,这个类型信息也有实际作用。它可以帮助 Graph 推断条件边可能指向哪些节点,生成结构图时会更准确。
Conditional Edge 的执行顺序是:
执行 source Node
↓
更新 State
↓
执行路由函数
↓
读取最新 State
↓
确定下一个 Node
需要注意的是,路由函数读取的是 source Node 更新之后的 State。
因此,Node 负责产生用于判断的结果,Router 只负责根据结果分流。
三、Conditional Edge示例
下面示例展示了根据文本长度选择不同的处理节点。
from typing import Literal, NotRequired
from typing_extensions import TypedDict
from langgraph.graph import START, END, StateGraph
class State(TypedDict):
text: str
length: NotRequired[int]
result: NotRequired[str]
def analyze(state: State):
return {
"length": len(state["text"])
}
def route_by_length(
state: State,
) -> Literal["short_text", "long_text"]:
if state["length"] < 10:
return "short_text"
return "long_text"
def handle_short_text(state: State):
return {
"result": "使用简短文本处理流程"
}
def handle_long_text(state: State):
return {
"result": "使用长文本处理流程"
}
builder = StateGraph(State)
builder.add_node("analyze", analyze)
builder.add_node("short_text", handle_short_text)
builder.add_node("long_text", handle_long_text)
builder.add_edge(START, "analyze")
builder.add_conditional_edges(
"analyze",
route_by_length,
)
builder.add_edge("short_text", END)
builder.add_edge("long_text", END)
graph = builder.compile()
result = graph.invoke({
"text": "LangGraph"
})
print(result)
StateGraph(State) 创建 Graph,并使用 State 定义整个 Graph 共享的数据结构。
Graph 定义完成后,需要调用 compile() 编译成可执行对象,再通过 invoke() 执行。
上例运行结果如下:
{
'text': 'LangGraph',
'length': 9,
'result': '使用简短文本处理流程'
}
节点 analyze 先计算文本长度,并返回:
{"length": 9}
LangGraph 将结果写入 State。
随后执行 route_by_length。因为 length 小于 10,所以返回 "short_text",Graph 进入 short_text 节点。
这里两个函数的职责比较清楚:
analyze
计算长度
route_by_length
根据长度选择路径
如果不希望 Router 直接返回 Node 名称,也可以使用 path_map:
def route_by_length(
state: State,
) -> Literal["short", "long"]:
return "short" if state["length"] < 10 else "long"
builder.add_conditional_edges(
"analyze",
route_by_length,
{
"short": "short_text",
"long": "long_text",
},
)
这时 Router 返回的是 "short" 或 "long",再由 path_map 映射到具体 Node。
如果路由结果本身有明确的业务含义,这种写法更容易维护。
四、循环与Agent Loop
Conditional Edge 也可以用来实现循环。
Tool Calling Agent 就是一个典型例子:
model
│
├── 有 Tool Call ──→ tools
│ │
│ └──→ model
│
└── 无 Tool Call ──→ END
tools → model 是固定路径。
model 后面走哪条路径,则取决于模型返回结果:
def route_after_model(
state: State,
) -> Literal["tools", "__end__"]:
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
然后注册条件边:
builder.add_conditional_edges(
"model",
route_after_model,
)
builder.add_edge("tools", "model")
START 和 END 是 LangGraph 内置的特殊节点,分别表示 Graph 的入口和结束位置。
从 Graph 角度看,Agent Loop 并没有特殊之处:
model
↓
检查 State
↓
tools 或 END
↓
tools 执行
↓
重新回到 model
Agent 是否继续执行,最终还是由明确的条件决定。
实际项目中通常要考虑这些条件:
- 是否存在 Tool Call
- Tool 是否执行成功
- 是否需要重试
- 当前结果是否满足结束条件
- 是否超过最大执行次数
只要存在循环,就应该有明确的退出条件。
否则 Router 一直返回前序节点,Graph 就会持续运行。
五、Command
Conditional Edge 的写法是:
Node
负责更新 State
Router
负责选择路径
有些场景没必要拆成两步。
例如审核节点:
审核通过
→ decision = approved
→ publish
审核失败
→ decision = rejected
→ revise
这里 State 更新和路径选择来自同一个判断。
这种情况可以直接返回 Command。
Command 定义在:
from langgraph.types import Command
常见写法:
Command(
update=...,
goto=...,
graph=...,
resume=...,
)
各参数作用如下:
| 参数 | 作用 | 常见用法 |
|---|---|---|
update | 更新当前 Graph 的 State | Node 执行完成后写入新的状态,例如审核结果、重试次数 |
goto | 指定接下来执行的 Node | 根据当前处理结果跳转到 publish、retry、tools 等节点 |
graph | 指定 goto 应作用于哪个 Graph | Subgraph 中需要跳转到父 Graph 时使用,例如 Command.PARENT |
resume | 为暂停的 Graph 提供恢复值 | 配合 interrupt() 使用,在人工确认或外部输入完成后继续执行 |
其中最常用的是 update 和 goto:
return Command(
update={"decision": "approved"},
goto="publish",
)
这表示当前 Node 执行完成后,将 decision 更新为 "approved",然后继续执行 publish。
graph 和 resume 的使用场景更具体。graph 主要用于 Subgraph 与父 Graph 之间的跳转,resume 则用于 Interrupt 恢复。本文先理解它们的用途,后续介绍 Subgraph 和 Human-in-the-loop 时再展开。
本质上,一次 Node 执行就完成两件事:
更新 State
+
决定下一步
Command 这种写法比较适合状态变化和路由本来就是同一次业务判断的情况。
六、Command示例
继续看审核流程:
from typing import Literal, NotRequired
from typing_extensions import TypedDict
from langgraph.graph import START, END, StateGraph
from langgraph.types import Command
class State(TypedDict):
score: int
decision: NotRequired[str]
result: NotRequired[str]
def review(
state: State,
) -> Command[Literal["publish", "revise"]]:
if state["score"] >= 80:
return Command(
update={"decision": "approved"},
goto="publish",
)
return Command(
update={"decision": "rejected"},
goto="revise",
)
def publish(state: State):
return {
"result": "进入发布流程"
}
def revise(state: State):
return {
"result": "返回修改流程"
}
builder = StateGraph(State)
builder.add_node("review", review)
builder.add_node("publish", publish)
builder.add_node("revise", revise)
builder.add_edge(START, "review")
builder.add_edge("publish", END)
builder.add_edge("revise", END)
graph = builder.compile()
result = graph.invoke({
"score": 92
})
print(result)
这里有一个新的类型写法:
Command[Literal["publish", "revise"]]
它表示 review 返回的是 Command,并且可能跳转到 "publish" 或 "revise"。
上例运行结果:
{
'score': 92,
'decision': 'approved',
'result': '进入发布流程'
}
score 为 92,所以 review 返回:
Command(
update={"decision": "approved"},
goto="publish",
)
LangGraph 先更新 decision,然后执行 publish。
这里不需要额外再配置 Conditional Edge。
七、Conditional Edge与Command
两种方式都能做动态路由,但适合的场景不同。
| 对比项 | Conditional Edge | Command |
|---|---|---|
| 路由位置 | 独立 Router | Node 内 |
| State 更新 | Node 返回 | update |
| 路径选择 | Router 返回 | goto |
| 适合场景 | 根据已有 State 分流 | 更新 State 后立即跳转 |
| 流程结构 | 更直观 | 更集中 |
| 跨 Graph 路由 | 不适合 | 支持 |
如果只是读取已有 State,再决定下一步:
def route(state):
if state["has_error"]:
return "retry"
return "finish"
使用 Conditional Edge 更合适。
如果判断过程中同时产生新的状态:
return Command(
update={
"retry_count": state["retry_count"] + 1,
"status": "retrying",
},
goto="retry",
)
使用 Command 更直接。
实际项目中不建议把所有路由都改成 Command。
如果大量 Node 都在内部通过 goto 决定下一步,只看 Graph Builder 很难知道完整流程。
对于结构比较固定的 Workflow,显式 Edge 和 Conditional Edge 通常更清楚。
对于运行过程中经常根据当前处理结果调整路径的 Agent,Command 会更方便。
八、路由设计注意事项
Router尽量只做判断
Router 最好只读取 State,然后返回路径:
def route(state):
if state["validation_failed"]:
return "retry"
return "continue"
如果 Router 里面还要调用模型、访问数据库或者请求外部 API,那么这些逻辑更适合放进 Node。
Router 越简单,Graph 的执行路径越清晰。
判断数据放进State
路由需要使用的数据,应显式保存在 State 中,例如:
validation_status
retry_count
tool_result
classification
permission
例如需要根据重试次数决定是否继续,就应该把 retry_count 放进 State,而不是使用模块级变量保存。
这样在查看 State 时,就能知道一次执行为什么进入某个分支。
Node出口不要混用
如果一个 Node 已经返回:
Command(goto="publish")
又给它配置:
builder.add_edge("review", "save")
需要特别注意。
Command 不会自动覆盖已经存在的 Edge。
因此,一个 Node 最好明确使用哪种方式控制后续路径:
固定路径
→ add_edge
根据 State 分支
→ add_conditional_edges
更新 State 并跳转
→ Command
这样 Graph 的结构更容易看懂,也更方便排查执行问题。
总结
LangGraph 中的动态路由主要解决一件事:当前 Node 执行完成以后,下一步执行谁。
固定路径使用普通 Edge。
如果只是根据当前 State 选择下一步,使用 Conditional Edge。Node 负责处理数据,Router 负责分流。
如果一次处理既要修改 State,又要决定下一步,使用 Command 更合适。实际开发时不需要刻意统一成一种写法。
下一篇继续介绍 Persistence、Checkpoint 和 Thread,看 LangGraph 如何保存 Graph State,以及一次执行结束以后如何继续使用之前的状态。
社区讨论
参与讨论
有问题或想法?欢迎继续讨论。