第 17 章并行执行与 Send API（Map-Reduce）

LLM 应用中存在大量天然可并行的任务形态：多个主题各自独立生成摘要、同一查询并行到 Web/知识库/数据库三类数据源检索、批量文档分别抽取结构化字段、多模型并行回答再由评审者择优。LangGraph 基于 Pregel 计算模型（Pregel Model，以 Google Pregel 图计算框架为蓝本的 Bulk Synchronous Parallel 模型），同一超步（Superstep）内被选中的多个节点会通过 asyncio 事件循环并发执行，到超步末自动执行 Barrier 汇聚（等待全部前驱节点完成且所有 Channel 写入合并后才调度后继节点），天然支持 Fan-out/Fan-in 语义。当并行任务数在编译期可确定时（例如固定 3 个数据源）使用静态 Fan-out；当任务数运行时才能确定（N 随输入变化，如批量主题 N 个、用户上传文档 N 份）时使用 Send API 做动态 Fan-out，实现经典的 Map-Reduce 并行模式。

17.1 静态 Fan-out 与 Pregel 并行语义

静态 Fan-out 指从一个节点出发，通过多条 add_edge 或条件边返回字符串列表，在编译期即可确定并行分支数。例如多源检索场景：

python

g.add_edge("planner", "web_retriever")
g.add_edge("planner", "db_retriever")
g.add_edge("planner", "kb_retriever")
# 或用条件边：g.add_conditional_edges("planner", lambda s: ["web_retriever","db_retriever","kb_retriever"])
g.add_edge("web_retriever", "aggregator")
g.add_edge("db_retriever", "aggregator")
g.add_edge("kb_retriever", "aggregator")

三条 retriever 节点在 planner 完成后的同一超步内并发执行（asyncio.gather 语义）；由于三条边都汇聚到 aggregator，Pregel 的 Fan-in/Barrier 语义自动让 aggregator 等待三个 retriever 全部完成（且每个 retriever 产生的 Channel 写入都合并到最新状态版本）后才在下一超步被调度。aggregator 执行时读取的是三个 retriever 结果合并后的完整 State，不需要做轮询或回调。

静态 Fan-out 的关键特性是共享全局 State：所有并行分支读取的是 planner 写入后的同一 State 版本（快照隔离），不会互相读到对方未提交的中间写入；各分支的返回值按字段 reducer 合并回全局 State。这使静态 Fan-out 非常适合"同一输入、多个不同处理、结果合并"的场景。其局限在于分支数在编译期固定——当输入是 N 个主题、N 份文档，N 每次调用都不同，无法预先 add_edge N 次，也无法在条件边中硬编码一个长度固定的列表（N 最大可能上百），此时需要 Send API。

Pregel 模型对并行的支撑有三点基础机制：(1) Plan 阶段：每个超步开始时根据 Channel 版本变化与 triggers 订阅关系计算本超步要调度的节点集合；(2) Execute 阶段：所有被选中节点通过 asyncio.gather 并发执行，单节点失败立即记录异常；(3) Update 阶段：全部节点成功完成后，将各节点返回的写入按 reducer 合并到 Channel，生成新版本 State；任一节点失败则本次写入全部不合并（超步原子性），异常上抛。Barrier 就位于 Update 阶段之前，保证后继节点读取的是完整的合并后状态。

17.2 Send API 与动态 Fan-out

Send 类定义在 langgraph.types 模块，签名为 Send(node_name: str, arg: dict)。其核心语义与使用约束必须严格遵守：

