add_node / add_edge / add_conditional_edges: ブループリントに中身を詰める
役割
StateGraph は 3 つのコアな登録メソッドを提供します。add_node は関数 / Runnable をノードとして包み (add_node:662)、add_edge は確定的なエッジを追加し (add_edge:915)、add_conditional_edges は分岐条件を追加します (add_conditional_edges:969)。これら 3 つのメソッドはすべて 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 とチャネル (channel) 購読への実際の変換は compile() 時に attach_node / attach_edge / attach_branch が行います。
これらはユーザー API 層の最外周に位置し、グラフを書くときのほぼすべての行がこの 3 つのメソッドを呼びます。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_edgesset に格納され、コンパイル時に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]は name をキーにした dict だからです。
主要ファイル
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の 3 段階で spec をself.nodesに格納します。add_edge:915-967— 単一起点ならedgesset、多起点ならwaiting_edgesset に入ります。起点にENDは不可、終点にSTARTは不可です。add_conditional_edges:969-1017—pathを Runnable に包み、BranchSpec.from_pathがendsを算出し、self.branches[source][name]に格納します。BranchSpec.from_path:83-120—path_mapの 3 形態 (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 が実際に作業をするのは次の 2 段です——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最後に推論結果に従って self.nodes 辞書に 3 段階で格納します——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,
)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(:) は使用不可。後者 2 つはcheckpoint_nsの連結を壊すためです。 add_edgeの起点がENDならエラー(END 不能当起点:939-940)。同様にSTARTは終点になれません (START 不能当终点:941-942)——これら 2 つの定数はグラフの入口/出口であり、逆転できません。- 多入辺はすべての起点が先に
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 を推論する「マジック」です。この 3 つのメソッドを理解すれば、StateGraph のすべての mutable API をほぼ読み終えたことになります。次は compile がこれらの spec をどう PregelNode に変換するか、実行モデルは Pregel エンジン を参照してください。
公式資料:LangGraph ドキュメント · README