多模态 AI 推理的服务化架构：视觉-语言模型的批处理与调度优化

随着 GPT-4V、LLaVA、Qwen-VL 等多模态大模型的崛起，AI 推理从纯文本扩展到图像、视频、音频等多维度输入。这种转变对推理服务架构提出了全新的挑战：如何处理异构模态的混合请求？如何在延迟和吞吐量之间取得最优平衡？本文深入剖析多模态推理服务化的核心架构设计与工程优化策略。

一、多模态推理的独特挑战

与纯文本 LLM 推理相比，多模态推理面临三个核心差异：

1.1 输入长度爆炸

一张 1080x1920 的图像，经 Vision Encoder（如 ViT-L/14）转换后产生 576 个 patch token。按照 OpenAI 的分块策略，高分辨率图像可能产生数千个视觉 token。一次多模态请求的 token 总量可能达到纯文本请求的 10-50 倍。

1.2 计算图异构

请求A: [图像 → Vision Encoder → 投影层] → LLM 解码 50 tokens
请求B: [纯文本2000 tokens] → LLM 解码 100 tokens
请求C: [3图像+文本] → Vision Encoder → 投影层 → LLM 解码 200 tokens

三条请求的计算路径完全不同：Vision Encoder 是计算密集型的 CNN/ViT 前向传播，而 LLM 解码是 memory-bound 的自回归生成。将它们混合在同一批次中需要精细的资源规划。

1.3 内存占用不均衡

视觉 token 的 KV Cache 占用远高于文本 token。假设 hidden_size=4096，一个视觉 token 的 KV Cache 约为 32KB（fp16），576 个视觉 token 就消耗约 18MB 的 GPU 显存。当批量处理含图像请求时，显存可能在 Prefill 阶段就爆满。

二、多模态推理服务的核心架构

2.1 分离式 Prefill-Decode 架构

传统 LLM 推理的 PD 分离（Prefill-Decode Separation）在多模态场景下需要演进为三阶段流水线：

class MultimodalInferencePipeline:
    """多模态推理三阶段流水线"""
    
    def __init__(self):
        self.vision_encoder = VisionEncoderEngine()  # ViT 推理引擎
        self.projector = MultiModalProjectorEngine()  # 线性投影引擎
        self.llm_engine = LLMEngine()  # LLM 推理引擎
        
    async def execute(self, request: MultimodalRequest):
        # 阶段 1: 视觉编码 (GPU-bound)
        if request.has_image:
            visual_tokens = await self.vision_encoder.encode(
                images=request.images,
                resolution=request.image_resolution
            )
            projected_tokens = await self.projector.forward(
                visual_tokens=visual_tokens,
                target_dtype=self.llm_engine.dtype
            )
        else:
            projected_tokens = None
        
        # 阶段 2: 多模态 Prefill (Memory-bound)
        input_ids = self.tokenize_mm_input(
            text=request.text,
            projected_visual=projected_tokens
        )
        kv_cache = await self.llm_engine.prefill(
            request_id=request.id,
            input_ids=input_ids
        )
        
        # 阶段 3: 自回归 Decode (Memory-bound)
        async for token in self.llm_engine.generate(
            request_id=request.id,
            kv_cache=kv_cache,
            max_tokens=request.max_tokens
        ):
            yield token

2.2 Vision Encoder 专用调度器

Vision Encoder 的计算特性与 LLM 完全不同，需要独立的调度策略：

pub struct VisionEncoderScheduler {
    encoder: Arc<ViTModel>,
    image_queue: PriorityQueue<ImageTask>,
    batch_timeout: Duration,
    max_batch_pixels: u64,  // 按像素总量而非图片数量限制批次
}

impl VisionEncoderScheduler {
    pub async fn schedule(&mut self) -> Vec<EncodeResult> {
        let mut current_batch_pixel_count: u64 = 0;
        let mut batch = Vec::new();
        let deadline = Instant::now() + self.batch_timeout;
        
        while let Some(task) = self.image_queue.peek() {
            let task_pixels = task.width as u64 * task.height as u64;
            
            // 超过像素预算，停止攒批
            if current_batch_pixel_count + task_pixels > self.max_batch_pixels 
                && !batch.is_empty() {
                break;
            }
            
            // 等待更多请求攒批，但有超时上限
            if Instant::now() >= deadline && !batch.is_empty() {
                break;
            }
            
            batch.push(self.image_queue.pop().unwrap());
            current_batch_pixel_count += task_pixels;
        }
        
        // 动态填充到统一尺寸进行批处理
        self.execute_batch(batch).await
    }
}

