第 88 课 - FastAPI 高性能模型服务

1. 学习目标

通过本课的学习，你将能够：

理解FastAPI的核心优势：明白为什么FastAPI比Flask更适合构建生产级的机器学习模型服务。
构建基础的FastAPI模型服务：使用FastAPI创建一个完整的、包含模型加载和推理逻辑的REST API。
利用异步特性优化性能：通过async def和异步操作，提升服务在I/O密集型场景下的吞吐量。
应用依赖注入和中间件：掌握FastAPI的高级功能，如依赖注入管理共享资源（模型实例）和中间件（如CORS）。
生成自动交互式文档：学会使用Swagger UI和ReDoc为API自动生成清晰的文档。

2. 核心概念

FastAPI 是一个现代、快速（高性能）的Web框架，用于构建API。它基于 ASGI 标准（如Uvicorn或Daphne服务器），而Flask基于WSGI标准。这个根本区别带来了几个关键优势：

高性能：基于ASGI，天然支持异步（async/await），能够高效处理大量并发请求，非常适合I/O密集型任务（如调用其他微服务、数据库操作）。
开发体验与效率：通过 Python类型提示 来定义请求体、路径参数、查询参数等。这不仅提供了自动数据验证和转换，还让代码更清晰、更易维护。
自动文档：它能根据你的代码和类型提示，自动生成两个交互式API文档页面（Swagger UI 和 ReDoc），极大方便了前后端联调和测试。
依赖注入系统：一个强大的工具，可以轻松地管理共享资源（例如我们的ML模型实例）、处理数据库连接、实现认证等。

简单来说，FastAPI就像是Flask的“增强版”和“现代化版”，在保持简洁的同时，提供了企业级应用所需的高性能和高级功能。

3. 代码示例

我们将复用上一课Flask的示例，但使用FastAPI重写，并展示其高级特性。

前提：确保安装了所需库。

pip install fastapi uvicorn scikit-learn

完整代码 (fastapi_model_server.py):

import joblib
import numpy as np
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from contextlib import asynccontextmanager
from fastapi.middleware.cors import CORSMiddleware

# 1. 定义请求和响应的数据模型 (Pydantic)
class PredictionRequest(BaseModel):
    features: list[float]  # 例如: [5.1, 3.5, 1.4, 0.2]

class PredictionResponse(BaseModel):
    prediction: int
    probability: list[float] | None = None  # 类型提示，可以是列表或None

# 2. 使用 FastAPI 的 lifespan 管理模型加载 (推荐方式)
# lifespan 替代了旧的 @app.on_event("startup") 装饰器
ml_model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 加载模型
    global ml_model
    model_path = "iris_model.joblib"  # 假设模型已保存在此路径
    ml_model = joblib.load(model_path)
    print("机器学习模型已加载。")
    yield
    # 清理资源 (如有必要)
    ml_model = None
    print("模型资源已释放。")

# 3. 创建 FastAPI 应用实例，并绑定 lifespan
app = FastAPI(
    title="Iris Flower Classifier API",
    description="使用FastAPI部署的鸢尾花分类模型。",
    version="1.0.0",
    lifespan=lifespan
)

# 4. 添加中间件 (例如：CORS)
# 允许所有源访问（仅用于开发！生产环境需配置具体域名）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 5. 定义一个简单的依赖项（函数），用于获取模型实例
# 这是FastAPI依赖注入的一个应用示例
async def get_model():
    if ml_model is None:
        raise HTTPException(status_code=503, detail="模型尚未加载或加载失败。")
    return ml_model

# 6. 定义路由和请求处理函数
@app.get("/", summary="健康检查", tags=["系统"])
async def read_root():
    return {"message": "FastAPI 模型服务运行正常"}

@app.post("/predict", summary="模型预测", response_model=PredictionResponse, tags=["预测"])
async def predict(
    request: PredictionRequest,
    model = Depends(get_model)  # 通过依赖注入获取模型
):
    """
    输入特征值，返回鸢尾花分类结果。
    - **features**: 4个特征值的列表，顺序为 [花萼长度, 花萼宽度, 花瓣长度, 花瓣宽度]
    """
    try:
        # 将输入列表转换为模型所需的二维数组
        input_data = np.array([request.features])
        # 进行预测
        prediction = model.predict(input_data)[0]
        # 获取概率（如果模型支持）
        proba = model.predict_proba(input_data)[0].tolist() if hasattr(model, 'predict_proba') else None

        return PredictionResponse(prediction=int(prediction), probability=proba)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"预测时发生错误: {str(e)}")

