LangGraph 实际构建机制：Runnables、Channels 与 Agent 架构解析

我花了整个周末深入 LangGraph 的源代码，试图理解抽象层之下实际发生了什么。以下是我的笔记、运行的实验以及最终形成的思维模型。

最让我惊讶的是其核心概念之简洁。图（Graph）、ReAct 智能体和深度智能体（Deep Agent）并非三种不同的机器。它们是一台机器（Pregel），只是配置量不同而已。

以下内容均在离线环境下运行：无网络、无实时模型，使用脚本化的模拟聊天模型。版本锁定为 langgraph 1.2.12、langchain-core 1.6.6、langchain 1.4.3、deepagents 0.7.20；内部实现在不同版本间会发生变化，因此请将类名视为快照。代码位于 samples 仓库中：langgraph_primitives_demo.py 生成了第 1–6 节中的所有打印输出，structured_cases.py 涵盖了第 5c 节。

开始前的旁注：Python 片段与 C# 对应项

如果你来自 C# 背景，以下几个 Python/LangGraph 术语会立刻出现。它们是普通的 Python 或普通的 LangGraph。

有两个 LangGraph 术语值得你现在在脑海中固定下来。

State（状态）是在图中流动的那个字典。每个节点接收它并返回对其的部分更新，而不是全新的状态。如果你熟悉 C#，可以将其想象为不可变记录加上 with { ... } 表达式，由引擎为你应用。

Channel（通道）是该状态的一个命名槽位，它拥有新值如何合并的规则。状态键 total 变成一个名为 total 的通道。规则在多个节点在同一轮次写入时至关重要：是新值替换旧值，还是将它们合并？用 C# 术语来说，通道大致是一个字段及其自身的合并函数配对，例如 Func merge。

如果你了解 C#，这个名字具有误导性。它不是 System.Threading.Channels：没有队列供消费者等待，也没有阻塞行为。它不是 Rx 的 IObservable：没有向订阅者推送值。它也不是 TPL Dataflow 块，尽管那是最接近的家族，因为“当某个依赖项被更新时节点运行”这一想法与数据流触发是相同的。最贴近的字面图像是共享状态对象中的一个带版本号的单元格：一个字段，其 Update(IEnumerable writes) 方法决定写入如何组合，加上一个随变化而递增的版本号。节点通过比较这些版本号被唤醒，而不是通过接收消息。（两个例外表现得更像队列：Topic 用于 Send，收集多个值；EphemeralValue 仅保留一轮的值。）

Reducer（归约器）就是那个合并函数：reducer(old, new) -> combined。你可以通过 Annotated 附加一个。如果你用过 LINQ 的 Aggregate，它的形状是一样的：将每个新值折叠到运行结果中。operator.add 作为列表上的归约器意味着“将新项目追加到旧项目之后”。

有了这些基础，这里有一个最小的有趣图。

1. 唯一的契约：Runnable

Runnable 是任何具备 invoke、ainvoke、stream、astream、batch 和 with_config 方法的东西。仅此而已。在 C# 中，你会称之为所有类都实现的一个接口，即 IRunnable。| 运算符将两个 Runnable 组合在一起（Python 允许类重载 | 运算符，类似于 C# 中的 | 运算符）：

chain = RunnableLambda(lambda x: x + 1) | RunnableLambda(lambda x: x * 10)
type(chain).__name__   # RunnableSequence
chain.invoke(1)        # 20

这之所以重要：编译后的 LangGraph 也是一个 Runnable。其 MRO（方法解析顺序，即 Python 的继承链）以 CompiledStateGraph → Pregel → PregelProtocol 开头。（MRO 是 Method Resolution Order，Python 的继承链：CompiledStateGraph : Pregel : PregelProtocol。）因此，整个图可以嵌入到链中，或者嵌入到另一个图的节点中。这就是后续子代理的工作原理。

1. compile() 生成什么

取一个最小的有趣图例：两个节点，其状态包含一个 overwrite 字段和一个 append 字段。节点只是一个接受状态并返回部分更新的函数：

import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class S(TypedDict):
    total: int                              # 裸键：每个超步一次写入
    visited: Annotated[list, operator.add]  # 归约器：追加

def a(s):                               # 节点 "a"：使 n 翻倍，记录 "a"
    return {"total": s["total"] * 2, "visited": ["a"]}

def b(s):                               # 节点 "b"：使 n 加 1，记录 "b"
    return {"total": s["total"] + 1, "visited": ["b"]}

g = StateGraph(S)
g.add_node("a", a)
g.add_node("b", b)
g.add_edge(START, "a")
g.add_edge("a", "b")
g.add_edge("b", END)
app = g.compile()
app.invoke({"total": 10, "visited": []})        # {'total': 21, 'visited': ['a', 'b']}

StateGraph 仅是一个构建器（类似于 C# 中 BuildServiceProvider() 之前的 ServiceCollection）。compile() 将其降低为基本原语。以下是演示程序针对编译后对象打印的内容：

这些行并非上述图代码的一部分。它们来自我编写的一个小型检查辅助函数（在演示脚本中为 show()）。它读取编译对象的公共属性 app.channels 和 app.nodes：

# `app` 是上面编译后的图（一个 Pregel 对象）
print("channels:", {name: type(ch).__name__ for name, ch in app.channels.items()
                    if not name.startswith("branch:")})    # 隐藏每边的通道以简洁起见
for name, node in app.nodes.items():
    if name == "__start__":
        continue                                            # 内部输入节点，为简洁起见跳过
    print(f"node {name!r}: triggers={node.triggers} bound={type(node.bound).__name__}"
          f" writers={len(node.writers)}")

channels: {'total': 'LastValue', 'visited': 'BinaryOperatorAggregate',
           '__start__': 'EphemeralValue', '__pregel_tasks': 'Topic'}
node 'a': triggers=['branch:to:a'] bound=RunnableCallable writers=2
node 'b': triggers=['branch:to:b'] bound=RunnableCallable writers=1

以 node 开头的行描述的是 PregelNode 对象，那么这里就是一个 PregelNode 的样子。这是来自 langgraph/pregel/_read.py 的类，已精简至此处相关的字段（我省略了错误处理程序和子图字段，文档字符串是我根据源代码浓缩而成的）：

