4、基于FastAPI的模型API服务：FastAPI异步特性、Pydantic数据校验、自动文档生成

模型训练好了，总不能一直躺在Jupyter Notebook里吧？

得把它「请」出来，变成一个能随时响应的服务。我个人习惯用FastAPI来做这件事。为什么？因为它快、写起来爽，而且自带文档——这对我们风控工程师来说太重要了。你想想看，每次给业务方或者前端同学对接接口，光写API文档就能写到手抽筋，FastAPI直接帮你生成了，省心不少。

4.1 FastAPI的异步特性：别让I/O卡住你的模型

先聊聊异步。说白了，异步就是「不等」。

传统的同步服务，一个请求来了，CPU去查数据库、调特征服务、跑模型——这些操作大部分时间都在等I/O（磁盘读写、网络传输）。等的时候，这个线程就卡住了，别的请求进不来。异步就不一样了，它会在等I/O的时候，主动让出CPU，去处理别的请求。

我在项目中遇到过一个问题：模型服务上线后，并发一上来，响应时间就飙升。后来发现是同步写法导致的。换成异步后，同样的机器，QPS翻了一倍。

核心要点：FastAPI天生支持异步。如果你的模型推理是纯CPU计算（比如LightGBM、XGBoost），其实异步收益不大；但如果你的服务需要先调特征平台、查Redis缓存、再调模型，那异步就是刚需。

来看一个简单的异步接口示例：

from fastapi import FastAPI
import uvicorn
import asyncio

app = FastAPI()

