FastAPI - Model Serving 해보기

조정훈·2024년 4월 12일
FastAPI

Inference 서버 코드 작성

개발환경 세팅

IRIS-API-SERVER 라는 폴더(이름맘대로)를 하나 만들어주고 conda create --name iris-api python=3.11 으로 가상환경 만들어주고 activate 후 pip install poetry 으로 poetry 설치한 다음 poetry init
뭐 영어 뜨면 대충 엔터치고 no 선택하고 마지막은 yes로 셋팅하면 폴더에 pyproject.toml 파일이 생길 것이다.
저번 머신러닝 파이프라인 실습했던 환경과 일치시켜보자
poetry add fastapi
poetry add mlflow==1.27.0
poetry add boto3
poetry add sqlalchemy==1.4.52
poetry add psycopg2-binary
poetry add scikit-learn
poetry add pandas
poetry add 'uvicorn[standard]' - standard로 깔아야 속도빠름(uvloop)

  • history 치면 명령어 쳤던거 기록 볼 수 있다

app폴더 생성하고 안에 main.py 만들자


FastAPI App 객체 정의하기

먼저 FastAPI 몸체가 될 App 객체를 정의합니다.
배포 시 healthcheck를 위한 엔드포인트도 하나 정의해둡니다.
main.py 에서 작업했다

# app/main.py
from fastapi import FastAPI
from app import routers

app = FastAPI()

app.include_router(routers.router)

@app.get("/healthcheck")
async def healthcheck():
   return {"status": "ok"}

작성 후 uvicorn app.main:app --reload 명령어로 서버 띄워보자

정상작동한 모습

그 후 http://127.0.0.1:8000/docs 가서 제대로 뜨는지 확인.


FastAPI Router 객체 정의하기

예측 API 함수를 작성합니다. API 함수는 별도의 Router 객체에 구현합니다.
기능 별로 API 엔드포인트를 모듈로 분리하여 개발하는 것이 유지보수성에 좋기 때문
routers/ml_router.py 파일에 작성합니다.

# app/routers/ml_router.py
import pandas as pd
from fastapi import APIRouter

from app.schemas.request import IrisReq
from app.schemas.response import ResponseModel, IrisResp
from app.core.model_registry import model_registry


router = APIRouter()


@router.post("/predict", response_model=ResponseModel)
def predict(request: IrisReq):
   result = model_registry.get_model("iris_model").predict(
       pd.DataFrame([request.model_dump()])
   )

   return ResponseModel(status_code=200, data=IrisResp(target=result))

FastAPI Request, Response 모델 정의하기

예측 API 함수를 작성하기 전에 Request, Response Body로 사용할 Pydantic Model을 정의합니다.
app/schemas/request.py response.py에 작성합니다.
Pydantic Base Model을 사용함으로써 입/출력 값의 검증을 자동으로 수행할 수 있습니다.

 # app/schemas/request.py
from pydantic import BaseModel, Field


class IrisReq(BaseModel):
   sepal_length: float = Field(..., example=5.1, 
                         description="Sepal length")
   sepal_width: float = Field(..., example=3.5, 
                          description="Sepal width")
   petal_length: float = Field(..., example=1.4, 
                         description="Petal length")
   petal_width: float = Field(..., example=0.2, 
                          description="Petal width")
# app/schemas/response.py
from pydantic import BaseModel, Field


class IrisResp(BaseModel):
   target: int = Field(..., example=0, 
                       description="Predicted class")

class ResponseModel(BaseModel):
   status_code: int = Field(..., example=200, 
                           description="Status code")
   data: IrisResp = Field(..., 
                         description="Response data")

FastAPI 예측 엔드포인트 작성하기

/predict라는 경로를 갖는 POST API를 작성합니다.

# app/routers/ml_router.py
import pandas as pd
from fastapi import APIRouter

from app.schemas.request import IrisReq
from app.schemas.response import ResponseModel, IrisResp
from app.core.model_registry import model_registry


router = APIRouter()


@router.post("/predict", response_model=ResponseModel)
def predict(request: IrisReq):
   result = model_registry.get_model("iris_model").predict(
       pd.DataFrame([request.model_dump()])
   )

   return ResponseModel(status_code=200, data=IrisResp(target=result))

설명

@router.post("/predict", response_model=ResponseModel)

-> 작성한 Response Model도 response_model로 지정해줍니다.


def predict(request: IrisReq):

->정의한 Pydantic Request Model을 request의 hint로 작성하여 Request Body Model로 지정합니다.


 result = model_registry.get_model("iris_model").predict(
       pd.DataFrame([request.model_dump()])
   )

-> 요청 데이터를 DataFrame으로 만들고, 모델 예측을 진행합니다.


return ResponseModel(status_code=200, data=IrisResp(target=result))

->예측된 결과를 Request Model 구조에 맞게에 담아 결과를 반환합니다.



모델 불러오는 함수 작성

Mlflow에서 모델을 불러오는 함수를 작성합니다.
이미 Mlflow 및 Minio 연동은 되어있다고 가정하고 다음과 같이 작성합니다. 

# app/utils.py
import mlflow


def load_model():
   print("Loading model")
   return mlflow.sklearn.load_model("models:/iris_model/production")

# app/core/model_registry.py
class ModelRegistry:
   def __init__(self):
       self.models = {}

   def register_model(self, model_name, model):
       self.models[model_name] = model

   def get_model(self, model_name):
       return self.models[model_name]

   def clear(self):
       self.models.clear()

model_registry = ModelRegistry()



모델 저장해둘 객체 정의

Mlflow에서 불러온 모델을 저장해둘 Model Repository 라는 객체를 생성해둡니다.
서버 실행 시 해당 객체에 필요한 모델들을 저장하고 요청마다 모델을 꺼내 사용할 수 있도록 합니다.

# app/core/model_registry.py
class ModelRegistry:
   def __init__(self):
       self.models = {}

   def register_model(self, model_name, model):
       self.models[model_name] = model

   def get_model(self, model_name):
       return self.models[model_name]

   def clear(self):
       self.models.clear()

model_registry = ModelRegistry()



모델 불러오기

서버가 실행될 때 모델을 불러올 수 있도록 lifespan 함수를 작성합니다.
lifespan 함수는 API 서버가 실행될 때 수행할 사전, 사후 작업을 정의할 수 있습니다.
이제 서버가 실행 될 때 load_model 이라는 함수를 통해 모델을 불러옵니다.
모델을 불러와 사전에 정의된 model_registry에 저장합니다.

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

from app import routers
from app.core.model_registry import model_registry
from app.utils import load_model


@asynccontextmanager
async def lifespan(app: FastAPI):
   model_registry.register_model("iris_model", load_model())

   yield

   model_registry.clear()


app = FastAPI(lifespan=lifespan)

app.include_router(routers.router)



FastAPI 서버 실행하기

코드 작성이 완료 되었으면, 준비된 다음 쉘 코드를 실행합니다.
bash start.sh 혹은 zsh start.sh로 실행합니다.
해당 쉘 코드에는 mlflow, minio의 접속 정보를 포함하고 있습니다.

# start.sh
export AWS_ACCESS_KEY_ID=mlflow_admin
export AWS_SECRET_ACCESS_KEY=mlflow_admin
export MLFLOW_S3_ENDPOINT_URL=http://localhost:9000
export MLFLOW_TRACKING_URI=http://localhost:5000

uvicorn app.main:app --reload

성공시 결과 화면


profile
조정훈
이전 포스트

FastAPI - 서버만들기

다음 포스트

Cloud Run - CD 파이프라인 구축

0개의 댓글