LangGraph 源码剖析：Runnable、Channel、Bubble-Up、Agent 与 Deep Agent 如何构建

我花了一个周末钻研 LangGraph 的源代码，试图理解这些抽象层之下究竟发生了什么。以下是我的笔记、我做过的实验，以及我最终形成的思维模型。

最让我惊讶的是，核心思想竟然如此小。一张图、一个 ReAct agent 和一个 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#，可以把它想成一个不可变 record 加上 with { ... } 表达式，由引擎替你应用。

Channel 是该状态中的一个具名槽位，它拥有如何合并新值的规则。一个状态键 total 会变成一个名为 total 的 channel。当多个节点在同一轮写入时，这条规则就很重要：新值应该替换旧值，还是与旧值合并？用 C# 的话说，channel 大致就是一个字段配上它自己的合并函数，类似 Func<T, T, T> merge。

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

Reducer 就是那个合并函数：reducer(old, new) -> combined。你用 Annotated 附加一个。如果你用过 LINQ 的 Aggregate，形状是一样的：把每个新值折叠进一个运行中的值。operator.add 作为列表上的 reducer 意味着“把新项追加到旧项上”。

有了这些，下面是最小的有趣图。

1. 唯一的契约：Runnable

一个 Runnable 就是任何具有 invoke、ainvoke、stream、astream、batch 和 with_config 的东西。就这样。在 C# 中，你会称其为一个所有东西都实现的接口，一个 IRunnable<TIn, TOut>。| 运算符组合其中两个（Python 允许类重载 |，就像 C# 的 operator |）：

1
2
3

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

为什么这很重要：编译后的 LangGraph 也是一个 Runnable。它的 MRO 以 CompiledStateGraph → Pregel → PregelProtocol 开头。（MRO 是方法解析顺序，Python 的继承链：CompiledStateGraph : Pregel : PregelProtocol。）所以整个图可以坐在一个链中，或另一个图的节点中。这就是后来子代理的工作方式。

2. compile() 产生什么

拿最小的有趣图来说：两个节点，状态有一个覆盖字段和一个追加字段。节点只是一个函数，它接收状态并返回部分更新：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

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

class S(TypedDict):
    total: int                              # bare key: one write per superstep
    visited: Annotated[list, operator.add]  # reducer: appends

def a(s):                               # node "a": doubles n, logs "a"
    return {"total": s["total"] * 2, "visited": ["a"]}

def b(s):                               # node "b": adds 1 to n, logs "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：

1
2
3
4
5
6
7
8

# `app` is the compiled graph from above (a Pregel object)
print("channels:", {name: type(ch).__name__ for name, ch in app.channels.items()
                    if not name.startswith("branch:")})    # hide the per-edge channels for brevity
for name, node in app.nodes.items():
    if name == "__start__":
        continue                                            # internal input node, skipped for brevity
    print(f"node {name!r}: triggers={node.triggers} bound={type(node.bound).__name__}"
          f" writers={len(node.writers)}")

其打印：

1
2
3
4

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 对象，所以这里有一个。这是来自 langgraph/pregel/_read.py 的类，裁剪到此处重要的字段（我省略了错误处理程序和子图字段，文档字符串是我的，从源代码的浓缩而来）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

class PregelNode:
    # A container, not a Runnable. The engine uses it to build a runnable task each time the node fires.

    channels: str | list[str]      # which channels to READ as input. For node "a": ['total', 'visited'],
                                   # so your function receives a dict with those two keys (the `s` in `def a(s)`)
    triggers: list[str]            # which channels WAKE the node: if any is written, the node runs next step.
                                   # For node "a": ['branch:to:a']
    bound: Runnable                # the actual logic: a Runnable wrapping your function, called with the input
    writers: list[Runnable]        # run AFTER `bound`; each takes its output and writes it to channels
                                   # (one stores your return value in state, one wakes the next node)
    mapper: Callable | None        # optional transform of the input before `bound` (None for node "a")

    retry_policy: Sequence[RetryPolicy] | None   # what happens if the node raises (None here)
    cache_policy: CachePolicy | None             # reuse results for identical input (None here)
    timeout: TimeoutPolicy | None                # per-invocation limit (async nodes only)
    tags: Sequence[str] | None                   # tracing labels
    metadata: Mapping[str, Any] | None           # tracing metadata

