大模型推理服务的请求队列与优先级调度：从先到先得到智能排序，推理资源的效率最大化

一、推理服务的排队困境：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(order=True)
class InferenceRequest:
    priority: int = field(compare=True)
    enqueue_time: float = field(compare=True)
    request_id: str = field(compare=False)
    prompt: str = field(compare=False)
    max_tokens: int = field(compare=False)
    user_tier: str = field(compare=False)
    callback: Callable[[Any], Awaitable[None]] = field(compare=False, default=None)

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 -= 1

2.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(
            requests=self.pending[:],
            created_at=time.time(),
            total_tokens=sum(r.max_tokens for r in self.pending),
        )
        self.pending = []
        return batch

四、边界分析与架构权衡

优先级抢占的实现复杂度：真正的优先级抢占需要中断正在执行的推理任务，但 GPU 推理一旦开始就无法中断。实际实现中，"抢占"只能发生在请求调度阶段——高优先级请求跳过队列直接进入执行，而非中断正在执行的请求。这意味着抢占的响应延迟取决于当前推理任务的执行时间。

饥饿防护与优先级的矛盾：饥饿防护机制为低优先级请求保留配额，但这会牺牲高优先级请求的响应速度。在极端流量场景下，保护低优先级可能导致付费用户体验下降。需要在商业策略和技术实现之间找到平衡。

批量合并的延迟权衡：请求合并需要等待队列积累到一定数量，这引入了额外的等待延迟。对于实时对话场景，即使 50ms 的等待也是不可接受的。解决方案是对实时请求跳过合并，直接执行；只对批量任务和离线分析进行合并。

队列状态的监控与可观测性：多级优先级队列的运行状态需要实时监控——各队列的积压长度、平均等待时间、调度器的吞吐量。缺乏可观测性会导致问题难以定位。建议暴露 Prometheus 指标并配置告警。

五、总结

推理服务的优先级调度将 GPU 资源分配从"先到先得"升级为"价值驱动"，确保高价值请求优先获得推理资源。多级优先级队列、动态优先级调整和请求批量合并三个机制组合，可以在保证付费用户体验的同时最大化 GPU 利用率。但抢占限制、饥饿防护、合并延迟和可观测性是需要持续关注的边界条件。落地建议：按用户等级和请求类型划分优先级；实时请求跳过合并直接执行；暴露队列指标并配置积压告警；定期审查优先级策略与商业目标的匹配度。