Vertex AI PipelinesによるMLモデル評価バッチの自動化手法

Vertex AI PipelinesでML評価バッチを自動化する：手動運用をゼロにするPython実装ハンズオン

2026年1月5日更新 2026年3月24日約14分で読めます

文字サイズ:

Vertex AI PipelinesでML評価バッチを自動化する：手動運用をゼロにするPython実装ハンズオン

この記事の要点

MLモデルの精度劣化を自動で検知
手動による評価・監視作業を撤廃
Vertex AI Pipelinesによる安定したワークフロー構築

あなたは今、自分が先月デプロイしたモデルの精度が、今日この瞬間も維持されていると自信を持って言えますか？

もし、その答えを確認するために、Jupyter Notebookを開き、CSVデータをダウンロードし、手元のスクリプトを走らせる必要があるなら、この記事はあなたのためのものです。

ビジネスの現場において、優秀なデータサイエンティストが「運用フェーズのルーチンワーク」に時間を奪われている状況は、大きな機会損失です。モデルを作ることは創造的で楽しいものですが、定期的な再評価レポートの作成は、必ずしも創造的な作業とは言えません。

しかし、この作業はビジネス価値を担保する上で極めて重要です。人が行う以上、見落としや誤りが発生するリスクも伴います。

今回は、Google Cloudのマネージドサービスである Vertex AI Pipelines を使い、この評価プロセスを完全に自動化する「評価バッチパイプライン」を構築します。大規模なMLOps基盤チームがいなくても、Pythonコードだけで自動化フローを作り上げることができます。まずは動くプロトタイプを作り、仮説を即座に形にして検証していきましょう。

なぜ「モデル評価」をバッチ化する必要があるのか

実装に入る前に、なぜエンジニアリングリソースを割くべきなのかを明確にしておきましょう。単に「楽をしたいから」だけではありません。そこにはビジネス上の明確なリスクが存在するからです。

「作った時が最高性能」の罠

機械学習モデルには、従来のソフトウェアとは異なる性質があります。それは「コードは変わらなくても、振る舞いが変わる（劣化する）」という点です。

データドリフト (Data Drift): 入力データの分布が、学習時のデータ分布から乖離していく現象。
コンセプトドリフト (Concept Drift): 入力と出力の関係性自体が変化してしまう現象。

例えば、ECサイトのレコメンドモデルなら、ユーザーの嗜好の変化や季節性のトレンド変化がこれに当たります。リリース直後に高い性能を示したモデルも、時間が経つにつれて性能が低下することがあります。

手動評価が引き起こす運用リスク

一般的な運用現場では、この劣化検知を「担当者が定期的にチェックする」という運用でカバーしようとすることがあります。しかし、これにはリスクが伴います。

属人化による評価基準のブレ: 担当者によって重視する指標が異なる場合、「モデルが劣化しているか否か」の判断基準が変わってしまう可能性があります。
検知の遅れ: 定期チェックでは、異常が発生してから検知までに時間がかかることがあります。その間、ビジネス機会の損失や誤った予測による損害が発生し続けるかもしれません。
形骸化: 状況によってはチェックがスキップされがちです。問題はそういった時に起こる可能性があります。

本チュートリアルのゴール：評価の完全自動化

今回目指すのは、以下の状態です。

定期的かつ自動的に最新データを取得し、推論を実行する。
事前に定義した厳格な指標でスコアを算出する。
スコアが閾値を下回った場合のみ、Slackやメールで人間に通知する。

つまり、「便りがないのは良い便り」という状態を作り出すことです。これにより、開発チームは新しいAIエージェントやモデルの研究・開発に集中できるようになります。

Step 0: 環境準備とアーキテクチャ設計

それでは、実際に手を動かしていきましょう。まずはVertex AI Pipelinesを動かすための舞台を整えます。

Vertex AI Pipelinesは、内部的にはKubeflow Pipelines (KFP) のマネージドサービスです。通常、Kubeflowを自前で運用する場合、基盤となるKubernetesクラスターのバージョン管理（例：v1.34といった最新版への追随や、廃止APIへの対応など）やノードのメンテナンスに多大な工数を割くことになります。Vertex AI Pipelinesを採用する最大のメリットは、こうしたインフラ管理をGoogle Cloudに完全にオフロードし、開発者が「パイプラインのロジック」そのものに集中できる点にあります。ビジネスへの最短距離を描くためには、マネージドサービスの活用が不可欠です。

