AI推理的请求批处理：Dynamic Batching与Continuous Batching的工程实现

AI推理的请求批处理Dynamic Batching与Continuous Batching的工程实现一、推理批处理的核心矛盾大模型推理存在一个天然的矛盾GPU计算是并行的但用户请求是串行到达的。单请求推理时GPU中的大量计算单元处于闲置状态。以A100为例一个典型的7B模型单请求推理只能利用不到15%的TFLOPS。批处理Batching通过将多个请求合并为一个批次一起送入GPU计算大幅提升硬件利用率。但批处理也带来了新的问题延迟增加——请求必须等待批次凑齐才能执行。三种批处理策略在延迟-吞吐的Pareto前沿上占据了不同的位置graph LR subgraph 静态批处理 R1[请求1] -- W1[等待批次凑满] R2[请求2] -- W1 R3[请求3] -- W1 W1 -- B1[固定Batch Size8br/一起推理] end subgraph 动态批处理 R4[请求4] -- T1[超时窗口] R5[请求5] -- T1 T1 -- B2[可变Batch Sizebr/到期就推理] end subgraph 连续批处理 R6[请求6] -- S[调度器] R7[请求7] -- S S -- B3[迭代级调度br/完成的请求立即返回br/新请求立即加入] end二、三种批处理策略的技术原理2.1 静态批处理Static Batching最简单的批处理方式等待凑满固定大小的批次或等待超时然后一次性推理。客户端感受的延迟 等待时间 推理时间。class StaticBatcher: def __init__(self, model, batch_size: int, max_wait_ms: int): self.model model self.batch_size batch_size self.max_wait_ms max_wait_ms self.request_queue: List[Request] [] self.lock threading.Lock() def infer(self, request: Request) - Response: start_time time.time() promised threading.Event() result_holder [] with self.lock: self.request_queue.append((request, promised, result_holder)) if len(self.request_queue) self.batch_size: self._flush() # 等待批次执行完成或超时 elapsed (time.time() - start_time) * 1000 if not promised.wait(timeout(self.max_wait_ms - elapsed) / 1000): with self.lock: self._flush() promised.wait() return result_holder[0] def _flush(self): if not self.request_queue: return batch self.request_queue[:] self.request_queue.clear() inputs [r[0].prompt for r in batch] outputs self.model.generate(inputs) for (_, event, holder), output in zip(batch, outputs): holder.append(output) event.set()静态批处理的致命缺陷长请求拖累整个批次。即使批次中有一个请求输出2000个token其他7个请求只输出50个token也必须等最长的那个完成才能一起返回。2.2 动态批处理Dynamic Batching在静态批处理上增加灵活性不严格要求凑满批次而是设定一个超时窗口。窗口内收集到的请求合并为一批执行下一批重新收集。关键参数是max_batch_size和max_wait_ms的权衡class DynamicBatcher: def __init__(self, model, max_batch_size: int, max_wait_ms: int): self.model model self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.pending: List[Tuple[Request, Event, List]] [] self.lock threading.Lock() self.flush_event threading.Event() self._timer_thread threading.Thread(targetself._timer_loop, daemonTrue) self._timer_thread.start() def _timer_loop(self): 定时刷新线程 while True: time.sleep(self.max_wait_ms / 1000) self.flush_event.set() def infer(self, request: Request) - Response: event threading.Event() result [] with self.lock: self.pending.append((request, event, result)) if len(self.pending) self.max_batch_size: self._execute_batch() event.wait() return result[0]动态批处理改善了延迟但没有解决长请求拖累短请求的问题——同一批次内的请求仍然同步等待。2.3 连续批处理Continuous Batching连续批处理也叫Iteration-level Batching是目前最先进的方式。核心思想在每次迭代生成一个token后完成的请求立即返回新请求可以动态加入下一轮迭代。class ContinuousBatcher: def __init__(self, model, max_batch_size: int): self.model model self.max_batch_size max_batch_size self.active_requests: Dict[str, RequestState] {} self.pending_queue: List[Request] [] self.lock threading.Lock() self.scheduler_thread threading.Thread( targetself._scheduler_loop, daemonTrue ) self.scheduler_thread.start() class RequestState: def __init__(self, request: Request): self.request request self.generated_tokens: List[int] [] self.kv_cache None self.finished False self.result_event threading.Event() self.result: Optional[Response] None def _scheduler_loop(self): 每轮迭代调度 while True: with self.lock: # 从等待队列中补充请求 while (len(self.active_requests) self.max_batch_size and self.pending_queue): req self.pending_queue.pop(0) self.active_requests[req.id] self.RequestState(req) if not self.active_requests: time.sleep(0.001) continue # 构建当前批次的输入包含KV Cache batch_inputs [ state for state in self.active_requests.values() ] # 执行一次迭代生成一个token/请求 next_tokens self.model.generate_next_token(batch_inputs) with self.lock: for state, token in zip(batch_inputs, next_tokens): state.generated_tokens.append(token) if token EOS_TOKEN or len(state.generated_tokens) state.request.max_tokens: state.finished True state.result Response( tokensstate.generated_tokens, finish_reasonstop ) state.result_event.set() del self.active_requests[state.request.id]连续批处理的优势在于请求级别的完成通知——每个请求一旦生成结束就立即返回无需等待批次中的其他请求。三、批处理延迟的量化模型批处理的延迟由三个部分构成Total_Latency Queueing_Delay Prefill_Time Decode_Time × Num_Tokens其中Queueing_Delay等待批次凑齐的时间与负载正相关Prefill_Time处理输入prompt的时间一次性与prompt长度和batch_size正相关Decode_Time每个解码迭代的时间与batch_size和总KV Cache大小相关定量分析batch_sizePrefill (512 tokens)Decode/tokenP50延迟 (100 tokens)145ms8ms845ms485ms12ms1285ms8140ms18ms1940ms16230ms28ms3030ms32420ms45ms4920ms数据基于Llama-2-7B在A10G上的实测FP16精度。可以看到batch_size从1增加到4吞吐量提升约3.2倍但延迟只增加52%。但batch_size超过8后延迟增长开始超过吞吐收益。graph TD A[请求到达] -- B{当前活跃br/请求数 max_batch?} B --|是| C[加入活跃列表br/开始Prefill] B --|否| D[加入等待队列] C -- E[Prefill阶段br/处理prompt tokens] E -- F[进入Decode循环] D -- G[等待调度] G -- B F -- H{生成EOS或br/达到max_tokens?} H --|否| I[生成下一个token] I -- J{本次迭代br/所有请求都完成?} J --|否| F J --|是| K[开始下一轮迭代] K -- L[从等待队列补充请求] L -- F H --|是| M[该请求返回结果] M -- N{活跃列表为空?} N --|否| F N --|是| B四、与vLLM/TGI的集成方式4.1 vLLM集成vLLM的核心创新是PagedAttention将KV Cache以页为单位管理实现了近乎零碎片的显存管理from vllm import LLM, SamplingParams # vLLM默认启用Continuous Batching llm LLM( modelmeta-llama/Llama-2-7b-hf, max_num_batched_tokens4096, # 批次级最大token数 max_num_seqs32, # 最大并发序列数 gpu_memory_utilization0.90, ) sampling_params SamplingParams( temperature0.7, max_tokens256, ) # 异步批量请求 prompts [Hello, , What is AI?, Tell me about ] outputs llm.generate(prompts, sampling_params)vLLM内部自动执行continuous batching当某个请求生成完毕输出EOS其KV Cache页被释放新请求的KV Cache立即分配空闲页无缝衔接到下一轮迭代。4.2 TGI集成Hugging Face的Text Generation Inference通过Rust实现的调度器支持continuous batchingdocker run -p 8080:80 \ -e MAX_BATCH_PREFILL_TOKENS4096 \ -e MAX_TOTAL_TOKENS8192 \ -e MAX_CONCURRENT_REQUESTS32 \ ghcr.io/huggingface/text-generation-inference:latest \ --model-id meta-llama/Llama-2-7b-hfTGI与vLLM的核心差异特性vLLMTGIKV Cache管理PagedAttention静态预分配分块调度粒度迭代级迭代级水印策略无支持限制延迟峰值量化支持AWQ/GPTQGPTQ/EETQ/bitsandbytes4.3 批量大小动态调整生产环境推荐根据负载自动调整max_batch_sizeclass AdaptiveBatchController: def __init__(self, target_p99_ms: float 2000): self.target_p99 target_p99_ms self.current_max_batch 8 self.p99_history collections.deque(maxlen60) def adjust(self, current_p99_ms: float): self.p99_history.append(current_p99_ms) avg_p99 sum(self.p99_history) / len(self.p99_history) if avg_p99 self.target_p99 * 1.2: # 延迟超标减小batch size self.current_max_batch max(1, self.current_max_batch - 2) elif avg_p99 self.target_p99 * 0.7 and self.current_max_batch 32: # 延迟有余量增大batch size以提升吞吐 self.current_max_batch 2 return self.current_max_batch五、总结批处理是提升AI推理吞吐量的核心手段但策略的选择直接决定了延迟和资源利用率静态批处理适合离线批量推理场景如数据标注对延迟不敏感动态批处理是过渡方案通过超时窗口缓解等待延迟但仍有长请求拖累问题连续批处理是生产环境的最佳实践通过迭代级调度最大化GPU利用率同时最小化首token延迟工程选型建议在线服务API优先使用vLLM/TGI的continuous batchingP99延迟目标设2s以内离线批处理可以使用更大的batch_size32~64配合静态批处理追求极致吞吐混合场景按请求优先级分级高优请求走小batch低延迟通道低优请求走大batch高吞吐通道批处理参数的调优没有银弹需要基于实际负载的token长度分布和SLA要求进行持续的AB测试。