引言：AI模型需要一条“通往生产的路”

在Notebook里跑通模型推理，和在生产环境中提供一个稳定、可扩展的AI推理API，中间隔着一条不小的鸿沟。

在本地，你手动导入模型、调用model.predict()、看到输出，一切都很美好。但到了生产环境，你要面对的是：多个用户同时请求、依赖版本冲突、环境不一致导致“在我机器上能跑”、模型加载耗时拖垮首次请求……这些问题，需要一个工程化的方案来解决。

本文的核心目标很直接：手把手带你构建一个基于FastAPI的AI模型推理服务，并用Docker把它容器化，让这个服务能在任何环境中稳定运行。我们将以一个图像分类场景为例，完整展示从零开始到服务启动的全过程。

---

一、为什么是FastAPI + Docker？

1.1 FastAPI：为AI推理量身打造的Web框架

在众多Python Web框架中，FastAPI之所以成为AI推理API的首选，有几个关键原因：

• 高性能：基于Starlette和Pydantic，异步支持让它在处理I/O密集型任务时表现优异，这对需要同时处理多个推理请求的场景至关重要。
• 自动生成API文档：启动服务后，/docs和/redoc端点自动生成交互式文档，前端同事或测试人员可以直接在浏览器中试用API。
• 类型安全：基于Python类型提示的请求/响应验证，能在请求到达业务逻辑之前就过滤掉格式错误的数据。

1.2 Docker：解决“环境漂移”的终极方案

AI项目的依赖管理是出了名的复杂——PyTorch版本、CUDA驱动、cuDNN版本、各Python包的兼容性……Docker通过将整个运行环境打包成镜像，确保了开发、测试、生产环境的一致性。

核心价值：你不再需要在新服务器上重新安装CUDA、配置环境变量、解决依赖冲突——只需要docker run，服务就跑起来了。

---

二、项目结构：从零搭建

我们以一个图像分类服务为例，完整的项目结构如下：

ai-inference-api/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI应用入口
│   ├── config.py            # 配置管理（Pydantic Settings）
│   ├── models/
│   │   ├── __init__.py
│   │   ├── schemas.py       # 请求/响应的Pydantic模型
│   │   └── predictor.py     # 模型加载与推理逻辑
│   └── routes/
│       ├── __init__.py
│       └── inference.py     # /predict路由定义
├── models/                  # 模型权重文件存放目录
│   └── resnet50.onnx        # 示例：导出的ONNX模型
├── requirements.txt
├── Dockerfile
├── .dockerignore
└── docker-compose.yml       # 可选：本地开发编排

---

三、核心代码实现

3.1 配置管理：告别硬编码

# app/config.py
from pydantic_settings import BaseSettings
from pathlib import Path

class Settings(BaseSettings):
    """应用配置，从环境变量读取"""
    model_path: Path = Path("./models/resnet50.onnx")
    model_input_name: str = "input"
    model_output_name: str = "output"
    image_size: int = 224
    log_level: str = "INFO"
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

3.2 数据模型：Pydantic定义接口契约

# app/models/schemas.py
from pydantic import BaseModel
from typing import Optional

class PredictRequest(BaseModel):
    """推理请求体"""
    image_url: Optional[str] = None      # 图片URL
    image_base64: Optional[str] = None   # Base64编码的图片
    
class PredictResponse(BaseModel):
    """推理响应体"""
    predicted_class: str
    confidence: float
    top5: Optional[list[dict]] = None    # Top-5预测结果

3.3 模型加载与推理：使用ONNX Runtime

选择ONNX而非直接加载PyTorch模型，原因在于ONNX Runtime在CPU上推理性能优异，且模型格式标准化，便于跨平台部署。

# app/models/predictor.py
import numpy as np
import onnxruntime as ort
from PIL import Image
import io
import requests
from typing import Tuple, List
from app.config import settings