class PregelNode: # 一个容器，而非 Runnable。引擎在节点触发时用它来构建可运行的任务。

channels: str | list[str] # 作为输入读取哪些通道。对于节点 "a"：['total', 'visited']， # 因此你的函数接收包含这两个键的字典（即 def a(s) 中的 s） triggers: list[str] # 哪些通道会唤醒节点：如果其中任何一个被写入，节点将在下一步运行。 # 对于节点 "a"：['branch:to:a'] bound: Runnable # 实际逻辑：一个包裹你函数的 Runnable，传入输入进行调用 writers: list[Runnable] # 在 bound 之后运行；每个 writer 获取其输出并将其写入通道 # （一个将返回值存储到状态中，另一个唤醒下一个节点） mapper: Callable | None # bound 之前对输入的可选转换（节点 "a" 为 None）

retry_policy: Sequence[RetryPolicy] | None # 如果节点抛出异常会发生什么（此处为 None） cache_policy: CachePolicy | None # 对相同输入重用结果（此处为 None） timeout: TimeoutPolicy | None # 每次调用的限制（仅限异步节点） tags: Sequence[str] | None # 追踪标签 metadata: Mapping[str, Any] | None # 追踪元数据

注意通道（它读取的内容）和触发器（什么唤醒它）之间的区别。它们被故意分开：一个节点可以读取多个通道，但仅由一个通道唤醒。用 C# 术语来说，PregelNode 是一个记录“我的输入、我的唤醒条件、我的处理程序、我的后处理”的结构，类似于带有过滤器的消息处理器注册，只不过这里的消息是通道更新。

逐行解读打印输出：

• 'total': 'LastValue'：total 字段变成了一个 LastValue 类型的通道。它保存一个值，写入操作会替换该值。这就是 S 中普通的 total: int 的含义。
• 'visited': 'BinaryOperatorAggregate'：visited 字段变成了一个通道，它将旧值和新值通过你的函数（这里是 operator.add）组合起来。这就是 Annotated[list, operator.add] 的含义，也是列表增长的原因。
• '__start__': 'EphemeralValue'：一个内部通道，用于将 invoke(...) 的输入带入图中。EphemeralValue 意味着它仅存在于一轮迭代中。
• '__pregel_tasks': 'Topic'：一个类似队列的内部通道。Send 数据包会进入这里（第 5c 节）；我们的图未使用它，但每个图都有它。
• 节点 'a'：triggers=['branch:to:a']：当通道 branch:to:a 被更新时，节点 a 运行。该通道是边 START → a 转换而成的结果，因此没有边对象，只有“当此通道更改时唤醒 a”。
• bound=RunnableCallable：实际运行的东西。它是对你普通函数 a 的 Runnable 包装器。
• writers=2 vs writers=1：在 a 运行后，有两个写入操作：一个将其返回的更新存储到 total/visited 中，另一个写入 branch:to:b 以唤醒节点 b。b 是最后一个节点，所以它只有一个 writer；它的边指向 END，不会唤醒任何节点。

注意通道列表中没有 branch:to:a 或 branch:to:b，因为我的过滤器隐藏了它们。它们确实存在，每条边对应一个，节点的触发器指向它们。

以下是相同的转换图示。第一个图表是你用 add_edge 绘制的内容；第二个图表是 compile() 实际构建的内容（我在图中添加了 __start__ 输入通道和每个节点的两个 writer，正如打印输出所示）：

你编写的代码（构建器）：

flowchart LR
   S([START]) --> a --> b --> E([END])

compile() 构建的内容（运行时）：

flowchart TD
   I["__start__ channel<br/>(你的输入)"] -->|wakes| CA["channel branch:to:a"]
   CA -->|wakes| NA["node a<br/>reads: total, visited<br/>bound = a(s)<br/>writer1 → total, visited<br/>writer2 → channel branch:to:b"]
   NA -->|wakes| NB["node b<br/>reads: total, visited<br/>bound = b(s)<br/>writer1 → total, visited<br/>(no writer2: its edge goes to END)"]

第二个图中节点之间的箭头消失了。剩下的是方框（节点）和命名的邮箱（通道）：“a”不知道“b”的存在，它只是将消息投递到 branch:to:b 中。

因此，心智模型如下：

• 通道持有状态（参见上方的侧边注释）。每个状态键都成为一个通道，其类型由合并规则决定：LastValue 保留跨步骤的最新值，但每个超级步骤最多接受一次写入（两个并行写入者会引发 InvalidUpdateError，如 5c 所示）；BinaryOperatorAggregate 应用你的归约器（operator.add），这就是为什么 visited 是追加而不是替换。额外的通道承载控制流：__start__ 保存输入，而边 a → b 变为一个名为 branch:to:b 的通道。双下划线仅是内部通道的命名约定（Python 的 __name__ 风格，而非关键字）。
• 节点是 PregelNodes。每个节点都有触发器（哪些通道唤醒它）、通道（它读取哪些内容）、bound（实际的 Runnable，这里是你函数的包装器）以及 writers（运行后写入什么）。
• 边消失了。运行时没有边对象。a → b 的含义是“a 的 writer 写入 branch:to:b，且 b 在 branch:to:b 上被触发”。get_graph() 重建了用于绘图的漂亮边列表，但引擎从不遍历它。

条件边和 Send 使用相同的机制：路由器 Runnable 作为额外的 writer 运行，并写入它选择的任何 branch:to:X 通道（或使用其自己的输入发送任务以进行 map-reduce）。

为了直观展示，向同一图添加一个路由器：现在 a 通过 g.add_conditional_edges("a", route, ["b", "c"]) 选择 b 或 c，其中 route(s) 在 s["total"] > 5 时返回 "b"，否则返回 "c"。作为代码（它复用了上面的 S、a 和 b，并添加了节点 c 和路由器）：

def c(s):                                # 节点 "c"：替代路径，重置 total
    return {"total": 0, "visited": ["c"]}

def route(s):                            # 路由器：返回下一个节点的名称
    return "b" if s["total"] > 5 else "c"