注意 channels（它读取什么）和 triggers（什么唤醒它）之间的分离。它们故意分开：一个节点可以读取许多通道，但只被一个唤醒。用 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 节）；我们的图不使用它，但每个图都有它。
• node '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 是最后一个，所以它只有第一个写入者；它的边通向 END，不唤醒任何东西。

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

这是同样的转换的图片。第一个图是你用 add_edge 画的；第二个是 compile() 实际构建的（我在图片中添加了 __start__ 输入通道和每个节点的两个写入者，如打印输出所示）：

你写的内容（构建器）：

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

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

flowchart TD
   I["__start__ channel<br/>(your input)"] -->|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 应用你的 reducer（operator.add），这就是为什么 visited 是追加而不是替换。额外的通道承载控制流：__start__ 保存输入，而边 a → b 会变成一个名为 branch:to:b 的通道。双下划线只是内部通道的命名约定（Python 的 __name__ 风格，不是关键字）。
• 节点是 PregelNode。每个节点都有 triggers（哪些通道唤醒它）、channels（它读取哪些通道）、bound（实际的 Runnable，这里是你函数的包装器）以及 writers（它运行后写入什么）。
• 边消失了。运行时不存在边对象。a → b 是“a 的写入者写入 branch:to:b，而 b 在 branch:to:b 上触发”。get_graph() 为绘图重建了漂亮的边列表，但引擎从不遍历它。

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

def c(s):                                # node "c": the alternative path, resets total
    return {"total": 0, "visited": ["c"]}

def route(s):                            # the router: returns the NAME of the next node
    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"])   # replaces 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)

我编译并运行了这个，并以同样的方式检查了它：

1
2
3
4
5

