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

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

一、你的 vLLM 在凌晨 3 点还在满功率跑空推理，月账单多了两万

AI 推理服务的负载有极强的波谷特征：白天办公时段（9:00-18:00）QPS 高，夜间（0:00-6:00）QPS 趋近于 0。如果你的推理服务在低负载时段仍然保持跟峰值一样的 Batch Size 和 GPU 占用，大部分 GPU 时间是在空跑——不是在等请求（GPU 利用率 5%），就是在处理凑不满 batch 的小请求组合（吞吐极低）。

Batch Size 是推理服务最重要的吞吐杠杆。大 batch 提升吞吐（GPU 并行度高），但同时增加延迟（并发请求排队）。小 batch 降低延迟但浪费 GPU 算力。自适应 Batch Size 的核心思想是：根据实时 QPS 动态调整 batch size，在高负载时用大 batch 提吞吐，低负载时用小 batch 降延迟——甚至缩容到 0（sleep 模式）。

实现自适应 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(level=logging.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_client=None): 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(target=self._schedule_loop, daemon=True) 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(timeout=5) 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 --replicas=0 # 配合 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-gpu=utilization.gpu", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=5, ) 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_size=1, max_batch_size=64, default_batch_size=8, high_qps_threshold=50, low_qps_threshold=5, idle_timeout_minutes=30, p95_latency_warning_ms=2000, p95_latency_critical_ms=5000, ) 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 跳到 200，batch 从 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 多高，只关心他多快拿到回复。

企业官网建设流程全解析