先聊一个我最近反复琢磨的事：在做智能体应用的时候，很多时候我们习惯性地把工作流写成“一条直线”——先做A，再跑B，然后C，最后汇总。这个思路没问题，但它有个隐藏的代价，那就是你所有的等待时间被串行地累加了。在LLM推理延迟普遍还在秒级、甚至复杂任务奔着几十秒去的今天，串行执行一个多步骤的智能体流程，用户体验往往就是“转圈圈转了大半天”。而并行工作流（Parallelization Workflow）这个设计模式，恰恰就是为了解决这个问题而存在的。

这篇文章里，我会用自己实际踩坑后的经验，把并行工作流在智能体开发中的核心思路、三种典型形态、完整可跑的代码范例，以及关键的设计决策和常遇到的坑，一次讲清楚。不管你是刚接触智能体开发的新手，还是已经在做复杂工作流编排的老手，这篇文章都值得你花十分钟看完，至少能帮你少走几次弯路。

1. 并行工作流是什么——先把它从“概念名词”变成能用的东西

很多讲设计模式的文章喜欢上来就扔定义，我尽量换个方式：并行工作流，本质上是让你智能体应用里的多个独立环节“同时开工”，而不是“排队开工”。

说得更直白一点，传统的智能体开发里，我们会因为“流程必须有序”而被迫把很多本来没有严格依赖关系的步骤硬生生排成顺序执行。举个例子，你要做一个“竞品分析报告”的智能体，流程拆解出来可能是：先分析A产品，再分析B产品，然后对比，最后写报告。但如果A和B的分析彼此之间没有依赖关系，那为什么不让它们同时跑？并行工作流干的就是这件事。

这套模式的核心价值不只是“快”，它还能带给你两点意外收获：

一是质量的提升。同一个问题，你让一个模型推一次和让多个模型从不同角度各推一次，再把结果综合起来，前者的容错率明显更低。在智能体应用里，很多“幻觉”问题和“单点盲区”问题，用并行投票的方式能得到有效抑制。

二是资源利用率的优化。现在的LLM服务基本都是按token计费、按并发调用设计的。你串行执行的时候，算力大部分时间在闲置，你自己还在那干等着；并行执行则是在同样的等待时间内，把算力“填满”。从这个角度看，并行不是炫技，是性价比。

这个模式适合谁？基本上只要你在做智能体相关的东西，都会用到。比如你在做一个客服机器人，多个知识库检索可以并行；你在做一个内容创作助手，多个素材收集通道可以并行；你在做一个数据分析Agent，多个维度的数据抽取同样可以并行。如果你目前做的东西还需要“等上一步的结果才能做下一步”，你大概率已经遇到了性能瓶颈，这篇文章正好适合你。

2. 为什么必须并行——串行流程的三个致命瓶颈

不夸张地说，把工作流做成串行，是智能体开发里最容易犯、也最容易被忽视的“性能杀手”。我见过不少项目前期跑得挺顺，结果一上真实用户，响应时间动不动就奔着一分钟去，尺子一量，基本上都是串行调用LLM拖的。

2.1 延迟的乘法效应

大模型单次推理的延迟，根据模型大小和输入输出长度不同，大概在几百毫秒到几十秒之间波动。串行流程里，这个延迟是被“乘法”放大的：你有5个步骤，每个步骤平均需要3秒，那么理论上的最少耗时就是15秒起步。但实际情况往往更糟，因为一个步骤的模型输出可能会很长，下一轮的输入会变长，导致后续步骤的延迟同步增长，整体时间动不动就变成两倍三倍。并行化则不同，它是“取最大值”而不是“取和”——5个步骤同时跑，总耗时大概率只取决于最慢的那一个分支。

2.2 Token成本的隐性浪费

这个点很多教程不会提，但实际项目里非常疼。串行流程里，每一步的模型输入往往包含了上一步的完整输出作为上下文。这就意味着，你在第N步付出的token成本，其实打包了前面N-1步的所有输出。等你跑到最后，几十页的中间结果都在上下文里，成本早就不是“每次调用的价格”了，而是“每一代都带着祖宗一起前进”的复利式堆积。并行流程天然隔离了这个积压：分支之间的上下文彼此独立，你不会因为某一步太啰嗦，就逼着后面所有步骤都跟着“买单”。

