构建可用的预测服务结合了API、Flask Web框架以及在Python应用中加载已保存机器学习模型的过程。详细的指导将引导此类服务的创建。这个动手练习将逐步指导您使用Flask创建一个简单的Web API。这个API将加载一个预训练模型(比如您在第二章中保存的模型),并提供一个端点,该端点通过HTTP接收输入数据,使用模型进行预测,并返回结果。前提条件在开始编码之前,请确保您已准备好以下各项:Python: 确保您的系统上已安装Python 3。Flask: 您需要安装Flask库。如果尚未安装,请打开您的终端或命令提示符并运行:pip install Flask joblib scikit-learn pandas(我们包含joblib、scikit-learn和pandas,因为它们常用于保存/加载模型和处理数据,如果您的模型使用不同的库如pickle,请进行调整)。已保存模型: 您需要一个使用joblib或pickle保存到文件的训练好的机器学习模型。对于本例,我们假设您有一个名为model.joblib的模型文件,它保存在您将创建Flask应用的同一目录中。这个模型应该经过训练,能够根据特定的输入特征进行预测。我们还将假设这个模型需要以Pandas DataFrame的形式作为输入。(可选)已保存的预处理器: 如果您的模型需要单独保存的特定预处理步骤(如缩放或编码)(例如,scaler.joblib或完整的pipeline.joblib),也请准备好该文件。为了本例的简单起见,我们假设模型文件包含必要的步骤,或者预处理足够简单,可以直接在Flask应用中完成。项目结构让我们保持文件整洁。为您的项目创建一个新目录,例如simple_ml_api。在这个目录中,放置您保存的模型文件(model.joblib)。您将在同一目录中创建您的Flask应用脚本,命名为app.py。simple_ml_api/ ├── app.py # 您的Flask应用代码 └── model.joblib # 您保存的模型文件步骤1:创建基本的Flask应用(app.py)在您的项目目录中打开一个名为app.py的新文件,首先导入必要的库并初始化Flask:import joblib import pandas as pd from flask import Flask, request, jsonify # 初始化Flask应用 app = Flask(__name__) # 加载训练好的模型(如果您有预处理器也一并加载) # 我们在应用启动时加载一次,而不是在路由函数内部 # 这样做更高效,因为它避免了每次请求都重新加载。 try: model = joblib.load("model.joblib") print("模型加载成功。") # 如果您有单独的预处理器,请在此处加载: # preprocessor = joblib.load("preprocessor.joblib") except FileNotFoundError: print("错误:未找到model.joblib。请确保模型文件位于正确的目录中。") model = None except Exception as e: print(f"加载模型时出错:{e}") model = None 这里,我们导入Flask用于Web服务器,request用于处理传入数据,jsonify用于创建JSON响应,joblib用于加载模型,以及pandas因为许多模型期望数据以DataFrame格式。我们使用app = Flask(__name__)初始化Flask应用。最重要的是,我们在初始化应用后立即加载model.joblib文件。这意味着模型在应用启动时只加载到内存一次。如果在预测函数内部加载,效率会非常低,因为它会为每一个预测请求重新从磁盘加载模型。我们还添加了基本的错误处理,以防模型文件丢失或无法加载。步骤2:定义预测端点现在,我们来创建处理预测请求的特定URL端点。我们将使用/predict路由,并指定它应接受HTTP POST请求,因为客户端将向其发送数据。# 将此代码添加到app.py中模型加载代码的下方 @app.route('/predict', methods=['POST']) def predict(): # 检查模型是否加载成功 if model is None: return jsonify({"error": "Model not loaded or failed to load."}), 500 # 1. 从POST请求中获取数据 try: data = request.get_json(force=True) print(f"收到的数据:{data}") # 记录收到的数据 # 确保数据是预期格式(例如,一个字典) if not isinstance(data, dict): raise ValueError("Input data must be a JSON object (dictionary).") # 2. 为模型准备数据 # 假设模型期望一个带有特定列名的Pandas DataFrame # 根据您模型的训练数据调整列名 # 示例:{'特征1': 值1, '特征2': 值2, ...} feature_values = list(data.values()) feature_names = list(data.keys()) # 或者明确定义期望的列 input_df = pd.DataFrame([feature_values], columns=feature_names) print(f"准备好的DataFrame:\n{input_df}") # 记录DataFrame # 如果您有预处理器,您将在此处应用它: # input_processed = preprocessor.transform(input_df) # prediction = model.predict(input_processed) # 3. 进行预测 prediction = model.predict(input_df) # 如果需要,将预测转换为标准的Python类型(例如,从numpy数组) # 确保输出可JSON序列化 output = prediction[0] if hasattr(output, 'item'): # 处理numpy类型 output = output.item() print(f"预测结果:{output}") # 记录预测结果 # 4. 将预测作为JSON响应返回 return jsonify({"prediction": output}) except ValueError as ve: print(f"值错误:{ve}") return jsonify({"error": f"Invalid input data format: {ve}"}), 400 except KeyError as ke: print(f"错误:{ke}") return jsonify({"error": f"Missing expected feature in input data: {ke}"}), 400 except Exception as e: # 捕获处理或预测过程中可能发生的其他错误 print(f"发生错误:{e}") return jsonify({"error": "An error occurred during prediction."}), 500 让我们分解一下predict函数:检查模型: 首先,它验证model在启动时是否加载成功。如果没有,则返回错误。获取数据: request.get_json(force=True)尝试将传入的请求正文解析为JSON。force=True有助于在内容类型未明确设置为application/json时进行解析,但客户端设置该类型是一个好习惯。我们添加了基本验证,检查接收到的数据是否为字典。准备数据: 这一步非常重要,完全取决于您的模型是如何训练的。许多scikit-learn模型期望输入是二维数组状结构(如Pandas DataFrame或NumPy数组),并带有特定顺序的特征列。这里,我们假设输入JSON是一个字典,如{"feature1": 1.0, "feature2": 2.5, ...}。我们将其转换为单行Pandas DataFrame。您必须调整列名和数据准备,以符合您的具体模型要求。 如果您保存了预处理器或管道,您将在此处应用其transform方法。预测: 我们调用model.predict()方法,传入准备好的数据(input_df)。格式化输出: 预测结果通常以NumPy数组的形式返回(即使是单个预测)。我们提取第一个元素(prediction[0]),并在必要时使用.item()将其转换为标准Python类型,确保它可以轻松转换为JSON。返回响应: jsonify({"prediction": output})创建一个包含预测结果的JSON响应。错误处理: try...except块会捕获潜在问题,例如缺失JSON数据、数据格式不正确(例如,缺少特征)或预测步骤本身中的错误,并返回带有相应HTTP状态码(客户端错误为400,服务器错误为500)的信息性JSON错误消息。步骤3:添加运行服务器的代码最后,添加标准的Python构造,使脚本可运行并启动Flask开发服务器:# 将此代码添加到app.py的末尾 if __name__ == '__main__': # 设置host='0.0.0.0'以使服务器可从网络上的其他设备访问 # 如果需要,使用与默认5000不同的端口 app.run(host='0.0.0.0', port=5000, debug=True)这段代码检查脚本是否被直接执行(而不是被导入)。app.run()启动Flask开发服务器。host='0.0.0.0'使服务器监听所有可用的网络接口,而不仅仅是localhost。如果您想从网络上的其他设备进行测试,或最终在容器中运行,这会很有用。port=5000指定端口号(5000是Flask的默认端口)。debug=True启用调试模式。这会在浏览器中提供更详细的错误消息,并在您保存代码更改时自动重启服务器。重要: 由于安全风险和性能开销,请勿在生产环境中使用debug=True。步骤4:运行您的Flask API现在您已准备好运行您的预测服务!打开您的终端或命令提示符。导航到您的项目目录(simple_ml_api)。确保您的model.joblib文件存在。运行应用:python app.py您应该会看到输出,表明模型已成功加载,并且Flask服务器正在运行,通常类似于:模型加载成功。 * 正在提供Flask应用 'app' * 调试模式:开启 警告:这是一个开发服务器。请勿在生产部署中使用它。请改用生产WSGI服务器。 * 运行在所有地址 (0.0.0.0) * 运行在 http://127.0.0.1:5000 * 运行在 http://[您的本地IP]:5000 按CTRL+C退出 * 使用stat重启 模型加载成功。 * 调试器已激活! * 调试器PIN码:...您的API现在已运行并在端口5000监听请求!步骤5:测试您的API您需要一种方式向正在运行的API发送带有JSON数据的POST请求。您可以使用curl(一个命令行工具)或使用requests库编写一个简单的Python脚本。示例输入数据:我们假设您的model.joblib期望四个特征,分别为sepal_length、sepal_width、petal_length和petal_width。您的输入JSON应如下所示:{ "sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2 }使用curl测试:打开另一个终端窗口(让第一个窗口继续运行服务器)并执行以下命令。如果需要,替换特征值。curl -X POST -H "Content-Type: application/json" \ -d '{"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2}' \ http://127.0.0.1:5000/predict-X POST:将HTTP方法指定为POST。-H "Content-Type: application/json":告诉服务器请求体包含JSON数据。-d '{...}':在请求体中提供JSON数据。http://127.0.0.1:5000/predict:您的API端点URL。使用Python requests测试:或者,您可以创建一个小的Python脚本(例如,test_api.py)或使用交互式Python会话:import requests import json # 您的Flask API端点URL url = 'http://127.0.0.1:5000/predict' # 输入数据,Python字典格式 # 根据您的模型调整特征名称和值 input_data = { "sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2 } # 发送带有JSON数据的POST请求 response = requests.post(url, json=input_data) # 检查请求是否成功(状态码200) if response.status_code == 200: # 打印API的JSON响应(预测) result = response.json() print(f"API响应: {result}") # 示例输出可能为:API响应:{'prediction': 0} 或 {'prediction': 'setosa'} else: # 如果请求失败,打印错误信息 print(f"错误: {response.status_code}") try: print(f"错误详情: {response.json()}") except json.JSONDecodeError: print(f"错误详情: {response.text}") 运行此脚本(python test_api.py)。预期输出:如果一切正常,curl和Python脚本都应收到来自您的API的JSON响应,类似于以下内容(实际预测值取决于您的模型):{ "prediction": 0 }或者(如果您的模型预测类别名称):{ "prediction": "setosa" }您还应该在运行app.py的终端中看到日志消息,显示接收到的数据、准备好的DataFrame和预测结果。digraph G { rankdir=LR; node [shape=box, style=filled, fillcolor="#e9ecef", fontname="sans-serif"]; edge [color="#495057", fontname="sans-serif"]; Client [label="客户端\n(curl 或 Python requests)", shape=oval, fillcolor="#a5d8ff"]; Flask [label="Flask 服务器 (app.py)\n运行在端口 5000", fillcolor="#bac8ff"]; Model [label="已加载模型\n(model.joblib)", shape=cylinder, fillcolor="#96f2d7"]; subgraph cluster_flask { label = "Flask应用内部"; style=filled; color="#dee2e6"; Flask; Model; } Client -> Flask [label=" POST /predict \n Content-Type: application/json \n Body: {'特征': 值, ...} "]; Flask -> Model [label=" 调用 model.predict(input_df) "]; Model -> Flask [label=" 返回预测结果 "]; Flask -> Client [label=" HTTP 200 OK \n Body: {'prediction': 结果} "]; }流程图显示客户端向正在运行的Flask应用的/predict端点发送带有JSON数据的POST请求。应用处理数据,使用加载的模型进行预测,并将结果作为JSON响应返回给客户端。完整的app.py示例代码以下是app.py的完整代码,以便查阅:import joblib import pandas as pd from flask import Flask, request, jsonify # 初始化Flask应用 app = Flask(__name__) # 加载训练好的模型 try: model = joblib.load("model.joblib") print("模型加载成功。") except FileNotFoundError: print("错误:未找到model.joblib。") model = None except Exception as e: print(f"加载模型时出错:{e}") model = None @app.route('/predict', methods=['POST']) def predict(): # 检查模型是否加载成功 if model is None: return jsonify({"error": "Model not loaded or failed to load."}), 500 # 1. 从POST请求中获取数据 try: data = request.get_json(force=True) print(f"收到的数据:{data}") if not isinstance(data, dict): raise ValueError("Input data must be a JSON object (dictionary).") # 2. 为模型准备数据 # 重要提示:调整特征名称以匹配您模型的训练数据 feature_values = list(data.values()) feature_names = list(data.keys()) # 或者使用预定义的列表:['sepal_length', 'sepal_width', ...] input_df = pd.DataFrame([feature_values], columns=feature_names) print(f"准备好的DataFrame:\n{input_df}") # 3. 进行预测 prediction = model.predict(input_df) # 4. 格式化输出 output = prediction[0] if hasattr(output, 'item'): # 处理numpy类型 output = output.item() print(f"预测结果:{output}") # 5. 将预测作为JSON响应返回 return jsonify({"prediction": output}) except ValueError as ve: print(f"值错误:{ve}") return jsonify({"error": f"Invalid input data format: {ve}"}), 400 except KeyError as ke: print(f"错误:{ke}") return jsonify({"error": f"Missing expected feature in input data: {ke}"}), 400 except Exception as e: print(f"发生错误:{e}") return jsonify({"error": "An error occurred during prediction."}), 500 if __name__ == '__main__': # 运行应用,可在网络上访问,并开启调试模式 # 记住在生产环境中将debug设置为False app.run(host='0.0.0.0', port=5000, debug=True)恭喜!您已成功使用Flask构建了一个基本的机器学习预测API。该服务将您保存的模型包装在一个Web服务器中,通过HTTP接受输入数据并返回预测结果。这是使您的模型可供其他应用或用户使用的基本模式。在下一章中,我们将讨论如何使用Docker打包此应用,使其更具可移植性且更易于部署。