channels: total=LastValue, visited=BinaryOperatorAggregate, __start__, __pregel_tasks=Topic,
          branch:to:a, branch:to:b, branch:to:c   (the last three are 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 的第二个写入者。它不是对 branch:to:b 的固定写入，而是 _route，它调用你的函数并写入答案所指定的任意通道。另一个通道从未被写入，因此该节点根本不会运行。当 total 从 10 开始时，a 使其变为 20，路由器选择 b，而 invoke 返回 {'total': 21, 'visited': ['a', 'b']}；c 永远不会触发。从 1 开始，路由器则选择 c。（_route 是我在这个版本中看到的内部写入者的名称；把它当作实现细节。）

3. 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" 显示每个节点一个超步：

1

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

“先写入，在屏障处应用”这一分离是最重要的一个思想。这就是为什么并行分支是安全的（它们彼此看不到），为什么存在 reducer（两个分支在同一轮写入同一个键需要一个 Annotated reducer，否则图会失败），以及为什么轮次之间的检查点是普通通道的一致快照（Deep Agents 的 DeltaChannel 是例外，见 5b）。

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

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

class S(TypedDict):
    items: list
    out: Annotated[list, operator.add]       # reducer, so parallel writes are appended

def fan(s):                                  # router: one Send per item
    return [Send("work", {"x": i}) for i in s["items"]]

def work(a):                                 # 'a' is the Send's argument, NOT the graph state
    return {"out": [a["x"] * 10]}

g = StateGraph(S)
g.add_node("work", work)
g.add_conditional_edges(START, fan, ["work"])
g.add_edge("work", END)
app = g.compile()

print(app.invoke({"items": [1, 2, 3], "out": []}))

用 stream_mode="debug" 进行流式处理并打印每个 task 事件，可以显示循环调度了什么（我运行过这个）：

1
2
3
4

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 归约器：三个任务在同一轮中写入了 out。

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

一句话总结：Pregel 循环以超步方式运行你的图，其中一个超步就是一轮，在这一轮中每个就绪节点都在同一份状态快照上运行，它们的所有写入都在屏障处合并，然后下一轮才开始。

4. 示例：一个 AI agent 就是一张图

在更困难的部分（冒泡、归约器、持久性）之前，先看看目前为止一切的价值所在：一个 AI agent 被表达为一张图。这听起来像是另一个物种，但 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：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

def create_agent(model, tools, ...):
    graph = StateGraph(state_schema=AgentState, ...)   # 'messages' etc. become channels

    # ---- node 1: "model" -- a plain function node ----
    def model_node(state, runtime):
        messages = state["messages"]                    # read the channel
        if system_message:
            messages = [system_message, *messages]
        ai_message = model.bind_tools(tools).invoke(messages)   # call the chat model
        return {"messages": [ai_message]}               # write: append to 'messages'
    graph.add_node("model", RunnableCallable(model_node))

    # ---- node 2: "tools" -- a ToolNode, itself a Runnable ----
    tool_node = ToolNode(tools)    # reads the last AIMessage's tool_calls, runs each tool,
    graph.add_node("tools", tool_node)                  # and writes one ToolMessage per call

    graph.add_edge(START, "model")                      # entry point

    # ---- router after "model": tool calls -> "tools", otherwise finish ----
    def model_to_tools(state):
        last_ai = last_ai_message(state["messages"])
        if not last_ai.tool_calls:                      # classic exit condition of an agent loop
            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]   # one task per pending tool call
    graph.add_conditional_edges("model", model_to_tools, ["tools", END])

    # ---- router after "tools": loop back, unless a return_direct tool ran ----
    def tools_to_model(state):
        if all_executed_tools_are_return_direct(state):
            return END
        return "model"                                  # let the model read the results
    graph.add_conditional_edges("tools", tools_to_model, ["model"])

    return graph.compile(checkpointer=checkpointer, ...)   # same compile() as section 2

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

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

当你传入中间件或响应格式时，真实函数会添加更多节点和边（这就是 Deep Agents 部分中那些额外的 ...middleware.before_agent 节点），但骨架就是如此。

4.2 使用它

下面是演示中调用 create_agent 的代码。为了保持离线和确定性，它使用一个脚本化模型，一个假的聊天模型，它首先请求调用工具 double(21)，然后在看到工具结果后作答。（真实的 agent 会在其位置传入真实的聊天模型。）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from langchain.agents import create_agent
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool

@tool                                    # turns a plain function into a tool the model can call
def double(n: int) -> str:
    """Double n."""                      # the docstring is the description the model sees
    return str(n * 2)

class Scripted(BaseChatModel):           # a fake model with a fixed script
    @property
    def _llm_type(self): return "scripted"
    def bind_tools(self, tools, **kw): return self.bind(tools=tools, **kw)
    def _generate(self, messages, stop=None, run_manager=None, **kw):
        done = [m for m in messages if isinstance(m, ToolMessage)]
        if not done:                     # turn 1: no tool result yet, so ask for the tool
            msg = AIMessage(content="", id="ai-1", tool_calls=[
                {"name": "double", "args": {"n": 21}, "id": "call-1", "type": "tool_call"}])
        else:                            # turn 2: tool result is in the history, so answer
            msg = AIMessage(content=f"answer {done[-1].content}", id="ai-2")
        return ChatResult(generations=[ChatGeneration(message=msg)])

agent = create_agent(Scripted(), tools=[double])      # one call builds the whole graph
out = agent.invoke({"messages": [HumanMessage("double 21")]})

像之前一样检查 agent 会打印出：

