AI 推理批处理调度合批策略、延迟约束与资源碎片整理一、你的推理服务 GPU 利用率 15%但 P99 延迟却到了 5 秒GPU 推理服务有个反直觉的现象单个请求 100ms并发 10 个请求 P99 到了 5 秒。原因不是 GPU 不够快而是每个请求独占一次 GPU kernel launch大量时间被 kernel 调度开销和显存零散分配吃掉了。推理批处理Batching的本质是把多个独立的推理请求打包成一个 batch一次 kernel launch 服务多个请求。但批处理引入了延迟权衡——等 request 攒够了才执行 batch越大的 batch 吞吐越高、但单个 request 的等待时间越长。这个 trade-off 是批处理调度的核心。有几种常见的合批策略动态批处理Dynamic Batching——在一定时间窗口内收集请求满了或超时就执行连续批处理Continuous Batching——请求到达后立即加入当前正在执行的 batch如果 KV cache 能容纳贪心批处理——按每个请求的预估计算量做 bin-packing同量级的请求放在一起。二、底层机制与原理剖析批处理调度的核心流程flowchart TD A[推理请求到达] -- B[请求队列] B -- C{Batch Scheduler} C -- D[策略1: 动态批处理] C -- E[策略2: 连续批处理] C -- F[策略3: 贪心打包] D -- D1[等待窗口内收集请求] D1 -- D2{窗口到期 OR 队列满?} D2 --|是| G[执行 Batch] D2 --|否| D1 E -- E1[请求到达立即加入] E1 -- E2{当前 Batch 有空位?} E2 --|是| G E2 --|否| E3[等待下一 Batch] F -- F1[按计算量分组] F1 -- F2[同量级打包] F2 -- G G -- H[GPU Kernel Launch] H -- I[并行推理] I -- J[拆分结果] J -- K[返回各请求]关键设计点动态批处理窗口时间窗口大小的选择决定了延迟-吞吐的 trade-off。窗口太短如 1msbatch 太小吞吐量上不去。窗口太长如 100msP99 延迟会被拉高。建议用自适应窗口——根据当前队列深度动态调整。连续批处理的 KV Cache最大的限制是显存。一个 batch 的所有请求共享一次 kernel launch但各自的 KV cache 独立占用显存。如果显存不足当前 batch 中的请求会被部分驱逐preemption这部分请求的前向计算需要从头重算。贪心打包的意义不同请求的计算量可能差 10 倍prompt 长度 10 tokens vs 1000 tokens。如果把长短混在一个 batch长请求会拖慢整个 batch——因为 kernel 必须等最慢的那个完成。贪心打包把长请求放一个 batch、短请求放另一个 batch避免相互拖累。三、生产级代码实现 推理批处理调度器 三种策略动态批处理 / 连续批处理 / 贪心打包 from dataclasses import dataclass, field from typing import List, Optional, Callable from collections import deque import time import threading import asyncio import heapq dataclass(orderTrue) class InferenceRequest: 推理请求 priority: int 0 # 优先级越小越优先 arrival_time: float field(default_factorytime.time, compareFalse) prompt_tokens: int 0 # Prompt Token 数估算计算量 max_new_tokens: int 256 callback: Optional[Callable] field(defaultNone, compareFalse) metadata: dict field(default_factorydict, compareFalse) # 延迟约束 max_wait_ms: float 100.0 # 最大等待延迟 deadline: float field(initFalse, compareFalse) def __post_init__(self): self.deadline self.arrival_time self.max_wait_ms / 1000.0 class DynamicBatcher: 动态批处理器 策略在时间窗口或容量限制内收集请求达到任一上限即执行。 设计决策用最大延迟约束代替固定窗口大小 每个请求有个 deadline保证不超过延迟预算。 def __init__( self, max_batch_size: int 32, max_delay_ms: float 10.0, inference_fn: Callable None, ): self.max_batch_size max_batch_size self.max_delay_ms max_delay_ms self.inference_fn inference_fn self._queue: deque deque() self._lock threading.Lock() self._running False # 统计指标 self.stats { total_batches: 0, total_requests: 0, avg_batch_size: 0.0, avg_wait_ms: 0.0, } def submit(self, request: InferenceRequest): 提交推理请求 with self._lock: self._queue.append(request) def run_loop(self): 批处理主循环在独立线程中运行 self._running True while self._running: batch self._collect_batch() if batch: self._execute_batch(batch) else: time.sleep(0.001) # 1ms 空闲等待 def _collect_batch(self) - List[InferenceRequest]: 收集一个 batch 逻辑 1. 如果队列深度 max_batch_size立即收集 2. 如果最早请求的 deadline 已过期立即收集 3. 否则等待 max_delay_ms 后收集 with self._lock: if not self._queue: return [] # 检查是否达到大小限制 if len(self._queue) self.max_batch_size: batch [] for _ in range(self.max_batch_size): batch.append(self._queue.popleft()) return batch # 检查最早请求是否已超 deadline now time.time() first_req self._queue[0] if now first_req.deadline: batch list(self._queue) self._queue.clear() return batch return [] # 没有足够的请求休眠后重试 def _execute_batch(self, batch: List[InferenceRequest]): 执行推理调用 if not self.inference_fn: return start_time time.time() # 统计等待时间 now time.time() for req in batch: wait_ms (now - req.arrival_time) * 1000 self._update_stats(wait_ms, len(batch)) try: # 执行推理 results self.inference_fn(batch) # 回调各请求 for req, result in zip(batch, results): if req.callback: req.callback(result) except Exception as e: # 推理失败时的错误处理 for req in batch: if req.callback: req.callback({error: str(e)}) def _update_stats(self, wait_ms: float, batch_size: int): n self.stats[total_batches] self.stats[total_batches] 1 self.stats[total_requests] batch_size self.stats[avg_batch_size] ( (self.stats[avg_batch_size] * n batch_size) / (n 1) ) self.stats[avg_wait_ms] ( (self.stats[avg_wait_ms] * n wait_ms) / (n 1) ) class ContinuousBatcher: 连续批处理器 策略在当前 batch 执行期间新请求可加入下一个 batch。 核心优势减少 GPU 空闲间隙。 核心限制KV Cache 显存占用。 def __init__( self, max_batch_tokens: int 8192, # 最大 Token 容量KV Cache 限制 max_sequence_length: int 2048, inference_fn: Callable None, ): self.max_batch_tokens max_batch_tokens self.max_sequence_length max_sequence_length self.inference_fn inference_fn self._pending: List[InferenceRequest] [] self._running_batch: Optional[List[InferenceRequest]] None self._lock threading.Lock() def submit(self, request: InferenceRequest): with self._lock: self._pending.append(request) def get_next_batch(self) - List[InferenceRequest]: 获取下一个要执行的 batch 设计决策优先从 pending 中取其次从 running 的未完成部分取。 同时预计算 KV Cache 的 Token 总量避免显存溢出。 with self._lock: batch [] total_tokens 0 # 优先处理 pending 请求 for req in list(self._pending): tokens req.prompt_tokens req.max_new_tokens if total_tokens tokens self.max_batch_tokens: break batch.append(req) self._pending.remove(req) total_tokens tokens return batch class GreedyBinPacker: 贪心打包调度器 策略按 prompt_tokens 对请求分组同组请求放入同一 batch。 避免长短请求混在一起导致的短板效应。 # Token 分组边界 BINS [0, 128, 512, 2048, 8192] def __init__(self, max_batch_size: int 16): self.max_batch_size max_batch_size self._bins {i: deque() for i in range(len(self.BINS) - 1)} self._lock threading.Lock() def _get_bin(self, tokens: int) - int: 根据 token 数分配桶 for i in range(len(self.BINS) - 1): if self.BINS[i] tokens self.BINS[i 1]: return i return len(self.BINS) - 2 # 最大桶 def submit(self, request: InferenceRequest): bin_idx self._get_bin(request.prompt_tokens) with self._lock: self._bins[bin_idx].append(request) def collect_batch(self) - List[InferenceRequest]: 收集一个 batch从非空桶中取满 max_batch_size with self._lock: # 按桶优先级收集先处理小请求桶延迟敏感 for bin_idx in sorted(self._bins.keys()): if self._bins[bin_idx]: batch [] while self._bins[bin_idx] and len(batch) self.max_batch_size: batch.append(self._bins[bin_idx].popleft()) return batch return [] # 自适应时间窗口 class AdaptiveBatcher(DynamicBatcher): 自适应批处理器 动态调整 max_delay_ms - 队列深度高时缩小窗口加速出队 - 队列深度低时扩大窗口等待更多请求凑 batch def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._current_delay self.max_delay_ms self._delay_min 1.0 self._delay_max 50.0 def _collect_batch(self) - List[InferenceRequest]: with self._lock: queue_len len(self._queue) # 自适应调整窗口 if queue_len self.max_batch_size * 3: # 队列积压严重最小化等待时间 self._current_delay self._delay_min elif queue_len self.max_batch_size: # 队列充实适中等待 self._current_delay self.max_delay_ms else: # 队列稀疏放宽等待时间 self._current_delay min(self._delay_max, self.max_delay_ms * 2) if queue_len self.max_batch_size: batch [] for _ in range(min(self.max_batch_size, queue_len)): batch.append(self._queue.popleft()) return batch return [] # 让主循环的 sleep 来控制节奏四、边界分析与架构权衡批处理的固有缺陷任何批处理都会引入额外的延迟。动态批处理的等待窗口、连续批处理的 batch 转换间隙。对于实时性要求极高的场景如实时语音交互延迟要求 100ms批处理的延迟可能不可接受。另外GPU 显存是批处理规模的硬上限——KV Cache 的显存占用随 batch size 线性增长当你把 batch size 从 1 提到 32 时显存占用可能是原来的 32 倍。贪心打包的适用性贪心打包在请求计算量差异大时效果最好。但如果大部分请求的计算量相似分桶的开销反而降低了 GPU 利用率。分桶的边界需要根据实际流量分布来校准。适用边界最适合离线批处理场景如对已有数据的批量推理。也适合高吞吐在线推理服务QPS 100单请求推理的 GPU 利用率太低。需要服务能够容忍毫秒级的额外延迟。禁用场景不适合实时性优先的场景如自动驾驶推理、高频交易。也不适合单个 GPU 同时服务大量不同模型的多租户场景——batch 内多个模型的 kernel launch 无法共享。五、总结推理批处理的核心 trade-off 是延迟与吞吐。动态批处理用时间窗口平衡两者连续批处理减少 GPU 空闲贪心打包避免短板效应。选择策略的关键是延迟敏感的场景用动态批处理 小窗口吞吐优先的场景用连续批处理请求规模差异大的场景用贪心打包。所有策略最终受限于 GPU 显存——KV Cache 决定了一个 batch 能装多少请求。
