3. RESTful API服务搭建：用Flask/FastAPI构建评分接口

各位同学，今天我们来聊聊模型部署中最关键的一环——API服务搭建。说白了，就是把我们训练好的风控模型，包装成一个能对外提供服务的接口。我做了这么多年风控，见过太多模型在Jupyter Notebook里跑得飞起，一到线上就各种翻车。嗯，这里面的坑，我今天一个一个给你们讲清楚。

核心要点：API服务是模型与业务系统之间的桥梁。设计得好，模型上线顺风顺水；设计得不好，运维同学天天找你喝茶。

3.1 为什么选择Flask和FastAPI？

我个人习惯用FastAPI做新项目，Flask维护老项目。为什么？

  • FastAPI：异步支持好，性能高，自动生成API文档。适合高并发的风控场景。
  • Flask：生态成熟，插件丰富。很多老牌风控系统还在用。

你想想看，一个风控评分接口，每秒可能要处理几百个请求。如果同步阻塞，那延迟就上去了。FastAPI的异步特性，在这时候就特别香。

3.2 请求与响应设计——别让前端同学骂你

我在项目中遇到过，前端同学拿着接口文档来找我，说「这个字段名怎么是驼峰的？我们统一用小写加下划线行不行？」。嗯，从那以后，我设计接口时都会先跟团队对齐规范。

一个标准的评分接口，请求和响应应该长这样：

# 请求体示例
{
  "user_id": "U123456",
  "features": {
    "age": 35,
    "income": 25000,
    "credit_history_length": 8,
    "existing_loans": 2,
    "late_payment_count": 1
  },
  "timestamp": "2024-01-15T10:30:00Z"
}

# 响应体示例
{
  "code": 0,
  "message": "success",
  "data": {
    "score": 725,
    "level": "A",
    "probability": 0.12,
    "reason_codes": ["R01", "R03"]
  },
  "request_id": "req_abc123"
}

我的经验：响应里一定要带request_id。出了问题，排查日志全靠它。我曾经因为没加这个字段，排查一个线上问题花了整整两天。

3.3 接口鉴权——别让坏人钻空子

风控接口暴露在外网，没有鉴权等于裸奔。我见过最离谱的案例，有人直接把评分接口挂在公网上，连个密码都没有。结果被人刷了几百万次，服务器直接挂了。

常用的鉴权方式：

方式 适用场景 安全性
API Key 内部系统调用 中等
JWT Token 外部第三方接入 高
OAuth2.0 开放平台 最高

我个人习惯用JWT。为什么？因为它无状态，不需要查数据库，性能好。而且可以携带用户信息，方便做日志审计。

# FastAPI JWT鉴权示例
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

app = FastAPI()
security = HTTPBearer()

SECRET_KEY = "your-secret-key-here"

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        payload = jwt.decode(
            credentials.credentials, 
            SECRET_KEY, 
            algorithms=["HS256"]
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token已过期")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="无效的Token")

@app.post("/api/v1/score")
async def get_score(
    request: ScoreRequest,
    user: dict = Depends(verify_token)
):
    # 业务逻辑
    pass

注意：密钥不要硬编码在代码里！我曾经见过有人把密钥提交到GitHub上，结果被爬虫抓取，整个系统被攻破。用环境变量或者密钥管理服务来存储。

3.4 限流——保护你的模型不被刷爆

限流这件事，说白了就是「别让一个用户把资源占完了」。风控模型的计算资源很贵，尤其是深度学习模型，一次推理可能就要几十毫秒。如果不限流，一个恶意用户就能把整个服务拖垮。

常用的限流算法：

  • 令牌桶：允许突发流量，适合风控场景
  • 漏桶：平滑流量，适合对延迟敏感的场景
  • 滑动窗口：精确控制，适合对精度要求高的场景

我在项目中用的是令牌桶+IP维度的限流。为什么？因为风控场景经常有突发流量，比如双十一大促。令牌桶允许一定程度的突发，不会误伤正常用户。

# FastAPI限流示例
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)

@app.post("/api/v1/score")
@limiter.limit("100/minute")  # 每分钟最多100次
async def get_score(request: ScoreRequest):
    # 业务逻辑
    pass

避坑指南：我曾经把限流阈值设得太低，结果双十一那天，正常用户都被限流了，业务方直接打电话骂我。后来我学乖了，限流阈值要留30%的余量，并且要加监控告警。

3.5 完整代码示例——一个能用的评分接口

说了这么多，我们来写一个能用的。这个接口我实际部署过，线上跑了半年，稳得很。

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np
import time
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="风控评分服务", version="1.0.0")

# 加载模型
model = joblib.load("credit_model.pkl")

class FeatureInput(BaseModel):
    age: float
    income: float
    credit_history_length: float
    existing_loans: int
    late_payment_count: int

class ScoreRequest(BaseModel):
    user_id: str
    features: FeatureInput
    timestamp: str = None

class ScoreResponse(BaseModel):
    code: int
    message: str
    data: dict
    request_id: str

@app.post("/api/v1/score", response_model=ScoreResponse)
async def get_score(request: ScoreRequest):
    start_time = time.time()
    request_id = f"req_{int(start_time * 1000)}"
    
    try:
        # 特征提取
        features = np.array([
            request.features.age,
            request.features.income,
            request.features.credit_history_length,
            request.features.existing_loans,
            request.features.late_payment_count
        ]).reshape(1, -1)
        
        # 模型推理
        probability = model.predict_proba(features)[0][1]
        score = int((1 - probability) * 1000)
        
        # 等级划分
        if score >= 800:
            level = "AAA"
        elif score >= 700:
            level = "AA"
        elif score >= 600:
            level = "A"
        else:
            level = "B"
        
        # 耗时监控
        elapsed = time.time() - start_time
        logger.info(f"Request {request_id} processed in {elapsed:.3f}s")
        
        return ScoreResponse(
            code=0,
            message="success",
            data={
                "score": score,
                "level": level,
                "probability": round(probability, 4),
                "reason_codes": ["R01"] if probability > 0.3 else []
            },
            request_id=request_id
        )
        
    except Exception as e:
        logger.error(f"Request {request_id} failed: {str(e)}")
        return ScoreResponse(
            code=-1,
            message=f"处理失败: {str(e)}",
            data={},
            request_id=request_id
        )

关键点：这个代码里我加了耗时监控和错误日志。线上排查问题，全靠这些日志。别嫌麻烦，等你被线上问题折磨过就懂了。

3.6 知识体系总览

下面这张图，是我做API服务搭建的心得总结。你照着这个结构来，基本不会出大问题。

RESTful API服务搭建知识体系 评分API服务 框架选择 请求与响应设计 接口鉴权 限流策略 Flask FastAPI JSON格式 统一响应体 API Key JWT Token 令牌桶 漏桶 滑动窗口 图：评分API服务知识体系结构

这张图把今天讲的内容串起来了。你从框架选择开始，到请求响应设计，再到鉴权和限流，每一步都有对应的技术选型和最佳实践。我建议你照着这个结构，自己动手搭一个试试。

最后说一句：API服务搭建这件事，看起来简单，但细节决定成败。我见过太多人栽在鉴权、限流这些「小」问题上。你把这些基础打牢了，后面模型上线、监控、迭代，都会顺很多。

专注资料整理

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