使用Flask部署機器學習模型

本文轉載自查看原文 2020-07-09 16:58 1290 機器學習/ 深度學習

作者|LAKSHAY ARORA
編譯|VK
來源|Analytics Vidhya

概述

部署機器學習模型是每個ML項目的一個關鍵
學習如何使用Flask將機器學習模型部署到生產中
模型部署是數據科學家訪談中的一個核心話題

介紹

我記得我早期在機器學習領域的日子。我喜歡處理多個問題，對機器學習項目的各個階段都很感興趣。和我之前的許多人一樣，我被模型整個生命周期的構建所吸引。

我和領域專家談過，項目經理和所有相關人員確保他們的投入被包括在模型中。但后來我遇到了一個障礙——我到底該如何把我的模型交給我的客戶呢？我不能給他們一個Jupyter notebook！

我所學的一切都集中在模型構建組件上。沒有多少人會談論如何部署你的機器學習模型。把你的模型投入生產意味着什么？它需要什么？

這些都是每個數據科學家需要回答的關鍵的職業定義問題。這就是為什么我決定寫下這個教程來演示如何使用Flask來部署機器學習模型。

我們將首先了解模型部署的概念，然后討論Flask是什么，如何安裝它，最后，我們將深入到一個問題陳述中，學習如何使用Flask部署機器學習模型。

什么是模型部署？

在典型的機器學習和深度學習項目中，我們通常從定義問題陳述開始，然后是數據收集和准備、數據理解和模型構建，對吧？

但是，最后，我們希望我們的模型能夠提供給最終用戶，以便他們能夠利用它。模型部署是任何機器學習項目的最后階段之一，可能有點棘手。如何將機器學習模型傳遞給客戶/利益相關者？當你的模型投入生產時，你需要注意哪些不同的事情？你怎么能開始部署一個模型呢？

Flask的作用來了。

什么是Flask？

Flask是一個用Python編寫的web應用程序框架。它有多個模塊，使web開發人員更容易編寫應用程序，而不必擔心協議管理、線程管理等細節。

Flask是開發web應用程序的選擇之一，它為我們提供了構建web應用程序所必需的工具和庫。

在本教程中，我們將利用Flask的資源來幫助我們部署自己的機器學習模型。你會喜歡用Flask工作的！

在機器上安裝Flask

安裝Flask簡單明了。在這里，我假設你已經安裝了Python 3和pip。要安裝Flask，需要運行以下命令：

sudo apt-get install python3-flask

就這樣！准備好深入到問題陳述中去，離部署機器學習模型更近一步。

理解問題陳述

在本節中，我們將使用Twitter數據集。我們的目標是在推特上發現仇恨言論。為了簡單起見，如果推特帶有種族主義或性別歧視情緒，我們說它包含仇恨言論。

我們將創建一個包含如下文本框的網頁（用戶可以搜索任何文本）：

對於任何搜索查詢，我們將實時抓取與該文本相關的tweet，對於所有這些被抓取的tweet，我們將使用仇恨言語檢測模型對種族主義和性別歧視tweet進行分類。

設置項目工作流

模型構建：我們將建立一個邏輯回歸模型管道來分類tweet是否包含仇恨言論。在這里，我們的重點不是如何建立一個非常精確的分類模型，而是看看如何使用Flask部署這個模型
安裝Twitter應用程序：我們將在Twitter開發人員的網站上創建一個Twitter應用程序，並獲取身份驗證密鑰。我們將編寫一個Python腳本來抓取與特定文本查詢相關的tweet
網頁模板：在這里，我們將設計一個用戶界面，用戶可以提交他的查詢
獲取Tweets：從用戶處獲取查詢后，我們將使用twitter API獲取與所搜索查詢相關的Tweets
預測類並發送結果：接下來，使用保存的模型預測tweets的類並將結果發送回網頁

下面是我們剛才看到的步驟的示意圖：

建立我們的機器學習模型

我們在映射到標簽的CSV文件中有關於Tweets的數據。我們將使用logistic回歸模型來預測tweet是否包含仇恨言論。

你可以在這里下載完整的代碼和數據集。

https://github.com/lakshay-arora/Hate-Speech-Classification-deployed-using-Flask/tree/master

首先導入一些必需的庫：

# 導入必需的庫
import pandas as pd
from sklearn.feature_extraction.text import ENGLISH_STOP_WORDS, TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.metrics import f1_score
from sklearn.model_selection import train_test_split

接下來，我們將讀取數據集並查看頂行：

# 讀取數據集
data = pd.read_csv('dataset/twitter_sentiments.csv')
# 查看頂行
data.head()

數據集有31962行和3列：

