Day 03 / 共 60 天 · 阶段 1 入门与心智
StateGraph 基础:内部结构与三大方法
昨天我们把 StateGraph 当黑盒用通了。今天把盖子掀开:它 __init__ 里初始化了哪 7 个内部容器、add_node 的真实签名藏了多少能力、add_edge 为什么要分两种边、compile 到底在编译什么。看完你会发现——所谓"图",落到代码里就是几个普通的 dict 和 set。
📍 阶段 1「入门与心智」(D01-06) · 你在第 3 天
D01 全景分包→
D02 跑通第一个图→
D03 StateGraph→
D04 节点与边→
D05 State/TypedDict→
D06 对话图+小结
💡 一句话锚点
StateGraph 是个纯粹的"建造器"(builder):它自己不会执行任何图,只负责把你的节点、边、状态 schema 攒进几个内部容器里,等你 compile() 时把这些"图纸零件"翻译成真正能跑的 Pregel 引擎。今天就是把这几个容器和三大方法看穿。L01
先认清:它是建造器,不是执行器
官方在类文档里用加粗警告把这点讲死了——别对着建造器直接 invoke:
libs/langgraph/langgraph/graph/state.py:139-144
!!! warning
`StateGraph` is a builder class and cannot be used directly for execution.
You must first call `.compile()` to create an executable graph that supports
methods like `invoke()`, `stream()`, `astream()`, and `ainvoke()`.
🤔 痛点:为什么要把"搭图"和"跑图"拆成两个对象?因为搭图是一次性的、结构性的(连线、校验),跑图是反复的、运行时的(喂数据、并行调度、存档)。混在一起会让类臃肿、职责不清。拆开后:
StateGraph 只管"结构对不对",CompiledStateGraph(继承自 Pregel)只管"怎么高效跑"。这是典型的建造者模式。类的泛型签名也值得一眼——它带了 4 个类型参数,暗示了后面要学的东西:
libs/langgraph/langgraph/graph/state.py:130
class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
StateT状态类型——传送带上工件的样子(D05 专讲)。ContextT运行时上下文类型——昨天那个 {"r":3.0}(D59 Runtime 深入)。InputT / OutputT输入/输出 schema——图对外可以长得和内部状态不一样(D11 专讲)。默认都等于 StateT。L02
7 个内部容器:图纸就是这几个字典
类顶部的属性声明,就是这张"图纸"的全部家当。记住它们,后面所有方法都在往这几个容器里塞东西:
libs/langgraph/langgraph/graph/state.py:201-207
edges: set[tuple[str, str]]
nodes: dict[str, StateNodeSpec[Any, ContextT]]
branches: defaultdict[str, dict[str, BranchSpec]]
channels: dict[str, BaseChannel]
managed: dict[str, ManagedValueSpec]
schemas: dict[type[Any], dict[str, BaseChannel | ManagedValueSpec]]
waiting_edges: set[tuple[tuple[str, ...], str]]
| 容器 | 类型 | 装什么(大白话) |
|---|---|---|
nodes | dict[名字→节点规格] | 所有工位。key 是节点名,value 是 StateNodeSpec(D04 拆) |
edges | set[(起, 终)] | 普通边:单起点→单终点的连线 |
waiting_edges | set[((多起点), 终)] | "汇合边":等多个节点全完成才走(L06 讲) |
branches | dict[源节点→{名字→分支}] | 条件边:跑完某节点动态决定去哪(D13-14) |
channels | dict[字段名→通道] | 状态每个字段对应一个"通道",管它怎么存/合并(阶段⑤) |
managed | dict[字段名→托管值] | 特殊的引擎托管字段(不由你直接写) |
schemas | dict[schema 类→字段表] | 登记过的所有 schema(状态/输入/输出)解析结果缓存 |
图注:所谓"图",在建造阶段就是这一堆普通字典和集合。没什么魔法。
L03
__init__ 逐行:一开始就把 3 个 schema 解析了
看 __init__ 的核心部分(跳过一堆弃用参数兼容代码):
libs/langgraph/langgraph/graph/state.py:251-269
self.nodes = {}
self.edges = set()
self.branches = defaultdict(dict)
self.schemas = {}
self.channels = {}
self.managed = {}
self.compiled = False
self.waiting_edges = set()
self.state_schema = state_schema
self.input_schema = cast(type[InputT], input_schema or state_schema)
self.output_schema = cast(type[OutputT], output_schema or state_schema)
self.context_schema = context_schema
self._node_defaults = _NodeDefaults()
self._add_schema(self.state_schema)
self._add_schema(self.input_schema, allow_managed=False)
self._add_schema(self.output_schema, allow_managed=False)
self.nodes = {} 等七个容器全初始化成空的 dict/set。此刻是一张完全空白的图纸。input_schema or state_schema没单独传 input/output schema 时,默认都等于 state_schema——这就是昨天例子里图对外和对内长得一样的原因。D11 会讲怎么让它们不一样。self._add_schema(state_schema)关键!构造时立刻解析状态 schema,把每个字段变成 channel 塞进 self.channels。所以昨天 StateGraph(State) 那一刻,x 字段就已经变成一个通道了。allow_managed=False输入/输出 schema 里不允许出现"托管字段"——那是引擎内部用的,不该暴露给输入输出。这里用参数把规则卡死。_add_schema 内部有个很妙的"同名字段冲突"处理,体现了设计的细致:
libs/langgraph/langgraph/graph/state.py:354-364
for key, channel in channels.items():
if key in self.channels:
if self.channels[key] != channel:
if isinstance(channel, LastValue):
pass # ← 同名但新的是 LastValue:容忍
else:
raise ValueError(
f"Channel '{key}' already exists with a different type"
)
else:
self.channels[key] = channel
🎨 设计取舍①:同名字段类型冲突,为什么"LastValue 放行、其他报错"?
多个 schema(状态/输入/输出)可能都声明了同名字段
x。若两处给的通道类型不同,通常是矛盾——直接 raise 让你早发现。但有一种例外:如果新来的是最普通的 LastValue(覆盖写,无 reducer),说明它没提出特殊合并要求,那就沿用已有的那个更具体的通道(pass 掉)。好处:input schema 里随手写个 x: int(默认 LastValue)不会覆盖掉 state schema 里精心定义的 x: Annotated[list, reducer]。这是"宽进严出"——能兼容的默默兼容,真矛盾才报错。L04
add_node 的真实签名:小方法藏大能力
昨天只用了 add_node("A", node) 两个参数,但真实签名有一长串关键字参数——每一个都是后面某一天的主题:
libs/langgraph/langgraph/graph/state.py:662-675(真正的实现签名,前面 375/444/517/586 是重载声明)
def add_node(
self,
node: str | StateNode[NodeInputT, ContextT],
action: StateNode[NodeInputT, ContextT] | None = None,
*,
defer: bool = False,
metadata: dict[str, Any] | None = None,
input_schema: type[NodeInputT] | None = None,
retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
cache_policy: CachePolicy | None = None,
error_handler: StateNode[Any, ContextT] | None = None,
destinations: dict[str, str] | tuple[str, ...] | None = None,
timeout: float | timedelta | TimeoutPolicy | None = None,
**kwargs,
) -> Self:
| 参数 | 作用 | 哪天深入 |
|---|---|---|
node / action | 节点名 + 节点函数(可只传函数,自动取函数名) | 本讲 L05 |
input_schema | 这个节点单独用的输入 schema(可与全局不同) | D05/D11 |
retry_policy | 失败重试策略 | D25/D52 |
cache_policy | 结果缓存策略 | D51 |
defer | 延迟到本次运行快结束才执行 | D21/汇合场景 |
destinations | 声明"这个节点可能跳去哪"(画图/Command 用) | D15 |
timeout | 单次执行超时 | D25 |
🎨 设计取舍②:为什么
add_node 有 5 个 @overload 重载声明?
翻源码你会看到 375 / 444 / 517 / 586 四五处都是 def add_node(...)——那些是 @overload(类型重载声明),只给类型检查器看,没有函数体(body 是 ...);真正干活的实现在 662。为什么这么麻烦:add_node 支持两种调用形态——add_node(函数)(自动取名)和 add_node("名字", 函数);还要区分传不传 input_schema。用重载能让 IDE 在每种形态下都给出精确的类型提示和补全。代价是源码里同一个方法名出现好几次,读的时候要认准"哪个才是有 body 的真实现"。技巧:找那个不以 ... 结尾、有真实语句的定义。L05
add_node 内部:取名、查重、打包成 spec
实现体里最核心的三段。第一段——自动推断节点名:
libs/langgraph/langgraph/graph/state.py:768-773
if not isinstance(node, str):
action = node
if isinstance(action, Runnable):
node = action.get_name()
else:
node = getattr(action, "__name__", action.__class__.__name__)
not isinstance(node, str)如果第一个参数不是字符串(说明你直接传了函数),就把它当成 action,然后自动取名字。action.__name__普通函数取 __name__(函数名)。所以昨天 add_node(node) 会得到节点名 "node"。这也解释了官方例子里 add_node(my_node) 节点名就叫 "my_node"。第二段——一堆护栏校验(这些就是昨天报错的来源):
libs/langgraph/langgraph/graph/state.py:792-801
if node in self.nodes:
raise ValueError(f"Node `{node}` already present.") # 重名
if node == END or node == START:
raise ValueError(f"Node `{node}` is reserved.") # 占用保留名
for character in (NS_SEP, NS_END):
if character in node:
raise ValueError(
f"'{character}' is a reserved character and is not allowed in the node names."
)
node in self.nodes节点名不能重复——因为 nodes 是 dict,重名会互相覆盖,必须早报错。node == START/END不能拿保留字 __start__/__end__ 当节点名。NS_SEP / NS_END名字里不能含特殊分隔符(子图命名空间用的字符,如 :)——D49 子图会看到它们的用途。这里提前把冲突堵死。第三段——打包成 StateNodeSpec 存进 nodes(这里取最常见的分支):
libs/langgraph/langgraph/graph/state.py:897-907
self.nodes[node] = StateNodeSpec[StateT, ContextT](
coerce_to_runnable(action, name=node, trace=False),
metadata,
input_schema=self.state_schema,
retry_policy=retry_policy,
cache_policy=cache_policy,
error_handler_node=handler_node_name,
ends=ends,
defer=defer,
timeout=timeout,
)
coerce_to_runnable(action)把你的普通函数统一包装成 Runnable(langchain-core 的统一可调用接口)。这样引擎不用关心你传的是函数、lambda 还是 Runnable,一律按 Runnable 调。这正是昨天说的"只依赖 langchain-core"的用处。StateNodeSpec(...)把节点的所有信息(行为、输入schema、重试、超时、能跳去哪 ends)打成一个规格对象存进 self.nodes[node]。明天 D04 就专门拆这个 spec。🍼 一句话总结 add_node取名 → 查重/校验 → 把函数包成 Runnable、连同各种策略打成一个 spec、塞进
self.nodes 字典。就这三步,没有执行任何东西。L06
add_edge:为什么要分"普通边"和"汇合边"
add_edge 接受两种起点:单个字符串,或一个字符串列表。它俩进不同的容器:
libs/langgraph/langgraph/graph/state.py:938-967
if isinstance(start_key, str):
if start_key == END:
raise ValueError("END cannot be a start node") # END 不能当起点
if end_key == START:
raise ValueError("START cannot be an end node") # START 不能当终点
...
self.edges.add((start_key, end_key)) # ← 普通边进 edges
return self
# start_key 是列表的情况:
for start in start_key:
if start not in self.nodes:
raise ValueError(f"Need to add_node `{start}` first")
...
self.waiting_edges.add((tuple(start_key), end_key)) # ← 汇合边进 waiting_edges
单起点 → self.edges最普通的边:A 完成 → 走 B。存成 (start, end) 元组进 edges 集合。多起点列表 → self.waiting_edges汇合边(fan-in):add_edge(["A","B","C"], "D") 表示"A、B、C 全部完成后才执行 D"。存进 waiting_edges,编译时会变成一个"同步屏障"通道。END 不能当起点 / START 不能当终点方向护栏:终点之后没有下一步,起点之前没有上一步。违反直接报错。💡 本质:两种边对应两种"触发时机"普通边是"OR 触发"——任一上游到了就能推动。汇合边是"AND 触发"——必须所有指定上游都完成。区分它俩,是为了表达"三个并行分支跑完再汇总"这种 map-reduce 结构。编译时(state.py:1547-1554),汇合边会被翻译成一个
NamedBarrierValue(命名屏障)通道——D32 会精讲这个"等齐了才放行"的通道。📝 对比两种写法
g.add_edge("A", "D"); g.add_edge("B", "D")(两条普通边):A 或 B 任一完成都会触发 D,可能触发两次。g.add_edge(["A","B"], "D")(一条汇合边):等 A 和 B 都完成,D 只触发一次。做"并行汇总"必须用后者。
L07
compile:把图纸翻译成能跑的引擎
compile 签名里最常用的就是 checkpointer(存档器)——这是"持久化"的入口:
libs/langgraph/langgraph/graph/state.py:1164-1175
def compile(
self,
checkpointer: Checkpointer = None,
*,
cache: BaseCache | None = None,
store: BaseStore | None = None,
interrupt_before: All | list[str] | None = None,
interrupt_after: All | list[str] | None = None,
debug: bool = False,
name: str | None = None,
...
) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
compile 内部干的四件大事(挑核心行):
libs/langgraph/langgraph/graph/state.py:1218 / 1248 / 1257-1273
checkpointer = ensure_valid_checkpointer(checkpointer) # ① 规整存档器
...
self.validate(interrupt=(...)) # ② 校验结构
# ③ 挑出输出通道、流式通道
output_channels = (...)
stream_channels = (...)
# ④ 造出 CompiledStateGraph(继承 Pregel),把节点/边翻译成通道与 PregelNode
① ensure_valid_checkpointer把你传的 checkpointer(存档器)规整好。传 None 就是不存档,传 InMemorySaver() 就是存内存(D35)。持久化能力从这里注入。② validate()结构体检:昨天见过的"必须有入口"、边指向的节点必须存在、interrupt 的节点名必须存在——全在这一步查。结构错误集中在 compile 时爆出。③ output/stream channels算出"图跑完输出哪些字段""流式模式下推哪些字段"。默认是全部状态字段。④ 造 CompiledStateGraph核心翻译:每个节点 spec 变成一个 PregelNode,每条边变成对通道的"订阅/写入"。这一步之后,图才真的能跑(D19-24 精讲 Pregel)。翻译的证据——把节点变成 PregelNode 就发生在 attach_node:
libs/langgraph/langgraph/graph/state.py:1518-1525(attach_node 里造 PregelNode)
self.nodes[key] = PregelNode(
triggers=[branch_channel], # 被哪个通道"触发"就跑
channels=(...), # 读哪些状态字段
mapper=mapper,
writers=[ChannelWrite(write_entries)], # 结果写回哪些通道
...
bound=node.runnable, # ← 你写的那个函数在这
)
看懂这段就够了:一个节点在运行期的本质 = "被某个通道触发 → 读若干通道 → 跑
bound(你的函数)→ 把结果写回若干通道"。"边"就退化成了"谁触发谁"的通道订阅关系。这套"通道触发"模型正是 Pregel(阶段④)的核心,今天先埋下种子。⚠️ 边界:compile 后再 add_node/add_edge 不生效编译后
self.compiled=True。此时再调 add_node/add_edge,源码只会 logger.warning 提醒"加到已编译的图上不会反映到编译结果"(state.py:778-782, 932-936),不报错但也不生效。所以:先把图搭全,最后再 compile;要改结构就重新搭一个建造器。L08
今日小结 + 动手 + 明日预告
🧠 今天你应该能回答
- StateGraph 是建造器还是执行器?(建造器,必须 compile 出 CompiledStateGraph 才能跑)
- 它内部靠哪 7 个容器存图纸?(nodes/edges/waiting_edges/branches/channels/managed/schemas)
- __init__ 里为什么一上来就 _add_schema?(立刻把状态字段解析成 channel 存进 self.channels)
- 为什么 add_node 有一堆 @overload?(给不同调用形态精确类型提示;真实现在 662 行、body 不是
...) - add_node 内部三步?(取名 → 查重校验 → 包成 StateNodeSpec 塞进 nodes)
- 普通边和汇合边的区别?(单起点 OR 触发 vs 多起点 AND 触发/同步屏障)
- compile 干的四件事?(规整存档器 / validate 校验 / 挑输出通道 / 翻译成 PregelNode)
✋ 10 分钟动手
# 1. 亲眼看 7 个内部容器的声明
sed -n '201,207p' libs/langgraph/langgraph/graph/state.py
# 2. 看 __init__ 如何初始化并立刻解析 schema
sed -n '251,269p' libs/langgraph/langgraph/graph/state.py
# 3. 找到 add_node 真正的实现(有 body 的那个)
sed -n '662,676p' libs/langgraph/langgraph/graph/state.py
# 4. 在 REPL 里搭个空图,打印内部容器看看
python3 -c "from langgraph.graph import StateGraph; from typing_extensions import TypedDict; \
class S(TypedDict): x:int
g=StateGraph(S); g.add_node('A', lambda s:{'x':s['x']+1}); \
print('nodes=',list(g.nodes)); print('channels=',list(g.channels))"
# 5. 看 add_edge 怎么分流普通边/汇合边
sed -n '938,967p' libs/langgraph/langgraph/graph/state.py
💡 明日预告 · Day 04今天见到节点被打包成
StateNodeSpec、编译后变成 PregelNode。明天 D04 就专门拆这两个:graph/_node.py 里 StateNodeSpec 每个字段的含义、节点函数的 9 种签名形态(要不要 config/writer/store/runtime)、以及边在编译后怎么变成"通道订阅"。