当前位置：首页 > news >正文

云原生 LLM 推理服务部署：从模型加载到请求调度的全链路优化

news 2026/6/9 20:56:45

云原生 LLM 推理服务部署：从模型加载到请求调度的全链路优化

一、LLM 推理的"延迟焦虑"：首 Token 要等 5 秒，用户早走了

LLM 推理服务部署到生产环境后，最常见的性能问题是首 Token 延迟（TTFT）过高。模型加载需要 10-30 秒，首次推理需要 2-5 秒预热，用户等不了这么久。更棘手的是，LLM 推理是计算密集型+内存密集型——KV Cache 占用大量 GPU 内存，并发请求时内存不够用，被迫排队等待。

全链路优化的核心是"从模型加载到请求调度，每个环节都减少延迟"。模型预热减少首次推理延迟，KV Cache 管理减少内存占用，连续批处理（Continuous Batching）提高吞吐量，请求调度优先级保证高优请求先处理。

二、全链路优化架构

graph TB subgraph 模型加载优化 A[模型分片加载<br/>Tensor Parallel] --> B[预热推理<br/>首次请求无冷启动] B --> C[模型缓存<br/>内存映射加载] end subgraph 推理优化 C --> D[Continuous Batching<br/>动态组批] D --> E[KV Cache 管理<br/>PagedAttention] E --> F[流式输出<br/>Token-by-Token] end subgraph 调度优化 F --> G[优先级队列<br/>高优请求先处理] G --> H[负载均衡<br/>最少连接路由] H --> I[弹性扩缩容<br/>GPU 感知 HPA] end

优化分三层：模型加载（分片+预热+缓存）、推理执行（连续批处理+KV Cache+流式输出）、请求调度（优先级+负载均衡+弹性扩缩）。每层优化独立，组合效果叠加。

三、实现

3.1 模型预热与缓存

import time from typing import Optional class ModelWarmup: """模型预热：消除首次推理延迟""" def __init__(self, model_loader): self.model_loader = model_loader self.model = None self.warmup_prompt = "Hello, this is a warmup request." def warmup(self, max_retries: int = 3) -> dict: """执行预热推理""" start = time.time() for attempt in range(max_retries): try: if self.model is None: self.model = self.model_loader.load() # 执行一次短推理，触发所有懒初始化 _ = self.model.generate( self.warmup_prompt, max_tokens=10 ) elapsed = time.time() - start return { 'status': 'success', 'warmup_time': f'{elapsed:.2f}s', 'attempt': attempt + 1, } except Exception as e: if attempt == max_retries - 1: return { 'status': 'failed', 'error': str(e), 'attempts': max_retries, } time.sleep(1) return {'status': 'failed', 'error': 'max retries exceeded'} class ModelCache: """模型缓存：使用内存映射加速加载""" def __init__(self, cache_dir: str = "/tmp/model_cache"): self.cache_dir = cache_dir self.loaded_models = {} def get_or_load( self, model_name: str, loader_fn ) -> object: """获取缓存的模型或加载新模型""" if model_name in self.loaded_models: return self.loaded_models[model_name] model = loader_fn(model_name) self.loaded_models[model_name] = model return model def evict(self, model_name: str) -> bool: """淘汰模型释放 GPU 内存""" if model_name in self.loaded_models: del self.loaded_models[model_name] # 触发 GPU 内存回收 import torch if torch.cuda.is_available(): torch.cuda.empty_cache() return True return False

3.2 连续批处理调度器

from dataclasses import dataclass from typing import List, Optional from collections import deque @dataclass class InferenceRequest: """推理请求""" request_id: str prompt: str max_tokens: int = 512 priority: int = 0 # 0=最高 created_at: float = 0.0 class ContinuousBatchScheduler: """连续批处理调度器""" def __init__(self, max_batch_size: int = 8): self.max_batch_size = max_batch_size self.pending: List[InferenceRequest] = [] self.running: List[InferenceRequest] = [] def add_request(self, request: InferenceRequest) -> None: """添加推理请求""" import time request.created_at = time.time() self.pending.append(request) # 按优先级排序（优先级数字越小越优先） self.pending.sort(key=lambda r: (r.priority, r.created_at)) def schedule_batch(self) -> List[InferenceRequest]: """调度下一批推理请求""" available_slots = self.max_batch_size - len(self.running) if available_slots <= 0 or not self.pending: return [] batch = self.pending[:available_slots] self.pending = self.pending[available_slots:] self.running.extend(batch) return batch def complete_request( self, request_id: str ) -> Optional[InferenceRequest]: """标记请求完成""" for i, req in enumerate(self.running): if req.request_id == request_id: return self.running.pop(i) return None def get_queue_depth(self) -> int: """获取队列深度""" return len(self.pending) def get_estimated_wait( self, priority: int = 0 ) -> float: """估算等待时间""" # 同优先级的前面有多少请求 ahead = sum( 1 for r in self.pending if r.priority <= priority ) # 每批处理 max_batch_size 个，每批约 2 秒 batches = (ahead + self.max_batch_size - 1) // self.max_batch_size return batches * 2.0 # 秒

3.3 KV Cache 管理

class KVCacheManager: """KV Cache 管理器：PagedAttention 简化实现""" def __init__( self, total_memory_gb: float = 40, page_size: int = 16, # 每个 page 的 token 数 page_memory_mb: float = 0.5, # 每个 page 的内存 ): self.page_size = page_size self.page_memory_mb = page_memory_mb total_pages = int( total_memory_gb * 1024 / page_memory_mb ) # 空闲页面池 self.free_pages = list(range(total_pages)) # 每个请求占用的页面 self.request_pages = {} def allocate( self, request_id: str, num_tokens: int ) -> List[int]: """为请求分配 KV Cache 页面""" num_pages = (num_tokens + self.page_size - 1) // self.page_size if len(self.free_pages) < num_pages: # 内存不足，尝试抢占低优先级请求的页面 freed = self._preempt_low_priority(num_pages - len(self.free_pages)) if len(self.free_pages) < num_pages: raise MemoryError( f"KV Cache 不足: 需要 {num_pages} 页, " f"可用 {len(self.free_pages)} 页" ) pages = self.free_pages[:num_pages] self.free_pages = self.free_pages[num_pages:] self.request_pages[request_id] = pages return pages def release(self, request_id: str) -> int: """释放请求的 KV Cache 页面""" if request_id not in self.request_pages: return 0 pages = self.request_pages.pop(request_id) self.free_pages.extend(pages) return len(pages) def _preempt_low_priority( self, num_needed: int ) -> int: """抢占低优先级请求的页面""" freed = 0 # 按 token 数降序抢占（大请求优先释放） sorted_requests = sorted( self.request_pages.items(), key=lambda x: len(x[1]), reverse=True, ) for req_id, pages in sorted_requests: if freed >= num_needed: break self.free_pages.extend(pages) del self.request_pages[req_id] freed += len(pages) return freed def get_utilization(self) -> float: """获取 KV Cache 利用率""" total = len(self.free_pages) + sum( len(p) for p in self.request_pages.values() ) used = sum(len(p) for p in self.request_pages.values()) return used / total if total > 0 else 0.0