# 7. 一个演示异步的端点（模拟I/O操作）
@app.get("/status-async", summary="异步状态检查", tags=["系统"])
async def async_status_check():
    import asyncio
    # 模拟一个耗时的I/O操作（如数据库查询、调用另一个API）
    await asyncio.sleep(1)
    return {"status": "all_systems_operational", "latency": "1s (simulated)"}

运行与测试:

# 运行服务 (使用uvicorn)
uvicorn fastapi_model_server:app --reload --host 0.0.0.0 --port 8000

启动后，访问 http://127.0.0.1:8000/docs 即可看到自动生成的 Swagger UI 交互式文档，你可以直接在里面测试你的API。

4. 实践练习

练习1 (基础): 添加一个新的GET端点 在上述代码中添加一个 /features 端点。它应该接受一个查询参数 name (例如 ?name=sepal_length)，并返回该特征的描述信息。使用Pydantic模型来定义响应。 预期输出示例: {"name": "sepal_length", "description": "花萼长度（厘米）"}

练习2 (进阶): 实现批量预测 创建一个新的端点 /predict-batch。它应该接受一个 PredictionRequest 的列表（即多个样本），并返回每个样本的预测结果列表。思考如何设计请求和响应的数据结构。 预期输出示例: {"predictions": [{"prediction": 0, "probability": [...]}, {"prediction": 1, "probability": [...]}]}

练习3 (挑战): 添加请求日志中间件 编写一个FastAPI中间件，记录每个请求的方法、路径和处理时间。将其应用到应用中。提示：可以使用 BaseHTTPMiddleware 或直接在 app 上使用 @app.middleware("http")。

5. 常见错误

忽略异步：即使你的函数内部是同步阻塞的（如普通的scikit-learn推理），使用 async def 定义路由也可以。FastAPI会在一个外部线程池中运行它，不会阻塞事件循环。但如果你想进一步优化I/O操作，才需要将其改为真正的异步调用。
忘记类型提示：FastAPI的核心功能依赖于类型提示。不添加类型提示，你将失去自动文档、自动验证和编辑器自动补全的便利。
混淆 @app.on_event 和 lifespan：lifespan 是管理应用生命周期资源的更现代、更推荐的方式。@app.on_event 虽然仍可用，但官方更推荐 lifespan。
生产环境使用 allow_origins=["*"]：在开发时，allow_origins=["*"] 允许所有源很方便。但在生产环境，必须将其配置为你的前端域名（如 ["https://your-frontend.com"]），以防止跨站请求伪造攻击。
将模型加载代码放在函数内部：模型加载是一个昂贵的操作，应该在应用启动时执行一次，而不是在每个请求时都执行。本例使用 lifespan 正确地实现了这一点。

6. 小结

本课我们学习了如何使用 FastAPI 来构建高性能的机器学习模型服务，完成了从Flask的过渡。

FastAPI优势：基于ASGI，性能高；基于类型提示，开发体验好，自带验证和文档；依赖注入系统强大。
构建服务三要素：1. 使用Pydantic模型定义数据结构；2. 使用路径操作装饰器（@app.get, @app.post）定义路由；3. 使用**lifespan** 管理应用生命周期（如加载模型）。
高级特性应用：我们演示了如何使用依赖注入来管理共享的模型实例，以及如何添加中间件（如CORS）。
异步支持：async def 函数能够充分利用ASGI服务器的能力，是构建高并发服务的关键。

FastAPI已经成为构建Python API的事实标准之一，尤其适合需要高性能和良好开发体验的机器学习模型部署场景。掌握它，将为你的模型上线之旅提供一个强大的工具。下一课，我们将学习如何将这个FastAPI服务容器化，使用Docker进行打包和部署。

练习编辑器

rust