3、基于Flask的模型API服务：Flask框架搭建、模型加载、请求/响应设计、接口测试

好，咱们进入第三章。这一章我打算聊聊怎么把训练好的模型「端」到线上。说白了，就是让模型能接受外部请求，然后返回预测结果。

你可能觉得这不就是写个接口嘛，有什么难的？嗯，我刚开始也这么想。直到有一次，我部署的模型服务在线上跑了三天，突然内存爆了——后来发现是模型加载方式有问题，每次请求都重新加载一次。那叫一个惨。

所以这一章，我会把我在实际项目中踩过的坑、总结的经验，都揉进代码里。咱们从零开始，搭一个能用的模型API服务。

核心目标：用Flask搭建一个轻量级模型服务，支持模型加载、请求解析、预测响应、接口测试。不依赖复杂框架，适合快速上线。

3.1 Flask框架搭建——别小看这几行代码

Flask是Python里最轻量的Web框架之一。我选它，说白了就两个原因：快、简单。你想想看，风控模型服务通常不需要高并发，Flask完全够用。

先搭个基础骨架：

from flask import Flask, request, jsonify
import joblib
import numpy as np

app = Flask(__name__)

@app.route('/health', methods=['GET'])
def health_check():
    return jsonify({'status': 'ok', 'message': '服务运行中'})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)

这段代码，我建议你直接复制。为什么？因为我在项目里就是这么用的。host设成'0.0.0.0'，这样局域网内其他机器也能访问。debug=False，线上千万别开debug模式，否则会暴露敏感信息。

我的习惯：端口号我一般用5000或8080。如果跟其他服务冲突，可以在启动时通过环境变量传入，比如 PORT=5001 python app.py。

3.2 模型加载——一次加载，多次使用

模型加载是个坑。我曾经见过有人把模型加载写在请求处理函数里，结果每次请求都重新load一次，响应时间从50ms直接飙到5秒。你想想看，这谁受得了？

正确的做法是：在服务启动时加载模型，然后全局复用。

# 全局变量，服务启动时加载
model = None
model_path = './models/credit_risk_model.pkl'

def load_model():
    global model
    try:
        model = joblib.load(model_path)
        print(f'模型加载成功，路径：{model_path}')
    except Exception as e:
        print(f'模型加载失败：{str(e)}')
        model = None

# 在启动前加载
load_model()

这里要注意几点：

  • 模型文件路径最好用绝对路径，或者通过配置文件传入。我遇到过相对路径在线上找不到的情况，排查了半天。
  • 加载失败时，要给一个明确的错误提示。别让服务静默启动，然后接口返回500。
  • 如果模型很大（比如超过1GB），可以考虑用lazy loading，但大部分风控模型都在几百MB以内，直接加载就行。

避坑指南：我曾经把模型文件放在项目根目录，结果部署时忘了复制，服务启动成功但接口全挂。后来我改成在启动脚本里检查模型文件是否存在，不存在就直接报错退出。这样至少不会出现「服务活着但不能用」的尴尬局面。

3.3 请求/响应设计——接口规范是命根子

接口设计，说白了就是约定。前端怎么传数据？后端怎么返回结果？这些不商量好，后面全是扯皮。

我一般用这种结构：

请求格式（POST /predict）

{
    "features": [0.12, 0.45, 0.78, 0.23, 0.91],
    "feature_names": ["age", "income", "credit_score", "debt_ratio", "loan_amount"]
}

响应格式

{
    "code": 0,
    "message": "success",
    "data": {
        "prediction": 0.87,
        "probability": 0.87,
        "risk_level": "high"
    },
    "request_id": "20240315-001"
}

为什么这么设计？我解释一下：

  • code：0表示成功，非0表示错误。这样前端判断起来很方便。
  • message：给人类看的错误信息。别返回那种「Error 10086」之类的，没人看得懂。
  • data：真正的预测结果。我习惯把原始分数和风险等级都返回，方便下游使用。
  • request_id：每次请求的唯一标识。排查问题的时候，这个字段能救命。

来看看完整的预测接口实现：