g = StateGraph(S)
g.add_node("a", a)
g.add_node("b", b)
g.add_node("c", c)
g.add_edge(START, "a")
g.add_conditional_edges("a", route, ["b", "c"])   # 替换 add_edge("a", "b")
g.add_edge("b", END)
g.add_edge("c", END)
app = g.compile()
app.invoke({"total": 10, "visited": []})   # {'total': 21, 'visited': ['a', 'b']}  (a: 20 > 5 -> b)
app.invoke({"total": 1,  "visited": []})   # {'total': 0,  'visited': ['a', 'c']}  (a: 2 -> c)

我编译并运行了此代码，并以相同的方式进行了检查：

channels: total=LastValue, visited=BinaryOperatorAggregate, __start__, __pregel_tasks=Topic,
          branch:to:a, branch:to:b, branch:to:c   (最后三个是 EphemeralValue)
node a: triggers=['branch:to:a']  writers=[state write, _route]
node b: triggers=['branch:to:b']  writers=[state write]
node c: triggers=['branch:to:c']  writers=[state write]

你写的：

flowchart LR
   S([START]) --> a
   a --> b --> E1([END])
   a --> c --> E2([END])

compile() 构建的：

flowchart TD
   NA["node a<br/>writer1: state write → total, visited<br/>writer2: _route(s) ← your router function"]
   NA -->|"router returns 'b'"| CB["writes branch:to:b"] -->|wakes| NB[node b]
   NA -->|"router returns 'c'"| CC["writes branch:to:c"] -->|wakes| NC[node c]

与线性情况的区别仅在于 a 的第二个 writer。不再是固定写入 branch:to:b，而是 _route，它调用你的函数并写入答案所命名的任意通道。另一个通道从未被写入，因此该节点根本不会运行。当 total 初始值为 10 时，a 使其变为 20，路由器选择 b，invoke 返回 {'total': 21, 'visited': ['a', 'b']}；c 从未触发。如果从 1 开始，路由器则选择 c。（_route 是我在此版本中看到的内部 writer 的名称；将其视为实现细节。）

1. Pregel：超级步骤循环

Pregel 是一种用于图计算的模型，描述于 2010 年 Google 的一篇论文中（“Pregel: A System for Large-Scale Graph Processing”，SIGMOD 2010）。它建立在更早期的“批量同步并行”概念之上。Apache Giraph、Spark GraphX 和 Flink Gelly 等开源系统提供了相同的以顶点为中心的模型。LangGraph 借鉴了这一思想来运行你的图（它还将其编译后的图类命名为 Pregel，这也是该模块被称为 pregel 的原因）。用一句话概括：工作按轮次进行，在每一轮中，所有就绪的节点在同一状态快照上并行运行，只有当它们全部完成后，其结果才会被合并并决定下一轮的操作。如果用 C# 来类比，可以想象为对就绪节点执行一次 Task.WhenAll(...)，其中每个任务读取相同的快照，并在下一轮之前通过屏障机制合并它们的结果。（LangGraph 的版本是一种适配：Google 的 Pregel 在图的顶点之间传递消息，而 LangGraph 的“顶点”是你的节点，消息则是通道写入。）如果你从事 .NET 开发，你可能已经接触过这种算法：Microsoft Agent Framework 的工作流引擎（包括其 C# 和 Python 版本）使用了其文档所称的“修改版的 Pregel 执行模型——一种基于超步处理的批量同步并行 (BSP) 方法”，因此它的执行器也在相同的超步中运行。有关详细信息，请参阅 Workflow Builder & Execution 中的“Execution Model: Supersteps”。在 langgraph/pregel/_loop.py 中，每一轮包含以下步骤：

1. tick：检查停止条件，然后 prepare_next_tasks 找出自上次运行以来触发通道发生更新的每个节点。
2. 运行这些任务（如果多于一个则并行运行）。任务从本轮开始时读取通道值并返回写入操作；其写入的内容对其兄弟节点不可见。
3. after_tick：apply_writes 通过每个通道的规则将所有写入合并到通道中，递增版本号，发出流输出，并且（如果有检查点器）保存检查点。

重复直到没有节点被触发。使用 invoke 运行双节点图会返回 {'total': 21, 'visited': ['a', 'b']}，而 stream_mode="updates" 会为每个节点显示一个超步：

[{'a': {'total': 20, 'visited': ['a']}}, {'b': {'total': 21, 'visited': ['b']}}]

“先写入，然后在屏障处应用”的分离是最核心的概念。这就是为什么并行分支是安全的（它们无法互相看到），为什么需要归约器（同一轮中两个分支向同一个键写入时需要 Annotated 归约器，否则图会失败），以及为什么轮次之间的检查点是普通通道的一致快照（Deep Agents DeltaChannel 是例外，参见第 5b 节）。

Send：当一个节点必须变成多个任务时

普通边表示“在 a 之后运行 b”，仅运行一次，并以共享状态作为输入。有时你直到运行时才知道需要多少任务：模型请求了三个工具，或者你有 N 份文档需要处理。Send(node, arg) 涵盖了这种情况。它是一个小数据包，意为“运行此节点一次，使用此私有输入”，路由器可以返回整个列表：

使用 stream_mode="debug" 进行流式传输并打印每个任务事件，展示了循环调度了什么（我运行了以下代码）：

task work {'x': 1} step 1
task work {'x': 2} step 1
task work {'x': 3} step 1
{'items': [1, 2, 3], 'out': [10, 20, 30]}

一个节点，三个任务，都在同一个超步中，每个任务都有自己的输入。谁做什么：

• 你（或 create_agent）编写路由器。它仅返回 Send 对象；不执行任何操作。
• Pregel 循环拥有任务。路由器的返回值被写入内部 __pregel_tasks 通道（第 2 节中的 Topic）。在下一个屏障处，循环读取该通道，并为每个数据包为命名节点构建一个任务，以数据包的参数作为其输入（内部调用 prepare_push_task_send，区别于普通的“由通道触发”的任务）。然后，运行器像执行任何其他超级步一样并行执行这些任务。
• 时机：路由器在一个超级步结束时运行；Send 任务在紧接着的下一个超级步中运行。随后它们通过正常通道写入结果，这就是为什么示例需要 operator.add 归约器：三个任务在同一轮次中写出。

C# 中的对应场景是 await Task.WhenAll(items.Select(i => Work(i)))，只不过列表是由图中间的路由器决定的，且每个调用都是一个一等公民任务，拥有自己的检查点写入。这正是 create_agent 为并行工具调用所做的事情，如下节所示。

