模型推理的自适应 Batch Size：根据负载动态调整吞吐

模型推理的自适应 Batch Size根据负载动态调整吞吐一、你的 vLLM 在凌晨 3 点还在满功率跑空推理月账单多了两万AI 推理服务的负载有极强的波谷特征白天办公时段9:00-18:00QPS 高夜间0:00-6:00QPS 趋近于 0。如果你的推理服务在低负载时段仍然保持跟峰值一样的 Batch Size 和 GPU 占用大部分 GPU 时间是在空跑——不是在等请求GPU 利用率 5%就是在处理凑不满 batch 的小请求组合吞吐极低。Batch Size 是推理服务最重要的吞吐杠杆。大 batch 提升吞吐GPU 并行度高但同时增加延迟并发请求排队。小 batch 降低延迟但浪费 GPU 算力。自适应 Batch Size 的核心思想是根据实时 QPS 动态调整 batch size在高负载时用大 batch 提吞吐低负载时用小 batch 降延迟——甚至缩容到 0sleep 模式。实现自适应 batch 需要两个组件负载感知器怎么知道现在忙不忙和调度策略知道了之后怎么调整。二、底层机制与原理剖析自适应 Batch Size 的四个状态和对应策略高负载状态QPS 阈值上限增大 batch size。GPU 利用率 80%请求队列在增长说明需要更高吞吐。增大 batch 让 GPU 单次处理更多请求。但要注意延迟往上走的趋势——batch 不是越大越好。低负载状态QPS 阈值下限减小 batch size。GPU 利用率 30%说明请求太少填不满 GPU。减小 batch 降低单个请求的等待时间不用等凑齐大 batch。如果长时间30 分钟低负载进入缩容模式。空闲状态QPS ≈ 0缩容到 0。释放 GPU 资源。有新的请求到来时通过 KEDA 或 Custom Metrics HPA 自动扩容恢复。冷启动延迟3-5 秒加载模型在这个场景是可接受的——因为全部请求都在 30 分钟的间隔之后到来多等 5 秒用户几乎感觉不到。延迟飙升P95 预警线不管当前负载如何立刻降低 batch size。优先保证用户体验延迟暂时牺牲吞吐。三、生产级代码实现 自适应 Batch Size 调度器 策略四个维度的指标QPS、队列深度、GPU 利用率、P95 延迟 驱动 batch size 的上下调整 import time import threading import logging import statistics from typing import Optional, Dict, List from dataclasses import dataclass, field from enum import Enum from collections import deque logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class LoadState(Enum): HIGH high # 高负载 → 增大 batch NORMAL normal # 正常 → 不变 LOW low # 低负载 → 减小 batch IDLE idle # 空闲 → 缩容 CRITICAL critical # 延迟过高 → 立刻降 batch dataclass class AdaptiveBatchConfig: 自适应 Batch 调度配置 # Batch 范围 min_batch_size: int 1 max_batch_size: int 64 default_batch_size: int 8 current_batch_size: int 8 # QPS 阈值基于 GPU 能力和模型复杂度设定 high_qps_threshold: int 50 # QPS 50 → 高负载 low_qps_threshold: int 5 # QPS 5 → 低负载 idle_qps_threshold: int 0 # QPS 0 → 空闲 idle_timeout_minutes: int 30 # 空闲 30 分钟 → 缩容 # 延迟阈值 p95_latency_warning_ms: float 2000 # P95 2s → 预警 p95_latency_critical_ms: float 5000 # P95 5s → 立刻降 batch # 调整参数 scale_up_factor: float 2.0 # 扩容时 batch 翻倍 scale_down_factor: float 0.5 # 缩容时 batch 减半 cooldown_seconds: int 60 # 两次调整之间的最小间隔防止抖动 # 滑动窗口 metrics_window_seconds: int 60 # 用于计算 QPS/延迟的时间窗口 # GPU 利用率阈值 gpu_high_threshold: float 0.80 # GPU 80% → 可能需要增大 batch gpu_low_threshold: float 0.30 # GPU 30% → 可能需要减小 batch class AdaptiveBatchScheduler: 自适应 Batch 调度器 运行在独立线程中周期性地检查系统状态并调整 batch size def __init__(self, config: AdaptiveBatchConfig, vllm_clientNone): self.config config self.vllm_client vllm_client # vLLM API client用于动态调整参数 # 滑动窗口数据 self._request_times: deque deque() # 记录每个请求的时间戳 self._latency_records: deque deque() # 记录每个请求的延迟 self._lock threading.Lock() # 状态追踪 self.last_adjustment_time: float 0 self.idle_since: Optional[float] None self.current_state: LoadState LoadState.NORMAL # 控制标志 self._running False self._thread: Optional[threading.Thread] None def start(self): 启动调度器 self._running True self._thread threading.Thread(targetself._schedule_loop, daemonTrue) self._thread.start() logger.info(AdaptiveBatchScheduler started (batch%d), self.config.current_batch_size) def stop(self): self._running False if self._thread: self._thread.join(timeout5) def record_request(self, latency_ms: float): 记录一次请求含延迟 with self._lock: now time.time() self._request_times.append(now) self._latency_records.append(latency_ms) def _schedule_loop(self): 主调度循环每 15 秒执行一次 while self._running: time.sleep(15) try: self._evaluate_and_adjust() except Exception as e: logger.error(Scheduler evaluation failed: %s, e) def _evaluate_and_adjust(self): 评估当前负载并决定 batch size 调整 now time.time() # 如果还在冷却期跳过 if now - self.last_adjustment_time self.config.cooldown_seconds: return # 清理过期数据 self._cleanup_expired(now) # 计算当前指标 qps self._compute_qps() p95_latency self._compute_p95_latency() gpu_util self._get_gpu_utilization() queue_depth self._get_queue_depth() # 状态判定 new_state self._determine_state(qps, p95_latency, gpu_util) if new_state self.current_state: # 状态没变但如果是空闲状态需要检查是否超时 if new_state LoadState.IDLE: if self.idle_since and (now - self.idle_since) self.config.idle_timeout_minutes * 60: self._trigger_scale_to_zero() return # 状态发生变化 → 执行调整 logger.info(State transition: %s → %s (QPS%.1f, P95%.0fms, GPU%.0f%%), self.current_state.value, new_state.value, qps, p95_latency, gpu_util * 100 if gpu_util else -1) self._execute_adjustment(new_state, qps, p95_latency) self.current_state new_state self.last_adjustment_time now def _determine_state(self, qps: float, p95_latency: float, gpu_util: Optional[float]) - LoadState: 负载状态判定优先级 1. CRITICAL延迟过高最高优先级 2. IDLE完全无流量 3. LOW / HIGH / NORMAL # 延迟过高 → 不论任何状态优先降 batch if p95_latency self.config.p95_latency_critical_ms: return LoadState.CRITICAL # 空闲 if qps self.config.idle_qps_threshold: if self.idle_since is None: self.idle_since time.time() return LoadState.IDLE else: self.idle_since None # 低负载 if qps self.config.low_qps_threshold: return LoadState.LOW # 高负载 if qps self.config.high_qps_threshold: return LoadState.HIGH # P95 预警——虽然不是 critical但值得关注 if p95_latency self.config.p95_latency_warning_ms: return LoadState.CRITICAL # 预警也走降 batch 逻辑 return LoadState.NORMAL def _execute_adjustment(self, new_state: LoadState, qps: float, latency: float): 执行 batch size 调整 old_batch self.config.current_batch_size if new_state LoadState.CRITICAL: # 延迟高 → 立刻降 batch new_batch max( self.config.min_batch_size, int(self.config.current_batch_size * self.config.scale_down_factor) ) logger.warning(P95 latency %.0fms %.0fms, reducing batch %d→%d, latency, self.config.p95_latency_warning_ms, old_batch, new_batch) elif new_state LoadState.HIGH: # 高负载 → 增大 batch new_batch min( self.config.max_batch_size, int(self.config.current_batch_size * self.config.scale_up_factor) ) elif new_state LoadState.LOW: # 低负载 → 减小 batch new_batch max( self.config.min_batch_size, int(self.config.current_batch_size * self.config.scale_down_factor) ) elif new_state LoadState.NORMAL: new_batch self.config.current_batch_size else: return if new_batch ! old_batch: self._apply_batch_size(new_batch) def _apply_batch_size(self, new_batch: int): 将新的 batch size 应用到推理引擎 self.config.current_batch_size new_batch logger.info(Batch size adjusted: %d, new_batch) # 生产环境通过 vLLM API 动态调整 # 实际 API 取决于推理引擎vLLM/TGI/TensorRT-LLM # 这里记录日志作为示例 if self.vllm_client: try: # vLLM 不支持运行时改 batch但可以改 max_num_seqs # 一些推理引擎支持通过 HTTP API 调整配置 pass except Exception as e: logger.error(Failed to update vLLM config: %s, e) def _trigger_scale_to_zero(self): 触发缩容到 0 logger.warning(Idle for %d minutes, triggering scale to zero, self.config.idle_timeout_minutes) # 生产环境通过 K8s API 缩减 Deployment replicas 到 0 # kubectl scale deployment vllm-service --replicas0 # 配合 KEDA ScaledJob 在下一个请求到来时自动扩容 def _compute_qps(self) - float: 计算滑动窗口内的 QPS with self._lock: now time.time() cutoff now - self.config.metrics_window_seconds recent [t for t in self._request_times if t cutoff] if len(recent) 2: return 0 return len(recent) / (max(recent) - min(recent)) if max(recent) ! min(recent) else len(recent) / 0.001 def _compute_p95_latency(self) - float: 计算滑动窗口的 P95 延迟 with self._lock: if len(self._latency_records) 20: return 0 sorted_latency sorted(self._latency_records) p95_idx int(len(sorted_latency) * 0.95) return sorted_latency[p95_idx] def _get_gpu_utilization(self) - Optional[float]: 获取 GPU 利用率 # 生产环境通过 nvidia-smi 或 DCGM 获取 try: import subprocess result subprocess.run( [nvidia-smi, --query-gpuutilization.gpu, --formatcsv,noheader,nounits], capture_outputTrue, textTrue, timeout5, ) if result.returncode 0: return float(result.stdout.strip()) / 100.0 except Exception: pass return None def _get_queue_depth(self) - int: 获取请求队列深度 # 通过 vLLM API 获取当前排队请求数 return 0 def _cleanup_expired(self, now: float): 清理滑动窗口中的过期数据 with self._lock: cutoff now - self.config.metrics_window_seconds * 2 while self._request_times and self._request_times[0] cutoff: self._request_times.popleft() if self._latency_records: self._latency_records.popleft() # --------------------------------------------------------------------------- # 启动示例 # --------------------------------------------------------------------------- if __name__ __main__: config AdaptiveBatchConfig( min_batch_size1, max_batch_size64, default_batch_size8, high_qps_threshold50, low_qps_threshold5, idle_timeout_minutes30, p95_latency_warning_ms2000, p95_latency_critical_ms5000, ) scheduler AdaptiveBatchScheduler(config) scheduler.start() # 模拟请求通常由 API Gateway 在请求进入时调用 record_request import random try: while True: latency random.gauss(500, 200) # 模拟延迟 scheduler.record_request(max(0, latency)) time.sleep(0.1) except KeyboardInterrupt: scheduler.stop()四、边界分析与架构权衡调节速率问题如果 QPS 突然从 0 跳到 200batch 从 1 翻倍到 2 → 4 → 8 → 16 → 32 需要经过 4 个调度周期60 秒冷却期太慢优化检测到负载跳变时跳过冷却期直接跳到对应 batch size查表缩容到 0 的冷启动问题GPU 推理服务从 0 扩容到 1 需要 30-60 秒模型加载 GPU 初始化如果用户突然在这个窗口内发起请求体验很差优化Keep-Warm 策略——在业务低峰期不缩容到 0而是保留 1 个实例Batch Size ≠ 并发度增大 batch 不是无限的——受 GPU 显存限制。Llama-3-70B 在 A100 80G 上最大 batch 可能只有 16-32当 batch 达到显存上限时更高的负载需要水平扩容加 GPU 节点而非垂直扩 batch五、结语自适应 Batch Size 调度本质是用实时负载指标驱动推理服务的吞吐和延迟平衡。在高 QPS 时增大 batch 提吞吐低 QPS 时减小 batch 降延迟空闲 30 分钟后缩容到 0 省成本。关键是四个指标的权重QPS 和队列深度反映有多忙P95 延迟反映用户感受到的慢GPU 利用率反映硬件用得多满。延迟保护必须是最优先级的——用户不关心你 GPU 多高只关心他多快拿到回复。