1. はじめに
このラボでは、Vertex AI でカスタム予測ルーチンを使用して、カスタムの前処理と後処理のロジックを記述する方法を学びます。このサンプルでは Scikit-learn を使用しますが、カスタム予測ルーチンは、XGBoost、PyTorch、TensorFlow などの他の Python ML フレームワークと連携できます。
学習内容
- カスタム予測ルーチンを使用してカスタム予測ロジックを作成する
- カスタムのサービス提供コンテナとモデルをローカルでテストする
- Vertex AI Prediction でカスタム サービング コンテナをテストする
2. Vertex AI の概要
このラボでは、Google Cloud で利用できる最新の AI プロダクトを使用します。Vertex AI は Google Cloud 全体の ML サービスを統合してシームレスな開発エクスペリエンスを提供します。以前は、AutoML でトレーニングしたモデルやカスタムモデルには、個別のサービスを介してアクセスする必要がありました。Vertex AI は、これらの個別のサービスを他の新しいプロダクトとともに 1 つの API へと結合します。既存のプロジェクトを Vertex AI に移行することもできます。
Vertex AI には、エンドツーエンドの ML ワークフローをサポートするさまざまなプロダクトが含まれています。このラボでは、Predictions と Workbench を中心に学習します。
3. ユースケースの概要
このラボでは、ランダム フォレスト回帰モデルを構築し、カット、明瞭さ、サイズなどの属性に基づいてダイヤモンドの価格を予測します。
カスタム前処理ロジックを記述して、サービング時のデータがモデルが想定する形式であることを確認します。また、予測を丸めて文字列に変換するカスタムの後処理ロジックを記述します。このロジックを記述するには、カスタム予測ルーチンを使用します。
カスタム予測ルーチンの概要
Vertex AI のビルド済みコンテナは、ML フレームワークの予測オペレーションを実行して予測リクエストを処理します。カスタム予測ルーチンの前に、予測を実行する前に入力を前処理する場合や、結果を返す前にモデルの予測を後処理する場合は、カスタム コンテナをビルドする必要があります。
カスタム サービング コンテナを構築するには、トレーニング済みモデルをラップし、HTTP リクエストをモデル入力に変換し、モデルの出力をレスポンスに変換する HTTP サーバーを作成する必要があります。
カスタム予測ルーチンでは、Vertex AI がサービング関連のコンポーネントを提供するため、ユーザーはモデルとデータ変換に集中できます。
作成するアプリの概要
ユーザー管理のノートブックをデプロイし、us-central1 にデプロイされたオンライン予測とモデル エンドポイントへのアクセスに使用するワークベンチ サブネット(以下の図 1 を参照)からなる、aiml-vpc という VPC ネットワークを設定します。
Figure1
4. チュートリアル API を有効にする
ステップ 1: Compute Engine API を有効にする
まだ有効になっていない場合は、[Compute Engine] に移動して [有効にする] を選択します。これはノートブック インスタンスを作成するために必要です。
ステップ 2: Artifact Registry API を有効にする
まだ有効になっていない場合は、[Artifact Registry] に移動して [有効にする] を選択します。これを使用してカスタムのサービス提供コンテナを作成します。
ステップ 3: Vertex AI API を有効にする
Cloud コンソールの [Vertex AI] セクションに移動し、[Vertex AI API を有効にする] をクリックします。
ステップ 4: Vertex AI Workbench インスタンスを作成する
Notebooks APII がまだ有効になっていない場合は、有効にします。
5. aiml-vpc を作成する
このチュートリアルでは、Cloud Shell で gcloud 構成を実装するために $variables を使用します。
Cloud Shell で、次のコマンドを実行します。
gcloud config list project
gcloud config set project [YOUR-PROJECT-NAME]
projectid=YOUR-PROJECT-NAME
echo $projectid
aiml-vpc を作成する
Cloud Shell で、次のコマンドを実行します。
gcloud compute networks create aiml-vpc --project=$projectid --subnet-mode=custom
ユーザー管理のノートブック サブネットを作成する
Cloud Shell 内で、workbench-subnet を作成します。
gcloud compute networks subnets create workbench-subnet --project=$projectid --range=172.16.10.0/28 --network=aiml-vpc --region=us-central1 --enable-private-ip-google-access
Cloud Router と NAT の構成
ユーザー管理のノートブックには外部 IP アドレスがないため、チュートリアルでは Cloud NAT を使用してソフトウェア パッケージをダウンロードします。Cloud NAT は下り(外向き)NAT 機能を提供します。つまり、インターネット ホストがユーザー管理ノートブックとの通信を開始できないため、ノートブックの安全性が向上します。
Cloud Shell 内で、リージョン クラウド ルーター us-central1 を作成します。
gcloud compute routers create cloud-router-us-central1-aiml-nat --network aiml-vpc --region us-central1
Cloud Shell 内で、リージョン Cloud NAT ゲートウェイ us-central1 を作成します。
gcloud compute routers nats create cloud-nat-us-central1 --router=cloud-router-us-central1-aiml-nat --auto-allocate-nat-external-ips --nat-all-subnet-ip-ranges --region us-central1
6. ユーザー管理のノートブックを作成する
ユーザー管理のサービス アカウントを作成する(ノートブック)
次のセクションでは、チュートリアルで使用する Vertex Workbench(ノートブック)に関連付けるユーザー マネージド サービス アカウントを作成します。
このチュートリアルでは、サービス アカウントに次のルールが適用されます。
Cloud Shell 内で、サービス アカウントを作成します。
gcloud iam service-accounts create user-managed-notebook-sa \
--display-name="user-managed-notebook-sa"
Cloud Shell で、ストレージ管理者のロールでサービス アカウントを更新します。
gcloud projects add-iam-policy-binding $projectid --member="serviceAccount:user-managed-notebook-sa@$projectid.iam.gserviceaccount.com" --role="roles/storage.admin"
Cloud Shell 内で、Vertex AI ユーザーロールを使用してサービス アカウントを更新します。
gcloud projects add-iam-policy-binding $projectid --member="serviceAccount:user-managed-notebook-sa@$projectid.iam.gserviceaccount.com" --role="roles/aiplatform.user"
Cloud Shell 内で、Artifact Registry 管理者ロールを持つサービス アカウントを更新します。
gcloud projects add-iam-policy-binding $projectid --member="serviceAccount:user-managed-notebook-sa@$projectid.iam.gserviceaccount.com" --role="roles/artifactregistry.admin"
Cloud Shell でサービス アカウントを一覧表示し、ユーザー管理ノートブックの作成時に使用するメールアドレスをメモします。
gcloud iam service-accounts list
ユーザー管理のノートブックを作成する
次のセクションでは、以前に作成したサービス アカウント user-managed-notebook-sa を組み込んだユーザー管理ノートブックを作成します。
Cloud Shell 内で private-client インスタンスを作成します。
gcloud notebooks instances create workbench-tutorial \
--vm-image-project=deeplearning-platform-release \
--vm-image-family=common-cpu-notebooks \
--machine-type=n1-standard-4 \
--location=us-central1-a \
--shielded-secure-boot \
--subnet-region=us-central1 \
--subnet=workbench-subnet \
--no-public-ip --service-account=user-managed-notebook-sa@$projectid.iam.gserviceaccount.com
7. トレーニング コードを作成する
ステップ 1: Cloud Storage バケットを作成する
モデルと前処理アーティファクトを Cloud Storage バケットに保存します。使用するバケットがプロジェクトにすでにある場合は、この手順をスキップできます。
ランチャーから新しいターミナル セッションを開きます。
ターミナルで次のコマンドを実行して、プロジェクトの環境変数を定義します。その際、your-cloud-project は実際のプロジェクト ID に置き換えてください。
PROJECT_ID='your-cloud-project'
次に、ターミナルで次のコマンドを実行して、プロジェクトに新しいバケットを作成します。
BUCKET="gs://${PROJECT_ID}-cpr-bucket"
gsutil mb -l us-central1 $BUCKET
ステップ 2: モデルをトレーニングする
ターミナルで cpr-codelab という新しいディレクトリを作成し、そのディレクトリに移動します。
mkdir cpr-codelab
cd cpr-codelab
ファイル ブラウザで、新しい cpr-codelab ディレクトリに移動し、ランチャーを使用して task.ipynb という新しい Python 3 ノートブックを作成します。
cpr-codelab ディレクトリは次のようになります。
+ cpr-codelab/
+ task.ipynb
ノートブックに次のコードを貼り付けます。
まず、requirements.txt ファイルを作成します。
%%writefile requirements.txt
fastapi
uvicorn==0.17.6
joblib~=1.1.1
numpy>=1.17.3, <1.24.0
scikit-learn~=1.0.0
pandas
google-cloud-storage>=2.2.1,<3.0.0dev
google-cloud-aiplatform[prediction]>=1.18.2
デプロイするモデルには、ノートブック環境とは異なる依存関係がプリインストールされます。そのため、モデルの依存関係をすべて requirements.txt にリストしてから、pip を使用してまったく同じ依存関係をノートブックにインストールします。その後、Vertex AI にデプロイする前にローカルでモデルをテストし、環境が一致していることを再確認します。
Pip は依存関係をノートブックにインストールします。
!pip install -U --user -r requirements.txt
pip のインストールが完了したら、カーネルを再起動する必要があります。
次に、モデルと前処理アーティファクトを保存するディレクトリを作成します。
USER_SRC_DIR = "src_dir"
!mkdir $USER_SRC_DIR
!mkdir model_artifacts
# copy the requirements to the source dir
!cp requirements.txt $USER_SRC_DIR/requirements.txt
cpr-codelab ディレクトリは次のようになります。
+ cpr-codelab/
+ model_artifacts/
+ scr_dir/
+ requirements.txt
+ task.ipynb
+ requirements.txt
ディレクトリ構造が設定されたので、いよいよモデルをトレーニングします。
まず、ライブラリをインポートします。
import seaborn as sns
import numpy as np
import pandas as pd
from sklearn import preprocessing
from sklearn.ensemble import RandomForestRegressor
from sklearn.pipeline import make_pipeline
from sklearn.compose import make_column_transformer
import joblib
import logging
# set logging to see the docker container logs
logging.basicConfig(level=logging.INFO)
次に、以下の変数を定義します。PROJECT_ID はプロジェクト ID に、BUCKET_NAME は前の手順で作成したバケットに置き換えてください。
REGION = "us-central1"
MODEL_ARTIFACT_DIR = "sklearn-model-artifacts"
REPOSITORY = "diamonds"
IMAGE = "sklearn-image"
MODEL_DISPLAY_NAME = "diamonds-cpr"
# Replace with your project
PROJECT_ID = "{PROJECT_ID}"
# Replace with your bucket
BUCKET_NAME = "gs://{BUCKET_NAME}"
Seaborn ライブラリからデータを読み込み、2 つのデータフレーム(1 つは特徴付き、もう 1 つはラベル付き)を作成します。
data = sns.load_dataset('diamonds', cache=True, data_home=None)
label = 'price'
y_train = data['price']
x_train = data.drop(columns=['price'])
トレーニング データを見てみましょう。各行がひし形を表していることがわかります。
x_train.head()
ラベル(対応する価格)があります。
y_train.head()
次に、sklearn 列変換を定義して、カテゴリ特徴量をワンホット エンコードし、数値特徴量をスケーリングします。
column_transform = make_column_transformer(
(preprocessing.OneHotEncoder(sparse=False), [1,2,3]),
(preprocessing.StandardScaler(), [0,4,5,6,7,8]))
ランダム フォレスト モデルを定義する
regr = RandomForestRegressor(max_depth=10, random_state=0)
次に、sklearn パイプラインを作成します。つまり、このパイプラインにフィードされるデータは、まずエンコード/スケーリングされてから、モデルに渡されます。
my_pipeline = make_pipeline(column_transform, regr)
パイプラインをトレーニング データに合わせる
my_pipeline.fit(x_train, y_train)
モデルを試して、期待どおりに機能することを確認しましょう。モデルで predict メソッドを呼び出して、テストサンプルを渡します。
my_pipeline.predict([[0.23, 'Ideal', 'E', 'SI2', 61.5, 55.0, 3.95, 3.98, 2.43]])
パイプラインを model_artifacts ディレクトリに保存し、Cloud Storage バケットにコピーできるようになりました。
joblib.dump(my_pipeline, 'model_artifacts/model.joblib')
!gsutil cp model_artifacts/model.joblib {BUCKET_NAME}/{MODEL_ARTIFACT_DIR}/
ステップ 3: 前処理アーティファクトを保存する
次に、前処理アーティファクトを作成します。このアーティファクトは、モデルサーバーの起動時にカスタム コンテナに読み込まれます。前処理アーティファクトは、ほぼすべての形式(pickle ファイルなど)にできますが、ここではディクショナリを JSON ファイルに書き出します。
clarity_dict={"Flawless": "FL",
"Internally Flawless": "IF",
"Very Very Slightly Included": "VVS1",
"Very Slightly Included": "VS2",
"Slightly Included": "S12",
"Included": "I3"}
トレーニング データの明瞭さの特徴は、常に省略形式(「Flawless」ではなく「FL」など)でした。配信時に、この機能のデータも省略されていることを確認します。これは、モデルが「FL」をワンホット エンコードする方法を「Flawless」ではありません。このカスタム前処理ロジックは後で記述します。ここでは、このルックアップ テーブルを JSON ファイルに保存してから、Cloud Storage バケットに書き込みます。
import json
with open("model_artifacts/preprocessor.json", "w") as f:
json.dump(clarity_dict, f)
!gsutil cp model_artifacts/preprocessor.json {BUCKET_NAME}/{MODEL_ARTIFACT_DIR}/
ローカルの cpr-codelab ディレクトリは次のようになります。
+ cpr-codelab/
+ model_artifacts/
+ model.joblib
+ preprocessor.json
+ scr_dir/
+ requirements.txt
+ task.ipynb
+ requirements.txt
8. CPR モデルサーバーを使用してカスタム サービング コンテナを構築する
モデルのトレーニングが完了し、前処理アーティファクトが保存されたので、カスタム サービング コンテナを構築します。通常、サービング コンテナを構築するには、モデルサーバーのコードを記述する必要があります。一方、カスタム予測ルーチンを使用すると、Vertex AI Prediction がモデルサーバーを生成し、カスタム コンテナ イメージを構築します。
カスタム サービング コンテナには、次の 3 つのコードが含まれます。
- モデルサーバー(SDK によって自動的に生成され、scr_dir/ に保存されます)
- モデルをホストする HTTP サーバー
- ルートやポートなどの設定を担当します。
- リクエスト本文のシリアル化解除、レスポンスのシリアル化、レスポンス ヘッダーの設定など、リクエストの処理のウェブサーバー要素を担当します。
- この例では、SDK で提供されているデフォルトの Handler、google.cloud.aiplatform.prediction.handler.PredictionHandler を使用します。
- 予測リクエストを処理するための ML ロジックを担当します。
これらの各コンポーネントは、ユースケースの要件に基づいてカスタマイズできます。この例では、予測器のみを実装します。
予測器は、カスタム前処理や後処理など、予測リクエストを処理する ML ロジックを担います。カスタム予測ロジックを作成するには、Vertex AI Predictor インターフェースをサブクラス化します。
このリリースのカスタム予測ルーチンには、再利用可能な XGBoost 予測と Sklearn 予測子が含まれていますが、別のフレームワークを使用する必要がある場合は、基本予測子をサブクラス化することで独自の予測を作成できます。
以下の Sklearn 予測器の例をご覧ください。このカスタムモデル サーバーを構築するために記述する必要があるのは、これだけです。
ノートブックに以下のコードを貼り付けて SklearnPredictor をサブクラス化し、src_dir/ の Python ファイルに書き込みます。この例では、読み込み、前処理、後処理の各メソッドのみをカスタマイズし、予測メソッドはカスタマイズしません。
%%writefile $USER_SRC_DIR/predictor.py
import joblib
import numpy as np
import json
from google.cloud import storage
from google.cloud.aiplatform.prediction.sklearn.predictor import SklearnPredictor
class CprPredictor(SklearnPredictor):
def __init__(self):
return
def load(self, artifacts_uri: str) -> None:
"""Loads the sklearn pipeline and preprocessing artifact."""
super().load(artifacts_uri)
# open preprocessing artifact
with open("preprocessor.json", "rb") as f:
self._preprocessor = json.load(f)
def preprocess(self, prediction_input: np.ndarray) -> np.ndarray:
"""Performs preprocessing by checking if clarity feature is in abbreviated form."""
inputs = super().preprocess(prediction_input)
for sample in inputs:
if sample[3] not in self._preprocessor.values():
sample[3] = self._preprocessor[sample[3]]
return inputs
def postprocess(self, prediction_results: np.ndarray) -> dict:
"""Performs postprocessing by rounding predictions and converting to str."""
return {"predictions": [f"${value}" for value in np.round(prediction_results)]}
それぞれの方法を詳しく見ていきましょう。
- 読み込みメソッドが前処理アーティファクトを読み込みます。このアーティファクトは、この場合はダイヤモンドの透明度の値を略語にマッピングする辞書です。
- 前処理メソッドでは、このアーティファクトを使用して、配信時に鮮明さ機能が省略された形式になるようにします。一致しない場合は、文字列全体が省略形に変換されます。
- 後処理メソッドは、予測値を $ 記号付きの文字列として返し、値を丸めます。
次に、Vertex AI Python SDK を使用してイメージをビルドします。カスタム予測ルーチンを使用すると、Dockerfile が生成され、イメージがビルドされます。
from google.cloud import aiplatform
aiplatform.init(project=PROJECT_ID, location=REGION)
import os
from google.cloud.aiplatform.prediction import LocalModel
from src_dir.predictor import CprPredictor # Should be path of variable $USER_SRC_DIR
local_model = LocalModel.build_cpr_model(
USER_SRC_DIR,
f"{REGION}-docker.pkg.dev/{PROJECT_ID}/{REPOSITORY}/{IMAGE}",
predictor=CprPredictor,
requirements_path=os.path.join(USER_SRC_DIR, "requirements.txt"),
)
予測用の 2 つのサンプルを含むテストファイルを作成します。一方のインスタンスには明瞭な略称が使われていますが、もう一方のインスタンスは最初に変換する必要があります。
import json
sample = {"instances": [
[0.23, 'Ideal', 'E', 'VS2', 61.5, 55.0, 3.95, 3.98, 2.43],
[0.29, 'Premium', 'J', 'Internally Flawless', 52.5, 49.0, 4.00, 2.13, 3.11]]}
with open('instances.json', 'w') as fp:
json.dump(sample, fp)
ローカルモデルをデプロイして、コンテナをローカルでテストします。
with local_model.deploy_to_local_endpoint(
artifact_uri = 'model_artifacts/', # local path to artifacts
) as local_endpoint:
predict_response = local_endpoint.predict(
request_file='instances.json',
headers={"Content-Type": "application/json"},
)
health_check_response = local_endpoint.run_health_check()
予測結果は、次のコマンドで確認できます。
predict_response.content
9. モデルを Vertex AI にデプロイする
コンテナをローカルでテストしたので、次はイメージを Artifact Registry に push し、モデルを Vertex AI Model Registry にアップロードします。
まず、Artifact Registry にアクセスできるように Docker を構成します。
!gcloud artifacts repositories create {REPOSITORY} --repository-format=docker \
--location=us-central1 --description="Docker repository"
!gcloud auth configure-docker {REGION}-docker.pkg.dev --quiet
次に、イメージを push します。
local_model.push_image()
モデルをアップロードします
model = aiplatform.Model.upload(local_model = local_model,
display_name=MODEL_DISPLAY_NAME,
artifact_uri=f"{BUCKET_NAME}/{MODEL_ARTIFACT_DIR}",)
モデルがアップロードされると、コンソールに表示されます。
次に、オンライン予測で使用できるようにモデルをデプロイします。カスタム予測ルーチンはバッチ予測でも機能するため、ユースケースでオンライン予測が必要ない場合はモデルをデプロイする必要はありません。
次に、イメージを push します。
endpoint = model.deploy(machine_type="n1-standard-2")
最後に、予測を取得して、デプロイされたモデルをテストします。
endpoint.predict(instances=[[0.23, 'Ideal', 'E', 'VS2', 61.5, 55.0, 3.95, 3.98, 2.43]])
お疲れさまでした
Vertex AI を使って次のことを行う方法を学びました。
- カスタム予測ルーチンを使用してカスタム前処理と後処理のロジックを作成する
Cosmopup は Codelab を素晴らしいと思っています!