RunnableBatch 实战：大模型批量推理并发优化与跨供应商实践-北京尧图网络

如果你的业务里有大批文本需要让大模型处理比如客服工单分类、商品评论打标、知识库入库前的清洗你最初的实现大概率是写一个 for 循环一条一条调用模型接口。功能没问题但跑起来很痛苦处理 100 条就是 100 次网络往返平均单条约 2 秒全程要 3 分多钟这还不算供应商偶尔限流、超时、重试带来的额外等待。我在做模拟项目X的批量推理Batch Inference优化时把重心从“调接口”挪到“利用 RunnableBatch 把请求合并成可并发的批”上之后同样的任务从 3 分多钟压缩到了 40 秒以内。这篇文章不讲虚的直接拆解 RunnableBatch 的工作机制再给出一个可复跑的跨模型供应商并行请求合并示例最后把我踩过的限流、乱序、超时、部分失败这些坑一并列出来。如果你正在做 LLM 应用的批处理模块或者你还没听过 RunnableBatch、不确定它和 asyncio.gather 有什么区别这篇文章适合你。1. 从“每条请求干等一个模型响应”说起批处理的真实成本1.1 批处理场景里的三种常见形态先说清楚什么样的问题算“批处理”。我总结下来大部分业务都逃不过三种形态。第一种是文本标注。你有一批评论、工单、政策文档希望模型给每条文本打上类别标签比如“退款投诉”“物流咨询”“产品缺陷”。这类任务输入独立、输出简单是最典型的可批量场景。第二种是信息抽取。给一段非结构化文本提取出结构化字段比如合同里的金额、日期、甲方乙方。这类任务通常还要求输出 JSON 结构逻辑更复杂但一批文本之间完全互不影响。第三种是内容改写或翻译。把一批文档摘要统一改成某种风格或者翻译成目标语言。这三种任务有一个共同特征每一条输入都需要一次独立的大模型推理但条与条之间没有依赖关系。既然没有依赖理论上就可以并行而并行正是 RunnableBatch 能够带来收益的前提。1.2 串行调用的隐性开销延迟墙、限流墙和时间碎片很多人对“串行慢”的理解停留在“时间等于一条乘以条数”但实际损失不止这么简单。我把串行调用模型的成本拆成三层。第一层叫延迟墙。大模型推理是个相对漫长的过程单条请求通常会花 1 到 5 秒这还取决于输出长度和供应商侧排队情况。你发一条请求大部分时间都花在等待模型生成上但 for 循环让这些等待时间完全串起来无法重叠。第二层叫限流墙。供应商虽然限流但限流不等于串行。很多供应商允许一定程度的并发。你用 for 循环等于主动放弃并发额度白白浪费了接口能力。而一旦某个请求被限流你又得做退避重试整体耗时进一步拉长。第三层叫时间碎片。每次请求都有连接建立、鉴权、路由、排队的过程。单条请求时这些开销不太明显但 100 条请求累加起来这部分时间碎片非常可观。串行方案等于把每条请求的路由和排队时间首尾相接全都变成了等待时间。用一张表来感受一下差距。假设平均单条响应 2 秒处理方式30 条60 条120 条纯串行 for 循环约 60 秒以上约 120 秒以上约 240 秒以上最大并发 10 的批处理约 8 秒起约 14 秒起约 26 秒起当然实际耗时还受供应商并发上限和网络波动影响但趋势很明显一旦数据量上了几十条串行方案的时间成本就完全不可接受了。1.3 异步未必是答案为什么需要“有顺序语义的并发”很多人第一反应是既然 for 循环慢那我用 asyncio.gather 并发调用不就行了思路没错但自己手写并发会引入一堆额外问题。你自己写并发至少要解决三件事控制最大并发数否则几十个请求一起涌向供应商立刻触发限流收集结果并按输入顺序对齐否则 30 条结果返回后你根本不知道哪条对应哪条处理单个请求异常时不能影响整个批次。一个经验丰富的开发者能把这三件事写对但代码量会显著膨胀而且每个供应商的接入方式都不一样你还要为每家单独维护一套并发逻辑。RunnableBatch 的价值在于它把“并发执行顺序对齐异常处理”封装成一个统一的接口。你只需要面对一个批量方法告诉它你要处理哪批输入它负责调度最后按原顺序把结果列表还给你。这就是它和手写异步最大的不同——不是能不能并发而是并发时能不能不操心一堆边缘细节。2. RunnableBatch 的工作原理一次调用瞬间拆成 N 个并行子任务2.1 先理解 Runnable 的统一约定要搞懂 RunnableBatch先得理解 Runnable 到底是个什么概念。在目前流行的 LLM 编排框架里Runnable 是一种统一抽象。任何对象只要实现了这个接口就具备相同的调用方式传一个输入进去得到一个输出。模型调用是 Runnable提示词模板是 Runnable输出解析器也是 Runnable。更妙的是多个 Runnable 按顺序组合成的完整链路本身仍然是一个 Runnable。这意味着 batch 调用并不只适用于“单个模型”。它可以作用于一整条处理链路。以前你写批量文本分类可能是先对每条文本做模板拼接再调模型再解析结果用了 RunnableBatch 之后你可以先把“模板模型解析器”打包成一个完整链路然后对这组链路批量调用。这个设计比 for 循环高的地方在于调用方只依赖统一接口不关心对象内部是 HTTP 调用、本地函数还是更复杂的组合链路。想要替换供应商、调整处理环节只需要替换这个 Runnable 实现。2.2 batch 内部发生了三件事当你调起一个 RunnableBatch 时框架内部并不是魔法般地让供应商合并多条数据而是替你做了三件事。第一接收一个列表输入并按输入顺序为每个元素生成一个独立的子任务。它不会改变你的数据顺序每个子任务都知道自己是第几位。第二根据你传入的配置比如最大并发数把这些子任务交给内部的执行器调度。这一步相当于自己实现 asyncio 时的信号量控制但框架帮你做了。你只负责声明“最多同时跑多少个”它会在执行器里控制实际并发量。第三收集所有子任务返回结果按输入时的位置重新拼装成列表返回给你。不管哪个子任务先完成最终列表顺序永远和输入列表顺序一致。我做一个小类比。你是一家餐厅的点菜员以前你每接到一桌客人的菜单就亲自跑一趟后厨高峰期后厨门口挤成一团。RunnableBatch 做的事情是你同时收集 20 桌客人的菜单一起交到后厨分单台后厨师傅按灶台数量并行做菜出菜时再按桌号归位。对点菜员来说他不需要知道哪个灶台做哪道菜只关心最后端上来的菜对应哪一桌。2.3 跨供应商合并的适配策略输入统一、输出归一、错误隔离跨模型供应商合并是这篇文章的重点。你可能会问单个模型可以 batch不同供应商的模型怎么合在一起关键就在于把不同供应商的调用都适配成 Runnable。框架不可能知道某商业模型 A 和某开源本地模型 B 的差异但只要你把两者的调用都封装成同类对象就可以放进同一个编排流程里。实际项目中我通常分三步做适配。第一步是统一输入。模型 A 和模型 B 可能希望不同的消息格式我在调用前把同一份原始输入转换成各自的输入格式让下游链路无感知。第二步是输出归一。模型 A 返回 JSON模型 B 返回一段带格式化标记的文本两者不能直接混用。我会为每个模型接一个自己的解析器把不同风格的输出统一成同一个结构体。这样即使下个月换了新的供应商也只需要新增一个适配器。第三步是错误隔离。一个供应商超时不应该拖垮另一个供应商的结果。用 RunnableBatch 的异常返回机制把失败的单条请求隔离出来而不是让整个批次崩掉。后面第 4 章会详细讲这个机制。3. 可复跑的跨供应商批量合并示例30 条输入 2 个模型供应商3.1 场景设定与选型理由下面给一个可以直接复用的示例场景。假设我在做一个模拟项目X它是某电商平台的评论情感分析服务。上游每天会推过来一批待分析评论我的任务是把每条评论分类成 positive、neutral、negative并给出置信度。为了平衡成本和稳定性我同时接了两家供应商供应商 A闭源商业模型质量高输出稳定但单价贵。供应商 B开源本地模型便宜响应快但输出格式偶尔不稳定。我选的策略是前 20 条交给供应商 A后 10 条交给供应商 B。这样既能验证混跑效果也方便在 A 涨价或限流时把流量切换到 B。为什么这样选型因为我不希望业务代码里去写“A 用哪种 HTTP 库、B 用哪种 SDK”。我要的是一个统一入口输入 30 条文本得到 30 个结构化结果每条结果都带有 provider 字段标明出处。RunnableBatch 正好适合这个需求。3.2 核心实现从单供应商 batch 到多供应商组合先定义统一返回结构。from pydantic import BaseModel class SentimentResult(BaseModel): text: str sentiment: str # positive / neutral / negative confidence: float provider: str # 是哪家供应商返回的然后分别封装两家供应商的调用。这里用伪代码表达核心逻辑实际接入时要替换成对应供应商 SDK。from orchestration_framework import RunnableLambda def call_provider_a(text: str) - dict: # 此处是供应商 A 的具体调用逻辑 # 返回原始响应可能是一段 JSON 文本 return {raw: A 的响应, provider: A} def call_provider_b(text: str) - dict: # 此处是供应商 B 的具体调用逻辑 # 它的输出格式可能和 A 完全不同 return {raw: B 的响应, provider: B} def parse_result(raw: dict) - SentimentResult: # 把两家供应商的原始输出统一解析成 SentimentResult # 真实项目里这里应使用结构化输出解析而不是手写字符串匹配 ... return SentimentResult(text..., sentiment..., confidence0.9, providerraw[provider]) provider_a_chain RunnableLambda(call_provider_a) | RunnableLambda(parse_result) provider_b_chain RunnableLambda(call_provider_b) | RunnableLambda(parse_result)接下来是关键。我们不把 30 条数据混在一个 batch 里发给同一个 Runnable因为两家供应商是不同的 Runnable。正确做法是先把输入按规则分组各自调用 batch再把结果按原顺序合并。def run_mixed_batch(items, split20): a_inputs items[:split] b_inputs items[split:] a_results provider_a_chain.batch( a_inputs, config{max_concurrency: 5} ) b_results provider_b_chain.batch( b_inputs, config{max_concurrency: 3} ) # 关键不要直接 a_results b_results # 如果你后续需要知道每条结果原先后面的输入是谁这里要带索引拼接 return a_results b_results上面这个写法已经能跑通但在更严肃的场景里我建议全程携带原始索引。def run_mixed_batch_with_index(items, split20): indexed list(enumerate(items)) # [(0, text0), (1, text1), ...] a_inputs indexed[:split] b_inputs indexed[split:] a_results provider_a_chain.batch( [x[1] for x in a_inputs], config{max_concurrency: 5} ) b_results provider_b_chain.batch( [x[1] for x in b_inputs], config{max_concurrency: 3} ) ordered [(idx, result) for (idx, _), result in zip(a_inputs, a_results)] ordered [(idx, result) for (idx, _), result in zip(b_inputs, b_results)] return [result for _, result in sorted(ordered, keylambda item: item[0])]这段代码看起来多一些但它解决了一个实际痛点当两条结果的内容完全相同时你能确认它们到底对应哪条原始输入。排序后的列表顺序一定与输入顺序一致。3.3 为什么该用 map 而不是自己写 for 循环看到这里你可能会想这看起来也没比 for 循环高端多少啊我把 batch 换成循环不也一样区别在调度策略。for 循环是同步阻塞的下一条请求必须等上一条结束。batch 则是把整个列表交给框架执行器执行器内部按你设置的并发数并行调度。一个是排队打饭一个是多个窗口同时出餐。如果你熟悉 Python 并发可能会说“我用 asyncio.gather 也能做到”。确实能但你对比一下两种方案的样板代码量关注点手写 asyncio.gatherRunnableBatch控制最大并发数自己写信号量配置 max_concurrency结果按输入顺序对齐自己收集后排序默认按输入顺序返回单条异常不影响整体自己捕获后聚合return_exceptions 参数多家供应商统一处理每家写一套适配封装成 Runnable 后统一调用我在很多项目里见过手写并发最头疼的问题并发上去了结果回来一乱排查半天发现是 as_completed 改变了返回顺序。RunnableBatch 把这类问题拦在了接口层。3.4 小批量的收益单条流式处理没有的乘法效应可能有读者会问我就 10 条数据量batch 有意义吗有但没有大数据量那么夸张。10 条数据如果每条约 2 秒串行要 20 秒。设置最大并发 5理论耗时降到 4 秒左右。这里面省掉的不仅是模型响应时间还有每条请求各自经历的连接建立、鉴权、路由时间。10 条可能省下 10 多秒对交互式任务已经感知很明显。更关键的是batch 带来的收益在你任务量增长时是“乘法效应”。30 条、300 条、3000 条只要并发配置合理耗时增长远小于线性增长。这就是为什么我建议哪怕你现在数据量不大也先把调用层改成 batch 模式后面量起来不用重构。4. 上线前必须填的五个坑限流、超时、重试、部分失败与结果对齐4.1 供应商限流模型与最大并发数的换算把请求从串行改成批量并发之后最容易踩的第一个坑就是并发一高被供应商限流。千万不要以为 max_concurrency 设得越大越好。每家供应商的限流维度不一样常见的有限制每分钟请求数、限制每分钟 token 数、限制最大并发连接数。你可能前 30 秒跑得很欢然后突然全部报 429因为整分钟的请求配额已经用完了。我在实际项目中是这样做的。先看供应商文档里写的是哪类限制。如果是每分钟请求数限制假设限制值是 600 RPM也就是每秒最多 10 个请求。那么一个批次 60 条数据即使 max_concurrency 设 20 也没意义因为每秒并发量按 10 控制更稳妥。更谨慎的做法是把请求摊开每批次之间加一个短暂 sleep。限流维度限制值示例对 batch 配置的影响每分钟请求数600 RPM单批条数 ÷ 预计耗时 ≤ 10 请求/秒每分钟 token 数100k TPM输入越长并发越要保守最大并发连接数50max_concurrency 不应超过 50对 Token 限制我要多说一句。同样是 30 条数据每条评论 100 个词和每条文档 3000 个词消耗的 token 完全不同。只算请求数是很容易翻车的建议把输入长度也纳入并发估算。4.2 超时与重试的正确姿势batch 普及后很多人以为超时配置只是“设一个总超时时间”。这理解是有问题的。如果整个 batch 设了一个 60 秒总超时结果某条请求卡了 55 秒才返回整个 batch 都要等它。更麻烦的是如果网络抖动导致其中一条连接挂起你没有单请求超时控制整批结果就悬空了。我的建议是分层配置单条请求的超时时间要短比如 30 到 60 秒这样某条请求卡住时它能独立超时batch 总的时间预算要更宽松比如 90 秒重试采用指数退避第一次失败等 1 秒第二次等 2 秒第三次等 4 秒不要在同一批次里对失败请求无限重试。另外需要注意并不是所有框架的 batch 都内置重试机制。如果框架没提供就在封装 Runnable 的内部处理重试而不是在 batch 外面套循环重试整个批次。因为重试整个批次会把本来已经成功的请求再发一遍造成重复消耗。4.3 部分失败的处理策略在真实业务里一个批次里某几条请求失败是常态不是异常。所以你要提前想清楚一批 30 条成功 27 条失败 3 条到底怎么处理如果你的 batch 调用默认抛异常那么其中一条失败可能导致你能拿到的结果全部丢失。这是最危险的错误姿势。更稳妥的方式是利用框架提供的异常返回参数让失败项变成异常对象放进结果列表。results chain.batch( all_inputs, config{max_concurrency: 8}, return_exceptionsTrue # 单条失败不会中断整个批次 ) success_items [] failed_items [] for input_text, result in zip(all_inputs, results): if isinstance(result, Exception): failed_items.append((input_text, result)) else: success_items.append((input_text, result))拿到 failed_items 之后根据业务决定是重试、落死信队列还是人工审核。我在模拟项目X里就是失败条目重新进队列延迟 5 秒后小批量重试。4.4 结果对齐与顺序保证RunnableBatch 的一个明确语义是返回列表顺序与输入列表顺序一致。这个特性在单体 batch 里基本不会出错但在跨组拼接时容易出问题。比如我们把 30 条拆成前 20 条给供应商 A、后 10 条给供应商 B两组 batch 内部顺序都没问题但在拼接时刻如果你直接用 a_result b_result结果仍然是前 20 条对前 20 条后 10 条对后 10 条整体顺序也正确。但一旦你以后想按“文本特征”动态拆分而不是按固定下标拆分简单拼接就会错位。我建议从一开始就带上原始索引或者至少让每条结果携带一个唯一的 request_id。这样下游无论怎么合并、排序都能找到原始对应关系。这个习惯看起来多余但在系统规模变大以后能帮你省掉大量排查时间。顺序对齐不是“不会出问题”而是“出了问题你很难发现”因为数据内容正确时你不会去怀疑对应关系。5. 实测对比与问题排查性能提升多少问题出在哪5.1 一组不严谨但真实可复现的压测数字先声明以下数字不是厂商基准只是我在本地环境和某测试项目里跑出来的趋势参考不同供应商、不同输入长度会有明显差异。我的测试条件是30 条商品评论平均每条 80 个汉字输入加提示词大约 500 token输出固定 3 个词加一个置信度。测试了两种链路纯串行 for 循环以及 RunnableBatch 混跑两家供应商。数据量串行方案Batch 方案按供应商拆分提升幅度30 条约 58 秒约 9 秒约 6.5 倍60 条约 118 秒约 16 秒约 7.4 倍120 条约 240 秒约 29 秒约 8.3 倍Batch 方案里还额外设置了连接复用否则提升幅度会打折。别把这几组数字当绝对标准重点看增长趋势串行方案基本线性增长batch 方案因为并发摊薄了延迟墙增长曲线明显平缓。5.2 现象一返回结果顺序错乱这是我自己最早踩过的坑也是很多人刚尝试 batch 时最容易懵的问题。现象是结果能拿到但和输入对不上号。排查链路是这样的。先确认是不是混用了异步收集方式。如果你在 RunnableBatch 内部自己包了一层 asyncio.gather并且用了 as_completed返回顺序就会变成完成顺序这属于“多次并发”叠加导致的错误。正确做法是去掉自定义异步逻辑完全交给 batch 去调度。第二步检查是不是跨组拼接导致的。比如你先把前 20 条发给供应商 A后 10 条发给供应商 B两组分别 batch最后按“先 A 后 B 的顺序拼”。如果拆分逻辑不是按下标而是按某种规则动态分组拼接时必须按原始索引排序。第三步检查是不是下游缓存导致的。如果解析器内部用了类似“记住上一次结果”的全局变量多线程并发时会出现结果张冠李戴。RunnableBatch 的并发环境下任何共享状态都可能是乱序元凶。5.3 现象二线程数和连接数暴涨有次测试时我观察到进程线程数突然从几十涨到几百HTTP 连接数也一路飙升最后连网关都开始超时。排查链路是先看 max_concurrency 是否真的传进了 config 对象里。有些框架的 batch 支持 config 参数但如果你用了错误的 config 键名框架会静默忽略然后采用默认并发。默认值有时是一个非常大的数字等于把整个批次同时发出去。再看是不是每次请求都新建了客户端连接。模型调用 SDK 一般会维护连接池但如果你在每次调用时都 new 一个客户端并发一高连接数就会爆炸。正确做法是让所有 batch 子任务共享同一个客户端实例。最后看供应商网关侧是否有连接复用。如果供应商侧不接受保持连接那你在客户端做再多的连接池优化也没用。这种情况只能在 batch 内部降低 max_concurrency。5.4 现象三不同供应商速率差异大导致整体被拖慢跨供应商合并时有个经典场景供应商 A 平均响应 1 秒供应商 B 平均响应 4 秒。如果你把两家放在同一个串行处理流程里或者放在同一个没有区分的 batch 里快的供应商会被慢的拖住。我对这个问题的处理方式是不为“合并而合并”。跨供应商合并的关键是入口统一和结果归一而不是非得让两条链路共享同一个并发配额。拆开跑反而更合理。比如 30 条数据供应商 A 输出质量好但慢我给它 10 条设 max_concurrency2供应商 B 快但格式不稳我给它 20 条设 max_concurrency8。两条链路的并发互不干扰最后在主流程里合并结果。这样速度快的那一侧不会因为另一侧慢而一直空等。用一句话总结跨模型供应商的“并行”可以是并行跑两条不同链路而不是把不同供应商的任务塞进同一个并发池里抢资源。5.5 一个容易忽略的内存问题RunnableBatch 默认会收集所有输出最后以列表形式返回。这在数据量几千条时没问题但到了几十万条时所有结果全堆在内存里很容易把进程内存打满。我的习惯是不要一口气把全量数据塞进一个 batch。把任务切成小批次比如每批 200 条处理完一批就立刻把结果写出去无论是写入数据库还是输出到文件然后再处理下一批。这样内存占用保持在一个稳定水位。还有一个小技巧在处理超大任务时把每条输入和结果的 request_id 一并收集起来这样即使进程中途崩溃也能根据已写入的数据反推出还有哪些没处理完。RunnableBatch 优化的是并发效率但它并不会替你解决“数据体量超过内存”的问题这是调用方需要主动设计的。再多说一句我实际用下来的习惯。每次上线批量任务前我都会先写一个小脚本分别压测串行、并发 5、并发 10、并发 20 四种配置的耗时和失败率再根据供应商限流文档确定最终的并发参数。不拍脑袋、不迷信默认值。Batch Inference 优化表面上是一个接口替换实际上是对延迟、并发、限流和容错这几件事的整体权衡。理解了 RunnableBatch 到底在调度层替你做了什么你在遇到问题时就不需要靠猜了。

RunnableBatch 实战：大模型批量推理并发优化与跨供应商实践

相关推荐

PSO-CNN-RF-ABKDE：多变量时序预测与不确定性量化实践

EasyMedia-ui：RTSP转FLV的开箱即用Web播放器

Upscayl 超分辨率实测：本地免费 AI 让老照片变清晰

更多资讯

电商小程序h5st签名机制解析：原理、风险与合规联调指南

AI Agent任务调度实战：规划、执行与并发控制的完整链路设计

两串电芯被动均衡方案：MP2642GR-Z与MK51DN512CLQ10实战

危险报文识别实战：从告警日志到分层过滤与规则引擎

拼多多客服机器人开发实战：从Webhook接入到自动回复

VS Code远程SSH开发C++环境搭建指南

今日更新

NetSpot 实战：用信道分析与频宽优化解决 Wi-Fi 卡顿

C++ split() 字符串分割的五种实现与性能实测

R语言坐标轴范围设置：从xlim/ylim到coord_cartesian的完整指南

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？