关键设计点：按像素总量而非图片数量限制批次。一张 4K 图像（3840x2160）的像素量是 1080p 图像的 4 倍，如果按图片数量批处理，会导致显存使用剧烈波动。

2.3 动态分辨率与 Token 预算管理

class ResolutionManager:
    """根据请求优先级和系统负载动态调整图像分辨率"""
    
    RESOLUTION_PRESETS = {
        "high": (1080, 1920, 576 tokens),
        "medium": (512, 512, 256 tokens),
        "low": (224, 224, 256 tokens),
        "thumbnail": (128, 128, 64 tokens),
    }
    
    def allocate_resolution(
        self, 
        request: MultimodalRequest,
        system_load: SystemMetrics
    ) -> ResolutionConfig:
        """根据系统负载动态选择分辨率"""
        
        gpu_memory_headroom = system_load.gpu_free_memory / system_load.gpu_total_memory
        queue_depth = system_load.pending_request_count
        
        # 高负载时降级分辨率
        if gpu_memory_headroom < 0.2 or queue_depth > 20:
            base_res = "low"
        elif gpu_memory_headroom < 0.4 or queue_depth > 10:
            base_res = "medium"
        else:
            base_res = request.requested_resolution
        
        # 预算敏感型请求可以进一步降级
        if request.priority == Priority.ECONOMY:
            base_res = self._downgrade(base_res)
        
        return ResolutionConfig(**self.RESOLUTION_PRESETS[base_res])

三、异构批处理策略

3.1 模态感知的连续批处理 (Modality-Aware Continuous Batching)

传统的 Continuous Batching 假设所有请求具有相似的计算模式。多模态场景下，我们需要将 Vision Encoder 和 LLM 的批处理分离：

class ModalityAwareBatcher:
    """感知模态差异的连续批处理器"""
    
    def __init__(self, config: BatcherConfig):
        self.vision_batcher = VisionBatcher(
            max_pixel_batch=config.max_pixel_batch,
            timeout_ms=config.vision_timeout_ms
        )
        self.llm_batcher = LLMBatcher(
            max_token_batch=config.max_prefill_tokens,
            timeout_ms=config.llm_timeout_ms
        )
    
    async def run(self, request_stream: AsyncIterator[MultimodalRequest]):
        # Vision Encoder 和 LLM 的批处理是异步并行的
        vision_task = asyncio.create_task(self._vision_batch_loop(request_stream))
        llm_task = asyncio.create_task(self._llm_batch_loop(request_stream))
        
        await asyncio.gather(vision_task, llm_task)
    
    async def _vision_batch_loop(self, stream):
        async for batch in self.vision_batcher.chunk(stream):
            # Vision Encoder 批处理：按像素预算攒批
            images, metadata = self._prepare_image_batch(batch)
            visual_features = await self.vision_encoder.forward(images)
            
            # 将视觉特征注入到对应请求的输入中
            for i, meta in enumerate(metadata):
                meta.request.attach_visual_features(visual_features[i])
    
    async def _llm_batch_loop(self, stream):
        async for batch in self.llm_batcher.chunk(stream):
            # LLM Prefill 批处理：按 token 预算攒批
            input_ids = self._prepare_mm_inputs(batch)
            await self.llm_engine.prefill_batch(input_ids)

3.2 混合模态请求的 Prefill 调度

当 Vision Encoder 和 LLM Prefill 共享 GPU 时，需要精细的时间片调度：

时间线:
0ms    100ms   200ms   300ms   400ms   500ms
|-------|-------|-------|-------|
[ViT Batch: 4 images]           [Prefill Batch: 含视觉请求+文本请求]
         [ViT Batch: 2 images]
                  [Prefill Batch: 文本请求]
                          [ViT Batch: 1 image]

实现策略：

pub struct ModalityTimeSlicer {
    visionengine: Arc<VisionEncoder>,
    llmengine: Arc<LLMEngine>,
    slot_duration: Duration,  // 200ms 时间片
}

impl ModalityTimeSlicer {
    pub async fn run_loop(&self) {
        let mut interval = tokio::time::interval(self.slot_duration);
        
        loop {
            interval.tick().await;
            
            // 检查 Vision Encoder 是否有积压
            let vision_pending = self.vision_engine.queue_depth().await;
            let llm_pending = self.llm_engine.queue_depth().await;
            
            // 加权公平调度
            let total_pending = vision_pending + llm_pending;
            if total_pending == 0 { continue; }
            
            let vision_ratio = vision_pending as f32 / total_pending as f32;
            
            if vision_ratio > 0.3 && vision_pending > 0 {
                // 分配时间片给 Vision Encoder
                self.run_vision_slot().await;
            } else {
                // 分配时间片给 LLM
                self.run_llm_slot().await;
            }
        }
    }
}

