从零部署一个模型，是用最小可用路径走通「训练→导出→封装→服务→容器化→压测→监控」这条完整闭环。 概念页讲清了推理、格式、服务是什么，但「动手把一个小模型真的跑成线上服务」是另一码事——它把几十个知识点串成一条肌肉记忆。这篇解决两个问题：部署的最小完整集是什么？每一步的坑在哪？ 跟着做一遍，你会得到一套可复用的部署骨架（skeleton）：一个带单元测试的推理类、一个 FastAPI 服务、一个 Dockerfile、一组压测与监控配置。这套骨架以后换模型、换引擎都能复用，也天然是一份简历上拿得出手的作品（作品集价值见 简历分析与包装）。

先决条件

本文假设你：会基础 Python、能在本地跑 PyTorch、机器上有 Docker。GPU 不是必需的——例子里的模型小到 CPU 也能跑。整个流程大约 2-3 小时。

一、整体流程地图：七步闭环

部署不是「写个接口把模型包起来」那么一步到位，而是七步环环相扣。缺任何一步，到线上就会以不同姿势翻车：不导出就绑死 Python、不封装就前后处理不一致、不压测就不知道容量、不监控就裸奔（这些坑的完整拆解见 常见陷阱与反模式）。

  训练            导出             封装             服务            容器化           压测            监控
 ┌──────┐   ┌────────────┐   ┌────────────┐   ┌────────────┐   ┌────────────┐   ┌────────────┐   ┌────────────┐
 │ PyTorch│──▶│ ONNX 模型   │──▶│ Inference   │──▶│ FastAPI     │──▶│ Docker      │──▶│ wrk/locust  │──▶│ Prometheus  │
 │ 训练    │   │ (.onnx)    │   │ 类(预处理/后  │   │ 服务        │   │ 镜像        │   │ 压测        │   │ 指标/告警    │
 └──────┘   │  + meta.json │   │ 处理)       │   │             │   │             │   │             │   │             │
            └────────────┘   └────────────┘   └────────────┘   └────────────┘   └────────────┘   └────────────┘
   step1        step2            step3            step4            step5           step6            step7
   精度基准       可复现交付       一致性           可访问性         可迁移性         可容量规划        可观测

七步各自回答一个问题：

步骤	产物	回答的问题	验收标准
① 训练	权重文件	模型精度基线是多少？	有明确验证指标，记录为「离线基线」
② 导出	ONNX + 元数据	换引擎能跑吗？	导入即用，输入输出 shape 固定
③ 封装	Inference 类	预处理/后处理与训练一致吗？	单测通过，与训练时代码共用同一份预处理
④ 服务	REST API	别人怎么调用？	健康检查 + 错误语义 + 文档
⑤ 容器化	镜像 + compose	到哪都能跑？	一条命令起服务
⑥ 压测	容量报告	能扛多少 QPS？	有 SLO 有结论
⑦ 监控	指标 + 告警	出问题谁能先知道？	黄金指标齐备

二、本文的小项目设定

我们要部署一个 CIFAR-10 图像分类器：MobileNetV2 在 CIFAR-10（10 类、32×32 小图）上微调几轮，精度约 80%。选它的理由：

• 模型小而真实：MobileNetV2 的 ONNX 权重约 14MB（FP32），CPU 单线程推理约 5-15ms，压测阶段不用 GPU 也能打出有意义的数据；
• 预处理有真实复杂度：归一化、resize、通道顺序，正好暴露「训练/推理预处理不一致」这个头号坑；
• 输入输出明确：输入是图片字节流，输出是 top-5 标签，天然适合演示 HTTP 服务。

仓库结构（后续每一步都会往这个目录里填东西）：

image-classifier/
├── train_export.py        # ① 训练 + 导出 ONNX
├── requirements.txt
├── app/
│   ├── main.py            # ④ FastAPI 服务
│   ├── inference.py       # ③ 推理类
│   └── metrics.py         # ⑦ Prometheus 指标
├── models/
│   ├── model.onnx         # ② 导出产物
│   └── meta.json          # 标签表、预处理参数
├── tests/
│   └── test_inference.py  # 推理类单元测试
├── Dockerfile             # ⑤ 容器化
├── docker-compose.yml
└── prometheus.yml         # ⑦ 监控配置

