去年给一个做智能客服的团队做模型部署，他们用PyTorch训练了一个意图分类模型，准确率95%，然后卡在了部署这一步。用Flask跑，QPS只有30，延迟200ms，上线一周就扛不住并发了。换FastAPI之后，同样的模型QPS到了400+，延迟降到了30ms。这篇文章把这个过程整理出来，从API设计到性能优化，一步步带你把模型部署到生产环境。

为什么选FastAPI

做模型服务的API框架，主流选择有Flask、Django REST Framework、FastAPI、Tornado。在模型服务这个场景下，FastAPI的优势非常明显：

异步支持是最大的卖点；模型推理本身就是计算密集型的，如果API框架也是同步的，一个请求在推理，其他请求全部排队等；FastAPI基于Starlette，天然支持async，可以在推理等待IO的时候处理其他请求；

自动文档省去了维护Swagger的工作量；模型服务的API通常需要被前端或其他后端调用，OpenAPI格式的自动文档让对接成本降低了一大截；访问/docs就能看到完整的API文档，还能直接在里面测试；

类型验证用Pydantic做请求/响应的数据校验；模型输入通常是结构化的（文本、图片URL、特征向量），用Pydantic定义好schema，框架自动校验，不用手写一堆if判断；

性能在Python Web框架里算顶级；基准测试中FastAPI的吞吐量是Flask的2-3倍，接近Node.js和Go的水平；对于模型服务来说，性能瓶颈通常在推理本身而不是API层，但API层的开销越小越好；

基础服务搭建

项目结构

一个典型的模型服务项目结构：

model_service/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI入口
│   ├── models.py         # Pydantic数据模型
│   ├── inference.py      # 推理逻辑
│   └── middleware.py     # 中间件
├── model/                # 模型文件
│   └── model.pt
├── requirements.txt
└── Dockerfile

定义API接口

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional
import torch, time

app = FastAPI(title="意图分类服务", version="1.0.0")

class PredictRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=1000)
    top_k: int = Field(default=3, ge=1, le=10)

class ModelManager:
    def __init__(self):
        self.model = None
    def load(self, path):
        self.model = torch.load(path)
        self.model.eval()
    def predict(self, text, top_k=3):
        with torch.no_grad():
            # 推理逻辑
            return [{"label": "类别", "score": 0.95}]

manager = ModelManager()

@app.on_event("startup")
async def startup():
    manager.load("model/model.pt")

@app.post("/predict")
async def predict(request: PredictRequest):
    start = time.time()
    result = manager.predict(request.text, request.top_k)
    return {"predictions": result, "latency_ms": (time.time()-start)*1000}

生产环境优化

并发推理

默认情况下模型推理是串行的，用线程池可以实现并发：

from concurrent.futures import ThreadPoolExecutor
import asyncio

executor = ThreadPoolExecutor(max_workers=4)

@app.post("/predict")
async def predict(request: PredictRequest):
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        executor, manager.predict, request.text, request.top_k
    )
    return {"predictions": result}

健康检查和监控

@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "model_loaded": manager.model is not None,
        "gpu_available": torch.cuda.is_available()
    }

Docker部署

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

踩坑记录

模型加载放在请求里是最常见的错误；每次请求都加载模型，延迟直接爆表；一定要在启动时加载（用@app.on_event("startup")或lifespan）；

GPU显存不够时推理会报CUDA OOM；解决办法是限制并发数，或者用CPU推理；torch.no_grad()不能省，它能减少一半的显存占用；

序列化开销在大模型响应时不可忽略；如果返回的是大量向量数据，用MessagePack或Protobuf替代JSON，体积能缩小一半；

并发推理

from concurrent.futures import ThreadPoolExecutor
import asyncio

executor = ThreadPoolExecutor(max_workers=4)

@app.post("/predict")
async def predict(request):
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, manager.predict, request.text)
    return {"result": result}

Docker部署

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

健康检查

@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "model_loaded": manager.model is not None,
        "gpu": torch.cuda.is_available()
    }

踩坑记录

模型加载放在请求里是最常见的错误——每次请求都加载模型，延迟直接爆表。一定要在启动时加载。GPU显存不够时推理会报CUDA OOM，用torch.no_grad()减少显存占用，并限制并发数；

写在最后

FastAPI做模型服务的API层，是目前Python生态里性价比最高的选择。它足够快、足够简单、生态足够好。配合Docker和Kubernetes，可以搭建出一套可靠的模型服务体系。如果后续需要gRPC支持，FastAPI也能和grpcio搭配使用。