KServe Custom Predictor 이미지 빌드 가이드 - v2 protocol

테스트 환경

KServe 0.13에서 테스트를 진행하였고, v2 protocol을 사용하는 predictor를 개발한다.

 

개요

KServe를 위한 custom Predictor 이미지를 빌드하는 과정을 소개한다.

 

KServe.Model 기본 클래스는 주로 세 가지 핸들러인 preprocess, predict, postprocess를 정의하고, 이 핸들러들은 순차적으로 실행된다.

• preprocess의 출력이 predict의 입력으로 전달된다.
• predict 핸들러는 모델에 대한 추론을 실행한다.
• postprocess 핸들러는 원시 예측 결과를 사용자 친화적인 추론 응답으로 변환한다.

추가적으로 load 핸들러가 있는데, 이는 로컬 파일 시스템이나 원격 모델 저장소에서 모델을 메모리로 로드하는 데 사용된다. 일반적으로는 모델 서버 클래스의 __init__ 함수에서 load 핸들러를 호출한다. 

 

프로젝트 구조

프로젝트 구조는 다음과 같다.

project_root/
│
├── predictor.py
├── requirements.txt
├── Dockerfile
└── model/
    └── model.joblib

 

1. 모델 생성 및 저장

먼저, scikit-learn을 사용하여 모델을 준비한다. 이 예제에서는 Iris 데이터셋을 사용한 SVM 분류기를 만들어 배포한다.

from sklearn import svm
from sklearn import datasets
from joblib import dump

# 데이터 로드
iris = datasets.load_iris()
X, y = iris.data, iris.target

# 모델 생성 및 학습
clf = svm.SVC(gamma='scale')
clf.fit(X, y)

# 모델을 'model.joblib' 파일로 저장
dump(clf, 'model.joblib')

 

2. Custom Predictor 구현

predictor.py 파일에 custom Predictor 클래스를 구현한다.

이 예제에서는 predict와 postprocess만 구현하였다.

import argparse

from typing import Dict, Union
import numpy as np
import kserve
from kserve import Model, ModelServer
from kserve import logging
from kserve.model import PredictorConfig
from kserve import (
    Model,
    ModelServer,
    InferInput,
    InferResponse,
)
from joblib import load
from kserve.utils.utils import generate_uuid

class IrisSVMModel(Model):
    def __init__(
        self,
        name: str,
        predictor_host: str,
        predictor_protocol: str,
        predictor_use_ssl: bool,
    ):
        super().__init__(
            name, PredictorConfig(predictor_host, predictor_protocol, predictor_use_ssl)
        )
        self.load()

    def load(self):
        self.model = load('/mnt/model/model.joblib')
        self.ready = True

    def predict(self, payload: Dict, headers: Dict[str, str] = None) -> Dict:
        data = payload.inputs[0].data
        inputs = np.array(data)
        results = self.model.predict(inputs).tolist()

        return InferResponse(
            model_name=self.name,
            infer_outputs=[InferInput(name="OUTPUT_0", datatype="INT64", shape=[len(results)], data=results)],
            response_id=generate_uuid()
        )

    def postprocess(
        self, infer_response: Union[Dict, InferResponse], headers: Dict[str, str] = None
    ) -> Union[Dict, InferResponse]:

        categories = ['setosa', 'versicolor', 'virginica']

        predictions = infer_response.outputs[0].data

        results = [categories[int(pred)] for pred in predictions]

        return InferResponse(
            model_name=self.name,
            infer_outputs=[InferInput(name="OUTPUT__0", datatype="BYTES", shape=[len(results)], data=results)],
            response_id=infer_response.id
        )


parser = argparse.ArgumentParser(parents=[kserve.model_server.parser])
args, _ = parser.parse_known_args()

if __name__ == "__main__":
    if args.configure_logging:
        logging.configure_logging(args.log_config_file)  # Configure kserve and uvicorn logger
    model = IrisSVMModel(
        args.model_name,
        predictor_host=args.predictor_host,
        predictor_protocol=args.predictor_protocol,
        predictor_use_ssl=args.predictor_use_ssl,
    )

    ModelServer().start([model])

3. requirements.txt 파일 작성

requirements.txt 파일에 필요한 의존성을 명시한다.

kserve==0.13.1
joblib
scikit-learn
numpy

 

4. Dockerfile 작성

FROM python:3.8-slim

COPY requirements.txt .
COPY predictor.py .
COPY model/ /mnt/model/

RUN pip install -U pip && pip install -r requirements.txt

ENTRYPOINT ["python", "-m", "predictor", "--predictor_protocol", "v2"]

 

5. Docker 이미지 빌드 및 푸시

이미지를 빌드하고 레지스트리에 푸시한다.

docker build -t registry.dd.io/shared/iris-predictor:v1 .
docker push registry.dd.io/shared/iris-predictor:v1

 

6. InferenceService 배포

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: iris-custom
spec:
  predictor:
    serviceAccountName: s3-sa
    containers:
      - name: kserve-container
        image: registry.dd.io/shared/iris-predictor:v1
        args:
        - --model_name=iris-custom

 

7. 테스트

InferenceService 상태를 확인한다.

[root@km iris-predictor-v1]# k get isvc -n kubeflow-user-example-com
NAME                   URL                                                                       READY   PREV   LATEST   PREVROLLEDOUTREVISION   LATESTREADYREVISION                    AGE
iris-custom            http://iris-custom.kubeflow-user-example-com.10.0.2.6.sslip.io            True           100                              iris-custom-predictor-00001            11s

 

샘플 데이터를 사용해 추론 요청을 전송한다.

import requests
import json
from pprint import pprint
import pandas as pd
df = pd.DataFrame({
    'sepal length (cm)': [5.1, 4.9, 4.7],
    'sepal width (cm)': [3.5, 3.0, 3.2],
    'petal length (cm)': [1.4, 1.4, 1.3],
    'petal width (cm)': [0.2, 0.2, 0.2]
})

# KServe v2 프로토콜에 맞는 payload 구성
payload = {
    "inputs": [
        {
            "name": "INPUT_0",
            "shape": list(df.shape),
            "datatype": "FP32",
            "data": df.values.tolist()
        }
    ]
}

resp = requests.post(
    "http://iris-custom.kubeflow-user-example-com.10.0.2.6.sslip.io/v2/models/iris-custom/infer", 
    json=payload
)   

resp.json()

 

결과를 확인한다.

{'model_name': 'iris-custom',
 'model_version': None,
 'id': 'c3b9f1af-10d5-47c1-962b-f142659ac03d',
 'parameters': None,
 'outputs': [{'name': 'OUTPUT__0',
   'shape': [3],
   'datatype': 'BYTES',
   'parameters': None,
   'data': ['setosa', 'setosa', 'setosa']}]}