三、① 训练并导出 ONNX

训练不是本文重点，但导出前的最后一件事很重要：记录离线基线指标。这是后面所有「线上是不是退化了」对比的锚点（详见 模型优化实战 的基线测量）。

# train_export.py —— 训练 + 导出 ONNX，CPU 可跑，约 15 分钟
import json
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from torchvision import datasets, transforms, models

# 训练/推理必须共用同一套预处理！这里定义一次，导出后写进 meta.json
mean, std = (0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)
train_tf = transforms.Compose([
    transforms.RandomCrop(32, padding=4),
    transforms.RandomHorizontalFlip(),
    transforms.ToTensor(),
    transforms.Normalize(mean, std),
])
eval_tf = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize(mean, std),
])

def main():
    torch.manual_seed(42)
    ds_train = datasets.CIFAR10("./data", train=True, download=True, transform=train_tf)
    ds_eval  = datasets.CIFAR10("./data", train=False, download=True, transform=eval_tf)
    model = models.mobilenet_v2(weights=None)
    model.classifier[1] = nn.Linear(model.last_channel, 10)  # 10 类
    opt = torch.optim.Adam(model.parameters(), lr=1e-3)

    for epoch in range(3):
        model.train()
        for x, y in DataLoader(ds_train, batch_size=128, shuffle=True, num_workers=2):
            opt.zero_grad()
            nn.functional.cross_entropy(model(x), y).backward()
            opt.step()

    model.eval()
    correct = total = 0
    with torch.no_grad():
        for x, y in DataLoader(ds_eval, batch_size=256, num_workers=2):
            correct += (model(x).argmax(1) == y).sum().item()
            total += y.size(0)
    acc = correct / total
    print(f"离线基线 accuracy = {acc:.4f}")  # 期望 ≈ 0.79-0.82

    # ---------- 导出 ONNX ----------
    dummy = torch.randn(1, 3, 32, 32)
    torch.onnx.export(
        model, dummy, "models/model.onnx",
        input_names=["input"], output_names=["logits"],
        dynamic_axes={"input": {0: "batch"}, "logits": {0: "batch"}},
        opset_version=17,
    )
    # 把标签表和预处理参数一起存下来，部署时用
    with open("models/meta.json", "w", encoding="utf-8") as f:
        json.dump({"labels": ds_train.classes, "mean": mean, "std": std,
                   "input_size": [3, 32, 32], "baseline_acc": acc}, f, indent=2)
    print("已导出 models/model.onnx 与 models/meta.json")

if __name__ == "__main__":
    main()

导出时的四个关键决策：

1. opset_version 用 17 或更高（PyTorch 2.x 默认即可），太低会缺算子支持；
2. dynamic_axes 给 batch 维开动态，否则服务端只能一次一张图；也可以后续在 ONNX Runtime 里配 session_options.intra_op_num_threads 调线程数；
3. mean/std 必须写进 meta.json——线上预处理和训练用同一份数值，这是防「离线好线上崩」的第一道保险；
4. 导出后用 ONNX Runtime 立即跑一遍比对输出，允许的数值差一般在 1e-4 量级，差异过大说明导出有算子问题（见 模型格式与转换）。

导出前先跑 torch.onnx.export 的 check=True

导出时默认会做一次「用 PyTorch 跑一遍、再用导出的图跑一遍、比对结果」的检查。千万别手滑关掉它（do_constant_folding=False 只在你确定要保留动态 shape 时用）。数值对不上，第一现场在导出，不在线上。

四、② ③ 封装推理类：预处理/后处理与训练完全一致

封装推理类（inference class）是整个骨架的心脏：服务代码只依赖这个类，不直接碰 ONNX 会话。这样以后换引擎（Triton、vLLM）、换量化版本，只改一个文件。

