カスタム予測ルーチン

カスタム予測ルーチン（CPR）を使用すると、HTTP サーバーの設定やコンテナのゼロからのビルドを行うことなく、前処理コードと後処理コードを使用してカスタム コンテナを簡単にビルドできます。前処理を使用して入力を正規化または変換する、外部サービスを呼び出して追加データを取得する、後処理を使用してモデル予測の形式を設定する、あるいはビジネス ロジックを実行する、などの処理が可能です。

次の図は、カスタム予測ルーチンがある場合とない場合の両方のユーザー ワークフローを示しています。

主な違いは次のとおりです。

  • モデルサーバーや Dockerfile を記述する必要はありません。モデルサーバー（モデルをホストする HTTP サーバー）が用意されています。

  • モデルをローカルでデプロイしてデバッグできるため、開発中のイテレーション サイクルを高速化できます。

カスタム コンテナをビルドしてデプロイする

このセクションでは、CPR を使用して前処理と後処理のロジックを備えたカスタム コンテナをビルドし、ローカル エンドポイントとオンライン エンドポイントの両方にデプロイする方法について説明します。

設定

環境に Vertex AI SDK for Python と 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 形式の予測リクエストのシリアル化解除など、ウェブサーバー関連のロジックをカスタマイズする必要がある、非常にまれなケースで役立ちます。

Predictor と Handler の両方を実装するサンプル ノートブックをご覧ください。

必須ではありませんが、コードの編成と再利用性を高めるために、デフォルトのハンドラで示されるように、ウェブサーバー ロジックを Handler に、ML ロジックを Predictor に実装することをおすすめします。

カスタム コンテナをビルドする

イメージにパッケージをインストールする必要がある場合は、カスタムコードと追加の requirements.txt ファイルをディレクトリに配置します。

Vertex AI SDK for Python を使用して、次のようにカスタム コンテナをビルドします。

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 に push します。

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},
)

モデルを Model Registry にアップロードしたら、モデルを使用してバッチ予測を取得できます。または、Vertex AI エンドポイントにデプロイしてオンライン予測を取得することもできます。

Vertex AI エンドポイントにデプロイする

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

モデルをデプロイすると、オンライン予測を取得できます。

ノートブックの例

このサンプルでは、Vertex AI Prediction を使用してカスタムの前処理と後処理を備えたモデルのデプロイ方法をいくつか紹介します。