必要なGoogle Cloudリソースの確認

このチュートリアルを進めるには、以下のリソースが必要です。

Google Cloud プロジェクト: 課金が有効になっていること。
Cloud Storage (GCS) バケット: パイプラインのアーティファクト（中間データやログ）を保存するステージングエリアとして機能します。
Vertex AI API の有効化: コンソールから「Vertex AI API」を検索し、有効化しておきましょう。
Service Account (SA): パイプライン実行用のアカウント。

特に重要なのが Service Account (SA) です。パイプラインはクラウド上で非同期に実行されるため、実行主体となるSAに適切な権限（IAM）を与えておく必要があります。

推奨される最小権限構成：

Vertex AI User: パイプラインの実行およびメタデータへのアクセス権限
Storage Object Admin: GCSバケットへの読み書き権限（アーティファクトの保存に必須）
BigQuery Data Editor: (BigQueryをデータソースとする場合) データの読み書き権限

ローカル開発環境のセットアップ（KFP SDK v2）

手元のPC（またはVertex AI Workbenchなどのノートブック環境）で、パイプラインを定義・コンパイルするためのSDKをインストールします。現在は KFP SDK v2 が標準となっており、v1系とは書き方が異なるため注意が必要です。

また、Vertex AIのエコシステムは進化が速く、利用可能なモデル（Geminiの最新版など）や機能が頻繁に更新されます。互換性を保つため、SDKは常に最新の安定版を利用することを推奨します。

# 仮想環境を作成してアクティベートすることを推奨します
# Vertex AI SDKとKFP SDKの最新版をインストール
pip install google-cloud-aiplatform kfp --upgrade

※ 特定のバージョンに依存するプロジェクトの場合は、requirements.txt 等でバージョンを固定してください。公式ドキュメントで最新のリリースノートを確認することをお勧めします。

Vertex AI Pipelinesの基本概念（DAG、コンポーネント）

実装前に、アーキテクチャを理解するための2つの用語だけ覚えてください。

コンポーネント (Component): 処理の最小単位です。Pythonの関数一つ一つがコンポーネントになります（例:「データをロードする関数」「モデルを評価する関数」）。これらは実行時にコンテナ化され、独立した環境で動作します。
パイプライン (Pipeline): コンポーネントをつなぎ合わせた一連の処理フローです。DAG（有向非巡回グラフ）として表現され、データの依存関係に基づいて実行順序が自動的に制御されます。

この構造により、例えば「データ前処理」コンポーネントだけを変更したい場合でも、パイプライン全体を作り直すことなく、モジュール単位での修正が可能になります。これは、アジャイルかつスピーディーな改善が求められるMLOpsにおいて非常に重要な特性です。

Step 1: 評価ロジックを「コンポーネント」に封じ込める

Step 0: 環境準備とアーキテクチャ設計 - Section Image

ここがパイプライン構築の核心部分です。普段手元のNotebookで書いているPythonスクリプトを、Vertex AI Pipelinesが理解できる「コンポーネント」という単位に変換します。

Vertex AI Pipelines（KFP v2）では、@component デコレータを使うことで、Python関数をそのままコンテナ化できます。複雑なDockerfileを一から書く必要はありません。このアプローチは、構造化データの予測モデルだけでなく、将来的にGemini等の生成AIモデルを評価するパイプラインへ拡張する際にも共通する重要な基礎となります。

データセット読み込みコンポーネントの実装

まず、評価用データを取得するコンポーネントを作成します。ここではシンプルに、GCS上のCSVファイルパスを受け取り、データセットをロードして次の工程に渡す処理を実装します。

ランタイム環境として、安定性と互換性を考慮し、Python 3.10系を指定します。

from kfp.v2.dsl import component, Output, Dataset

@component(
    base_image="python:3.10",  # 推奨: 最新の安定版ランタイムを指定
    packages_to_install=["pandas", "google-cloud-storage"]
)
def load_evaluation_data(
    source_gcs_path: str,
    output_dataset: Output[Dataset]
):
    import pandas as pd
    # google-cloud-storageは認証済み環境で自動的にクレデンシャルを使用
    
    # GCSからデータを読み込む
    # 実務ではBigQueryからの取得や、Feature Storeとの連携も一般的です
    print(f"Loading data from {source_gcs_path}...")
    df = pd.read_csv(source_gcs_path)
    
    # 必要に応じて最小限の前処理やバリデーションをここに記述
    print(f"Loaded dataset with {len(df)} rows")

    # KFPが管理するパスにCSVとして保存し、次のコンポーネントへ渡す
    # output_dataset.path はシステム側で自動生成されるパスです
    df.to_csv(output_dataset.path, index=False)

