上一篇已经介绍了 LangGraph 为什么会引入 State、Node 和 Edge,也写了一个最小的 StateGraph 示例。
真正用 LangGraph 构建稍复杂一些的 Graph 时,还需要理解这些问题:Node 返回的数据去了哪里?后面的 Node 为什么能读取前面产生的结果?Edge 在其中负责什么?定义完 Graph 以后,为什么还需要 compile()?
这些问题都指向同一件事:State、Node 和 Edge 在一次 Graph 运行中怎样配合。
本篇我们继续深入这些概念,理解它们在实际运行时怎样配合。
一、StateGraph 的结构
通过上一篇我们已初步理解以下三个概念:
State:任务当前有哪些数据
Node:当前步骤执行什么
Edge:完成以后执行哪里
在 StateGraph 中,它们会组合成一个完整的执行结构。
例如有一个订单计算流程:
START
↓
calculate_subtotal
↓
apply_discount
↓
calculate_total
↓
END
整个任务需要维护:
price: 商品单价
quantity: 购买数量
subtotal: 小计金额
discount: 折扣优惠
total: 总计应付
这些数据属于 State。
三个计算步骤分别是三个 Node。
Node 之间的执行顺序由 Edge 描述。
对应到代码,大致就是:
builder = StateGraph(OrderState)
builder.add_node("calculate_subtotal", calculate_subtotal)
builder.add_node("apply_discount", apply_discount)
builder.add_node("calculate_total", calculate_total)
builder.add_edge(START, "calculate_subtotal")
builder.add_edge("calculate_subtotal", "apply_discount")
builder.add_edge("apply_discount", "calculate_total")
builder.add_edge("calculate_total", END)
Graph 与普通函数的执行有一个很重要的区别。
普通函数调用经常是:
result = function_a()
function_b(result)
function_a 的返回值直接传给 function_b。
LangGraph 中更常见的运行过程是:
Node A
↓
返回 State 的部分更新
↓
更新当前 State
↓
进入下一个 Node
↓
Node B 读取更新后的 State
因此,Node 之间共享的主要是 State。
Edge 负责描述执行关系,并不承担普通函数调用中“传递返回值”的职责。
二、State 的数据范围
下面继续使用订单示例。
State 可以定义为:
from typing_extensions import TypedDict
class OrderState(TypedDict):
price: float
quantity: int
subtotal: float
discount: float
total: float
再用它创建 StateGraph:
from langgraph.graph import StateGraph
builder = StateGraph(OrderState)
这里的 OrderState 描述了这张 Graph 运行期间会使用哪些状态字段。
例如计算小计的 Node:
def calculate_subtotal(state: OrderState):
subtotal = state["price"] * state["quantity"]
return {
"subtotal": subtotal
}
它读取两个字段:
price
quantity
最终只返回:
subtotal
假设执行前的 State 是:
{
"price": 100.0,
"quantity": 3,
"subtotal": 0.0,
"discount": 0.0,
"total": 0.0,
}
Node 返回:
{
"subtotal": 300.0
}
这里返回的并不是完整 State,只包含这一步产生的更新。
执行完成后,当前 State 变成:
{
"price": 100.0,
"quantity": 3,
"subtotal": 300.0,
"discount": 0.0,
"total": 0.0,
}
后面的 Node 读取到的已经是这份更新后的状态。这就是前一篇提到的 Partial State Update。
State 是 Node 之间的数据约定
在实际项目里,可以把 State 看成各个 Node 共同使用的一份数据约定。
例如计算折扣的节点只需要读取 subtotal:
def apply_discount(state: OrderState):
discount = 30.0 if state["subtotal"] >= 300 else 0.0
return {
"discount": discount
}
它不需要知道 subtotal 是怎样计算出来的,也不需要直接调用:
calculate_subtotal()
两个 Node 的依赖关系通过 State 建立:
calculate_subtotal
↓
更新 subtotal
↓
State
↓
读取 subtotal
↓
apply_discount
这样做的意义在简单程序中并不明显。
如果只有两三个函数,直接调用完全没有问题。
但当一个结果后面可能被多个步骤使用,或者任务需要暂停、恢复、重试时,把跨步骤的数据统一放进 State,会比散落在函数参数、局部变量和数据库字段中更易管理与维护。
三、Node 的执行范围
Node 是 StateGraph 中实际执行业务代码的地方。
最常见的 Node 就是一个普通 Python 函数:
def calculate_subtotal(state: OrderState):
return {
"subtotal": state["price"] * state["quantity"]
}
然后注册到 Graph:
builder.add_node(
"calculate_subtotal",
calculate_subtotal
)
Node 可以完成的工作没有限定为 LLM 调用。
例如:
调用模型
执行 Tool
查询数据库
调用 HTTP API
读取文件
检查业务规则
转换数据
生成结果
都可以放进 Node。
因此,Agent 中的一个模型调用可以是 Node,一个 Tool 执行可以是 Node,一段完全没有 AI 调用的普通 Python 函数同样可以是 Node。
Node 适合表示完整步骤
写 Graph 时很容易出现一个倾向:函数既然可以成为 Node,就把每个函数都注册成 Node。
例如文本处理代码:
strip_text
↓
lower_text
↓
remove_spaces
↓
validate_text
如果这些操作每次都会连续执行,也没有单独重试、路由或保存状态的需求,拆成四个 Node 只会让 Graph 变长。
放在一个 Node 中更清楚:
def normalize_text(state):
text = state["text"].strip()
text = text.lower()
text = " ".join(text.split())
return {
"text": text
}
实际设计时,Node 更适合对应一个相对完整的执行步骤。
例如研究 Agent 中的:
搜索资料
↓
评估资料
↓
生成报告
↓
人工审核
这些步骤各自都有明确的输入、输出和运行意义,后续也可能分别加入重试、路由或者人工确认。
这样的步骤成为独立 Node,会让 Graph 更容易阅读和维护。
Node 只负责当前步骤
还有一个细节很重要。
下面这个 Node:
def calculate_subtotal(state: OrderState):
return {
"subtotal": state["price"] * state["quantity"]
}
只负责计算 subtotal。
函数内部没有:
return apply_discount(...)
也没有处理:
计算完成后应该执行哪个 Node
流程由 Graph 本身管理。
因此在设计 Node 时,可以尽量保持一个清楚的边界:
读取当前需要的数据
↓
完成当前工作
↓
返回产生的状态更新
至于后面执行什么,由 Edge 或其他路由机制负责。
业务处理和流程控制分开以后,一个 Node 的代码通常也会更容易测试。
四、Edge 的执行关系
普通 Edge 描述固定的执行顺序。
例如:
builder.add_edge(
"calculate_subtotal",
"apply_discount"
)
表示 calculate_subtotal 完成后,继续进入 apply_discount。
对应的 Graph 是:
calculate_subtotal
↓
apply_discount
这里容易产生一个误解。
calculate_subtotal 返回:
{
"subtotal": 300.0
}
并不是 Edge 把这个字典直接传给 apply_discount。
完整过程是:
calculate_subtotal
↓
返回 {"subtotal": 300}
↓
应用到当前 State
↓
沿 Edge 继续执行
↓
apply_discount
↓
读取更新后的 State
State 和 Edge 因而承担不同职责:
State 保存执行过程中持续变化的数据
Edge 描述 Node 之间的执行关系
在只有固定流程的 Graph 中,这种区别很明显。
后面我们学习 Conditional Edge 后,执行路径还可以根据 State 动态变化,但数据仍然保存在 State 中。
START 和 END
除了普通 Node,Graph 中还经常会看到:
from langgraph.graph import START, END
START 表示 Graph 的入口。
例如:
builder.add_edge(
START,
"calculate_subtotal"
)
表示一次运行从 calculate_subtotal 开始。
END 表示执行结束:
builder.add_edge(
"calculate_total",
END
)
因此整张图是:
START
↓
calculate_subtotal
↓
apply_discount
↓
calculate_total
↓
END
START 和 END 不需要实现对应的 Python 函数,也不负责处理业务数据。
它们主要用于描述 Graph 的边界。
五、StateGraph 的构建过程
State、Node 和 Edge 最终都定义在 StateGraph 中。
创建方式很简单:
builder = StateGraph(OrderState)
这里使用 builder 作为变量名,会比直接叫 graph 更容易区分后面的两个阶段。
因为此时我们还在描述 Graph:
使用什么 State
有哪些 Node
Node 怎样连接
从哪里开始
在哪里结束
例如:
builder = StateGraph(OrderState)
builder.add_node(
"calculate_subtotal",
calculate_subtotal
)
builder.add_node(
"apply_discount",
apply_discount
)
builder.add_node(
"calculate_total",
calculate_total
)
builder.add_edge(
START,
"calculate_subtotal"
)
builder.add_edge(
"calculate_subtotal",
"apply_discount"
)
builder.add_edge(
"apply_discount",
"calculate_total"
)
builder.add_edge(
"calculate_total",
END
)
这段代码只是完成 Graph 的定义。
定义完成以后,还需要:
graph = builder.compile()
之后才能:
graph.invoke(...)
构建与运行是两个阶段
Graph 的构建、编译、执行流程如下:
StateGraph
↓
定义 Node 和 Edge
↓
compile()
↓
CompiledStateGraph
↓
invoke()
StateGraph 是 Graph 的构建对象。
compile() 会生成真正可以执行的 Graph。
所以常见代码结构是:
builder = StateGraph(State)
builder.add_node(...)
builder.add_edge(...)
graph = builder.compile()
result = graph.invoke(...)
后面的 Persistence、Interrupt 等能力也会和编译后的 Graph 运行过程发生关系。
整个流程可以这样简单区分:
compile() 之前主要是在定义 Graph
compile() 之后才开始运行 Graph
六、Graph 的执行过程
我们使用上面的示例完整说明 Graph 的执行过程,这里并不会引入 LLM 和 Tool,避免其他组件影响对 StateGraph 的理解。
代码如下:
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class OrderState(TypedDict):
price: float
quantity: int
subtotal: float
discount: float
total: float
def calculate_subtotal(state: OrderState):
subtotal = state["price"] * state["quantity"]
print(f"calculate_subtotal: {subtotal}")
return {
"subtotal": subtotal
}
def apply_discount(state: OrderState):
discount = 30.0 if state["subtotal"] >= 300 else 0.0
print(f"apply_discount: {discount}")
return {
"discount": discount
}
def calculate_total(state: OrderState):
total = state["subtotal"] - state["discount"]
print(f"calculate_total: {total}")
return {
"total": total
}
builder = StateGraph(OrderState)
builder.add_node(
"calculate_subtotal",
calculate_subtotal
)
builder.add_node(
"apply_discount",
apply_discount
)
builder.add_node(
"calculate_total",
calculate_total
)
builder.add_edge(
START,
"calculate_subtotal"
)
builder.add_edge(
"calculate_subtotal",
"apply_discount"
)
builder.add_edge(
"apply_discount",
"calculate_total"
)
builder.add_edge(
"calculate_total",
END
)
graph = builder.compile()
result = graph.invoke(
{
"price": 100.0,
"quantity": 3,
"subtotal": 0.0,
"discount": 0.0,
"total": 0.0,
}
)
print(result)
上例运行结果:
calculate_subtotal: 300.0
apply_discount: 30.0
calculate_total: 270.0
{'price': 100.0, 'quantity': 3, 'subtotal': 300.0, 'discount': 30.0, 'total': 270.0}
这个例子计算逻辑本身很简单,重点在 State 的变化。
开始执行时:
price = 100
quantity = 3
subtotal = 0
discount = 0
total = 0
Graph 从 START 进入 calculate_subtotal。
这个 Node 读取:
price = 100
quantity = 3
返回:
{
"subtotal": 300.0
}
执行完这一步以后,State 中的 subtotal 更新为:
subtotal = 300
接下来执行 apply_discount。
此时它读取到的已经是:
subtotal = 300
因此返回:
{
"discount": 30.0
}
State 再次更新。
之后执行 calculate_total:
subtotal = 300
discount = 30
得到:
total = 270
最终到达 END。
把状态变化连起来,就是:
初始 State
↓
calculate_subtotal
↓
subtotal = 300
↓
apply_discount
↓
discount = 30
↓
calculate_total
↓
total = 270
↓
最终 State
从这个结果中,可以明确看到两个过程。
一个是 State 持续更新:
State
↓
State
↓
State
另一个是 Graph 按照 Edge 持续推进:
Node
↓
Node
↓
Node
两者同时发生,构成了一次完整的 Graph Run。
七、Graph Run 的运行方式
对于前面的线性 Graph,一次:
graph.invoke(...)
大致经历下面的过程:
接收输入 State
↓
找到入口 Node
↓
执行 Node
↓
得到 State Update
↓
更新当前 State
↓
根据 Edge 进入后续 Node
↓
继续执行
↓
Graph 结束
↓
返回最终 State
整个过程总结如下:
第一,Graph 中始终存在一份当前 State。
第二,每个 Node 都在当前 State 的基础上执行,并产生新的状态更新。
第三,Graph 根据已经定义的 Edge,决定哪些 Node 后续可以继续执行。
后面的很多 LangGraph 功能都建立在这套运行方式上。
例如,普通 Edge 表示:
A → B
Conditional Edge 会变成:
┌→ B
A → 判断
└→ C
Reducer处理的问题是,一个 State 字段收到更新以后,新旧数据应该怎样合并。
Persistence处理的是,某一步执行完成以后,State 怎样保存下来。
Interrupt处理的是,Graph 执行过程中怎样暂停,并在以后继续。
它们处理的问题不同,但都发生在同一套 Graph 运行模型中。
八、实际建模边界
理解 State、Node 和 Edge 并不难。
实际项目里更容易出现问题的地方,是 Graph 怎么拆。
如果粒度过细,一张简单的工作流很快会出现十几个甚至几十个 Node,阅读成本反而比普通代码更高。
因此设计 Node 时,可以从“这一步是否需要被 Graph 单独管理”来判断。
例如:
查询资料
调用模型
执行 Tool
人工审核
写入外部系统
这些步骤通常具有独立运行意义。
某一步可能失败,需要重试;某一步可能需要根据结果走不同路径;人工审核还可能暂停很长时间。
这类步骤比较适合作为 Node。
而下面这样的实现细节:
字符串 trim
大小写转换
格式整理
简单字段计算
如果只属于某个步骤内部的处理,通常没有必要单独做成 Node。
State 也采用相同原则。
任务需要跨步骤保存的数据进入 State。
当前函数内部的临时变量留在函数内部。
Edge 则只表达有实际流程意义的节点关系。
这样设计出来的 Graph 通常会比较稳定:
State
保存跨步骤的数据
Node
表示需要独立运行和管理的步骤
Edge
描述步骤之间的执行关系
StateGraph
组织整个工作流
总结
本文详细说明了 Graph 中 State、Node 和 Edge 这些对象在运行时怎样配合。
一次 Graph 的执行过程中,Node 读取当前 State,完成当前步骤并返回部分状态更新;LangGraph 将更新应用到 State,然后按照 Edge 继续执行后续 Node。StateGraph 负责定义整个结构,经过 compile() 后得到可以执行的 Graph。
后面的 Reducer、条件路由、持久化和中断机制,都是基于这套执行过程继续扩展出来的。
社区讨论
参与讨论
有问题或想法?欢迎继续讨论。