id：每行的唯一編號
label：對於正常的tweet，它是0；對於種族主義或性別歧視的tweet，它將是1。有29720個0和2242個1
tweet：在Twitter上發布的tweet

現在，我們將使用scikit learn的train_test_split函數將數據分為訓練和測試。我們只將20%的數據用於測試。我們將對標簽列上的數據進行分層抽樣，以便目標標簽在訓練和測試數據中的分布相同：

# 分為訓練集和測試集
train, test = train_test_split(data, test_size = 0.2, stratify = data['label'], random_state=21)

# 獲取大小
train.shape, test.shape
## >> ((25569, 3), (6393, 3))

現在，我們將使用TfidfVectorizer 創建tweet列的TF-IDF向量，並將參數lowercase設為True，以便它首先將文本轉換為小寫。我們還將保持max features為1000，並傳遞scikit learn 庫中預定義的停用詞列表。

首先，創建TFidfVectorizer的對象，構建模型並將模型與訓練數據tweets匹配：

# 創建TfidfVectorizer對象
tfidf_vectorizer = TfidfVectorizer(lowercase= True, max_features=1000, stop_words=ENGLISH_STOP_WORDS)

# 擬合模型
tfidf_vectorizer.fit(train.tweet)

利用模型對訓練和測試數據的推文進行變換:

# #轉換訓練和測試數據
train_idf = tfidf_vectorizer.transform(train.tweet)
test_idf  = tfidf_vectorizer.transform(test.tweet)

現在，我們將創建一個Logistic回歸模型的對象。

請記住，我們的重點不是建立一個非常精確的分類模型，而是看我們如何部署這個預測模型來獲得結果。

# 創建線性回歸模型的對象
model_LR = LogisticRegression()

# 用訓練數據擬合模型
model_LR.fit(train_idf, train.label)

# 預測訓練數據的標簽
predict_train = model_LR.predict(train_idf)

# 在測試數據預測模型
predict_test = model_LR.predict(test_idf)

# f1得分
f1_score(y_true= train.label, y_pred= predict_train)
## >> 0.4888178913738019

f1_score(y_true= test.label, y_pred= predict_test)
## >> 0.45751633986928114

讓我們定義管道的步驟：

步驟1：創建一個tweet文本的TF-IDF向量，其中包含上面定義的1000個特征
步驟2：使用邏輯回歸模型預測目標標簽

當我們對管道對象使用fit()函數時，這兩個步驟都會執行。在模型訓練過程之后，我們使用predict())函數來生成預測。

# 定義管道的階段
pipeline = Pipeline(steps= [('tfidf', TfidfVectorizer(lowercase=True,
                                                      max_features=1000,
                                                      stop_words= ENGLISH_STOP_WORDS)),
                            ('model', LogisticRegression())])

# 用訓練數據擬合管道模型                          
pipeline.fit(train.tweet, train.label)

現在，我們將使用一個示例tweet測試管道：

# 示例tweet
text = ["Virat Kohli, AB de Villiers set to auction their 'Green Day' kits from 2016 IPL match to raise funds"]

# 使用管道預測標簽
pipeline.predict(text)
## >> array([0])

我們已經成功地構建了機器學習管道，我們將使用joblib庫中的dump函數保存這個管道對象。只需傳遞管道對象和文件名：

# 導入joblib
from joblib import dump

# 保存管道模型
dump(pipeline, filename="text_classification.joblib")

它將創建一個文件“text_classification.joblib“. 現在，我們將打開另一個Python文件，並使用joblib庫的load函數來加載管道模型。

讓我們看看如何使用保存的模型：

# 導入joblib
from joblib import load

# tweet文本示例
text = ["Virat Kohli, AB de Villiers set to auction their 'Green Day' kits from 2016 IPL match to raise funds"]

# 加載保存的pipleine模型
pipeline = load("text_classification.joblib")

# 對tweet文本樣本的預測
pipeline.predict(text)
## >> array([0])

設置Twitter API

我們需要做的第一件事是從Twitter開發人員網站獲取API key, API secret key, access token，access token secret。這些密鑰將幫助API進行身份驗證。首先，轉到這一頁並填寫表格。

https://developer.twitter.com/en/apps/create

一旦你填好表格，你就會拿到key。

安裝tweepy

現在，我們將安裝tweepy，它是一個Python庫，允許我們訪問Twitter API。

!pip3 install tweepy

導入所需的庫並添加從Twitter接收到的身份驗證密鑰。Tweepy試圖使身份驗證對你來說盡可能無痛。

要開始這個過程，需要創建OAuthHandler實例並傳遞API key和API secret key。然后使用access token和access token secret對實例進行身份驗證。

# 導入所需庫
import tweepy
import time
import pandas as pd
pd.set_option('display.max_colwidth', 1000)