2.3 故障传播的放大型

串行链条是“一条线上一根绳上的蚂蚱”，中间任何一步失败了，后面全都停摆。你可以加大重试机制，但重试的成本同样会在链条上被放大——如果第2步失败了，后面几步都得跟着重跑。并行化的工作流因为各分支相对独立，一个分支失败并不会拖垮全局，其他分支的结果可以先落袋，失败的那个分支单独做重试或降级处理即可。这在真实生产环境里，属于“少踩一个坑就赚回本”的收益。

3. 并行工作流的三种典型形态——先看清地图再出发

看了网上不少智能体框架的实现，也自己动手搓过几个，我倾向于把并行工作流总结成三种基本形态。理解了这三种形态，你就基本掌握了并行化设计的核心思维。

3.1 Sectioning：任务切片，分头处理

这是最直接、最常用的一种形态。它的思路是：把一个完整任务拆成多个互不依赖的子任务，然后同时执行，最后把结果按既定结构拼起来。

典型案例就是长文档分析。你有一篇几万字的文档，如果一次性塞给模型，一来容易突破上下文窗口限制，二来模型对超长输入的注意力会衰减，细节丢失严重。正确的做法是：把文档切成多个段落，每个段落作为一个独立任务，并行送去提取关键信息、打标签、生成摘要，最后把各段落结果汇总成整篇文档的结构化摘要。

这种形态对“切分质量”要求较高：你必须保证各切片之间没有强耦合关系，否则并行出来的结果拼不到一起去。我在实际项目里有个经验：不能按固定字符数硬切，那样很容易把一句话的逻辑砍断。更稳的做法是按语义边界切，比如按段落、按Markdown章节、按代码函数块来分。

3.2 Voting：同题多解，投票定案

第二种形态也比较常见。它的思路是：同一个问题，让多个模型、或多轮独立推理各自给出答案，再通过投票、打分或交叉验证的方式决定最终答案。

这个思路在现实中很像“专家组会诊”。比如你要做一道难度较高的阅读理解题，与其让一个大模型拍脑袋给你一个答案，不如同时开三个独立会话跑同一个问题，把三个结果汇集起来，如果两个人都说选B，另一个说选C，那你选B的置信度就高很多。另一个典型的场景是代码审查：一个模型负责查安全性漏洞，一个模型负责查性能隐患，另一个模型负责查代码规范。它们都在看同一段代码，但关注维度不同，最后汇总出的报告就比单个模型“一条龙”式审查全面得多。

Voting形态需要注意的是结果聚合策略：什么时候用“少数服从多数”，什么时候用“置信度加权”，什么时候干脆打回重写？这个我在后面专门讲。

3.3 Aggregation：分功协作，统一汇编

第三种形态和Sectioning容易混淆，但本质上有区别：Sectioning是“拆蛋糕”，Aggregation是“各做一道菜，拼成一桌席”。

Aggregation的思路是：为同一个目标设置多个“专业分工”的并行分支，每个分支产出不同格式或不同侧重点的内容，最后汇总成一个统一的综合性结果。举个例子，你要生成一份关于某新产品的市场分析报告，你可以同时启动三条线：一条线专门整理产品技术参数和功能列表，一条线专门搜索市场反馈和用户评价，一条线专门分析同类竞品的定价和定位。三条线结果汇总后，再由一个汇总器把它们串成一份完整的报告。

这里最讲究的是“汇总器”的设计，它需要能够读懂各分支产出的差异，并且巧妙的消解彼此重叠的部分，而不是机械地拼一段文字上去。我在项目中常采用的做法是：除了让各分支产出内容本身，还要让它们附带上结构化的元信息，比如标题、要点列表、关键词，汇总器根据这些元信息来做编排，效果会好很多。

4. 从零落地：一个并行工作流的最小可运行实现

