LLM 推理调优方法：Batching 与 KV Cache 怎么做对照实验

LLM 推理调优方法Batching 与 KV Cache 怎么做对照实验Batching 和 KV Cache 是否有效取决于模型、请求长度与并发分布。调优时固定软硬件和数据集分别记录 TTFT、TPOT、吞吐、失败率与显存再用重复实验比较方案。1. 并发上升后 P99 延迟增加静态 Batching 通常等待 Batch 填满或超时后统一执行。当请求的输入、输出长度差异较大时短请求可能被长请求拖住形成队头阻塞。可以按长度分桶并记录各桶排队时间再与 Continuous Batching 对照。更糟糕的是显存中为每个 Request 预分配的 Key-Value CacheKV Cache由于采用连续物理内存分配产生了大量显存碎片。当并发升高时显存分配器无法找到足够大的连续空间存储新的 KV 矩阵直接触发了昂贵的 CPU-GPU KV 交换Swapping最终导致整条推理流水线产生严重停顿。2. 调度链路与显存耗尽静态 Batching 造成的 Head-of-Line 阻塞Prefill 阶段属于 Compute-bound计算密集型大矩阵乘法能够充分填满 Tensor Core 的流水线而 Decode 阶段则是典型的 Memory-bandwidth bound内存带宽密集型每一个生成的 Token 都需要把数十 GB 的模型权重和前面所有 Token 的 KV Cache 从显存重新加载到 SRAM 中。在静态 Batching 下一旦 Batch 内部某个超长 Request 没有结束整组请求占用的显存和计算资源就无法释放。这种现象被称为队头阻塞Head-of-Line Blocking。解决这一瓶颈的关键在于实现两项核心工程重构Continuous Batching连续批处理 / 迭代级调度不再在 Request 粒度上组装 Batch而是在每个 Decode Token 的 Step 粒度上动态调整 Batch 组装。生成完 EOS 的请求立即出队并释放资源队列中的新 Prefill 请求顺理成章地插入到当前 Step 的空余 Slot 中。PagedAttention页式注意力机制借鉴操作系统虚拟内存的分页思想将 KV Cache 切分为固定大小的 Block如 16 个 Token 一个 Block通过物理 Block 表进行动态映射。不再要求显存物理连续从根本上解决显存碎片化。下面是带有分支决策与降级防线的高并发推理调度决策链路通过这套逻辑调度器可以在每个毫秒级 Iteration 内精细化管理 GPU 的每一块显存和算力 Slot不再浪费任何一个 Decode 周期。3. 示例 Python/C 服务端请求调度与 KV Cache 智能管理实现在工程落地中仅靠理论是不够的。需要在 C / Python 服务端写出一个具备高吞吐、物理显存页表映射以及熔断降级能力的调度管理器。下面的 Python 示例示范了一个基于 PagedAttention 映射逻辑和 Continuous Batching 策略的线程安全调度器。代码中加入了确定性的显存越界防线、超时期限裁切以及 Token 生成异常的平滑退避处理import time import threading from typing import List, Dict, Optional from dataclasses import dataclass dataclass class Request: req_id: str prompt_tokens: List[int] max_new_tokens: int created_time: float timeout_ms: float 1500.0 generated_tokens: Optional[List[int]] None def __post_init__(self): if self.generated_tokens is None: self.generated_tokens [] class BlockAllocator: 模拟 PagedAttention 物理显存块分配器 def __init__(self, total_blocks: int, block_size: int 16): self.total_blocks total_blocks self.block_size block_size self.free_blocks set(range(total_blocks)) self.lock threading.Lock() def allocate(self, num_blocks: int) - List[int]: with self.lock: if len(self.free_blocks) num_blocks: raise MemoryError(GPU Block Allocator OOM: 物理显存页不足) allocated [] for _ in range(num_blocks): allocated.append(self.free_blocks.pop()) return allocated def free(self, block_ids: List[int]): with self.lock: for b_id in block_ids: self.free_blocks.add(b_id) class ContinuousBatchScheduler: def __init__(self, max_batch_size: int, allocator: BlockAllocator): self.max_batch_size max_batch_size self.allocator allocator self.waiting_queue: List[Request] [] self.running_batch: List[Request] [] self.kv_block_map: Dict[str, List[int]] {} self.queue_lock threading.Lock() def add_request(self, req: Request) - bool: 安全入队增加队列积压防护 with self.queue_lock: if len(self.waiting_queue) 500: # 高并发下触发过载熔断 return False self.waiting_queue.append(req) return True def step_schedule(self) - List[str]: 每个 Iteration 执行一次的连续批处理调度逻辑 now time.time() completed_req_ids [] with self.queue_lock: # 1. 检查运行中与等待队列的超时请求防止无意义算力浪费 self._prune_timeouts(now) # 2. 从等待队列补充新请求到 running_batch while len(self.running_batch) self.max_batch_size and self.waiting_queue: candidate self.waiting_queue[0] needed_blocks (len(candidate.prompt_tokens) self.allocator.block_size - 1) // self.allocator.block_size try: blocks self.allocator.allocate(needed_blocks) self.kv_block_map[candidate.req_id] blocks self.running_batch.append(self.waiting_queue.pop(0)) except MemoryError: # 显存块不足暂停调度新请求优先保证运行中任务 break # 3. 模拟 Decode 计算 Step finished_in_this_step [] for req in self.running_batch: # 模拟生成 1 个 Token fake_token 101 req.generated_tokens.append(fake_token) # 判断是否达到终止条件 if len(req.generated_tokens) req.max_new_tokens: finished_in_this_step.append(req) # 4. 回收已完成请求的显存 Block 资源 with self.queue_lock: for req in finished_in_this_step: self.running_batch.remove(req) blocks_to_free self.kv_block_map.pop(req.req_id, []) self.allocator.free(blocks_to_free) completed_req_ids.append(req.req_id) return completed_req_ids def _prune_timeouts(self, current_time: float): 超时降级防御防线 valid_waiting [] for req in self.waiting_queue: if (current_time - req.created_time) * 1000.0 req.timeout_ms: # 记录日志或指标放弃处理 continue valid_waiting.append(req) self.waiting_queue valid_waiting # 示例调度器部署验证 if __name__ __main__: allocator BlockAllocator(total_blocks1024, block_size16) scheduler ContinuousBatchScheduler(max_batch_size8, allocatorallocator) # 模拟写入 10 个请求 for i in range(10): req Request( req_idfreq_{i}, prompt_tokens[1, 2, 3, 4, 5], max_new_tokens5 (i % 3), created_timetime.time() ) success scheduler.add_request(req) assert success, 请求入队不应失败 # 执行 10 个迭代 Step for step in range(10): done scheduler.step_schedule() if done: print(fStep {step}: 完成请求 - {done})这段代码展示了连续批处理与 KV 页分配的协作方式但示例级计数器不能证明显存一定不溢出。接入真实引擎时还要处理 CUDA 分配失败、请求取消和页表回收并用 OOM 注入验证清理路径。4. 验证口径与记录方法使用vegeta与自定义 Python 脚本生成 Poisson 到达流量。并发阶梯从单实例容量基线逐步增加每一档的实际请求数都写入报告。压测收集到的核心数据指标如下表所示负载档位调度策略延迟吞吐GPU 显存失败与 OOM基线静态 Batching 连续 KV统计 P50/P99统计 Tokens/s采集占用与碎片统计失败类型基线Continuous PagedKV统计 P50/P99统计 Tokens/s采集占用与碎片统计失败类型中档静态 Batching 连续 KV统计 P50/P99统计 Tokens/s采集占用与碎片统计失败类型中档Continuous PagedKV统计 P50/P99统计 Tokens/s采集占用与碎片统计失败类型容量边界静态 Batching 连续 KV统计 P50/P99统计 Tokens/s采集占用与碎片记录是否 OOM容量边界Continuous PagedKV统计 P50/P99统计 Tokens/s采集占用与碎片记录是否 OOM5. 延迟与成本交汇点的工程防线降级策略与超时熔断在大模型推理系统调优的最终阶段单纯追求吞吐量和稳定延迟是不够的工程落地需要考虑算力成本与服务可接受度SLA之间的平衡。入口 Prompt 长度动态裁切根据当前队列积压深度自动调整系统可接受的最大输入 Token 数。当队列积压超过阈值时对非核心业务的大上下文请求自动执行尾部剪枝或拒绝处理。多级 KV Cache 缓存复用Prefix Caching对于包含固定 System Prompt 或长上下文 Agent 提示词的请求将前缀 Token 的 KV Cache 在 CUDA 显存中持久化共享。减少重复 Prefill 的计算耗时直接跳过计算密集阶段。把延迟、吞吐与成本的取舍写入可观测的控制回路并通过过载与恢复测试检查服务是否能按预期降级。收尾