Send: the PUSH task primitive for node fan-out

源码版本1.2.9

Responsibilities

Send is LangGraph's "deliver a task to a specific node" primitive, defined in libs/langgraph/langgraph/types.py (Send class:664). A Send(node="foo", arg={"k": v}) means "on the next step, invoke node foo with arg as input". The difference from ordinary node triggering is: ordinary nodes (PULL) are passively triggered by channel subscription relationships, while Send (PUSH) is the previous node actively declaring "I want foo to run once, with this specific arg".

It sits between conditional edges and the TASKS channel: add_conditional_edges(START, lambda s: [Send("gen", {"x": 1}), Send("gen", {"x": 2})]) writes the list returned by the conditional function into the TASKS channel (goto as sends:62). On the next round, prepare_next_tasks pulls them out of tasks_channel.get(), assembling each Send into an independent PregelExecutableTask (PUSH tasks:421).

Design motivation

• Natural expression of map-reduce: The classic scenario is "run the same node N times in parallel with different inputs". Send("generate", {"subject": s}) returns a list from a conditional edge; each Send is an independent call, Pregel executes them concurrently, and the reducer merges the N results back into the main state. No need to manually expand N edges in the graph.
• PUSH / PULL dichotomy: Node triggering falls into two categories — PULL (passive, triggered by updates to subscribed channels) and PUSH (active, explicitly delivered by Send). prepare_next_tasks consumes PUSH tasks from the TASKS channel first, then computes PULL tasks via trigger_to_nodes (PUSH tasks:440).
• arg does not have to be the main state: Send.arg can be any object and is not forced to conform to the graph's main state schema — the target node's input_schema decides the shape of arg, allowing "heterogeneous inputs to the same node", e.g. Send("review", ReviewContext(doc=...)) vs Send("review", DefaultContext()).
• Send is equivalent to Command.goto: Command(goto=Send(...)) goes through the same TASKS channel in map_command (goto as sends:62). So conditional edges and node return values converge on the same PUSH task scheduling — a unified abstraction.
• Each Send is an independent task: N elements in a Send list become N independent PregelExecutableTasks, each with its own task_id and path. They can be retried and error-handled individually — one failure does not affect other concurrent Sends.

Key files

• Send class:664 — The Send(node, arg, *, timeout=None) data class; __slots__ = ("node", "arg", "timeout").
• Send __init__:718 — Constructor; timeout goes through TimeoutPolicy.coerce.
• Send __hash__:739 — hash((node, arg, timeout)), used as the dedup key for the futures dict.
• prepare_next_tasks:392 — The main scheduling function; consumes PUSH tasks first, then computes PULL candidate nodes.
• PUSH tasks:440 — Takes the Send sequence out of the TASKS channel; each is passed to prepare_single_task and assembled into a PregelExecutableTask.
• PULL tasks:470 — PULL path; computes which nodes should run based on trigger_to_nodes; does not consume TASKS.
• goto as sends:62 — map_command writes the Send from Command.goto into the TASKS channel, unifying the PUSH path.
• add_conditional_edges:969 — Conditional edge entry; the conditional function returns a Send or a list of Sends.
• PregelExecutableTask:627 — The executable form of a task; holds name / input / proc / writes / triggers / path and so on.
• PregelRunner.tick:176 — Concurrently executes all PUSH + PULL tasks for this step; tasks delivered by Send are run here.

Data flow

Send is actually consumed at the top of 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)

The TASKS channel is a Topic[Send] — a channel type that supports multiple writes and multiple reads (Topic:23). Each Send is pulled out of tasks_channel.get(); (PUSH, idx) is passed as the task_id prefix to prepare_single_task, which pulls the corresponding PregelNode config (proc / writers / retry_policy, etc.) from processes[send.node] and assembles send.arg as input into a PregelExecutableTask. Send.arg does not enter the main state channel — it is passed directly to the node proc as task.input. What the node function receives is this arg, not a full snapshot of the main graph state.

There are two paths for Send to enter the TASKS channel: (1) a conditional function configured by add_conditional_edges returns a list of Send, and attach_branch writes them into TASKS; (2) a node function returns Command(goto=Send(...)), and map_command in _io.py writes the Send into TASKS (goto as sends:62). Both paths converge — write into the TASKS channel, consume on the next round by prepare_next_tasks.

Boundaries and failure modes

• Send must target a registered node: When processes[send.node] does not exist, prepare_single_task returns None and the Send is silently dropped — no raise, but the task does not run. If you notice a Send not taking effect during debugging, first check whether the target node name was registered in add_node.
• Send.arg does not go through the state reducer: Send's arg is passed directly as node input, without going through the main state channel's reducer merge. If the target node expects to read the main state, it must obtain it inside the node function via config[CONFIG_KEY_READ] or a state: State parameter — arg is just "extra input".
• Send list dedup relies on hash: Send.__hash__ uses the (node, arg, timeout) triple (Send __hash__:739). If two Sends to the same node have an unhashable arg (like a dict), it raises TypeError directly — Send is used as a dict key in the dedup path of _call.
• The same node can be triggered by both PULL and PUSH: If Send("foo", arg) delivers to foo while one of foo's trigger channels is also updated, prepare_next_tasks will list both the PUSH task and the PULL task, and PregelRunner.tick will run both concurrently — this is allowed, but be aware the node function may be called twice within the same step.
• Send.timeout is per-task: Send(node, arg, timeout=10) only applies to this PUSH task and does not affect the target node's default timeout config (Send __init__:718).
• The TASKS channel is a Topic, not a LastValue: Topic supports accumulated multiple writes (Topic:23), so multiple nodes can all write Sends into it during the same superstep, and the next round consumes them all at once — this is the foundation of map-reduce.

Summary

Send is the carrier for PUSH tasks, complementing ordinary PULL nodes. Conditional edges and Command.goto converge on the same TASKS channel and are uniformly scheduled by prepare_next_tasks. For how tasks are run concurrently see /func/concurrency; for the conditional-edge-to-node relationship see libs/langgraph/langgraph/graph/_branch.py.

See official docs: LangGraph docs · README.