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