理论讲再多，不落地都是空谈。下面我给出一个我自己在项目里经常用的基础代码骨架，用的是Python和asyncio。你完全可以复制过去，根据自己用的LLM SDK替换一下核心调用即可。

4.1 整体架构与关键依赖

我建议的架构是“任务拆解器 + 并行执行器 + 结果聚合器”。

• 任务拆解器：把用户请求拆成一串独立子任务，每个子任务自带参数、提示词和输出约定。
• 并行执行器：负责并发调用LLM，管理并发上限、超时和重试。
• 结果聚合器：把各个分支的输出整理成最终结构。

我常用的依赖很简单：asyncio负责并发，json负责结构化传参，HTTP库就用你LLM服务商提供的异步客户端，正常的异步SDK都行。

4.2 完整代码示例

下面这段代码模拟的是“多文档并行摘要”的场景，你把它稍微改改就能扩展到其他任务。

import asyncio import json from dataclasses import dataclass, field from typing import Dict, List, Optional # ---------- 1. 基础数据结构 ---------- @dataclass class ParallelTask: """单个并行子任务""" task_id: str name: str prompt: str params: Dict = field(default_factory=dict) @dataclass class TaskResult: """单个子任务的执行结果""" task_id: str name: str success: bool output: Optional[str] = None error: Optional[str] = None latency: float = 0.0 # ---------- 2. 模拟异步LLM调用 ---------- # 正常开发时把这里替换成你所用大模型平台的异步SDK调用即可 async def call_llm(prompt: str, **kwargs) -> str: # 模拟网络延迟 await asyncio.sleep(1.0) # 这里只是一个模拟返回 return f"模拟结果: {prompt[:20]} ..." # ---------- 3. 并行执行器 ---------- class ParallelExecutor: def __init__(self, concurrency: int = 3, timeout: int = 30): self.semaphore = asyncio.Semaphore(concurrency) # 控制最大并发数 self.timeout = timeout async def run_single(self, task: ParallelTask) -> TaskResult: start_time = asyncio.get_event_loop().time() async with self.semaphore: try: output = await asyncio.wait_for( call_llm(task.prompt, **task.params), timeout=self.timeout ) latency = asyncio.get_event_loop().time() - start_time return TaskResult( task_id=task.task_id, name=task.name, success=True, output=output, latency=latency ) except asyncio.TimeoutError: latency = asyncio.get_event_loop().time() - start_time return TaskResult( task_id=task.task_id, name=task.name, success=False, error="timeout", latency=latency ) except Exception as e: latency = asyncio.get_event_loop().time() - start_time return TaskResult( task_id=task.task_id, name=task.name, success=False, error=str(e), latency=latency ) async def execute(self, tasks: List[ParallelTask]) -> List[TaskResult]: """并发跑完所有任务，返回全部结果""" results = await asyncio.gather( *[self.run_single(task) for task in tasks] ) return list(results) # ---------- 4. 任务拆解器和聚合器 ---------- def build_tasks(doc_chunks: List[str]) -> List[ParallelTask]: """把长文档按章节切分后构建为并行子任务""" tasks = [] for idx, chunk in enumerate(doc_chunks): tasks.append( ParallelTask( task_id=f"doc_{idx}", name=f"段落{idx}摘要", prompt=f"请对以下文本生成简洁摘要，不超过100字：\n\n{chunk}", params={"temperature": 0.3} ) ) return tasks def aggregate_results(results: List[TaskResult]) -> str: """把各段的摘要汇总为完整摘要（简单版）""" sections = [] for r in sorted(results, key=lambda x: x.task_id): if r.success: sections.append(f"[{r.name}] {r.output}") else: sections.append(f"[{r.name}] 处理失败: {r.error}") return "\n\n".join(sections) # ---------- 5. 主流程 ---------- async def main(): # 这里代表你从某个长文档里切出来的多个段落 doc_chunks = [ "第一章：项目背景与目标……", "第二章：技术方案选型……", "第三章：实施计划与排期……", ] tasks = build_tasks(doc_chunks) executor = ParallelExecutor(concurrency=3, timeout=30) results = await executor.execute(tasks) final_report = aggregate_results(results) print("=== 各子任务耗时 ===") for r in results: print(f"{r.name}: 成功={r.success}, 耗时={r.latency:.2f}s, 错误={r.error}") print("\n=== 最终汇总结果 ===") print(final_report) if __name__ == "__main__": asyncio.run(main())

