---
type: study
created: 2026-08-11
updated: 2026-08-11
sensitivity: private
status: active
tags:
  - study
  - langchain
  - langgraph
  - python
  - state-machine
topic: LangGraph Graph API 的状态、节点、边与控制流
sources:
  - https://docs.langchain.com/oss/python/langgraph/graph-api
  - https://docs.langchain.com/oss/python/langgraph/use-graph-api
last_verified: 2026-08-11
---
# Graph API：状态、节点、边与控制流

## 核心模型

Graph API 由三个核心元素组成：

1. **State**：当前应用快照及每个字段的合并规则；
2. **Node**：读取 State、执行计算或副作用、返回部分状态更新；
3. **Edge**：决定下一批执行哪些 Node。

一句话记忆：节点做工作，边决定下一步，reducer 决定并发或连续更新如何进入状态。

图必须先 compile 才能执行。compile 会校验结构，并在此注入 checkpointer、store、cache、静态 breakpoint 等运行时能力。

## State 设计

### schema 选择

| schema | 优点 | 代价与边界 | 建议 |
|---|---|---|---|
| TypedDict | 轻量、快、类型提示清晰 | 无运行时递归校验 | 默认首选 |
| dataclass | 支持默认值，仍较轻量 | 运行时校验弱于 Pydantic | 需要默认值时优先 |
| Pydantic BaseModel | 输入运行时校验与类型转换 | 性能较低；图输出不是模型实例；只校验首节点输入 | 仅在确需边界校验时使用 |

Pydantic 的已知边界：

- 校验只发生在进入第一个节点的输入，不覆盖后续节点输出；
- 错误栈不会天然标明具体节点；
- 图的最终返回值不是 Pydantic 实例；
- 递归校验可能成为性能热点；
- 消息字段应使用 AnyMessage，避免跨 wire 序列化问题。

### overall、input、output 与 private schema

可以把 schema 分成：

- **OverallState**：图内部所有 channel 的并集；
- **InputState**：调用者允许传入的字段；
- **OutputState**：调用者最终可见的字段；
- **PrivateState**：仅某些节点之间传递的中间字段。

节点的输入注解限制它读取什么，但节点可以向图已声明的其他 channel 写入。显式 output schema 只过滤 invoke 的返回值。

> [!warning] private 不是安全隔离
> private channel 不会自动从 streaming 中脱敏。values 模式默认可能发出全部 state channel。要限制内容，显式传 output_keys，或使用 updates 只发节点实际更新。敏感字段仍需在协议和观测层单独治理。

### State 只接收部分更新

节点不要原地修改 State，也不需要返回整个 State，只返回变化字段。这样 reducer 才能统一应用更新，checkpoint 的写入语义也更清晰。

## Reducer：状态更新的真正契约

每个 State key 都有独立 reducer。reducer 是二元函数：

- 左参数：该 key 当前累计值；
- 右参数：本次节点返回的新更新；
- 返回值：写回 State 的新值。

### 默认 reducer

未声明 reducer 时，右值覆盖左值。节点返回某个 key 不会影响其他未返回 key。

### 自定义 reducer

TypedDict 字段通过 Annotated 绑定 reducer。列表常用 operator.add，但生产代码要明确：

- 并行写入的合并顺序是否影响结果；
- reducer 是否满足所需结合性；
- 重试或 replay 是否可能重复追加；
- 是否需要稳定 ID 去重。

### 消息 reducer

简单的 operator.add 只能追加，无法按 ID 更新已有消息。消息历史应优先用 add_messages：

- 新消息追加；
- 相同 message ID 的消息替换；
- 输入字典反序列化为 LangChain Message；
- 允许人工修改已有消息。

MessagesState 已预置 messages 字段与 add_messages，可继承后增加业务字段。

### Overwrite

当某字段有 reducer，但某次更新必须完整重置时，用 Overwrite 绕过 reducer。

并行同一 super-step 内，同一个 key 最多只能有一个 Overwrite 写入；多个并行节点同时 Overwrite 会触发 InvalidUpdateError。

## Node 契约

节点可以是同步或异步 Python 函数，运行时按类型注解注入以下参数：

| 参数 | 内容 |
|---|---|
| state | 当前图状态 |
| config | RunnableConfig，包括 thread_id、tags、metadata 等 |
| runtime | context、store、stream writer、execution info、server info、heartbeat、control |

函数会被转换为 RunnableLambda，从而获得 batch、async、tracing 等能力。未显式命名时，节点名默认为函数名。