# api key
api_key = "Enter API Key Here"
# api secret key
api_secret_key = "Enter API Secret Key Here."
# access token
access_token = "Enter Access Token Here"
# access token secret
access_token_secret = "Enter Access Token Secret Here."

# 授權API Key
authentication = tweepy.OAuthHandler(api_key, api_secret_key)

#對用戶access token和access token secret授權
authentication.set_access_token(access_token, access_token_secret)

# 調用api
api = tweepy.API(authentication, wait_on_rate_limit=True)

接下來，我們將定義一個函數“get_related_tweets”，它將接受參數text_query並返回與該特定文本查詢相關的50條tweets。我們將使用搜索API從Twitter獲取結果。

搜索API的一些參數是：

q–最多500個字符的搜索查詢字符串
geocode–返回位於給定緯度/經度的給定半徑內的用戶的tweets
lang–將tweets限制為給定的語言，由ISO 639-1代碼給出
result_type–指定希望接收的搜索結果類型。當前默認值為“mixed”。有效值包括：
- mixed：返回包含流行和實時結果
- recent：僅返回最近的結果
- popular：只返回最流行的結果
count–每頁嘗試檢索的結果數。一次最多可以請求100條tweets
max_id–僅返回id小於（即早於）或等於指定id的狀態。使用此選項，可以自動獲取大量唯一的tweets

我們將為給定的文本查詢請求50條tweet以及tweet創建時間、tweet id和tweet文本，函數將返回所有tweet的數據幀：

def get_related_tweets(text_query):
    # 存儲推文的列表
    tweets_list = []
    # 推特數量
    count = 50
    try:
        # 從查詢中提取單個tweets
        for tweet in api.search(q=text_query, count=count):
            print(tweet.text)
            # 添加到包含所有tweets的列表
            tweets_list.append({'created_at': tweet.created_at,
                                'tweet_id': tweet.id,
                                'tweet_text': tweet.text})
        return pd.DataFrame.from_dict(tweets_list)

    except BaseException as e:
        print('failed on_status,', str(e))
        time.sleep(3)

創建網頁

在這里，我們將創建一個類似於以下內容的網頁：

它將有一個文本框，用戶可以在其中鍵入文本查詢，然后單擊“搜索”按鈕以獲取搜索文本查詢的結果。

我們需要添加表單標記來收集搜索容器中的數據，在表單標記中，我們將方法post和name作為“search”傳遞。通過提供這個方法，我們的后端代碼將能夠知道我們已經收到了一些名為“search”的數據，在后端，我們需要處理這些數據並發送一些數據。

這只是HTML文件的一部分。你可以在此處下載完整的代碼和與此項目相關的其他文件。

https://github.com/lakshay-arora/Hate-Speech-Classification-deployed-using-Flask/tree/master

將網頁與模型連接

我們已經完成了前端部分，現在我們連接模型和網頁。第一步是加載保存的管道模型，我們將定義一個函數requestResults，該函數將獲取所請求查詢的tweets，並使用管道獲取標簽並返回要發送的最終結果。

# 導入所需庫
from flask import Flask, render_template, request, redirect, url_for
from joblib import load
from get_tweets import get_related_tweets

# 加載管道對象
pipeline = load("text_classification.joblib")

# 獲取特定文本查詢的結果
def requestResults(name):
    # 獲取推特文本
    tweets = get_related_tweets(name)
    # 獲取預測
    tweets['prediction'] = pipeline.predict(tweets['tweet_text'])
    # 獲取預測的不同標簽的值計數
    data = str(tweets.prediction.value_counts()) + '\n\n'
    return data + str(tweets)

現在，首先，創建Flask類的一個對象，該對象將以當前模塊名作為參數。route函數將告訴Flask應用程序下一步要在網頁上呈現哪個URL。

當Flask服務器運行時，Flask應用程序將路由到默認URL路徑並調用home函數，它將呈現home.html文件。

現在，每當有人發送文本查詢時，Flask將檢測post方法並調用get_data函數，在這里我們將使用名稱搜索獲取表單數據，然后重定向到success函數。

最后，success函數將使用requestResults函數獲取數據並將其發送回網頁。

# 啟動flask
app = Flask(__name__)

# 渲染網頁
@app.route('/')
def home():
    return render_template('home.html')

# 當post方法檢測到時，則重定向到success函數
@app.route('/', methods=['POST', 'GET'])
def get_data():
    if request.method == 'POST':
        user = request.form['search']
        return redirect(url_for('success', name=user))

# 獲取請求查詢的數據
@app.route('/success/<name>')
def success(name):
    return "<xmp>" + str(requestResults(name)) + " </xmp> "

現在，調用run函數啟動Flask服務器：