FastAPI 模型服务化：从 Notebook 到生产 API—

模型训练完成后，调用方还需要稳定的输入格式、响应和错误提示。服务化要把预处理、推理和结果校验组织成接口，并处理超时、并发与模型版本。下面先做一个小接口，再补齐运行时容易遗漏的细节。

FastAPI 是 Python 模型服务化的标准答案：快（Starlette + Pydantic）、自带文档、类型安全。这篇按生产标准写一个推理服务，包括那些教程不讲的脏活。

前置阅读：建议先读模型压缩与部署、MLOps 入门、Python 环境管理。

最小骨架：能跑 vs 能上生产的距离

教程版（能跑）：

from fastapi import FastAPI
import joblib

app = FastAPI()
model = joblib.load("model.pkl")

@app.post("/predict")
def predict(features: list[float]):
    return {"prediction": model.predict([features])[0]}

生产版和它的差距在六件事：请求校验、生命周期管理、错误处理、并发模型、监控、配置化。逐个补。

生产级写法：六块拼图

① Pydantic 校验：把脏请求挡在门外

from pydantic import BaseModel, Field

class PredictRequest(BaseModel):
    features: list[float] = Field(..., min_length=10, max_length=10,
                                  description="10 维特征向量")
    request_id: str | None = None

class PredictResponse(BaseModel):
    prediction: int
    probability: float
    model_version: str

请求字段不对（长度错、类型错）FastAPI 自动返回 422——不让脏数据碰到模型。免费的 /docs 页面（OpenAPI 自动生成）联调时前后端都感激你。

② 生命周期：模型只加载一次

from contextlib import asynccontextmanager
from fastapi import FastAPI

ml = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    ml["model"] = joblib.load("model.pkl")          # 启动时加载
    ml["version"] = "v2026-08-30"
    yield
    ml.clear()                                       # 关闭时清理

app = FastAPI(lifespan=lifespan)

别在请求函数里加载模型——每个请求加载一次 500MB 的模型，延迟直接起飞。lifespan 保证「启动加载、全期复用」。

③ 并发模型：别让 GPU 推理被「并发」搞崩

FastAPI 的 def（线程池）和 async def（协程）选择：

CPU 推理（sklearn/小模型）：用 def——阻塞调用扔线程池，别堵事件循环。
GPU 推理：GPU 一次只能服务一个 kernel 流，多线程并发调 GPU 反而互相干扰。正确姿势：推理放单一 worker，用请求队列攒 batch：

import asyncio

batch_queue: asyncio.Queue = asyncio.Queue(maxsize=64)

async def batch_worker():
    """后台协程：攒 16 条或等 10ms，凑批推理"""
    while True:
        batch, futures = [], []
        item, fut = await batch_queue.get()
        batch.append(item); futures.append(fut)
        deadline = asyncio.get_event_loop().time() + 0.01
        while len(batch) < 16:
            try:
                item, fut = await asyncio.wait_for(batch_queue.get(),
                    timeout=max(0, deadline - asyncio.get_event_loop().time()))
                batch.append(item); futures.append(fut)
            except asyncio.TimeoutError:
                break
        results = ml["model"].predict_proba(batch)   # 一次 batch 推理
        for fut, r in zip(futures, results):
            fut.set_result(r.tolist())

动态批处理（dynamic batching）是推理服务的吞吐密码：batch=16 的 GPU 推理耗时远小于 16 次单条——vLLM 的连续批处理（推理优化篇）是同一思想。

④ 错误处理与超时

from fastapi import HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_handler(request, exc):
    logger.exception("inference failed")
    return JSONResponse(status_code=500,
        content={"error": "internal", "request_id": request.state.req_id})

@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest):
    try:
        proba = ml["model"].predict_proba([req.features])[0]
    except ValueError as e:
        raise HTTPException(422, f"特征校验失败: {e}")
    return PredictResponse(prediction=int(proba.argmax()),
                           probability=float(proba.max()),
                           model_version=ml["version"])

⑤ 监控埋点：没有指标的服务是黑盒

import time
from prometheus_client import Counter, Histogram, generate_latest

REQ_CNT = Counter("predict_requests_total", "请求数", ["status"])
REQ_LAT = Histogram("predict_latency_seconds", "推理延迟")

