前面几篇介绍了 LangGraph 的 State、Node、Edge 和状态更新机制。到这里,Graph 已经可以按预先定义的路径运行。

但 Agent 的执行路径通常不是固定的。

模型返回 Tool Call 时,需要进入 Tool 节点;没有 Tool Call 时,可以直接结束。检索结果不满足要求时,可能重新检索,也可能进入其他处理流程。某一步执行失败后,也可能选择重试、降级或终止。

这些场景最终都要解决一个问题:当前 Node 执行完成后,下一个 Node 怎么确定。

LangGraph 中常用的方式有三种:

  • 固定路径使用普通 Edge
  • 根据 State 选择路径时使用 Conditional Edge
  • Node 需要同时更新 State 和决定下一步时使用 Command

本文重点介绍后两种。

一、Graph中的执行路径

最简单的 Graph 只有固定顺序:

代码片段Text
load

process

save

直接使用普通 Edge:

代码片段Python
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 中更常见的是分支:

代码片段Text
analyze

   ├── 条件满足 ──→ process

   └── 条件不满足 → fallback

或者:

代码片段Text
model

  ├── 有 Tool Call ──→ tools ──→ model

  └── 无 Tool Call ──→ END

这些路径无法在构建 Graph 时完全确定,需要等 Node 执行结束后,根据当前 State 再决定下一步。

如果需要根据 State 判断决定下一步执行哪个节点,就要使用动态路由。

二、Conditional Edge

Conditional Edge,也就是条件边,用于根据 State 动态选择后续节点。

对应的方法是:

代码片段Python
StateGraph.add_conditional_edges(
    source,
    path,
    path_map=None,
)

三个参数分别是:

  • source:从哪个 Node 开始分支
  • path:路由函数
  • path_map:可选,把路由结果映射到具体 Node

例如:

代码片段Python
builder.add_conditional_edges(
    "check",
    route_after_check,
)

这里的 route_after_check 是一个普通 Python 函数:

代码片段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 用来限制函数可能返回的值:

代码片段Python
Literal["process", "fallback"]

表示这里只会返回 "process""fallback"

对于 LangGraph,这个类型信息也有实际作用。它可以帮助 Graph 推断条件边可能指向哪些节点,生成结构图时会更准确。

Conditional Edge 的执行顺序是:

代码片段Text
执行 source Node

更新 State

执行路由函数

读取最新 State

确定下一个 Node

需要注意的是,路由函数读取的是 source Node 更新之后的 State。

因此,Node 负责产生用于判断的结果,Router 只负责根据结果分流。

三、Conditional Edge示例

下面示例展示了根据文本长度选择不同的处理节点。

代码片段Python
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
{
    'text': 'LangGraph',
    'length': 9,
    'result': '使用简短文本处理流程'
}

节点 analyze 先计算文本长度,并返回:

代码片段Python
{"length": 9}

LangGraph 将结果写入 State。

随后执行 route_by_length。因为 length 小于 10,所以返回 "short_text",Graph 进入 short_text 节点。

这里两个函数的职责比较清楚:

代码片段Text
analyze
计算长度

route_by_length
根据长度选择路径

如果不希望 Router 直接返回 Node 名称,也可以使用 path_map

代码片段Python
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 就是一个典型例子:

代码片段Text
model

  ├── 有 Tool Call ──→ tools
  │                      │
  │                      └──→ model

  └── 无 Tool Call ──→ END

tools → model 是固定路径。

model 后面走哪条路径,则取决于模型返回结果:

代码片段Python
def route_after_model(
    state: State,
) -> Literal["tools", "__end__"]:
    last_message = state["messages"][-1]
 
    if last_message.tool_calls:
        return "tools"
 
    return END

然后注册条件边:

代码片段Python
builder.add_conditional_edges(
    "model",
    route_after_model,
)
 
builder.add_edge("tools", "model")

STARTEND 是 LangGraph 内置的特殊节点,分别表示 Graph 的入口和结束位置。

从 Graph 角度看,Agent Loop 并没有特殊之处:

代码片段Text
model

检查 State

tools 或 END

tools 执行

重新回到 model

Agent 是否继续执行,最终还是由明确的条件决定。