返回位置：Send 对象只能在条件边的 path 函数返回值中出现（返回单个 Send 或 List[Send]），不能在普通节点函数内"直接 return Send"——节点函数返回 Send 不会触发派发，Send 对象会被当作普通 dict 尝试合并到 State，引发类型错误。
arg 是目标节点的局部 State：Send 派发出去的任务其入参是 arg dict，目标节点被调用时接收到的 State 就是这个 dict（而不是父图全局 State）；任务的返回值按返回 dict 的键与父图 State Schema 字段匹配后（经 reducer）合并回父图全局 State。这是 Send 与静态 Fan-out 的关键区别——静态 Fan-out 共享同一份全局 State，Send 每个任务独立入参、互相隔离。
同一节点可被多个 Send 指向：N 个 Send 指向同一个 worker 节点时，worker 会在同一超步内被 asyncio 并发调用 N 次，每次传入不同 arg，N 个返回值按 reducer 合并。
Reducer 是强制需求：因为多个 worker 并发返回结果，目标结果字段如果是 LastValue 通道（未声明 reducer）会在 Update 阶段触发 InvalidUpdateError: Can receive only one value per step——同一超步内多个节点对同一 LastValue 字段多次写入发生冲突。list 等需要聚合的字段必须声明 reducer：Annotated[list, operator.add]（列表拼接）、Annotated[list, add_messages]（消息合并，去重、按 ID 覆盖）。

典型 Map-Reduce 结构由四类节点 + 一个路由函数组成：

Map 节点：读取父输入，生成待处理任务列表（topics、docs、urls、chunk 等），写入全局 State 的 items 字段；本身不做并行工作，只做任务切分。
路由函数（条件边 path）：读取 items 字段，返回 [Send("worker", {"item": x, "item_id": i}) for i, x in enumerate(items)]，把每个 item 独立派发给 worker。返回值必须是 Send 列表（可混合其他节点名）。
Worker 节点：输入类型为 WorkerState（局部 State，只含 item/item_id 等必要字段），处理完成后返回部分结果（如 {"summaries": [summary_x]}），结果字段必须有 reducer。
Reduce 节点：在所有 worker 完成后的下一超步被调度，读取合并后的 results/summaries 列表，做最终汇总、排序、总结、写入 final_output 字段，产出最终输出。

并发度控制。LangGraph 默认将 List[Send] 中的全部任务在同一超步并发启动，asyncio 在 IO 密集场景（LLM 调用、HTTP 请求、数据库查询）能高效调度数千协程，协程本身切换成本极低；但 LLM API 通常有 RPM（每分钟请求数）/TPM（每分钟 token 数）速率限制，并发过高会立即触发 429 Too Many Requests。两种限流方式：(1) Worker 内信号量限流：在 Worker 节点内使用 asyncio.Semaphore(N) 包裹真正的 LLM/IO 调用，保证同时在飞的请求数不超过 N，这种方式简单且所有任务在同一超步启动、按信号量分批执行；(2) 分批 Send：在 Map 阶段不一次性派发全部任务，而是每批 N 个 Send 派发后经过 reducer 聚合，再由下一个路由节点检查是否还有未处理 items，派掉下一批，通过图结构本身控制并发度，这种方式每批都形成独立超步，checkpoint 更细粒度，故障恢复更友好。CPU 密集任务（大规模 embedding 计算、本地模型推理、图像处理）应使用 ProcessPoolExecutor 或把计算放到外部服务，不要在 asyncio 事件循环内做阻塞运算。

错误隔离。默认情况下，任一 Worker 抛出未捕获异常会导致整个超步失败（Pregel 超步原子性），异常冒泡到 invoke/stream 调用方，整个 Map-Reduce 任务终止。若希望单个任务失败不影响其他任务（部分失败返回部分结果），必须在 Worker 内部 try/except，将错误信息包装为普通返回值（如 {"summaries": [], "errors": [{"item_id": i, "error": str(e), "retryable": True}]}），由 Reduce 节点统一做错误处理——对 retryable 错误可再次派发 Send 重试、对 fatal 错误记录日志并在最终输出中标注失败项。

Send 的持久化。pending_sends 列表（本超步尚未完成的 Send 任务）在每个超步末随 Checkpoint 一起落盘。进程崩溃、服务重启或人工 interrupt 后恢复执行时，未完成的 Send 任务会被重新调度（基于最后一次成功的 Checkpoint 中的 pending_sends），已完成的 worker 结果已合并到 Channel，不会重复执行，避免 Map-Reduce 中途失败后从头再来。这要求 Send 的 arg 必须可序列化（可 JSON/pickle 序列化），不要放入 LLM 实例、DB 连接、锁、lambda、打开的文件句柄等不可 pickle 对象。