一句话总结：Pregel 循环以超级步的方式运行你的图，其中超级步是指一轮迭代，在该轮中所有就绪节点都在相同的状态快照上运行，并且它们的所有写入都在下一轮开始前于屏障处合并。

1. 示例：AI 智能体只是一个图

在探讨更复杂的部分（气泡式上升、归约器、持久化）之前，这里展示到目前为止所有内容的应用成果：AI 智能体被表达为一个图。这听起来像是不同的物种，但 create_agent(model, tools=[...]) 返回的是一个 CompiledStateGraph，与第 2 节中的 app 属于同一种对象类型。因此，我们可以阅读它是如何构建的，然后运行它，再查看它生成的图。

4.1 create_agent 的代码

create_agent 并非魔法；它是一个使用与第 2 节相同的构建器调用来创建 StateGraph 并编译它的函数。以下是 langchain/agents/factory.py（langchain 1.4.3）的相关核心代码，由我简化为无中间件且无结构化输出的路径；原始文件约有 2,000 行，标记为 ... 的主体部分已省略。节点和路由器函数完整展示，源自 model_node、_make_model_to_tools_edge 和 _make_tools_to_model_edge 的简化版本：

def create_agent(model, tools, ...): graph = StateGraph(state_schema=AgentState, ...) # 'messages' 等成为通道

# ---- 节点 1：“模型”——一个普通函数节点 ---- def model_node(state, runtime): messages = state["messages"] # 读取通道 if system_message: messages = [system_message, *messages] ai_message = model.bind_tools(tools).invoke(messages) # 调用聊天模型 return {"messages": [ai_message]} # 写入：追加到 'messages' graph.add_node("model", RunnableCallable(model_node))

# ---- 节点 2：“工具”——一个 ToolNode，本身是一个 Runnable ---- tool_node = ToolNode(tools) # 读取最后一条 AIMessage 的 tool_calls，运行每个工具， graph.add_node("tools", tool_node) # 并为每次调用写入一条 ToolMessage

graph.add_edge(START, "model") # 入口点

# ---- “模型”之后的路由器：工具调用 -> “工具”，否则结束 ---- def model_to_tools(state): last_ai = last_ai_message(state["messages"]) if not last_ai.tool_calls: # 智能体循环的经典退出条件 return END pending = [c for c in last_ai.tool_calls if c["id"] not in answered_ids(state)] return [Send("tools", [call]) for call in pending] # 每个待处理的工具调用对应一个任务 graph.add_conditional_edges("model", model_to_tools, ["tools", END])

# ---- “工具”之后的路由器：循环回传，除非运行了 return_direct 工具 ---- def tools_to_model(state): if all_executed_tools_are_return_direct(state): return END return "model" # 让模型读取结果 graph.add_conditional_edges("tools", tools_to_model, ["model"])

return graph.compile(checkpointer=checkpointer, ...) # 与第 2 节相同的 compile()

有三点值得注意，因为它们与我们之前看到的内容相联系：

• 节点是状态的普通函数。model_node 读取 messages 通道并返回一个部分更新，reducer 会将其追加。这正是第 2 节中节点 a 所做的。ToolNode 只是另一个 Runnable。
• 循环由两个条件边组成。model_to_tools 和 tools_to_model 是与第 2 节中的 route 相同类型的路由器。模型 → 工具 → 模型的循环构成了整个智能体循环。
• 并行工具调用通过 Send 实现。当模型同时请求两个工具时，路由器为每个调用返回一个 Send("tools", [call])，因此每个调用都在下一个超级步骤中作为独立任务运行（Send 机制在第 3 节末尾有解释）。在我们的单工具运行中，只有一个待处理调用，因此只有一个任务。

当你传递中间件或响应格式时，实际函数会添加更多的节点和边（这就是 Deep Agents 部分中额外的 ...middleware.before_agent 节点的作用），但其骨架如下所示。

4.2 使用它

以下是演示中调用 create_agent 的代码。为了保持离线和确定性，它使用了一个脚本化模型，这是一个假的聊天模型，首先要求调用工具 double(21)，然后，一旦看到工具的结果，便进行回答。（真实的智能体会在此处传入一个真正的聊天模型。）

检查 agent 的方式与之前相同，输出如下：

nodes   : ['model', 'tools']
channels: {'messages': 'BinaryOperatorAggregate', 'jump_to': 'EphemeralValue',
           'structured_response': 'LastValue', '__start__': ..., '__pregel_tasks': ...}
node 'model': bound=RunnableCallable
node 'tools': bound=ToolNode
edges   : [('__start__','model'), ('model','__end__',cond), ('model','tools',cond), ('tools','model',cond)]
messages: HumanMessage 'double 21' → AIMessage[tool_call double] → ToolMessage '42' → AIMessage 'answer 42'

4.3 它生成的图

create_agent 构建的图很小。让编译后的对象自行绘制（agent.get_graph().draw_mermaid()）会得到以下内容，其中虚线箭头表示条件边：

graph TD;
    __start__([__start__]) --> model(model)
    model -.->|tool calls present| tools(tools)
    model -.->|no tool calls| __end__([__end__])
    tools -.-> model

这是我运行 get_graph().draw_mermaid() 得到的结果，并添加了我用来解释路由的两个边标签。使用我们的运行流程遍历该图：__start__ 将 HumanMessage 放入 messages 通道并唤醒 model。脚本化模型回复了一个工具调用，因此路由器将运行路径发送给 tools。tools 执行 double(21) 并追加一条 ToolMessage('42')，这再次唤醒了 model。这次回复没有工具调用，因此路由器转向 __end__。这就是上面 messages: 行所描述的内容：经过两次 model，一次 tools。

这就是整个 ReAct 循环：

• 状态是一个带有追加归约器的 messages 通道（因此对话会累积，就像上面访问的那样）。
• model 是一个调用聊天模型的节点。tools 是一个 ToolNode：一个 Runnable，它读取最后一条 AIMessage 的 tool_calls，运行每个工具并写入 ToolMessages。
• 循环是 model 的一个条件边：存在工具调用 → tools，否则 → __end__。而 tools → model 是一条普通的反向边。图中的循环即为代理循环；Pregel 循环只是在有触发器时持续滴答作响。

