LangGraph并行节点并发执行

可视化演示

展示 Map-Reduce 并行执行模型：Map 阶段分发任务到多个 Worker 并发执行，Reduce 阶段合并结果。以及并发控制与错误处理策略。

Map并行拆分：任务扇出

婚宴筹备最紧张的时刻，是主厨在开工前把厚厚一沓菜单摊开，飞快地勾画出每个帮厨该负责的菜品。这个“拆菜单、派活”的动作，对应 LangGraph 里的 Map 阶段。入口节点收到一个输入后，并不直接处理它，而是把它拆成许多互不依赖的子任务，每个子任务带着自己的一份“小菜单”（独立的状态快照）流向不同的执行分支。

在 LangGraph 中，这个“派单”由一个叫 Send 的指令完成。你可以把 Send 想象成一张张派单小票：每张票上写着一个子任务的函数和数据。更巧妙的是，这个小票可以动态开具——Map 阶段根据输入内容决定开几张票、每张票的数据结构是什么。这就像主厨看到客人有海鲜过敏备注，临时把某道菜拆成“免海鲜”版单独派一张单，不需要事先列好所有情况。

因为我们想让每位帮厨全神贯注炒自己的菜，所以每个子任务都拥有独立的子状态。它们可能共享一份“总菜单”（全局配置），但各自的进度、结论都写在自己的小本子上，互不干扰。这样设计的好处是：任何两个分支之间不会因为抢同一支笔而打架，后续的并发执行才有安全的基础。当一个入口节点调用 send() 生成多个任务时，LangGraph 会把这些分支当作同一个父节点的多个“分身”，并发窗口会同时调度它们。

1. 子任务与并发窗口的比值：N 是待处理的子任务数，M 是允许同时执行的并发窗口（比如线程池大小）。

$$\displaystyle \frac{N}{M}$$

2. 上取整：⌈·⌉ 表示向上取整。7 个子任务、3 个并发窗口，需要 3 轮而不是 2 轮，因为最后一轮哪怕只有 1 个任务也得执行完。

$$\displaystyle \left\lceil \frac{7}{3} \right\rceil = 3$$

3. 总时间构成：轮数乘以单任务耗时得到执行时间，再加上 Map 拆分开销和 Reduce 合并开销，就得到整体耗时。

$$\displaystyle t_{\text{map}} + t_{\text{reduce}}$$

这个公式一眼就能看出并行收益来自哪里：如果 N 很大而 M 很小，时间主要花在排队轮次上；如果 M 足够覆盖 N，一轮就能跑完，剩下就是拆与合的固定开销。主厨看完各灶台的完成情况后，判断的正是这个“怎么排班最省时间”的问题。

\displaystyle T = \left\lceil \frac{N}{M} \right\rceil \times \tau + t_{\text{map}} + t_{\text{reduce}}

并行执行机制：Super-step 同步

派单结束后，后厨进入最热闹的环节：六口灶台同时开火，油锅滋滋作响。每一位帮厨都不知道旁边的人在炒什么，只专心做好手上的菜。LangGraph 的并行执行机制与此惊人地相似——多个并行分支同时处于 running 状态，每个分支只处理自己的子状态，彼此之间不需要直接通信。

但热闹背后必须有秩序。LangGraph 采用了一种叫 “super-step“（超级步）的同步模型：每一轮被唤醒的节点全部执行完毕后，框架才会推进到下一轮。你可以把它理解为后厨的”齐步走”口令——所有灶台火候到点同时起锅，不会出现某个灶台还在炒、另一位已经乱入下一轮的情况。任何依赖上一步结果的节点，都必须等这一轮所有并行分支归档之后才能被唤醒，这正是数字化后厨和真实后厨一样需要的纪律。

那各位帮厨做完菜，如何把自己的成果“登记在案”？答案是 Channel——后厨公告板。每个并行分支执行结束后，会把结果写到指定的 Channel 上。如果多个分支同时往同一个 Channel 写入，就由 Channel 的写入规则决定如何处置：有些规则是“保留最后一个值”，这相当于公告板上最后贴上的一张便利贴盖住了之前的；有些规则是“追加成列表”，像把每位厨师的作品单依次钉在公告板下面。写入规则的协调，保证了所有分支对状态的修改以可预期的方式合并。

super-step 的同步保证换来的是状态一致性：读到的每个值都代表“上一轮所有分支的最终结果”，不会出现读到一半的脏数据。正如后厨的秩序保证了每道菜都是完整出锅——虽然六口灶台同时开火，但传菜口永远只接收已经完成的菜品。

Reduce结果合并：扇入汇聚