四、KV Cache 分层管理

4.1 视觉 KV Cache 的特殊处理

视觉 token 的 KV Cache 具有以下特性：

1. 一次性消费：视觉 token 只在 Prefill 阶段被访问一次（不像文本 token 被每个 decode step 访问）
2. 占用大但访问频率低：一旦 LLM 开始 decode，视觉 token 的 KV Cache 很少被再次访问
3. 可压缩性强：视觉特征本身冗余度高，可以压缩存储

class TieredKVCacheManager:
    """分层管理视觉和文本的 KV Cache"""
    
    def __init__(self, gpu_memory_budget: int, cpu_memory_budget: int):
        # GPU 缓存：当前活跃 decode 的文本 KV
        self.gpu_text_cache = GPUKVCache(capacity=gpu_memory_budget * 0.5)
        
        # GPU 缓存：正在 Prefill 的视觉 KV（短期保留）
        self.gpu_visual_cache = GPUKVCache(capacity=gpu_memory_budget * 0.3)
        
        # CPU 缓存：已完成 Prefill 的视觉 KV（长期保留）
        self.cpu_visual_cache = CPUKVCache(capacity=cpu_memory_budget)
    
    async def on_prefill_complete(self, request_id: str, kv_cache: KVCache):
        """Prefill 完成后，视觉 KV 迁移到 CPU"""
        
        # 从 KV Cache 中分离视觉和文本部分
        visual_kv = kv_cache.extract_range(
            start=0, 
            end=visual_token_count
        )
        text_kv = kv_cache.extract_range(
            start=visual_token_count,
            end=kv_cache.total_tokens
        )
        
        # 视觉 KV 异步搬到 CPU
        await self.cpu_visual_cache.put_async(request_id, visual_kv)
        
        # 文本 KV 留在 GPU
        await self.gpu_text_cache.put(request_id, text_kv)
    
    async def on_eviction(self, request_id: str):
        """请求完成或驱逐时，清理相关缓存"""
        self.gpu_text_cache.remove(request_id)
        self.cpu_visual_cache.remove(request_id)

4.2 视觉 KV Cache 压缩

class VisualKVCacheCompressor:
    """对视觉 KV Cache 进行低秩压缩"""
    
    def __init__(self, hidden_size: int, compression_ratio: int = 4):
        self.compression_ratio = compression_ratio
        # 压缩矩阵：从 hidden_size 降到 hidden_size/ratio
        self.compress_k = nn.Linear(
            hidden_size, hidden_size // compression_ratio, bias=False
        )
        self.compress_v = nn.Linear(
            hidden_size, hidden_size // compression_ratio, bias=False
        )
    
    def compress(self, kv_cache: Tensor) -> CompressedKVCache:
        """
        压缩视觉 KV Cache，将 576 个 token 的 KV 压缩为
        576/ratio 个 token 的低秩表示
        """
        k_cache = kv_cache[0]  # [seq_len, hidden_size]
        v_cache = kv_cache[1]
        
        # 压缩: [576, 4096] → [576/4, 4096] = [144, 4096]
        compressed_k = self.compress_k(k_cache)
        compressed_v = self.compress_v(v_cache)
        
        return CompressedKVCache(
            k=compressed_k,
            v=compressed_v,
            original_seq_len=k_cache.shape[0],
            decompress_fn=self._decompress
        )
    
    def _decompress(self, compressed: Tensor) -> Tensor:
        """在需要访问原始 KV Cache 时解压缩"""
        return F.linear(compressed, self.compress_k.weight.t())

---

五、动态批大小与 Token 预算

5.1 自适应 Token 预算算法

class AdaptiveTokenBudget:
    """根据系统状态动态调整 Prefill Token 预算"""
    
    def __init__(self):
        self.base_budget = 2048  # 基础 token 预算
        self.min_budget = 512
        self.max_budget = 8192
        
    def calculate(self, metrics: SystemMetrics) -> int:
        """根据 GPU 显存压力和队列深度计算预算"""
        
        # 显存压力因子 (0.0 - 1.0)
        memory_pressure = 1.0 - (
            metrics.gpu_free_memory / metrics.gpu_total_memory
        )
        
        # 队列深度因子
        queue_factor = min(metrics.queue_depth / 30.0, 1.0)
        
        # 显存压力越大、队列越深，预算越小
        # 但队列深时适当增大预算以提高吞吐
        effective_pressure = memory_pressure * 0.7 + queue_factor * 0.3
        
        budget = int(
            self.base_budget * (1.0 - effective_pressure * 0.6)
        )
        
        return max(self.min_budget, min(self.max_budget, budget))
    
    def adjust_for_modality(
        self, 
        base_budget: int, 
        requests: List[MultimodalRequest]
    ) -> int:
        """根据批次中的模态组成调整预算"""
        
        # 计算批次中视觉 token 的预期占比
        visual_token_total = sum(
            r.expected_visual_tokens for r in requests
        )
        visual_ratio = visual_token_total / max(
            sum(r.total_tokens for r in requests), 1
        )
        
        # 视觉 token 需要更多临时内存（注意力计算中的中间激活）
        # 所以对含视觉请求的批次，降低 token 预算
        if visual_ratio > 0.5:
            return int(base_budget * 0.6)
        elif visual_ratio > 0.2:
            return int(base_budget * 0.8)
        
        return base_budget

