第 19 讲 · Send 动态并行与函数式 API

学习目标

看懂 Send 的完整生命周期：从产出到变成并行 PUSH 任务。
理解 map-reduce 在 LangGraph 里的实现。
理解函数式 API（@entrypoint / @task）如何编译成 Pregel。

一、Send：运行时动态扇出

回顾第 7/8 讲：条件边的路由函数可以返回 Send(node, arg) 列表。 Send 解决的是"运行时才知道要并行多少个子任务"的问题——比如对 N 篇文档分别处理，N 在运行时才确定。

Send 的生命周期

types.py 的 Send（664–752 行）。它走的是 PUSH 任务路径，与普通边触发（PULL）并列：

sequenceDiagram
    participant E as 条件边/Branch
    participant CW as ChannelWrite
    participant TC as TASKS(Topic)通道
    participant PN as prepare_next_tasks
    participant Node as 目标节点(并行多份)
    E->>E: 返回 [Send("worker", arg1), Send("worker", arg2), ...]
    E->>CW: writer 处理 destinations
    CW->>TC: _assemble_writes: Send → (TASKS, Send)
    Note over TC: 超步结束, TASKS 通道更新
    PN->>TC: 消费 pending sends
    PN->>Node: 每个 Send 生成一个 PUSH 任务
    Note over Node: 多个 worker 并行执行, input=Send.arg

关键代码位置：

Send → TASKS：_write.py 的 _assemble_writes（172–192 行，第 13 讲见过）。
消费 Send：_algo.py 的 prepare_next_tasks（441–466 行，第 10 讲）。
构建 PUSH 任务：_algo.py 的 prepare_push_task_send（938–1107 行）—— 任务 input = packet.arg（每个 Send 携带的独立参数，不是全图 state）。

map-reduce 三段式

Send docstring（683–708 行）给的经典例子：

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.types import Send

class State(TypedDict):
    subjects: list[str]
    jokes: Annotated[list, operator.add]   # reduce 用聚合通道

# map: 为每个 subject 扇出一个子任务
def continue_to_jokes(state):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

# 每个 worker 产出一条, 经 operator.add reduce 聚合
def generate_joke(state):
    return {"jokes": [f"joke about {state['subject']}"]}

map：continue_to_jokes 按运行时数据扇出 N 个 Send。
parallel：N 个 generate_joke 在同一超步并行（第 12 讲 max_concurrency 控并行度）。
reduce：jokes 用 Annotated[list, operator.add]（第 4 讲）把 N 份结果合并；若需要严格"等全部到齐"，配合屏障通道（第 5 讲）。

注意 worker 的 state 参数是 Send.arg（{"subject": s}），不是完整图 state—— 这是 PUSH 任务与 PULL 任务的本质区别。

二、函数式 API：@entrypoint / @task

libs/langgraph/langgraph/func/__init__.py。函数式 API 给你一种"不画图、直接写函数"的方式，但底层仍然编译成 Pregel，享受同样的 checkpoint/中断/重试能力。

@task：可调度的子任务

@task 装饰器（109–251 行）把一个函数包装成 _TaskFunction（59–106 行）。调用它返回一个 future，可 .result() / await：

from langgraph.func import task, entrypoint

@task
def double(x: int) -> int:
    return x * 2

调用链（与第 13 讲的 PUSH 路径相通）：

double(3)
  → _TaskFunction.__call__
  → _call_with_options (_call.py 276–298)
  → CONFIG_KEY_CALL → _runner._call (_runner.py 700–786)
  → schedule_task → accept_push (_loop.py 543–580)
  → prepare_push_task_functional (_algo.py 800–935)
  → 返回 future

可见 @task 和 Send 是 PUSH 任务的两条路径：Send 是图路由产出的，@task 是函数调用产出的。两者都通过 accept_push 动态加入当前超步的任务。

@entrypoint：把函数编译成图

entrypoint.__call__（516–620 行）把被装饰的函数编译成一个最小 Pregel：

单节点 PregelNode(bound=函数, triggers=[START])。
通道：START（EphemeralValue 输入）、END（LastValue 输出）、 PREVIOUS（跨 thread 持久化"上一次的返回值"）。
writers：返回值写 END，save 值写 PREVIOUS。

from langgraph.checkpoint.memory import InMemorySaver

@entrypoint(checkpointer=InMemorySaver())
def workflow(inputs):
    a = double(inputs["x"])     # @task, 并行调度
    b = double(inputs["y"])
    return a.result() + b.result()   # fan-out + reduce

entrypoint.final：返回值与存档值分离

entrypoint.final（475–514 行）允许"对外返回 A，但 checkpoint 存 B"：

@entrypoint(checkpointer=saver)
def counter(n, *, previous=None):
    total = (previous or 0) + n
    return entrypoint.final(value=total, save=total)  # 返回 total, 也存 total 供下次 previous

适用于"对外暴露的结果"和"内部累积状态"不一致的场景（计数器、累加器）。

使用场景

批处理 fan-out：对一批文档/查询/子问题并行处理（Send 或 @task）。
轻量 workflow：不想画 StateGraph，用 @entrypoint + @task 写得像普通 async 代码，但白拿 checkpoint/中断/重试。
Map-Reduce 聚合：搜索结果合并、多 Agent 投票、并行评分汇总。
@entrypoint + interrupt：函数式 API 同样支持 interrupt() + Command(resume=...)（第 18 讲）。

动手实验

实现上面的 jokes map-reduce，设 max_concurrency 观察并行度，确认 jokes 被聚合。
用 @entrypoint + @task 写一个并行 fan-out，对比 future.result() 聚合的结果。
用 entrypoint.final 写一个计数器，多次 invoke 同 thread，观察 previous 累积。

阅读作业

精读 types.py 的 Send（664–752 行）与 _algo.py 的 prepare_push_task_send（938–1107 行）。
精读 func/__init__.py 的 task（109–251 行）与 entrypoint.__call__（516–620 行）。
浏览 _call.py 的 _call_with_options（276–298 行）与 _runner.py 的 _call（700–786 行）。

小结

Send 走 PUSH 路径：写 TASKS 通道 → prepare_next_tasks 消费 → 并行子任务（input=arg），实现运行时 map-reduce。
函数式 API 仍编译成 Pregel：@entrypoint = 单节点图，@task = 另一条 PUSH 路径。
entrypoint.final 分离"返回值"与"存档值"。

最后一讲：流式输出体系与 ReAct Agent 拆解——把全课串成一个完整 Agent。