视觉服务化与推理加速

引言

模型训好了不等于能上线。线上服务要面对并发请求、突发流量、显存约束与尾延迟。一个常见的误区是只优化模型推理时间，却忽略了预处理、序列化与网络开销，最终端到端延迟并没有改善。视觉服务的核心是在吞吐、延迟、成本之间找平衡。

本文按「服务方案 → 批处理 → 加速手段 → 指标与压测 → 代码」的顺序拆解，并给出可直接运行的批处理服务与压测代码。

目录

1. 服务化方案概览
2. Triton Inference Server
3. TorchServe
4. 动态批处理与并发
5. 推理加速手段
6. 性能指标与压测
7. 代码：FastAPI 批处理服务
8. 代码：并发压测
9. 方案对比表
10. 加速手段收益表
11. 服务监控与可观测性
12. 弹性伸缩与成本
13. 多模型编排与级联
14. 权衡取舍
15. 常见坑清单
16. 小结

1. 服务化方案概览

选择建议：多模型、多框架、要极致吞吐，用 Triton；纯 PyTorch 且要快速上线，用 TorchServe；有特殊业务逻辑或轻量需求，用 FastAPI 自建。

2. Triton Inference Server

Triton 是 NVIDIA 开源的推理服务器，核心特性：

• 多框架：同时托管 TensorRT、ONNX、PyTorch、TensorFlow 模型。
• 动态批处理：把短时间内的多个请求合并成一个批，提升 GPU 利用率。
• 多实例：同一模型起多个实例并行，提高并发。
• 模型版本与热更新：支持灰度与回滚。
• 指标导出：Prometheus 格式的延迟与吞吐指标。

一个 Triton 模型仓库的目录结构包含 config.pbtxt 与版本目录，配置里指定最大批大小、动态批处理窗口与实例数。

3. TorchServe

TorchServe 是 PyTorch 官方服务框架：

• 用 .mar 打包模型与处理器（handler），handler 里定义预处理与后处理。
• 支持模型版本、A/B 测试与自动扩缩。
• 内置指标与日志。
• 适合已经用 PyTorch 且不想引入额外复杂度的团队。

它的动态批处理能力弱于 Triton，高吞吐场景通常还是选 Triton。

4. 动态批处理与并发

动态批处理是提升吞吐最有效的手段：GPU 擅长并行，批大小从 1 提到 8 常能把吞吐提升数倍，而单请求延迟只小幅上升。

• 批处理窗口：等待极短时间（如几毫秒）收集请求凑批。
• 最大批大小：受显存限制，超了要拆分。
• 并发实例：单实例无法打满 GPU 时，起多个实例并行。
• 优先级队列：对延迟敏感的请求优先处理。

并发模型的选择：Python 的 GIL 限制多线程，常用多进程（gunicorn 多 worker）或异步（asyncio）配合批处理。

5. 推理加速手段

关键洞察：先测量瓶颈在哪。如果预处理占了一半时间，只优化模型推理收益有限，应该把 resize 与归一化也搬到 GPU。

6. 性能指标与压测

服务化关注三类指标：

• 吞吐：QPS 或每秒处理的图片数，看整体处理能力。
• 延迟：P50、P90、P99，尾延迟决定用户体验，只看平均会误导。
• 资源：GPU 利用率、显存占用、CPU 使用率，判断是否饱和。

压测方法：用固定并发逐步加压，记录不同并发下的吞吐与延迟曲线，找到吞吐饱和点与延迟拐点。压测工具可用 locust、wrk 或自写脚本。

7. 代码：FastAPI 批处理服务

下面是一个带批处理的推理服务，用后台批处理线程把请求聚合成批。

import asyncio
import numpy as np
import torch
from fastapi import FastAPI
from pydantic import BaseModel
from torchvision import models

app = FastAPI()
model = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1).eval()

queue: asyncio.Queue = asyncio.Queue()

class Request(BaseModel):
    data: list[float]        # 展平的 3x224x224 输入

async def batch_worker(max_batch=8, wait_ms=5):
    while True:
        batch, futures = [], []
        # 等第一个请求
        item, fut = await queue.get()
        batch.append(item)
        futures.append(fut)
        # 在窗口内继续收集
        deadline = asyncio.get_event_loop().time() + wait_ms / 1000
        while len(batch) < max_batch:
            timeout = deadline - asyncio.get_event_loop().time()
            if timeout <= 0:
                break
            try:
                item, fut = await asyncio.wait_for(queue.get(), timeout)
                batch.append(item)
                futures.append(fut)
            except asyncio.TimeoutError:
                break
        x = torch.tensor(np.stack(batch), dtype=torch.float32)
        with torch.inference_mode():
            out = model(x).softmax(dim=1)
        for fut, row in zip(futures, out):
            fut.set_result(row.tolist())

@app.on_event("startup")
async def startup():
    asyncio.create_task(batch_worker())

@app.post("/predict")
async def predict(req: Request):
    x = np.array(req.data, dtype=np.float32).reshape(3, 224, 224)
    fut = asyncio.get_event_loop().create_future()
    await queue.put((x, fut))
    return {"probs": await fut}

这套模式把零散请求聚合成批，显著提升 GPU 利用率。生产环境建议直接用 Triton 的动态批处理，自建仅在需要特殊逻辑时使用。

8. 代码：并发压测

用 asyncio 并发压测，统计 P50 与 P99 延迟。

import asyncio
import time
import numpy as np
import httpx

async def one_request(client, payload):
    start = time.perf_counter()
    await client.post("http://localhost:8000/predict", json=payload)
    return (time.perf_counter() - start) * 1000