实际项目中通常要考虑这些条件:

  • 是否存在 Tool Call
  • Tool 是否执行成功
  • 是否需要重试
  • 当前结果是否满足结束条件
  • 是否超过最大执行次数

只要存在循环,就应该有明确的退出条件。

否则 Router 一直返回前序节点,Graph 就会持续运行。

五、Command

Conditional Edge 的写法是:

代码片段Text
Node
负责更新 State

Router
负责选择路径

有些场景没必要拆成两步。

例如审核节点:

代码片段Text
审核通过
→ decision = approved
→ publish

审核失败
→ decision = rejected
→ revise

这里 State 更新和路径选择来自同一个判断。

这种情况可以直接返回 Command

Command 定义在:

代码片段Python
from langgraph.types import Command

常见写法:

代码片段Python
Command(
    update=...,
    goto=...,
    graph=...,
    resume=...,
)

各参数作用如下:

参数作用常见用法
update更新当前 Graph 的 StateNode 执行完成后写入新的状态,例如审核结果、重试次数
goto指定接下来执行的 Node根据当前处理结果跳转到 publishretrytools 等节点
graph指定 goto 应作用于哪个 GraphSubgraph 中需要跳转到父 Graph 时使用,例如 Command.PARENT
resume为暂停的 Graph 提供恢复值配合 interrupt() 使用,在人工确认或外部输入完成后继续执行

其中最常用的是 updategoto

代码片段Python
return Command(
    update={"decision": "approved"},
    goto="publish",
)

这表示当前 Node 执行完成后,将 decision 更新为 "approved",然后继续执行 publish

graphresume 的使用场景更具体。graph 主要用于 Subgraph 与父 Graph 之间的跳转,resume 则用于 Interrupt 恢复。本文先理解它们的用途,后续介绍 Subgraph 和 Human-in-the-loop 时再展开。

本质上,一次 Node 执行就完成两件事:

代码片段Text
更新 State
+
决定下一步

Command 这种写法比较适合状态变化和路由本来就是同一次业务判断的情况。

六、Command示例

继续看审核流程:

代码片段Python
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)

这里有一个新的类型写法:

代码片段Python
Command[Literal["publish", "revise"]]

它表示 review 返回的是 Command,并且可能跳转到 "publish""revise"

上例运行结果:

代码片段Text
{
    'score': 92,
    'decision': 'approved',
    'result': '进入发布流程'
}

score 为 92,所以 review 返回:

代码片段Python
Command(
    update={"decision": "approved"},
    goto="publish",
)

LangGraph 先更新 decision,然后执行 publish

这里不需要额外再配置 Conditional Edge。

七、Conditional Edge与Command

两种方式都能做动态路由,但适合的场景不同。

对比项Conditional EdgeCommand
路由位置独立 RouterNode 内
State 更新Node 返回update
路径选择Router 返回goto
适合场景根据已有 State 分流更新 State 后立即跳转
流程结构更直观更集中
跨 Graph 路由不适合支持

如果只是读取已有 State,再决定下一步:

代码片段Python
def route(state):
    if state["has_error"]:
        return "retry"
 
    return "finish"

使用 Conditional Edge 更合适。

如果判断过程中同时产生新的状态:

代码片段Python
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,然后返回路径:

代码片段Python
def route(state):
    if state["validation_failed"]:
        return "retry"
 
    return "continue"

如果 Router 里面还要调用模型、访问数据库或者请求外部 API,那么这些逻辑更适合放进 Node。

Router 越简单,Graph 的执行路径越清晰。

判断数据放进State

路由需要使用的数据,应显式保存在 State 中,例如:

代码片段Text
validation_status
retry_count
tool_result
classification
permission

例如需要根据重试次数决定是否继续,就应该把 retry_count 放进 State,而不是使用模块级变量保存。

这样在查看 State 时,就能知道一次执行为什么进入某个分支。

Node出口不要混用

如果一个 Node 已经返回:

代码片段Python
Command(goto="publish")

又给它配置:

代码片段Python
builder.add_edge("review", "save")

需要特别注意。

Command 不会自动覆盖已经存在的 Edge。

因此,一个 Node 最好明确使用哪种方式控制后续路径:

代码片段Text
固定路径
→ 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,以及一次执行结束以后如何继续使用之前的状态。