理解 LangGraph 的核心模型：State、Node、Edge 与 StateGraph

上一篇已经介绍了 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 }

它读取两个字段：

pricequantity

最终只返回：

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有哪些 NodeNode 怎样连接从哪里开始在哪里结束

例如：

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() 之前主要是在定义 Graphcompile() 之后才开始运行 Graph

六、Graph 的执行过程

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

代码如下：

from typing_extensions import TypedDictfrom 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.0apply_discount: 30.0calculate_total: 270.0{'price': 100.0, 'quantity': 3, 'subtotal': 300.0, 'discount': 30.0, 'total': 270.0}

这个例子计算逻辑本身很简单，重点在 State 的变化。

开始执行时：

price = 100quantity = 3subtotal = 0discount = 0total = 0

Graph 从START进入calculate_subtotal。

这个 Node 读取：

price = 100quantity = 3

返回：

{ "subtotal": 300.0}

执行完这一步以后，State 中的subtotal更新为：

subtotal = 300

接下来执行apply_discount。

此时它读取到的已经是：

subtotal = 300

因此返回：

{ "discount": 30.0}

State 再次更新。

之后执行calculate_total：

subtotal = 300discount = 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 会变成：

┌→ BA → 判断 └→ 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、条件路由、持久化和中断机制，都是基于这套执行过程继续扩展出来的。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】