MLOps - FastAPI 서빙

API

- 애플리케이션 프로그래밍 인터페이스

- 컴퓨터나 컴퓨터 프로그램 사이의 연결

- 일종의 소프트웨어 인터페이스이며 다른 종류의 소프트웨어에 서비스를 제공한다.

 

FastAPI

빠르게 API를 만듦

모던하고 빠르다.

pip install "fastapi[all]"

모든 의존성을 설치

 

(myenv) (base) dinoqos@jangjeong-uui-MacBookAir 07_api_serving % uvicorn fastapi_tutorial:app --reload --host 0.0.0.0

 

 

docs로 받는 값을 확인 가능

 

path 파라미터

qurey 파라미터

 

 

path 파리미터 qurey 파라미터 차이

 

path는 주소에서 입력

qurey는 바디에 값을 넣어서 실행

 

 

데이터 스키마

 

입력출력을 조금 더 체계적으로 관리

 

from fastapi import FastAPI
from pydantic import BaseModel


#
# Data out schema
#
class ItemOut(BaseModel):
    item_id: int
    item_body: str


# Create a FastAPI instance
app = FastAPI()


@app.get("/item/", response_model=ItemOut)
def read_item_with_query_and_pydantic(item_id: int) -> ItemOut:
    item_body = str(item_id + 1)
    return ItemOut(item_id=item_id, item_body=item_body)

 

pydantic BaseModel을 상속

 

입력되게 되면 +1을 해서 output하게 해줌

 

 

CRUD

생성 읽기 갱신 삭제

 

POST

GET

PUT

DELETE

 

from fastapi import FastAPI
from pydantic import BaseModel

# create a Fastapi instance
app = FastAPI()

#
# database
#
ITEMS = {0: "default data"}

데이터베이스를 딕셔너리 형태로 만듦

 

#
# create
#
class ItemCreateIn(BaseModel):
    item_body: str


class ItemCreateOut(BaseModel):
    item_id: int
    item_body: str


@app.post("/item/", response_model=ItemCreateOut)
def create_item(item_create_in: ItemCreateIn) -> ItemCreateOut:
    item_id = len(ITEMS)
    item_body = item_create_in.item_body
    ITEMS[item_id] = item_body
    return ItemCreateOut(item_id=item_id, item_body=item_body)

post 메소드를 통해 create을 할 수 있음

1. item_create_in: ItemCreateIn: 이 함수는 ItemCreateIn 형식의 인자인 item_create_in을 받습니다. 이는 새로운 항목의 내용을 포함하는 객체입니다.
2. -> ItemCreateOut: 이 함수는 ItemCreateOut 형식의 값을 반환합니다. 이는 새로운 항목이 생성된 후의 응답 형식을 나타냅니다.
3. item_id = len(ITEMS): ITEMS 딕셔너리의 길이를 이용하여 새로운 item_id를 생성합니다. 이렇게 함으로써 새로운 항목이 추가될 때마다 item_id가 증가하게 됩니다.
4. item_body = item_create_in.item_body: 입력으로 받은 item_create_in 객체의 item_body 필드 값을 가져와 item_body 변수에 저장합니다. 이는 새로운 항목의 내용을 의미합니다.
5. ITEMS[item_id] = item_body: ITEMS 딕셔너리에 새로운 항목을 추가합니다. 이 항목의 키는 새로운 item_id이고 값은 새로운 항목의 내용인 item_body입니다.
6. return ItemCreateOut(item_id=item_id, item_body=item_body): ItemCreateOut 클래스의 객체를 생성하여 반환합니다. 이 객체는 새로운 항목의 item_id와 item_body를 포함하고 있습니다. 이것이 클라이언트에게 반환되어 새 항목이 성공적으로 생성되었음을 알려줍니다.

 

 

read

#
# read
#
class ItemGetOut(BaseModel):
    item_id: int
    item_body: str