5.2 优先级感知的批处理

@dataclass
class BatchAllocation:
    """批次分配结果"""
    vision_batch: List[ImageTask]
    text_batch: List[TextRequest]
    mm_batch: List[MultimodalRequest]  # 已完成视觉编码的请求
    
class PriorityBatchAllocator:
    """优先级感知的异构批分配"""
    
    def allocate(
        self, 
        vision_queue: Deque[ImageTask],
        llm_queue: Deque[LLMRequest],
        priority_weights: Dict[Priority, float]
    ) -> BatchAllocation:
        
        # 按优先级排序
        vision_queue = self._sort_by_priority(vision_queue)
        llm_queue = self._sort_by_priority(llm_queue)
        
        # 高优先级请求优先进入批次
        vision_batch = []
        token_budget = self.current_token_budget
        
        for task in vision_queue:
            task_tokens = task.estimated_token_cost()
            if token_budget - task_tokens >= 0:
                vision_batch.append(task)
                token_budget -= task_tokens
            else:
                break
        
        # LLM 批次分配考虑已完成的 Vision 结果
        mm_batch = []
        text_batch = []
        
        for req in llm_queue:
            if req.has_visual_features and req.visual_features_ready:
                mm_batch.append(req)
            elif not req.has_image:
                text_batch.append(req)
        
        return BatchAllocation(
            vision_batch=vision_batch,
            text_batch=text_batch,
            mm_batch=mm_batch
        )

---

六、生产环境工程实践

6.1 请求路由与负载均衡

多模态请求应该根据其特征路由到最合适的推理实例：

class MultimodalRequestRouter:
    """多模态请求路由器"""
    
    def __init__(self, pool: InferencePool):
        self.pool = pool
        
    def route(self, request: MultimodalRequest) -> InferenceInstance:
        """选择最优推理实例"""
        
        candidates = self.pool.get_healthy_instances()
        
        if request.has_image:
            # 图像请求需要 Vision Encoder 支持
            candidates = [c for c in candidates if c.supports_vision]
            
            # 优先选择 Vision Encoder 队列较短的实例
            candidates.sort(
                key=lambda c: c.vision_queue_depth
            )
        else:
            # 纯文本请求可以路由到无 Vision Encoder 的轻量实例
            # 优先选择 LLM 解码队列较短的实例
            candidates.sort(
                key=lambda c: c.llm_decode_queue_depth
            )
        
        # 选择负载最轻的候选
        return candidates[0] if candidates else None

6.2 超时与降级策略

class MultimodalFallbackManager:
    """多模态请求的超时与降级管理"""
    
    TIMEOUT_CONFIG = {
        Stage.VISION_ENCODE: 5.0,    # 视觉编码超时: 5s
        Stage.PROJECTION: 1.0,        # 投影层超时: 1s
        Stage.PREFILL: 10.0,          # Prefill 超时: 10s
        Stage.DECODE_FIRST: 2.0,      # 首 token 超时: 2s
        Stage.DECODE_NEXT: 0.5,       # 后续 token 超时: 500ms
    }
    
    async def execute_with_fallback(
        self, 
        request: MultimodalRequest
    ) -> AsyncIterator[Token]:
        
        try:
            # 视觉编码阶段：超时则降级为纯文本
            visual_features = await asyncio.wait_for(
                self.encode_vision(request),
                timeout=self.TIMEOUT_CONFIG[Stage.VISION_ENCODE]
            )
        except asyncio.TimeoutError:
            logger.warning(
                f"Vision encode timeout for {request.id}, "
                f"falling back to text-only"
            )
            request = request.to_text_only()
            visual_features = None
        
        try:
            # Prefill 阶段
            kv_cache = await asyncio.wait_for(
                self.prefill(request, visual_features),
                timeout=self.TIMEOUT_CONFIG[Stage.PREFILL]
            )
        except asyncio.TimeoutError:
            # Prefill 超时：重定向到冷启动实例
            raise FallbackToColdInstance(request)
        
        # Decode 阶段：流式输出
        async for token in self.generate(kv_cache, timeout_config=self.TIMEOUT_CONFIG):
            yield token

