4. RESTful API 服务架构：Flask/FastAPI 搭建精算模型评分接口

精算模型从离线训练到线上实时评分，核心环节是构建一个稳定、低延迟的 RESTful API 服务。本章聚焦于使用 Python 生态中最流行的两个 Web 框架——Flask 与 FastAPI——搭建模型评分接口，并详细定义输入输出的 JSON Schema 设计规范，确保接口的可读性、可维护性与类型安全。

4.1 框架选型对比：Flask vs FastAPI

Flask 轻量灵活，生态成熟；FastAPI 原生支持异步、自动生成 OpenAPI 文档，且内置数据校验。下表从精算模型服务的关键维度进行对比：

对比维度 Flask FastAPI
性能（高并发） 同步 WSGI，需配合 Gunicorn + 多 worker 原生异步 ASGI，支持高并发 I/O 场景
数据校验 需手动或借助 Marshmallow/Pydantic 内置 Pydantic 模型，自动校验与序列化
API 文档 需集成 Flasgger 或手动编写 自动生成 Swagger UI / ReDoc
模型加载方式 全局变量 / Application Context Lifespan 事件 / 依赖注入
学习成本 低，文档丰富 中，需理解异步与类型注解
适用场景 中小规模、快速原型、传统精算团队 高吞吐、实时风控、微服务架构
选型建议： 若团队已有 Flask 技术栈且并发要求不高，Flask 足够胜任；若面向高频实时评分（如车险实时核保、健康险在线核赔），推荐 FastAPI 以获得更好的异步性能与自动文档能力。

4.2 JSON Schema 设计原则

精算模型评分接口的输入输出必须严格定义，避免因字段缺失、类型错误或边界值导致服务异常。设计时应遵循以下原则：

  • 字段命名： 统一使用 snake_case，与 Python 变量风格一致。
  • 类型明确： 数值型使用 float 或 int，类别型使用 string 并枚举合法值。
  • 必填与可选： 使用 required 列表标记必填字段，可选字段提供默认值或 null。
  • 范围约束： 对年龄、保额、保费等字段设置 minimum / maximum。
  • 版本控制： 在 URL 路径或请求头中携带版本号，如 /v1/score。

4.3 输入 JSON Schema 示例（车险评分）

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "AutoInsuranceScoreInput",
  "type": "object",
  "properties": {
    "policy_id": {
      "type": "string",
      "description": "保单唯一标识"
    },
    "driver_age": {
      "type": "integer",
      "minimum": 18,
      "maximum": 100,
      "description": "驾驶员年龄"
    },
    "vehicle_value": {
      "type": "number",
      "minimum": 0,
      "description": "车辆价值（元）"
    },
    "claim_history": {
      "type": "integer",
      "minimum": 0,
      "description": "过去3年出险次数"
    },
    "coverage_type": {
      "type": "string",
      "enum": ["comprehensive", "third_party", "minimum"],
      "description": "险种类型"
    },
    "region_code": {
      "type": "string",
      "pattern": "^[A-Z]{2}$",
      "description": "地区代码（两位大写字母）"
    }
  },
  "required": ["policy_id", "driver_age", "vehicle_value", "coverage_type"]
}

4.4 输出 JSON Schema 示例

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "AutoInsuranceScoreOutput",
  "type": "object",
  "properties": {
    "policy_id": {
      "type": "string",
      "description": "请求对应的保单ID"
    },
    "risk_score": {
      "type": "number",
      "minimum": 0,
      "maximum": 100,
      "description": "风险评分（0-100，越高风险越大）"
    },
    "predicted_loss_cost": {
      "type": "number",
      "minimum": 0,
      "description": "预测损失成本（元）"
    },
    "score_components": {
      "type": "object",
      "properties": {
        "age_factor": { "type": "number" },
        "vehicle_factor": { "type": "number" },
        "claim_factor": { "type": "number" }
      },
      "description": "评分因子分解"
    },
    "model_version": {
      "type": "string",
      "description": "模型版本号"
    },
    "timestamp": {
      "type": "string",
      "format": "date-time",
      "description": "评分时间戳（ISO 8601）"
    }
  },
  "required": ["policy_id", "risk_score", "model_version", "timestamp"]
}
关键设计点： 输出中始终返回 policy_id 用于请求溯源；score_components 提供可解释性，便于精算师审核；model_version 与 timestamp 支持线上回滚与监控。

4.5 FastAPI 实现精算评分接口

以下代码展示如何使用 FastAPI + Pydantic 模型快速搭建评分端点：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, conint, confloat
from typing import Optional
import joblib
import numpy as np
from datetime import datetime

