【LangGraph基础】图构建与编译解析
目录
本文从源码层面拆解 LangGraph 的 StateGraph 是如何从「几个 API 调用」变成「一张可执行的 Pregel 计算图」的。核心结论只有一句话：
add_node/add_edge/add_conditional_edges只是「声明」，compile()才是「构建」。声明阶段产出的是一堆普通的数据结构（StateNodeSpec、BranchSpec、edges集合），编译阶段把它们翻译成执行引擎真正认的PregelNode+channels+writers。
全文围绕三个示例展开（均位于 pregel/ 目录）：
| 示例 | 对应阶段 | 回答的问题 |
|---|---|---|
graph_builder_internals.py | 构建 | add_node / add_edge / add_conditional_edges 内部做了什么、产物是什么 |
compile_internals.py | 编译 | compile() 有哪些步骤、步骤如何衔接 |
compiled_graph.py | 编译后 | 编译出来的计算图长什么样、执行引擎怎么用它 |
一、两个阶段的心智模型
StateGraph 是一个 builder 类，不能直接执行（state.py:131-145 文档明确说明必须 .compile() 后才有 invoke/stream）。所以整个流程天然分两段：
flowchart LR
subgraph build["① 构建阶段（builder）"]
A["add_node"] --> N["self.nodes<br/>dict[str, StateNodeSpec]"]
B["add_edge"] --> E["self.edges / waiting_edges<br/>set[tuple]"]
C["add_conditional_edges"] --> BR["self.branches<br/>dict[source][name]=BranchSpec"]
end
subgraph compile["② 编译阶段（compile）"]
N --> P["attach_node"]
E --> AE["attach_edge"]
BR --> AB["attach_branch"]
P --> G["PregelNode + channels"]
AE --> G
AB --> G
end
G --> X["可执行 Pregel 图"]
- 构建阶段：三个方法只往
StateGraph实例的容器里塞数据，不产生任何可执行逻辑。 - 编译阶段：
compile()读取这些容器，把它们翻译成PregelNode/channels/writers。
二、构建阶段：三个 API 记录了什么
2.1 容器初始化
StateGraph.__init__（state.py:252-259）先建好几个空容器：
self.nodes = {} # dict[str, StateNodeSpec] ← add_node 的产物
self.edges = set() # set[tuple[str, str]] ← add_edge 单源的产物
self.branches = defaultdict(dict) # dict[source][name] ← add_conditional_edges 的产物
self.channels = {} # dict[str, BaseChannel] ← 状态字段通道
self.waiting_edges = set() # set[tuple[tuple[str,...], str]] ← add_edge 多源的产物
self.compiled = False这五个容器就是后续一切的「原料仓库」。
2.2 add_node：从函数到 StateNodeSpec
add_node（state.py:782-920）内部做了五件事：
flowchart TD
A["解析节点名<br/>非 str 时取 __name__"] --> B["校验<br/>重名 / 保留名 / 非法字符 | 和 :"]
B --> C["推断 schema<br/>input_schema + ends(Command[Literal])"]
C --> D["若有 error_handler<br/>额外注册 __error_handler__{node}"]
D --> E["coerce_to_runnable(action)<br/>包装成 Runnable"]
E --> F["self.nodes[node] = StateNodeSpec"]
产物：StateNodeSpec（graph/_node.py:91-103）：
class StateNodeSpec(Generic[NodeInputT, ContextT]):
runnable: StateNode # coerce_to_runnable 包装后的 Runnable
metadata: dict[str, Any] | None
input_schema: type # 输入 schema（第一参数类型推断）
retry_policy: ... # 重试策略
cache_policy: ... # 缓存策略
is_error_handler: bool = False
error_handler_node: str | None = None # 该节点的 error handler 节点名
ends: tuple | dict | None # 从 Command[Literal[...]] 推断的可达目标
defer: bool = False
timeout: TimeoutPolicy | None = None
trace_policy: TracePolicy | None = None关键点：这一步只包装了函数、记录了配置，还没生成 PregelNode、也没有触发通道。
2.3 add_edge：边就是集合里的一个元组
add_edge（state.py:928-980）极其简单：
if isinstance(start_key, str):
self.edges.add((start_key, end_key)) # 单源
else:
self.waiting_edges.add((tuple(start_key), end_key)) # 多源产物：
- 单源边 →
self.edges（set[tuple[str, str]]），如('__start__', 'agent') - 多源边 →
self.waiting_edges（set[tuple[tuple[str,...], str]]），表示「等所有源都完成后才触发目标」
2.4 add_conditional_edges：路由函数到 BranchSpec
add_conditional_edges（state.py:982-1030）：
flowchart TD
A["coerce_to_runnable(path)<br/>包装路由函数"] --> B["取分支名<br/>path.name 或 'condition'"]
B --> C["BranchSpec.from_path<br/>归一化 path_map"]
C --> D["self.branches[source][name] = BranchSpec"]
BranchSpec.from_path（graph/_branch.py:88-120）把 path_map 归一化成 dict[Hashable, str]：
- 传
dict→ 直接 copy - 传
list→{name: name for name in path_map} - 都不传 → 从路由函数返回类型的
Literal[...]推断
产物：BranchSpec（graph/_branch.py:83-86）：
class BranchSpec(NamedTuple):
path: Runnable # 包装后的路由函数
ends: dict[Hashable, str] | None # {返回值: 目标节点}，如 {'tools': 'tools', 'end': '__end__'}
input_schema: type | None = None注意：BranchSpec 只是一个描述——「用这个函数路由，返回值按这个表映射到目标」。真正的动态路由逻辑要到编译时才被挂到源节点上。
三、编译阶段：compile() 的 11 个步骤
compile()（state.py:1177-1401）把上面的声明翻译成可执行图。整体分 11 步：
flowchart TD
A["① ensure_valid_checkpointer<br/>处理 checkpointer"] --> B["② serde allowlist<br/>序列化白名单(可选)"]
B --> C["③ validate<br/>验证图结构"]
C --> D["④ 准备 output/stream channels<br/>__root__ 或字段列表"]
D --> E["⑤ 应用 builder defaults<br/>默认 retry/cache/timeout/error_handler"]
E --> F["⑥ node_error_handler_map<br/>{节点: handler节点}"]
F --> G["⑦ 构造 CompiledStateGraph<br/>nodes={} 空，channels={状态字段+START}"]
G --> H["⑧ attach_node × N<br/>建 branch:to:{key} + PregelNode"]
H --> I["⑨ attach_edge × N<br/>普通边 → 源节点 writer"]
I --> J["⑩ attach_branch × N<br/>条件边 → 源节点 branch writer"]
J --> K["⑪ validate<br/>返回编译好的图"]
3.1 步骤 ①~⑥：纯数据准备
前六步不碰图结构，只在 builder 上做校验和配置下放：
| 步骤 | 做什么 | 产物 |
|---|---|---|
| ① | ensure_valid_checkpointer | 规范化 checkpointer（None/False/实例） |
| ② | strict-msgpack 下收集所有 schema | 序列化白名单 serde_allowlist |
| ③ | 验证节点名、边端点存在性 | 无（校验通过才继续） |
| ④ | 确定输出通道 | output_channels / stream_channels |
| ⑤ | 默认 retry_policy/cache_policy/timeout/error_handler 下放给 spec（per-node 优先） | 修改后的 StateNodeSpec |
| ⑥ | 收集「节点 → error handler 节点」 | node_error_handler_map |
3.2 步骤 ⑦：构造空壳
state.py:1346-1370 构造 CompiledStateGraph，此时：
nodes={} # 空，等 ⑧ 填充
channels={
**self.channels, # 状态字段通道（如 messages → BinaryOperatorAggregate）
**self.managed, # managed values
START: EphemeralValue(self.input_schema), # 入口通道
}这是关键：⑦ 时 channels 里只有「状态字段通道」，还没有 branch:to:{node}。那些触发通道是 ⑧ 才新建的。
3.3 步骤 ⑧⑨⑩：三者的衔接核心
这三个步骤是编译的精髓，它们都往源节点的 writers 列表里追加 writer：
flowchart LR
subgraph attach["对每个节点执行"]
N8["⑧ attach_node<br/>新建 branch:to:{key} 通道<br/>PregelNode(writers=[状态更新, Command路由])"]
end
subgraph edge["对每条边执行"]
N9["⑨ attach_edge(start, end)<br/>给 start.writers 追加<br/>ChannelWrite(branch:to:{end})"]
end
subgraph branch["对每条条件边执行"]
N10["⑩ attach_branch(source, name)<br/>给 source.writers 追加<br/>branch.run(...) 动态路由"]
end
N8 --> N9 --> N10
⑧ attach_node（state.py:1444-1549）为每个节点：
- 确定
output_keys（START 用input_schema，普通节点用builder.channels） - 定义
_get_updates：把节点返回的dict转成[(key, value), ...]元组 - 构造两个
ChannelWriteTupleEntry（状态更新 +_control_branch处理 Command 路由） - 调用
_compile_node：新建branch:to:{key}通道（EphemeralValue），生成PregelNode(triggers=[branch:to:{key}], channels=输入字段, writers=[ChannelWrite(write_entries)])
⑨ attach_edge（state.py:1551-1578）给源节点追加一个 writer：
self.nodes[starts].writers.append(
ChannelWrite((ChannelWriteEntry(_CHANNEL_BRANCH_TO.format(end), None),))
)即：「边」被翻译成「源节点执行完后，往目标节点的 branch:to:{end} 通道写一个信号」。
⑩ attach_branch（state.py:1578-1624）给源节点追加动态路由 writer：
self.nodes[start].writers.append(branch.run(get_writes, reader))get_writes 把路由函数选中的目标包成 ChannelWriteEntry(branch:to:{target})，运行时动态决定写哪个通道。
3.4 用实际输出看衔接
以「ReAct 图」为例（compile_internals.py），编译后三个节点的 writers 分别是：
'__start__': writers = [ ['<tuple>','<tuple>'], ['branch:to:agent'] ]
└── ⑧ 状态更新+Command ──┘ └── ⑨ attach_edge(START→agent) ──┘
'agent': writers = [ ['<tuple>','<tuple>'], <branch:RunnableCallable> ]
└── ⑧ ──┘ └── ⑩ attach_branch(agent) ──┘
'tools': writers = [ ['<tuple>','<tuple>'], ['branch:to:agent'] ]
└── ⑧ ──┘ └── ⑨ attach_edge(tools→agent) ──┘一眼能看出：writers 列表 = ⑧ 的状态更新 writer + ⑨/⑩ 追加的边/分支 writer。而目标节点的 triggers=[branch:to:{target}] 一旦被写就触发。这就是「边 → 触发」的编译本质。
四、编译后的计算图：执行引擎看到的东西
编译完成后，CompiledStateGraph 里有三样执行引擎真正关心的结构（见 compiled_graph.py）。
4.1 channels：状态通道
每个状态字段对应一个 channel，类型由 reducer 决定：
flowchart LR
F["State 字段"] -->|"Annotated[x, add_messages]"| B["BinaryOperatorAggregate<br/>(追加消息)"]
F -->|"Annotated[int, operator.add]"| A["BinaryOperatorAggregate<br/>(累加)"]
F -->|"str（无 reducer）"| L["LastValue<br/>(覆盖写)"]
F -->|"触发通道"| E["EphemeralValue<br/>(branch:to:{node})"]
实际输出：
messages BinaryOperatorAggregate reducer=add_messages
token_count BinaryOperatorAggregate reducer=add
weather LastValue # 覆盖写
branch:to:weather EphemeralValue # 触发通道BinaryOperatorAggregate：reducer 语义（追加/累加），多个节点写同一字段时合并而非覆盖。LastValue：普通字段，后写覆盖先写。EphemeralValue：触发通道，只做「本步有没有被写」的信号，用完即清。
4.2 trigger_to_nodes：通道 → 订阅节点
这是执行引擎的反向索引，回答「某通道被写时，该唤醒哪些节点」：
__start__ -> ['planner']
branch:to:planner -> ['planner']
branch:to:weather -> ['weather']
branch:to:news -> ['news']
branch:to:summarize -> ['summarize']prepare_next_tasks 每轮就是靠它，把「上一步被写的通道」映射成「下一步要跑的节点集合」。
4.3 节点的「订阅 / 读 / 写」三视图
每个 PregelNode（pregel/_read.py:97-150）有三组关键字段：
| 字段 | 含义 | 示例 |
|---|---|---|
triggers | 哪些通道被写会触发它 | ['branch:to:weather'] |
channels | 读哪些通道作为输入 | ['plan', 'token_count'] |
writers | 执行完往哪些通道写 | [状态更新, branch:to:summarize] |
4.4 条件边的运行时形态
条件边在编译时被 attach_branch 包成一层动态 writer：路由函数 path() 运行时返回目标，再经 BranchSpec.ends 映射成 branch:to:{target} 写入。所以：
flowchart LR
P["planner 执行完"] --> R["route() 返回 ['weather','news']"]
R -->|"ends 映射"| W["写 branch:to:weather<br/>写 branch:to:news"]
W --> T1["触发 weather"]
W --> T2["触发 news"]
T1 --> S["同一 superstep 内并行"]
T2 --> S
一个路由返回列表，就会在同一 superstep 内并行唤醒多个分支。
4.5 运行时：reducer 与 superstep
编译只是静态结构，真正动起来看两个运行时现象：
reducer 合并并行写入（compiled_graph.py 场景 5）：
输入 token_count=1
三个节点各写 token_count：planner=100, weather=50, news=50
最终 token_count = 201 ← 1+100+50+50（BinaryOperatorAggregate 累加）superstep 边界（stream_mode="updates"）：
flowchart LR
S1["superstep 1<br/>planner"] --> S2["superstep 2<br/>weather + news 并行"]
S2 --> S3["superstep 3<br/>summarize"]
并行分支在同一个 superstep 内完成，各自写入 token_count 后，在 superstep 边界由 reducer 一次性合并——这正是 Pregel/BSP（批量同步并行）模型的体现。
五、总结
整条链路可以压缩成一张图：
flowchart TD
subgraph build["构建：只记录声明"]
A["add_node"] --> NS["StateNodeSpec<br/>(runnable + 配置)"]
B["add_edge"] --> ED["edges / waiting_edges<br/>(set[tuple])"]
C["add_conditional_edges"] --> BS["BranchSpec<br/>(path + ends 映射)"]
end
subgraph compile["编译：翻译成执行图"]
NS --> AT["attach_node<br/>→ PregelNode + branch:to:{key}"]
ED --> AE["attach_edge<br/>→ 源节点追加 ChannelWrite"]
BS --> AB["attach_branch<br/>→ 源节点追加动态路由 writer"]
end
AT --> G["channels + PregelNode + writers"]
AE --> G
AB --> G
G --> R["执行：写 channel → 触发下一批节点<br/>(superstep 循环)"]
三个最值得记住的点：
- 两个阶段：
add_*只声明、compile才构建，中间产物是StateNodeSpec/BranchSpec/edges集合。 - 边 = 触发：普通边和条件边最终都被翻译成「源节点的 writer 写往目标节点的
branch:to:{target}通道」，而目标节点的triggers订阅了该通道。 - 通道类型决定语义：
BinaryOperatorAggregate（reducer 合并）、LastValue（覆盖）、EphemeralValue（触发信号）、Topic（Send 队列）——同一个「写 channel」动作，因通道类型不同而语义不同。
掌握这条「声明 → 数据结构 → 触发图」的链路，就能看懂 LangGraph 源码里大部分构建与编译相关的实现。
附录：完整源代码
示例一：graph_builder_internals.py（构建阶段：三个 API 的中间产物）
"""详解 add_node / add_edge / add_conditional_edges 内部做了什么。
关键认知：StateGraph 是「builder」，三个方法只负责「记录声明」，真正「构建执行图」
发生在 compile()。所以产物分两层：
【builder 阶段产物】（调 add_* 时产生，存进 StateGraph 实例）
- self.nodes : dict[str, StateNodeSpec] ← add_node 的产物
- self.edges : set[tuple[str, str]] ← add_edge(单源) 的产物
- self.waiting_edges : set[tuple[tuple[str, ...], str]] ← add_edge(多源) 的产物
- self.branches : dict[source][name] = BranchSpec ← add_conditional_edges 的产物
【compile 阶段产物】（compile() 时由上面四条构建成可执行图）
- self.nodes : dict[str, PregelNode] ← 执行节点（triggers/channels/writers/bound）
- self.channels : dict[str, BaseChannel] ← 含 branch:to:{node} 触发通道
- self.edges : 执行边（含条件边展开）
本示例用一个最小 ReAct 图，在 compile 前后分别打印这些中间产物，看清每一步生成了什么。
"""
import operator
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
# ---- 三个普通函数：节点 / 节点 / 路由 ----
def agent_node(state: State) -> dict:
"""节点1：调用模型（此处仅示意，不真的调 LLM）。"""
return {"messages": [("ai", "agent 思考中...")]}
def tools_node(state: State) -> dict:
"""节点2：执行工具。"""
return {"messages": [("tool", "工具结果")]}
def route(state: State) -> Literal["tools", "end"]:
"""条件路由：最后一条消息带 tool_calls 就去 tools，否则结束。"""
return "end"
# ============================================================
# 构建图：调用三个方法，观察它们往 builder 里塞了什么
# ============================================================
builder = StateGraph(State)
# ---------- add_node：包装 runnable + 记录 StateNodeSpec ----------
builder.add_node("agent", agent_node)
builder.add_node("tools", tools_node)
# ---------- add_edge：记录一条固定边 ----------
builder.add_edge(START, "agent")
# ---------- add_conditional_edges：包装路由 + 记录 BranchSpec ----------
builder.add_conditional_edges("agent", route, {"tools": "tools", "end": END})
# ---------- add_edge：工具执行完回到 agent，形成循环 ----------
builder.add_edge("tools", "agent")
print("=" * 70)
print("① builder 阶段产物（compile 之前）")
print("=" * 70)
print("\n[add_node 的产物] builder.nodes：dict[str, StateNodeSpec]")
for name, spec in builder.nodes.items():
print(f" {name!r}:")
print(f" runnable 类型 = {type(spec.runnable).__name__}")
print(f" input_schema = {spec.input_schema.__name__}")
print(f" ends(从 Command[Literal] 推断) = {spec.ends}")
print("\n[add_edge 的产物] builder.edges：set[tuple[start, end]]")
print(f" {builder.edges}")
print("\n[add_edge(多源) 的产物] builder.waiting_edges：set[tuple[tuple, end]]")
print(f" {builder.waiting_edges}")
print("\n[add_conditional_edges 的产物] builder.branches：dict[source][name]=BranchSpec")
for source, branches in builder.branches.items():
for name, b in branches.items():
print(f" {source!r} -> {name!r}:")
print(f" path 类型 = {type(b.path).__name__}")
print(f" ends(path_map) = {b.ends}")
# ============================================================
# compile：把上面的声明构建成可执行图
# ============================================================
graph = builder.compile()
print("\n" + "=" * 70)
print("② compile 阶段产物（compile 之后）")
print("=" * 70)
print("\n[结果产物] graph.nodes：dict[str, PregelNode]")
for name, node in graph.nodes.items():
writers = []
for w in node.writers:
if hasattr(w, "writes"):
# ChannelWriteEntry 有 .channel；ChannelWriteTupleEntry 只有 .mapper
writers.append([getattr(e, "channel", "<tuple>") for e in w.writes])
print(f" {name!r}:")
print(f" triggers = {node.triggers} # 哪些 channel 被写会触发它")
print(f" channels = {node.channels} # 读哪些 channel 作为输入")
print(f" writers 写往 = {writers} # 执行后往哪些 channel 写")
print("\n[结果产物] graph.channels：dict[str, BaseChannel]")
for ch_name, ch in graph.channels.items():
print(f" {ch_name!r}: {type(ch).__name__}")
print("\n[结果产物] graph.get_graph().edges：执行边（Edge 结构）")
for e in graph.get_graph().edges:
print(f" Edge(source={e.source!r}, target={e.target!r}, conditional={e.conditional})")
print("\n[图结构] mermaid：")
print(graph.get_graph().draw_mermaid())示例二：compile_internals.py（编译阶段：compile 的 11 个步骤）
"""compile() 内部发生了什么：11 个步骤的逐步拆解。
compile() 把 builder 阶段的声明（StateNodeSpec / edges / waiting_edges / branches）
转成可执行图（PregelNode / channels / writers）。步骤链见 state.py:1177-1401：
① ensure_valid_checkpointer 处理 checkpointer 参数（None/False/实例）
② serde allowlist strict-msgpack 下构建序列化白名单（可选）
③ validate 验证图结构（节点名 / 边端点存在性）
④ 准备 output/stream channels 确定输出通道（__root__ 或状态字段列表）
⑤ 应用 builder defaults 默认 error_handler/retry/cache/timeout 下放给 spec
⑥ node_error_handler_map 生成 {节点名: error_handler 节点名}
⑦ 构造 CompiledStateGraph nodes={} 空、channels={状态字段 + managed + START}
⑧ attach_node × N 每个节点 → branch:to:{key} 通道 + PregelNode
⑨ attach_edge × N 普通边 → 给【源节点】追加 writer（写 branch:to:{目标}）
⑩ attach_branch × N 条件边 → 给【源节点】追加 branch writer（动态路由）
⑪ validate 最终验证，返回编译好的图
核心衔接：⑧⑨⑩ 都往「源节点的 writers 列表」里追加 writer，
每个 writer 都写往【目标节点的 branch:to:{target}】通道，从而把边变成「触发」。
本示例构建一个 ReAct 图，打印 compile 前后每个步骤的产物，看 writer 如何一步步累积。
"""
import operator
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
def agent_node(state: State) -> dict:
return {"messages": [("ai", "思考中...")]}
def tools_node(state: State) -> dict:
return {"messages": [("tool", "工具结果")]}
def route(state: State) -> Literal["tools", "end"]:
return "end"
# ---- builder 阶段：声明（前一个示例已详解，这里快速带过）----
builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", tools_node)
builder.add_edge(START, "agent") # → self.edges
builder.add_conditional_edges("agent", route, {"tools": "tools", "end": END}) # → self.branches
builder.add_edge("tools", "agent") # → self.edges
print("=" * 72)
print("步骤 ⑦ 之前：builder 阶段的输入")
print("=" * 72)
print(f" nodes = {list(builder.nodes)}")
print(f" edges = {builder.edges}")
print(f" branches = { {s: list(ns) for s, ns in builder.branches.items()} }")
# ---- compile：一次性跑完 ①~⑪ ----
graph = builder.compile()
print("\n" + "=" * 72)
print("步骤 ⑦⑧ 的产物：channels + PregelNode")
print("=" * 72)
print("\n⑦ 构造 CompiledStateGraph 时初始化的 channels：")
print(" （状态字段 + managed + START 的 EphemeralValue）")
for ch, val in graph.channels.items():
tag = " ← ⑦ 初始就有" if ch in ("messages", "__pregel_tasks", "__start__") else " ← ⑧ attach_node 新建"
print(f" {ch!r}: {type(val).__name__}{tag}")
print("\n⑧ attach_node 给每个节点构建 PregelNode：")
print(" （triggers = [branch:to:{key}]，writers = [状态更新, Command 路由]）")
for name, node in graph.nodes.items():
writers = []
for w in node.writers:
if hasattr(w, "writes"):
writers.append([getattr(e, "channel", "<tuple>") for e in w.writes])
else:
writers.append(f"<branch:{type(w).__name__}>")
print(f" {name!r}:")
print(f" triggers = {node.triggers}")
print(f" writers = {writers}")
print("\n" + "=" * 72)
print("步骤 ⑨⑩ 的产物：普通边 / 条件边如何「追加」到源节点 writers")
print("=" * 72)
print("""
⑨ attach_edge(START, 'agent') → 给 __start__.writers 追加 ChannelWrite(branch:to:agent)
⑨ attach_edge('tools', 'agent') → 给 tools.writers 追加 ChannelWrite(branch:to:agent)
⑩ attach_branch('agent', route) → 给 agent.writers 追加 branch.run(...)（动态路由 writer）
所以最终：
__start__ 触发时，往 branch:to:agent 写 → 触发 agent
agent 的条件分支 writer 往 branch:to:tools 或 branch:to:__end__ 写 → 路由
tools 执行完，往 branch:to:agent 写 → 回到 agent（循环）
""")
print("=" * 72)
print("步骤 ⑪ 的产物：最终验证后的执行边（Edge 结构）")
print("=" * 72)
for e in graph.get_graph().edges:
print(f" Edge(source={e.source!r}, target={e.target!r}, conditional={e.conditional})")
print("\n图结构 mermaid：")
print(graph.get_graph().draw_mermaid())示例三：compiled_graph.py（编译后的计算图内部结构）
"""展示 graph.compile() 之后，Pregel 计算图的内部结构。
用接近真实 agent 的复杂 State，覆盖：
- LastValue channel（普通字段，覆盖写）
- BinaryOperatorAggregate channel（reducer：追加 / 累加）
- 边 → branch:to:xxx 信号 channel
- 条件边 → ends 映射 + _control_branch 动态写
- 并行分支 → 一个 superstep 内唤醒多个节点
"""
from operator import add as add_reducer
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages] # 累积
plan: str # 覆盖
token_count: Annotated[int, add_reducer] # 累加
weather: str # 覆盖
news: str # 覆盖
def planner(state: AgentState) -> dict:
return {"plan": "并行查询天气和新闻", "token_count": 100}
def call_weather(state: AgentState) -> dict:
return {"weather": "sunny", "token_count": 50}
def call_news(state: AgentState) -> dict:
return {"news": "market up", "token_count": 50}
def summarize(state: AgentState) -> dict:
return {"messages": [("ai", f"{state['weather']} / {state['news']}")]}
def route(state: AgentState) -> list[str]:
"""返回列表 → 同一 superstep 内并行触发多个分支。"""
return ["weather", "news"]
g = StateGraph(AgentState)
g.add_node("planner", planner)
g.add_node("weather", call_weather)
g.add_node("news", call_news)
g.add_node("summarize", summarize)
g.add_edge(START, "planner")
g.add_conditional_edges(
"planner", route, {"weather": "weather", "news": "news"}
)
g.add_edge("weather", "summarize")
g.add_edge("news", "summarize")
g.add_edge("summarize", END)
app = g.compile()
BAR = "=" * 72
# ============ 1. channels ============
print(BAR)
print("1. channels（状态通道）")
print(BAR)
for name, ch in app.channels.items():
detail = ""
if hasattr(ch, "reducer"):
# 看 reducer 是哪个函数
fn = getattr(ch.reducer, "func", ch.reducer)
detail = f" reducer={getattr(fn, '__name__', fn)}"
print(f" {name:<22} {type(ch).__name__:<26}{detail}")
# ============ 2. trigger_to_nodes ============
print("\n" + BAR)
print("2. trigger_to_nodes（channel → 订阅它的节点）")
print(BAR)
for trigger, nodes in app.trigger_to_nodes.items():
print(f" {trigger:<22} -> {list(nodes)}")
# ============ 3. 条件边的定义（builder 上的 BranchSpec）============
print("\n" + BAR)
print("3. 条件边：builder.branches 里的 BranchSpec")
print(BAR)
for src, branches in g.branches.items():
for name, spec in branches.items():
print(f" source={src} name={name}")
print(f" ends（path 返回值 → 目标节点）: {spec.ends}")
# ============ 4. 节点详情 ============
print("\n" + BAR)
print("4. 节点详情（订阅 / 读 / 写）")
print(BAR)
for name, node in app.nodes.items():
static_writes = [] # 编译时就能确定的 channel（静态边）
dynamic_mappers = [] # 运行时才能确定（state 更新 / Command / Send）
for w in node.writers:
for entry in getattr(w, "writes", ()):
if ch := getattr(entry, "channel", None):
static_writes.append(ch)
elif mapper := getattr(entry, "mapper", None):
dynamic_mappers.append(getattr(mapper, "__name__", str(mapper)))
# 条件边：编译时被 attach_branch 包成一层 Runnable，写入目标由 path 运行时决定
has_branch = name in g.branches
print(f"\n node [{name}]")
print(f" 订阅 triggers = {node.triggers}")
print(f" 读取 channels = {node.channels}")
print(f" 静态边写入 = {static_writes}")
print(f" 动态写入 mapper = {dynamic_mappers}")
if has_branch:
print(f" 条件边 = path() 返回值经 ends 映射决定")
# ============ 5. 运行时：reducer 的实际效果 ============
print("\n" + BAR)
print("5. 运行时：看 reducer 怎么合并并行写入")
print(BAR)
result = app.invoke({"messages": [("user", "查询")], "token_count": 1})
print(f" 输入 token_count=1")
print(f" 三个节点各写 token_count: planner=100, weather=50, news=50")
print(f" 最终 token_count = {result['token_count']} ← 1+100+50+50（累加）")
print(f" 最终 messages 条数 = {len(result['messages'])} ← 追加而来")
print(f" 最终 {result['messages'][-1].content!r}")
# ============ 6. 执行顺序（superstep）============
print("\n" + BAR)
print("6. 执行顺序：superstep 边界")
print(BAR)
for chunk in app.stream(
{"messages": [("user", "查询")], "token_count": 0}, stream_mode="updates"
):
for node, update in chunk.items():
print(f" superstep 内节点 [{node}] 更新: {list(update.keys())}")