使用 Flask REST API 进行 Python 机器学习预测

本文是关于在机器学习或人工智能 (AI) 系统的上下文中使用 Python 进行实时预测，并使用 Flask REST API。 这里公开的架构可以看作是从概念验证 (PoC) 到机器学习应用程序的最小可行产品 (MVP) 的一种方式。

在设计实时解决方案时，Python 并不是人们能想到的第一选择。 但是由于 Tensorflow 和 Scikit-Learn 是 Python 支持的一些最常用的机器学习库，它在许多 Jupyter Notebook PoC 中使用起来很方便。

使该解决方案可行的原因在于，与预测相比，训练需要大量时间。 如果您将训练视为观看电影并预测有关问题的答案的过程，那么在每个新问题之后不必重新观看电影似乎是非常有效的。

训练是“电影”的一种压缩视图，预测是从压缩视图中检索信息。 无论电影是复杂的还是长的，它都应该非常快。

让我们用 Python 中的一个快速 [Flask] 示例来实现它！

通用机器学习架构

让我们首先概述一个通用的训练和预测架构流程：

首先，创建一个训练管道以根据目标函数了解过去的数据。

这应该输出两个关键元素：

1. 特征工程功能：训练时使用的转换应该在预测时重用。
2. 模型参数：最终选择的算法和超参数应该保存下来，以便在预测时重复使用

请注意，在训练期间完成的特征工程应小心保存，以便适用于预测。 在此过程中可能出现的许多其他问题中的一个常见问题是特征缩放，这对于许多算法来说都是必需的。

如果特征 X1 从值 1 缩放到 1000 并使用函数f(x) = x/max(X1)重新缩放到 [0,1] 范围，如果预测集的值是 2000，会发生什么？

应该提前考虑一些仔细的调整，以便映射函数返回一致的输出，这些输出将在预测时正确计算。

机器学习训练与预测

这里有一个主要问题需要解决。 为什么我们一开始就将训练和预测分开？

绝对正确的是，在机器学习示例和课程的上下文中，预先知道所有数据（包括要预测的数据），构建预测器的一种非常简单的方法是将训练和预测数据（通常称为一个测试集）。

然后，需要在“训练集”上进行训练并在“测试集”上进行预测以获得结果，同时对训练和测试数据进行特征工程，在相同且唯一的管道中进行训练和预测.

但是，在现实生活中的系统中，您通常有训练数据，而要预测的数据在处理时就进来了。 换句话说，你一次看电影，后来你有一些问题，这意味着答案应该简单快捷。

此外，通常不需要在每次新数据进入时重新训练整个模型，因为训练需要时间（对于某些图像集可能需要数周）并且随着时间的推移应该足够稳定。

这就是为什么在许多系统上训练和预测可以，甚至应该明确分开，这也更好地反映了智能系统（人工或非人工）的学习方式。

与过拟合的联系

训练和预测的分离也是解决过拟合问题的好方法。

在统计学中，过拟合是“产生的分析过于接近或精确地对应于一组特定的数据，因此可能无法拟合其他数据或可靠地预测未来的观察结果”。

过度拟合在具有许多特征的数据集或训练数据有限的数据集中尤为明显。 在这两种情况下，与预测器可以验证的数据相比，数据包含的信息太多，其中一些甚至可能与预测变量无关。 在这种情况下，噪声本身可以被解释为一个信号。

控制过度拟合的一个好方法是对部分数据进行训练，并在我们拥有基本事实的另一部分进行预测。 因此，如果我们训练的数据能够代表系统的实际情况及其未来状态，那么新数据的预期误差大致就是该数据集的测量误差。

因此，如果我们设计一个适当的训练和预测管道以及正确的数据拆分，我们不仅可以解决过度拟合问题，而且可以重用该架构来预测新数据。

最后一步是控制新数据的错误是否与预期相同。 总有一个转变（实际误差总是低于预期的），应该确定什么是可接受的转变——但这不是本文的主题。

用于预测的 REST API

这就是明确区分训练和预测的地方。 如果我们保存了特征工程方法和模型参数，那么我们可以使用这些元素构建一个简单的 REST API。

这里的关键是在 API 启动时加载模型和参数。 一旦启动并存储在内存中，每个 API 调用都会触发特征工程计算和 ML 算法的“预测”方法。 两者通常都足够快以确保实时响应。

API 可以设计为接受要预测的唯一示例，或几个不同的示例（批量预测）。

这是实现这一原则的最小 Python/Flask 代码，带有 JSON 输入和 JSON 输出（问题输入，回答输出）：

 app = Flask(__name__) @app.route('/api/makecalc/', methods=['POST']) def makecalc(): """ Function run at each API call No need to re-load the model """ # reads the received json jsonfile = request.get_json() res = dict() for key in jsonfile.keys(): # calculates and predicts res[key] = model.predict(doTheCalculation(key)) # returns a json file return jsonify(res) if __name__ == '__main__': # Model is loaded when the API is launched model = pickle.load(open('modelfile', 'rb')) app.run(debug=True)