完整示例：批量主题摘要 Map-Reduce。

python

# langgraph==1.2.x
import asyncio, operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
LLM_SEMAPHORE = asyncio.Semaphore(3)  # 并发度上限 3

class MapState(TypedDict):
    topics: list[str]
    summaries: Annotated[list[str], operator.add]
    errors: Annotated[list[dict], operator.add]

class WorkerState(TypedDict):
    item_id: int
    item: str

def map_node(s: MapState) -> dict:
    return {"summaries": [], "errors": []}

def dispatch(s: MapState):
    # 路由函数：返回 Send 列表
    return [Send("worker", {"item_id": i, "item": t})
            for i, t in enumerate(s["topics"])]

async def worker(s: WorkerState) -> dict:
    try:
        async with LLM_SEMAPHORE:
            msg = await llm.ainvoke([("user", f"请用一句话总结：{s['item']}")])
        return {"summaries": [msg.content], "errors": []}
    except Exception as e:
        return {"summaries": [], "errors": [{"item_id": s["item_id"], "error": str(e)}]}

def reduce_node(s: MapState) -> dict:
    # Reducer 已自动合并 summaries/errors，此处做最终排序与报告
    ordered = sorted(zip(s.get("summaries", []),
                         [i for i in range(len(s.get("summaries", [])))]))
    return {"summaries": [x[0] for x in ordered]}

g = StateGraph(MapState)
g.add_node("map", map_node)
g.add_node("worker", worker, retry=__import__("langgraph.errors", fromlist=["RetryPolicy"]).RetryPolicy(max_attempts=2))
g.add_node("reduce", reduce_node)
g.add_edge(START, "map")
g.add_conditional_edges("map", dispatch, {"worker": "worker"})
g.add_edge("worker", "reduce")
g.add_edge("reduce", END)
graph = g.compile()

result = graph.invoke({"topics": ["量子计算", "核聚变", "脑机接口", "生成式AI", "自动驾驶"]})
for i, s in enumerate(result["summaries"], 1):
    print(f"{i}. {s}")

进阶示例：多源检索 Fan-out。对同一查询并发到 Web/知识库/数据库三类数据源，分支数固定为 3，用静态 Fan-out 更清晰（三条 add_edge 汇聚到 aggregator）。若要把"查询 N 个不同知识库"做成动态 N（知识库列表从用户配置或数据库读取），则用 Send：Map 节点读取当前可用知识库列表，dispatch 路由函数 Send N 个 retriever worker。两种模式可以在同一张图里组合使用（例如外层静态 Fan-out 到三种数据源类型，每种类型内部又 Send 到该类型的 N 个具体数据源）。

本章小结

LangGraph 基于 Pregel BSP 模型天然支持同超步节点 asyncio 并发与 Barrier 汇聚。静态 Fan-out 通过多条 add_edge 或条件边返回字符串列表实现，分支数编译期固定，所有分支共享同一份全局 State 快照，结果按 reducer 合并；动态 Fan-out 通过 Send API 实现，分支数运行时由输入数据决定，每个 Send 携带独立局部 State 派发给目标节点，worker 被并发调用 N 次，每个 worker 独立入参、结果按 reducer 合并。Map-Reduce 标准结构由 Map 节点、dispatch 路由函数、Worker 节点、Reduce 节点组成，结果聚合字段必须声明 reducer（list 用 Annotated[list, operator.add]、消息用 add_messages）。并发度通过 asyncio.Semaphore（限流）或分批 Send（图级分批）控制；错误隔离必须在 Worker 内 try/except 包装；pending_sends 随 Checkpoint 持久化支持故障恢复，要求 Send arg 可序列化。CPU 密集任务需迁出 asyncio 线程池到 ProcessPool 或外部服务。

常见错误

