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。窗口太短（如 1ms），batch 太小，吞吐量上不去。窗口太长（如 100ms），P99 延迟会被拉高。建议用自适应窗口——根据当前队列深度动态调整。
连续批处理的 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(order=True)
class InferenceRequest:
"""推理请求"""
priority: int = 0 # 优先级（越小越优先）
arrival_time: float = field(default_factory=time.time, compare=False)
prompt_tokens: int = 0 # Prompt Token 数（估算计算量）
max_new_tokens: int = 256
callback: Optional[Callable] = field(default=None, compare=False)
metadata: dict = field(default_factory=dict, compare=False)
# 延迟约束
max_wait_ms: float = 100.0 # 最大等待延迟
deadline: float = field(init=False, compare=False)
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 能装多少请求。