import uuid
from flask import Flask, request, jsonify

@app.route('/predict', methods=['POST'])
def predict():
    # 1. 检查模型是否加载
    if model is None:
        return jsonify({
            'code': 500,
            'message': '模型未加载，请联系管理员',
            'data': None,
            'request_id': str(uuid.uuid4())
        }), 500

    # 2. 解析请求
    try:
        data = request.get_json()
        features = data.get('features')
        feature_names = data.get('feature_names')
    except Exception as e:
        return jsonify({
            'code': 400,
            'message': f'请求格式错误：{str(e)}',
            'data': None,
            'request_id': str(uuid.uuid4())
        }), 400

    # 3. 校验特征
    if not features or len(features) == 0:
        return jsonify({
            'code': 400,
            'message': '特征不能为空',
            'data': None,
            'request_id': str(uuid.uuid4())
        }), 400

    # 4. 预测
    try:
        features_array = np.array(features).reshape(1, -1)
        prediction = model.predict(features_array)[0]
        probability = model.predict_proba(features_array)[0][1]

        # 风险等级映射
        if probability >= 0.7:
            risk_level = 'high'
        elif probability >= 0.4:
            risk_level = 'medium'
        else:
            risk_level = 'low'

        return jsonify({
            'code': 0,
            'message': 'success',
            'data': {
                'prediction': float(prediction),
                'probability': float(probability),
                'risk_level': risk_level
            },
            'request_id': str(uuid.uuid4())
        })
    except Exception as e:
        return jsonify({
            'code': 500,
            'message': f'预测失败：{str(e)}',
            'data': None,
            'request_id': str(uuid.uuid4())
        }), 500

我的经验：异常处理一定要细。我见过很多接口，不管什么错误都返回「系统异常」，前端根本没法排查。我建议至少区分三类错误：请求格式错误（400）、模型未加载（500）、预测执行错误（500）。这样定位问题快很多。

3.4 接口测试——别等上线再后悔

接口写完了，怎么测？我一般用两种方式：

方式一：curl命令（快速验证）

curl -X POST http://localhost:5000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [0.12, 0.45, 0.78, 0.23, 0.91], "feature_names": ["age", "income", "credit_score", "debt_ratio", "loan_amount"]}'

方式二：Python脚本（批量测试）

import requests
import json

url = 'http://localhost:5000/predict'
test_cases = [
    {'features': [0.12, 0.45, 0.78, 0.23, 0.91]},
    {'features': [0.01, 0.02, 0.03, 0.04, 0.05]},
    {'features': [0.99, 0.98, 0.97, 0.96, 0.95]},
]

for i, case in enumerate(test_cases):
    response = requests.post(url, json=case)
    result = response.json()
    print(f'测试用例{i+1}: code={result["code"]}, risk_level={result["data"]["risk_level"]}')

我建议你至少测这几种情况：

  • 正常请求：特征完整，返回200
  • 空特征：返回400，提示特征不能为空
  • 特征数量不对：返回400，提示特征维度错误
  • 模型未加载：返回500，提示模型未加载

避坑指南：我曾经上线前只测了正常情况，结果线上有个请求传了字符串而不是数字，模型直接报错。后来我加了特征类型校验，确保每个特征都是float类型。这个校验虽然简单，但能挡住80%的异常请求。

3.5 完整服务结构

最后，我把整个服务的目录结构整理一下，方便你参考：

model_api/
├── app.py              # Flask主程序
├── models/
│   └── credit_risk_model.pkl  # 模型文件
├── config.py           # 配置文件
├── requirements.txt    # 依赖列表
└── test/
    └── test_api.py     # 接口测试脚本

requirements.txt内容：

flask==2.3.3
joblib==1.3.2
numpy==1.24.3
requests==2.31.0

嗯，这一章的内容就这些。从框架搭建到模型加载，从接口设计到测试验证，每一步我都把实际项目中的经验融进去了。你照着这个结构搭，基本不会出大问题。

记住一句话：模型服务不是写出来就完事了，能稳定运行、方便排查问题，才是真本事。

专注资料整理

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