本章聚焦于将训练好的 scikit-learn 模型部署为在线推理 API，并提供从模型包装到性能优化的全面指南。核心概念包括使用 FastAPI 快速搭建服务、利用 Docker 容器化部署以及通过 nginx 实现流量控制。读者将学会如何将模型封装为 HTTP 服务，通过 Docker 部署并使用 docker-compose 管理多服务架构。此外，章节还介绍了性能优化的关键点，如模型预加载、调整 worker 数量、批量预测以及使用异步框架提升高并发处理能力。读者还将了解灰度发布策略，如通过 nginx 按比例分流新模型流量，以及使用 pyarmor 加密模型文件以保护敏感信息。完成学习后，读者能够独立完成模型部署、配置性能优化策略、实施灰度发布，并运用监控工具跟踪服务运行状态。

模型部署:从 pickle 到 REST API

训好模型不是终点, 让用户用上才是。这一章我们把 scikit-learn 模型包装成 HTTP 服务, 用 Docker 部署, 用 nginx 限流。

部署的 4 种姿势

方式	适用	例子
在线推理 API	实时请求	FastAPI + Docker
批处理	离线跑全量	Spark / Airflow
嵌入式	移动端/IoT	TFLite / ONNX Runtime
流式	持续到达的数据	Kafka + Flink

本章专注最常见的在线推理 API。

方案 1: 最快的 FastAPI 服务

# app.py
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI(title="ML Inference API")
model = joblib.load("model.pkl")

class Input(BaseModel):
    features: list[float]

class Output(BaseModel):
    prediction: int
    probability: float

@app.post("/predict", response_model=Output)
def predict(inp: Input):
    X = np.array(inp.features).reshape(1, -1)
    pred = model.predict(X)[0]
    prob = model.predict_proba(X)[0].max()
    return Output(prediction=int(pred), probability=float(prob))

@app.get("/health")
def health():
    return {"status": "ok"}

启动: uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

客户端调用

# curl 测试
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [5.1, 3.5, 1.4, 0.2]}'

# 输出
{"prediction": 0, "probability": 0.98}

Python 客户端:

import requests
r = requests.post("http://api.example.com/predict", json={"features": [5.1, 3.5, 1.4, 0.2]})
print(r.json())

方案 2: Docker 打包

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 装依赖 (单独一层, 利用缓存)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码和模型
COPY app.py .
COPY model.pkl .

EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# build
docker build -t ml-api:v1 .

# run
docker run -d --name ml-api -p 8000:8000 ml-api:v1

方案 3: docker-compose 多服务

# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      - MODEL_PATH=/models/model.pkl
    volumes:
      - ./models:/models:ro
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      retries: 3
  
  nginx:
    image: nginx:alpine
    ports: ["80:80"]
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - api

# nginx.conf
http {
    upstream api {
        server api:8000;
    }
    server {
        listen 80;
        location / {
            proxy_pass http://api;
            proxy_set_header X-Real-IP $remote_addr;
        }
        # 限流: 每秒 100 请求
        limit_req_zone $binary_remote_addr zone=api:10m rate=100r/s;
    }
}

启动: docker-compose up -d

性能优化: 4 个关键点

模型加载一次, 复用进程: 不要每个请求都 joblib.load
worker 数量: workers = 2 * CPU_cores + 1 (uvicorn 默认是 1)
批量预测: 客户端一次发 100 条, 比 100 次单条快 5-10x
异步框架: 高并发用 uvicorn[standard] (uvloop + httptools)

# 异步 + 批量
from fastapi import FastAPI
import asyncio

app = FastAPI()
model = joblib.load("model.pkl")  # 启动时加载一次

@app.post("/predict_batch")
async def predict_batch(items: list[Input]):
    X = np.array([item.features for item in items])
    preds = model.predict(X)
    probs = model.predict_proba(X).max(axis=1)
    return [{"prediction": int(p), "probability": float(pr)} for p, pr in zip(preds, probs)]

灰度发布: 10% 流量走新模型

import random

@app.post("/predict")
def predict(inp: Input):
    # 10% 流量用 v2, 90% 走 v1
    if random.random() < 0.1:
        model = model_v2
    else:
        model = model_v1
    
    return model.predict(...)

更稳的做法: nginx 按比例分流

split_clients $request_id $model_version {
    90% "v1";
    10% "v2";
}

模型加密:不让人看到权重

模型 .pkl 文件可能含敏感 IP, 用 pyarmor 加密:

pip install pyarmor
pyarmor gen --recursive app.py

加密后的 .py 反编译不了, 但 .pkl 还是要存安全的地方。

监控:请求延迟 + 错误率

import time
from prometheus_client import Counter, Histogram

REQUEST_COUNT = Counter("api_requests_total", "Total requests", ["endpoint", "status"])
REQUEST_LATENCY = Histogram("api_latency_seconds", "Request latency", ["endpoint"])

@app.middleware("http")
async def metrics_middleware(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    REQUEST_COUNT.labels(request.url.path, response.status_code).inc()
    REQUEST_LATENCY.labels(request.url.path).observe(duration)
    
    return response

# /metrics 暴露给 Prometheus 抓
@app.get("/metrics")
def metrics():
    return Response(prometheus_client.generate_latest(), media_type="text/plain")

部署清单

模型在 MLflow Registry 标 Production
FastAPI 启动时加载模型 (不是每个请求)
Dockerfile 多阶段, 装依赖单独一层
docker-compose.yml + nginx 反代
healthcheck 端点
/metrics 暴露 Prometheus 指标
限流 (nginx limit_req)
日志到 stdout (docker logs 看)
灰度发布 10% → 50% → 100%

小结

部署 = FastAPI 包装 + Docker 打包 + nginx 限流
模型加载放启动时, 不用每个请求 load
workers = 2*CPU + 1, 批量预测比单条快 5-10x
灰度发布用随机数或 nginx split_clients
监控: 延迟 (Histogram) + 错误率 (Counter) + Prometheus 抓

练习思考

用 Docker 部署一个 sklearn 模型, 用 curl 测试, 记录平均响应时间。
加 nginx 限流 (rate=10r/s), 用 ab 压测看 503 出现频率。
模型 .pkl 怎么安全分发? 想过用 S3 + 签名 URL 吗?

章末小测验

检验你对《模型部署:从 pickle 到 REST API》的掌握程度。

FastAPI 部署模型时, joblib.load 应该放在哪?

uwsgi/uvicorn 的 workers 数量经验值是?

学完这章, 你可能想看

同课程MLOps 入门

线上监控与数据漂移

性能监控、数据漂移检测、PSI / KS 检验, 报警链路。

40 分钟

讨论区(0)

加载评论中...