@app.get("/health")
async def health_check():
    # 模拟一个异步I/O操作，比如查数据库
    await asyncio.sleep(0.1)
    return {"status": "ok"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

注意那个async def。只要函数前面加了async，FastAPI就会用异步事件循环来调度它。如果你的模型推理是同步的（比如用model.predict()），可以用run_in_executor把它丢到线程池里跑，避免阻塞事件循环。

小技巧：我曾经在线上踩过一个坑——在异步函数里直接调了同步的模型推理，结果高并发时事件循环被卡死。后来改成await loop.run_in_executor(None, model.predict, data)，问题就解决了。

4.2 Pydantic数据校验：把脏数据挡在门外

风控模型最怕什么？怕脏数据。特征缺失、类型不对、数值越界——这些都会让模型输出奇怪的结果，甚至直接报错。

Pydantic就是干这个的。它让你用Python的类型注解来定义数据模型，然后自动帮你做校验。说白了，就是给数据上个「安检门」。

from pydantic import BaseModel, Field, validator
from typing import Optional

class RiskRequest(BaseModel):
    user_id: str = Field(..., min_length=1, max_length=32, description="用户ID")
    age: int = Field(..., ge=18, le=100, description="年龄")
    income: float = Field(..., ge=0, description="月收入")
    loan_amount: float = Field(..., ge=100, le=1000000, description="借款金额")
    credit_score: Optional[int] = Field(None, ge=300, le=850, description="信用分")

    @validator("income")
    def income_must_be_reasonable(cls, v):
        if v > 1000000:
            raise ValueError("月收入超过100万？不太合理吧")
        return v

class RiskResponse(BaseModel):
    risk_score: float = Field(..., ge=0, le=1, description="风险评分")
    decision: str = Field(..., description="决策结果：approve/reject/review")
    reason: Optional[str] = Field(None, description="拒绝原因")

你看，Field里可以指定长度范围、数值范围，甚至自定义校验逻辑。比如income_must_be_reasonable这个校验器，就是防止有人传个离谱的数值进来。

注意：Pydantic的校验是在请求进入路由函数之前执行的。如果数据不合法，FastAPI会自动返回422状态码和详细的错误信息。这意味着你的模型根本不会接触到脏数据——这是好事，但也意味着你要提前想好，哪些字段是必须的，哪些可以容忍缺失。

我在项目中遇到过这样一个场景：业务方传过来的age字段有时候是字符串，有时候是整数。Pydantic默认会尝试类型转换（比如把"25"转成25），但如果你不想让它转，可以设置strict=True。我个人建议在风控场景下尽量用严格模式，避免隐式转换带来的坑。

4.3 自动文档生成：接口文档不用再手写了

FastAPI最让我喜欢的一点，就是它自动生成文档。你只要把Pydantic模型定义好，路由写清楚，文档就出来了。

默认有两个文档地址：

  • /docs — Swagger UI，交互式文档，可以直接在页面上测试接口
  • /redoc — ReDoc，更简洁的文档风格

为什么这对风控模型服务很重要？因为你的接口通常要对接多个下游系统——比如前端H5页面、APP端、甚至其他后端服务。每个对接方都需要知道：请求参数是什么？返回格式是什么？错误码怎么定义？

有了自动文档，你只需要维护代码，文档自动同步。改代码就是改文档，再也不用担心「文档和代码对不上」这种尴尬事了。

from fastapi import FastAPI, HTTPException
from typing import List

app = FastAPI(
    title="风控模型API服务",
    description="提供信用评分、反欺诈等模型推理服务",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

@app.post("/predict", 
          response_model=RiskResponse,
          summary="风险评分预测",
          description="输入用户特征，返回风险评分和决策建议")
async def predict(request: RiskRequest):
    # 这里调用你的模型
    # 假设模型返回一个风险分数
    risk_score = 0.75
    decision = "reject" if risk_score > 0.7 else "approve"
    return RiskResponse(risk_score=risk_score, decision=decision, reason="信用评分过低")

你看，我在@app.post里加了summary和description，这些都会显示在文档里。Pydantic模型里的description字段也会被自动提取，变成文档中的字段说明。

避坑指南：我曾经在文档里暴露了太多内部信息——比如模型版本号、特征列表。后来发现这有安全风险。建议在description里只写必要的业务说明，不要泄露模型细节。另外，生产环境可以考虑关闭/docs和/redoc，或者加上认证。

4.4 完整示例：一个带校验和文档的风控API

把上面这些串起来，就是一个完整的模型API服务了。我贴一个我常用的模板：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator
import uvicorn
import asyncio
import joblib
import numpy as np

# 加载模型（假设是训练好的LightGBM）
# model = joblib.load("lgb_model.pkl")

app = FastAPI(
    title="风控模型API",
    description="信用评分预测服务",
    version="2.0.0"
)

class PredictRequest(BaseModel):
    features: List[float] = Field(..., min_items=10, max_items=50, 
                                  description="特征向量，长度10-50")
    user_id: str = Field(..., min_length=1, max_length=32)

    @validator("features")
    def check_feature_range(cls, v):
        for i, f in enumerate(v):
            if f < -1e6 or f > 1e6:
                raise ValueError(f"特征{i}的值{f}超出合理范围")
        return v

class PredictResponse(BaseModel):
    score: float = Field(..., ge=0, le=1)
    level: str = Field(..., pattern="^(low|medium|high)$")
    user_id: str

@app.post("/v1/predict", 
          response_model=PredictResponse,
          summary="风险评分预测",
          description="输入特征向量，返回风险评分和等级")
async def predict(request: PredictRequest):
    try:
        # 异步执行模型推理
        loop = asyncio.get_event_loop()
        features = np.array(request.features).reshape(1, -1)
        # score = await loop.run_in_executor(None, model.predict, features)
        score = 0.65  # 模拟结果
        
        if score < 0.3:
            level = "low"
        elif score < 0.7:
            level = "medium"
        else:
            level = "high"
            
        return PredictResponse(score=score, level=level, user_id=request.user_id)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"模型推理失败: {str(e)}")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, workers=4)

这个例子涵盖了：异步推理、Pydantic校验、自动文档、错误处理。你把它跑起来，访问http://localhost:8000/docs就能看到漂亮的交互式文档了。

4.5 知识体系总览

下面这张图是我自己画的，把本章的核心逻辑串起来了：

FastAPI模型API服务核心架构 客户端请求 FastAPI 异步路由 async def predict() Pydantic数据校验 Field / validator 异步模型推理 run_in_executor 自动文档生成 /docs /redoc JSON响应返回

这张图展示了请求从客户端进来，经过FastAPI路由，分流到三个核心组件：Pydantic做数据校验、异步引擎做模型推理、自动文档生成器。最后汇聚成JSON响应返回给客户端。三个组件各司其职，缺一不可。

好了，这一章就聊到这儿。FastAPI的异步特性、Pydantic的数据校验、自动文档生成——这三个东西组合起来，基本就是一个生产级模型API服务的骨架了。你可以在本地跑跑看，感受一下那种「写完代码就有文档」的爽快感。

公众号:蓝海资料掘金营，微信deep3321
← 上一章 📖 返回目录 下一章 →