# app/inference.py —— 推理类，与训练共用预处理参数
import json
import threading
import numpy as np
import onnxruntime as ort
from PIL import Image


class ImageClassifier:
    def __init__(self, onnx_path: str, meta_path: str):
        with open(meta_path, encoding="utf-8") as f:
            meta = json.load(f)
        self.labels = meta["labels"]
        self.mean = np.array(meta["mean"], dtype=np.float32).reshape(3, 1, 1)
        self.std = np.array(meta["std"], dtype=np.float32).reshape(3, 1, 1)
        # 单例会话，全局只创建一次（见常见坑⑤：别每个请求建会话）
        self.sess = ort.InferenceSession(onnx_path,
                                         providers=["CPUExecutionProvider"])
        # onnxruntime 会话线程安全，可共享；加锁是为了多线程下排队一致
        self._lock = threading.Lock()

    def preprocess(self, image_bytes: bytes) -> np.ndarray:
        # 和训练 eval_tf 逐项对应：ToTensor + Normalize
        img = Image.open(BytesIO(image_bytes)).convert("RGB").resize((32, 32))
        arr = np.asarray(img, dtype=np.float32) / 255.0          # ToTensor
        arr = arr.transpose(2, 0, 1)                              # HWC -> CHW
        arr = (arr - self.mean) / self.std                        # Normalize
        return arr[None, ...]                                     # (1,3,32,32)

    def predict(self, image_bytes: bytes) -> dict:
        x = self.preprocess(image_bytes)
        with self._lock:
            logits = self.sess.run(["logits"], {"input": x})[0]
        probs = np.exp(logits - logits.max(axis=1, keepdims=True))
        probs = probs / probs.sum(axis=1, keepdims=True)          # softmax
        top = np.argsort(probs[0])[::-1][:5]
        return {"top5": [{"label": self.labels[i], "prob": float(probs[0][i])}
                         for i in top]}

为什么坚持「预处理放服务端」

训练侧的图片增强（RandomCrop 等）只该出现在训练数据加载里；但 ToTensor + Normalize 是输入分布的一部分，必须由服务端复现。把预处理留在客户端会导致「模型看到的分布和训练时不一样」，这是 常见陷阱与反模式 里排第一的坑。

配套单元测试（占位 3 行，真实 20 行）：

# tests/test_inference.py
def test_preprocess_matches_training():
    # 用训练集的同一张图：把 ToTensor+Normalize 的手算结果和 preprocess 比对
    # 断言 np.allclose(...) —— 这是「前后处理一致性」的机器化验收
    ...

五、④ FastAPI 服务化

服务层只做三件事：接请求、调推理类、返回结构化响应。业务逻辑（重试、鉴权、限流）不写在这里，留给网关（模型网关与灰度发布 会讲）。完整的 FastAPI 案例拆解见 FastAPI + Docker 在线服务。

# app/main.py
import time
from fastapi import FastAPI, File, UploadFile, HTTPException
from app.inference import ImageClassifier
from app.metrics import metrics

app = FastAPI(title="image-classifier")
clf = ImageClassifier("models/model.onnx", "models/meta.json")


@app.get("/healthz")
def healthz():
    return {"status": "ok"}          # 给 K8s/Docker 健康检查用，不带推理逻辑


@app.post("/predict")
def predict(file: UploadFile = File(...)):
    data = file.file.read()
    t0 = time.perf_counter()
    try:
        result = clf.predict(data)
    except Exception as e:           # 图片损坏等，返回 400 而非 500
        raise HTTPException(status_code=400, detail=f"bad image: {e}")
    latency_ms = (time.perf_counter() - t0) * 1000
    metrics.observe(latency_ms)      # ⑦ 埋点：计数、直方图
    return {"result": result, "latency_ms": round(latency_ms, 2)}

启动命令（生产不要用 uvicorn 单 worker，见坑⑤）：

pip install -r requirements.txt
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
curl -F "file=@cat.jpg" http://localhost:8000/predict
curl http://localhost:8000/healthz