4.3 这个骨架的几个关键细节

很多人第一次写并发调用LLM的代码，会忽略信号量Semaphore的用法。没有它，你的代码确实也能并行，但并发数会完全失控——如果你的任务列表有50项，你就一瞬间打出50个并发请求，LLM平台那边不给你返回一堆429限流错误才怪。用asyncio.Semaphore包一下，等于给并发池装了水龙头，想放多少水你说了算。

asyncio.wait_for也是必记的用法。LLM调用经常因为网络抖动或服务端过载而卡住，如果不加超时控制，你的整个gather会一直傻等，用户体验直接崩掉。给它一个超时上限，到点就放弃，配合失败重试或降级策略，稳定的多。

有一点你得特别注意：asyncio.gather默认是有return_exceptions=False的，如果一个任务抛了没被捕获的异常，整个gather会跟着抛异常。所以我习惯在每个子任务内部就把异常吞掉，转成TaskResult里的error字段，保证一个分支挂了不影响全局。这在并行工作流里是个非常重要的代码习惯。

4.4 更进一步的建议：结构化输出

在上面的骨架里，每个子任务返回的是“纯文本摘要”。在实际业务中，我强烈建议你用结构化JSON模式来约束子任务的输出。比如让每个分支返回：

{ "title": "段落标题", "keywords": ["关键词1", "关键词2"], "summary": "摘要内容", "sentiment": "positive" }

这样聚合器拿到的就不是一段难以解析的散文，而是干净的数据。你可以用这些结构化字段做排序、过滤、甚至后续路由。这个习惯一旦养成，你的聚合器代码会好写十倍。具体的做法就是，在子任务提示词里明确加上“请只输出JSON，不要输出其他内容”，并在你的LLM调用里开启JSON模式（很多主流平台都支持这种方式）。

5. 关键设计决策：粒度、并发数与聚合策略怎么定

学会了跑通一个demo，接下来真正让你头疼的往往是：这个东西放到真实业务里，参数怎么调？任务怎么拆？并发开多大才稳？下面这几个决策点是我在项目里反复校准过的，每个背后都有真实的教训。

5.1 并行任务的粒度——拆到什么程度才是最合适

切得太粗，每个分支的任务还是太重，单次LLM调用的延迟依然很高，并行收益有限；切得太细，又会出现管理开销过大、分支之间结果碎片化严重、聚合困难的问题。我自己的经验法则是：让每个子任务的单次LLM调用控制在3000~5000 token以内的输入输出规模。这样既不会触碰上下文窗口的天花板，也不会让单次调用延迟长得离谱。

另一个判断标准是：如果一个子任务需要“看到”另一个子任务的结果才能继续做，那它们就不适合并行，需要重新考虑拆法。我在项目里吃过一次亏：把一个“先总结再翻译”的流程硬拆成了两个并行分支，结果翻译分支因为拿不到总结内容，硬着头皮翻译了全文，最后聚合阶段只能把翻译结果又重新总结了一遍，白白多花了一倍成本。从此以后，我每次拆任务前都会先画一张“依赖关系图”，只有那些没有箭头相连的节点才允许分到同一批并行。

5.2 并发数的选择——不是越大越好

很多新手以为并发数开得越大越好，其实这是一个严重的误区。实际项目里，LLM平台都有配额限制，你并发开得再大，超过平台的RPM（每分钟请求数）或TPM（每分钟token数）配额，照样被限流。而且过高的并发还会导致服务端响应变慢，单次调用延迟反而上升，得不偿失。

比较稳的做法是：先查清楚你的API账号配额的30%~50%，把它作为并发上限。比如你的配额是每分钟600次请求，那并发数设置在20~30左右就够了，千万不要一上来就开100。还要注意平台常常另有单账号的并发连接数限制，这个数是无法通过调代码绕过的。

