一、引言1.1 大模型推理的吞吐量困境大语言模型LLM的推理服务面临一个核心矛盾低延迟 vs 高吞吐。当多个用户同时请求推理时传统的推理方式要么串行排队延迟高要么使用静态批处理浪费算力。DeepSeek-V3/R1 这类千亿参数模型在商用场景下推理吞吐量直接决定了服务成本。你有没有想过为什么 vLLM、TensorRT-LLM 这类推理引擎能做到比原始实现数倍甚至数十倍的吞吐量提升答案就在Continuous Batching连续批处理和PagedAttention分页注意力这两项核心技术中。本文将带你从零实现一个简化的 Continuous Batching 调度引擎深入理解其核心原理并用 Python 代码一步步搭建可运行的推理调度系统。1.2 什么是 Continuous Batching传统的 LLM 推理采用Static Batching静态批处理将多个请求收集成一个批次一起执行前向传播。但 LLM 的 token 生成是自回归的——每个请求的生成长度不同。静态批处理中短请求生成完毕后只能空闲等待长请求完成造成 GPU 利用率严重浪费。Continuous Batching的核心思想是在 token 级别动态调度。每当一个请求生成一个 token系统就检查是否可以插入新请求或移除已完成请求保持批次始终处于饱满状态。上图清晰地展示了两种方式的区别静态批处理中批次大小固定不变而在连续批处理中批次大小随时间动态调整最大化 GPU 利用率。1.3 为什么这很重要对于 DeepSeek-V3 这类大规模 MoE 模型推理时的计算瓶颈主要在注意力机制和 FFN 层。Continuous Batching 可以-提升吞吐量 2-5 倍GPU 计算单元不再空闲-降低平均延迟短请求不会被长请求拖累-更好的资源利用根据请求到达动态调整二、核心技术原理在动手编码之前我们需要先理解两个核心技术Continuous Batching 调度和 KV Cache 管理。2.1 自回归生成的计算特性LLM 推理分为两个阶段Prefill预填充阶段处理整个输入 prompt一次计算生成所有输入 token 的 KV Cache然后输出第一个 token。这个阶段是计算密集型compute-bound。Decode解码阶段逐 token 生成每一步都需读取所有历史 KV Cache计算注意力并写入新的 KV Cache。这个阶段是显存带宽密集型memory-bound。传统静态批处理的问题在于所有请求同时进入 Prefill 阶段又同时在 Decode 阶段进退。如果一个请求长度为 10另一个长度为 1000前者在完成 10 个 token 后就无事可做。2.2 Continuous Batching 调度策略Continuous Batching 由 2022 年的 ORCA 论文提出其核心思想是在每个解码步骤维护一个动态调度队列调度循环: 1. 收集所有活跃请求的上一步输出 token 2. 将新到达请求插入待处理队列 3. 从待处理队列取请求进入 Prefill 阶段 4. 将 Prefill 完成的请求加入 Decode 批次 5. 执行一次批量前向传播包含 Prefill Decode 混合 6. 将已生成完毕的请求移除 7. 重复步骤 12.3 KV Cache 管理挑战Continuous Batching 带来了一个关键问题KV Cache 的动态管理。每个请求的 KV Cache 大小取决于其 prompt 长度 已生成 token 数。动态调度意味着 KV Cache 的分配和释放必须高效。传统方法在 GPU 显存中预先分配固定大小的块但会出现显存碎片和浪费。PagedAttentionvLLM 的核心创新借鉴了操作系统虚拟内存的分页思想- 将 KV Cache 切分成固定大小的Block/Page- 通过Block Table实现逻辑地址到物理地址的映射- 按需分配减少碎片我们在本文的实现中会简化这部分用 Python 模拟 Block-based 的 KV Cache 管理器。2.4 显存带宽瓶颈详解要理解 Continuous Batching 为什么能大幅提升吞吐量需要先理解 Decode 阶段的显存带宽瓶颈。在 Decode 阶段Transformer 每步计算的核心操作是加载完整的 KV Cache大小 batch_size × seq_len × num_layers × 2 × d_head × num_heads和模型权重然后进行一次注意力计算和 FFN 计算。对于一个 70B 参数模型单次 Decode 需要从 HBM 加载约 140GB 的权重数据而 GPU HBM 带宽如 A100 为 2TB/s决定了这一步的耗时约为 70μs。当 batch_size 从 1 增加到 8 时权重加载量不变仍然 140GB但 KV Cache 加载量变为 8 倍。这意味着权重加载的开销被分摊到了更多请求上。这就是批处理提升吞吐量的根本原因——更高的计算/访存比。Continuous Batching 的优势在于它不像静态批处理那样等待所有请求同步而是让每个请求独立地进入和退出 Decode 阶段始终保持最大 batch_size从而让 GPU 的显存带宽始终处于饱和状态。2.5 与 MoE 模型的协同效应DeepSeek-V3 采用 MoEMixture of Experts架构这为 Continuous Batching 带来了额外的挑战和机遇。挑战MoE 模型的每个 token 只激活部分 Expert。不同请求可能激活不同 Expert导致 GPU 上的 Expert 负载不均衡。当 batch 中包含请求 A激活 Expert 1、3、5和请求 B激活 Expert 2、4、6时所有 Expert 都需要被加载计算效率下降。机遇Continuous Batching 可以通过调度策略优先将激活相同 Expert 的请求组成 sub-batch减少 Expert 切换开销。这正是 DeepSeek-V3 推理优化的核心方向之一。这种三、系统设计与架构3.1 整体架构我们的推理调度系统分为以下几个核心模块┌─────────────────────────────────────────┐ │ InferenceScheduler │ │ ┌──────────┐ ┌───────────┐ │ │ │ Request │ │ Batch │ │ │ │ Queue │──│ Manager │ │ │ └──────────┘ └───────────┘ │ │ │ │ │ │ ┌─────▼──────────────▼─────┐ │ │ │ Token Generator │ │ │ │ (模拟前向传播) │ │ │ └──────────────────────────┘ │ │ │ │ │ │ ┌─────▼──────────────▼─────┐ │ │ │ KV Cache Manager │ │ │ │ (Block-based 管理) │ │ │ └──────────────────────────┘ │ └─────────────────────────────────────────┘3.2 数据结构设计Request请求对象每个推理请求包含-request_id: 唯一标识-prompt_tokens: 输入 token ID 列表-generated_tokens: 已生成的 token ID 列表-max_tokens: 最大生成长度-state: 当前状态WAITING | PREFILL | DECODE | DONE-arrival_time: 到达时间-kv_cache_blocks: 已分配的 KV Cache block 列表Batch批次对象每个批次包含-batch_id: 唯一标识-requests: 当前批次中的请求列表-batch_size: 批次大小-max_seq_len: 当前批次最大序列长度3.3 调度算法我们实现两种调度策略以便对比FCFS先来先服务传统方式攒够批次大小后一起推理。Continuous Batching迭代级调度每步解码后重新评估批次组成def schedule(self): # 1. 处理新到达请求进入等待队列 new_requests self._poll_new_requests() self.waiting_queue.extend(new_requests) # 2. 从等待队列中取出请求进入 Prefill while self._can_add_to_batch(): req self.waiting_queue.pop(0) self._start_prefill(req) # 3. 检查已完成请求并移除 completed [r for r in self.active_requests if r.is_done()] for r in completed: self._free_kv_cache(r) self.active_requests.remove(r) # 4. 对活跃请求执行一步 Decode if self.active_requests: output self._forward(self.active_requests) self._update_generation(output)四、从零实现 Continuous Batching 引擎现在进入实战环节。我们将用 Python 实现一个完整的 Continuous Batching 调度引擎。为了清晰展示核心逻辑我们模拟 Transformer 前向传播聚焦调度逻辑本身。4.1 定义基础数据结构首先我们需要定义 Request 和 Batch 的数据结构from dataclasses import dataclass, field from typing import Optional, List, Dict import time import random from enum import Enum import numpy as np class RequestState(Enum): WAITING waiting # 排队等待中 PREFILL prefill # 预填充阶段 DECODE decode # 解码阶段 DONE done # 完成 dataclass class Request: request_id: str prompt_tokens: List[int] max_tokens: int arrival_time: float # 运行时状态 state: RequestState RequestState.WAITING generated_tokens: List[int] field(default_factorylist) num_prefill_tokens: int 0 kv_cache_blocks: List[int] field(default_factorylist) start_time: Optional[float] None end_time: Optional[float] None property def input_len(self) - int: return len(self.prompt_tokens) property def output_len(self) - int: return len(self.generated_tokens) property def total_len(self) - int: return self.input_len self.output_len def is_done(self) - bool: return self.state RequestState.DONE def get_elapsed(self) - float: if self.start_time is None: return 0.0 end self.end_time if self.end_time else time.time() return end - self.start_time def __repr__(self): return (fRequest(id{self.request_id}, fstate{self.state.value}, fprompt{self.input_len}, fgenerated{self.output_len}, fmax{self.max_tokens}))4.2 Block-based KV Cache 管理器接下来实现 KV Cache 管理器采用类似 PagedAttention 的分块策略dataclass class Block: KV Cache 中的一个 Block固定大小 block_id: int is_free: bool True request_id: Optional[str] None # 在实际实现中这里存储 K、V 张量 # 我们使用模拟数据表示 k_data: Optional[np.ndarray] None v_data: Optional[np.ndarray] None class BlockManager: 基于 Block 的 KV Cache 管理器 类似虚拟内存的分页管理 def __init__(self, num_blocks: int, block_size: int 16): self.num_blocks num_blocks self.block_size block_size # 每个 block 可存储的 token 数 self.blocks {i: Block(block_idi) for i in range(num_blocks)} self.free_blocks list(range(num_blocks)) self.allocated {} # request_id - [block_ids] def alloc(self, request_id: str, num_tokens: int) - List[int]: 为请求分配 KV Cache block 向上取整每个 block 可存储 block_size 个 token 的 K/V needed_blocks (num_tokens self.block_size - 1) // self.block_size if len(self.free_blocks) needed_blocks: raise MemoryError(f显存不足需要 {needed_blocks} blocks f空闲 {len(self.free_blocks)} blocks) allocated [] for _ in range(needed_blocks): block_id self.free_blocks.pop(0) self.blocks[block_id].is_free False self.blocks[block_id].request_id request_id allocated.append(block_id) self.allocated[request_id] allocated return allocated def append_block(self, request_id: str) - int: 为已有请求追加一个 block生成新 token 时 if not self.free_blocks: raise MemoryError(显存不足无法追加 block) block_id self.free_blocks.pop(0) self.blocks[block_id].is_free False self.blocks[block_id].request_id request_id if request_id in self.allocated: self.allocated[request_id].append(block_id) else: self.allocated[request_id] [block_id] return block_id def free(self, request_id: str): 释放请求所占用的所有 block if request_id not in self.allocated: return for block_id in self.allocated[request_id]: self.blocks[block_id].is_free True self.blocks[block_id].request_id None self.free_blocks.append(block_id) del self.allocated[request_id] def get_usage(self) - float: 返回显存使用率 used self.num_blocks - len(self.free_blocks) return used / self.num_blocks def is_allocated(self, request_id: str) - bool: return request_id in self.allocated4.3 模拟 Transformer 前向传播为了专注于调度逻辑我们用一个简化的模拟器替代真正的 Transformer 推理class SimulatedModel: 模拟 Transformer 前向传播 用计算时间模拟真实推理延迟 def __init__(self, prefill_time_per_token: float 0.001, # Prefill 每个 token 耗时 decode_time_per_token: float 0.005, # Decode 每个 token 耗时 batch_overhead: float 0.001): # 每增加一个请求的额外耗时 self.prefill_time_per_token prefill_time_per_token self.decode_time_per_token decode_time_per_token self.batch_overhead batch_overhead self.total_compute_time 0.0 self.num_prefill_steps 0 self.num_decode_steps 0 def forward(self, batch: List[Request]) - Dict[str, int]: 模拟一次批量前向传播 返回每个请求新生成的 token ID 在真实场景中这里会执行: 1. 拼接 batch 中所有请求的输入 2. 填充/掩码到相同长度 3. Transformer 前向传播 4. 采样得到新 token if not batch: return {} # 区分 Prefill 和 Decode 请求 prefill_reqs [r for r in batch if r.state RequestState.PREFILL] decode_reqs [r for r in batch if r.state RequestState.DECODE] wall_time 0.0 # Prefill 阶段计算 if prefill_reqs: total_prefill_tokens sum(r.num_prefill_tokens for r in prefill_reqs) prefill_time (total_prefill_tokens * self.prefill_time_per_token len(prefill_reqs) * self.batch_overhead) wall_time prefill_time self.num_prefill_steps 1 # Decode 阶段计算 if decode_reqs: decode_time (max(len(decode_reqs), 1) * self.decode_time_per_token len(decode_reqs) * self.batch_overhead * 0.5) wall_time decode_time self.num_decode_steps 1 # 模拟执行耗时 self._simulate_execution(wall_time) # 模拟输出生成随机 token ID outputs {} for req in batch: outputs[req.request_id] random.randint(0, 32000) return outputs def _simulate_execution(self, wall_time: float): 模拟执行耗时在真实系统中就是 GPU 计算时间 time.sleep(wall_time * 0.001) # 缩放以便快速演示 self.total_compute_time wall_time4.4 核心调度引擎这是整个系统的核心——Continuous Batching 调度器class ContinuousBatchingScheduler: Continuous Batching 调度引擎 在每个 token 生成步骤后动态调整批次 def __init__(self, model: SimulatedModel, block_manager: BlockManager, max_batch_size: int 8, max_waiting_time: float 1.0): self.model model self.block_manager block_manager self.max_batch_size max_batch_size self.max_waiting_time max_waiting_time # 请求队列 self.waiting_queue: List[Request] [] self.active_requests: List[Request] [] self.completed_requests: List[Request] [] # 统计 self.timestep 0 self.total_latency 0.0 def add_request(self, prompt_length: int, max_tokens: int) - str: 添加新推理请求 request_id freq-{len(self.completed_requests) len(self.active_requests) len(self.waiting_queue) 1} prompt_tokens [0] * prompt_length # 模拟输入 token req Request( request_idrequest_id, prompt_tokensprompt_tokens, max_tokensmax_tokens, arrival_timetime.time() ) self.waiting_queue.append(req) return request_id def step(self) - bool: 执行一步调度 推理 返回值是否有活跃请求 self.timestep 1 # 调度阶段 self._schedule() if not self.active_requests: return False # 推理阶段 outputs self.model.forward(self.active_requests) # 更新阶段 self._update(outputs) return True def _schedule(self): 调度决策 1. 将等待队列中的请求加入如果批次有空闲 2. 移除已完成请求 # 移除已完成请求 done_reqs [] for req in self.active_requests: if req.is_done(): done_reqs.append(req) self.completed_requests.append(req) req.end_time time.time() for req in done_reqs: self.block_manager.free(req.request_id) self.active_requests.remove(req) # 从等待队列取请求加入 while (len(self.active_requests) self.max_batch_size and self.waiting_queue): req self.waiting_queue[0] # 检查是否需要分配 KV Cache if not self.block_manager.is_allocated(req.request_id): try: self.block_manager.alloc( req.request_id, req.input_len ) except MemoryError: break # 显存不足停止添加 # 从等待队列移除加入活跃列表 self.waiting_queue.pop(0) req.state RequestState.PREFILL req.num_prefill_tokens req.input_len req.start_time time.time() self.active_requests.append(req) # # 等待超时策略如果请求等待太久强制加入 # if (time.time() - req.arrival_time self.max_waiting_time # and len(self.active_requests) self.max_batch_size): # continue # 将 Prefill 阶段完成后的请求切换到 Decode 状态 for req in self.active_requests: if req.state RequestState.PREFILL: req.state RequestState.DECODE def _update(self, outputs: Dict[str, int]): 根据模型输出更新请求状态 for req in self.active_requests: if req.state RequestState.DONE: continue new_token outputs.get(req.request_id, 0) req.generated_tokens.append(new_token) req.output_len # 检查是否生成完毕 if req.output_len req.max_tokens: req.state RequestState.DONE continue # 是否为 Prefill 完成后第一个 token已在上面完成切换 # 检查是否需要追加 KV Cache block total_tokens req.input_len req.output_len allocated_capacity len(self.block_manager.allocated.get( req.request_id, [])) * self.block_manager.block_size if total_tokens allocated_capacity: try: self.block_manager.append_block(req.request_id) except MemoryError: # 显存不足强制完成 req.state RequestState.DONE def get_stats(self) - Dict: 获取运行统计 avg_latency 0 if self.completed_requests: latencies [r.get_elapsed() for r in self.completed_requests] avg_latency sum(latencies) / len(latencies) return { total_steps: self.timestep, completed: len(self.completed_requests), active: len(self.active_requests), waiting: len(self.waiting_queue), avg_latency: avg_latency, gpu_usage: self.block_manager.get_usage(), compute_time: self.model.total_compute_time, } def run_until_complete(self, max_steps: int 10000): 运行直到所有请求完成 while self.active_requests or self.waiting_queue: has_active self.step() if not has_active and not self.waiting_queue: break if self.timestep max_steps: print(f [警告] 达到最大步数 {max_steps}) break4.5 传统静态批处理实现为了对比性能我们也实现一个传统 FCFS 静态批处理器class StaticBatchingScheduler: 传统静态批处理调度器 攒够 batch_size 个请求后一起推理 def __init__(self, model: SimulatedModel, block_manager: BlockManager, batch_size: int 4): self.model model self.block_manager block_manager self.batch_size batch_size self.waiting_queue: List[Request] [] self.completed_requests: List[Request] [] self.timestep 0 def add_request(self, prompt_length: int, max_tokens: int) - str: request_id (fstatic-req-{len(self.completed_requests) f{len(self.waiting_queue) 1}) prompt_tokens [0] * prompt_length req Request( request_idrequest_id, prompt_tokensprompt_tokens, max_tokensmax_tokens, arrival_timetime.time() ) self.waiting_queue.append(req) return request_id def run_until_complete(self, max_steps: int 100000): 执行所有请求 while self.waiting_queue: # 取出 batch_size 个请求 batch [] while len(batch) self.batch_size and self.waiting_queue: req self.waiting_queue.pop(0) req.state RequestState.PREFILL req.num_prefill_tokens req.input_len req.start_time time.time() self.block_manager.alloc(req.request_id, req.input_len) batch.append(req) if not batch: break # Prefill 阶段所有请求一起 self.model.forward(batch) for req in batch: req.state RequestState.DECODE # 生成第一个 token req.generated_tokens.append(random.randint(0, 32000)) # Decode 阶段所有请求同步进行 # 所有请求必须等到最长的那个完成 max_output max(req.max_tokens for req in batch) for step in range(max_output): self.timestep 1 # 只保留尚未完成的请求 active [r for r in batch if r.output_len r.max_tokens] if not active: break self.model.forward(active) for req in active: req.generated_tokens.append(random.randint(0, 32000)) # 管理 KV Cache total_tokens req.input_len req.output_len capacity (len(self.block_manager.allocated.get( req.request_id, [])) * self.block_manager.block_size) if total_tokens capacity: self.block_manager.append_block(req.request_id) if req.output_len req.max_tokens: req.state RequestState.DONE req.end_time time.time() self.completed_requests.append(req) self.block_manager.free(req.request_id) # 清理未通过循环完成的请求 for req in batch: if req.state ! RequestState.DONE: req.state RequestState.DONE req.end_time time.time() self.completed_requests.append(req) self.block_manager.free(req.request_id)五、性能对比与结果分析5.1 编写测试基准现在我们来编写测试代码对比两种调度策略def run_benchmark(use_continuous_batching: bool, num_requests: int 20, batch_size: int 4, max_batch_size: int 8, prompt_length_range(10, 100), max_tokens_range(20, 200)): 运行基准测试 参数: use_continuous_batching: 是否使用连续批处理 num_requests: 请求数量 batch_size: 静态批处理大小 max_batch_size: 连续批处理最大批次大小 # 创建共享组件 model SimulatedModel( prefill_time_per_token0.001, decode_time_per_token0.005, batch_overhead0.002 ) block_manager BlockManager(num_blocks1024, block_size16) # 创建调度器 if use_continuous_batching: scheduler ContinuousBatchingScheduler( modelmodel, block_managerblock_manager, max_batch_sizemax_batch_size ) mode Continuous Batching else: scheduler StaticBatchingScheduler( modelmodel, block_managerblock_manager, batch_sizebatch_size ) mode Static Batching # 生成随机请求 random.seed(42) requests_config [] for i in range(num_requests): prompt_len random.randint(*prompt_length_range) max_tok random.randint(*max_tokens_range) requests_config.append((prompt_len, max_tok)) # 记录开始时间 start_time time.time() # 添加请求 for prompt_len, max_tok in requests_config: scheduler.add_request(prompt_len, max_tok) # 执行推理 if use_continuous_batching: scheduler.run_until_complete() else: scheduler.run_until_complete() # 统计 total_time time.time() - start_time stats scheduler.get_stats() if use_continuous_batching else { completed: len(scheduler.completed_requests), compute_time: model.total_compute_time, } # 计算吞吐量 total_output_tokens sum( req.output_len for req in scheduler.completed_requests ) throughput total_output_tokens / total_time if total_time 0 else 0 print(f\n {mode} 性能报告 ) print(f 请求总数: {num_requests}) print(f 总耗时: {total_time:.3f}秒) print(f 实际计算时间: {model.total_compute_time:.3f}秒) print(f 总输出 Token: {total_output_tokens}) print(f 吞吐量: {throughput:.1f} tokens/秒) print(f 完成请求数: {stats.get(completed, 0)}) print(f 平均延迟: {stats.get(avg_latency, 0):.3f}秒) return { mode: mode, total_time: total_time, compute_time: model.total_compute_time, throughput: throughput, total_tokens: total_output_tokens, completed: stats.get(completed, 0), } # 运行对比 print( * 60) print(Continuous Batching vs Static Batching 性能对比) print( * 60) # 测试配置混合长度请求模拟真实场景 results [] for use_cb in [True, False]: result run_benchmark( use_continuous_batchinguse_cb, num_requests20, batch_size4, max_batch_size8, prompt_length_range(10, 100), max_tokens_range(20, 200) ) results.append(result) # 加速比 if len(results) 2: speedup results[1][total_time] / results[0][total_time] print(f\n{ * 60}) print(fContinuous Batching 加速比: {speedup:.2f}x) print(f(相比 Static Batching 总耗时减少 f{(1 - results[0][total_time] / results[1][total_time]) * 100:.1f}%))5.2 预期结果分析运行上述基准测试你将会观察到以下现象1. 吞吐量提升显著指标Static BatchingContinuous Batching提升总耗时基准降低 40-60%~2xToken 吞吐量基准提升 80-150%~2xGPU 利用率60-70%90-95%30%2. 短请求延迟大幅降低在静态批处理中一个 prompt10、max_tokens20 的短请求必须等待同批次中 prompt100、max_tokens200 的长请求完成后才能结束。而在 Continuous Batching 中短请求完成即可释放资源。3. 资源利用更加平滑Static Batching GPU 利用率: ████████████████░░░░░░░████████████████░░░░░░░... Continuous Batching GPU 利用率: ████████████████████████████████████████████...六、进阶优化技术6.1 调度策略深入我们在基础实现中使用了最简单的 FCFS 调度。实际生产环境如 vLLM、TensorRT-LLM还实现了更复杂的调度策略SJFShortest Job First优先调度预估完成时间短的请求。可以减少平均延迟但可能导致长请求饿死。基于 QoS 的优先级调度不同用户/请求有不同的服务质量要求。实时交互请求优先于批量处理请求。混合 Prefill/Decode 调度让 Prefill计算密集和 Decode显存带宽密集阶段同时进行充分利用 GPU 不同资源。class PriorityScheduler(ContinuousBatchingScheduler): 带优先级的调度器扩展 def _schedule(self): # 先按优先级排序假设 Request 有 priority 字段 self.waiting_queue.sort(keylambda r: (-r.priority, r.arrival_time)) super()._schedule()6.2 KV Cache 优化Prefix Caching前缀缓存如果多个请求共享公共前缀如系统 prompt可以共享同一段 KV Cache。DeepSeek 的 MLAMulti-head Latent Attention设计就从架构层面优化了 KV Cache 效率。KV Cache 量化将 KV Cache 从 FP16 压缩到 INT4/INT8可在同样显存下支持更长的上下文。我们在 6 月 3 日的文章「从零实现 KV Cache 量化推理引擎」中有详细实现。Memory Management- 预留一部分显存作为紧急缓冲区- 实现基于 LRU 的 block 回收策略- 使用异步内存传输pinned memory CUDA stream6.3 与 Prefix Caching 的结合Continuous Batching 与 Prefix Caching 是天作之合请求1: [System Prompt] [用户问题A] [生成回答A] 请求2: [System Prompt] [用户问题B] [生成回答B] ^ └── 共享相同的 System Prompt KV Cache实时实现中Block Manager 会维护一个全局的 Hash Table以 token 序列的 hash 为 keyblock ID 为 value。新请求进入时先匹配前缀缓存。6.5 处理长上下文当请求上下文达到 128K tokens 时KV Cache 大小成为瓶颈。以 DeepSeek-V3 为例其采用了 MLAMulti-head Latent Attention架构大幅压缩了 KV Cache 的体积。即使如此在 128K 上下文下KV Cache 依然可达数 GB。常见的优化策略窗口注意力Window Attention只保留最后 N 个 token 的 KV Cache丢弃历史信息滑动窗口 全局 Token类似 LongChat 和 Mistral 的方案部分 tokenclass WindowedBlockManager(BlockManager): 带滑动窗口的 Block 管理器 def __init__(self, num_blocks: int, block_size: int 16, window_size: int 4096): super().__init__(num_blocks, block_size) self.window_size window_size def append_block(self, request_id: str) - int: total_tokens ... # 计算当前总 token 数 if total_tokens self.window_size: # 淘汰最早的 block oldest self.allocated[request_id].pop(0) self.blocks[oldest].is_free True self.blocks[oldest].request_id None self.free_blocks.append(oldest) return super().append_block(request_id)七、总结与展望7.1 核心要点回顾本文从零实现了一个 Continuous Batching 推理引擎核心要点包括Continuous Batching 解决了 LLM 推理中的木桶效应不让短请求被长请求拖累Block-based KV Cache 管理是实现动态批处理的基础调度策略直接影响服务质量FCFS 简单可靠优先级调度更灵活吞吐量提升通常在 2-5x具体取决于请求长度分布的多样性7.2 与生产引擎的差距我们的实现做了大量简化真实生产引擎vLLM、TensorRT-LLM、SGLang还有以下复杂性CUDA Kernel 优化FlashAttention、PagedAttention 的自定义 CUDA kernel多 GPU 张量并行模型分片到多卡流水线并行不同层分配到不同 GPU动态 Batching 的显存管理内存池、碎片整理请求级别的超时和取消防止资源无限占用7.3 生产部署实践建议如果要在生产环境中部署 Continuous Batching 推理引擎以下几点经验值得注意显存预算规划建议为 KV Cache 预留 30-40% 的 GPU 显存其余用于模型权重和激活值。可以通过max_num_seqs和max_seq_len_to_capture参数控制最大并发请求数和上下文长度。监控指标重点关注以下指标——调度延迟决定首 token 响应时间、批次利用率实际 batch_size / 最大 batch_size、KV Cache 使用率和 OOM 事件数。这些指标直接反映推理服务的健康状态。负载均衡策略在多 GPU 部署中建议使用 Consistent Hashing 将请求分配到不同的 GPU同时为每个 GPU 独立运行调度器实例避免跨 GPU 的调度竞争。降级策略当负载超过阈值时应当有明确的降级方案——例如将非实时请求降级到静态批处理模式或拒绝新请求直到负载恢复正常。7.4 未来趋势LLM 推理引擎的未来方向包括Speculative Decoding推测解码用小模型起草大模型验证已在 5 月 28 日的文章中详细讲解Disaggrated Inference分离式推理将 Prefill 和 Decode 阶段部署到不同 GPU各自优化In-flight Batching在 GPU 计算过程中动态插入新请求MoE 专用调度针对 DeepSeek-V3 这类 MoE 模型的 Expert 级别的负载均衡延伸阅读如果你对 DeepSeek 推理优化更感兴趣推荐阅读 DeepSeek 实战指南掌握这些实用技巧让你的开发效率翻倍其中详细介绍了 DeepSeek API 调用、上下文窗口优化和成本控制等实用技巧。希望本文能帮助你深入理解现代 LLM 推理引擎的核心原理。动手运行代码观察不同请求分布下的性能表现才能真正掌握这些技术。好文章从技术深度开始。✍️