app = FastAPI(title="精算模型评分服务", version="1.0.0")

# 全局加载模型（生产环境建议使用Lifespan事件）
model = joblib.load("actuarial_model_v1.pkl")

# ---------- 输入模型 ----------
class ScoreInput(BaseModel):
    policy_id: str = Field(..., description="保单唯一标识")
    driver_age: conint(ge=18, le=100) = Field(..., description="驾驶员年龄")
    vehicle_value: confloat(ge=0) = Field(..., description="车辆价值（元）")
    claim_history: conint(ge=0) = Field(0, description="过去3年出险次数")
    coverage_type: str = Field(..., pattern="^(comprehensive|third_party|minimum)$")
    region_code: Optional[str] = Field(None, pattern="^[A-Z]{2}$")

# ---------- 输出模型 ----------
class ScoreOutput(BaseModel):
    policy_id: str
    risk_score: confloat(ge=0, le=100)
    predicted_loss_cost: confloat(ge=0)
    score_components: dict
    model_version: str = "1.0.0"
    timestamp: str

@app.post("/v1/score", response_model=ScoreOutput)
async def predict_score(input_data: ScoreInput):
    try:
        # 特征工程（示例）
        features = np.array([
            input_data.driver_age,
            input_data.vehicle_value,
            input_data.claim_history,
            1 if input_data.coverage_type == "comprehensive" else 0,
            1 if input_data.coverage_type == "third_party" else 0
        ]).reshape(1, -1)

        # 模型推理
        risk_score = float(model.predict(features)[0])
        loss_cost = risk_score * 1200.0  # 示例换算

        return ScoreOutput(
            policy_id=input_data.policy_id,
            risk_score=round(risk_score, 2),
            predicted_loss_cost=round(loss_cost, 2),
            score_components={
                "age_factor": round(risk_score * 0.3, 4),
                "vehicle_factor": round(risk_score * 0.5, 4),
                "claim_factor": round(risk_score * 0.2, 4)
            },
            timestamp=datetime.utcnow().isoformat()
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
生产增强： 实际部署时应添加请求速率限制、超时控制、日志结构化（如 JSON 格式）、健康检查端点 /health 以及 Prometheus 指标暴露。

4.6 服务架构与请求流程

下图展示精算模型评分服务的完整请求链路与内部组件协作关系：

精算模型评分服务架构图 客户端 / App 负载均衡 (Nginx) FastAPI 服务 Pydantic 校验 特征工程 模型推理引擎 响应组装 模型存储 (S3/FS) 监控 / 日志 说明：请求经负载均衡分发至 FastAPI 实例，依次经过校验、特征工程、模型推理与响应组装，最终返回结构化评分结果。

4.7 接口测试与验证

使用 curl 或 Python requests 库验证接口正确性：

curl -X POST "http://localhost:8000/v1/score" \
  -H "Content-Type: application/json" \
  -d '{
    "policy_id": "POL2024001",
    "driver_age": 35,
    "vehicle_value": 150000.0,
    "claim_history": 1,
    "coverage_type": "comprehensive",
    "region_code": "BJ"
  }'

预期返回示例：

{
  "policy_id": "POL2024001",
  "risk_score": 42.57,
  "predicted_loss_cost": 51084.00,
  "score_components": {
    "age_factor": 12.771,
    "vehicle_factor": 21.285,
    "claim_factor": 8.514
  },
  "model_version": "1.0.0",
  "timestamp": "2025-03-21T14:30:00.123456"
}
常见错误： 输入字段类型不匹配（如 driver_age 传入字符串）、必填字段缺失、枚举值拼写错误（如 "comprehensive" 误写为 "compre"）。FastAPI 会自动返回 422 状态码及详细错误信息，建议在客户端做好预处理。

4.8 部署与性能考量

  • 并发模型： FastAPI 使用 Uvicorn 作为 ASGI 服务器，建议配合 Gunicorn + Uvicorn Worker 实现多进程。
  • 模型预热： 服务启动时预先加载模型到内存，避免首次请求冷启动延迟。
  • 超时设置： 评分接口建议设置 5-10 秒超时，防止模型推理异常导致连接挂起。
  • 批量评分： 对于高吞吐场景，可设计批量端点 /v1/score/batch，接受数组输入，利用向量化计算提升吞吐。
  • 安全加固： 添加 API Key 认证、请求体大小限制、CORS 策略，防止恶意调用。

通过本章内容，你已掌握使用 Flask/FastAPI 搭建精算模型评分接口的核心方法，并能够设计符合生产要求的 JSON Schema。下一环节将深入模型版本管理与灰度发布策略。

← 上一章 📖 返回目录 下一章 →