1
2
3
4
5
6
7

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 通道，带有 append 归约器（因此对话会累积，就像上面的 visited 一样）。
• model 是一个调用聊天模型的节点。tools 是一个 ToolNode：一个 Runnable，它读取最后一个 AIMessage 的 tool_calls，运行每个工具并写入 ToolMessage。
• 循环是从 model 出发的一条条件边：有工具调用 → tools，否则 → __end__。而 tools → model 是一条普通的回边。图中的环就是智能体循环；Pregel 循环只是在有东西被触发时持续运转。

引擎中没有任何智能体特有的东西。“智能体”是一种图形状。

5. 图冒泡

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

节点所做的某些事情不是返回值。interrupt("approve?") 必须停止整个运行，而不仅仅是节点。LangGraph 将其实现为异常，很像在 C# 中抛出一个自定义异常，由栈上更上层的 catch 故意处理：

1
2
3
4
5

GraphBubbleUp(Exception)
  ├─ GraphDrained        (cooperative drain)
  ├─ GraphInterrupt      (interrupt(), human-in-the-loop)
  │    └─ NodeInterrupt
  └─ ParentCommand       (a subgraph telling its parent what to do)

“冒泡”正是这个意思：异常在节点内部引发，并向上穿过运行器传播。根据 _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() 的节点。需要一个检查点保存器，因为暂停必须保存在某处：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

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

def ask(s):
    answer = interrupt("approve?")       # pauses the whole run here; returns the resume value next time
    return {"visited": [f"answer={answer}"]}

h = StateGraph(S)                        # same state class S as before
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"}}   # identifies the saved conversation/run

print("first call :", happ.invoke({"total": 0, "visited": []}, cfg))   # runs until interrupt(), returns
print("next       :", happ.get_state(cfg).next)                        # which node is waiting
print("resumed    :", happ.invoke(Command(resume="yes"), cfg))         # continue, answering "yes"

它会打印（中断 id 被缩短了）：

1
2
3

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

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

注意“恢复”到底是什么：节点会从头再次运行，这一次 interrupt() 返回恢复值而不是抛出异常。正是这种重新执行，使得 interrupt() 之前的副作用必须是幂等的。

6. 以及 Deep Agents

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

它在技术栈中的位置。 想象三层，每一层都建立在下一层之上：

1
2
3
4
5

deepagents   create_deep_agent   a ready-made "harness": files, subagents, summarization, skills, memory
   |  calls
langchain    create_agent        the model <-> tools loop, plus the middleware system
   |  builds
langgraph    StateGraph / Pregel the runtime: channels, supersteps, checkpoints

用 C# 的话说：LangGraph 是运行时（类似 ASP.NET 的宿主和管道），create_agent 是一个最小应用模板，而 Deep Agents 是在该模板上预装一组中间件的、有主见的入门套件。这就是为什么下面的图看起来和第 4 节的一样。

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

1
2
3
4
5

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 而不是普通的 reducer 通道，所以我不会假设该代理的通道类型等同于普通代理的。
• 额外工具： 来自文件系统中间件的 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）：

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

中间件还可以贡献 tools 和额外的状态键。我运行了这个来检查：

1
2
3
4
5
6
7
8
9

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()])

1
2

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、上传/下载）的对象。后端决定文件实际存放的位置，因此同一个 agent 可以针对不同的存储运行。它是一个仓库/策略接口，类似于 .NET 中的 IFileProvider。包中的后端有：

注意：这与 LangGraph 检查点器不是同一回事，后者有时也被称为状态后端（第 5b 节）。检查点器存储图的快照；Deep Agents 后端存储agent 的文件。使用默认的 StateBackend 时，两者相遇：其文档字符串说读写通过 CONFIG_KEY_READ 和 CONFIG_KEY_SEND 进行，因此文件写入只是对 files 的普通通道写入，在屏障处应用并像任何其他状态一样被检查点化。这就是为什么 files 在上面的输出中显示为一个通道，以及为什么默认不需要磁盘。StateBackend 如果在图运行之外使用会抛出异常；要预置文件，请将 {"files": {...}} 传递给 invoke。

