커스텀 예측 루틴

커스텀 예측 루틴(CPR)을 사용하면 HTTP 서버를 설정하거나 컨테이너를 처음부터 새로 만드는 세부사항을 다루지 않고 사전/사후 처리 코드로 커스텀 컨테이너를 쉽게 빌드할 수 있습니다. 사전 처리를 사용하여 입력을 정규화/변환하거나 외부 서비스를 호출하여 추가 데이터를 가져올 수 있으며, 사후 처리를 사용하여 모델 예측 형식을 지정하거나 비즈니스 로직을 실행할 수 있습니다.

다음 다이어그램은 커스텀 예측 루틴이 있거나 없는 사용자 워크플로를 보여줍니다.

가장 큰 차이점은 다음과 같습니다.

  • 모델 서버나 Dockerfile을 작성할 필요가 없습니다. 모델을 호스팅하는 HTTP 서버인 모델 서버가 제공됩니다.

  • 로컬에서 모델을 배포 및 디버깅하여 개발 중에 반복 주기를 앞당길 수 있습니다.

커스텀 컨테이너 빌드 및 배포

이 섹션에서는 CPR을 사용하여 사전/사후 처리 로직으로 커스텀 컨테이너를 빌드하고 로컬 및 온라인 엔드포인트 모두에 배포하는 방법을 설명합니다.

설정

환경에 Vertex AI SDK와 Docker가 설치되어 있어야 합니다.

커스텀 Predictor 작성

Predictor 인터페이스를 구현합니다.

class Predictor(ABC):
    """Interface of the Predictor class for Custom Prediction Routines.
    The Predictor is responsible for the ML logic for processing a prediction request.
    Specifically, the Predictor must define:
    (1) How to load all model artifacts used during prediction into memory.
    (2) The logic that should be executed at predict time.
    When using the default PredictionHandler, the Predictor will be invoked as follows:
      predictor.postprocess(predictor.predict(predictor.preprocess(prediction_input)))
    """

    @abstractmethod
    def load(self, artifacts_uri: str) -> None:
        """Loads the model artifact.
        Args:
            artifacts_uri (str):
                Required. The value of the environment variable AIP_STORAGE_URI.
        """
        pass

    def preprocess(self, prediction_input: Any) -> Any:
        """Preprocesses the prediction input before doing the prediction.
        Args:
            prediction_input (Any):
                Required. The prediction input that needs to be preprocessed.
        Returns:
            The preprocessed prediction input.
        """
        return prediction_input

    @abstractmethod
    def predict(self, instances: Any) -> Any:
        """Performs prediction.
        Args:
            instances (Any):
                Required. The instance(s) used for performing prediction.
        Returns:
            Prediction results.
        """
        pass

    def postprocess(self, prediction_results: Any) -> Any:
        """Postprocesses the prediction results.
        Args:
            prediction_results (Any):
                Required. The prediction results.
        Returns:
            The postprocessed prediction results.
        """
        return prediction_results

예시는 Sklearn의 Predictor 구현을 참조하세요.

커스텀 Handler 작성(선택사항)

커스텀 핸들러는 원시 요청 객체에 액세스할 수 있으므로, 추가 요청/응답 헤더를 지원하거나 JSON 외 형식의 예측 요청을 역직렬화하는 등 드물지만 웹 서버 관련 로직을 맞춤설정해야 하는 경우에 유용합니다.

다음은 예측자와 핸들러를 모두 구현하는 샘플 노트북입니다.

필수는 아니지만 코드 구성과 재사용성을 높이려면 기본 핸들러에 표시된 것처럼 핸들러에 웹 서버 로직을 구현하고 예측자에 ML 로직을 구현하는 것이 좋습니다.

커스텀 컨테이너 빌드

이미지의 패키지를 설치해야 하는 경우 커스텀 코드와 추가 requirements.txt 파일을 디렉터리에 추가합니다.

Vertex SDK를 사용하여 다음과 같이 커스텀 컨테이너를 빌드합니다.

from google.cloud.aiplatform.prediction import LocalModel

# {import your predictor and handler}

local_model = LocalModel.build_cpr_model(
    {PATH_TO_THE_SOURCE_DIR},
    f"{REGION}-docker.pkg.dev/{PROJECT_ID}/{REPOSITORY}/{IMAGE}",
    predictor={PREDICTOR_CLASS},
    handler={HANDLER_CLASS},
    requirements_path={PATH_TO_REQUIREMENTS_TXT},
)

컨테이너 사양을 조사하여 이미지 URI 및 환경 변수와 같은 유용한 정보를 얻을 수 있습니다.

local_model.get_serving_container_spec()

로컬로 컨테이너 실행(선택사항)

이 단계는 더 빠른 반복을 위해 유용한 로컬에서 컨테이너를 실행하고 테스트하려는 경우에만 필요합니다. 다음 예시에서는 로컬 엔드포인트에 배포하고 예측 요청(요청 본문 형식)을 보냅니다.

with local_model.deploy_to_local_endpoint(
    artifact_uri={GCS_PATH_TO_MODEL_ARTIFACTS},
    credential_path={PATH_TO_CREDENTIALS},
) as local_endpoint:
    health_check_response = local_endpoint.run_health_check()
    predict_response = local_endpoint.predict(
        request_file={PATH_TO_INPUT_FILE},
        headers={ANY_NEEDED_HEADERS},
    )

상태 점검 및 예측 응답을 출력합니다.

print(health_check_response, health_check_response.content)
print(predict_response, predict_response.content)

모든 컨테이너 로그를 출력합니다.

local_endpoint.print_container_logs(show_all=True)

Vertex AI Model Registry에 업로드

모델은 모델 아티팩트(학습의 파일)에 액세스해야 하므로 이를 Google Cloud Storage에 업로드해야 합니다.

Artifact Registry로 이미지를 푸시합니다.

local_model.push_image()

그런 다음 Model Registry에 업로드합니다.

from google.cloud import aiplatform

model = aiplatform.Model.upload(
    local_model=local_model,
    display_name={MODEL_DISPLAY_NAME},
    artifact_uri={GCS_PATH_TO_MODEL_ARTIFACTS},
)

모델을 Vertex AI Model Registry에 업로드하면 이 모델을 사용하여 일괄 예측을 수행하거나 Vertex AI 엔드포인트에 배포하여 온라인 예측을 수행할 수 있습니다.

Vertex AI 엔드포인트에 배포

endpoint = model.deploy(machine_type="n1-standard-4")

배포하면 온라인 예측을 수행할 수 있습니다.

노트북 샘플

이 샘플은 Vertex AI 예측에서 커스텀 전처리 및후처리를 사용하여 모델을 배포할 수 있는 다양한 방법을 보여줍니다.