Deploy Machine Learning Model บน Production ด้วย FastAPI, Uvicorn และ Docker

6 min readSep 23, 2020

Architecture

สถาปัตยกรรมใน Workshop นี้ จะมีโครงสร้างดังภาพด้านล่าง จากภาพ HTTP Traffic ที่มาจาก Internet จะวิ่งไปยัง Uvicon Server ที่ Port 7001 โดย Uvicon จะทำหน้าที่ในการรัน Python Web Application แบบ Asynchronous Process ที่มีการพัฒนาด้วย FastAPI Framework โดยมี /getclass เป็น API Endpoint (http://hostname:7001/getclass) ซึ่งมีการรับข้อมูลเป็น JSON Format (จาก HTTP POST Method) แล้วส่งผลการ Predict ด้วย Model ที่พัฒนาโดยใช้ Tensorflow Framework กลับมาเป็น JSON Format เช่นเดียวกัน ซึ่งเราจะเห็นว่า Component ทั้งหมดที่กล่าวมานั้นจะถูกบรรจุอยู่ภายใน Docker Container เพียง Container เดียว

Project Structure

ยกตัวอย่างด้วยการ Train Neural Network Model อย่างง่ายเพื่อจำแนกข้อมูล 3 Class แล้ว Save Model เป็นไฟล์ model1.h5 เพื่อนำไปบรรจุลงใน Docker Container โดยไฟล์ทั้งหมดใน Project นี้ จะจัดเก็บใน Folder ชื่อ basic_model ซึ่งภายใน basic_model จะประกอบด้วยไฟล์ และ Folder ดังต่อไปนี้


├── model_deploy
│   ├── docker-compose.yml
│   └── python
│       ├── Dockerfile
│       ├── api.py
│       ├── model1.h5
│       ├── .env
│       └── requirements.txt
└── train_model
    ├── train_classification_model.ipynb
    ├── model1.h5
    └── loadtest.py

สร้าง Environment ใหม่ ตั้งชื่อเป็น basic_model สำหรับรัน Python 3.6.8 และติดตั้ง Library ที่จำเป็น

conda create -n basic_model python=3.6.8 fastapi uvicorn python-dotenv pydantic locust plotly scikit-learn seaborn jupyter -c conda-forge

ลบ Environment ที่เคยสร้าง

conda remove — name basic_model — all

import library ที่จำเป็น

import matplotlib.pyplot as plt
from tensorflow.keras.layers import Dense
from tensorflow.keras.models import Sequential
from tensorflow.keras.utils import to_categorical
from sklearn.datasets import make_blobs
from sklearn.model_selection import train_test_split
from sklearn.metrics import confusion_matrix
from sklearn.metrics import classification_report
import pandas as pd
import plotly.express as px
import plotly
import plotly.graph_objs as go
import seaborn as sn
import numpy as np

แบ่ง Dataset เป็น 2 ส่วน สำหรับการ Train 60% และสำหรับการ Test อีก 40%

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.4, shuffle= True)
X_train.shape, X_test.shape, y_train.shape, y_test.shape

นำ Dataset ส่วนที่ Train มาแปลงเป็น DataFrame โดยเปลี่ยนชนิดข้อมูลใน Column “class” เป็น String

X_train_pd = pd.DataFrame(X_train, columns=[‘x’, ‘y’])
y_train_pd = pd.DataFrame(y_train, columns=[‘class’])
df = pd.concat([X_train_pd, y_train_pd], axis=1)
df[“class”] = df[“class”].astype(str)
fig = px.scatter(df, x=”x”, y=”y”, color=”class”)
fig.show()

นิยาม, Compile และ Train Model

model = Sequential()
model.add(Dense(50, input_dim=2, activation=’relu’, kernel_initializer=’he_uniform’))
model.add(Dense(3, activation=’softmax’))
model.compile(loss=’categorical_crossentropy’, optimizer=’adam’, metrics=[‘accuracy’])

his = model.fit(X_train, y_train, validation_data=(X_test, y_test), epochs=200, verbose=1)

Plot Loss

plotly.offline.init_notebook_mode(connected=True)
h1 = go.Scatter(y=his.history[‘loss’],
mode=”lines”, line=dict(
width=2,
color=’blue’),
name=”loss”
)
h2 = go.Scatter(y=his.history[‘val_loss’],
mode=”lines”, line=dict(
width=2,
color=’red’),
name=”val_loss”
)
data = [h1,h2]
layout1 = go.Layout(title=’Loss’,
xaxis=dict(title=’epochs’),
yaxis=dict(title=’’))
fig1 = go.Figure(data = data, layout=layout1)
plotly.offline.iplot(fig1, filename=”Intent Classification”)

Plot Accuracy

h1 = go.Scatter(y=his.history[‘accuracy’],
mode=”lines”, line=dict(
width=2,
color=’blue’),
name=”acc”)
h2 = go.Scatter(y=his.history[‘val_accuracy’],
mode=”lines”, line=dict(
width=2,
color=’red’),
name=”val_acc” )
data = [h1,h2]
layout1 = go.Layout(title=’Accuracy’,
xaxis=dict(title=’epochs’),
yaxis=dict(title=’’))
fig1 = go.Figure(data = data, layout=layout1)
plotly.offline.iplot(fig1, filename=”Intent Classification”)

Model Predict จาก Test Dataset

predicted_classes = model.predict_classes(X_test)
predicted_classes.shape

เตรียมผลเฉลยของ Test Dataset สำหรับสร้างตาราง Confusion Matrix

y_true = np.argmax(y_test,axis = 1)
y_true.shape

คำนวณค่า Confusion Matrix

cm = confusion_matrix(y_true, predicted_classes)