@app.get("/item/", response_model=ItemGetOut)
def read_item(item_id: int) -> ItemGetOut:
    item_body = ITEMS.get(item_id, "Not valid id")
    return ItemGetOut(item_id=item_id, item_body=item_body)

id값이 입력되면 ITEMS에서 조회하여 반환.

 

update

#
# update
#
class ItemUpdateIn(BaseModel):
    item_id: int
    item_body: str


class ItemUpdateOut(BaseModel):
    item_id: int
    item_body: str


@app.put("/item/", response_model=ItemUpdateOut)
def update_item(item_update_in: ItemUpdateIn) -> ItemUpdateOut:
    item_id = item_update_in.item_id
    item_body = item_update_in.item_body
    if item_id not in ITEMS:
        item_body = "Not valid id"
    else:
        ITEMS[item_id] = item_body
    return ItemUpdateOut(item_id=item_id, item_body=item_body)

예외처리

 

delete

#
# delete
#
class ItemDeleteIn(BaseModel):
    item_id: int


class ItemDeleteOut(BaseModel):
    item_id: int
    item_body: str


@app.delete("/item/", response_model=ItemDeleteOut)
def delete_item(item_delete_in: ItemDeleteIn) -> ItemDeleteOut:
    item_id = item_delete_in.item_id
    if item_id not in ITEMS:
        item_body = "Not valid id"
    else:
        item_body = ITEMS[item_id]
        del ITEMS[item_id]
    return ItemDeleteOut(item_id=item_id, item_body=item_body)

iitem_id가 입력되면 del 함수를 통해 제거

 

 

post

 

get

 

update

 

delete

 

CRUD 가능

uvicorn fastapi_crud_tutorial:app --reload --host 0.0.0.0

실행 터미널 코드

 

 

 

 

 

API 서빙

모델에 추론을 요청할 수 있는 API를 노출해 모델을 서빙

 

배치 서빙

- 1시간 단위의 추천 상품 추론 후 저장, 저장된 추론값을 사용

 

API 서빙

- 개/고양이 이미지에 대한 분류를 요청 즉시 추론 후 결과값 반환

 

API모델로 노출

 

 

CORE

BENTOML

tesorflow extended

torch serve

등

 

모델을 환경을 보전해주고 API를 노출 후 모델을 넣음

 

다시한번 학습을 시키고 모델을 생성하여 다운로드를 한다.

import mlflow
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel


#
# load model
#
model = mlflow.pyfunc.load_model("./downloads/my_model")


#
# Data in and out schema
#
class PredictIn(BaseModel):
    sepal_length: float
    sepal_width: float
    petal_length: float
    petal_width: float


class PredictOut(BaseModel):
    iris_class: str


#
# Write fastapi app
#

app = FastAPI()


@app.post("/predict", response_model=PredictOut)
def predict(data: PredictIn) -> PredictOut:
    df = pd.DataFrame([data.dict()])
    df.columns = df.columns.str.replace("_", " ")
    df = df.add_suffix(" (cm)")
    pred = model.predict(df).item()
    return PredictOut(iris_class=pred)

 

추론값을 입력받은 후 아웃풋값을 반환하게끔 한다.

 

데이터를 딕셔너리 형태로 저장되게끔 df로 만든 . ㅎ

컬럼에서 사이즈 값을 추론할 때 다른 str값이 들어가 에러가 일어나는것을 방지하기 위해 _를 공백으로 변환

cm로 열이름을 지정

pred를 통해 예측 (item로 단일값 반환)

 

여기서 pandas 에러가 발생했는데, 해당 에러는

train.py에서 따로 pandas를 import 해줘야 데이터프레임을 읽을 수 있다는 점을 알게 되었습니다.

    def predict(self, X):
        import pandas as pd

        X_pred = self.clf.predict(X)
        X_pred_df = pd.Series(X_pred).map({0: "virginica", 1: "setosa", 2: "versicolor"})
        return X_pred_df

 

 

API 서빙 이미지

docker build -t api-serving -f api.Dockerfile .

 

docker run -p 8000:8000 api-serving