单 worker 陷阱

uvicorn --workers 1 在本地调试没问题，但线上 1 个进程 = 1 份模型 + 单线程处理请求。--workers 4 时每个 worker 各加载一份模型（吃 4 份内存），这是部署新手最常踩的显存/内存账（显存泄漏与 OOM 有完整分析）。

六、⑤ Docker 化

容器化的目标是一条命令在任何机器上复现。镜像分两步：先装依赖、再拷模型与代码，让模型层与依赖层分开缓存。

# Dockerfile —— 基于 slim 镜像，体积小一半以上
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app/ app/
COPY models/ models/
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml —— 一条命令起服务 + 监控
services:
  api:
    build: .
    ports: ["8000:8000"]
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/healthz"]
      interval: 30s
      timeout: 5s
      retries: 3
  prometheus:
    image: prom/prometheus:v2.53.0
    volumes: ["./prometheus.yml:/etc/prometheus/prometheus.yml"]
    ports: ["9090:9090"]

构建并验证：

docker compose up -d --build
docker compose ps            # 状态 healthy
curl -F "file=@cat.jpg" http://localhost:8000/predict

依赖 requirements.txt 用锁死版本的方式写，别用 >=：

fastapi==0.115.6
uvicorn[standard]==0.32.1
onnxruntime==1.19.2
Pillow==11.0.0
numpy==1.26.4
prometheus-client==0.21.1

七、⑥ 本地压测：先拿基线，再谈优化

用 wrk 先打一个粗糙基线（5 秒、4 线程、100 连接）。图片请求用 POST 不方便 wrk，先在服务里加一个 GET /predict_path?url= 不好——更简单：压 /healthz 拿「服务层裸吞吐」，再单独压推理函数。 更完整的做法是压一个内嵌测试图片的路径，工具与方法论见 压测与容量规划。

# 压服务层裸吞吐（不含推理）
wrk -t4 -c100 -d10s http://localhost:8000/healthz
# 结果里看 Requests/sec 与 Latency P99，先记录为「服务层基线」

# 压真实推理路径：用 locust 写脚本，POST 图片文件
pip install locust

# locustfile.py —— 真实推理路径压测
import io
from PIL import Image
from locust import HttpUser, task, between

class ImageUser(HttpUser):
    wait_time = between(0.05, 0.15)          # 模拟 5-15% 的请求间隔
    img_bytes = io.BytesIO()                  # 模块级生成一张固定测试图，避免压测机 IO 变瓶颈
    Image.new("RGB", (32, 32), (128, 64, 200)).save(img_bytes, format="PNG")

    @task
    def predict(self):
        self.client.post("/predict", files={"file": ("t.png", self.img_bytes.getvalue())})

压测的三个纪律（详见 压测与容量规划）：

1. 压测机不能成为瓶颈——测试图放内存，不在循环里读磁盘；
2. 预热至少 30-60 秒再记数，跳过 JIT/缓存热身阶段的虚高数据；
3. 记录环境：压测机规格、worker 数、并发数、时长，否则数字不可复现。

八、⑦ 接入 Prometheus 指标

用 prometheus-client 埋三类黄金指标：请求计数、延迟直方图、错误计数。

# app/metrics.py —— 全局唯一的指标对象
from prometheus_client import Counter, Histogram

REQUESTS = Counter("predict_requests_total", "推理请求总数",
                   ["model", "version"])
ERRORS = Counter("predict_errors_total", "推理失败数",
                 ["model", "version", "error_type"])
LATENCY = Histogram("predict_latency_seconds",
                    "推理延迟（秒）",
                    buckets=(0.005, 0.01, 0.02, 0.05, 0.1, 0.25, 0.5, 1.0))

def observe(latency_ms: float):
    REQUESTS.labels(model="mobilenetv2", version="v1").inc()
    LATENCY.observe(latency_ms / 1000.0)

在 main.py 里暴露 /metrics 给 Prometheus 抓取：

from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from fastapi import Response