専門家の視点:

base_image: 実行コンテナのベースイメージです。Vertex AIのエコシステムは急速に進化しているため、古すぎるバージョン（Python 3.7など）は避け、python:3.10 以降の利用を強く推奨します。
packages_to_install: 実行時に動的にインストールするライブラリです。本番運用では、インストール時間を短縮し再現性を担保するために、これらを含んだカスタムコンテナイメージを事前にビルドしておくのがベストプラクティスです。
Output[Dataset]: データの出力インターフェースです。具体的な保存先（GCSバケット内のアーティファクトリポジトリ）はVertex AI Pipelinesが自動管理するため、コード内でパスをハードコーディングする必要がなくなります。

モデル推論・評価指標算出コンポーネントの実装

次に、ロードされたデータを使ってモデルの推論を行い、精度を計算するコンポーネントです。ここでは例としてscikit-learnモデルを扱いますが、ロジックを差し替えればTensorFlowやPyTorch、あるいはGemini APIを用いたLLM評価（LLM-as-a-Judge）にも応用可能です。

from kfp.v2.dsl import component, Input, Output, Dataset, Metrics, Model

@component(
    base_image="python:3.10",
    packages_to_install=["pandas", "scikit-learn", "joblib", "google-cloud-storage"]
)
def evaluate_model(
    input_dataset: Input[Dataset],
    model_gcs_path: str,
    metrics: Output[Metrics]
) -> float:
    import pandas as pd
    import joblib
    from sklearn.metrics import accuracy_score, f1_score
    from google.cloud import storage
    import tempfile
    
    # 1. データセットの読み込み（前のコンポーネントから渡されたパスを参照）
    df = pd.read_csv(input_dataset.path)
    
    # 想定するデータ構造に合わせて調整
    X = df.drop(columns=["target"])
    y_true = df["target"]

    # 2. モデルのダウンロードとロード
    # Vertex AI Model Registryを使用している場合も、実体はGCS上のアーティファクトです
    client = storage.Client()
    with tempfile.NamedTemporaryFile() as tmp:
        # model_gcs_path = "gs://bucket/path/to/model.pkl" 形式を想定
        # 注: 実装を簡略化しています。実務ではパス解析の堅牢化が必要です
        bucket_name = model_gcs_path.replace("gs://", "").split("/")[0]
        blob_name = "/".join(model_gcs_path.replace("gs://", "").split("/")[1:])
        
        bucket = client.bucket(bucket_name)
        blob = bucket.blob(blob_name)
        blob.download_to_filename(tmp.name)
        
        # モデルロード（セキュリティ上の注意: 信頼できるソースのpklのみ使用すること）
        model = joblib.load(tmp.name)

    # 3. 推論実行
    y_pred = model.predict(X)

    # 4. 評価指標の算出
    acc = accuracy_score(y_true, y_pred)
    f1 = f1_score(y_true, y_pred, average='macro')

    print(f"Evaluation Result - Accuracy: {acc}, F1: {f1}")

    # 5. Vertex AI PipelinesのUIで可視化するためにMetricsに記録
    metrics.log_metric("accuracy", acc)
    metrics.log_metric("f1_score", f1)

    # 条件分岐（デプロイ可否判定など）のために主要な指標を戻り値として返す
    return float(f1)

このコンポーネントは、入力データとモデルを受け取り、評価結果をVertex AIのUI上で追跡可能なMetricsとして記録します。さらに、後続のステップ（例：精度が閾値を超えた場合のみデプロイする条件分岐）で利用するために、F1スコアを関数の戻り値として返しています。

このように評価ロジックをコンポーネント化しておけば、モデルのアルゴリズムが変わっても、あるいは評価対象が生成AIのエージェントに変わっても、パイプラインの全体構造を維持したまま柔軟に対応できます。

Step 2: パイプラインの結合とコンパイル

部品（コンポーネント）ができたら、それらを組み合わせてパイプラインを定義します。

from kfp.v2 import dsl
from kfp.v2 import compiler