引擎中不存在任何特定于代理的东西。“Agent”只是一种图结构。

1. 图的冒泡

将这一点与上一节中的 Send 快速做个对比：两者都是节点或路由器交给循环处理的事物，但方向相反。Send 是向下的数据流：创建任务的请求，写入通道并在下一个屏障处被拾取。Bubble-up（冒泡）是向上的控制信号：一个停止工作的异常，从节点内部传送到拥有运行代码的地方。它们是独立的机制。它们在一点交汇：由 Send 创建的任务是普通任务，因此如果它调用 interrupt() 或引发 GraphBubbleUp，该信号会像其他任何信号一样通过循环冒泡上来（例如，其工具要求人类批准智能体）。

节点执行的一些操作并非返回值。interrupt("approve?") 必须停止整个运行，而不仅仅是当前节点。LangGraph 将其实现为异常，类似于在 C# 中抛出自定义异常并由堆栈更上层的 catch 块故意处理的情况：

GraphBubbleUp(Exception)
  ├─ GraphDrained        (协作式排空)
  ├─ GraphInterrupt      (interrupt(), 人在回路中)
  │    └─ NodeInterrupt
  └─ ParentCommand       (子图告知父图如何操作)

“冒泡”字面意思正是如此：异常在节点内部引发，并通过运行器向上遍历。根据 _retry.py 和 _runner.py：

• 重试包装器 (run_with_retry) 故意不重试 GraphBubbleUp；普通异常会经过你的 RetryPolicy；
• GraphInterrupt 会将其待处理的写入保存到检查点，从而使暂停得以保留；
• ParentCommand 由外层图处理，这就是子图中的 Command(graph=Command.PARENT, ...) 能够到达其父图的方式；根图会抑制 GraphInterrupt 并将中断作为结果返回，而不是抛出异常。

Command 只是一个包含四个可选字段的小数据类：update、goto、resume、graph。你在两个地方使用它，每个地方关注的字段对有所不同：

• 从节点返回，用于在一步中完成“更新状态并路由”：Command(update={"total": 1}, goto="b")。update 的应用方式如同普通的节点返回值，而 goto（节点名称、名称列表或 Send 数据包）选择下一个运行的内容，无需依赖边。graph=Command.PARENT 将其指向外层图而非当前图（即上述的 ParentCommand）。
• 作为 invoke 的输入传递，以继续暂停的运行：Command(resume="yes")。通常 invoke 接受新状态；这里它被告知“不要重新开始，保存的运行正在等待答案，这就是答案”。resume 是待处理的 interrupt() 调用返回的值。

下面的演示使用了第二种形式。第一种形式（update/goto）在第 5c 节中展示。

该演示展示了带有内存保存器的完整周期。首先是图，其中一个节点调用 interrupt()。需要检查点，因为暂停必须保存在某处：

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt

def ask(s):
    answer = interrupt("approve?")       # 在此处暂停整个运行；下次返回恢复值
    return {"visited": [f"answer={answer}"]}

h = StateGraph(S)                        # 与之前相同的状态类 S
h.add_node("ask", ask)
h.add_edge(START, "ask")
h.add_edge("ask", END)
happ = h.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "t"}}   # 标识保存的对话/运行

print("first call :", happ.invoke({"total": 0, "visited": []}, cfg))   # 运行直到 interrupt()，然后返回
print("next       :", happ.get_state(cfg).next)                        # 哪个节点在等待
print("resumed    :", happ.invoke(Command(resume="yes"), cfg))         # 继续，回答“yes”

输出如下（中断 ID 已缩短）：

first call : {'total': 0, 'visited': [], '__interrupt__': [Interrupt(value='approve?', id='014f…')]}
next       : ('ask',)
resumed    : {'total': 0, 'visited': ['answer=yes']}

阅读代码：第一次调用不会抛出异常。它正常返回，待处理的问题位于 __interrupt__ 下。get_state(cfg).next 表明节点 ask 是正在等待的那个。第二次调用传入的是 Command(resume="yes") 而非新输入；它会找到同一 thread_id 的保存检查点并继续执行。

注意 “resume” 的真正含义：节点从头再次运行，这次 interrupt() 返回 resume 值而不是抛出异常。这种重新执行意味着在 interrupt() 之前的副作用必须是幂等的。

1. 以及深度智能体（Deep Agents）

它的来源。create_deep_agent 位于 deepagents 包中，这是 LangChain 的一个独立开源（MIT）项目，发布在 langchain-ai/deepagents。其包元数据将其描述为“内置文件系统、上下文管理、子智能体委派、技能和长期记忆的代理框架”，并标记为 Beta（我使用的是 0.7.20）。它不属于 langgraph 或 langchain 的一部分；它依赖于 langchain。其源代码从 langchain.agents 导入 create_agent，且其文档字符串指出多个参数是“透传给 create_agent 的”。

它在技术栈中的位置。考虑三个层次，每一层都建立在下一层之上：

deepagents   create_deep_agent   现成的“框架”：文件、子智能体、摘要、技能、记忆
   | 调用
langchain    create_agent        模型 <-> 工具循环，以及中间件系统
   | 构建
langgraph    StateGraph / Pregel 运行时：通道、超级步骤、检查点

用 C# 术语来说：LangGraph 是运行时（类似于 ASP.NET 的主机和管道），create_agent 是一个最小化的应用模板，而 Deep Agents 是一个有偏见的入门套件，在该模板上预安装了一组中间件。这就是为什么下面的图与第 4 节中的图看起来相同的原因。

create_deep_agent(model=..., tools=[...]) 也返回一个 CompiledStateGraph，具有相同的 model 和 tools 节点，以及一个中间件节点。演示程序打印的内容如下：

nodes   : ['model', 'tools', 'PatchToolCallsMiddleware.before_agent']
channels: messages: DeltaChannel, files: DeltaChannel,
          _summarization_event, _summarization_session_id, ...
tools the model sees: ['delete','double','edit_file','execute','glob','grep','ls','read_file','task','write_file']
edges   : ... ('model','model',cond), ('model','tools',cond), ('tools','model',cond) ...

因此，Deep Agents 是带有额外状态和额外工具的代理图，这些是由中间件添加的：

