「MLモデルを開発したけど、本番環境にデプロイする方法がわからない」「Vertex AIの設定項目が多すぎて、何から手をつければいいかわからない」——SES現場でMLプロジェクトに参画すると、こうした課題に直面することが増えています。
**Gemini CLI（Google Antigravity）を使えば、Vertex AIのモデル開発からデプロイまでのワークフロー全体を効率化できます。**自然言語でインフラ構成を指示し、コード生成・設定・デプロイを一貫して行えます。
この記事では、Gemini CLIを活用したVertex AIでのMLモデルデプロイの実践手法を、環境構築からモニタリングまで体系的に解説します。
Vertex AIの概要とGemini CLIの活用ポイント
Vertex AIとは
Vertex AIはGoogle Cloudが提供する統合MLプラットフォームで、以下の機能を提供します。
- AutoML: コードなしでMLモデルを構築
- カスタムトレーニング: TensorFlow/PyTorchでのモデルトレーニング
- Model Registry: モデルのバージョン管理
- Prediction: オンライン/バッチ予測エンドポイント
- Pipelines: MLワークフローの自動化
- Feature Store: 特徴量の管理・共有
- Model Monitoring: モデルの性能監視
Gemini CLIの活用ポイント
# Gemini CLIでVertex AIの構成を自然言語で指示
gemini "Vertex AIでカスタムモデルをトレーニングして、
オンライン予測エンドポイントにデプロイするパイプラインを構築して。
TensorFlow 2.15、GPU使用、Auto Scalingあり。"Gemini CLIがVertex AI開発で特に有効な理由は以下の通りです。
| 活用ポイント | 詳細 |
|---|---|
| Google Cloud深度理解 | GCPサービスの連携パターンを熟知 |
| Pythonコード生成 | Vertex AI SDKの使い方を正確に生成 |
| YAML/JSON生成 | パイプライン定義やモデル設定を自動生成 |
| gcloud CLI連携 | デプロイコマンドの組み立てを支援 |
| コスト最適化 | マシンタイプ選定やスケーリング設定を提案 |
環境構築
前提条件
# Google Cloud SDKの確認
gcloud version
gcloud auth list
# プロジェクト設定
gcloud config set project YOUR_PROJECT_ID
gcloud config set compute/region asia-northeast1
# 必要なAPIの有効化
gcloud services enable aiplatform.googleapis.com
gcloud services enable storage.googleapis.com
gcloud services enable cloudbuild.googleapis.com
gcloud services enable artifactregistry.googleapis.com
# Vertex AI Python SDKのインストール
pip install google-cloud-aiplatform[all]プロジェクト設定
# GEMINI.md
## Vertex AI プロジェクト設定
- プロジェクト: your-project-id
- リージョン: asia-northeast1（東京）
- Python: 3.11+
- フレームワーク: TensorFlow 2.15 / PyTorch 2.2
- MLflow: メトリクス追跡に使用
- コンテナレジストリ: Artifact Registryプロジェクト構成
ml-project/
├── GEMINI.md
├── pyproject.toml
├── src/
│ ├── data/
│ │ ├── preprocessing.py
│ │ └── feature_engineering.py
│ ├── model/
│ │ ├── architecture.py
│ │ ├── train.py
│ │ └── evaluate.py
│ ├── serving/
│ │ ├── predict.py
│ │ └── Dockerfile
│ └── pipeline/
│ ├── components.py
│ └── pipeline.py
├── configs/
│ ├── training.yaml
│ └── serving.yaml
├── tests/
│ ├── test_model.py
│ └── test_serving.py
└── notebooks/
└── exploration.ipynbカスタムモデルのトレーニング
トレーニングコードの生成
gemini "以下の仕様でTensorFlowのモデルトレーニングコードを生成して：
- 画像分類モデル（EfficientNetV2ベース）
- 転移学習（ImageNet事前学習済み）
- データ: GCSバケットの画像データ（train/val/test分割済み）
- データ拡張: ランダムフリップ、回転、明度変更
- 最適化: Adam（学習率スケジューリング付き）
- メトリクス: accuracy, precision, recall, F1
- Vertex AI Training用のDockerfile付き"Gemini CLIが生成するトレーニングコード：
# src/model/train.py
import os
import argparse
import tensorflow as tf
from google.cloud import storage, aiplatform
def create_model(num_classes: int, input_shape: tuple = (224, 224, 3)):
"""EfficientNetV2ベースの転移学習モデル"""
base_model = tf.keras.applications.EfficientNetV2B0(
include_top=False,
weights='imagenet',
input_shape=input_shape
)
base_model.trainable = False # 初期はフリーズ
model = tf.keras.Sequential([
base_model,
tf.keras.layers.GlobalAveragePooling2D(),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Dense(256, activation='relu'),
tf.keras.layers.Dropout(0.2),
tf.keras.layers.Dense(num_classes, activation='softmax')
])
return model
def create_data_pipeline(data_dir: str, batch_size: int = 32):
"""データパイプラインの構築"""
data_augmentation = tf.keras.Sequential([
tf.keras.layers.RandomFlip("horizontal"),
tf.keras.layers.RandomRotation(0.2),
tf.keras.layers.RandomBrightness(0.2),
tf.keras.layers.RandomContrast(0.2),
])
train_ds = tf.keras.utils.image_dataset_from_directory(
os.path.join(data_dir, 'train'),
image_size=(224, 224),
batch_size=batch_size,
).map(lambda x, y: (data_augmentation(x), y))
val_ds = tf.keras.utils.image_dataset_from_directory(
os.path.join(data_dir, 'val'),
image_size=(224, 224),
batch_size=batch_size,
)
return train_ds.prefetch(tf.data.AUTOTUNE), val_ds.prefetch(tf.data.AUTOTUNE)
def train(args):
"""モデルのトレーニング"""
train_ds, val_ds = create_data_pipeline(args.data_dir, args.batch_size)
num_classes = len(train_ds.class_names) if hasattr(train_ds, 'class_names') else args.num_classes
model = create_model(num_classes)
# 学習率スケジューリング
lr_schedule = tf.keras.optimizers.schedules.CosineDecay(
initial_learning_rate=args.learning_rate,
decay_steps=args.epochs * len(train_ds),
)
model.compile(
optimizer=tf.keras.optimizers.Adam(learning_rate=lr_schedule),
loss='sparse_categorical_crossentropy',
metrics=['accuracy',
tf.keras.metrics.Precision(name='precision'),
tf.keras.metrics.Recall(name='recall')]
)
callbacks = [
tf.keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True),
tf.keras.callbacks.TensorBoard(log_dir=os.path.join(args.model_dir, 'logs')),
]
history = model.fit(
train_ds,
validation_data=val_ds,
epochs=args.epochs,
callbacks=callbacks,
)
# モデル保存
model.save(os.path.join(args.model_dir, 'saved_model'))
# メトリクスをVertex AIに報告
aiplatform.log_metrics({
'accuracy': float(history.history['val_accuracy'][-1]),
'precision': float(history.history['val_precision'][-1]),
'recall': float(history.history['val_recall'][-1]),
})
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--data-dir', type=str, required=True)
parser.add_argument('--model-dir', type=str, default=os.environ.get('AIP_MODEL_DIR', '/tmp/model'))
parser.add_argument('--epochs', type=int, default=30)
parser.add_argument('--batch-size', type=int, default=32)
parser.add_argument('--learning-rate', type=float, default=1e-3)
parser.add_argument('--num-classes', type=int, default=10)
args = parser.parse_args()
train(args)Vertex AI Trainingジョブの実行
gemini "以下のトレーニングジョブをVertex AIで実行するgcloudコマンドを生成して：
- カスタムDockerイメージ使用
- GPU: NVIDIA T4 × 1
- トレーニングデータ: gs://my-bucket/data/
- 最大実行時間: 4時間
- asia-northeast1リージョン"# Dockerイメージのビルドとプッシュ
gcloud builds submit \
--tag asia-northeast1-docker.pkg.dev/PROJECT_ID/ml-images/training:latest \
--timeout=1800s
# トレーニングジョブの実行
gcloud ai custom-jobs create \
--region=asia-northeast1 \
--display-name="image-classification-training" \
--worker-pool-spec="machine-type=n1-standard-8,accelerator-type=NVIDIA_TESLA_T4,accelerator-count=1,replica-count=1,container-image-uri=asia-northeast1-docker.pkg.dev/PROJECT_ID/ml-images/training:latest" \
--args="--data-dir=gs://my-bucket/data/,--model-dir=gs://my-bucket/models/,--epochs=30,--batch-size=64"モデルのデプロイ
Model Registryへの登録
gemini "トレーニング済みモデルをVertex AI Model Registryに登録して、
オンライン予測エンドポイントにデプロイするコードを生成して。
オートスケーリング設定も含めて。"# src/serving/deploy.py
from google.cloud import aiplatform
def deploy_model():
"""モデルのデプロイ"""
aiplatform.init(
project='your-project-id',
location='asia-northeast1'
)
# モデルの登録
model = aiplatform.Model.upload(
display_name='image-classifier-v1',
artifact_uri='gs://my-bucket/models/saved_model/',
serving_container_image_uri='asia-northeast1-docker.pkg.dev/PROJECT_ID/ml-images/serving:latest',
serving_container_predict_route='/predict',
serving_container_health_route='/health',
labels={'version': 'v1', 'team': 'ml-team'},
)
# エンドポイントの作成
endpoint = aiplatform.Endpoint.create(
display_name='image-classifier-endpoint',
labels={'service': 'image-classification'},
)
# モデルのデプロイ
model.deploy(
endpoint=endpoint,
deployed_model_display_name='image-classifier-v1',
machine_type='n1-standard-4',
min_replica_count=1,
max_replica_count=5,
accelerator_type='NVIDIA_TESLA_T4',
accelerator_count=1,
traffic_percentage=100,
deploy_request_timeout=1200,
)
return endpoint
if __name__ == '__main__':
endpoint = deploy_model()
print(f"Endpoint: {endpoint.resource_name}")サービングコードの生成
# src/serving/predict.py
import os
import numpy as np
import tensorflow as tf
from flask import Flask, request, jsonify
app = Flask(__name__)
model = None
def load_model():
global model
model_path = os.environ.get('AIP_STORAGE_URI', '/models/saved_model')
model = tf.keras.models.load_model(model_path)
@app.route('/health', methods=['GET'])
def health():
return jsonify({'status': 'healthy'}), 200
@app.route('/predict', methods=['POST'])
def predict():
if model is None:
load_model()
data = request.get_json()
instances = data.get('instances', [])
predictions = []
for instance in instances:
img = tf.image.decode_image(tf.io.read_file(instance['image_uri']))
img = tf.image.resize(img, [224, 224])
img = tf.expand_dims(img, 0)
pred = model.predict(img)
class_idx = int(np.argmax(pred[0]))
confidence = float(pred[0][class_idx])
predictions.append({
'class_id': class_idx,
'confidence': confidence,
})
return jsonify({'predictions': predictions})
if __name__ == '__main__':
load_model()
app.run(host='0.0.0.0', port=int(os.environ.get('AIP_HTTP_PORT', 8080)))Vertex AI Pipelinesの構築
パイプライン定義
gemini "以下のMLパイプラインをVertex AI Pipelinesで構築して：
1. データ前処理（GCSからデータ読み込み → クレンジング → 特徴量エンジニアリング）
2. モデルトレーニング（ハイパーパラメータチューニング付き）
3. モデル評価（精度が閾値以上なら次へ）
4. Model Registryへの登録
5. エンドポイントへのデプロイ（カナリアリリース）
KFP v2の形式で出力して。"# src/pipeline/pipeline.py
from kfp import dsl
from kfp.dsl import Input, Output, Dataset, Model, Metrics
from google_cloud_pipeline_components.v1 import custom_job, model, endpoint
@dsl.component(base_image='python:3.11')
def preprocess_data(
raw_data_uri: str,
output_data: Output[Dataset],
):
"""データ前処理コンポーネント"""
import pandas as pd
from google.cloud import storage
# データ読み込みと前処理
df = pd.read_csv(raw_data_uri)
df = df.dropna()
df = df.drop_duplicates()
# 特徴量エンジニアリング
# ... 省略 ...
df.to_csv(output_data.path, index=False)
@dsl.component(base_image='python:3.11')
def evaluate_model(
model_artifact: Input[Model],
test_data: Input[Dataset],
metrics: Output[Metrics],
accuracy_threshold: float = 0.85,
) -> bool:
"""モデル評価コンポーネント"""
import tensorflow as tf
import pandas as pd
import numpy as np
model = tf.keras.models.load_model(model_artifact.path)
test_df = pd.read_csv(test_data.path)
accuracy = float(np.mean(predictions == labels))
metrics.log_metric('accuracy', accuracy)
return accuracy >= accuracy_threshold
@dsl.pipeline(
name='ml-training-pipeline',
description='End-to-end ML pipeline with Vertex AI',
)
def training_pipeline(
project: str,
location: str = 'asia-northeast1',
raw_data_uri: str = 'gs://my-bucket/raw-data.csv',
):
# Step 1: データ前処理
preprocess_task = preprocess_data(raw_data_uri=raw_data_uri)
# Step 2: トレーニング
training_task = custom_job.CustomTrainingJobOp(
display_name='model-training',
worker_pool_specs=[{
'machine_spec': {
'machine_type': 'n1-standard-8',
'accelerator_type': 'NVIDIA_TESLA_T4',
'accelerator_count': 1,
},
'replica_count': 1,
'container_spec': {
'image_uri': 'training-image:latest',
'args': ['--data-dir', preprocess_task.outputs['output_data']],
},
}],
project=project,
location=location,
)
# Step 3: 評価
eval_task = evaluate_model(
model_artifact=training_task.outputs['model'],
test_data=preprocess_task.outputs['output_data'],
)
# Step 4-5: 条件付きデプロイ
with dsl.Condition(eval_task.output == True):
deploy_task = endpoint.ModelDeployOp(
model=training_task.outputs['model'],
endpoint='projects/PROJECT/locations/REGION/endpoints/ENDPOINT_ID',
traffic_split={'0': 10}, # カナリア: 10%
)モデルモニタリング
パフォーマンス監視の設定
gemini "Vertex AIのModel Monitoringを設定して。
以下の項目を監視して：
- 予測レイテンシ（P50, P95, P99）
- スループット（QPS）
- 入力データドリフト
- 予測ドリフト
- リソース使用率（CPU/GPU/メモリ）
アラート閾値も設定して。"# monitoring/setup_monitoring.py
from google.cloud import aiplatform
def setup_monitoring(endpoint_name: str):
"""モデルモニタリングの設定"""
aiplatform.init(project='your-project-id', location='asia-northeast1')
endpoint = aiplatform.Endpoint(endpoint_name)
# モニタリングジョブの作成
monitoring_job = aiplatform.ModelDeploymentMonitoringJob.create(
display_name='image-classifier-monitoring',
endpoint=endpoint,
logging_sampling_strategy={
'random_sample_config': {'sample_rate': 0.1}
},
schedule_config={'monitor_interval': {'seconds': 3600}},
alert_config={
'email_alert_config': {
'user_emails': ['[email protected]']
}
},
objective_configs=[
{
'deployed_model_id': endpoint.traffic_split.keys()[0],
'objective_config': {
'training_dataset': {
'gcs_source': {'uris': ['gs://my-bucket/training-data/']},
'data_format': 'csv',
},
'training_prediction_skew_detection_config': {
'skew_thresholds': {
'feature_1': {'value': 0.3},
'feature_2': {'value': 0.3},
}
},
'prediction_drift_detection_config': {
'drift_thresholds': {
'prediction': {'value': 0.2},
}
},
}
}
],
)
return monitoring_jobコスト最適化
マシンタイプの選定
gemini "以下の要件でVertex AIのコストを最適化して：
- トレーニング: 画像分類、データセット10万枚
- 推論: 平均100 QPS、ピーク500 QPS
- 予算: 月額30万円以内
マシンタイプ・スケーリング設定・Spot VMの活用方法を提案して。"| 用途 | 推奨マシンタイプ | 月額コスト概算 |
|---|---|---|
| トレーニング | n1-standard-8 + T4 GPU | ¥15,000/回 |
| 推論（通常時） | n1-standard-4 × 2台 | ¥60,000/月 |
| 推論（ピーク時） | オートスケール最大5台 | ¥150,000/月 |
| バッチ予測 | n1-standard-16（Spot VM） | ¥5,000/回 |
SES現場での活用パターン
パターン1：既存モデルのVertex AI移行
gemini "ローカルで開発されたPyTorchモデルをVertex AIにデプロイして。
以下の条件で：
- model.pth（PyTorch 2.2）
- REST API形式で公開
- 認証あり（IAMベース）
- ログ出力（Cloud Logging連携）"パターン2：A/Bテストの実装
gemini "Vertex AIエンドポイントでモデルのA/Bテストを設定して。
モデルv1に90%、モデルv2に10%のトラフィックを分配。
各モデルの性能メトリクスを比較するダッシュボードも作成して。"パターン3：Feature Storeの構築
gemini "Vertex AI Feature Storeを構築して。
以下の特徴量を管理して：
- ユーザー属性（年齢、地域、会員種別）
- 行動データ（最終ログイン、購入回数、閲覧カテゴリ）
- リアルタイム特徴量（直近1時間のアクション数）
オンラインサービング・オフラインバッチ両方に対応して。"まとめ
Gemini CLIとVertex AIの組み合わせは、SES現場でのML開発を大幅に効率化します。
- 開発フェーズ: トレーニングコード・パイプラインの自動生成
- デプロイフェーズ: モデル登録・エンドポイント設定の自動化
- 運用フェーズ: モニタリング・A/Bテスト・コスト最適化
特にGemini CLIはGoogle Cloudのサービスを深く理解しているため、Vertex AI特有の設定やベストプラクティスを正確に反映したコードを生成できます。
まずは既存のモデルをVertex AIにデプロイするところから始めて、段階的にパイプラインやモニタリングを追加していきましょう。
