1.1 为什么用"图",而不是链或 DAG

先回答最根本的设计问题。LLM 应用的编排,经历了三代抽象:

Agent 的核心循环是 "思考 → 行动 → 观察 → 再思考",这是一个带反馈的环。DAG 表达不了环,所以 LangGraph 选择了"可含环的有向图 + 一套受控的调度算法"。那套算法,就是下一节的 Pregel。

1.2 内核:Pregel + BSP(批量同步并行)

LangGraph 的运行时,借鉴了 Google 的图计算系统 Pregel,采用 BSP(Bulk Synchronous Parallel,批量同步并行) 模型。编译后的图,内部就是一个 Pregel 对象。

BSP 把计算切成一连串 超步(super-step),每个超步内部分三个阶段,严格按顺序:

这套机制带来三个对 Agent 至关重要的性质:

• 确定性的并行:同一超步内的节点并行跑,但它们看到的是"上一超步结束时"的快照,互相看不到彼此本超步的写入——所以并行分支不会有竞态。这就是 BSP 的"同步屏障"。
• 天然支持环:循环 = "更新激活了上游节点 → 下个超步它又被选中执行"。循环不是特殊语法,而是调度的自然结果。
• 每个超步是一个原子断点:超步之间状态干净一致,可以存快照、可以中断、可以恢复——持久化和人工介入由此而来(见 1.7)。

1.3 超步执行模型:跟着一次运行走一遍

用最经典的 ReAct 图(agent ⇄ tools)走一遍,看每个超步发生了什么。

注意:"节点被激活"的唯一触发条件,是它订阅的通道在上一超步被写入了。边的本质,就是把上游节点的"写"和下游节点的"订阅"连起来。这也是为什么 LangGraph 能优雅支持并行——只要多个节点订阅了同一个被更新的通道,它们就在同一超步一起被激活、一起跑。

1.4 Channel:State 的真身

你写的 State 是个 TypedDict,但在运行时,State 的每个字段,其实是一个 Channel(通道)。这是最反直觉、也最关键的内核概念。

节点之间并不直接传参,而是通过读/写通道来通信:节点从它订阅的通道读值,执行后把结果写回某些通道;写入会激活订阅了这些通道的下游节点。通道有不同类型,决定"写入如何被存储":

1.5 Reducer = Channel 的更新规则

现在第 2 章那句 Annotated[list, add_messages] 就有了精确含义:你是在给这个字段指定它背后通道的"写入归并函数"。

复制import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # BinaryOperatorAggregate 通道:追加+按id去重
    visited:  Annotated[list, operator.add]  # 累积:列表拼接
    score:    Annotated[int, operator.add]   # 累加
    answer:   str                          # LastValue 通道:覆盖

当多个并行节点在同一超步都写了 score,reducer operator.add 决定把它们相加而不是互相覆盖。reducer 不是"语法糖",它是通道在并行/循环下保证数据正确的核心契约。

1.6 compile() 到底干了什么

builder.compile() 不是简单"打包",它把你声明式的图,翻译成一个可执行的 Pregel 运行时。

• 校验图结构:检查有没有悬空节点、START/END 是否连通、节点名冲突等。
• 把 State 字段实例化成 Channel:根据每个字段有没有 reducer,决定它是 LastValue 还是聚合通道。
• 把节点编译成 PregelNode(actor):确定每个节点"订阅哪些通道(触发条件)、写入哪些通道"。边和条件边在这一步被翻译成订阅关系。
• 挂载运行时选项:checkpointer(持久化)、interrupt_before/after(中断点)、store(长期记忆)等。

复制graph = builder.compile(
    checkpointer=memory,            # 每个超步存快照
    interrupt_before=["tools"],     # 在 tools 超步前设断点
    store=store,                    # 跨会话长期记忆
)
# 编译产物是 CompiledStateGraph,本质是一个 Pregel,也是一个 Runnable
print(type(graph))  # langgraph.graph.state.CompiledStateGraph
graph.get_graph().draw_mermaid()   # 还能导出可视化(mermaid/png)

