KServe Transformer 개발 가이드

테스트 환경

KServe 0.13에서 테스트를 진행하였고, v2 protocol을 사용했다.

 

개요

KServe Transformer는 머신러닝 모델의 입력 전처리와 출력 후처리를 담당하는 컴포넌트다. 

사용자가 KServe endpoint에 추론 요청을 보내면, transformer는 사용자가 입력한 input 데이터를 모델에 맞게 전처리한 후 predictor로 요청을 전송한다. 

추론이 완료되면 predictor는 transformer로 결괏값을 전송하고, transformer는 결괏값을 사용자 친화적인 값으로 수정 후 사용자에게 전달한다.

 

이 가이드에서는 Iris 꽃 분류 모델을 위한 Transformer를 개발하는 과정을 설명한다.

전처리가 필요하지 않은 데이터라, 전처리 과정은 필요없지만, 구조를 설명하기 위해 단순히 입력받은 값을 그대로 predictor로 전달하도록 했고, 추론의 결과를 카테고리와 매핑해 사용자에게 전달하는 후처리 과정을 추가했다.

 

Model 준비

아래 링크에서 생성한 모델을 기반으로 transformer를 구현한다.

KServe를 이용한 scikit-learn 모델 배포 및 사용 가이드

 

Transformer 클래스 구현

전체적인 구조는 아래와 같으며, preprocess와 postprocess 함수에 로직을 구현하면 된다.

전처리, 후처리가 모두 필요한 경우가 아니라면 둘 중 하나만 구현해도 상관없다.

import argparse
from typing import Dict, Union
from kserve import (
    Model,
    ModelServer,
    model_server,
    InferInput,
    InferRequest,
    InferResponse,
)
from kserve.model import PredictorConfig

class IrisTransformer(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.ready = True


    def preprocess(
        self, payload: Union[Dict, InferRequest], headers: Dict[str, str] = None
    ) -> Union[Dict, InferRequest]:

        # InferRequest 객체 처리
        data = payload.inputs[0].data

        # 모델 입력 형식으로 변환
        infer_inputs = [InferInput(name="INPUT__0", datatype="FP32", shape=[len(data), 4], data=data)]
        return InferRequest(model_name=self.name, infer_inputs=infer_inputs)

    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=[model_server.parser])
args, _ = parser.parse_known_args()

if __name__ == "__main__":

    model = IrisTransformer(
        args.model_name,
        predictor_host=args.predictor_host,
        predictor_protocol=args.predictor_protocol,
        predictor_use_ssl=args.predictor_use_ssl,
    )
    ModelServer().start([model])

 

이미지 빌드를 위한 requirements.txt 작성

이미지에 설치해야할 파이썬 패키지 리스트를 작성한다.

kserve==0.13.1
argparse
pandas

 

이미지 빌드를 위한 Dockerfile 작성

KServe에 배포한 모델이 v2 protocol을 사용하기에 `--predictor_protocol`을 v2로 설정했다.

v1 protocol을 사용하는 경우에는 `--predictor_protocol`을 별도로 지정하지 않아도 된다.

FROM python:3.8-slim
COPY requirements.txt .
COPY transformer transformer
RUN pip install -U pip && pip install -r requirements.txt
ENTRYPOINT ["python", "-m", "transformer.iris-transformer", "--predictor_protocol", "v2"]

 

Docker 이미지 빌드

전체 디렉토리 구조는 다음과 같다.

├── Dockerfile
├── requirements.txt
└── transformer
    └── iris-transformer.py

 

아래의 명령어를 사용해 이미지를 빌드하고, private container registry에 push한다.

docker build -t registry.dd.io/shared/iris-transformer .
docker push registry.dd.io/shared/iris-transformer

 

Harbor 접근을 위한 서비스 계정을 생성

Harbor에 접근하기 위한 secret을 생성한다.

kubectl create secret generic docker-config-secret --from-file=.dockerconfigjson=/root/.docker/config.json --type=kubernetes.io/dockerconfigjson -n kubeflow-user-example-com

 

S3, Harbor에 접근하기 위한 ServiceAccount를 생성한다.

apiVersion: v1
kind: ServiceAccount
metadata:
  name: s3-sa
  namespace: kubeflow-user-example-com
secrets:
  - name: s3-secret
  - name: docker-config-secret

 

InferenceService 생성

Transformer를 포함한 InferenceService를 배포하기 위해 다음과 같은 YAML 파일을 작성

apiVersion: "serving.kserve.io/v1beta1"
kind: "InferenceService"
metadata:
  name: "iris-svm-transformer"
spec:
  predictor:
    serviceAccountName: s3-sa
    model:
      modelFormat:
        name: sklearn
      runtime: kserve-sklearnserver
      protocolVersion: v2
      storageUri: "s3://sandbox/iris_svm/v1/"
  transformer:
    containers:
      - image: registry.dd.io/shared/iris-transformer:latest
        name: kserve-container

 

테스트

배포된 모델에 예측 요청을 보낸다.

import requests
import json
import pandas as pd
from pprint import pprint
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-svm-transformer.kubeflow-user-example-com.10.0.2.6.sslip.io/v2/models/iris-svm-transformer/infer", 
    data=json.dumps(payload)
)   

pprint(resp.json())

 

 결괏값을 보면 transformer를 통해 카테고리와 매핑된 값이 리턴된 것을 확인할 수 있다.

{'id': '65f0f7d8-8c8a-46b3-a685-5871c59e1ed4',
 'model_name': 'iris-svm-transformer',
 'model_version': None,
 'outputs': [{'data': ['setosa', 'setosa', 'setosa'],
              'datatype': 'BYTES',
              'name': 'OUTPUT__0',
              'parameters': None,
              'shape': [3]}],
 'parameters': None}