帮厨们陆续熄火，端着成品走向传菜口。此时的主厨不再下锅炒菜，而是站在摆盘区，把一盘盘菜按宴席顺序排列、核对、装饰。这个“收拢”动作就是 LangGraph 的 Reduce 阶段：它把所有并行分支的输出汇聚到一个节点上，按照预订规则整合成最终结果。

LangGraph 里的 Reduce 有两种实现思路。第一种是“站着收”——设置一个 Reducer 节点，它等所有并行分支完成后被唤醒，拿到完整的输出列表，做排序、筛选、聚合这类重量级操作。另一种是“边收边合”——依靠 Channel 的写入规则增量合并，比如统计型 reducer 每收到一个值就累加一次，不用等全部结束就已有部分结果。选择哪种，取决于你要的是“全部到齐再算总分”还是“边收分边记录”。

这里最关键的概念是 Reducer 函数本身。它定义了如何把多条结果合并成一个：是拼接、求和、加权平均，还是取置信度最高的那条？自定义合并策略的灵活性，让并行模式不只适用于同质化任务。你可以让一个分支写 SQL 查询结果、另一个分支写向量检索结果，最后在 Reduce 阶段用加权融合把它们变成统一的推荐列表——像主厨把冷菜、热菜、甜点编排成完整的婚宴菜单。

主厨在传菜口每收一道菜就勾掉菜单上的一项，全部的菜到齐才能开席——Reduce 就是这样一道“验收关”。它保证了并行分支各自为战的成果，最终被组织成一个有序、完整的整体交付给下游节点。

并发调度控制

只有六口灶台的后厨，却接到了二十道菜的任务。主厨必须做出选择：一口气全塞给帮厨，灶台不够、案板不够、调味料也会抢破头。真正聪明的调度是“分批上灶”——每轮只推六道菜，做完一轮再上新菜。LangGraph 的并发调度控制，处理的就是这个“灶台排班”问题。

LangGraph 底层通常跑在线程池或异步执行器上，max_workers 之类的参数直接决定了并发窗口的大小。并发窗口不是越大越好：窗口太大，大量分支同时抢占内存和回调栈，反而让单任务变慢；窗口太小，任务排队时间变长，并行红利所剩无几。实用的策略是让窗口大小接近执行器的核心并发度，就像六口灶台的后厨，派七道菜不如派六道菜节奏更稳。

除了窗口大小，LangGraph 还允许你控制“分批派单”的节奏。Map 阶段可以不是一次性生成全部子任务，而是每轮生成一部分，等上一批完成后再继续派下一批——这对应 Send API 的连续调用。你可以把它想成主厨不会一次性把所有菜单都拍到帮厨面前，而是按上菜节奏一盘一盘地派。配合 recursion limit（递归深度上限），这种动态分批调度能避免“任务爆炸”：在循环或递归场景中，只要还有剩余配额就继续派单，配额耗尽则优雅终止。

调度的艺术是“有限资源做最多的事”。当你在生产环境里调并行度时，那个最合适的窗口值往往要通过压测来找——正如主厨知道六口灶台最稳，但偶尔也会根据客流量临时调整到五口或七口，核心原则是不让任何一口灶台闲太久，也不让任何一位帮厨手忙脚乱。

错误处理策略

热闹的后厨也会出事：一位帮厨火候没看住，把红烧肉烧糊了。这时候整桌宴席怎么办？是让这位帮厨重做一份，还是干脆取消这道菜，让宴席继续？LangGraph 并行执行的错误处理，本质上就是这些预案的工程化实现。

默认行为是“整桌停下”：只要任何一个并行分支抛出异常，整个 super-step 立刻终止，图进入失败状态。这听起来很残酷，但你换位思考一下——主厨无法确定其他灶台的结果是否会被这道糊菜影响，与其端上一桌东拼西凑的菜，不如让宴席停下来统一处理。不过，LangGraph 也提供了“重做这一道”的选项：给节点配置重试策略（retry policy），设置最大重试次数和退避间隔。这就像帮厨被允许再炒一次，第二次心里更有底，成功率自然更高。

还有更进阶的降级方案：在 Reduce 阶段手动捕获部分分支的结果，即使某个分支失败，也把其他成功分支的结果做合并输出。这种“部分降级”像是主厨说：“红烧肉没了，但其他菜照常上，我们用备用的梅菜扣肉顶上。”它最适合那些“部分成功仍有价值”的场景，比如多源检索时一个数据库挂了，其他数据库的结果依然可以用来回答问题。