1.7 快照(Checkpoint):持久化/时间旅行/HITL 的统一底座

这是 LangGraph 相对裸框架最大的工程优势,而它直接来自超步模型。

因为每个超步结束时状态是干净一致的,LangGraph 可以在每个超步后存一份"所有通道值"的快照(checkpoint),用 thread_id 标识属于哪个会话,用一个递增的 checkpoint_id 标识第几步。于是同一套机制解锁了四件事:

复制config = {"configurable": {"thread_id": "t1"}}
graph.invoke(inputs, config)

# 查看完整历史快照(时间旅行的入口)
for snap in graph.get_state_history(config):
    print(snap.config["configurable"]["checkpoint_id"], snap.next)

# 从某个历史 checkpoint 恢复并重跑
graph.invoke(None, {"configurable": {"thread_id":"t1", "checkpoint_id":"..."}})

1.8 Send:运行时动态 map-reduce(并行扇出)

前面的边在编译期就固定了。但有时你要在运行时根据数据量动态地"扇出"——比如把一个清单的 N 项分给 N 个并行任务处理。这就是 Send。

在条件边(或节点)里返回一个 Send 列表,LangGraph 会在下个超步为每个 Send 创建一个独立的并行任务,各带自己的局部状态。它们的输出再经累积型通道归并(reduce)。这就是经典的 map-reduce。

复制from langgraph.types import Send

# map:把 subjects 列表里每一项,扇出给一个 generate_joke 节点实例
def fan_out(state):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

builder.add_conditional_edges("pick_subjects", fan_out, ["generate_joke"])

# reduce:每个 generate_joke 把结果写进累积通道 jokes(Annotated[list, operator.add])
def generate_joke(state):
    return {"jokes": [f"关于 {state['subject']} 的笑话…"]}

1.9 Command:把"更新状态"和"决定去哪"合二为一

传统写法里,节点负责"改状态",边负责"决定下一步去哪",两者分离。Command 让一个节点同时做这两件事——很多动态路由场景因此变得干净。

复制from langgraph.types import Command
from typing import Literal

def router_node(state) -> Command[Literal["node_a", "node_b"]]:
    # 一次返回:既更新 state,又决定跳到哪个节点
    return Command(
        update={"foo": "bar"},      # 写状态
        goto="node_a",               # 路由(替代条件边)
    )

Command 还有几个高级用法,在多智能体里特别有用:

• goto=Send("worker", {...}):更新状态的同时做动态扇出。
• graph=Command.PARENT:从子图跳到父图的节点——多智能体里一个 sub-agent 干完直接把控制权交回上层 supervisor。
• Command(resume=value):配合新版 interrupt(),把人工输入送回暂停点继续(见第 3 章 HITL)。

1.10 编译后的图,也是一个 Runnable

这点把 LangGraph 和整个 LangChain 生态缝合在了一起。

CompiledStateGraph 实现了 LangChain 的 Runnable 接口,所以它和一个普通的 LCEL 链长得一样,有 invoke / stream / batch 及对应的 a* 异步版本。这意味着:

• 一个完整的 Agent 图,可以当作一个"组件",嵌进更大的 LCEL 管道里。
• 一个图可以作为子图(subgraph),直接当节点塞进另一个图(第 3 章)。
• LangSmith 的 trace、流式、回调,全都开箱即用,因为它们都面向 Runnable 接口。

复制graph.invoke(inputs, config)              # 同步,拿最终结果
for ev in graph.stream(inputs, config): ...  # 流式,逐超步
await graph.ainvoke(inputs, config)       # 异步
graph.batch([in1, in2])                  # 批量

1.11 一图总结(背下来就能讲原理)

一句话讲给面试官:"LangGraph 把 Agent 建模成可含环的有向图,运行时用 Pregel/BSP 的超步模型调度——State 的每个字段是带 reducer 的通道,节点订阅通道、写通道来通信;每个超步后存全量快照,持久化、时间旅行、人工介入都基于它;Send 做动态并行,Command 统一状态更新与路由;编译产物还是个 Runnable,能无缝接入 LangChain 生态。"