源码 types.py:664，本体极简——就是"目标节点 + 要喂给它的参数"的封装：

# types.py:664
class Send:
    __slots__ = ("node", "arg", "timeout")     # types.py:711
    node: str                                   # 目标节点名
    arg: Any                                    # 喂给它的输入（可与主状态完全不同）
    timeout: TimeoutPolicy | None

    def __init__(self, /, node: str, arg: Any, *,
                 timeout: float | timedelta | TimeoutPolicy | None = None) -> None:
        self.node = node
        self.arg = arg
        self.timeout = TimeoutPolicy.coerce(timeout)     # 把数字/timedelta 统一成策略对象

    def __hash__(self) -> int:                  # types.py:738
        return hash((self.node, self.arg, self.timeout))

    def __eq__(self, value: object) -> bool:    # types.py:746
        return (isinstance(value, Send) and self.node == value.node
                and self.arg == value.arg and self.timeout == value.timeout)

官方性能基准 bench/fanout_to_subgraph.py:16 就是标准 map-reduce。核心那句路由函数：

# bench/fanout_to_subgraph.py:11
class OverallState(TypedDict):
    subjects: list[str]
    jokes: Annotated[list[str], operator.add]        # ← reduce 靠这个累加器

# bench/fanout_to_subgraph.py:16 —— map：给每个主题发一个 Send
async def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

builder.add_conditional_edges(START, continue_to_jokes)   # 条件边返回 Send 列表
builder.add_node("generate_joke", subgraphc)              # 目标是一个子图
builder.add_edge("generate_joke", END)

Send 被发出后，会写进一个特殊通道 TASKS。无论来自条件边（Day 14 _finish）还是 Command 的 goto（Day 15 map_command），最终都归到这里。回看 pregel/_io.py:67：

# pregel/_io.py:66（map_command 里处理 goto 是 Send 的分支）
for send in sends:
    if isinstance(send, Send):
        yield (NULL_TASK_ID, TASKS, send)          # ← Send 写进 TASKS 通道

准备下一超步任务时，prepare_next_tasks 会先把 TASKS 通道里所有待处理的 Send 消费掉，每个变成一个 PUSH 任务。pregel/_algo.py:442：

# pregel/_algo.py:442
tasks_channel = cast(Topic[Send] | None, channels.get(TASKS))
if tasks_channel and tasks_channel.is_available():
    for idx, _ in enumerate(tasks_channel.get()):        # 遍历每个待处理 Send
        if task := prepare_single_task(
            (PUSH, idx),                                  # ← PUSH 任务：由 Send 推来
            ...
        ):

每个 Send 变成 PUSH 任务，由 prepare_push_task_send 把它组装成真正可执行的任务。pregel/_algo.py:938：

# pregel/_algo.py:938（裁剪核心）
def prepare_push_task_send(task_path, ..., channels, processes, step, ...):
    if len(task_path) == 2:
        idx = cast(int, task_path[1])                     # 第几个 Send
        if not channels[TASKS].is_available():
            return
        sends: Sequence[Send] = channels[TASKS].get()
        if idx < 0 or idx >= len(sends):
            return                                        # ← 边界：下标越界，安静跳过
        packet = sends[idx]
        if not isinstance(packet, Send):
            logger.warning(f"Ignoring invalid packet type {type(packet)} ...")
            return                                        # ← 边界：不是 Send，警告跳过
        if packet.node not in processes:
            logger.warning(f"Ignoring unknown node name {packet.node} ...")
            return                                        # ← 边界：目标节点不存在，警告跳过
        proc = processes[packet.node]                     # 找到目标节点的执行进程
        # ...（用 packet.node、idx 等算出确定性 task_id）

map 出去 N 个任务，它们并行往同一个通道写结果。能安全并行写，全靠 Day 07-10 讲的 reducer。没有 reducer 会怎样？

✋ 10 分钟动手

# 1. 读 Send 类
sed -n '664,753p' libs/langgraph/langgraph/types.py

# 2. 读官方 fanout 基准（map-reduce 范本）
sed -n '1,60p' libs/langgraph/bench/fanout_to_subgraph.py

# 3. 跑一个最小 map-reduce（注意 jokes 的 reducer）
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from typing import TypedDict, Annotated
import operator
class S(TypedDict):
    subjects: list; jokes: Annotated[list, operator.add]
def fan(state): return [Send("joke", {"s": x}) for x in state["subjects"]]
def joke(state): return {"jokes": [f"关于{state['s']}的笑话"]}
g = StateGraph(S)
g.add_node("joke", joke)
g.add_conditional_edges(START, fan)   # 返回 Send 列表
g.add_edge("joke", END)
print(g.compile().invoke({"subjects": ["猫","狗","鱼"], "jokes": []}))
PY

# 4. 把 jokes 的 Annotated 去掉，观察 InvalidUpdateError（体会 reducer 的必要性）