Day 03 / 共 60 天 · 阶段 1 入门与心智

StateGraph 基础:内部结构与三大方法

昨天我们把 StateGraph 当黑盒用通了。今天把盖子掀开:它 __init__ 里初始化了哪 7 个内部容器、add_node 的真实签名藏了多少能力、add_edge 为什么要分两种边、compile 到底在编译什么。看完你会发现——所谓"图",落到代码里就是几个普通的 dictset

📍 阶段 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]]
容器类型装什么(大白话)
nodesdict[名字→节点规格]所有工位。key 是节点名,value 是 StateNodeSpec(D04 拆)
edgesset[(起, 终)]普通边:单起点→单终点的连线
waiting_edgesset[((多起点), 终)]"汇合边":等多个节点完成才走(L06 讲)
branchesdict[源节点→{名字→分支}]条件边:跑完某节点动态决定去哪(D13-14)
channelsdict[字段名→通道]状态每个字段对应一个"通道",管它怎么存/合并(阶段⑤)
manageddict[字段名→托管值]特殊的引擎托管字段(不由你直接写)
schemasdict[schema 类→字段表]登记过的所有 schema(状态/输入/输出)解析结果缓存
数据结构:StateGraph 建造器内部长这样 graph : StateGraph nodes {}"A"→spec, "B"→spec edges set()(START,"A"),("A",END) branches {}条件边(D13) channels {}"x"→LastValue/BinOp waiting_edges set()汇合边 schemas / managedschema 解析缓存 compiled: bool = False(编译后置 True) add_node/add_edge 本质:往上面这些普通 dict/set 里塞条目
图注:所谓"图",在建造阶段就是这一堆普通字典和集合。没什么魔法。
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.pyStateNodeSpec 每个字段的含义、节点函数的 9 种签名形态(要不要 config/writer/store/runtime)、以及边在编译后怎么变成"通道订阅"。
← Day 02 · 跑通第一个图 Day 04 · 节点与边 →