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

Nuttachot Promrit

• 9 min read
ภาพจาก https://unsplash.com/@spacex

บทความโดย ผศ.ดร.ณัฐโชติ พรหมฤทธิ์
ภาควิชาคอมพิวเตอร์
คณะวิทยาศาสตร์
มหาวิทยาลัยศิลปากร

ในการให้คำปรึกษาโครงงานวิทยานิพนธ์ (Project) บ่อยครั้งที่เมื่อนักศึกษาได้พัฒนา Deep Learning Model ที่มีประสิทธิภาพพอจะใช้งานได้ระดับหนึ่งแล้ว ผมก็จะให้พวกเขานำ Model ไป Deploy บน Local Host หรือ Cloud Server เพื่อประกอบร่างเป็นชิ้นงาน โดยมีการเรียกใช้ Model ผ่าน RestFul API ซึ่งส่วนใหญ่แล้วเราก็จะให้เด็กๆ ไปศึกษาวิธีการ Deploy Model ด้วยตัวเอง หรือถามจากรุ่นพี่ใน Lab แต่ก็มักจะพบว่ามันเป็นวิธีที่ไม่สามารถเอาไปใช้จริงเป็น Production ได้ หรือใช้เวลาในการศึกษากันค่อนข้างนาน

เพื่อให้ผู้อ่านเห็นแนวทางการนำ Machine Learning Model ไปใช้งานจริง ในบทความนี้เราจะได้ทำ Workshop ด้วยการ Deploy Model บน Production Environment ด้วยวิธีการที่ไม่ยากมากนัก โดยใช้ FastAPI, Uvicorn และ Docker Container รวมทั้งการทำ Load Testing ด้วย Locust ครับ

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.9 โดยที่มี Package/Library ต่างๆ ได้แก่ FastAPI, Uvicorn, Python-dotenv, Pydantic, Tensorflow, Locust, Plotly, Scikit-learn, Seaborn รวมทั้ง Jupyter Notebook ซึ่งมีขั้นตอนดังต่อไปนี้

  • สำหรับผู้อ่านที่ยังไม่ได้ติดตั้ง Miniconda ท่านสามารถ Download และติดตั้ง Miniconda ได้จาก https://docs.conda.io/en/latest/miniconda.html
  • สร้าง Environment ใหม่ ตั้งชื่อเป็น basic_model สำหรับรัน Python 3.9 และติดตั้ง Library ที่จำเป็น รวมทั้ง Jupyter Notebook โดยใช้คำสั่ง conda create -n
conda create -n basic_model python=3.9 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.7
  • เปิด 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

import tensorflow as tf

to_categorical = tf.keras.utils.to_categorical
load_model = tf.keras.models.load_model

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 plotly.figure_factory as ff

import numpy as np
x, 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_val, y_train, y_val = train_test_split(x, y, test_size=0.4, shuffle= True)

x_train.shape, x_val.shape, y_train.shape, y_val.shape

((1800, 2), (1200, 2), (1800,), (1200,))

  • นำ 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_val = to_categorical(y_val)
  • นิยาม, Compile และ Train Model
model = tf.keras.Sequential()

model.add(tf.keras.layers.Dense(50, input_dim=2, activation='relu', kernel_initializer='he_uniform'))
model.add(tf.keras.layers.Dense(3, activation='softmax'))

model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])
his = model.fit(x_train, y_train, validation_data=(x_val, y_val), epochs=200, verbose=1)
  • Plot Loss
h1 = go.Scatter(y=his.history['loss'], 
                    mode="lines",
                    line=dict(
                        width=2,
                        color='blue'),
                        name="acc"
                   )
h2 = go.Scatter(y=his.history['val_loss'], 
                    mode="lines",
                    line=dict(
                        width=2,
                        color='red'),
                        name="val_acc"
                   )

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)
  • 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)
  • Model Predict จาก Test Dataset
predicted_classes = np.argmax(model.predict(x_val), axis=-1)
predicted_classes.shape

(1200,)

  • เตรียม Label
labels = ['A', 'B', 'C']
  • คำนวณค่า Confusion Matrix
y_true = np.argmax(y_val, axis=1)
cm = confusion_matrix(y_true, predicted_classes)
  • นิยาม Function cm_plot แล้ว Plot Confusion Matrix
def cm_plot(cm, labels):
    x = labels
    y = labels

    z_text = [[str(y) for y in x] for x in cm]
    fig = ff.create_annotated_heatmap(cm, x=x, y=y, annotation_text=z_text, colorscale='blues')

    fig.update_layout(title_text='Confusion Matrix')

    fig.add_annotation(dict(font=dict(color="black",size=13),
                            x=0.5,
                            y=-0.15,
                            showarrow=False,
                            text="Predicted Value",
                            xref="paper",
                            yref="paper"
                           ))

    fig.add_annotation(dict(font=dict(color="black",size=13),
                            x=-0.20,
                            y=0.5,
                            showarrow=False,
                            text="Real Value",
                            textangle=-90,
                            xref="paper",
                            yref="paper"
                           ))

    fig.update_layout(margin=dict(t=50, l=200))
    fig['layout']['yaxis']['autorange'] = "reversed"

    fig['data'][0]['showscale'] = True
    fig.show()