class ImageClassifier:
    """图像分类器，封装ONNX模型推理"""
    
    def __init__(self):
        # 初始化ONNX Runtime会话
        self.session = ort.InferenceSession(str(settings.model_path))
        
        # 加载类别标签
        self.classes = self._load_classes()
        
        # 预热：执行一次空推理，确保模型就绪
        dummy_input = np.random.randn(1, 3, settings.image_size, settings.image_size).astype(np.float32)
        _ = self.session.run([settings.model_output_name], {settings.model_input_name: dummy_input})
        print(f"Model loaded and warmed up. Input: {settings.model_input_name}, Output: {settings.model_output_name}")
    
    def _load_classes(self) -> List[str]:
        """从文件加载ImageNet类别名"""
        class_path = settings.model_path.parent / "imagenet_classes.txt"
        with open(class_path, "r") as f:
            return [line.strip() for line in f.readlines()]
    
    def preprocess(self, image_bytes: bytes) -> np.ndarray:
        """图片预处理：调整大小、归一化、转CHW格式"""
        img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
        img = img.resize((settings.image_size, settings.image_size))
        
        img_np = np.array(img).astype(np.float32) / 255.0
        # 标准化（ImageNet标准均值与标准差）
        mean = np.array([0.485, 0.456, 0.406])
        std = np.array([0.229, 0.224, 0.225])
        img_np = (img_np - mean) / std
        # HWC → CHW
        img_np = img_np.transpose(2, 0, 1)
        # 添加batch维度
        return img_np[None, ...].astype(np.float32)
    
    def predict(self, image_bytes: bytes) -> Tuple[str, float, List[dict]]:
        """执行推理，返回预测类别、置信度和Top-5"""
        input_tensor = self.preprocess(image_bytes)
        
        outputs = self.session.run(
            [settings.model_output_name],
            {settings.model_input_name: input_tensor}
        )
        
        probs = outputs[0][0]
        top5_indices = np.argsort(probs)[-5:][::-1]
        
        top_class_idx = top5_indices[0]
        top_class = self.classes[top_class_idx]
        top_confidence = float(probs[top_class_idx])
        
        top5 = [
            {"class": self.classes[idx], "confidence": float(probs[idx])}
            for idx in top5_indices
        ]
        
        return top_class, top_confidence, top5

3.4 FastAPI路由：依赖注入与生命周期管理

模型实例应当在服务启动时加载一次，而不是在每个请求中重新加载。FastAPI提供了lifespan机制来实现这一模式。

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.models.predictor import ImageClassifier
from app.routes import inference