• 额外通道：files（保留在图状态中的虚拟文件系统）以及用于长对话的摘要簿记。这里的 messages 是一个 DeltaChannel 而不是普通的归约通道，因此我不假设代理的通道类型等于普通代理的类型。
• 额外工具：来自文件系统中间件的 ls、read_file、write_file、edit_file、glob、grep、execute，以及 task。execute 列在模型可见的工具中，但默认的 StateBackend 没有 execute（我检查过：hasattr(StateBackend, "execute") 为 False）；只有当后端实现了 SandboxBackendProtocol 时它才会运行，否则返回错误。因此默认情况下没有 shell 访问权限，且 files 是临时的图状态。
• 作为节点的中间件钩子：PatchToolCallsMiddleware.before_agent 在第一次模型调用之前运行（它修复悬空的工具调用）。只有覆盖 before_agent、before_model、after_model 或 after_agent 的中间件才会添加节点（我确认 M.before_model 和 M.after_model 出现在 nodes 中）；包装模型或工具调用的中间件则不会。

什么是中间件？

中间件是在不编辑循环的情况下扩展 create_agent 的方式。这与 ASP.NET 中间件或 DelegatingHandler 的理念相同：你编写一个类，将它们列表传递给 create_agent(..., middleware=[...])，然后循环会在定义的点上调用你的钩子。AgentMiddleware 有两类钩子（来自 langchain/agents/middleware/types.py）：

• 节点钩子（Node hooks），在步骤之间运行：before_agent、before_model、after_model、after_agent。它们接收状态并返回状态更新（或 None），还可以设置 jump_to 以重定向循环。你覆盖的每一个钩子都会成为图中的实际节点。
• 包装钩子（Wrap hooks），围绕调用运行：wrap_model_call(request, handler) 和 wrap_tool_call(request, handler)。你调用 handler(request) 以继续执行，因此你可以修改请求、重试、短路或后处理响应。这些钩子存在于现有的模型和工具节点内部，因此不会增加新的节点。

中间件还可以提供工具和额外的状态键。我运行了以下代码进行检查：

class Audit(AgentMiddleware):
    def before_model(self, state, runtime):             # a node hook
        log.append(f"before_model: {len(state['messages'])} msgs"); return None
    def wrap_model_call(self, request, handler):        # a wrap hook
        log.append("wrap_model_call: enter")
        resp = handler(request)                         # the next layer, ending in the model
        log.append("wrap_model_call: exit"); return resp

agent = create_agent(fake_model, middleware=[Audit()])

nodes: ['Audit.before_model', '__end__', '__start__', 'model']
log  : ['before_model: 1 msgs', 'wrap_model_call: enter', 'wrap_model_call: exit']

Audit.before_model 是一个新节点；wrap_model_call 没有向图中添加任何内容，但围绕模型调用进行了运行。这正是上面 Deep Agents 输出中的 PatchToolCallsMiddleware.before_agent 节点。Deep Agents 主要是一个精心策划的中间件列表（文件系统、摘要、工具调用修补、可选技能和记忆）。

状态后端是什么？

Deep Agents 的文件工具（ls、read_file、write_file、edit_file、glob、grep）不直接操作磁盘。它们调用一个后端，即实现 BackendProtocol 的对象（ls、read、write、edit、grep、glob、delete、upload/download）。后端决定文件实际存储的位置，因此同一个代理可以针对不同的存储运行。这是一个仓库/策略接口，类似于 .NET 中的 IFileProvider。包中包含的后端如下：

注意：这与 LangGraph 检查点器不同，后者有时也被称为状态后端（第 5b 节）。检查点器存储图的快照；Deep Agents 后端存储代理的文件。使用默认的 StateBackend 时两者会合流：其文档字符串指出读写操作通过 CONFIG_KEY_READ 和 CONFIG_KEY_SEND 进行，因此文件写入只是对 files 通道的普通写入，在屏障处应用并按常规状态进行检入。这就是为什么文件中显示为上述输出中的通道，以及为什么默认情况下不需要磁盘的原因。如果在图运行之外使用 StateBackend 会引发异常；若要预填充文件，请向 invoke 传递 {"files": {...}}。

旁注：Squad SDK 中也存在相同的模式。Squad 的 SDK（bradygaster/squad, packages/squad-sdk）在其 StorageProvider 接口中具有匹配的理念（read、write、append、list、delete、mkdir、rename、stat、createIfAbsent...），包括 FSStorageProvider（默认）、InMemoryStorageProvider 和 SQLiteStorageProvider：一个狭窄的接口，多种实现，一种默认实现。它由 Dina Berry（diberry；#567, #640）添加。Squad 还有一个独立的高级 StateBackend（工作区文件、孤立的 git 分支，或加上 git notes），但这并非同一回事：一个适配器将其包装为 StorageProvider。源码固定在我阅读的提交版本：storage-provider.ts, state-backend.ts, State Backends 文档。回到深度智能体。

什么是 DeltaChannel？

在深度智能体图中，消息和文件都使用 DeltaChannel，而不是普通的归约器通道。它解决的问题是：普通归约器通道会在每一步检查点其完整值，因此长对话或大型虚拟文件系统会被复制到每个检查点中。DeltaChannel（beta 版，langgraph/channels/delta.py）则仅在普通检查点中存储一个哨兵值，并通过重放之前的写入操作来重建值，定期写入完整快照（默认每 1000 次更新一次）。这是事件溯源：写入日志是真相，快照是一种优化手段。

代价体现在其自身的文档字符串中：归约器必须是确定性的且对批处理不变（reducer(reducer(s, xs), ys) == reducer(s, xs + ys)），并且检查点保存器必须保留祖先写入记录（get_delta_channel_history）。第 5b 节涵盖了这对检查点的含义。

深度智能体能力清单

根据已安装包的布局（deepagents/middleware 和 deepagents/backends），以及我上面观察到的内容：

