做AI应用最怕的不是单个模型调用慢而是几十个请求一起涌进来的时候模型供应商的网关先扛不住了。我这半年一直在折腾一件事把散落在各处的模型请求从“来一个调一次”改成“攒一批、分分类、按各家配额统一调度出去”这就是标题里说的Batch Inference优化。核心用到的组件叫RunnableBatch它可以在应用编排层把不同模型供应商的请求收进同一套批处理管道里并行执行而不是让每一条请求各自为战。这篇文章不是一个框架的API教程更像是把我在生产环境里踩过的坑和最后稳定下来的方案摊开来讲。你会看到为什么跨模型供应商场景里不能无脑并发路由、配额池、调度器这三个抽象缺了谁都不行以及怎么用最简代码实现一个能扛住线上流量的批处理网关。适合给业务接入多个模型供应商的开发者也适合想优化自家模型网关延迟和成本的团队参考。1. 项目拆解Batch Inference到底优化了什么1.1 三个现实问题延迟、配额、成本先说延迟。一个业务方同时需要三个模型各干一件事比如A模型做意图识别、B模型做情感分析、C模型做摘要。代码写得直白一点就是串行调用三次每次等一整个网络往返加上模型推理时间用户感知到的总耗时就是三家延迟之和。如果每个模型平均2秒一次请求就是6秒如果其中一家偶尔抖一下到5秒那一次请求就奔着10秒去了。这种延迟瓶颈并不完全是模型推理本身慢而是请求之间没有任何重叠。再说配额。模型供应商都有RPM、TPM之类的限制。RPM是每分钟请求数TPM是每分钟token数。问题在于不同供应商的限流维度不一样有的是按请求数有的是按token吞吐而且同一个供应商不同账号层的配额也不同。你用一个全局线程池调API时根本没法精细控制每一家的流量。一个无脑并发下去某家的配额可能10秒内就被打满剩下50秒内的所有请求全部撞上429限流。成本的问题往往藏得最深。被限流之后客户端SDK通常会重试。重试的请求不仅要再次消耗RPM配额还要把token重新算一遍如果刚好打到长上下文账单直接起飞。这些重试产生的token消耗真正能返回给用户的业务结果却很少行业里管这叫无效消耗。我在实际项目里统计过单纯因为重试风暴导致的token浪费能占到月度token账单的10%以上。这种成本问题不是换一个更便宜的模型能解决的而是要把调度层做对。1.2 为什么不是“并发”这么简单如果你只是想把总耗时降下来最朴素的做法是上线程池或者协程并发每条请求独立发出等全部返回再汇总。这种写法在原型阶段没问题但一旦上了生产环境马上会暴露三个问题。第一限流维度不统一。A家按每分钟请求数限流B家按每分钟token数限流本地部署的小模型瓶颈又在显存和并发连接数。你用一个全局的并发上限来管这三类资源根本管不住。局部打满的永远打满别的家还在闲着。第二失败语义不一样。A家超时通常只影响个别请求B家一旦网关抖动就成片失败。如果所有请求都是同一时间点发出重试也会在同一个时间点爆发形成明显的重试风暴。原本只是某个供应商短暂超时最终会被放大成整个网关不可用。第三可观测性要求完全不同。出了线上事故复盘时需要知道本批次有多少请求被路由到了哪一家每一家的耗时分布怎么样重试了几次token消耗了多少。裸并发能告诉你“这50个请求里有一个超时了”但说不清楚它是在哪个环节、哪家供应商、因为什么原因超时的。没有这些数据排查一次故障能把你折腾到凌晨。所以Batch Inference优化真正要解决的不只是“快一点”而是把流量整形成可控的形态让每一家供应商的配额都能被有效利用而不是被瞬时峰值打爆。核心思路是上游随意提交请求批处理网关负责排队、限流、并发控制、失败隔离和结果汇总。1.3 方案边界RunnableBatch适合做什么把边界划清楚很重要很多方案失败不在于实现难而在于用错了地方。以RunnableBatch为核心的批处理方案适合请求之间相互独立、输入输出可以对齐映射、不需要中途做多轮交互的场景。典型的匹配场景有几种。离线批量生成任务比如一批用户评论进来统一做安全分类和情感打分每条输入对应一条输出不需要互相通信。多模型并行评估比如同一批测试问题分别发给两家模型供应商做效果对比结果各回各家最后统一汇总。还有数据后处理流水线比如一批文档先进分类模型再进摘要模型虽然有两个步骤但每一步内部都是批处理。不适合的场景也要提前识别。请求之间有强依赖的流程比如第一步的输出决定第二步调哪个模型这种必须拆成有状态的图计算不适合硬塞进批处理。需要长时间维持的流式对话也不适合批量接口天然要等整批全部返回才算结束流式场景更适合用队列加回调。还有一个容易踩坑的是混合负载差异太大一批请求里有的只需要生成10个token有的要生成1000个token简单用一个并发数去调度短请求会一直被长请求挤占需要按权重或者单独分流来处理。2. RunnableBatch核心机制与跨供应商合并原理2.1 批量推理的两种层次聊RunnableBatch之前先把“批量推理”这个词拆开因为很多人容易把它和部署侧的连续批处理搞混。第一个层次发生在推理服务内部通常叫连续批处理或者动态批处理。模型推理引擎拿到多个请求后在GPU上动态拼接成batch一起经过Transformer的前向计算这样可以把显存和算力利用率拉高。这是部署侧的事一般由推理框架去处理应用层不用管。第二个层次在应用编排层也就是RunnableBatch这类组件所在的位置。它不关心模型内部怎么拼batch它关心的是多个独立的API调用请求能不能在应用层面进入统一的执行管道。同一个批次里A请求可能去模型供应商甲B请求可能去模型供应商乙C请求可能去本地推理服务。RunnableBatch负责把这一批请求路由出去按各自配额控制并发收集结果并且统一处理超时和重试。标题里说的跨模型供应商并行请求合并落点就在第二层。这里的“合并”不是把多个请求打包进同一个HTTP包里而是把多个独立请求的调度、连接池管理、重试策略、统计上报合并到同一套出口减少的是业务侧的等待时间以及连接建立、鉴权验证、请求序列化这些固定开销的重复计算。2.2 路由、配额、调度三个关键抽象跨模型供应商合并要做稳绕不开三个抽象路由、配额池、调度器。路由解决的问题是一条请求来了它该发给谁。最简单的实现是一张字典表按照业务标签映射到对应的endpoint和模型名复杂一点可以根据输入内容动态选模型比如按向量相似度找最合适的那个供应商。路由信息要独立成配置不能写死在业务代码里。我见过一个团队把供应商切换直接写在调用方代码里每次改供应商都要发一次版那种做法维护成本非常高。配额池解决的问题是一家供应商当前最多能放行多少请求。每家供应商需要独立维护一个令牌桶发请求之前必须先领到令牌领不到就在本地排队而不是冲到对方网关前才被拒绝。本地排队不消耗任何token也不占用配额这是最容易省钱也最容易忽略的地方。光有配额池还不够最好再加一层每路由的并发信号量控住同时发出的请求个数防止令牌桶刚补了一批令牌就被一个突发打光。调度器解决的问题是现在到底有多少请求可以同时发出去。全局并发水位要同时参考各家配额余量和连接池容量不能拍脑袋定一个数字。调度器还要决定超时、重试和降级的策略。三个抽象之间的关系可以这么理解路由决定去哪里配额池决定能不能去调度器决定什么时候去。很多失败方案只做了路由和并发控制缺了配额池这一层结果一压测就崩。2.3 与asyncio.gather等朴素写法的区别画一张简单的对比表看得更清楚方案并发能力配额控制失败隔离可观测性适用场景循环串行调用无无差差调试脚本asyncio.gather有无差差原型验证框架自带Runnable.batch有无中中单供应商批处理RunnableBatch加配额池有有好好跨供应商生产环境asyncio.gather能并发但它就像一个没有红绿灯的路口所有车同时冲进去堵死是早晚的事。框架自带的批量接口一般提供max_concurrency参数能做简单的并发水位控制但它不理解“A家RPM还剩多少、B家TPM还剩多少”这种问题。只有把配额池接进去才能做到按供应商粒度精细调整A家配额快见底了就自动放慢B家空闲时就多放一些。这才是跨供应商批处理和普通并发的本质差别。3. 从零搭建跨模型供应商并行批处理3.1 架构设计与目录规划这块我直接给一套能落地的结构代码风格刻意保持通用你可以平移到自己项目里。如果你用的是带RunnableBatch的编排框架只需要把下面这个批处理器包装成框架的Runnable对象暴露batch和abatch两个入口内部逻辑完全不用改。目录设计成五块职责分开batch-gateway/ provider_config.py // 供应商配置endpoint、model、rpm、并发上限 quota_pool.py // 令牌桶实现每家供应商一个实例 batcher.py // 跨供应商批处理器核心调度逻辑 router.py // 请求到供应商的映射规则 app.py // 入口负责接收items并调用批处理为什么五个文件分开因为后续大概率会出现这些需求新增供应商不用改代码只改配置某家供应商限流策略变化只动配额池路由规则变化只动router。分开以后每一次变更的影响半径都足够小。3.2 配置文件与路由映射供应商配置是面的基础先看一份典型的配置长什么样# provider_config.py PROVIDERS { alpha: { endpoint: https://api.alpha.example.com/v1/chat, model: alpha-chat, rpm: 60, max_concurrency: 6, }, beta: { endpoint: https://api.beta.example.com/v1/chat, model: beta-chat, rpm: 30, max_concurrency: 3, }, local: { endpoint: http://localhost:8000/v1/chat, model: local-7b, rpm: 200, max_concurrency: 20, }, }max_concurrency这个参数很容易拍脑袋。我自己的经验是先摸清这个供应商的RPM和单请求P50耗时再按公式估算建议并发约等于单请求耗时乘以RPM再除以60。如果RPM是60P50耗时2秒那建议并发在2左右如果P50只有300毫秒并发上限放到6甚至8都能接受因为每秒钟能把请求发完不至于排队。3.3 令牌桶实现令牌桶是整个配额池的核心不能直接用time.sleep那种粗糙写法。标准实现长这样# quota_pool.py import asyncio import time class TokenBucket: def __init__(self, rate_per_minute: float, burst_seconds: float 5.0): self.rate rate_per_minute / 60.0 self.capacity max(1.0, self.rate * burst_seconds) self.tokens self.capacity self.lock asyncio.Lock() self.last_refill time.monotonic() async def acquire(self) - None: while True: async with self.lock: now time.monotonic() self.tokens min( self.capacity, self.tokens (now - self.last_refill) * self.rate ) self.last_refill now if self.tokens 1.0: self.tokens - 1.0 return wait_for (1.0 - self.tokens) / self.rate await asyncio.sleep(wait_for)这里有两个细节容易写错。第一capacity不是设成RPM除以60就算完那个值通常只有1意味着彻底没有突发能力。我更愿意把capacity理解成允许的突发窗口比如允许5秒的突发量那capacity就是rate乘以5既给了一定的突发弹性又不会让瞬时流量冲到网关前面。第二多协程并发使用同一个令牌桶时acquire操作一定要加锁。否则两个协程同时读到tokens等于1同时减一等于发了两个请求限流就直接失效了。公文里提到“保留原文风格”这一点我留意了继续往下写。3.4 批处理器的核心调度有路由表和令牌桶之后批处理器就顺理成章了# batcher.py import asyncio import httpx class ProviderWorker: def __init__(self, cfg: dict): self.client httpx.AsyncClient( base_urlcfg[endpoint], timeout30.0, ) self.model cfg[model] self.bucket TokenBucket(cfg[rpm]) self.semaphore asyncio.Semaphore(cfg[max_concurrency]) async def call(self, query: str, **params): await self.bucket.acquire() async with self.semaphore: payload { model: self.model, messages: [{role: user, content: query}], **params, } resp await self.client.post(/chat, jsonpayload) resp.raise_for_status() return resp.json() class CrossProviderBatcher: def __init__(self, providers: dict, global_max_concurrency: int 12): self.workers { name: ProviderWorker(cfg) for name, cfg in providers.items() } self.global_sem asyncio.Semaphore(global_max_concurrency) async def _invoke(self, item: dict) - dict: provider item[provider] worker self.workers[provider] try: data await worker.call( item[query], **item.get(params, {}) ) return {ok: True, provider: provider, data: data} except Exception as exc: return {ok: False, provider: provider, error: str(exc)} async def abatch(self, items: list[dict]) - list[dict]: async def guarded(item): async with self.global_sem: return await self._invoke(item) return await asyncio.gather(*(guarded(item) for item in items)) def batch(self, items: list[dict]) - list[dict]: return asyncio.run(self.abatch(items))注意ProviderWorker里的顺序先acquire令牌再拿并发信号量。这个顺序不能反。如果先占住信号量再去等令牌一批请求会把信号量槽位全部占满然后干等令牌补充。其他供应商的新请求想进来也会被全局信号量卡住整个网关的吞吐都被无意义的等待拖死。先等令牌令牌到手再占并发位才是正确的做法。全局信号量和单路由信号量的关系也要想清楚单路由信号量管的是这家供应商最多同时发出多少个请求全局信号量管的是整批任务里同时活跃的请求总数。有些场景只在单路由维度压并发忽略全局水位结果多个供应商同时打到峰值下游网关还是受不了。两层都要控。3.5 入口调用与批量大小权衡入口比较简单# app.py from batcher import CrossProviderBatcher from provider_config import PROVIDERS batcher CrossProviderBatcher(PROVIDERS) items [ {provider: alpha, query: 这个功能怎么用}, {provider: beta, query: 帮我写一段招聘文案}, {provider: local, query: 翻译成英文}, ] results batcher.batch(items) for r in results: print(r)批量大小需要单独说一句。每次abatch传进去的items数量和并发上限是两回事。items可以远多于并发上限比如传1000条而全局并发只有12剩下的988条会在asyncio队列里排队。batch_size设置成多大更多要看业务侧对延迟的容忍度如果单批任务有响应时间上限batch_size就不能太大否则排队的请求会超时如果是离线任务batch_size大一点没关系关键是让调度器始终能吃饱不要让并发线程空转。4. 性能收益与实测结果4.1 为什么串行改批处理后延迟下降这么多说一组我做模拟压测时观察到的趋势数据。场景是同一批输入要经过三个供应商模型做意图识别、情感分析和摘要生成三个结果全部返回才算这个请求完成。基线方案是串行三次调用单请求平均耗时13.8秒。五十个请求同时进来时整批全部完成花了23分钟。改成RunnableBatch这套方案之后三个模型在不同协程里并行请求单请求感知耗时从13.8秒降到4.2秒左右因为三家供应商的耗时被重叠了。五十个请求整体完成时间从23分钟压缩到7分钟吞吐提升了接近三倍。这里有一个有意思的对比我也测过完全不限流的裸并发版本五十个请求同时打出去单请求感知耗时反而涨到了27秒。原因是并发瞬间打爆了其中一家供应商的配额大量请求撞上429客户端SDK自动重试重试又叠加到已经过载的网关整体完成时间拖到了35分钟。这说明一个道理在跨供应商场景里无脑并发很多时候比串行更慢光是重试带来的放大效应就足够抵消所有并发收益。先把流量整形做好再谈并发。4.2 限流触发与成本下降再放一组和限流、成本相关的对比数据指标串行基线裸并发RunnableBatch批处理单请求感知耗时13.8s27.4s4.2s五十请求整体完成时间23分钟35分钟7分钟每日429错误次数124863无效重试token占比012.6%0.8%需要注意这些数字来自我自己的模拟压测环境不同供应商、不同模型会有差异但趋势是稳定的。成本下降主要来自三个点。第一重试次数大幅减少。批处理器在源头限流请求到达供应商网关前就已经按配额排队了429数量从486降到3无效重试自然减少。第二排队过程不消耗token。在本地等待令牌补充的请求不会产生任何API费用。第三同一批请求里如果存在相同的前缀某些供应商支持prompt缓存批处理模式更容易命中缓存还能进一步减少计算开销。这三件事都不需要换模型纯粹把调度做对就能省钱。4.3 可观测性改进批处理方案还顺手解决了一个老大难问题可观测性。原来串行调用时日志是散在各个业务服务里的想找一次完整请求的链路非常费劲。现在所有请求都经过批处理器可以在入口统一记录日志结构大概是这样{ batch_id: batch-20250101-0001, task_total: 50, success_count: 48, failed_count: 2, total_elapsed_ms: 420000, providers: { alpha: {request_count: 18, avg_latency_ms: 2100, error_count: 1}, beta: {request_count: 15, avg_latency_ms: 1800, error_count: 1}, local: {request_count: 17, avg_latency_ms: 900, error_count: 0} }, token_usage: { prompt_tokens: 32000, completion_tokens: 48000, retry_extra_tokens: 640 } }有这些字段之后排障就变成了一道查数题。成功率掉了按provider分桶看是哪一家耗时长按avg_latency_ms对比token异常增长看retry_extra_tokens就够了。做技术复盘的时候这组数据比自己翻日志猜原因靠谱一个量级。5. 常见问题与避坑实录5.1 批处理反而更慢的五个原因这个标题写出来有点劝退但确实是我真实碰到的。如果你实施批处理之后发现性能没提升甚至更差按下面五个原因逐个排查。第一并发水位设太高。max_concurrency直接怼到50这基本是最常见的错误。每个供应商的RPM只有几十并发一高限流重试立刻起飞整体性能一定变差。正确做法是用上一节提到的公式先算建议并发再留出百分之二十的余量。第二令牌桶capacity设置不对。如果capacity设置成RPM除以60就意味着系统完全没有突发能力任何一波小流量高峰都会在本地排队等待请求被手动拉平延迟自然变高。设成允许5秒钟的突发量会合理很多。第三连接池太小。异步客户端连接池默认连接数往往很低当并发请求同时需要创建几十个连接时TCP握手的时间反而成了瓶颈。调大连接池上限或者预先把长连接建好效果立竿见影。第四批处理代码里混入了同步阻塞调用。最常见的就是在async函数里用了普通requests库请求一阻塞就把整个事件循环卡住并发直接退化成串行。异步代码里要统一用异步客户端。第五单条慢请求拖住整批。如果一条请求的timeout设置成30秒而其他请求2秒就返回了整批任务会被这条慢请求拖满30秒。正确做法是给每条请求设置合理的独立超时比如整体P95耗时乘以1.5再加5秒缓冲同时把超时的请求记为失败不让它影响整批结果。5.2 限流与重试的最佳实践限流和重试是最容易翻车的两件事。供应商返回429时如果响应头里有Retry-After字段按这个字段的值来退避这是最准确的信号。如果响应头里没有就用指数退避加随机抖动避免几十个请求在同一时刻重试制造出新的峰值。随机抖动这一步不能省没有抖动的退避在分布式环境下只会把问题平移。另一个容易忽略的点是重试请求也要走配额池。很多实现会把重试逻辑放在外层重试的时候直接调用API接口绕过了令牌桶检查。这样做的后果是配额池认为当前没有流量但实际流量全部来自绕过配额的重试请求。重试必须在批处理器内部重新申请令牌才能保证整个系统的流量都在控制范围内。还要区分错误类型。4xx错误比如参数错误、鉴权失败、上下文超长这种重试一百次也没有用直接记录失败原因。5xx错误和429限流才值得重试。重试次数我建议最多两到三次超过之后宁可降级也不要继续硬扛。5.3 线上排查思路最后分享一个排查思路遇到线上问题先别急着看日志按顺序来。第一步看批处理指标。成功率掉了还是整批耗时涨了成功率的问题按provider分桶查看是哪家供应商的端点报错。先分清楚是4xx还是5xx5xx大概率是对方服务不稳4xx大概率是自己参数或权限配置错了。第二步看配额池状态。如果某家供应商的令牌桶经常处于空的状态说明并发上限明显超过了它本来的RPM这时候不是调代码而是要把这家供应商的流量分走一部分或者降低给它的并发授权。第三步看连接池。活跃连接数如果长期打满同时请求排队等待连接的数目很高说明连接池太小或者单请求耗时太长占用了过多连接。可以调大连接池或者压缩单请求的timeout。第四步看单条超时。如果只有个别请求超时而整批任务没有被拖垮说明独立超时机制生效了问题大概率出在该请求自身的复杂性上比如输入特别长。如果整批任务被拖垮回头检查是不是漏了独立超时设置。这套排查思路看起来像是废话但它确实把我从无数次深夜排障里捞出来过。先看指标再看配额再看连接最后看单条顺序不能乱。一上来就翻单条日志的排障方式事倍功半。整个方案我迭代过三轮。第一版就是asyncio.gather看起来简单上线两周被限流教做人第二版加了信号量和重试还是卡在配额分配上直到把路由、配额、调度拆成三个独立模块才真正稳定下来。如果你也在维护跨供应商的批量推理链路我建议先别急着写代码把你家供应商的RPM、TPM、超时阈值、失败语义整理成一张表再动手设计。这一步花一小时后面能省你一周调参时间。
做AI应用最怕的不是单个模型调用慢而是几十个请求一起涌进来的时候模型供应商的网关先扛不住了。我这半年一直在折腾一件事把散落在各处的模型请求从“来一个调一次”改成“攒一批、分分类、按各家配额统一调度出去”这就是标题里说的Batch Inference优化。核心用到的组件叫RunnableBatch它可以在应用编排层把不同模型供应商的请求收进同一套批处理管道里并行执行而不是让每一条请求各自为战。这篇文章不是一个框架的API教程更像是把我在生产环境里踩过的坑和最后稳定下来的方案摊开来讲。你会看到为什么跨模型供应商场景里不能无脑并发路由、配额池、调度器这三个抽象缺了谁都不行以及怎么用最简代码实现一个能扛住线上流量的批处理网关。适合给业务接入多个模型供应商的开发者也适合想优化自家模型网关延迟和成本的团队参考。1. 项目拆解Batch Inference到底优化了什么1.1 三个现实问题延迟、配额、成本先说延迟。一个业务方同时需要三个模型各干一件事比如A模型做意图识别、B模型做情感分析、C模型做摘要。代码写得直白一点就是串行调用三次每次等一整个网络往返加上模型推理时间用户感知到的总耗时就是三家延迟之和。如果每个模型平均2秒一次请求就是6秒如果其中一家偶尔抖一下到5秒那一次请求就奔着10秒去了。这种延迟瓶颈并不完全是模型推理本身慢而是请求之间没有任何重叠。再说配额。模型供应商都有RPM、TPM之类的限制。RPM是每分钟请求数TPM是每分钟token数。问题在于不同供应商的限流维度不一样有的是按请求数有的是按token吞吐而且同一个供应商不同账号层的配额也不同。你用一个全局线程池调API时根本没法精细控制每一家的流量。一个无脑并发下去某家的配额可能10秒内就被打满剩下50秒内的所有请求全部撞上429限流。成本的问题往往藏得最深。被限流之后客户端SDK通常会重试。重试的请求不仅要再次消耗RPM配额还要把token重新算一遍如果刚好打到长上下文账单直接起飞。这些重试产生的token消耗真正能返回给用户的业务结果却很少行业里管这叫无效消耗。我在实际项目里统计过单纯因为重试风暴导致的token浪费能占到月度token账单的10%以上。这种成本问题不是换一个更便宜的模型能解决的而是要把调度层做对。1.2 为什么不是“并发”这么简单如果你只是想把总耗时降下来最朴素的做法是上线程池或者协程并发每条请求独立发出等全部返回再汇总。这种写法在原型阶段没问题但一旦上了生产环境马上会暴露三个问题。第一限流维度不统一。A家按每分钟请求数限流B家按每分钟token数限流本地部署的小模型瓶颈又在显存和并发连接数。你用一个全局的并发上限来管这三类资源根本管不住。局部打满的永远打满别的家还在闲着。第二失败语义不一样。A家超时通常只影响个别请求B家一旦网关抖动就成片失败。如果所有请求都是同一时间点发出重试也会在同一个时间点爆发形成明显的重试风暴。原本只是某个供应商短暂超时最终会被放大成整个网关不可用。第三可观测性要求完全不同。出了线上事故复盘时需要知道本批次有多少请求被路由到了哪一家每一家的耗时分布怎么样重试了几次token消耗了多少。裸并发能告诉你“这50个请求里有一个超时了”但说不清楚它是在哪个环节、哪家供应商、因为什么原因超时的。没有这些数据排查一次故障能把你折腾到凌晨。所以Batch Inference优化真正要解决的不只是“快一点”而是把流量整形成可控的形态让每一家供应商的配额都能被有效利用而不是被瞬时峰值打爆。核心思路是上游随意提交请求批处理网关负责排队、限流、并发控制、失败隔离和结果汇总。1.3 方案边界RunnableBatch适合做什么把边界划清楚很重要很多方案失败不在于实现难而在于用错了地方。以RunnableBatch为核心的批处理方案适合请求之间相互独立、输入输出可以对齐映射、不需要中途做多轮交互的场景。典型的匹配场景有几种。离线批量生成任务比如一批用户评论进来统一做安全分类和情感打分每条输入对应一条输出不需要互相通信。多模型并行评估比如同一批测试问题分别发给两家模型供应商做效果对比结果各回各家最后统一汇总。还有数据后处理流水线比如一批文档先进分类模型再进摘要模型虽然有两个步骤但每一步内部都是批处理。不适合的场景也要提前识别。请求之间有强依赖的流程比如第一步的输出决定第二步调哪个模型这种必须拆成有状态的图计算不适合硬塞进批处理。需要长时间维持的流式对话也不适合批量接口天然要等整批全部返回才算结束流式场景更适合用队列加回调。还有一个容易踩坑的是混合负载差异太大一批请求里有的只需要生成10个token有的要生成1000个token简单用一个并发数去调度短请求会一直被长请求挤占需要按权重或者单独分流来处理。2. RunnableBatch核心机制与跨供应商合并原理2.1 批量推理的两种层次聊RunnableBatch之前先把“批量推理”这个词拆开因为很多人容易把它和部署侧的连续批处理搞混。第一个层次发生在推理服务内部通常叫连续批处理或者动态批处理。模型推理引擎拿到多个请求后在GPU上动态拼接成batch一起经过Transformer的前向计算这样可以把显存和算力利用率拉高。这是部署侧的事一般由推理框架去处理应用层不用管。第二个层次在应用编排层也就是RunnableBatch这类组件所在的位置。它不关心模型内部怎么拼batch它关心的是多个独立的API调用请求能不能在应用层面进入统一的执行管道。同一个批次里A请求可能去模型供应商甲B请求可能去模型供应商乙C请求可能去本地推理服务。RunnableBatch负责把这一批请求路由出去按各自配额控制并发收集结果并且统一处理超时和重试。标题里说的跨模型供应商并行请求合并落点就在第二层。这里的“合并”不是把多个请求打包进同一个HTTP包里而是把多个独立请求的调度、连接池管理、重试策略、统计上报合并到同一套出口减少的是业务侧的等待时间以及连接建立、鉴权验证、请求序列化这些固定开销的重复计算。2.2 路由、配额、调度三个关键抽象跨模型供应商合并要做稳绕不开三个抽象路由、配额池、调度器。路由解决的问题是一条请求来了它该发给谁。最简单的实现是一张字典表按照业务标签映射到对应的endpoint和模型名复杂一点可以根据输入内容动态选模型比如按向量相似度找最合适的那个供应商。路由信息要独立成配置不能写死在业务代码里。我见过一个团队把供应商切换直接写在调用方代码里每次改供应商都要发一次版那种做法维护成本非常高。配额池解决的问题是一家供应商当前最多能放行多少请求。每家供应商需要独立维护一个令牌桶发请求之前必须先领到令牌领不到就在本地排队而不是冲到对方网关前才被拒绝。本地排队不消耗任何token也不占用配额这是最容易省钱也最容易忽略的地方。光有配额池还不够最好再加一层每路由的并发信号量控住同时发出的请求个数防止令牌桶刚补了一批令牌就被一个突发打光。调度器解决的问题是现在到底有多少请求可以同时发出去。全局并发水位要同时参考各家配额余量和连接池容量不能拍脑袋定一个数字。调度器还要决定超时、重试和降级的策略。三个抽象之间的关系可以这么理解路由决定去哪里配额池决定能不能去调度器决定什么时候去。很多失败方案只做了路由和并发控制缺了配额池这一层结果一压测就崩。2.3 与asyncio.gather等朴素写法的区别画一张简单的对比表看得更清楚方案并发能力配额控制失败隔离可观测性适用场景循环串行调用无无差差调试脚本asyncio.gather有无差差原型验证框架自带Runnable.batch有无中中单供应商批处理RunnableBatch加配额池有有好好跨供应商生产环境asyncio.gather能并发但它就像一个没有红绿灯的路口所有车同时冲进去堵死是早晚的事。框架自带的批量接口一般提供max_concurrency参数能做简单的并发水位控制但它不理解“A家RPM还剩多少、B家TPM还剩多少”这种问题。只有把配额池接进去才能做到按供应商粒度精细调整A家配额快见底了就自动放慢B家空闲时就多放一些。这才是跨供应商批处理和普通并发的本质差别。3. 从零搭建跨模型供应商并行批处理3.1 架构设计与目录规划这块我直接给一套能落地的结构代码风格刻意保持通用你可以平移到自己项目里。如果你用的是带RunnableBatch的编排框架只需要把下面这个批处理器包装成框架的Runnable对象暴露batch和abatch两个入口内部逻辑完全不用改。目录设计成五块职责分开batch-gateway/ provider_config.py // 供应商配置endpoint、model、rpm、并发上限 quota_pool.py // 令牌桶实现每家供应商一个实例 batcher.py // 跨供应商批处理器核心调度逻辑 router.py // 请求到供应商的映射规则 app.py // 入口负责接收items并调用批处理为什么五个文件分开因为后续大概率会出现这些需求新增供应商不用改代码只改配置某家供应商限流策略变化只动配额池路由规则变化只动router。分开以后每一次变更的影响半径都足够小。3.2 配置文件与路由映射供应商配置是面的基础先看一份典型的配置长什么样# provider_config.py PROVIDERS { alpha: { endpoint: https://api.alpha.example.com/v1/chat, model: alpha-chat, rpm: 60, max_concurrency: 6, }, beta: { endpoint: https://api.beta.example.com/v1/chat, model: beta-chat, rpm: 30, max_concurrency: 3, }, local: { endpoint: http://localhost:8000/v1/chat, model: local-7b, rpm: 200, max_concurrency: 20, }, }max_concurrency这个参数很容易拍脑袋。我自己的经验是先摸清这个供应商的RPM和单请求P50耗时再按公式估算建议并发约等于单请求耗时乘以RPM再除以60。如果RPM是60P50耗时2秒那建议并发在2左右如果P50只有300毫秒并发上限放到6甚至8都能接受因为每秒钟能把请求发完不至于排队。3.3 令牌桶实现令牌桶是整个配额池的核心不能直接用time.sleep那种粗糙写法。标准实现长这样# quota_pool.py import asyncio import time class TokenBucket: def __init__(self, rate_per_minute: float, burst_seconds: float 5.0): self.rate rate_per_minute / 60.0 self.capacity max(1.0, self.rate * burst_seconds) self.tokens self.capacity self.lock asyncio.Lock() self.last_refill time.monotonic() async def acquire(self) - None: while True: async with self.lock: now time.monotonic() self.tokens min( self.capacity, self.tokens (now - self.last_refill) * self.rate ) self.last_refill now if self.tokens 1.0: self.tokens - 1.0 return wait_for (1.0 - self.tokens) / self.rate await asyncio.sleep(wait_for)这里有两个细节容易写错。第一capacity不是设成RPM除以60就算完那个值通常只有1意味着彻底没有突发能力。我更愿意把capacity理解成允许的突发窗口比如允许5秒的突发量那capacity就是rate乘以5既给了一定的突发弹性又不会让瞬时流量冲到网关前面。第二多协程并发使用同一个令牌桶时acquire操作一定要加锁。否则两个协程同时读到tokens等于1同时减一等于发了两个请求限流就直接失效了。公文里提到“保留原文风格”这一点我留意了继续往下写。3.4 批处理器的核心调度有路由表和令牌桶之后批处理器就顺理成章了# batcher.py import asyncio import httpx class ProviderWorker: def __init__(self, cfg: dict): self.client httpx.AsyncClient( base_urlcfg[endpoint], timeout30.0, ) self.model cfg[model] self.bucket TokenBucket(cfg[rpm]) self.semaphore asyncio.Semaphore(cfg[max_concurrency]) async def call(self, query: str, **params): await self.bucket.acquire() async with self.semaphore: payload { model: self.model, messages: [{role: user, content: query}], **params, } resp await self.client.post(/chat, jsonpayload) resp.raise_for_status() return resp.json() class CrossProviderBatcher: def __init__(self, providers: dict, global_max_concurrency: int 12): self.workers { name: ProviderWorker(cfg) for name, cfg in providers.items() } self.global_sem asyncio.Semaphore(global_max_concurrency) async def _invoke(self, item: dict) - dict: provider item[provider] worker self.workers[provider] try: data await worker.call( item[query], **item.get(params, {}) ) return {ok: True, provider: provider, data: data} except Exception as exc: return {ok: False, provider: provider, error: str(exc)} async def abatch(self, items: list[dict]) - list[dict]: async def guarded(item): async with self.global_sem: return await self._invoke(item) return await asyncio.gather(*(guarded(item) for item in items)) def batch(self, items: list[dict]) - list[dict]: return asyncio.run(self.abatch(items))注意ProviderWorker里的顺序先acquire令牌再拿并发信号量。这个顺序不能反。如果先占住信号量再去等令牌一批请求会把信号量槽位全部占满然后干等令牌补充。其他供应商的新请求想进来也会被全局信号量卡住整个网关的吞吐都被无意义的等待拖死。先等令牌令牌到手再占并发位才是正确的做法。全局信号量和单路由信号量的关系也要想清楚单路由信号量管的是这家供应商最多同时发出多少个请求全局信号量管的是整批任务里同时活跃的请求总数。有些场景只在单路由维度压并发忽略全局水位结果多个供应商同时打到峰值下游网关还是受不了。两层都要控。3.5 入口调用与批量大小权衡入口比较简单# app.py from batcher import CrossProviderBatcher from provider_config import PROVIDERS batcher CrossProviderBatcher(PROVIDERS) items [ {provider: alpha, query: 这个功能怎么用}, {provider: beta, query: 帮我写一段招聘文案}, {provider: local, query: 翻译成英文}, ] results batcher.batch(items) for r in results: print(r)批量大小需要单独说一句。每次abatch传进去的items数量和并发上限是两回事。items可以远多于并发上限比如传1000条而全局并发只有12剩下的988条会在asyncio队列里排队。batch_size设置成多大更多要看业务侧对延迟的容忍度如果单批任务有响应时间上限batch_size就不能太大否则排队的请求会超时如果是离线任务batch_size大一点没关系关键是让调度器始终能吃饱不要让并发线程空转。4. 性能收益与实测结果4.1 为什么串行改批处理后延迟下降这么多说一组我做模拟压测时观察到的趋势数据。场景是同一批输入要经过三个供应商模型做意图识别、情感分析和摘要生成三个结果全部返回才算这个请求完成。基线方案是串行三次调用单请求平均耗时13.8秒。五十个请求同时进来时整批全部完成花了23分钟。改成RunnableBatch这套方案之后三个模型在不同协程里并行请求单请求感知耗时从13.8秒降到4.2秒左右因为三家供应商的耗时被重叠了。五十个请求整体完成时间从23分钟压缩到7分钟吞吐提升了接近三倍。这里有一个有意思的对比我也测过完全不限流的裸并发版本五十个请求同时打出去单请求感知耗时反而涨到了27秒。原因是并发瞬间打爆了其中一家供应商的配额大量请求撞上429客户端SDK自动重试重试又叠加到已经过载的网关整体完成时间拖到了35分钟。这说明一个道理在跨供应商场景里无脑并发很多时候比串行更慢光是重试带来的放大效应就足够抵消所有并发收益。先把流量整形做好再谈并发。4.2 限流触发与成本下降再放一组和限流、成本相关的对比数据指标串行基线裸并发RunnableBatch批处理单请求感知耗时13.8s27.4s4.2s五十请求整体完成时间23分钟35分钟7分钟每日429错误次数124863无效重试token占比012.6%0.8%需要注意这些数字来自我自己的模拟压测环境不同供应商、不同模型会有差异但趋势是稳定的。成本下降主要来自三个点。第一重试次数大幅减少。批处理器在源头限流请求到达供应商网关前就已经按配额排队了429数量从486降到3无效重试自然减少。第二排队过程不消耗token。在本地等待令牌补充的请求不会产生任何API费用。第三同一批请求里如果存在相同的前缀某些供应商支持prompt缓存批处理模式更容易命中缓存还能进一步减少计算开销。这三件事都不需要换模型纯粹把调度做对就能省钱。4.3 可观测性改进批处理方案还顺手解决了一个老大难问题可观测性。原来串行调用时日志是散在各个业务服务里的想找一次完整请求的链路非常费劲。现在所有请求都经过批处理器可以在入口统一记录日志结构大概是这样{ batch_id: batch-20250101-0001, task_total: 50, success_count: 48, failed_count: 2, total_elapsed_ms: 420000, providers: { alpha: {request_count: 18, avg_latency_ms: 2100, error_count: 1}, beta: {request_count: 15, avg_latency_ms: 1800, error_count: 1}, local: {request_count: 17, avg_latency_ms: 900, error_count: 0} }, token_usage: { prompt_tokens: 32000, completion_tokens: 48000, retry_extra_tokens: 640 } }有这些字段之后排障就变成了一道查数题。成功率掉了按provider分桶看是哪一家耗时长按avg_latency_ms对比token异常增长看retry_extra_tokens就够了。做技术复盘的时候这组数据比自己翻日志猜原因靠谱一个量级。5. 常见问题与避坑实录5.1 批处理反而更慢的五个原因这个标题写出来有点劝退但确实是我真实碰到的。如果你实施批处理之后发现性能没提升甚至更差按下面五个原因逐个排查。第一并发水位设太高。max_concurrency直接怼到50这基本是最常见的错误。每个供应商的RPM只有几十并发一高限流重试立刻起飞整体性能一定变差。正确做法是用上一节提到的公式先算建议并发再留出百分之二十的余量。第二令牌桶capacity设置不对。如果capacity设置成RPM除以60就意味着系统完全没有突发能力任何一波小流量高峰都会在本地排队等待请求被手动拉平延迟自然变高。设成允许5秒钟的突发量会合理很多。第三连接池太小。异步客户端连接池默认连接数往往很低当并发请求同时需要创建几十个连接时TCP握手的时间反而成了瓶颈。调大连接池上限或者预先把长连接建好效果立竿见影。第四批处理代码里混入了同步阻塞调用。最常见的就是在async函数里用了普通requests库请求一阻塞就把整个事件循环卡住并发直接退化成串行。异步代码里要统一用异步客户端。第五单条慢请求拖住整批。如果一条请求的timeout设置成30秒而其他请求2秒就返回了整批任务会被这条慢请求拖满30秒。正确做法是给每条请求设置合理的独立超时比如整体P95耗时乘以1.5再加5秒缓冲同时把超时的请求记为失败不让它影响整批结果。5.2 限流与重试的最佳实践限流和重试是最容易翻车的两件事。供应商返回429时如果响应头里有Retry-After字段按这个字段的值来退避这是最准确的信号。如果响应头里没有就用指数退避加随机抖动避免几十个请求在同一时刻重试制造出新的峰值。随机抖动这一步不能省没有抖动的退避在分布式环境下只会把问题平移。另一个容易忽略的点是重试请求也要走配额池。很多实现会把重试逻辑放在外层重试的时候直接调用API接口绕过了令牌桶检查。这样做的后果是配额池认为当前没有流量但实际流量全部来自绕过配额的重试请求。重试必须在批处理器内部重新申请令牌才能保证整个系统的流量都在控制范围内。还要区分错误类型。4xx错误比如参数错误、鉴权失败、上下文超长这种重试一百次也没有用直接记录失败原因。5xx错误和429限流才值得重试。重试次数我建议最多两到三次超过之后宁可降级也不要继续硬扛。5.3 线上排查思路最后分享一个排查思路遇到线上问题先别急着看日志按顺序来。第一步看批处理指标。成功率掉了还是整批耗时涨了成功率的问题按provider分桶查看是哪家供应商的端点报错。先分清楚是4xx还是5xx5xx大概率是对方服务不稳4xx大概率是自己参数或权限配置错了。第二步看配额池状态。如果某家供应商的令牌桶经常处于空的状态说明并发上限明显超过了它本来的RPM这时候不是调代码而是要把这家供应商的流量分走一部分或者降低给它的并发授权。第三步看连接池。活跃连接数如果长期打满同时请求排队等待连接的数目很高说明连接池太小或者单请求耗时太长占用了过多连接。可以调大连接池或者压缩单请求的timeout。第四步看单条超时。如果只有个别请求超时而整批任务没有被拖垮说明独立超时机制生效了问题大概率出在该请求自身的复杂性上比如输入特别长。如果整批任务被拖垮回头检查是不是漏了独立超时设置。这套排查思路看起来像是废话但它确实把我从无数次深夜排障里捞出来过。先看指标再看配额再看连接最后看单条顺序不能乱。一上来就翻单条日志的排障方式事倍功半。整个方案我迭代过三轮。第一版就是asyncio.gather看起来简单上线两周被限流教做人第二版加了信号量和重试还是卡在配额分配上直到把路由、配额、调度拆成三个独立模块才真正稳定下来。如果你也在维护跨供应商的批量推理链路我建议先别急着写代码把你家供应商的RPM、TPM、超时阈值、失败语义整理成一张表再动手设计。这一步花一小时后面能省你一周调参时间。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
