上一篇已经介绍了 LangGraph 为什么会引入 State、Node 和 Edge,也写了一个最小的 StateGraph 示例。

真正用 LangGraph 构建稍复杂一些的 Graph 时,还需要理解这些问题:Node 返回的数据去了哪里?后面的 Node 为什么能读取前面产生的结果?Edge 在其中负责什么?定义完 Graph 以后,为什么还需要 compile()

这些问题都指向同一件事:State、Node 和 Edge 在一次 Graph 运行中怎样配合。

本篇我们继续深入这些概念,理解它们在实际运行时怎样配合。

一、StateGraph 的结构

通过上一篇我们已初步理解以下三个概念:

代码片段Text
State:任务当前有哪些数据
Node:当前步骤执行什么
Edge:完成以后执行哪里

StateGraph 中,它们会组合成一个完整的执行结构。

例如有一个订单计算流程:

代码片段Text
START

calculate_subtotal

apply_discount

calculate_total

END

整个任务需要维护:

代码片段Text
price: 商品单价
quantity: 购买数量 
subtotal: 小计金额
discount: 折扣优惠
total: 总计应付

这些数据属于 State。

三个计算步骤分别是三个 Node。

Node 之间的执行顺序由 Edge 描述。

对应到代码,大致就是:

代码片段Python
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 与普通函数的执行有一个很重要的区别。

普通函数调用经常是:

代码片段Python
result = function_a()
function_b(result)

function_a 的返回值直接传给 function_b

LangGraph 中更常见的运行过程是:

代码片段Text
Node A

返回 State 的部分更新

更新当前 State

进入下一个 Node

Node B 读取更新后的 State

因此,Node 之间共享的主要是 State。

Edge 负责描述执行关系,并不承担普通函数调用中“传递返回值”的职责。

二、State 的数据范围

下面继续使用订单示例。

State 可以定义为:

代码片段Python
from typing_extensions import TypedDict
 
 
class OrderState(TypedDict):
    price: float
    quantity: int
    subtotal: float
    discount: float
    total: float

再用它创建 StateGraph

代码片段Python
from langgraph.graph import StateGraph
 
builder = StateGraph(OrderState)

这里的 OrderState 描述了这张 Graph 运行期间会使用哪些状态字段。

例如计算小计的 Node:

代码片段Python
def calculate_subtotal(state: OrderState):
    subtotal = state["price"] * state["quantity"]
 
    return {
        "subtotal": subtotal
    }

它读取两个字段:

代码片段Text
price
quantity

最终只返回:

代码片段Text
subtotal

假设执行前的 State 是:

代码片段Python
{
    "price": 100.0,
    "quantity": 3,
    "subtotal": 0.0,
    "discount": 0.0,
    "total": 0.0,
}

Node 返回:

代码片段Python
{
    "subtotal": 300.0
}

这里返回的并不是完整 State,只包含这一步产生的更新。

执行完成后,当前 State 变成:

代码片段Python
{
    "price": 100.0,
    "quantity": 3,
    "subtotal": 300.0,
    "discount": 0.0,
    "total": 0.0,
}

后面的 Node 读取到的已经是这份更新后的状态。这就是前一篇提到的 Partial State Update。

State 是 Node 之间的数据约定

在实际项目里,可以把 State 看成各个 Node 共同使用的一份数据约定。

例如计算折扣的节点只需要读取 subtotal

代码片段Python
def apply_discount(state: OrderState):
    discount = 30.0 if state["subtotal"] >= 300 else 0.0
 
    return {
        "discount": discount
    }

它不需要知道 subtotal 是怎样计算出来的,也不需要直接调用:

代码片段Python
calculate_subtotal()

两个 Node 的依赖关系通过 State 建立:

代码片段Text
calculate_subtotal

更新 subtotal

      State

读取 subtotal

 apply_discount

这样做的意义在简单程序中并不明显。

如果只有两三个函数,直接调用完全没有问题。

但当一个结果后面可能被多个步骤使用,或者任务需要暂停、恢复、重试时,把跨步骤的数据统一放进 State,会比散落在函数参数、局部变量和数据库字段中更易管理与维护。

三、Node 的执行范围

Node 是 StateGraph 中实际执行业务代码的地方。

最常见的 Node 就是一个普通 Python 函数:

代码片段Python
def calculate_subtotal(state: OrderState):
    return {
        "subtotal": state["price"] * state["quantity"]
    }

然后注册到 Graph:

代码片段Python
builder.add_node(
    "calculate_subtotal",
    calculate_subtotal
)

Node 可以完成的工作没有限定为 LLM 调用。

例如:

代码片段Text
调用模型
执行 Tool
查询数据库
调用 HTTP API
读取文件
检查业务规则
转换数据
生成结果

都可以放进 Node。

因此,Agent 中的一个模型调用可以是 Node,一个 Tool 执行可以是 Node,一段完全没有 AI 调用的普通 Python 函数同样可以是 Node。

Node 适合表示完整步骤

写 Graph 时很容易出现一个倾向:函数既然可以成为 Node,就把每个函数都注册成 Node。

例如文本处理代码:

代码片段Text
strip_text

lower_text

remove_spaces

validate_text

如果这些操作每次都会连续执行,也没有单独重试、路由或保存状态的需求,拆成四个 Node 只会让 Graph 变长。

放在一个 Node 中更清楚:

代码片段Python
def normalize_text(state):
    text = state["text"].strip()
    text = text.lower()
    text = " ".join(text.split())
 
    return {
        "text": text
    }

实际设计时,Node 更适合对应一个相对完整的执行步骤。

例如研究 Agent 中的:

代码片段Text
搜索资料

评估资料

生成报告

人工审核

这些步骤各自都有明确的输入、输出和运行意义,后续也可能分别加入重试、路由或者人工确认。

这样的步骤成为独立 Node,会让 Graph 更容易阅读和维护。

Node 只负责当前步骤

还有一个细节很重要。

下面这个 Node:

代码片段Python
def calculate_subtotal(state: OrderState):
    return {
        "subtotal": state["price"] * state["quantity"]
    }

只负责计算 subtotal

函数内部没有:

代码片段Python
return apply_discount(...)

也没有处理:

代码片段Text
计算完成后应该执行哪个 Node

流程由 Graph 本身管理。

因此在设计 Node 时,可以尽量保持一个清楚的边界:

代码片段Text
读取当前需要的数据

完成当前工作

返回产生的状态更新

至于后面执行什么,由 Edge 或其他路由机制负责。

业务处理和流程控制分开以后,一个 Node 的代码通常也会更容易测试。

四、Edge 的执行关系

普通 Edge 描述固定的执行顺序。

例如:

代码片段Python
builder.add_edge(
    "calculate_subtotal",
    "apply_discount"
)

表示 calculate_subtotal 完成后,继续进入 apply_discount

对应的 Graph 是:

代码片段Text
calculate_subtotal

 apply_discount

这里容易产生一个误解。

calculate_subtotal 返回:

代码片段Python
{
    "subtotal": 300.0
}

并不是 Edge 把这个字典直接传给 apply_discount

完整过程是:

代码片段Text
calculate_subtotal

返回 {"subtotal": 300}

应用到当前 State

沿 Edge 继续执行

apply_discount

读取更新后的 State

State 和 Edge 因而承担不同职责:

代码片段Text
State    保存执行过程中持续变化的数据
Edge     描述 Node 之间的执行关系

在只有固定流程的 Graph 中,这种区别很明显。

后面我们学习 Conditional Edge 后,执行路径还可以根据 State 动态变化,但数据仍然保存在 State 中。

START 和 END

除了普通 Node,Graph 中还经常会看到:

代码片段Python
from langgraph.graph import START, END

START 表示 Graph 的入口。

例如:

代码片段Python
builder.add_edge(
    START,
    "calculate_subtotal"
)

表示一次运行从 calculate_subtotal 开始。

END 表示执行结束:

代码片段Python
builder.add_edge(
    "calculate_total",
    END
)

因此整张图是:

代码片段Text
START

calculate_subtotal

apply_discount

calculate_total

END

STARTEND 不需要实现对应的 Python 函数,也不负责处理业务数据。

它们主要用于描述 Graph 的边界。

五、StateGraph 的构建过程

State、Node 和 Edge 最终都定义在 StateGraph 中。

创建方式很简单:

代码片段Python
builder = StateGraph(OrderState)

这里使用 builder 作为变量名,会比直接叫 graph 更容易区分后面的两个阶段。

因为此时我们还在描述 Graph:

代码片段Text
使用什么 State
有哪些 Node
Node 怎样连接
从哪里开始
在哪里结束

例如:

代码片段Python
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 的定义。

定义完成以后,还需要:

代码片段Python
graph = builder.compile()

之后才能:

代码片段Python
graph.invoke(...)

构建与运行是两个阶段

Graph 的构建、编译、执行流程如下:

代码片段Text
StateGraph

定义 Node 和 Edge

compile()

CompiledStateGraph

invoke()

StateGraph 是 Graph 的构建对象。

compile() 会生成真正可以执行的 Graph。

所以常见代码结构是:

代码片段Python
builder = StateGraph(State)
 
builder.add_node(...)
builder.add_edge(...)
 