请注意，该 API 可用于根据新数据进行预测，但我不建议将其用于训练模型。 可以使用它，但这会使模型训练代码复杂化，并且在内存资源方面可能要求更高。

实施示例 - 共享单车

让我们以 Kaggle 数据集，共享单车为例。 假设我们是一家共享单车公司，想要预测每天的自行车租赁数量，以便更好地管理自行车的维护、物流和其他方面的业务。

租金主要取决于天气状况，因此通过天气预报，该公司可以更好地了解租金何时达到高峰，并尽量避免在这些日子进行维护。

首先，我们训练一个模型并将其保存为可以在 Jupyter 笔记本中看到的泡菜对象。

模型训练和性能这里不涉及，这只是一个理解全过程的例子。

然后我们编写将在每次 API 调用时完成的数据转换：

 import numpy as np import pandas as pd from datetime import date def doTheCalculation(data): data['dayofyear']=(data['dteday']- data['dteday'].apply(lambda x: date(x.year,1,1)) .astype('datetime64[ns]')).apply(lambda x: x.days) X = np.array(data[['instant','season','yr','holiday','weekday','workingday', 'weathersit','temp','atemp','hum','windspeed','dayofyear']]) return X

这只是一个变量（一年中的一天）的计算，包括月份和精确的日期。 还有一些列的选择及其各自的顺序要保留。

然后，我们需要用 Flask 编写 REST API：

 from flask import Flask, request, redirect, url_for, flash, jsonify from features_calculation import doTheCalculation import json, pickle import pandas as pd import numpy as np app = Flask(__name__) @app.route('/api/makecalc/', methods=['POST']) def makecalc(): """ Function run at each API call """ jsonfile = request.get_json() data = pd.read_json(json.dumps(jsonfile),orient='index',convert_dates=['dteday']) print(data) res = dict() ypred = model.predict(doTheCalculation(data)) for i in range(len(ypred)): res[i] = ypred[i] return jsonify(res) if __name__ == '__main__': modelfile = 'modelfile.pickle' model = pickle.load(open(modelfile, 'rb')) print("loaded OK") app.run(debug=True)

运行这个程序，它将默认在端口 5000 上提供 API。

如果我们在本地测试一个请求，仍然使用 Python：

 import requests, json url = '[http://127.0.0.1:5000/api/makecalc/](http://127.0.0.1:5000/api/makecalc/)' text = json.dumps({"0":{"instant":1,"dteday":"2011-01-01T00:00:00.000Z","season":1,"yr":0,"mnth":1,"holiday":0,"weekday":6,"workingday":0,"weathersit":2,"temp":0.344167,"atemp":0.363625,"hum":0.805833,"windspeed":0.160446}, "1":{"instant":2,"dteday":"2011-01-02T00:00:00.000Z","season":1,"yr":0,"mnth":1,"holiday":0,"weekday":3,"workingday":0,"weathersit":2,"temp":0.363478,"atemp":0.353739,"hum":0.696087,"windspeed":0.248539}, "2":{"instant":3,"dteday":"2011-01-03T00:00:00.000Z","season":1,"yr":0,"mnth":1,"holiday":0,"weekday":1,"workingday":1,"weathersit":1,"temp":0.196364,"atemp":0.189405,"hum":0.437273,"windspeed":0.248309}})

该请求包含提供给模型的所有信息。 因此，我们的模型将响应指定日期的自行车租赁预测（这里我们有三个）。

 headers = {'content-type': 'application/json', 'Accept-Charset': 'UTF-8'} r = requests.post(url, data=text, headers=headers) print(r,r.text) <Response [200]> { "0": 1063, "1": 1028, "2": 1399 }

而已！ 该服务可轻松用于任何公司的应用程序，用于维护计划或让用户了解自行车交通、需求和出租自行车的可用性。

把它们放在一起

许多机器学习系统，尤其是 PoC 的主要缺陷是混合训练和预测。

如果它们被仔细分离，可以很容易地为 MVP 执行实时预测，使用 Python/Flask 的开发成本和工作量非常低，特别是对于许多 PoC，它最初是使用 Scikit-learn、Tensorflow 开发的，或任何其他 Python 机器学习库。

但是，这可能不适用于所有应用程序，尤其是特征工程繁重的应用程序，或者检索最接近匹配的应用程序需要在每次调用时获得最新数据。

无论如何，您是否需要一遍又一遍地看电影来回答有关它们的问题？ 同样的规则也适用于机器学习！