add_node / add_edge / add_conditional_edges:往蓝图里填内容

源码版本1.2.9

职责

StateGraph 提供三个核心注册方法:add_node 把一个函数 / Runnable 包成节点 (add_node:662),add_edge 加一条确定的边 (add_edge:915),add_conditional_edges 加一个分支条件 (add_conditional_edges:969)。这三个方法都返回 Self,所以可以链式调:builder.add_node("a", a).add_node("b", b).add_edge("a", "b").add_conditional_edges("b", route)。它们都只往 builder.nodes / builder.edges / builder.branches 这几个容器里追加,真正的翻译成 PregelNode + 通道订阅要等 compile() 时由 attach_node / attach_edge / attach_branch 完成。

它们的位置在用户 API 层最外面:你写图时的每一行几乎都在调这三个方法。add_node 是最复杂的——它会从函数签名推 input_schema、从返回类型 Literal[...] 推 destinations、把 error_handler 包成一个独立的 __error_handler__{node} 节点 (error handler 注入:856-870)。add_edge 比较直接,要么进 edges set,要么进 waiting_edges set (多入边,要等所有起点都完成)。add_conditional_edges 最薄,只把 path 包成 Runnable,再交给 BranchSpec.from_path(from_path:89)算出 ends 字典存进 branches[source]。

设计动机

为什么 add_node 要从函数签名推这么多东西?

• 少写样板:不显式传 input_schema 时,从第一个参数的类型注解推 (inferred input schema:815-825),让 def my_node(state: MyState): ... 自动用 MyState;返回类型是 Literal["a", "b", "__end__"] 时自动当作 destinations (Literal destinations:840-846),不用写 destinations=...。
• 节点名可省:node 参数既可以是字符串也可以是函数本身,函数对象会取 __name__ 当节点名 (node 名推断:768-773)——Runnable 则用 get_name()。
• error_handler 是独立节点(error_handler 节点:857-870)——它不是 spec 的字段,而是生成一个名为 __error_handler__{node} 的常规节点,再在 error_handler_node 字段里记下指针。这样 handler 自己也是 Pregel 节点,享受 retry / metadata / tracing 等所有节点能力。
• 多入边语义统一:add_edge(["a", "b"], "c") 不是「a 跑完跑 c,b 跑完也跑 c」,而是「a 和 b 都跑完才跑 c」(multi-start 语义:917-921),存进 waiting_edges set 等编译时翻译成 NamedBarrierValue 通道。
• 条件边的返回值可省 path_map:不传 path_map 时,从 path 函数的返回类型 Literal["a", "b"] 推 (Literal 推 path_map:103-115),让 def route(state) -> Literal["a", "b"]: ... 自动工作;都没给时图可视化会假设它可能跳到任意节点 (warning:994-997)。
• 同节点同名的 branch 报错(branch 重名:1009-1012)——一个节点可以有多个 conditional edge,但每个的 condition name 必须唯一,因为 branches[source] 是 dict by name。

关键文件

• add_node 签名:662-676 — 接收 node / action / defer / metadata / input_schema / retry_policy / cache_policy / error_handler / destinations / timeout。
• node 名推断:768-790 — 字符串就用字符串,函数用 __name__,Runnable 用 get_name();保留字 START / END / NS_SEP / NS_END 会报错。
• 签名类型推断:803-848 — 从 __call__ 的 type hints 推 inferred_input_schema 和返回类型里的 Command[Literal[...]] destinations。
• error_handler 注入:856-870 — 生成 __error_handler__{node} 独立节点存进 self.nodes,字段 is_error_handler=True。
• 存 StateNodeSpec:872-907 — 按 input_schema / inferred_input_schema / self.state_schema 三档把 spec 存进 self.nodes。
• add_edge:915-967 — 单起点进 edges set,多起点进 waiting_edges set,起点不能是 END、终点不能是 START。
• add_conditional_edges:969-1017 — 把 path 包成 Runnable,BranchSpec.from_path 算 ends,存进 self.branches[source][name]。
• BranchSpec.from_path:83-120 — 处理 path_map 三种形态:dict / list / 都没给就从返回类型 Literal 推;另外推 input_schema。
• add_sequence:1019-1044 — 一次性注册一串节点(语法糖,内部还是循环 add_node)。
• attach_edge:1537-1561 — 编译期:单入边向起点节点的 writers 加一条 ChannelWrite 指向终点通道;多入边注册一个 NamedBarrierValue 通道当 join。
• attach_branch:1563-1596 — 编译期:把 BranchSpec 翻成 ChannelWrite.register_writer,根据 ends 决定写到哪个目标通道。

