大模型推理服务的请求队列与优先级调度从先到先得到智能排序推理资源的效率最大化一、推理服务的排队困境GPU 资源的稀缺与请求的洪峰大模型推理服务面临一个核心矛盾GPU 资源昂贵且有限但推理请求的到达模式是突发性的。在流量高峰期请求队列迅速积压所有请求按 FIFO 顺序处理导致高优先级请求如付费用户、实时对话与低优先级请求如批量处理、离线分析同等对待。付费用户等待 10 秒才收到响应而批量任务占用了大量 GPU 时间。更深层的问题是不同请求的价值和紧急度不同。一个实时对话请求如果延迟超过 2 秒用户体验就会严重下降而一个文档摘要请求延迟 30 秒也是可接受的。FIFO 调度无法区分这些差异导致 GPU 资源在低价值请求上浪费高价值请求却得不到及时处理。二、优先级调度的架构与队列模型flowchart TD A[推理请求入口] -- B[请求分类器] B -- B1[实时对话: P0 最高优先级] B -- B2[API 调用: P1 高优先级] B -- B3[批量任务: P2 低优先级] B -- B4[离线分析: P3 最低优先级] B1 -- C[多级优先级队列] B2 -- C B3 -- C B4 -- C C -- D[调度器] D -- D1[优先级抢占: P0 可抢占 P2/P3] D -- D2[权重分配: P1:P2:P3 5:3:2] D -- D3[饥饿防护: 低优先级保底配额] D -- E[GPU 推理引擎] E -- F[结果返回]2.1 多级优先级队列# priority_scheduler.py — 推理服务优先级调度器 # 设计意图实现多级优先级队列支持优先级抢占和饥饿防护 # 确保 GPU 资源优先服务于高价值请求 import asyncio import time from dataclasses import dataclass, field from enum import IntEnum from typing import Any, Callable, Awaitable class Priority(IntEnum): REALTIME 0 # 实时对话延迟敏感 API_CALL 1 # API 调用中等延迟 BATCH 2 # 批量任务延迟不敏感 OFFLINE 3 # 离线分析可长时间等待 dataclass(orderTrue) class InferenceRequest: priority: int field(compareTrue) enqueue_time: float field(compareTrue) request_id: str field(compareFalse) prompt: str field(compareFalse) max_tokens: int field(compareFalse) user_tier: str field(compareFalse) callback: Callable[[Any], Awaitable[None]] field(compareFalse, defaultNone) class PriorityScheduler: def __init__(self, max_concurrent: int 4): self.queues: dict[Priority, list[InferenceRequest]] { p: [] for p in Priority } self.max_concurrent max_concurrent self.active_count 0 self._lock asyncio.Lock() self._condition asyncio.Condition(self._lock) # 饥饿防护每个优先级的最小处理配额 self.quota {Priority.REALTIME: 0, Priority.API_CALL: 0, Priority.BATCH: 0, Priority.OFFLINE: 0} self.quota_reset_interval 100 # 每100个请求重置配额 self.processed_count 0 async def enqueue(self, request: InferenceRequest) - None: 将请求加入对应优先级队列 async with self._lock: self.queues[Priority(request.priority)].append(request) self._condition.notify_all() async def dequeue(self) - InferenceRequest | None: 按优先级策略取出下一个请求 async with self._condition: while self._all_empty(): await self._condition.wait() # 饥饿防护检查如果低优先级等待过久提升其配额 self._check_starvation() # 按优先级顺序查找可用请求 for priority in Priority: queue self.queues[priority] # 检查该优先级是否有配额饥饿防护可能限制高优先级 if queue and self.quota[priority] 0: request queue.pop(0) self.processed_count 1 if self.processed_count self.quota_reset_interval: self._reset_quota() return request return None def _all_empty(self) - bool: return all(len(q) 0 for q in self.queues.values()) def _check_starvation(self) - None: 检测低优先级是否被饥饿 now time.time() for priority in [Priority.BATCH, Priority.OFFLINE]: queue self.queues[priority] if queue and now - queue[0].enqueue_time 30: # 低优先级等待超过30秒限制高优先级配额 self.quota[Priority.REALTIME] max(self.quota[Priority.REALTIME] - 5, 0) self.quota[priority] 5 def _reset_quota(self) - None: 重置配额计数 self.quota {p: 0 for p in Priority} self.processed_count 0 async def process_loop(self, inference_engine: Callable) - None: 调度主循环 while True: if self.active_count self.max_concurrent: await asyncio.sleep(0.1) continue request await self.dequeue() if request is None: continue self.active_count 1 asyncio.create_task(self._execute_request(request, inference_engine)) async def _execute_request( self, request: InferenceRequest, inference_engine: Callable, ) - None: 执行推理请求 try: result await inference_engine(request.prompt, request.max_tokens) if request.callback: await request.callback(result) except Exception as e: if request.callback: await request.callback({error: str(e)}) finally: self.active_count - 12.2 动态优先级调整# dynamic_priority.py — 动态优先级调整策略 # 设计意图根据请求的等待时间和用户等级动态调整优先级 # 避免低优先级请求被无限期饥饿 import time class DynamicPriorityAdjuster: # 等待时间阈值秒超过此时间提升优先级 WAIT_THRESHOLDS { 0: float(inf), # P0 不需要提升 1: 10, # P1 等待10秒后提升 2: 30, # P2 等待30秒后提升 3: 60, # P3 等待60秒后提升 } # 用户等级优先级加成 TIER_BOOST { enterprise: -1, # 企业用户提升一级 pro: 0, # 专业用户不变 free: 1, # 免费用户降低一级 } classmethod def adjust_priority(cls, request: InferenceRequest) - int: 动态调整请求优先级 base_priority request.priority wait_time time.time() - request.enqueue_time # 等待时间提升 threshold cls.WAIT_THRESHOLDS.get(base_priority, float(inf)) if wait_time threshold: base_priority max(0, base_priority - 1) # 用户等级加成 tier_boost cls.TIER_BOOST.get(request.user_tier, 0) adjusted max(0, min(3, base_priority tier_boost)) return adjusted三、批量推理与请求合并3.1 请求合并策略# batch_merger.py — 推理请求批量合并 # 设计意图将多个短请求合并为一个 Batch 推理 // 提升 GPU 利用率降低单请求延迟 import asyncio import time from dataclasses import dataclass dataclass class BatchGroup: requests: list[InferenceRequest] created_at: float total_tokens: int class BatchMerger: def __init__( self, max_batch_size: int 8, max_wait_ms: int 50, max_total_tokens: int 4096, ): self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.max_total_tokens max_total_tokens self.pending: list[InferenceRequest] [] self._lock asyncio.Lock() async def add_request(self, request: InferenceRequest) - BatchGroup | None: 添加请求到待合并队列满足条件时返回一个 Batch async with self._lock: self.pending.append(request) # 检查是否满足批量条件 if len(self.pending) self.max_batch_size: return self._create_batch() if sum(r.max_tokens for r in self.pending) self.max_total_tokens: return self._create_batch() return None async def wait_and_flush(self) - BatchGroup | None: 等待超时后刷新当前待合并队列 await asyncio.sleep(self.max_wait_ms / 1000) async with self._lock: if self.pending: return self._create_batch() return None def _create_batch(self) - BatchGroup: batch BatchGroup( requestsself.pending[:], created_attime.time(), total_tokenssum(r.max_tokens for r in self.pending), ) self.pending [] return batch四、边界分析与架构权衡优先级抢占的实现复杂度真正的优先级抢占需要中断正在执行的推理任务但 GPU 推理一旦开始就无法中断。实际实现中抢占只能发生在请求调度阶段——高优先级请求跳过队列直接进入执行而非中断正在执行的请求。这意味着抢占的响应延迟取决于当前推理任务的执行时间。饥饿防护与优先级的矛盾饥饿防护机制为低优先级请求保留配额但这会牺牲高优先级请求的响应速度。在极端流量场景下保护低优先级可能导致付费用户体验下降。需要在商业策略和技术实现之间找到平衡。批量合并的延迟权衡请求合并需要等待队列积累到一定数量这引入了额外的等待延迟。对于实时对话场景即使 50ms 的等待也是不可接受的。解决方案是对实时请求跳过合并直接执行只对批量任务和离线分析进行合并。队列状态的监控与可观测性多级优先级队列的运行状态需要实时监控——各队列的积压长度、平均等待时间、调度器的吞吐量。缺乏可观测性会导致问题难以定位。建议暴露 Prometheus 指标并配置告警。五、总结推理服务的优先级调度将 GPU 资源分配从先到先得升级为价值驱动确保高价值请求优先获得推理资源。多级优先级队列、动态优先级调整和请求批量合并三个机制组合可以在保证付费用户体验的同时最大化 GPU 利用率。但抢占限制、饥饿防护、合并延迟和可观测性是需要持续关注的边界条件。落地建议按用户等级和请求类型划分优先级实时请求跳过合并直接执行暴露队列指标并配置积压告警定期审查优先级策略与商业目标的匹配度。