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

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

• ผู้อ่านที่ใช้ macOS และ Linux รวมทั้ง Linux Distribution on Windows สามารถสร้างไฟล์ และ Folder เตรียมไว้ก่อน ตามตัวอย่างด้านล่าง (ยกเว้นไฟล์ model1.h5 และ train_classification_model.ipynb)

Create a new Conda Environment

• เราจะสร้าง Environment ชื่อ basic_model สำหรับรัน Python 3.6.8 โดยที่มี Package/Library ต่างๆ ได้แก่ FastAPI, Uvicorn, Python-dotenv, Pydantic, Tensorflow, Locust, Plotly, Scikit-learn, Seaborn รวมทั้ง Jupyter Notebook ซึ่งมีขั้นตอนดังต่อไปนี้
• สร้าง Environment ใหม่ ตั้งชื่อเป็น basic_model สำหรับรัน Python 3.6.8 และติดตั้ง Library ที่จำเป็น รวมทั้ง Jupyter Notebook โดยใช้คำสั่ง conda create -n

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
• ก่อนลบ ออกจาก Environment ด้วยคำสั่ง conda deavtivate
• ใช้ sudo หน้าคำสั่ง conda บน macOS เมื่อมีการสร้าง หรือลบ Environment

เข้าใช้ Environment ใหม่ โดยพิมพ์คำสั่ง conda activate ตามด้วยชื่อ Environment

conda activate basic_model

• ติดตั้ง tensorflow

pip install tensorflow==2.3.0

• เปิด Jupyter Notebook

jupyter notebook

• ไปที่ Folder train_model สร้าง Notebook ใหม่ โดยกดที่ New->Python 3
• พิมพ์ print(‘Hello API’) ใน Cell แรก แล้วกด Shift+Enter เพื่อรันโปรแกรมและสร้าง Cell ใหม่ไปพร้อมกัน

• กดที่ Untitled พิมพ์ชื่อไฟล์เป็น train_classification_model แล้วกด Rename

Training and Save Model

เราจะใช้ make_blobs() Function ของ scikit-learn Library ในการสร้าง Dataset ขนาด 2 มิติ ที่มีเพียง 3 Class ตามตัวอย่างด้านล่าง

• Import Library ที่จำเป็น แล้วสร้าง Dataset

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 npX, y = make_blobs(n_samples=3000, centers=3, n_features=2, cluster_std=2, random_state=2)

• แบ่ง 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 เพื่อทำให้สามารถแสดงสีแบบไม่ต่อเนื่องได้ แล้วนำไป Plot

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()

• เราจะเข้ารหัสผลเฉลย แบบ One-Hot Encoding เพื่อที่ว่าเมื่อ Model มีการ Predict ว่าเป็น Class ไหน มันจะให้ค่าความมั่นใจ (Confidence) กลับมาด้วย

y_train = to_categorical(y_train)
y_test = to_categorical(y_test)

• นิยาม, 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)

res = predict_model.predict(a)
np.argmax(res, axis=1)

• Copy Model ที่ Train แล้วไปยัง Folder basic_model/model_deploy/python (บน Windows Command Line ใช้คำสั่ง copy แทน cp)

cp model1.h5 ../model_deploy/python/

• ขณะนี้ใน Project ของเราจะประกอบด้วยไฟล์ และ 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

==> FastAPI and Uvicorn <==

• ปัจจุบันมี Python Web Framework อยู่เป็นจำนวนมาก อย่างเช่น Flask, Falcon, Starlette, Sanic, Tornado และ FastAPI แต่ที่ FastAPI เป็นตัวเลือกอันดับต้นๆ สำหรับการพัฒนา RestFul API ก็เพราะมันสามารถเขียน Code ได้สั้นและเข้าใจง่ายเหมือนกับ Flask Framework แต่มีความเร็วที่สูงกว่า โดย FastAPI จะทำงานร่วมกับ Uvicorn Server (ASGI Server) ในการรองรับการทำงานแบบ Asynchronous รวมทั้งยังมีการสร้าง API Document ให้แบบอัตโนมัติ โดยมีขั้นตอนดังนี้
• แก้ไขไฟล์ app.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}

