并行工作流这种智能体开发模式，最初吸引我的不是一个理论概念，而是一个让我印象深刻的低效场景：几个子任务彼此完全没有依赖——查库存、看天气预报、比对两份合同里的条款、搜索行业新闻、算一组财务指标——但初版代码却用一个循环按顺序执行，整个流程磨了半分钟才出结果。问题不在单个动作慢，而是它们明明可以同时进行，却被硬生生排成了串行。后来我把调度方式改成并行工作流，总耗时直接掉到五六秒，改动量不大，收益却非常明显。
这篇文章就来拆解这种智能体开发模式：它到底解决了什么问题、判断该不该用的依据是什么、不用任何编排框架怎么手工实现、实战里如何设计扇出扇入，以及我在实际项目中踩过的那些坑。适合正在做智能体应用、后端服务或数据管道的开发者参考，尤其是那种想优化整体耗时，却又不知道该从哪里动刀的人。
1. 并行工作流到底在解决什么问题：一组互不相干的任务，却排队等了几十秒
1.1 先从一个最朴素的角度理解并行
假设你有十个任务，每个任务单独跑要两秒。如果它们之间没有任何依赖关系，串行执行的总耗时就是二十秒。但逻辑上没有任何理由让第二个任务非得等第一个跑完才能开始，它们完全可以同时跑。把这十个任务同时发出去，理想情况下总耗时接近两秒，而不是二十秒。并行工作流解决的就是这个“无谓等待”的问题。
打个比方：厨房里本来只有一口锅和一个厨师，四道菜只能一道一道做。如果让三个厨师同时用三口锅，只要菜与菜之间不存在“必须等上一道完成才能继续”的依赖，整桌菜的出餐时间就能大幅缩短。智能体场景也一样：要调用多个工具、请求多个接口、让模型分别处理多个片段，多数时候这些调用彼此独立，而且本质上是外部IO请求，不涉及共享内存，非常适合并发发起。
1.2 智能体开发里最常见的三种并行形态
根据我在实际项目里的观察，并行工作流在设计层面通常表现为三种形态，大家可以对照自己的场景看看属于哪一种。
第一种是并列子任务型：几个独立任务同时执行，最后汇总结果。典型例子是智能体一次性发起多个工具调用，比如查询天气、搜索新闻、读取日历，全部返回后再统一交给模型整理。这个形态最简单，理解成“多路发起、一起收口”就行。
第二种是扇出扇入型：一个任务先拆分成N个子任务，分别执行后再把结果合并。典型例子是把一个大文档按章节切块，每一块并行做摘要，最后把所有摘要合成一篇总报告。这种形态就是经典的map-reduce思维，核心在于“拆分粒度”和“合并逻辑”。
第三种是动态分支型：流程运行到某个节点时，根据当时的输入动态决定要产生几个并行分支。比如智能体先分析用户请求，判断需要查询哪些数据库，然后才对筛选出的多个数据源并行发起查询。分支数量不是预先写死的，而是运行时算出来的，灵活度最高，也最贴合智能体“边想边做”的特性。
1.3 一个容易被误解的点：并行不等于流程图上画了几条线
很多团队在流程设计器里画了分叉，就以为实现了并行，这是常见误区。并行是执行层面的“同时发生”，而不是流程图画面上的“分叉”。判断真伪的标准很简单：看总耗时。如果总耗时接近各个分支耗时的最大值，那就是真并行；如果总耗时还是各分支耗时之和，那只是视觉分叉，底层仍然是顺序调度，问题一点都没解决。
我见过不少项目在框架配置里开了分支，但由于执行引擎本身是单线程逐个调度，实际提升有限。检查方式也不复杂：把每个分支的起始时间和结束时间打出来，看它们的时间段是否重叠。只要时间段有重叠，才是真正在并行跑。
2. 该并行还是该串行：判断依据不是任务多少，而是依赖关系
2.1 核心判据永远是数据依赖
要不要并行，第一个要问的问题不是“这个任务快不快”，也不是“任务多不多”，而是“任务之间有没有数据依赖”。如果B必须拿到A的输出才能开始，那么A和B天生就只能串行；反过来，如果两者只是“看起来相关”，但数据流上没有直接传递关系，那就可以考虑并行。
我一般会先画一张数据依赖图：节点是子任务，边是数据流向。画完之后，凡是两个节点之间不存在路径相连、且它们各自依赖的前置节点都已经满足时，就可以并行。举个例子：从合同里抽取金额和从邮件里提取待办事项，这两件事之间没有数据依赖，完全可以并行；但“先扫描合同得到文本，再从文本里抽取金额”就不能并行，因为第二步的输入是第一步的输出。
这个判断过程有点像排家务：洗衣机和烤箱都在转的时候，你洗菜不冲突；但你不可能在菜还没洗好的时候就炒菜。数据依赖决定了工序的先后关系，没有依赖关系的工序才能同时做。
2.2 不适合并行的两类典型情况
知道什么时候不该用并行，比知道什么时候该用更重要。我踩过的场景里，有两类情况强烈不适合并行。
第一类是强依赖链路。典型的如智能体里的“规划→执行→验证→修正”闭环：验证必须拿到执行结果，修正必须拿到验证反馈。这种链路天然是串行循环，强行并行只会引入大量临时方案，最后还是要合并回一条链。
第二类是共享资源独占场景。多个任务同时写同一个数据库表、同一个会话上下文、同一个全局配置，很容易出现脏写和状态错乱。这类场景需要先做资源隔离才能并行。比如不同子任务写入不同的行、不同的表，或者给每条子任务分配独立的工作目录，让它们互不干扰。判断方法很简单：看这些子任务在写什么、写到哪里。只要写入点是同一个，就要警惕。
2.3 收益不是线性的：并行上限与开销估算
把串行改成并行，最怕的是盲目扩大并行度。这里有一个粗略但实用的估算思路：整体耗时约等于串行部分耗时加上并行部分耗时除以并行度。假设一个流程里80%的步骤可以并行，20%必须串行，那么理论上加速上限只有五倍左右，不管你把并行度拉到多大都突破不了这个上限。花了大力气优化后的收益，其实一开始就能算出来。
同时还要算上并行本身的开销：线程或协程的创建、子任务结果的传输、最终合并的逻辑处理，这些都会增加额外耗时。如果单个子任务只有几十毫秒，而合并逻辑本身要几百毫秒，那并行化很可能是负优化。我的经验是：单任务耗时要百毫秒级别以上，且任务数量达到一定规模，并行化带来的收益才值得投入成本。
关于并行度，还有一个来自实践的教训：如果不是经过压测，我不建议凭感觉设置过大的并发数。合理做法是从较小的并发度试起，逐步增加到错误率开始上升为止，这个值往往才是当前环境下的最优并发度。
3. 不依赖框架的最小实现：线程池和异步IO两条路线
3.1 为什么先不急着上编排框架
在智能体开发里，很多人一提到工作流就会想到某个编排框架。框架当然有用，但在引入框架之前，我建议先用最简单的代码把并行的核心机制跑通。因为所有并行工作流的底层逻辑其实都是同一件事：把N个子任务同时发出去，等结果回来后合并。这个逻辑在代码层面就是map-reduce。
用不依赖框架的最小实现，能让注意力集中在三个关键点上：如何并发发起子任务、如何有效收集结果、如何处理失败。把这三个点弄明白之后，再上框架时你会很清楚框架替你做了什么，而不是把它当黑盒。
3.2 线程池版实现：concurrent.futures 就够了
如果子任务主要是IO密集型的——比如调用接口、读取文件、请求模型API——线程池是最容易上手的方案。很多人一听到多线程就担心共享变量和锁的问题，但在并行工作流模式下，每个子任务只负责返回自己的结果，不在任务之间传递可变对象，安全性的顾虑会少很多。
from concurrent.futures import ThreadPoolExecutor, as_completed def process(item): # 子任务逻辑：这里可以是调用模型接口、查询数据库等 return {"id": item["id"], "summary": f"processed {item['id']}"} def run_parallel(items, max_workers=8): results = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: # 先把所有任务提交给线程池，让池内部自己排队调度 futures = {pool.submit(process, item): item for item in items} for future in as_completed(futures): item = futures[future] try: result = future.result() results.append(result) except Exception as exc: results.append({"id": item["id"], "error": str(exc)}) return results用as_completed而不是按提交顺序取结果，是为了让先完成的任务先被处理，这样整体收尾速度更快。另一个需要养成的好习惯是：子任务始终携带业务ID返回，后续合并时可以按ID还原顺序，而不是依赖返回顺序。
3.3 异步IO版实现：asyncio 加信号量限流
如果项目里用的接口客户端本身是异步的，那么asyncio是比线程池更轻量的实现路线。它没有线程切换的开销，适合大量轻量IO并发。不过异步代码有个问题一定要处理：如果不加限流，几百个任务同时发起，外部接口很容易被打爆。
import asyncio async def call_one(item): # 模拟一次异步IO调用 await asyncio.sleep(0.5) return item async def run_many(items, limit=8): sem = asyncio.Semaphore(limit) async def bounded(item): async with sem: return await call_one(item) results = await asyncio.gather( *(bounded(item) for item in items), return_exceptions=True ) return results这里的Semaphore(limit)说的就是同时只有limit个任务在真实执行，其余任务排队等待信号量。实操中这个值很关键，它能防止并发数瞬间冲高导致请求失败。return_exceptions=True也很重要，它的作用是让某一个子任务失败时，不会影响其他子任务继续执行。
路线怎么选，我的建议是：如果依赖库已经提供异步客户端，优先走asyncio；如果项目里大量使用的是同步SDK，线程池更省事。两者并要在同一段逻辑里混用，否则反而增加心智负担。
3.4 从单机到多机：队列加工人的基本思路
单机并行受限于CPU核数、线程数或协程数，但如果任务量再大、执行时间再长，就要考虑把并行扩展到多台机器。这个阶段最通用的设计是任务队列加工人进程模式：生产者把任务写入队列，多个工人进程从队列里拉取任务执行，跑完把结果写回结果队列或数据库。
这个模式在智能体场景里非常常见。比如有一个任务队列存放用户请求，多个消费端进程共享GPU或模型服务，去跑推理任务，再把结果汇总到结果表。并行度不再受单机限制，也方便做失败重试和断点续传。和单机最根本的区别在于，你需要把任务和结果都显式持久化，而不是保存在内存变量里。
4. 扇出扇入实战：一批合同等着被抽关键字段
4.1 场景设定与流量分析
讲完理论基础，用一个我在某合同处理项目中实际做过的例子说明完整的扇出扇入流程。假设输入是100份PDF合同，目标是从每一份合同里抽取甲方、乙方、金额、日期等关键字段，最终汇总成一张Excel表。
先看串行版：每份合同先做OCR得到文本，再让模型从文本里抽取字段，两份操作加起来大约三秒。100份合同串行跑就是五分钟，用户等得着急，优化空间非常明显。这100份合同之间没有任何数据依赖，每一份只需要读自己的源文件，写入各自独立的结果对象，天然适合并行工作流。
4.2 流程设计：扇出、并发执行、扇入汇总
整体流程分成三个阶段。扇出阶段把100份合同文件路径构造成100个独立任务；并发执行阶段用线程池控制合理并发度，每个任务内部完成“OCR+字段抽取”；扇入汇总阶段收集所有结果，按文件名排序，把成功的行和失败的行分开处理，最后导出Excel。
这个流程每个阶段边界清晰，失败重试只发生在单个合同级别，不会影响其他合同的结果。我特别建议保持这种边界设计：扇入汇总就是纯粹的数据整理，不要再混入子任务逻辑。
4.3 核心代码骨架
import time import pandas as pd from concurrent.futures import ThreadPoolExecutor CONTRACTS = [...] # 100个PDF文件路径 def extract_one(path): # 1. OCR 得到文本 text = ocr_file(path) # 2. 模型抽取字段 fields = llm_extract(text) return {"file": path, **fields} def retry_call(func, arg, max_retries=2): last_err = None for attempt in range(max_retries + 1): try: return func(arg) except Exception as exc: last_err = exc time.sleep(min(2 * attempt, 8)) return {"file": arg, "error": str(last_err)} rows = [] with ThreadPoolExecutor(max_workers=8) as pool: # 一次性提交全部任务，线程池内部排队调度 futures = [pool.submit(retry_call, extract_one, path) for path in CONTRACTS] for f in futures: rows.append(f.result()) df = pd.DataFrame(rows) df.to_excel("contracts_result.xlsx", index=False)这里有一个很多人忽略的细节：把任务一次性提交给线程池之后，再按futures的顺序收集结果。因为线程池内部会把提交上来的任务排队执行，最终返回的顺序和提交顺序一致，所以收集的时候不需要花额外精力去重排。上面代码里的retry_call包装了失败重试逻辑，单份合同偶发失败会自动重试两次，而不是一遇到报错就中断整个流程。
4.4 实测数据与并发度选择
实际跑下来，8个worker处理100份合同，一批大约在四五十秒，相比串行五分钟确实提升巨大。我试过把并发度加到16，发现耗时没有明显下降，反而偶发出现接口报错。原因是上游OCR和模型接口有限流，并发超过阈值就开始拒绝请求。
这个例子很能说明问题：智能体场景里的并行工作流，瓶颈往往不在本地计算能力，而在外部服务能承受多大的并发。并发度不是越大越好，合理值需要实测。我的习惯是先用4开跑，观察耗时和错误率，然后逐步加到8、12、16，直到错误率明显上升为止，回退到上一档就是当前比较合理的并发度。
5. 并行工作流的四个典型事故现场
5.1 共享状态灾难：全局列表被并发追加
第一个翻车场景发生在我的早期版本里。我为了图省事，让每个子任务在跑完后直接往一个全局结果列表里添加数据。跑了没几次，发现结果要么数量对不上，要么顺序诡异。排查到最终原因：多个线程在几乎同一时间向同一个列表执行追加操作，产生竞争写入，数据被覆盖或错乱。
正确做法其实已经写在最简单的代码模板里了：每个子任务只返回自己那份结果，列表追加的工作统一由收集端完成。如果确实需要多线程传递中间数据，唯一的共享通道应该用线程安全的队列，除此之外尽量不要在子任务里改动任何共享可变对象。
5.2 上游限流：并发一高反而更慢
这是个特别反直觉的坑。曾经有一批任务，串行跑耗时在正常范围，我把并发度加到50之后，外部API开始大量返回限流错误，整个流程耗时不降反而飙升到串行的两倍以上。原因很简单：并发请求超出了服务端每秒请求数或每分钟令牌数限制，大部分请求都在等待或失败重试。
这类问题的解决思路是分级治理。第一层用信号量做本地的并发控制，让整体请求速率稳定在上游允许的范围内。第二层是重试要有指数退避，不能同一时间疯狂重试。第三层是在调用前估算一下token用量或请求量，做令牌桶。最根本的还是先摸清外部服务的限制到底是多少，再据此设定并发上限。
5.3 结果顺序错乱：按完成顺序拼接导致数据错位
如果子任务本身有业务顺序，比如按章节顺序处理文档，最后合并时要按章节拼接，那么并行处理天然会打乱完成顺序。假设第5章先处理完了，第3章反而后完成，如果收集端直接按返回顺序拼接，最终文档就是乱的。
我的解决方案是给每个子任务一个明确的ID，收集时按ID重排而不是按返回顺序重排。实际编码中可以通过字典暂存每一个结果，最后按原始ID列表顺序取数据。另外，每个子任务带上trace_id也值得推荐，它能让你在日志里把同一个子任务的完整生命周期串起来，排查问题时非常方便。
5.4 部分失败的处理策略：只重跑失败项，不整体回滚
有一批任务跑到第97个时挂了，整体回滚重来显然不现实。正确思路是保证任务粒度足够小，让一个子任务失败不影响其他子任务，然后在收集阶段把失败项单独挑出来，统一做二次重试。如果二次重试仍然失败，再把这些失败项单独标记，人工介入或降级处理。
这个思路落到工程上，还需要有幂等设计。结果表里用唯一业务ID做去重，重跑时跳过已完成项，这样就天然支持断点续传。整体重跑是最后手段，大多数情况下不要轻易使用。
6. 再往前走一步：动态并行、可观测性与落地心得
6.1 动态并行：让分支数量在运行时决定
在真实智能体场景里，并行的分支数量经常不是预先确定的。智能体可能先分析用户请求，决定要调哪些工具，再把这些工具调用同时发出去。这个流程的第一步算决策节点，第二步才算执行节点。设计上的关键就是把两者分开：决策节点根据输入输出N个任务对象，执行节点再把这N个任务并行执行。
用静态流程图去穷举所有分支组合，是不现实的，因为输入变化会让组合数量爆炸。更好的方式是用代码动态生成任务列表，框架或线程池只负责调度，不要试图把所有分支都画在图上。
6.2 并行流程的可观测性：别等出错才后悔
并行流程排查问题比串行流程麻烦得多，因为多个分支同时跑，日志混在一起，出错了很难定位是哪一路的锅。我的做法很朴素但有效：每一路子任务带同一条trace_id，进入节点、离开节点、失败重试各打一条结构化日志。收集阶段汇总之后，可以依据trace_id回看某一分支的全链路记录。
这套机制在任务量小的时候看着有点多余，但一旦规模上来，它能救命。有次一个数据管道半夜报错，靠trace_id直接定位到特定分支的限流重试，十分钟内就处理完了，如果没有trace，光靠时间反推日志得折腾好一阵。
6.3 一点实际操作体会
并行工作流本质上是在用资源换时间，把互不依赖的调用同时发出去，换来整体响应时间的大幅下降。对智能体应用来说，用户最能直接感知的就是耗时，因此梳理一遍流程里那些串行步骤里是否存在互不依赖的调用，往往是性价比很高的优化点。
我每次接到新需求，都会先画数据依赖图，再决定哪些边可以拆开；并行化之后，先小规模试水位，再逐步加并发。给每个任务标上ID、做好幂等、留好trace，这套基础功夫做到位了，并行工作流会从一个看起来很酷的概念变成真正稳省时间的基础设计模式。
