Send:ノードファンアウトの PUSH タスクプリミティブ
役割
Send は LangGraph の「指定ノードへのタスク投げ込み」プリミティブで、libs/langgraph/langgraph/types.py で定義されています(Send class:664)。1 つの Send(node="foo", arg={"k": v}) は「次ステップで arg を入力としてノード foo を呼ぶ」ことを意味します。通常ノードによる発火との違いは、通常ノード (PULL) はチャネル購読関係で受動的に発火するのに対し、Send (PUSH) は前のノードが「foo を 1 回走らせたい、この特定の arg を付けて」と能動的に宣言する点です。
位置としては条件辺と TASKS チャネルの間にあります。add_conditional_edges(START, lambda s: [Send("gen", {"x": 1}), Send("gen", {"x": 2})]) は条件関数が返した Send リストを TASKS チャネルに書き込み (goto as sends:62)、次ラウンドの prepare_next_tasks が tasks_channel.get() でそれらを取り出し、各 Send を独立した PregelExecutableTask として組み立てます(PUSH tasks:421)。
設計動機
- map-reduce の自然な表現:典型的なシナリオは「同一ノードを N 回並行に走らせ、毎回入力を変える」です。
Send("generate", {"subject": s})を条件辺でリストとして返すと、各 Send が 1 回の独立呼び出しになり、Pregel が自然に並行実行し、reducer が N 回の結果を主状態にマージします。グラフ内で N 本のエッジを手動で展開する必要はありません。 - PUSH / PULL の二分:ノードの発火は 2 種類に分かれます——PULL (受動、購読チャネルの更新で発火)、PUSH (能動、
Sendで明示的に投げ込む)。prepare_next_tasksはまずTASKSチャネルの PUSH タスクを消費し、その後trigger_to_nodesで PULL タスクを算出します(PUSH tasks:440)。 argは主状態でなくてよい:Send.argは任意のオブジェクトで、グラフの主状態 schema に従う必要はありません——ターゲットノードのinput_schemaがargの形状を決め、「同一ノードに異種入力を投げる」ことを許可します。例えばSend("review", ReviewContext(doc=...))vsSend("review", DefaultContext())。SendとCommand.gotoは等価:Command(goto=Send(...))はmap_commandで同じTASKSチャネルを経由し (goto as sends:62)、条件辺とノード戻り値の 2 つの入り口が最終的に同じ PUSH タスクスケジュールに合流し、抽象を統一します。- 各
Sendは独立 task:Sendリスト内の N 個の要素は N 個の独立したPregelExecutableTaskで、それぞれ task_id と path を持ち、個別に retry、個別にエラー処理できます——1 つが失敗しても他の並行 Send に影響しません。
主要ファイル
Send class:664—Send(node, arg, *, timeout=None)データクラス、__slots__ = ("node", "arg", "timeout")。Send __init__:718— コンストラクタ、timeoutはTimeoutPolicy.coerceを経由します。Send __hash__:739—hash((node, arg, timeout))。futuresdict のキーとして重複排除します。prepare_next_tasks:392— メインスケジュール関数。PUSH タスクを先に消費し、その後 PULL 候補ノードを算出します。PUSH tasks:440—TASKSチャネルから Send シーケンスを取り出し、各々でprepare_single_taskを呼んでPregelExecutableTaskを組み立てます。PULL tasks:470— PULL パス。trigger_to_nodesでどのノードを走らせるか算出し、TASKSは消費しません。goto as sends:62—map_commandがCommand.gotoのSendをTASKSチャネルに書き込み、PUSH パスを統一します。add_conditional_edges:969— 条件辺の入り口。条件関数はSendまたはSendリストを返します。PregelExecutableTask:627— タスクの実行可能形態。name/input/proc/writes/triggers/pathなどを持ちます。PregelRunner.tick:176— このステップのすべての PUSH + PULL タスクを並行実行し、Sendで投げ込まれたタスクはここで走ります。
データフロー
Send が実際に消費されるのは prepare_next_tasks(prepare_next_tasks:392) の先頭です:
input_cache: dict[INPUT_CACHE_KEY_TYPE, Any] = {}
checkpoint_id_bytes = binascii.unhexlify(checkpoint["id"].replace("-", ""))
null_version = checkpoint_null_version(checkpoint)
tasks: list[PregelTask | PregelExecutableTask] = []
# Consume pending tasks
tasks_channel = cast(Topic[Send] | None, channels.get(TASKS))
if tasks_channel and tasks_channel.is_available():
for idx, _ in enumerate(tasks_channel.get()):
if task := prepare_single_task(
(PUSH, idx),
None,
checkpoint=checkpoint,
checkpoint_id_bytes=checkpoint_id_bytes,
checkpoint_null_version=null_version,
pending_writes=pending_writes,
processes=processes,
channels=channels,
managed=managed,
config=config,
step=step,
stop=stop,
for_execution=for_execution,
store=store,
checkpointer=checkpointer,
manager=manager,
input_cache=input_cache,
cache_policy=cache_policy,
retry_policy=retry_policy,
):
tasks.append(task)TASKS チャネルは Topic[Send] です——複数回の書き込み、複数回の読み出しをサポートするチャネル型です(Topic:23)。各 Send は tasks_channel.get() で取り出され、(PUSH, idx) が task_id のプレフィックスとして prepare_single_task に渡されます。後者は processes[send.node] から対応する PregelNode 設定 (proc / writers / retry_policy など) を取り出し、send.arg を input として PregelExecutableTask を組み立てます。Send.arg は主状態チャネルに入りません——直接 task.input としてノードの proc に渡され、ノード関数が受け取るのはこの arg で、グラフ主状態の全量スナップショットではありません。
Send が TASKS チャネルに入る経路は 2 つあります。(1) add_conditional_edges で設定された条件関数が Send リストを返し、attach_branch がそれらを TASKS に書き込みます。(2) ノード関数が Command(goto=Send(...)) を返し、map_command が _io.py で Send を TASKS に書き込みます(goto as sends:62)。2 つの経路は最終的に同じ TASKS チャネルに書き込み、次ラウンドの prepare_next_tasks が消費します。
境界と失敗
Sendはターゲットが登録済みノードでなければならない:processes[send.node]が存在しない場合、prepare_single_taskはNoneを返しその Send は暗黙に破棄されます——raise はしませんが、タスクは走りません。デバッグ時に Send が効かない場合は、まずターゲットノード名がadd_nodeで登録されているか確認してください。Send.argは状態 reducer を経由しない:Sendのargは直接ノードの input になり、主状態チャネルの reducer マージを経由しません。ターゲットノードが主状態を読みたい場合は、そのノード関数内でconfig[CONFIG_KEY_READ]またはstate: Stateパラメータで取得する必要があります——argは「追加の入力」です。Sendリストの重複排除は hash に依存:Send.__hash__は(node, arg, timeout)の 3 要組で計算し (Send __hash__:739)、同一ノードの 2 つの Send のargがハッシュ不可オブジェクト (dict など) だと直接TypeErrorになります——Sendは_callの dedup パスで dict キーとして使われます。- 同一ノードが PULL と PUSH で同時に発火可能:
Send("foo", arg)がfooに投げ込まれ、同時にfooのある trigger チャネルが更新された場合、prepare_next_tasksは PUSH タスクと PULL タスクを両方列出し、PregelRunner.tickが両方を並行実行します——これは許可されますが、ノード関数が同じステップで 2 回呼ばれる可能性があることに注意してください。 Send.timeoutは単独で効く:Send(node, arg, timeout=10)はその PUSH タスクにのみ効き、ターゲットノードのデフォルトtimeout設定には影響しません(Send __init__:718)。TASKSチャネルはLastValueではなくTopic:Topicは複数回書き込みの累積をサポートし (Topic:23)、同一スーパーステップで複数ノードがSendを書き込め、次ラウンドで一気に消費し切れます——これが map-reduce の基盤です。
まとめ
Send は PUSH タスクの担体で、通常の PULL ノードを補完します。条件辺と Command.goto の 2 つの入り口が同じ TASKS チャネルに合流し、prepare_next_tasks が統一的にスケジュールします。task がどう並行実行されるかは /func/concurrency、条件辺からノードへの関係は libs/langgraph/langgraph/graph/_branch.py を参照してください。公式資料:LangGraph ドキュメント · README