• 中间件模块：filesystem（文件系统）、subagents（子智能体）、async subagents（异步子智能体）、summarization（摘要）、memory（记忆）、skills（技能）、permissions（权限）、rubric（评分标准）、patch-tool-calls（修补工具调用）、prompt caching（提示缓存）、message eviction（消息驱逐）、overflow clipping（溢出裁剪）、unsupported-content（不支持的内容）、tool exclusion（工具排除）。
• 后端（文件和执行实际去向的地方）：state（图状态，演示中看到的默认值）、store（存储）、composite（组合）、filesystem（文件系统）、local shell（本地 Shell）、sandbox（沙箱）、LangSmith、context hub（上下文中心）。
• 只有定义 before_*/after_* 钩子的中间件才会成为图节点（我们看到了 PatchToolCallsMiddleware.before_agent）；其他中间件则在原地包装模型或工具调用，因此它们不会出现在节点中。我是从这一个观察到的节点和模块布局推断出这一规则的，而非阅读了中间件加载器。
• 在这个默认构建中，模型的工具有七个文件系统/Shell 工具、task（任务）和我自己的 double（副本）。我没有看到 todo-list（待办事项列表）工具，所以我不声称它有。

task 是一个调用图的图

子智能体机制是最有趣的部分，它是普通的 LangGraph。在 deepagents/middleware/subagents.py 中，task 是一个 StructuredTool。其主体准备一个新的状态（对于普通子智能体仅为 {"messages": [HumanMessage(description)]} 加上非私有状态键），然后调用：

result = subagent.invoke(subagent_state, subagent_config)
return _return_command_with_state_update(result, runtime.tool_call_id)

因为编译后的图是一个 Runnable（第 1 节），子智能体仅仅是另一个从工具调用内部调用的编译智能体。它的结果以 Command(update={..., "messages": [ToolMessage(...)]}) 的形式返回，父级的 ToolNode 将其应用于父级的通道。子智能体的上下文窗口是隔离的；只有其最终消息和选定的状态键会返回。

5b. 归约器、通道类型和检查点，真正的内容

检查点。使用检查点保存器时，每个超级步骤边界都会写入一行。演示将两节点图保存在 InMemorySaver 中并列出它：

step=-1 source=input values={'__start__': {'total': 10, 'visited': []}} step= 0 source=loop values={'total': 10, 'visited': []} step= 1 source=loop values={'total': 20, 'visited': ['a']} step= 2 source=loop values={'total': 21, 'visited': ['a', 'b']} versions_seen keys: ['__input__', '__start__', 'a', 'b'] channel_versions: {'__start__': '2', 'total': '4', 'visited': '4', 'branch:to:a': '3', 'branch:to:b': '4'}

检查点（checkpoint）是一个小型字典，包含 channel_values、channel_versions、versions_seen、updated_channels、一个 ID、时间戳和 v。其中两个字段负责调度：

• channel_versions 是每个通道的单调递增版本号（实际的字符串是零填充计数器加上随机后缀；这里已简化）。
• versions_seen[node] 记录每个节点上次消费的触发器版本。prepare_next_tasks 会在某个触发通道的版本号比该节点所见的版本更新时运行该节点。这就是整个“下一步运行什么”的规则，也是为什么从检查点恢复不需要额外记账的原因。

Step -1 存储原始输入；steps 0..N 是循环超级步骤。由已运行但尚未应用的写入操作单独存储为针对其父检查点的待处理写入（put_writes）。这样，并行超级步骤中间的崩溃不会丢失已经完成的分支，并且中断状态得以保留。

保存器（state backend）契约。BaseCheckpointSaver 是存储接口。核心方法包括 put（写入检查点）、put_writes（任务的待处理写入）、get_tuple 和 list（读取），以及异步对应方法（aput、aget_tuple 等），还有线程删除/修剪辅助方法和序列化器（此处为 JsonPlusSerializer）。行通过 thread_id、checkpoint_ns（子图命名空间）和 checkpoint_id 寻址；新的 thread id 意味着一个新的空历史。我只运行了 InMemorySaver；我尚未测试 SQLite 或 Postgres 保存器，因此不对它们做任何声明。

注意：DeltaChannel。Deep Agents 的消息和文件使用 DeltaChannel（langgraph/channels/delta.py 中的 beta 版本）。根据我对该源码的阅读，它并非每次运行都进行检查点记录，而是在普通步骤中检查点记录一个哨兵值，并从祖先写入中重建值，同时定期生成完整快照，因此它需要确定性的归约器和保留祖先历史的保存器。“每个检查点都是完整快照”仅适用于普通通道。

5c. 当事情变得结构化或出错时：冲突、路由、子图、重试

前面的章节展示了正常路径。本节涵盖当事情出错或变得更加结构化时会发生的情况：两个节点争夺同一个键，一个节点需要同时更新状态并选择去向，图中包含另一个图，一个节点抛出异常，以及你想控制状态保存的频率或实时观察它。我在离线环境下运行了以下所有代码片段（samples 仓库中的 structured_cases.py）；输出按所见引用。

两个并行节点写入相同的键

回想第 3 节中的超级步骤规则：同一超级步骤中的所有节点都读取旧状态，它们的写入在屏障处一起应用。那么如果两个节点写入相同的键会怎样？对于像 total: int 这样的普通字段，通道是 LastValue，只能容纳一个值。“哪一个写入获胜？”没有合理的答案，因此 LangGraph 拒绝猜测：

class S(TypedDict):
    total: int

g = StateGraph(S)
g.add_node("a", lambda s: {"total": 1})
g.add_node("b", lambda s: {"total": 1})
g.add_edge(START, "a")                           # a 和 b 都从 START 开始，
g.add_edge(START, "b")                           # 所以它们在同一个超级步骤中运行
g.add_edge("a", END)
g.add_edge("b", END)
g.compile().invoke({"total": 0})

InvalidUpdateError: At key 'total': Can receive only one value per step. Use an Annotated key to handle multiple values.

错误信息中已经包含了修复方案。如果你将键声明为 Annotated[list, operator.add]，通道就会变成 BinaryOperatorAggregate，它知道如何合并多次写入，此时同一个图会返回 {'total': ['a', 'b']}。用 C# 的概念来类比：这相当于两个线程赋值给同一个字段（一种你必须禁止的竞争条件）与两个线程通过锁调用 list.AddRange（一种你定义的合并操作）之间的区别。归约器（reducer）就是那个锁。

命令：在一次返回中更新状态并选择下一个节点

通常，节点返回状态更新，由边决定下一步运行什么。有时节点本身就知道该去哪里（例如路由器，或决定移交控制的智能体）。使用 Command 可以让它同时完成这两件事，如第 5 节所定义：