### Runtime 中的重要信息

- **context**：模型选择、依赖、数据库连接等运行期依赖，不污染持久化 State；
- **store**：跨 thread 长期数据；
- **execution_info**：thread_id、run_id、checkpoint_id、task_id、node_attempt、首次尝试时间；
- **server_info**：在 Agent Server 中提供 assistant_id、graph_id 与已认证用户；
- **heartbeat**：刷新 idle timeout；
- **drain_requested / drain_reason**：收到优雅退出请求后跳过昂贵工作；
- **stream_writer**：从节点发出自定义流事件。

server_info 在本地直接运行图时为 None。业务代码必须显式兼容本地和 Server 两种上下文。

### 节点重执行与幂等

checkpointer 在 super-step 边界保存完整 checkpoint，不会在一个普通函数中间保存执行位置。节点在 interrupt、失败恢复或重试后，会从函数开头重新执行。

因此外部写操作必须采用：

- idempotency key；
- upsert；
- read-before-write；
- 业务唯一索引；
- 把不可重复副作用拆成独立 task 或节点。

如果一个节点内部包含多个独立操作，可把每个操作包装成 Functional API task。已完成 task 的结果会在恢复时复用，减少整节点重做，但 task 与 interrupt 的顺序必须保持确定。

### Node cache

节点缓存需要同时：

1. 为节点配置 CachePolicy；
2. compile 时配置 cache 实现。

CachePolicy 可定义 key_func 与 ttl。默认 key 基于输入 hash；若输入包含时间、权限或租户信息，必须确认 key 不会造成越权复用。

## Edge 与执行语义

### 四类基础边

- **normal edge**：A 永远到 B；
- **conditional edge**：路由函数根据 State 返回一个或多个目标；
- **START edge**：定义入口；
- **conditional START edge**：根据输入选择不同入口。

END 表示终止节点。推荐使用 add_edge(START, node) 和 add_edge(node, END)，而不是旧式 set_entry_point 与 set_finish_point。

### 一个节点只选一种路由机制

从同一节点出发，不要同时混用：

- 静态 add_edge；
- conditional edge；
- Command.goto；
- 工具返回的动态 goto。

Command 或工具的 goto 只会增加动态边，不会覆盖静态边。混用时多个目标都会执行，常造成隐蔽的重复写入或重复副作用。

### 并行 super-step

一个节点有多个普通 outgoing edges 时，目标节点在下一 super-step 并行执行。fan-in 节点会在其上游并行分支完成后运行。

关键语义：

- 同一 super-step 的 State 更新整体提交；
- 某一分支失败时，本轮完整 state update 不提交；
- 配置 checkpointer 后，成功分支的 task writes 会保留为 pending writes；
- 恢复时只重跑失败分支；
- 并行 reducer 的更新顺序不保证稳定；
- 若顺序影响业务，应把排序键与结果一起写入，汇总后显式排序。

max_concurrency 控制图调用的最大并发任务数，不能只依赖下游 API 自己限流。

### defer

当 fan-out 分支长度不同，汇总节点可能需要等所有 pending tasks 完成。把汇总节点设为 defer，表示它在其他待执行任务全部结束后再运行，适合 map-reduce 的最终汇总。

## 动态分发：Send

Send 解决“构图时不知道 worker 数量”的问题。conditional edge 返回若干 Send，每个 Send 指定：

- 目标节点名；
- 传给该次 worker 的独立 State。

worker 输出通常写入带 reducer 的共享 key，随后由汇总节点合成结果。适用于：

- 动态报告章节；
- 未知数量的文档分析；
- 按输入规模展开的批处理；
- orchestrator-worker。

Send 还可以为某次动态调用覆盖节点默认 timeout。大规模 fan-out 必须同时设计：

- max_concurrency；
- 单项 timeout；
- 重试预算；
- 结果去重；
- 汇总节点对部分失败的策略。

## Command：状态更新与控制流合一

Command 有四个核心参数：

| 参数 | 用途 |
|---|---|
| update | 更新 State |
| goto | 跳转到指定节点 |
| graph | 指向父图，常用 Command.PARENT |
| resume | 为 interrupt 提供恢复值 |

### 三种使用位置

1. **节点返回值**：update + goto，在同一步内更新并路由；
2. **invoke、stream 输入**：只用 resume，可同时带 update；
3. **工具返回值**：更新 graph state，并可动态路由。

节点返回 Command 时应使用 Command[Literal[...]] 类型注解列出可能目标，否则图可视化无法识别动态边。

