基于Flask与Gunicorn的线性回归模型API服务化部署实践
模型服务化架构设计
将训练完成的机器学习模型转化为可供外部调用的Web服务，是实现算法业务价值的核心环节。在此过程中，Flask因其轻量级和灵活性，常被用于构建RESTful API路由；而Gunicorn作为高性能的WSGI HTTP服务器，能够有效处理并发请求，保障推理服务的稳定性与吞吐量。
项目目录规划
为了保证代码的可维护性，建议将数据、模型产物与业务逻辑分离。以下是推荐的项目结构：
ml_deployment_project/
├── data/ # 存放原始数据集与测试集
├── models/ # 存放序列化后的模型文件
├── src/
│ ├── train.py # 模型训练与持久化脚本
│ ├── api_server.py # Flask API 服务端代码
│ └── test_client.py # 模拟客户端请求脚本
└── requirements.txt # 环境依赖清单
环境初始化
在开始部署前，需确保系统已安装Python 3.8及以上版本。创建虚拟环境并安装必要的依赖库：
python -m venv venv
source venv/bin/activate
pip install flask gunicorn scikit-learn pandas joblib requests
模型训练与持久化
首先，我们需要训练线性回归模型并将其序列化到磁盘，以便Web服务在启动时直接加载。以下脚本读取房地产数据集，完成训练后将模型保存至models/目录，并导出一份测试集。
import pandas as pd
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split
import joblib
import os
# 确保输出目录存在
os.makedirs("models", exist_ok=True)
os.makedirs("data", exist_ok=True)
# 加载特征与目标变量
raw_data = pd.read_csv("data/real_estate.csv")
features = raw_data.drop(columns=["Price"])
target = raw_data["Price"]
# 划分训练集与测试集
X_train, X_test, y_train, y_test = train_test_split(features, target, test_size=0.2, random_state=42)
# 实例化并训练线性回归模型
lr_model = LinearRegression()
lr_model.fit(X_train, y_train)
# 序列化模型与测试数据
joblib.dump(lr_model, "models/real_estate_lr.pkl")
X_test.to_csv("data/real_estate_test.csv", index=False)
print("模型训练完成并已持久化。")
构建Flask推理服务
接下来，使用Flask构建接收HTTP请求并返回预测结果的API端点。为了提升响应速度，模型应在应用启动时一次性加载到内存中，避免每次请求都读取磁盘。
from flask import Flask, request, jsonify
import joblib
import numpy as np
app = Flask(__name__)
# 预加载模型至全局内存
model_instance = joblib.load("models/real_estate_lr.pkl")
@app.route("/api/v1/predict", methods=["POST"])
def predict_price():
# 解析JSON请求体
payload = request.get_json()
if not payload or "features" not in payload:
return jsonify({"error": "Missing 'features' in request body"}), 400
# 转换输入数据并进行维度调整
input_array = np.array(payload["features"]).reshape(1, -1)
# 执行推理
prediction = model_instance.predict(input_array)
return jsonify({
"status": "success",
"predicted_value": float(prediction[0])
})
if __name__ == "__main__":
# 开发环境下的直接运行方式
app.run(host="0.0.0.0", port=5000, debug=True)
使用Gunicorn启动生产级服务
在实际生产环境中，不应直接使用Flask内置的开发服务器。Gunicorn可以通过多进程模型充分利用多核CPU资源，处理高并发请求。使用以下命令启动服务，绑定4个Worker进程并监听8000端口：
gunicorn -w 4 -b 0.0.0.0:8000 api_server:app
客户端调用与验证
服务启动后，可以通过编写Python客户端脚本，向API端点发送测试数据以验证推理链路是否畅通。以下代码从测试集中抽取部分样本，并逐一向服务器发起POST请求。
import requests
import pandas as pd
# 读取先前保存的测试集
test_dataset = pd.read_csv("data/real_estate_test.csv")
# 截取前5条数据作为测试样本
sample_rows = test_dataset.head(5).values.tolist()
api_endpoint = "http://127.0.0.1:8000/api/v1/predict"
for i, row_data in enumerate(sample_rows):
response = requests.post(api_endpoint, json={"features": row_data})
if response.status_code == 200:
result = response.json()
print(f"样本 {i+1} 预测结果: {result['predicted_value']:.2f}")
else:
print(f"样本 {i+1} 请求失败: {response.text}")