จาก Code ด้านบน เรามีการนิยาม Function หลัก 3 Function ได้แก่ loadModel(), predict() และ get_class() ซึ่ง get_class() จะรับ Input Parameter แบบ JSON Format จาก HTTP POST Method ซึ่งมีการนิยามชนิดข้อมูลด้วย Pydantic Library

• เข้าใช้ basic_model Environment โดยพิมพ์คำสั่ง conda activate ตามด้วยชื่อ Environment

conda activate basic_model

• ไปที่ Folder basic_model/model_deploy/python รัน Python Web Application (app.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=nuttachot
API_PASSWORD=password

• แก้ไขไฟล์ app.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 ดังภาพด้านล่าง

ผู้อ่านสามารถเพิ่มระดับการรักษาความปลอดภัยโดยการติดตั้ง SSL Certificate บนเว็บไซต์ และการพิสูจน์ตัวตนโดยใช้ API Gateway ได้จากบทความต่อไปนี้

• วิธีติดตั้ง VPS และ Let’s Encrypt ด้วย Docker Container แบบง่ายๆ
• สร้าง API Gateway และระบบ Monitoring Microservice ด้วย Kong, Prometheus และ Grafana แบบง่ายๆ

Deployment

เราจะนำ Docker เข้ามาช่วยในการบรรจุ Software ทั้งหมดให้อยู่ในรูปของ Docker Image ซึ่งหลังจากนั้นก็จะนำ Docker Image ไปรัน (Docker Container) ในเครื่องไหนก็ได้ โดยทุกเครื่องจะมีสภาพแวดล้อมในการรันเหมือนกันทั้งหมด ไม่ว่าจะเป็นเครื่องสำหรับการ Development หรือ Production Server

*สำหรับผู้อ่านที่ยังไม่ได้ติดตั้ง Docker สามารถทำตามคำแนะนำจาก Link ต่อไปนี้ : ติดตั้งบน Mac / ติดตั้งบน Windows

**ท่านสามารถทำความเข้าใจแนวคิดของ Docker ได้จาก

• Docker tutorial for beginners : Part 1
• Docker tutorial for beginners : Part 2

เพื่อจะสร้าง Docker Container เราจะมีการแก้ไขไฟล์ docker-compose.yml, requirements.txt และ Dockerfile ดังต่อไปนี้

• แก้ไขไฟล์ docker-compose.yml

version: '3'
services:
  test_api:
    container_name: test_api
    build: python/
    restart: always
    networks:
      - default
      
    ports:
      - 7001:80
      
networks:
  default:
    external:
      name: basic_model_network

• แก้ไขไฟล์ 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

• สร้าง Bridge network โดยตั้งชื่อเป็น basic_model_network ด้วยคำสั่ง docker network create

docker network create basic_model_network

• ไปที่ Folder basic_model/model_deploy แล้ว Build Image และ Run Container ด้วยคำสั่ง docker-compose up

docker-compose up -d

• ดู Container ที่กำลังรันทั้งหมดที่ docker-compose.yml ดูแล

docker-compose ps

Testing the API Request

กลับมาที่ Jupyter Notebook เพื่อทดลองเรียกใช้งาน API ด้วยคำสั่งต่อไปนี้

import requests
from requests.auth import HTTPBasicAuth----------------------------------URL = 'http://localhost:7001/getclass'
data = {
  "x": -2.521156, 
  "y": -5.015865
}
response = requests.post(URL, json=data, auth=HTTPBasicAuth('nuttachot', 'password'))-----------------------------------if response.status_code == 200:
    res = response.json()['results']
    print(res)

Load Testing with Locust

Locust เป็น Open Source Load Testing Framwork ที่มีการกำหนดวิธีการทดสอบด้วยการเขียน Script โดยใช้ Python Code ซึ่งปัจจุบันมีผู้ใช้งานอยู่พอสมควร

ในการทำ Load Testing ด้วย Locust จะมีขั้นตอนดังนี้

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

from locust import HttpUser, task, between
import json
class 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

• จะเห็นว่า เมื่อทำตามขั้นตอนทั้งหมดนี้เราก็สามารถ Deploy Model และทดสอบความสามารถในการรองรับ load ของ Model ที่จะขึ้นใช้งานจริงบน Production ได้แล้ว

— — → ขอบคุณครับ < — —-