使用 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 機器學習庫。

但是，這可能不適用於所有應用程序，尤其是特徵工程繁重的應用程序，或者檢索最接近匹配的應用程序需要在每次調用時獲得最新數據。

無論如何，您是否需要一遍又一遍地看電影來回答有關它們的問題？ 同樣的規則也適用於機器學習！