แสดงตาราง Confusion Matrix ด้วย Heatmap

df_cm = pd.DataFrame(cm, range(3), range(3))
plt.figure(figsize=(10,7))
sn.set(font_scale=1.2)
sn.heatmap(df_cm, annot=True, fmt=’d’, annot_kws={“size”: 14})
plt.show()

แสดง Precision, Recall, F1-score

label = [‘0’, ‘1’, ‘2’]
print(classification_report(y_true, predicted_classes, target_names=label, digits=4))

Save Model

filepath=’model1.h5'
model.save(filepath)

Load Model

from tensorflow.keras.models import load_model
predict_model = load_model(filepath)
predict_model.summary()

Predict Model ที่ Load มา

a = np.array([[-2.521156, -5.015865]])
predict_model.predict(a)

Copy Model ที่ Train แล้วไปยัง Folder

cp model1.h5 C:\Users\Firstll\Desktop\basic_model\model_deploy\python

FastAPI and Uvicorn

แก้ไขไฟล์ api.py ด้วย Code Editor ของVisual Studio Code ตามตัวอย่างด้านล่าง

from tensorflow.keras.models import load_model
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
app = FastAPI()
class Data(BaseModel):
x:float
y:float
def loadModel():
global predict_model
predict_model = load_model(‘model1.h5’)
loadModel()
async def predict(data):
classNameCat = {0:’class_A’, 1:’class_B’, 2:’class_C’}
X = np.array([[data.x, data.y]])
pred = predict_model.predict(X)
res = np.argmax(pred, axis=1)[0]
category = classNameCat[res]
confidence = float(pred[0][res])

return category, confidence
@app.post(“/getclass/”)
async def get_class(data: Data):
category, confidence = await predict(data)
res = {‘class’: category, ‘confidence’:confidence}
return {‘results’: res}

รัน Python Web Application (api.py) ด้วยคำส่ง uvicorn api:ap แล้วกด Allow

uvicorn api:app — host 0.0.0.0 — port 80 — reload

API Documentation

FastAPI จะสร้าง API Document ให้โดยอัตโนมัติ โดยเราสามารถทดลองใช้งาน API ได้จาก URL http://localhost/docs

ไปที่ /getclass แล้วกด Try it out>>แก้ไข Request body แบบ JSON Format กด Execute แล้วดูผลลัพธ์จากการ Predict

{
“x”: -2.521156,
“y”: -5.015865
}

Basic Authen

โดยผู้เขียนจะยกตัวอย่างการพิสูจน์ตัวตนแบบ Basic Authen ด้วย Username และ Password ดังต่อไปนี้

แก้ไขไฟล์ .env โดยการกำหนด Username และ Password ตามตัวอย่างด้านล่าง

API_USERNAME=rynlapat
API_PASSWORD=password

แก้ไขไฟล์ api.py

from tensorflow.keras.models import load_model
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from starlette.status import HTTP_401_UNAUTHORIZED
import secrets
import os
from dotenv import load_dotenv
load_dotenv(os.path.join(‘.env’))
API_USERNAME = os.getenv(“API_USERNAME”)
API_PASSWORD = os.getenv(“API_PASSWORD”)
security = HTTPBasic()
def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
correct_username = secrets.compare_digest(credentials.username, API_USERNAME)
correct_password = secrets.compare_digest(credentials.password, API_PASSWORD)
if not (correct_username and correct_password):
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail=’Incorrect username or password’,
headers={‘WWW-Authenticate’: ‘Basic’},
)
return credentials.username
app = FastAPI()
class Data(BaseModel):
x:float
y:float
def loadModel():
global predict_model
predict_model = load_model(‘model1.h5’)
loadModel()
async def predict(data):
classNameCat = {0:’class_A’, 1:’class_B’, 2:’class_C’}
X = np.array([[data.x, data.y]])
pred = predict_model.predict(X)
res = np.argmax(pred, axis=1)[0]
category = classNameCat[res]
confidence = float(pred[0][res])

return category, confidence
@app.post(“/getclass/”)
async def get_class(data: Data, username: str = Depends(get_current_username)):
category, confidence = await predict(data)
res = {‘class’: category, ‘confidence’:confidence}
return {‘results’: res}

ใช้งาน API อีกครั้ง โดยเมื่อเรากด Execute จะต้องมีการพิสูจน์ตัวตนด้วย Username และ Password

แก้ไขไฟล์ requirements.txt

python-dotenv
fastapi
uvicorn
pydantic
tensorflow

แก้ไขไฟล์ Dockerfile

FROM python:3.6.8-slim-stretch
RUN apt-get update && apt-get install -y python-pip \
&& apt-get clean
WORKDIR /app
COPY api.py .env model1.h5 requirements.txt ./
RUN pip install — no-cache-dir -r requirements.txt
CMD uvicorn api:app — host 0.0.0.0 — port 80 — workers 6

Load Testing with Locust

แก้ไขไฟล์ loadtest.py

from locust import HttpUser, task, between
import jsonclass QuickstartUser(HttpUser):
min_wait = 1000
max_wait = 2000 @task
def test_api(self): data = {“x”:-2.521156, “y”:-5.015865}
self.client.post(
url=”/getclass”,
data=json.dumps(data),
auth=(“nuttachot”, “password”))

ไปที่ Folder basic_model/train_model แล้วรันไฟล์ loadtest.py ด้วยคำสั่ง locust -f loadtest.py

locust -f loadtest.py — host=http://localhost:7001

กด Allow แล้วไปยัง URL http://localhost:8089 กำหนดจำนวน User และ Spawn rate เท่ากับ 20 แล้วกด Start swarming

ลองไป Load Testing ใน IP VM Address ที่ Port 8089