忘记 reducer 导致 "Can receive only one value per step"。现象：多个 Send 指向同一 worker 执行时抛 InvalidUpdateError。原因：结果字段未声明 reducer，多个并发写入互相冲突，LastValue Channel 不允许同一超步多次写入。修复：list 字段声明 Annotated[list, operator.add]；消息字段用 Annotated[list, add_messages]；数值字段若需累加自定义 reducer。
Send 的 arg 传完整 State 导致序列化开销巨大。现象：Checkpoint 写入明显变慢、pending_sends 体积暴增、内存占用高。原因：arg 传了包含全量 messages、长文档、工具历史的完整父 State。修复：arg 只传 worker 必要字段（如 {"item_id": i, "item": x}），大体积数据通过 Store 或工具按需读取。
Send 中传不可序列化对象。现象：checkpointer 写入时 TypeError/PicklingError，或故障恢复时反序列化失败。原因：arg 中放入了 LLM 实例、DB 连接、threading.Lock、lambda、打开的文件句柄。修复：arg 只放基础类型、dict、Pydantic 模型；LLM/DB 在 worker 函数内部按需创建或通过 config 上下文（get_store/get_config）获取。
Send 返回位置错误（在节点函数内直接 return Send）。现象：worker 不被调度，Send 对象被当作普通 dict 尝试写入 State，类型错误或字段错误。原因：Send 只能在条件路由的 path 函数返回值中出现，节点函数 return 值必须是 State 增量 dict。修复：节点返回普通 state 增量（如写入 items 字段），在 add_conditional_edges 挂的 path 函数中返回 Send 列表。
并发度过高触发 API rate limit 导致雪崩。现象：大量 429 错误、所有重试一起打导致 API 长期不可用、整批任务失败。原因：Send N 个 worker（N 上百）全部同时调 LLM，瞬时流量远超 RPM 限额。修复：asyncio.Semaphore 严格限流（常用 N=5-20）；批量任务分批 Send 循环（每批 10 个，批间经过 reduce 节点）；对 429 配置 RetryPolicy 指数退避+jitter；必要时在 LLM 客户端加令牌桶。

生产建议

Worker 节点保持纯函数式：输入完全来自 arg、输出走 reducer、不直接读写全局可变对象、不依赖全局状态（除 config/store），便于重试、故障恢复与单元测试；
大批量任务（>50）采用分批 Send 模式，每批大小按 API RPM 限额设置（一般 10-20），每批经 reducer 汇聚到中间结果字段后再派发下一批，避免一次性并发上百请求；
Worker 内统一加 try/except 包装，返回结构化错误对象（item_id、error_type、message、retryable 布尔），Reduce 节点对 retryable 错误可收集后做二次派发（重新 Send）或写入死信队列告警；
Send 的 arg 中携带稳定的 item_id（业务主键，不要依赖列表索引），便于幂等重试（worker 先按 item_id 查询是否已处理）与审计追踪；
IO 密集任务（LLM/HTTP/DB 查询）直接用 async 节点+Semaphore；CPU 密集任务（大规模 embedding 本地计算、图像处理、复杂数据转换）通过 ProcessPoolExecutor 或外部任务队列（Celery/Arq/RQ）承载，避免阻塞 asyncio 事件循环拖慢所有并发任务；
Map-Reduce 图必须配置 checkpointer 并开启 checkpoints 流模式，长任务中断后可 resume，已完成的 worker 不会重复执行（基于 Checkpoint 中已合并的 Channel 值恢复）；
Send 派发前对 items 做去重（按 item_id），避免重复数据生成重复 worker；
监控 Map-Reduce 任务：在 custom 流模式中推送 {"phase":"map_progress","done":15,"total":100} 进度事件，前端实时展示并行任务完成百分比。

C 语言

语言点

深度

新标准

面试整理

第 17 章 并行执行与 Send API（Map-Reduce） ​

17.1 静态 Fan-out 与 Pregel 并行语义 ​

17.2 Send API 与动态 Fan-out ​

本章小结 ​

常见错误 ​