async def bench(concurrency=16, total=200):
    payload = {"data": np.random.randn(3 * 224 * 224).astype(float).tolist()}
    async with httpx.AsyncClient(timeout=30) as client:
        latencies = []
        sem = asyncio.Semaphore(concurrency)

        async def worker():
            async with sem:
                latencies.append(await one_request(client, payload))

        await asyncio.gather(*[worker() for _ in range(total)])

    lat = np.array(latencies)
    print(f"P50={np.percentile(lat, 50):.1f}ms P99={np.percentile(lat, 99):.1f}ms")
    print(f"throughput={total / (lat.sum() / 1000 / concurrency):.1f} qps")

asyncio.run(bench())

压测时要区分「单请求延迟」与「并发下延迟」。高并发下延迟上升是正常的，关键看拐点出现在哪里。

9. 方案对比表

10. 加速手段收益表

11. 服务监控与可观测性

上线只是开始，监控决定服务能不能稳住。

• 延迟分布：记录 P50、P90、P99，按模型版本与接口维度切分。
• 吞吐与饱和度：QPS、GPU 利用率、显存占用、队列长度。
• 错误率：区分请求错误、推理错误与超时，分别告警。
• 输入漂移：统计输入分辨率、亮度、置信度分布，发现分布偏移。
• 业务指标：预测类别分布、平均置信度，异常时能快速定位是数据问题还是模型问题。
• 追踪：给每个请求打 trace id，串联预处理、推理、后处理耗时，定位瓶颈。

一个实用做法是「影子流量」：把新模型与老模型同时推理，只返回老模型结果，对比两者差异后再切换。

12. 弹性伸缩与成本

推理服务的成本主要在 GPU。伸缩与调度直接决定单位成本。

• 水平伸缩：按队列长度或 GPU 利用率增减实例，避免长期空转。
• 模型常驻与按需加载：高频模型常驻，低频模型按需加载，节省显存。
• 混部：把多个小模型放到同一张 GPU，提高利用率。
• 请求分级：对延迟不敏感的批量任务用低优先级队列，填满空闲算力。
• 量化换成本：INT8 后同样的 GPU 能承载更多并发，直接降低单位成本。
• 冷启动优化：预加载权重、缓存编译结果，缩短扩容后的可用时间。

成本优化的前提是准确的容量规划：先测出单实例在目标延迟下的最大吞吐，再按峰值流量估算实例数，留出安全余量。

13. 多模型编排与级联

真实业务常常是多个模型串联或并联，编排方式影响整体延迟。

• 串联级联：检测 → 裁剪 → 分类，前级输出是后级输入，总延迟是各级之和。
• 并联聚合：多个模型对同一输入独立推理，结果融合，延迟取最慢者。
• 早退：简单样本走轻量分支提前返回，难样本才走完整链路，降低平均延迟。
• 异步流水线：把各阶段放到不同 worker，请求在阶段间流动，提升吞吐。

import time

def cascade(pipeline, image):
    # pipeline 是 (name, fn) 列表，逐级传递
    trace = {}
    x = image
    for name, fn in pipeline:
        start = time.perf_counter()
        x = fn(x)
        trace[name] = round((time.perf_counter() - start) * 1000, 2)
    return x, trace

stages = [
    ("detect", lambda im: [im]),
    ("classify", lambda boxes: boxes),
]
result, trace = cascade(stages, "img")
print("stage latency ms:", trace)

逐级记录耗时是定位流水线瓶颈的基础，也是容量规划的依据。检测与分类的级联在工程上极为常见，压缩时要整体考虑精度与延迟的累积。

14. 权衡取舍

• 批大小与延迟：批越大吞吐越高，但单请求延迟上升，实时场景要限批。
• 精度与吞吐：INT8 提升吞吐但可能掉点，看业务能否接受。
• 单模型与多模型共享 GPU：共享省钱但互相争抢资源，隔离性差。
• 自建与用现成：自建灵活但工作量大，Triton 省事但学习曲线陡。
• 单机与分布式：单机简单，分布式要处理负载均衡与状态一致性。

15. 常见坑清单

• 预处理成瓶颈：图像解码与 resize 在 CPU 上串行，GPU 空转。
• 只测平均延迟：忽略 P99，尾延迟拖垮体验。
• 批处理超时设置过大：为凑批等太久，单请求延迟飙升。
• 忘记预热：首个请求包含编译与显存分配，延迟异常高。
• 显存泄漏：长跑后显存不释放，最终 OOM，需定期压测验证。
• 序列化开销：大张量用 JSON 传输极慢，改用二进制或 protobuf。
• 冷启动慢：模型加载耗时，需常驻或预热池。
• 忽略并发下的 GIL：纯 Python 多线程无法并行，用多进程或异步。
• 压测环境与线上不一致：CPU、GPU、网络不同，指标不可迁移。
• 无监控：上线后不监控延迟与错误率，问题靠用户反馈发现。
• 忽略影子流量：新模型直接切流，出问题才发现精度下降。
• 容量规划拍脑袋：未测单实例上限就按峰值估实例数，扩容后仍超时。
• 批处理无上限：最大批大小设置过大，突发流量下显存 OOM。

16. 小结

视觉服务化的主线是：选方案（Triton 优先，特殊需求自建）→ 开动态批处理 → 按平台加速（TensorRT、ONNX Runtime）→ 把预处理也纳入优化 → 用 P50 与 P99 压测找拐点 → 上线监控与灰度。记住三条：先测量再优化、批大小是吞吐与延迟的旋钮、端到端延迟才是用户感知的延迟。服务化的收益往往大于模型本身的微调，值得投入工程资源。

延伸阅读

• 模型压缩与端侧部署 — 量化与端侧运行时
• 模型训练与迁移学习 — 训练到导出的完整链路
• 推理性能基准测试 — 科学的性能评测方法
• 计算机视觉任务全景与工程链路 — 服务化在链路中的位置