4. FastAPI 基础：安装、路由、请求与响应模型、Pydantic 数据校验

在实时风控系统中，Serving 层需要以毫秒级延迟处理海量请求，并提供高并发、高可用的 HTTP 接口。FastAPI 凭借其异步原生支持、自动数据校验与 OpenAPI 文档生成，成为风控模型 Serving 的首选框架。本章将系统讲解 FastAPI 的核心基础，为后续构建生产级风控接口打下坚实根基。

4.1 FastAPI 安装与环境配置

FastAPI 基于 Python 3.8+，推荐使用虚拟环境隔离项目依赖。安装命令如下：

# 创建并激活虚拟环境（以 venv 为例）
python -m venv risk_serving_env
source risk_serving_env/bin/activate  # Linux/Mac
# risk_serving_env\Scripts\activate   # Windows

# 安装 FastAPI 与生产级 ASGI 服务器 uvicorn
pip install fastapi uvicorn[standard]

# 可选：安装 Pydantic 数据校验扩展（FastAPI 已内置，但可单独升级）
pip install pydantic[email,dotenv]

安装完成后，可通过以下命令验证版本：

python -c "import fastapi; print(fastapi.__version__)"

4.2 路由：定义风控接口的入口

路由（Route）将 HTTP 方法与 URL 路径映射到具体的 Python 函数（即路径操作函数）。FastAPI 使用装饰器语法声明路由，支持 GET、POST、PUT、DELETE 等所有标准 HTTP 方法。

from fastapi import FastAPI

app = FastAPI(title="实时风控模型 Serving API", version="1.0.0")

# GET 路由：健康检查
@app.get("/health")
async def health_check():
    return {"status": "ok", "service": "risk-serving"}

# POST 路由：风控评分入口
@app.post("/v1/score")
async def get_score():
    # 暂未实现具体逻辑
    return {"score": 0.85, "decision": "pass"}

路径参数与查询参数是风控接口中常用的参数传递方式：

• 路径参数：用于标识资源，如 /user/{user_id}/risk，适合固定资源定位。
• 查询参数：用于过滤或可选参数，如 /score?amount=5000&channel=web，适合非必填字段。

# 路径参数示例：查询指定用户的黑名单状态
@app.get("/user/{user_id}/blacklist")
async def check_blacklist(user_id: str):
    # 模拟查询
    return {"user_id": user_id, "in_blacklist": False}

# 查询参数示例：带可选参数的风控查询
@app.get("/v1/check")
async def check_risk(amount: float, channel: str = "default"):
    return {"amount": amount, "channel": channel, "risk_level": "low"}

4.3 请求与响应模型：Pydantic 数据校验

Pydantic 是 FastAPI 的数据校验核心。通过定义数据模型（继承 BaseModel），FastAPI 自动完成请求体的解析、类型校验、字段约束与错误提示。在风控场景中，这确保了进入模型的特征数据格式正确、范围合理。

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

# 请求模型：风控评分输入
class RiskScoreRequest(BaseModel):
    user_id: str = Field(..., min_length=8, max_length=32, description="用户唯一标识")
    amount: float = Field(..., gt=0, le=1000000, description="交易金额（元）")
    channel: str = Field(default="web", pattern="^(web|app|api)$", description="渠道来源")
    device_fingerprint: Optional[str] = Field(None, max_length=64, description="设备指纹")

    # 自定义校验：金额不能为负数（Field 的 gt 已保证，此处仅为示例）
    @field_validator("amount")
    @classmethod
    def amount_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError("amount must be positive")
        return v

# 响应模型：风控评分输出
class RiskScoreResponse(BaseModel):
    score: float = Field(..., ge=0, le=1, description="风险评分 0-1")
    decision: str = Field(..., pattern="^(pass|review|reject)$", description="决策结果")
    reason: Optional[str] = Field(None, max_length=256, description="拒绝原因")

在路由中使用 Pydantic 模型：

@app.post("/v1/score", response_model=RiskScoreResponse)
async def get_score(request: RiskScoreRequest):
    # 模拟风控模型计算
    risk_score = min(1.0, request.amount / 100000)
    decision = "reject" if risk_score > 0.8 else "pass"
    return RiskScoreResponse(
        score=risk_score,
        decision=decision,
        reason="金额过高触发风控" if decision == "reject" else None
    )

4.4 请求与响应模型的核心机制

下图展示了 FastAPI 处理一个 POST 请求的完整数据流，从 HTTP 请求到 Pydantic 校验，再到业务逻辑与响应序列化：

4.5 字段校验与自定义验证器

Pydantic v2 提供了丰富的字段校验方式，风控场景中常用以下约束：

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

class TransactionRequest(BaseModel):
    user_id: str = Field(..., min_length=8, max_length=32)
    amount: float = Field(..., gt=0, le=1_000_000)
    channel: str = Field(..., pattern="^(web|app|api|pos)$")
    id_card: Optional[str] = Field(None, pattern=r"^\d{17}[\dXx]$")

    @field_validator("amount")
    @classmethod
    def check_amount_precision(cls, v):
        # 金额最多保留两位小数
        if round(v, 2) != v:
            raise ValueError("amount must have at most 2 decimal places")
        return v

    @field_validator("user_id")
    @classmethod
    def user_id_no_special_chars(cls, v):
        if not v.isalnum():
            raise ValueError("user_id must be alphanumeric")
        return v

4.6 启动服务与交互式文档

使用 uvicorn 启动 FastAPI 应用，推荐开发模式开启热重载：

# 启动服务，监听 8000 端口
uvicorn main:app --reload --host 0.0.0.0 --port 8000

启动后，FastAPI 自动生成两套交互式 API 文档：

• Swagger UI：http://localhost:8000/docs — 可视化测试接口，支持直接发送请求并查看响应。
• ReDoc：http://localhost:8000/redoc — 更清晰的文档视图，适合分享给团队。

4.7 完整示例：风控评分接口

综合本章知识点，实现一个完整的实时风控评分接口：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, field_validator
from typing import Optional

app = FastAPI(title="实时风控模型 Serving", version="1.0.0")

# ---------- 数据模型 ----------
class RiskRequest(BaseModel):
    user_id: str = Field(..., min_length=8, max_length=32)
    amount: float = Field(..., gt=0, le=1_000_000)
    channel: str = Field(..., pattern="^(web|app|api)$")
    device_id: Optional[str] = Field(None, max_length=64)

    @field_validator("amount")
    @classmethod
    def precision_check(cls, v):
        if round(v, 2) != v:
            raise ValueError("amount supports only 2 decimal places")
        return v

class RiskResponse(BaseModel):
    score: float = Field(..., ge=0, le=1)
    decision: str = Field(..., pattern="^(pass|review|reject)$")
    reason: Optional[str] = None

# ---------- 模拟风控模型 ----------
def mock_risk_model(user_id: str, amount: float, channel: str) -> tuple:
    # 实际项目中替换为 ML 模型推理
    base_score = min(1.0, amount / 500000)
    if channel == "api":
        base_score *= 1.2  # API 渠道风险略高
    score = min(1.0, base_score)
    if score > 0.8:
        return score, "reject", "高风险交易"
    elif score > 0.5:
        return score, "review", "需人工审核"
    return score, "pass", None

# ---------- 路由 ----------
@app.post("/v1/risk/score", response_model=RiskResponse)
async def risk_score(request: RiskRequest):
    score, decision, reason = mock_risk_model(
        request.user_id, request.amount, request.channel
    )
    return RiskResponse(score=score, decision=decision, reason=reason)

@app.get("/health")
async def health():
    return {"status": "healthy", "model_version": "1.0.0"}