@app.middleware("http")
async def metrics_mw(request, call_next):
    t = time.time()
    resp = await call_next(request)
    REQ_LAT.observe(time.time() - t)
    REQ_CNT.labels(status=resp.status_code).inc()
    return resp

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

延迟 P99、QPS、错误率、模型版本——Grafana 看板的四大件。MLOps 篇的模型监控（PSI 漂移）从这里接特征日志。

⑥ 配置与多环境

模型路径、batch 大小、阈值全部走环境变量/pydantic-settings——镜像不变，配置切换环境。配合 Docker 部署，docker run -e MODEL_PATH=/models/v2/ 即可换模型。

压测：上线前必须过的关

# locust 或 wrk 压测，看三个数：P99 延迟、吞吐、错误率
locust -f load_test.py --host http://localhost:8000 -u 100 -r 10 -t 60s

压测的判读：延迟随并发线性涨 → 接近饱和；错误率突增 → 队列打满/超时；吞吐平台期 → 这就是容量上限，按它规划副本数。Uvicorn 多进程：uvicorn app:app --workers 4（CPU 推理），GPU 推理单进程多副本（每副本一张卡）。

踩坑排查

现象	原因	解法
第一个请求特别慢	模型懒加载/预热缺失	启动时跑一条 warmup 请求
并发上去延迟爆炸	GPU 被多线程争抢	单 worker + 队列攒 batch
内存持续上涨	请求里创建大对象未释放	检查全局状态；worker 定期重启
偶发 422 但前端说字段对	类型隐式转换失败（字符串数字）	Pydantic 严格模式 + 明确错误信息
多 worker 时指标翻倍	每进程一份 Prometheus 注册表	用 multiprocess 模式或单独 metrics 端口

练习

把一个 sklearn 模型包成 FastAPI 服务：含 Pydantic 校验、lifespan 加载、/metrics 端点。
实现动态批处理 worker，对比 batch=1 和 batch=16 的吞吐（用 locust 压）。
写 Dockerfile 打包服务，容器内跑通并压测，记录单副本容量。
故意发 100 个畸形请求，验证服务返回 422 且进程不崩——错误处理的生命力。

面试常问

Q：FastAPI 和 Flask 做模型服务的区别？ Flask 同步 + 无类型校验（手写），性能受制于 WSGI；FastAPI 异步原生（ASGI）、Pydantic 类型安全、自动 OpenAPI 文档、性能接近 Node/Go 量级。模型服务的主流选择已是 FastAPI；Flask 生态老项目维护可继续，新项目没必要。

Q：GPU 推理服务为什么不能简单多线程？ GPU kernel 执行默认单流串行，多线程调用只是排队且增加同步开销；多进程各自加载模型则显存 ×N。正解：单进程 + 请求队列 + 动态批处理，或多进程每进程一卡（多卡机器）。这也是 Triton Inference Server / vLLM 等专用推理服务器存在的意义——复杂调度别自己重造，规模大就换它们。

Q：模型热更新怎么做？ 三层方案：① 简单——重启服务（滚动更新避免停机）；② 双缓冲——加载新模型到备用变量，切换指针后释放旧模型（注意线程安全）；③ 模型仓库——从 MLflow Registry 按别名拉取，定时轮询版本变化自动加载。生产推荐 ①+③ 组合：注册表管版本，K8s 滚动更新管切换。

Q：推理服务的 SLO 怎么定？ 从业务倒推：用户交互场景 P99 < 200ms（含网络），异步批处理按吞吐。拆解预算：网络 20ms + 排队 + 推理 + 序列化。定了 SLO 再压测验证容量，按 P99 而不是平均值规划副本——平均值会掩盖长尾。

Q：什么时候该从 FastAPI 换专用推理服务器？ 信号：需要动态批处理的精细调度、多模型编排（ensemble）、多后端（PyTorch/TF/ONNX 混部）、张量并行——这些在 Triton/vLLM/TFServing 里开箱即用。FastAPI 的优势是灵活（任意 Python 逻辑），专用服务器的优势是推理调度深度。常见架构：FastAPI 做业务编排层，后面挂 Triton 做纯推理。

FastAPI 模型服务化：从 Notebook 到生产 API——ML 工程的最后一公里