旁注：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 文档。回到 Deep Agents。

DeltaChannel 是什么？

Deep Agents 图中的 messages 和 files 都是 DeltaChannel，而不是普通的 reducer 通道。它解决的问题：普通的 reducer 通道每一步都会检查点化其完整值，因此长对话或大型虚拟文件系统会被复制到每个检查点中。DeltaChannel（beta，langgraph/channels/delta.py）改为在普通检查点中只存储一个哨兵，并通过 reducer 重放先前的写入来重建值，每隔一段时间写入完整快照（默认每 1000 次更新）。这是事件溯源：写入日志是真相，快照是优化。

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

Deep Agents 能力清单

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

• 中间件模块：文件系统、子代理、异步子代理、摘要、记忆、技能、权限、评分标准、补丁工具调用、提示缓存、消息驱逐、溢出裁剪、不支持内容、工具排除。
• 后端（files/execute 实际存放的位置）：state（图状态，演示中看到的默认值）、store、composite、filesystem、本地 shell、sandbox、LangSmith、context hub。
• 只有定义了 before_*/after_* 钩子的中间件才会成为图节点（我们看到了 PatchToolCallsMiddleware.before_agent）；其他的则就地包裹模型或工具调用，因此不会出现在 nodes 中。我是从这一个观察到的节点和模块布局推断出这条规则的，而不是通过阅读中间件加载器。
• 在这个默认构建中，模型的工具是七个文件系统/shell 工具、task，以及我自己的 double。我没有看到待办清单工具，所以我不会声称有。

task 是一个图调用另一个图

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

1
2

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 中并列出它：

1
2
3
4
5
6

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'}

检查点是一个小字典：channel_values、channel_versions、versions_seen、updated_channels、一个 id、一个时间戳和 v。有两个字段负责调度：

• channel_versions 是每个通道单调递增的版本号（真实字符串是零填充计数器加随机后缀；我缩短了它们）。
• versions_seen[node] 记录每个节点上次消费的触发版本。当触发通道的版本比该节点已见过的更新时，prepare_next_tasks 就会运行该节点。这就是“接下来运行什么”的全部规则，也是为什么从检查点恢复不需要额外的簿记。

步骤 -1 存储原始输入；步骤 0..N 是循环超步。已运行但尚未应用的任务所做的写入，会作为待处理写入单独存储在其父检查点（put_writes）下。这就是为什么并行超步中途崩溃不会丢失已经完成的分支，以及 interrupt 状态如何得以保留。

保存器（状态后端）契约。BaseCheckpointSaver 是存储接缝。核心方法是 put（一个检查点）、put_writes（某个任务的待处理写入）、get_tuple 和 list（读取），以及异步孪生方法（aput、aget_tuple、……），再加上线程删除/修剪辅助方法和一个 serde（这里是 JsonPlusSerializer）。行通过 thread_id、checkpoint_ns（子图命名空间）和 checkpoint_id 来寻址；新的线程 id 就是新的空历史。我只运行了 InMemorySaver；我没有测试过 SQLite 或 Postgres 保存器，所以我不对它们做任何断言。

注意事项：DeltaChannel。Deep Agents 的 messages 和 files 使用 DeltaChannel（在 langgraph/channels/delta.py 中为 beta）。根据我对该源代码的阅读（而非运行），它在普通步骤上对哨兵值进行检查点保存，并从祖先写入重建值，同时定期进行完整快照，因此它需要确定性的归约器和能保留祖先历史的保存器。“每个检查点都是完整快照”仅对普通通道成立。

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

前面的章节展示的是一帆风顺的路径。这一节讲的是出错或结构更复杂时会发生什么：两个节点争抢同一个键、一个节点既要更新状态又要决定下一步走向、图里套图、节点抛出异常，以及你想控制状态保存的频率或实时观察它。下面每个代码片段我都在离线环境跑过（structured_cases.py 在示例仓库里）；输出按所见原样引用。