数据流

add_node 真正干活的就这两段——推断 input schema 和返回类型里的 destinations,然后把 spec 存进 self.nodes:

if (
    isfunction(action)
    or ismethod(action)
    or ismethod(getattr(action, "__call__", None))
) and (
    hints := get_type_hints(getattr(action, "__call__"))
    or get_type_hints(action)
):
    if input_schema is None:
        first_parameter_name = next(
            iter(inspect.signature(cast(FunctionType, action)).parameters.keys())
        )
        if input_hint := hints.get(first_parameter_name):
            if isinstance(input_hint, type) and get_type_hints(input_hint):
                inferred_input_schema = input_hint
    if rtn := hints.get("return"):
        rtn_origin = get_origin(rtn)
        if rtn_origin is Union:
            rtn_args = get_args(rtn)
            for arg in rtn_args:
                arg_origin = get_origin(arg)
                if arg_origin is Command:
                    rtn = arg
                    rtn_origin = arg_origin
                    break
        if (
            rtn_origin is Command
            and (rargs := get_args(rtn))
            and get_origin(rargs[0]) is Literal
            and (vals := get_args(rargs[0]))
        ):
            ends = vals

(类型推断:806-846)

最后按推断结果三档存进 self.nodes 字典——input_schema 显式 / inferred_input_schema 推出 / fallback 到 self.state_schema:

if input_schema is not None:
    self.nodes[node] = StateNodeSpec[NodeInputT, ContextT](
        coerce_to_runnable(action, name=node, trace=False),
        metadata,
        input_schema=input_schema,
        retry_policy=retry_policy,
        cache_policy=cache_policy,
        error_handler_node=handler_node_name,
        ends=ends,
        defer=defer,
        timeout=timeout,
    )

(StateNodeSpec:872-883)

add_conditional_edges 倒是非常薄——核心是把 path 包成 Runnable,再让 BranchSpec.from_path 算 ends 字典:

path = coerce_to_runnable(path, name=None, trace=True)
name = path.name or "condition"
if name in self.branches[source]:
    raise ValueError(
        f"Branch with name `{path.name}` already exists for node `{source}`"
    )
self.branches[source][name] = BranchSpec.from_path(path, path_map, True)
if schema := self.branches[source][name].input_schema:
    self._add_schema(schema)
return self

(add_conditional_edges 实现:1005-1017)

边界与失败

• 节点名重复直接报错(重名:792-793)——Node \` already present`,不允许覆盖。
• 节点名保留字(reserved:794-801)——START / END / NS_SEP (|) / NS_END (:) 都不能用,后两个会破坏 checkpoint_ns 的拼接。
• add_edge 起点是 END 报错(END 不能当起点:939-940),同理 START 不能当终点(START 不能当终点:941-942)——这两个常量是图入口出口,不能被反转。
• 多入边要求所有起点先 add_node(多入边校验:956-964)——不然编译时 attach_edge 找不到起点的 writers。
• 同节点同名 branch 报错(branch 重名:1009-1012),但不同名的多个 branch 可以共存——这是 conditional + deterministic 混合路由的基础。
• add_node 在已编译图上只 warning 不报错(compiled warning:778-782),改动不会反映到已编译实例——典型顺序坑。

小结

add_node / add_edge / add_conditional_edges 是 StateGraph 的写入接口,本身不复杂,复杂的是从函数签名推 schema 和 destinations 的「魔法」。理解了这三个方法,基本就读完了 StateGraph 的所有 mutable API;下一步是 compile 怎么把这些 spec 翻译成 PregelNode,运行模型见 Pregel 引擎。

对照官方资料:LangGraph 文档 · README。