cm_plot(cm, labels)
  • แสดง Precision, Recall, F1-score
print(classification_report(y_true, predicted_classes, target_names=labels, digits=4))
  • Save Model
filepath='model1.h5'
model.save(filepath)
  • ทดลอง Load Model
predict_model = load_model(filepath) 
predict_model.summary()
  • ทดลอง Predict จาก Model ที่ Load มาใหม่
a = np.array([[-0.210738, -13.1719]])

predicted_score = predict_model.predict(a)

predicted_score

array([[9.9999821e-01, 5.2171343e-09, 1.7500638e-06]], dtype=float32)

predicted_classes = np.argmax(predicted_score, axis=-1)

predicted_classes

array([0])

  • Copy Model ที่ Train แล้วไปยัง Folder basic_model/model_deploy/python (บน Windows ใช้คำสั่ง copy model1.h5 .\..\model_deploy\python แทน)
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

ข้อมูลจาก https://www.techempower.com/benchmarks
ข้อมูลจาก https://www.techempower.com/benchmarks

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

  • แก้ไขไฟล์ api.py ด้วย Code Editor ยอดนิยมอย่างเช่น Visual Studio Code ตามตัวอย่างด้านล่าง
import tensorflow as tf
load_model = tf.keras.models.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 (api.py) ด้วยคำส่ง uvicorn api:ap
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": -0.210738, 
  "y": -13.1719
}

Basic Authen

อย่างไรก็ตามในการใช้งานจริงเราต้องคำนึงถึงการรักษาความปลอดภัยด้วย เช่นในเรื่องการพิสูจน์ตัวตนก่อนการใช้งาน โดยผู้เขียนจะยกตัวอย่างการพิสูจน์ตัวตนแบบ Basic Authen ด้วย Username และ Password ดังต่อไปนี้

  • แก้ไขไฟล์ .env โดยการกำหนด Username และ Password ตามตัวอย่างด้านล่าง
API_USERNAME=nuttachot
API_PASSWORD=password
  • แก้ไขไฟล์ api.py ดังต่อไปนี้
import tensorflow as tf
load_model = tf.keras.models.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 ได้จากบทความต่อไปนี้

Deployment

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

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

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

เพื่อจะสร้าง 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==2.7
  • แก้ไขไฟล์ Dockerfile
FROM python:3.9-slim
RUN apt-get update && apt-get install -y \
&& 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 บน Command Line
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': -0.210738, 
  'y': -13.1719
}

response = requests.post(URL, json=data, auth=HTTPBasicAuth('nuttachot', 'password'))
if response.status_code == 200:
    res = response.json()['results']
    print(res)

{'class': 'class_A', 'confidence': 0.9999982118606567}

Load Testing with Locust

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

ข้อมูลจาก https://locust.io

ในการทำ 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':-0.210738, 'y':-13.1719}
        self.client.post(
            url='/getclass',
            data=json.dumps(data),
            auth=('nuttachot', 'password')
            
        )
  • เปิด Terminal ไปที่ Folder basic_model/train_model รันคำสั่ง conda activate basic_model แล้วรันไฟล์ loadtest.py ด้วยคำสั่ง locust -f loadtest.py
locust -f loadtest.py --host=http://localhost:7001
  • ไปที่ URL http://localhost:8089 กำหนดจำนวน User และ Spawn rate เท่ากับ 20 แล้วกด Start swarming

จะเห็นว่า API ของเรามี Response Time โดยเฉลี่ย (Median) เท่ากับ 46ms และจำนวน Request ต่อวินาทีเท่ากับ 12.5 Request

  • กดที่ Edit ทดลองเพิ่มจำนวน User และ Spawn rate เท่ากับ 50 แล้วกด Start swarming

Response Time เฉลี่ย (Median) จะเพิ่มเป็น 51ms และจำนวน Request ต่อวินาทีเท่ากับ 31.7 Request

  • เพิ่มจำนวน User และ Spawn rate เท่ากับ 100 แล้วกด Start swarming

Response Time เฉลี่ย (Median) จะเพิ่มเป็น 56ms และจำนวน Request ต่อวินาทีเท่ากับ 61.2 Request

  • เพิ่มจำนวน User และ Spawn rate เท่ากับ 150 แล้วกด Start swarming

Response Time เฉลี่ย (Median) จะเพิ่มเป็น 71ms และจำนวน Request ต่อวินาทีเท่ากับ 82.2 Request

  • เพิ่มจำนวน User และ Spawn rate เท่ากับ 200 แล้วกด Start swarming

จากภาพด้านบน เมื่อเพิ่มจำนวน User และ Spawn rate เท่ากับ 200 แล้ว Response Time เฉลี่ย (Median) จะเพิ่มเป็น 320ms แต่พบว่ามันจะไม่สามารถรองรับ Request ที่ยิงมาจาก Locust ได้เพิ่มขึ้นมากนัก (91.8 RPS) และจากกราฟด้านล่างที่มีลักษณะขึ้น ๆ ลง ๆ แสดงให้เห็นว่า API ของเราจะไม่สามารถรองรับ Request แบบสบายๆ ได้เหมือนปกติ

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

Bonus -> ตัวอย่างการ Deploy Model บน Google Colab และการเรียก API