先聊一个我最近反复琢磨的事在做智能体应用的时候很多时候我们习惯性地把工作流写成“一条直线”——先做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_factorydict) 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), timeoutself.timeout ) latency asyncio.get_event_loop().time() - start_time return TaskResult( task_idtask.task_id, nametask.name, successTrue, outputoutput, latencylatency ) except asyncio.TimeoutError: latency asyncio.get_event_loop().time() - start_time return TaskResult( task_idtask.task_id, nametask.name, successFalse, errortimeout, latencylatency ) except Exception as e: latency asyncio.get_event_loop().time() - start_time return TaskResult( task_idtask.task_id, nametask.name, successFalse, errorstr(e), latencylatency ) 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_idfdoc_{idx}, namef段落{idx}摘要, promptf请对以下文本生成简洁摘要不超过100字\n\n{chunk}, params{temperature: 0.3} ) ) return tasks def aggregate_results(results: List[TaskResult]) - str: 把各段的摘要汇总为完整摘要简单版 sections [] for r in sorted(results, keylambda 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(concurrency3, timeout30) 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_exceptionsFalse的如果一个任务抛了没被捕获的异常整个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、模型名、输入输出长度、耗时、错误信息。没有这些数据你的并发参数调优和故障排查都只能靠猜到时候踩坑了连证据都找不到。如果你现在已经有了一个还不错的并行工作流跑起来下一步可以试着做“动态规划”不再固定一组并行任务而是让上一步的汇总输出自动决定下一步要并行展开哪些子任务。那种“会自我规划”的智能体才是并行工作流真正威力最大的形态。等我有更多实践心得再来和你聊。