@app.get("/metrics")
def metrics_endpoint():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

# prometheus.yml
scrape_configs:
  - job_name: image-classifier
    static_configs:
      - targets: ["api:8000"]        # compose 网络内用服务名
    scrape_interval: 10s

启动后 docker compose up -d prometheus，打开 http://localhost:9090 查询：

# 每秒请求数（5 分钟平均）
rate(predict_requests_total[5m])
# P99 延迟
histogram_quantile(0.99, sum by (le) (rate(predict_latency_seconds_bucket[5m])))

到这里，最小部署闭环已完整：能训练、能导出、能服务、能容器化、能压测、能观测。完整指标设计、告警规则与 Grafana 面板见 可观测性落地。

九、上线清单：别急着说「我部署好了」

上生产前逐项打勾，任何一项是「No」就先别上：

• [ ] 健康检查路径 /healthz 就绪，容器 healthcheck 配好；
• [ ] 结构化日志已接（JSON 格式，含 request_id，见 可观测性落地）；
• [ ] /metrics 有黄金三指标，Prometheus 抓取正常；
• [ ] 有回滚预案：旧镜像保留，一条命令能切回（见 发布策略：灰度与回滚）；
• [ ] 压测报告有结论：峰值 QPS、P99、饱和点；
• [ ] 模型版本与代码版本对应关系有记录（模型注册，见 MLOps 流水线）；
• [ ] 显存/内存上限已测，OOM 风险有预案（常见陷阱与反模式 坑④⑤）。

十、常见坑（本流程中最容易翻车的三个）

1. 预处理不一致：训练里 ToTensor+Normalize，服务端忘掉归一化 → 精度掉 20 个点还查不出原因。解法：预处理代码只在 train_export.py 定义一次，写进 meta.json，推理类读它。
2. 导出后没验证：ONNX 和 PyTorch 输出对不上就上线。解法：导出脚本末尾自动比对，np.allclose(atol=1e-4)。
3. 压测数字不可复现：没记环境、没预热、压测机 IO 成瓶颈。解法：把压测命令和结果写进 README（简历分析与包装 有模板）。

十一、进阶方向：从骨架到生产

这套骨架是地基，往上走有四条路，按你的目标选：

方向	怎么做	参考
性能	量化 INT8（先 PTQ）、换 TensorRT 引擎	模型优化实战、TensorRT 边缘部署
多模型/高吞吐	换 Triton，启用动态批处理	NVIDIA Triton 多模型服务
LLM	换 vLLM，PagedAttention + 连续批处理	vLLM 大模型推理服务、LLM 推理
规模化	K8s + KServe/BentoML，自动扩缩容	框架与平台怎么选

检查清单

• [ ] 七步闭环全部走通，每一步有产物、有验证；
• [ ] 离线基线 accuracy 记录在 meta.json，线上指标可对比；
• [ ] 推理类单测通过，预处理与训练完全一致；
• [ ] 容器 docker compose up 一条命令可用；
• [ ] 压测报告含环境、参数、QPS/P99 结论；
• [ ] 黄金指标在 Grafana 可见，告警有人响应。

延伸阅读

• 常见陷阱与反模式 —— 本流程每个步骤对应的翻车现场与根因
• 压测与容量规划 —— 从粗糙基线到容量公式的完整方法论
• 可观测性落地 —— 指标埋点、告警与日志的完整配置
• 学习路径：三条路线 —— 这套骨架在「系统精进」路线第 8 周的定位
• FastAPI + Docker 在线服务 —— 服务化的深度案例
• 简历分析与包装 —— 怎么把这套项目写进简历

参考资料

• PyTorch ONNX 导出文档：https://pytorch.org/docs/stable/onnx.html
• ONNX Runtime 官方文档：https://onnxruntime.ai/docs/
• FastAPI 官方文档：https://fastapi.tiangolo.com/
• prometheus/client_python：https://github.com/prometheus/client_python
• wrk：https://github.com/wg/wrk
• Locust 官方文档：https://docs.locust.io/