如果你做的是对延迟敏感的用户端应用，建议给并发数设置一个动态调整逻辑：初始阶段用小一点的并发值，根据最近几批任务的P95延迟动态调低或调高，维持延迟在一个可接受的区间内。

5.3 聚合策略怎么选——不是无脑“少数服从多数”

Voting形态的聚合策略要看你做的是什么类型任务。对于有标准答案的客观题（比如分类、实体抽取、判断题），“少数服从多数”是合理的，简单粗暴。但对于需要创造性的任务（比如写文案、写代码、设计方案），少数服从多数往往是灾难——因为大部分人写得都平庸，少数人的灵光一现反而被淹死了。

我在实践中通常用这三种策略：

• 一致性优先策略：多个分支给出的答案高度一致时直接采用；如果分歧较大，不做投票，而是把各分支结果都作为背景材料，让一个“仲裁者”模型重新综合决定。
• 置信度加权策略：让每个分支不仅要给出答案，还要给出置信度评分，加权求和后取最高者。这个对提示词设计要求较高，但对主观性任务非常有效。
• 质量门控策略：先让多个分支各自产出，再由聚合器或一个专门的“评审模型”对每个候选结果打分排序，选最高分输出。这相当于在并行之后又串了一个评审环节，效果最好，但成本也最高。

6. 踩坑实录：并行工作流的典型问题与排查清单

这部分我整理了几类实践中出现频率极高的问题，每一类都是真实踩过的坑，希望你看完能直接避开。

6.1 限流与超时——并发一上去就连环报错

现象：并发调到10以上，开始出现大量429或503错误，重试之后又是连环超时。

排查思路：第一，先确认你的并发数没有超过账号配额。第二，检查是否每个子任务都在争抢同一个模型名导致热点问题（可以适当冗余选择模型）。第三，确认你的网络出口带宽和HTTP连接池大小是否匹配并发数。

解决建议：在调用层实现“指数退避重试机制”，比如第一次失败等1秒重试，第二次等2秒，第三次等4秒。最多重试3次就放弃。千万不要用固定间隔的重试，那会让限流更严重。另外，在代码里实现“令牌桶”或简单的“每分钟请求计数”，提前预防限流，比触发了再去应对更稳。

6.2 部分分支失败——聚合结果缺胳膊少腿

现象：12个并行任务里有2个失败了，最终报告里那2个部分是空的，整体质量明显下降。

排查思路：看失败任务的错误类型分布，是超时、限流、还是内容审核拦截？如果是内容审核拦截，问题往往出在提示词或输出内容本身。如果是超时，大概率是单任务太复杂了，可以考虑拆小一点。

解决建议：对失败分支不要直接放弃，而是做“降级处理”。常见的降级方案有：一是把失败分支的任务交给一个更小更快的模型去跑，牺牲一点质量保证流程完整；二是尝试合并到相邻的成功分支里去处理；三是在最终结果里主动标注“部分数据缺失”，而不是假装完整。诚信输出比强行补全要靠谱得多。

6.3 聚合结果逻辑错乱——各分支质量都很好但拼不出好文章

现象：每个分支的单点输出质量都不错，但聚合器拼接后的结果读起来逻辑不通，重复内容多，衔接生硬。

排查思路：问题通常出在任务拆解时段落边界切得不干净，或者各分支的提示词没有统一输出规范，导致它们的内容在颗粒度和表达风格上差异很大。

解决建议：一是统一各分支的输出模板，比如都要求“先给结论，再给论据，最后引用来源”。二是在聚合器里增加“去重”逻辑，可以用字符串相似度或向量相似度来识别重复段落。三是最关键的一招：不要让聚合器一次性把所有分支结果生硬拼接，而是给它一段“组装指导”——说明哪些分支结果需要放在开头、哪些放中间、哪些是用来补充细节的，聚合器才能编排得更像人写的。

6.4 上下文长度不均衡——有些分支超长，有些分支超短

现象：同时跑10个任务，有的分支输出5000字，有的只输出50字，聚合阶段无法统一处理。

排查思路：常见原因是各分支的提示词约束力度不一致，或输入素材信息密度差异太大。