6.3 监控与可观测性

class MultimodalMetricsCollector:
    """多模态推理专用指标收集"""
    
    def __init__(self, registry: MetricsRegistry):
        # 核心延迟指标
        self.vision_latency = registry.histogram(
            "mm_vision_encode_seconds",
            buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
        )
        self.prefill_latency = registry.histogram(
            "mm_prefill_seconds",
            buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
        )
        self.ttft = registry.histogram(  # Time To First Token
            "mm_ttft_seconds",
            buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0]
        )
        
        # 批次组成指标
        self.batch_modality_ratio = registry.gauge(
            "mm_batch_visual_ratio",
            description="视觉 token 在批次中的占比"
        )
        
        # KV Cache 分层指标
        self.visual_cache_hit_rate = registry.gauge(
            "mm_visual_cache_hit_rate",
            description="视觉 KV Cache 命中率"
        )
        self.visual_cache_tier_distribution = registry.gauge_vec(
            "mm_visual_cache_tiers",
            description="视觉 KV Cache 分层分布",
            labels=["tier"],  # gpu, cpu, disk
        )

---

七、前沿优化方向

7.1 视觉 Token 重排序与稀疏注意力

研究表明，视觉 token 之间存在大量冗余。通过对视觉 token 进行聚类重排序，使用稀疏注意力（Sparse Attention）可以显著减少视觉部分的计算量：

class VisualTokenReducer:
    """视觉 Token 压缩与重排序"""
    
    def __init__(self, target_tokens: int = 128):
        self.target_tokens = target_tokens
        self.clusterer = KMeans(n_clusters=target_tokens)
    
    def reduce(self, visual_tokens: Tensor) -> Tensor:
        """
        将 576 个视觉 token 通过 K-Means 聚类压缩为 128 个
        保留最具代表性的视觉信息
        """
        batch_size, seq_len, hidden_dim = visual_tokens.shape
        
        # 对序列维度进行聚类
        tokens_2d = visual_tokens.reshape(-1, hidden_dim).cpu()
        
        # K-Means 聚类
        self.clusterer.fit(tokens_2d.numpy())
        centroids = torch.from_numpy(self.clusterer.cluster_centers_)
        
        return centroids.unsqueeze(0)  # [1, 128, hidden_dim]

7.2 推测解码在多模态的扩展

推测解码（Speculative Decoding）可以扩展至多模态场景：使用小型 Vision Encoder（如 ViT-Ti/16）快速生成"草图"视觉 token，再由大型 Vision Encoder 验证修正：

class MultimodalSpeculativeDecoder:
    """多模态推测解码器"""
    
    async def speculative_visual_encode(
        self, 
        image: Image,
        draft_model: VisionEncoder,
        target_model: VisionEncoder,
        acceptance_threshold: float = 0.95
    ) -> Tensor:
        # 1. 小型模型快速生成草图的视觉 token
        draft_features = await draft_model.encode(image)
        
        # 2. 大型模型生成目标视觉 token
        target_features = await target_model.encode(image)
        
        # 3. 计算相似度，如果足够接近则接受草图
        cosine_sim = F.cosine_similarity(
            draft_features.flatten(),
            target_features.flatten(),
            dim=0
        )
        
        if cosine_sim > acceptance_threshold:
            # 接受草图，节省一次大模型推理
            return draft_features
        else:
            # 使用精确结果
            return target_features

---

八、总结

多模态 AI 推理服务化是一项系统工程，需要在以下几个维度取得平衡：

1. 异构计算调度：Vision Encoder 和 LLM 的计算特性完全不同，需要独立的调度器协同工作
2. 分层内存管理：视觉 KV Cache 的一次性消费特性使其适合分层存储策略
3. 动态资源分配：图像分辨率、token 批大小、时间片分配都需要根据系统负载实时调整
4. 生产级可靠性：超时降级、故障转移、优先级调度是服务保障的基石

随着端侧多模态模型（如 Phi-3.5-Vision、Qwen2-VL-2B）的兴起，如何在资源受限设备上高效运行多模态推理将成为下一个工程挑战。未来，我们可能会看到更多硬件层面的多模态加速——例如直接在显示流水线中集成视觉编码器，实现从像素到语义的零拷贝传输。

---

延伸阅读