def router(s):
    return Command(update={"out": ["r"]}, goto="t")   # 写入状态 AND 选择下一个节点

g = StateGraph(C)                       # C 包含: total: int, out: Annotated[list, operator.add]
g.add_node("r", router)
g.add_node("t", lambda s: {"out": ["t"]})
g.add_edge(START, "r")
g.add_edge("t", END)                                   # 注意：没有从 r 到 t 的边
g.compile().invoke({"total": 0, "out": []})

{'total': 0, 'out': ['r', 't']}

图中不存在 r → t 的边。goto 在运行时创建了这种路由，而 out 显示两个节点按顺序都执行了。你已经见过的 Command 的其他两种用法是：resume= 用于在中断后继续执行（第 5 节），以及 graph=Command.PARENT 用于当你在子图内部时，将更新发送给外层图（下一小节）。Send（第 3 节）是第三种路由工具：使用 goto 选择一个下一个节点，使用 Send 启动具有不同输入的多个任务。

图中的图（子图）

因为编译后的图是一个 Runnable（第 1 节），你可以像传递任何函数一样将它传递给 add_node。父图将其视为一个节点；在内部，它会运行自己的超级步骤。它的检查点存在于自己的命名空间中，因此不会与父图的检查点冲突，而 stream(..., subgraphs=True) 允许你查看内部情况：

child = StateGraph(C)                        # C 是与上面相同的状态模式
child.add_node("t", lambda s: {"out": ["t"]})
child.add_edge(START, "t")
child.add_edge("t", END)

parent = StateGraph(C)
parent.add_node("child", child.compile())    # 将编译后的图用作节点
parent.add_edge(START, "child")
parent.add_edge("child", END)
app = parent.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "1"}}
for event in app.stream({"total": 0, "out": []}, cfg, subgraphs=True):
    print(event)

(('child:512f5449-103f-f070-0e96-928c949f2c95',), {'t': {'out': ['t']}})
((), {'child': {'total': 0, 'out': ['t']}})

每个事件都是 (path, update)。第一个事件的路径非空：表示节点 t 在子图内完成。第二个事件的路径为空 ()：表示父图看到整个子图节点完成。路径即为命名空间。这也是为什么第 6 节中的 task 工具被称为“一个图调用另一个图”。

重试：当节点抛出异常时会发生什么

一个引发异常的节点通常会导致整个运行失败。附加一个 RetryPolicy，循环将仅重新运行该节点。我遇到的一个陷阱是：默认的 retry_on 故意排除了 ValueError（编程错误类型的异常默认不重试），因此我的不稳定节点需要显式指定 retry_on=ValueError：

def flaky(s):
    n["i"] += 1
    if n["i"] < 3: raise ValueError("boom")        # 失败两次，然后成功
    return {"out": ["ok"]}

g.add_node("f", flaky, retry_policy=RetryPolicy(retry_on=ValueError, initial_interval=0.01))

{'total': 0, 'out': ['ok']}      # n["i"] == 3：两次失败，第三次尝试成功

重试机制作用于节点层面，而非图层面，因此同一超级步骤中其他节点已完成的工作不会被重复执行。此外，第 5 节（GraphBubbleUp：中断、父级命令）中的控制流异常永远不会被重试，因为它们不属于故障。在 C# 中，这相当于将 Polly 的 Retry 策略包裹在一次调用中，并通过 retry_on 异常过滤器进行控制。

减少保存频率与实时监控

有两个参数可以调节这套机制的成本和可见性：

• 持久化策略（durability on invoke/stream）决定检查点写入的时机：“sync”（在下一步开始前写入；最安全但最慢）、“async”（在后台运行，与下一步并行）或“exit”（仅在运行时结束时写入；最快，但崩溃会导致正在运行的状态丢失）。这些描述基于参数的文档定义，而非我的实测结果。该参数的签名默认值为 None；由于未验证其实际解析结果，我不在此声明有效默认值。
• 流模式（Stream modes）选择流输出的内容：values（每一步后的完整状态）、updates（仅每个节点的返回值，即上文所示内容）、checkpoints（检查点）、tasks（任务）、debug（调试信息）、messages（LLM 令牌）以及 custom（节点通过流写入器写入的任何自定义内容）。相同的运行过程，不同的视图呈现。

深度代理附加功能（仅列出，未实际测试）

仅从模块清单来看，以下功能我并未实际运行：skills（技能）、memory（记忆，包括提示注入或 BaseStore）、permissions（权限）以及 interrupt_on（人工介入的中断）。子代理可以是任何兼容的可运行对象，task 默认隔离上下文；实验性的 fork 模式会继承父级状态。

读取状态回溯与时间旅行

get_state(cfg) 返回一个 StateSnapshot（包含 values、next、config、metadata、created_at、parent_config、tasks、interrupts），而 get_state_history(cfg) 则遍历父级链。时间旅行只需选择一个较早的检查点快照并从中发起调用即可：

state before b = {'total': 20, 'visited': ['a']}  ->  invoke(None, that config) = {'total': 21, 'visited': ['a', 'b']}

Store 是另一回事。Checkpointer 是针对单线程运行状态的。BaseStore（示例：InMemoryStore）是基于命名空间元组和键寻址的跨线程长期存储：store.put(("users","tamir"), "pref", {...})。深度代理的记忆功能和基于存储的文件后端均建立在这一概念之上；此处我仅验证了基础的 InMemoryStore。

最后的思考

阅读完所有代码后，令我印象深刻的是其中几乎没有魔法可言。图本质上是一组通道以及当这些通道发生变化时被唤醒的节点。边只是对通道的写入操作，超级步骤提供了清晰的读、写、合并节奏，而偶尔出现的特殊异常或 Command 则是节点从内部引导循环的方式。代理就是在这台机器上增加一个模型节点和一个工具节点形成的循环，而深度代理则是在此基础上再次叠加中间件，增加通道、工具和钩子节点。一旦以这种方式理解，文档、堆栈跟踪和检查点都变得合情合理，我也不再将 LangGraph 视为黑盒。

如果你想亲自尝试，本文中的所有示例均可在 samples 仓库中离线运行，无需 API 密钥。请锁定文章开头提到的版本，因为内部实现在不同版本间可能会发生变化。