@dsl.pipeline(
    name="evaluation-pipeline-v1",
    description="定期的なモデル精度評価パイプライン"
)
def evaluation_pipeline(
    eval_data_path: str,
    model_path: str,
    threshold: float = 0.8
):
    # コンポーネント1: データロード
    load_task = load_evaluation_data(
        source_gcs_path=eval_data_path
    )

    # コンポーネント2: 評価実行
    # load_task.outputs["output_dataset"] で前のタスクの出力を受け取る
    eval_task = evaluate_model(
        input_dataset=load_task.outputs["output_dataset"],
        model_gcs_path=model_path
    )

    # コンパイル実行
compiler.Compiler().compile(
    pipeline_func=evaluation_pipeline,
    package_path="evaluation_pipeline.json"
)

このコードを実行すると、evaluation_pipeline.json というファイルが生成されます。これがパイプラインの設計図です。Pythonの関数呼び出しのように見えますが、実際にはload_taskとeval_taskの間の依存関係（DAG）を定義しているだけです。

依存関係の魔法:
eval_taskの引数にload_task.outputs[...]を渡すことで、KFPは「load_taskが終わらないとeval_taskは開始できない」と自動的に理解し、実行順序を制御します。

Step 3: 定期実行スケジュールの設定と通知連携

Step 2: パイプラインの結合とコンパイル - Section Image

設計図であるパイプライン定義が完成したら、いよいよVertex AI上で稼働させます。しかし、単にパイプラインを回すだけでは運用とは言えません。「精度が許容範囲を下回った場合のみ、即座に人間へアラートを飛ばす」という条件付きロジックを組み込むことで、初めて自律的な監視システムとして機能します。

条件分岐の実装：dsl.Condition

先ほど定義したパイプラインに、Kubeflow Pipelines SDK（KFP）の制御フロー機能である dsl.Condition を追加します。これにより、評価スコアに応じた動的な分岐が可能になります。

from kfp import dsl

@dsl.pipeline(
    name="evaluation-pipeline-with-alert",
    description="精度劣化時に通知を行うパイプライン"
)
def evaluation_pipeline(
    eval_data_path: str,
    model_path: str,
    threshold: float = 0.8
):
    # 1. データロード
    load_task = load_evaluation_data(source_gcs_path=eval_data_path)
    
    # 2. モデル評価
    eval_task = evaluate_model(
        input_dataset=load_task.outputs["output_dataset"],
        model_gcs_path=model_path
    )

    # 3. 条件分岐: スコアが閾値を下回った場合のみ実行
    # Pythonのif文ではなく、dsl.Conditionを使用することでDAG上の分岐として定義されます
    with dsl.Condition(
        eval_task.output < threshold,
        name="accuracy-drop-detected"
    ):
        # ここで通知コンポーネントを呼び出します
        # 実践的には、Slack通知やPub/Subへのメッセージ発行を行うコンポーネントを配置します
        send_slack_notification(
            message=f"Warning: Model accuracy dropped below {threshold}. Current: {eval_task.output}",
            webhook_url="YOUR_SLACK_WEBHOOK_SECRET" # Secret Managerの使用を推奨
        )

専門家の視点：
ここで重要なのは、Python標準の if 文ではなく dsl.Condition を使用する点です。if 文はパイプラインのコンパイル時に評価されてしまいますが、dsl.Condition はパイプラインの実行時に評価結果に基づいて動的にパスを決定します。

また、通知内容の生成において、最新のVertex AIでは Geminiモデル を活用するケースも増えています。単に「精度低下」と通知するだけでなく、評価レポートのJSONをGeminiに渡し、「どのセグメントで予測を外しているか」の要約を生成させてからSlackに投げる、といった高度な運用も可能です。

定期実行（Schedule）の設定

パイプラインをデプロイし、定期的に実行されるようスケジュールを設定します。Vertex AI SDK for Pythonを使用すれば、Cron形式で柔軟にスケジュールを定義できます。

from google.cloud import aiplatform

# プロジェクトとロケーションの初期化
aiplatform.init(project="your-project-id", location="us-central1")

# パイプラインジョブの定義（まだ実行はされません）
job = aiplatform.PipelineJob(
    display_name="weekly-model-evaluation",
    template_path="evaluation_pipeline.json", # コンパイル済みのYAML/JSONパス
    pipeline_root="gs://your-bucket/pipeline_root",
    parameter_values={
        "eval_data_path": "gs://your-bucket/data/eval.csv",
        "model_path": "gs://your-bucket/models/model_v1.pkl",
        "threshold": 0.8
    },
    # 重要: 評価パイプラインではキャッシュを無効化し、常に最新データ/コードで実行されるようにする
    enable_caching=False 
)

