Graph API：状态、节点、边与控制流

核心模型

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

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

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

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

State 设计

schema 选择

Pydantic 的已知边界：

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

overall、input、output 与 private schema

可以把 schema 分成：

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

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

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 函数，运行时按类型注解注入以下参数：

函数会被转换为 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 有四个核心参数：

三种使用位置

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
• Use the Graph API