解决建议：在任务拆解阶段就给每个分支设定明确的输出长度上限，比如“最多输出300字”。在聚合器里，如果发现某分支输出严重偏离预期长度，可以考虑二次摘要或直接丢弃异常值。另外，很多LLM服务支持在请求参数里设置max_tokens，这是硬约束，比提示词软约束可靠得多。

7. 并行工作流的典型应用场景——照着用就行

理论说完了，如果你还在纠结“我这个项目到底要不要用并行”，我列几个我实际做过的应用场景，你对照着看。

7.1 长文档知识库问答

这是并行工作流用得最猛的一个场景。做法就是把知识库文档离线切成块，用户提问时，不是把所有文档都塞给模型，而是先通过检索挑出Top K相关片段，再并行让模型分别基于每个片段回答问题，最后聚合。这比让模型在一大堆资料里盲目找答案要准得多。特别是当检索结果来自完全不同的文档主题时，并行能让每个模型都聚焦在“自己的一亩三分地”上，不被无关信息干扰。

7.2 多维度代码审查

给一个代码仓库做审查，传统做法是把整个仓库丢给一个模型让它“全面审查”，结果往往什么细节都发现不了，得到的全是大而空的话。正确做法是先拆出多个维度：安全审查、性能审查、可读性审查、测试覆盖审查、依赖安全审查等等，每个维度一个独立分支，并行跑。每个分支的提示词都按维度深度定制，最后合并成一份像模像样的Code Review报告。我试过用这个方法，单个维度发现的问题数量和质量都远高于“一把梭”式的单次审查。

7.3 多方案生成与比选

做产品设计或技术选型时，需要多个方案。你不妨一次性开多个并行的“方案生成器”，每个分支指定不同的方向参数，比如“低成本方案”“高性能方案”“易维护方案”，各分支独立生成完整方案后，再由一个评审模型统一打分比选。这样产出的候选方案差异性很大，不会出现“同一个模型换几个词换个格式就是一版新方案”的情况。这个场景我用过几次，效果非常惊艳，特别是能让评审模型对方案间的取舍做深入分析的时候，站出来说“方案A成本最低但后期维护成本高、方案B前期贵但长期更划算”这种有洞察力的结论。

7.4 内容创作的多角度素材收集

做一篇深度长文或一份行业研究报告，选题确定后，往往需要收集多个角度的素材。以前的做法是写一个脚本串行地搜资料、整理、写长文。现在的做法是并行开“行业趋势分析”“竞争格局梳理”“技术路径对比”“用户案例搜集”等几条线，每条线在前面说的Aggregation模式里有一个独立的分工。因为不同分支的提示词差异很大，模型在同一时间产生不同维度的素材，步调一致，速度也更稳。

8. 最后再分享一点个人的实际体会

做并行工作流有一段时间了，我最大的感受是：并行不是银弹，但它是一把值得随身携带的好锤子。

它解决的是“等待时间”的浪费，但并没有解决“思考质量”的问题。如果你的任务本身就有强依赖链条，即使硬切并行，也只是把“串行等待”变成了“并行等待+更复杂的聚合纠错”，成本反而更高。所以，拿到一个需求，先画依赖关系图，再判断哪些环节值得并行，这比无脑上并发重要得多。

另外，生产环境里我建议你给并行工作流加一个“总开关”，做成配置项：workflow.parallel_enabled = true/false。因为有时候你切到某个第三方模型，它的并发能力或成本模型跟之前的平台差异很大，允许一键退回串行模式，会成为你在每次上线前压测的一个保底选项。还有，日志里一定要记录每个分支的task_id、模型名、输入输出长度、耗时、错误信息。没有这些数据，你的并发参数调优和故障排查都只能靠猜，到时候踩坑了连证据都找不到。

如果你现在已经有了一个还不错的并行工作流跑起来，下一步可以试着做“动态规划”：不再固定一组并行任务，而是让上一步的汇总输出自动决定下一步要并行展开哪些子任务。那种“会自我规划”的智能体，才是并行工作流真正威力最大的形态。等我有更多实践心得，再来和你聊。