# 全局模型实例
model: ImageClassifier = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理：启动时加载模型，关闭时清理资源"""
    global model
    print("Loading model...")
    model = ImageClassifier()
    app.state.model = model  # 存入app.state，供依赖注入获取
    print("Model loaded successfully.")
    yield
    # 可在此添加清理逻辑
    print("Shutting down...")

app = FastAPI(
    title="AI Image Classifier API",
    description="基于ONNX + FastAPI的图像分类服务",
    version="1.0.0",
    lifespan=lifespan
)

# 允许跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(inference.router)

# app/routes/inference.py
from fastapi import APIRouter, Request, HTTPException
from fastapi import File, UploadFile
from app.models.schemas import PredictResponse
from app.models.predictor import ImageClassifier

router = APIRouter(prefix="/api/v1", tags=["inference"])

def get_model(request: Request) -> ImageClassifier:
    """依赖注入：从app.state获取模型实例"""
    model = request.app.state.model
    if model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    return model

@router.post("/predict", response_model=PredictResponse)
async def predict(request: Request, file: UploadFile = File(...)):
    """图片分类推理接口"""
    # 验证文件类型
    if not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="File must be an image")
    
    # 读取文件
    image_bytes = await file.read()
    if len(image_bytes) == 0:
        raise HTTPException(status_code=400, detail="Empty file")
    
    # 执行推理
    model = get_model(request)
    try:
        predicted_class, confidence, top5 = model.predict(image_bytes)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Inference failed: {str(e)}")
    
    return PredictResponse(
        predicted_class=predicted_class,
        confidence=confidence,
        top5=top5
    )

@router.get("/health")
async def health(request: Request):
    """健康检查端点"""
    model = get_model(request)
    return {"status": "healthy", "model_loaded": model is not None}

为什么模型加载放在lifespan里？

在没有lifespan机制之前，常见的做法是在模块顶层加载模型，但这会导致uvicorn在导入时就开始加载，可能阻塞启动流程。使用lifespan，模型在应用启动完成后再加载，并可通过app.state全局访问，既清晰又安全。

---

四、Docker化：让服务随处可跑

4.1 Dockerfile最佳实践

选择python:3.11-slim而非Alpine作为基础镜像，因为科学计算库在musl libc上存在兼容风险。

# Dockerfile
# 多阶段构建：减少最终镜像体积
FROM python:3.11-slim AS builder

WORKDIR /build

# 安装编译依赖（部分Python包需要编译）
RUN apt-get update && apt-get install -y \
    gcc \
    gfortran \
    libopenblas-dev \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖并安装
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# ---- 运行时阶段 ----
FROM python:3.11-slim

WORKDIR /app

# 从builder复制已安装的包
COPY --from=builder /root/.local /root/.local

# 设置PATH
ENV PATH=/root/.local/bin:$PATH

# 复制应用代码
COPY app/ ./app/
COPY models/ ./models/

# 复制启动脚本或直接运行
COPY docker-entrypoint.sh .
RUN chmod +x docker-entrypoint.sh

# 暴露端口
EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s \
    CMD python -c "import requests; requests.get('http://localhost:8000/api/v1/health')" || exit 1

ENTRYPOINT ["./docker-entrypoint.sh"]

# docker-entrypoint.sh
#!/bin/bash
uvicorn app.main:app --host 0.0.0.0 --port 8000

4.2 requirements.txt

fastapi[all]==0.104.1
uvicorn[standard]==0.24.0
onnxruntime==1.15.1
numpy==1.24.3
Pillow==10.1.0
pydantic-settings==2.1.0
requests==2.31.0

4.3 构建与运行

# 构建镜像
docker build -t ai-classifier:latest .

# 运行容器
docker run -d -p 8000:8000 --name ai-service ai-classifier:latest

# 测试API
curl -X POST http://localhost:8000/api/v1/predict \
  -F "file=@cat.jpg"

4.4 使用docker-compose（推荐本地开发）

# docker-compose.yml
version: '3.8'

services:
  inference:
    build: .
    ports:
      - "8000:8000"
    environment:
      - LOG_LEVEL=DEBUG
    volumes:
      # 本地代码挂载，实现热更新（开发模式）
      - ./app:/app/app
      - ./models:/app/models
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

---

五、关键设计决策解读

5.1 为什么模型要“预热”？

在predictor.__init__中执行一次dummy推理，这被称为“预热”（warm-up）。其作用是确保ONNX Runtime完成所有初始化操作（如内存分配、算子优化），避免第一个真实请求经历额外的冷启动延迟。生产环境中，这能将首次请求的响应时间从数秒降低到毫秒级。

5.2 为什么选择ONNX而非原生PyTorch？

ONNX Runtime是一个跨平台推理引擎，在CPU上经过大量优化，性能往往优于PyTorch的CPU推理。同时，ONNX模型是可移植的——同一个.onnx文件可以在不同框架、不同硬件上运行。

5.3 依赖注入模式的价值

通过Depends(get_model)将模型实例注入路由，而不是在路由内部直接访问全局变量，带来了几个好处：

• 可测试性：单元测试时可以注入Mock模型
• 清晰的依赖关系：路由函数的参数明确表达了它需要什么
• 统一错误处理：如果模型未加载，get_model抛出503，所有路由统一响应

---

六、生产环境扩展建议

6.1 增加请求队列与并发控制

当多个请求同时到达时，如果模型不支持并行推理（如GPU显存限制），需要引入队列机制。可以使用asyncio.Queue或Celery实现异步任务队列。

6.2 添加监控指标

暴露Prometheus指标，包括：

• 请求总数与延迟分布
• 推理耗时
• 模型加载状态
• GPU利用率（如果使用GPU）

6.3 灰度发布与回滚

在生产环境中，模型更新不应直接全量替换。可以通过容器镜像标签管理不同版本，配合Kubernetes的滚动更新策略，实现灰度发布。

---

总结

从零开始，我们完成了一个完整的AI推理服务：

1. FastAPI提供了高性能的API层和自动文档
2. ONNX Runtime实现了高效的模型推理
3. Docker保证了环境一致性，让服务可随处部署

这套方案的核心思想是将AI模型当作软件工程的一部分来对待——配置管理、依赖注入、生命周期管理、容器化——这些在传统后端开发中已是常识，在AI工程化中同样不可或缺。