这个公式揭示了一个反直觉的真相：并行不是“人多力量大”，而是“人多摊上事的概率也大”。十个人各端一杯水，总有一个人可能手抖。所以优秀的 Agent 工程师在设计并行图时，总会在关键分支上加重试和降级策略，避免一场婚宴因为一道菜毁于一旦。

适用场景案例

不是每道菜都适合让多位厨师同时做——佛跳墙需要文火慢炖，中途加人反而添乱。判断一个任务是否适合 LangGraph 并行节点，核心标准是看它能不能被拆成“互不依赖的平行工序”。如果子任务之间需要频繁交换中间结果，或者每一步都依赖上一步的输出，那就不该上并行，老老实实串行反而更可靠。

适合的场景长什么样？第一类是批处理数据：把一份包含一千篇文档的语料库拆成十个片段，每个片段独立做摘要，最后合并成一份总报告。这像宴席上的凉菜拼盘——每盘独立制作，最后摆在一起。第二类是多源检索：同一个查询同时发给多个知识库或搜索引擎，收集所有结果后做排序融合，能显著缩短用户等待时间。第三类是子 Agent 规划：让多个子 Agent 分别调研市场、竞品、成本，最后汇总成商业分析报告，每个子 Agent 都是独立动作、独立产出。

但也有明显的局限性。如果所有分支都要竞争同一份全局状态，频繁写入引起冲突，状态一致性就难以保证；如果任务之间存在隐形依赖——比如分支 A 需要分支 B 中途的一个数值才能继续——硬塞进并行只会得到错误结果。判断的方式一句话：能不能把任务画成“一棵树”而不是“一条链”？树可以并行修剪枝干，链只能逐节推进。掌握了这个判断标准，你就能在合适的场景里放心地开足火力。

主厨深知：哪怕后厨有三十口灶台，如果菜单设计本身必须一道菜接一道菜地炖，那并行也救不了宴席的进度。你能做的，是把可并行的部分尽量拆细、摊平，让每一口灶台都有活干，同时接受“总有些步骤快不起来”的现实。懂取舍，才是用好并行节点的最高境界。

AI 应用场景

Python 演示

# -*- coding: utf-8 -*-"""LangGraph Map-Reduce 并发执行模拟：多城市空气质量查询与汇总"""from concurrent.futures import ThreadPoolExecutor, as_completedimport time  def fetch_city_aqi(city):    time.sleep(0.3)    aqi = sum(ord(ch) for ch in city) % 100    return {"city": city, "aqi": aqi}  class CityAqiPipeline:    def __init__(self, cities, workers):        self._cities = cities        self._workers = workers        self._results = []     def map_phase(self):        with ThreadPoolExecutor(max_workers=self._workers) as pool:            futures = [pool.submit(fetch_city_aqi, c) for c in self._cities]            for fut in as_completed(futures):                self._results.append(fut.result())     def reduce_phase(self):        total = sum(item["aqi"] for item in self._results)        best = max(self._results, key=lambda item: item["aqi"])        avg = total / len(self._results)        return {"best_city": best["city"], "best_aqi": best["aqi"], "avg_aqi": avg}  if __name__ == "__main__":    cities = ["北京", "上海", "广州", "深圳", "成都", "杭州"]    pipe = CityAqiPipeline(cities, workers=3)    start = time.time()    pipe.map_phase()    summary = pipe.reduce_phase()    print("汇总结果:", summary)    print("总耗时: %.2f 秒" % (time.time() - start))

1. from concurrent.futures import ThreadPoolExecutor, as_completed引入线程池工具，相当于 LangGraph 底层的执行器。
2. with ThreadPoolExecutor(max_workers=self._workers) as pool:创建并发窗口为 _workers 的线程池，超过窗口的任务会排队等待，这就是“灶台排班”。
3. [pool.submit(fetch_city_aqi, c) for c in self._cities]对每个城市提交一个异步任务，对应 Map 阶段的 Send 派单。
4. for fut in as_completed(futures):每完成一个任务就取一次结果，对应 LangGraph 在 super-step 内持续收集分支输出。
5. self._results.append(fut.result())把每个并行分支的输出收集到列表，对应 Channel 的追加式写入规则。
6. total = sum(item["aqi"] for item in self._results)Reduce 阶段的第一部分：将所有子结果累加，对应自定义 reducer 的求和逻辑。
7. best = max(self._results, key=lambda item: item["aqi"])用 max 取最优城市，对应“取最大值”风格的自定义合并策略。
8. with ThreadPoolExecutor(...) as pool使用上下文管理器确保所有线程资源在 Map 阶段结束后自动释放，防止资源泄漏。
9. time.sleep(0.3)模拟真实网络请求的耗时，让并行效果可视化地体现为总耗时的显著下降。

常见误区

面试题