只需要路由、不更新 State 时，优先 conditional edge；需要更新并路由时才用 Command。

### Command.PARENT

子图节点可以返回 graph=Command.PARENT 跳到最近父图节点。若同时更新父子图共享 key，父图 State 必须为该 key 配置 reducer，否则跨图更新无法正确合并。

### Command.resume

作为 invoke 或 stream 输入时，只有 Command(resume=...) 是预期模式。不要用 Command(update=...) 继续一个已结束的普通多轮对话：

- 恢复 interrupt：同一 thread_id + Command(resume=value)；
- 普通新一轮对话：同一 thread_id + 普通输入字典。

### 工具返回 Command

工具更新消息历史时，Command.update 必须包含对应 ToolMessage，保证每个 AI tool call 后都有合法工具结果。使用预置 ToolNode 时，它会传播工具返回的 Command；自定义工具节点必须手动处理。

## 顺序、分支与循环

### 顺序

add_sequence 是顺序节点的简写。无论使用简写还是显式 edge，节点边界仍决定 checkpoint、stream、trace 与恢复粒度。

### 条件分支

conditional edge 可以返回一个目标，也可以返回多个目标并行执行。路由函数应该：

- 只读取 State，不产生副作用；
- 输出有限、可类型化；
- 对未知分类有安全默认分支；
- 不把权限决策交给不可信模型文本。

### 循环

循环必须同时具备：

- 业务终止条件，路由到 END；
- recursion_limit 作为防御性上限；
- 对超限的用户可理解降级。

recursion_limit 统计 super-step。当前 step 可从 config metadata 读取，也可以把 RemainingSteps 放入 State，在触顶前进入 graceful fallback。相比外部捕获 GraphRecursionError，RemainingSteps 能让图正常完成、保存中间结果并返回部分答案。

### Async

IO-bound 节点应改为 async def，并在内部 await 异步客户端，图调用使用 ainvoke 或 astream。同步阻塞 I/O 不会因为图是异步就自动并发。

节点 timeout 只支持 async 节点；同步节点配置 timeout 会在 compile 时失败。

## 运行期配置

模型、系统提示词、连接实例等每次调用可能变化但不属于业务状态，应通过 context_schema 与 invoke 的 context 参数传递，而不是放入 State。

区分三类数据：

- **State**：要 checkpoint、参与流程恢复的业务数据；
- **Context**：一次 run 内的依赖或配置；
- **Store**：跨 thread 的长期应用数据。

把这些混在一起会带来 checkpoint 膨胀、配置泄漏或恢复后读取过期依赖。

## 节点级可靠性配置

Graph API 可为节点配置：

- retry_policy；
- async timeout 或 TimeoutPolicy；
- cache_policy；
- error_handler。

set_node_defaults 可设全图默认值，节点显式配置优先。默认值不继承到 subgraph。error handler 不会捕获自己，且其结果不应缓存；具体生命周期见 [[03-持久化记忆与容错]]。

## 图变更的技术边界

已有 checkpoint 时：

- 已完成 thread 可以修改全部拓扑；
- interrupted thread 可以增删边、增加节点；
- interrupted thread 不能安全地重命名或删除即将进入的节点；
- State 新增或删除 key 通常兼容；
- State key 重命名会让旧值丢失；
- key 类型不兼容改变可能使旧 checkpoint 无法加载。

业务语义是否应对老 thread 生效是另一层问题，详见 [[05-生产化测试迁移与案例]]。

## 最小设计审查清单

- State 是否只存原始、需要恢复的数据？
- 每个 key 的 reducer 是否明确，是否能处理并发写入？
- 消息是否使用 add_messages，而不是无脑追加？
- private 字段是否可能被 values stream 泄漏？
- 节点副作用是否幂等？
- 同一节点是否混用了静态边和 Command.goto？
- fan-out 是否设置并发、timeout 与部分失败策略？
- 循环是否同时有业务终止条件和 recursion_limit？
- runtime context、State 与 Store 是否分层？
- interrupted thread 上线前是否检查节点名和 State 兼容？

## 关联笔记

- [[01-LangGraph基础与API选择]]
- [[03-持久化记忆与容错]]
- [[04-流式HITL与子图]]
- [[05-生产化测试迁移与案例]]

## 来源

- [Graph API overview](https://docs.langchain.com/oss/python/langgraph/graph-api)
- [Use the Graph API](https://docs.langchain.com/oss/python/langgraph/use-graph-api)