graph = builder.compile()
 
result = graph.invoke(...)

后面的 Persistence、Interrupt 等能力也会和编译后的 Graph 运行过程发生关系。

整个流程可以这样简单区分:

代码片段Text
compile() 之前主要是在定义 Graph
compile() 之后才开始运行 Graph

六、Graph 的执行过程

我们使用上面的示例完整说明 Graph 的执行过程,这里并不会引入 LLM 和 Tool,避免其他组件影响对 StateGraph 的理解。

代码如下:

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

上例运行结果:

代码片段Text
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 的变化。

开始执行时:

代码片段Text
price       = 100
quantity    = 3
subtotal    = 0
discount    = 0
total       = 0

Graph 从 START 进入 calculate_subtotal

这个 Node 读取:

代码片段Text
price = 100
quantity = 3

返回:

代码片段Python
{
    "subtotal": 300.0
}

执行完这一步以后,State 中的 subtotal 更新为:

代码片段Text
subtotal = 300

接下来执行 apply_discount

此时它读取到的已经是:

代码片段Text
subtotal = 300

因此返回:

代码片段Python
{
    "discount": 30.0
}

State 再次更新。

之后执行 calculate_total

代码片段Text
subtotal = 300
discount = 30

得到:

代码片段Text
total = 270

最终到达 END

把状态变化连起来,就是:

代码片段Text
初始 State

calculate_subtotal

subtotal = 300

apply_discount

discount = 30

calculate_total

total = 270

最终 State

从这个结果中,可以明确看到两个过程。

一个是 State 持续更新:

代码片段Text
State

State

State

另一个是 Graph 按照 Edge 持续推进:

代码片段Text
Node

Node

Node

两者同时发生,构成了一次完整的 Graph Run。

七、Graph Run 的运行方式

对于前面的线性 Graph,一次:

代码片段Python
graph.invoke(...)

大致经历下面的过程:

代码片段Text
接收输入 State

找到入口 Node

执行 Node

得到 State Update

更新当前 State

根据 Edge 进入后续 Node

继续执行

Graph 结束

返回最终 State

整个过程总结如下:

第一,Graph 中始终存在一份当前 State。

第二,每个 Node 都在当前 State 的基础上执行,并产生新的状态更新。

第三,Graph 根据已经定义的 Edge,决定哪些 Node 后续可以继续执行。

后面的很多 LangGraph 功能都建立在这套运行方式上。

例如,普通 Edge 表示:

代码片段Text
A → B

Conditional Edge 会变成:

代码片段Text
        ┌→ B
A → 判断
        └→ C

Reducer处理的问题是,一个 State 字段收到更新以后,新旧数据应该怎样合并。

Persistence处理的是,某一步执行完成以后,State 怎样保存下来。

Interrupt处理的是,Graph 执行过程中怎样暂停,并在以后继续。

它们处理的问题不同,但都发生在同一套 Graph 运行模型中。

八、实际建模边界

理解 State、Node 和 Edge 并不难。

实际项目里更容易出现问题的地方,是 Graph 怎么拆。

如果粒度过细,一张简单的工作流很快会出现十几个甚至几十个 Node,阅读成本反而比普通代码更高。

因此设计 Node 时,可以从“这一步是否需要被 Graph 单独管理”来判断。

例如:

代码片段Text
查询资料
调用模型
执行 Tool
人工审核
写入外部系统

这些步骤通常具有独立运行意义。

某一步可能失败,需要重试;某一步可能需要根据结果走不同路径;人工审核还可能暂停很长时间。

这类步骤比较适合作为 Node。

而下面这样的实现细节:

代码片段Text
字符串 trim
大小写转换
格式整理
简单字段计算

如果只属于某个步骤内部的处理,通常没有必要单独做成 Node。

State 也采用相同原则。

任务需要跨步骤保存的数据进入 State。

当前函数内部的临时变量留在函数内部。

Edge 则只表达有实际流程意义的节点关系。

这样设计出来的 Graph 通常会比较稳定:

代码片段Text
State
保存跨步骤的数据

Node
表示需要独立运行和管理的步骤

Edge
描述步骤之间的执行关系

StateGraph
组织整个工作流

总结

本文详细说明了 Graph 中 State、Node 和 Edge 这些对象在运行时怎样配合。

一次 Graph 的执行过程中,Node 读取当前 State,完成当前步骤并返回部分状态更新;LangGraph 将更新应用到 State,然后按照 Edge 继续执行后续 Node。StateGraph 负责定义整个结构,经过 compile() 后得到可以执行的 Graph。

后面的 Reducer、条件路由、持久化和中断机制,都是基于这套执行过程继续扩展出来的。