# スケジュール実行の作成 (Cron形式)
# 例: 毎週月曜日の朝9時(JST)に実行
# タイムゾーン指定がない場合、UTC基準となるため注意が必要です
pipeline_schedule = aiplatform.PipelineJobSchedule(
    pipeline_job=job,
    display_name="weekly-eval-schedule"
)

pipeline_schedule.create(
    cron="0 9 * * 1",  # 分 時 日 月 曜日
    max_concurrent_run_count=1, # 重複実行を防ぐ
    max_run_count=None # 無期限に実行
)

print(f"Schedule created: {pipeline_schedule.resource_name}")

これで設定は完了です。毎週月曜日の朝、あなたがコーヒーを飲んでいる間に、Vertex AIが自動的にコンピュートリソースを立ち上げ、最新データでモデルを評価し、問題がある場合のみSlackに通知を届けます。

運用のヒント:
Vertex AIのエコシステムは急速に進化しています。現在、Vertex AI Pipelinesはより複雑なワークフローの中核として機能しており、評価結果をトリガーに Vertex AI Agents を起動して再学習ジョブを自律的に構成するといった高度なオーケストレーションも視野に入ります。まずはこの基本的な「定期評価＋異常通知」のループを確立することが、MLOpsの第一歩となります。

発展：MLOps基盤への拡張に向けて

今回構築したパイプラインは、本格的なMLOpsへの入り口に過ぎません。この「評価の自動化」が安定稼働し始めたら、次は以下のような拡張を行い、システム全体を進化させていくことが重要です。

近年では、従来の機械学習モデルに加え、LLM（大規模言語モデル）を含むシステムの運用管理（LLMOps）への関心も高まっていますが、パイプラインによる自動化という根底の考え方は共通しています。

再学習パイプラインとの連携

評価スコアが閾値を下回ったことを検知した場合、単に通知を送るだけでなく、自動的に再学習パイプライン（Continuous Training）をトリガーする仕組みを構築できます。

Vertex AI Pipelinesでは、条件分岐コンポーネントを使用し、評価結果に応じて学習ジョブ（TrainingPipelineなど）を実行するフローを定義可能です。これにより、モデルの精度劣化（ドリフト）に対して、人手を介さずに即座に対応する自律的なシステムが実現します。

Vertex AI Model Registryとの統合

本ハンズオンではGCS上のモデルファイルを直接扱いましたが、本番運用、特にチーム開発においては Vertex AI Model Registry の活用が推奨されます。

Model Registryを利用することで、モデルのバージョン管理、メタデータ（精度指標や学習パラメータ）の追跡、デプロイ承認フローを一元管理できます。パイプラインからは「最新の承認済み（Approved）モデル」や「特定のエイリアスが付与されたモデル」を動的に取得するよう実装することで、ガバナンスの効いた運用が可能になります。

CI/CDによるパイプライン管理

パイプラインの定義コード自体もアプリケーションコードと同様にGitで管理し、変更がプッシュされたら自動的にコンパイルとテストが行われるCI/CD（継続的インテグレーション/継続的デリバリー）環境を構築しましょう。

Google Cloudであれば、Cloud Buildと連携させることで、「パイプライン定義の更新」から「Vertex AIへのデプロイ（テンプレートのアップロード）」までを完全に自動化できます。これはGitOpsの実践として非常に効果的です。

まとめ

job.run() - Section Image 3

Vertex AI Pipelinesを用いたモデル評価の自動化は、決してハードルの高いものではありません。Pythonとクラウドの基本的な知識があれば、十分に実装可能です。

重要なのは、「手動でも運用できるから」という現状維持バイアスを捨てることです。手動運用は、運用担当者の時間を奪うだけでなく、操作ミスや属人化といったビジネス上のリスク要因にもなりえます。

評価ロジックをコンポーネント化する
パイプラインとしてつなぎ合わせる
条件分岐で異常検知時のアクションを定義する
スケジュール実行を設定する

この4ステップを実践することで、「守りの運用」から解放され、より創造的な「攻めの開発」や、AIエージェント開発などの新しい技術課題に注力できるようになります。まずは小さなプロトタイプから、この自動化のアプローチを試してみてください。いかがでしょうか？皆さんの現場でも、今日から始められるはずです。

Vertex AI PipelinesでML評価バッチを自動化する：手動運用をゼロにするPython実装ハンズオン - Conclusion Image

コメントは1週間で消えます

コメントを読み込み中...