两个并行节点写入同一个键

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

1
2
3
4
5
6
7
8
9
10
11

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 and b both start from START,
g.add_edge(START, "b")                           # so they run in the same superstep
g.add_edge("a", END)
g.add_edge("b", END)
g.compile().invoke({"total": 0})

1

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：一次返回同时更新状态并选择下一个节点

通常节点返回一个状态更新，由边决定接下来运行什么。有时节点自己知道该去哪里（一个路由器，一个决定交接的 agent）。返回一个 Command 就能让它同时做这两件事，如第 5 节所定义：

1
2
3
4
5
6
7
8
9

def router(s):
    return Command(update={"out": ["r"]}, goto="t")   # write state AND pick the next node

g = StateGraph(C)                       # C has: 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)                                   # note: no edge from r to t
g.compile().invoke({"total": 0, "out": []})

1

{'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) 让你能看到内部：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

child = StateGraph(C)                        # C is the same state schema as above
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())    # a compiled graph used as a node
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)

1
2

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

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

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

抛出异常的节点通常会让整个运行失败。挂上一个 RetryPolicy，循环就只重跑那个节点。我踩到的一个坑：默认的 retry_on 故意排除了 ValueError（编程错误类异常默认不重试），所以我那个不稳定的节点需要显式加上 retry_on=ValueError：

1
2
3
4
5
6

def flaky(s):
    n["i"] += 1
    if n["i"] < 3: raise ValueError("boom")        # fails twice, then works
    return {"out": ["ok"]}

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

1

{'total': 0, 'out': ['ok']}      # n["i"] == 3: two failures, third attempt succeeded

重试是围绕节点的，不是围绕图的，所以同一超步里其他节点已完成的工作不会重复。而第 5 节里的控制流异常（GraphBubbleUp：中断、父命令）永远不会重试，因为它们不是失败。用 C# 说，这就是 Polly 的 Retry 策略包住一次调用，异常过滤器是 retry_on。

减少保存频率，以及实时观察

有两个旋钮控制着这整套机制的开销和可见性：

• durability 在 invoke/stream 上决定检查点何时写入：'sync'（下一步开始前；最安全，最慢）、'async'（在下一步运行的同时于后台写入），或 'exit'（仅在运行结束时；最快，但崩溃会丢失进行中的运行）。这些描述来自该参数的文档含义，并非我实测所得。签名默认值是 None；我没有验证它实际解析为什么，所以我不给出有效默认值。
• 流模式决定 stream 发出什么：values（每步之后的完整状态）、updates（仅每个节点的返回值，也就是你上面看到的内容）、checkpoints、tasks、debug、messages（LLM token），以及 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) 会沿父链向上遍历。时间旅行就是挑一个更早的快照并从它开始调用：

1

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

Store 是另一回事。检查点是按线程的运行状态。BaseStore（演示：InMemoryStore）是按命名空间元组和键寻址的跨线程长期记忆：store.put(("users","tamir"), "pref", {...})。Deep Agents 的 memory 和由 store 支持的文件后端就建立在这个理念之上；我在这里只验证了普通的 InMemoryStore。

最后一点思考

读完所有这些代码后，留在我脑海中的是其中几乎没有魔法。图就是一组通道，以及当这些通道变化时被唤醒的节点。边只是对通道的写入，超步给你一个干净的读、写、合并节奏，而偶尔出现的特殊异常或 Command 则是节点从内部操控循环的方式。一个代理就是同一台机器，只是在一个循环里放了一个模型节点和一个工具节点，而一个 Deep Agent 还是同一台机器，只是中间件添加了通道、工具和钩子节点。一旦我这样看待它，文档、堆栈跟踪和检查点就都开始说得通了，我也不再把 LangGraph 当作黑盒。

如果你想自己动手试试，本文中的所有内容都可以离线运行，无需 API key，就在 samples 仓库 中。请锁定文章开头列出的版本，因为内部实现会在不同版本之间变化。