はじめに はじめまして。 KINTO テクノロジーズで KINTO Unlimited Android アプリを開発している JR.Liang です。 本記事では、KINTO Unlimited アプリにて提供する「これなにガイド」スキャン機能の AR エフェクトについて、Android における技術的な検証を紹介します。 特に MediaPipe のソリューションを用いて幅広い Android デバイスで AR エフェクトを実現した実装にフォーカスします。 これなにガイドとは 「これなにガイド」は AR（拡張現実）を活用して、車内スイッチの用途や使い方をテキストと動画で案内する機能です。紹介動画をご覧ください。 https://youtube.com/watch?v=E8zfNzuHr7g&embeds_referring_euri=https%3A%2F%2Fcorp.kinto-jp.com%2F&source_ve_path=MjM4NTE 上記の紹介動画は iOS アプリでの動作を示しています。スイッチ上に表示された黄色の丸 🟡 が、AR 技術で実現した仮想コンテンツです。 機能全体の仕組みは以下の流れです。本記事では 3 番目（描画）に関する内容を扱います。 1. アプリのカメラを起動、カメラ画像を取得 2. 機械学習における物体認識を用いて、車内のスイッチを検出 3. 検出した座標を元に、ボタンとテキストをフレーム上に描画 4. ボタンをタップして、当該スイッチのテキストと動画を表示 Android AR 技術検証の経緯 当初の Android 版「これなにガイド」のスキャン機能では、Canvas を利用して毎フレーム検出される座標に描画する実装でした。そのため検出の時間差により、スマホ（カメラ）を動かすと描画のズレが生じていました。 2D Canvas 幸い、MediaPipe のソリューションである Instant Motion Tracking モジュールで 素早くかつ安定した AR エフェクトを実現できることがわかり、Android への導入を検証しました。 3D OpenGL MediaPipe Instant Motion Tracking MediaPipe は Google が開発したオープンソースの ML フレームワークで、顔検出・手のトラッキング・姿勢推定などリアルタイム映像処理のソリューションを提供します。 その中の Instant Motion Tracking は、現実世界のシーン上に 3D 仮想コンテンツをリアルタイムで正確に配置できる AR トラッキング機能です。初期化や厳密なキャリブレーションが不要で、静止面や動いている面の上にコンテンツを置くことが可能です。 @ card Android + MediaPipe AR アーキテクチャ graph TB A(Android CameraX) --> |Camera Frame| B(Instant Motion Tracking) B --> |Camera Image| C(TensorFlow Object Detection) C --> |Detections Information| B(Instant Motion Tracking) B --> |Output Stream| D(Android Surface Rendering) CameraX で取得したフレームを Instant Motion Tracking に渡し、TensorFlow Lite で物体検出した情報を元に AR コンテンツを描画・追従させるパイプラインです。 MediaPipe ライブラリの作成 MediaPipe では Bazel を使用してパッケージをビルドします。Android に適合する AAR として書き出してアプリに組み込みます。 https://chuoling.github.io/mediapipe/getting_started/android_archive_library.html AAR をビルドする BUILD ファイルを作成し、 instant_motion_tracking を基盤とした定義を記述します。 load("//mediapipe/java/com/google/mediapipe:mediapipe_aar.bzl", "mediapipe_aar") mediapipe_aar( name = "mediapipe_ar", calculators = ["//mediapipe/graphs/instant_motion_tracking:instant_motion_tracking_deps"] ) MediaPipe は C++ が中核のため、C++ ランタイムである libc++_shared.so を AAR に同梱する必要があります。 https://github.com/google-ai-edge/mediapipe/blob/v0.10.32/third_party/BUILD#L399-L403 また Instant Motion Tracking では画像処理ライブラリ OpenCV を利用し、AR トラッキングを行います。 https://github.com/google-ai-edge/mediapipe/blob/v0.10.32/WORKSPACE#L649-L655 上記サードパーティのライブラリを含めて、以下のコマンドで AAR をビルドします。 bazel build -c opt --strip=ALWAYS \ --host_crosstool_top=@bazel_tools//tools/cpp:toolchain \ --fat_apk_cpu=arm64-v8a \ --linkopt=-Wl,-z,max-page-size=16384 \ //path/to/the/aar/build/mediapipe_ar:mediapipe_ar.aar 市場に流通している Android デバイスは主に arm64-v8a アーキテクチャのため、AAR のサイズを抑える目的で fat_apk_cpu=arm64-v8a にします。 C++ ライブラリの 16KB page-size に対応するため、 max-page-size=16384 を追加します。 また AAR を利用するにはグラフ構造を定義するファイル（ binarypb ）が必要です。 bazel build -c opt mediapipe/graphs/instant_motion_tracking:instant_motion_tracking.binarypb Instant Motion Tracking の導入 AAR をアプリに組み込んで、Android 側の実装を解説していきます。 下記は AAR に組み込んだ instant_motion_tracking の全体構造です。 instant_motion_tracking.pbtxt の構成 グラフ定義ファイル instant_motion_tracking.pbtxt は、Calculator（処理ノード）・入出力ストリーム・サイドパケットの 3 要素で構成されます。 Calculator 各 Calculator がパイプライン上でどの処理を担うかを示します。 Calculator 役割 ImageTransformationCalculator カメラフレームを 320×320（FIT）にリサイズ。物体検出モデルの入力サイズに合わせる GpuBufferToImageFrameCalculator GPU テクスチャを CPU の ImageFrame に変換。TensorFlow Lite 推論に使用 StickerManagerCalculator Sticker Proto をパースし、初期アンカーの座標・回転・スケール・レンダリング種別に分解 RegionTrackingSubgraph ボックストラッキングでアンカー位置を追従。内部に TrackedAnchorManagerCalculator （アンカー管理）と BoxTrackingSubgraphGpu （GPU トラッキング）を持つ MatricesManagerCalculator トラッキング結果・回転・スケール・FOV・アスペクト比から OpenGL 用 4×4 モデル行列を生成 GlAnimationOverlayCalculator モデル行列とテクスチャを用いて、元のカメラフレーム上に AR コンテンツを OpenGL で描画し output_video として出力 input_stream / output_stream input_stream はフレームごとに Android 側から送信するデータ、 output_stream はグラフの処理結果です。 ストリーム名 C++ 型 方向 用途 input_video GpuBuffer Input カメラフレーム sticker_proto_string String(Serialized Proto) Input ステッカーの座標・スケール等（Sticker Proto） sticker_sentinels vector Input 座標をリセットするステッカー ID の配列 gif_textures vector Input AR コンテンツの Bitmap テクスチャ配列 gif_aspect_ratios vector Input 各テクスチャのアスペクト比 output_video GpuBuffer Output AR 描画済みフレーム input_side_packet input_side_packet は初期化時に一度だけ渡す定数で、グラフ実行中は変化しません。 パケット名 用途 vertical_fov_radians カメラの垂直 FOV（ラジアン） aspect_ratio カメラのアスペクト比 width / height カメラ解像度 gif_texture デフォルトテクスチャ（1x1 プレースホルダ） gif_asset_name AR テクスチャ描画用のポリゴンメッシュ（ .obj ）ファイル名 Android への導入に当たって、公式サンプルのコードを参考にします。 https://github.com/google-ai-edge/mediapipe/tree/master/mediapipe/examples/android/src/java/com/google/mediapipe/apps/instantmotiontracking 1. 初期化 MediaPipe を使用する前に、ネイティブライブラリの読み込みとアセットマネージャーの初期化が必要です。 companion object { init { System.loadLibrary("mediapipe_jni") System.loadLibrary("opencv_java4") } } // onCreate 相当の処理 AndroidAssetUtil.initializeNativeAssetManager(context) mediapipe_jni : MediaPipe のコア処理を行う JNI ライブラリ opencv_java4 : AR トラッキングに使用する OpenCV ライブラリ initializeNativeAssetManager : ネイティブコードからアセット（binarypb 等）にアクセスするために必要 2. カメラを起動する 公式サンプルを参考に、以下の順序でパイプラインを構築します。 データフロー： CameraX → ExternalTextureConverter → FrameProcessor → SurfaceView 2.1 EGL 環境と FrameProcessor の初期化 val eglManager = EglManager(null) val frameProcessor = FrameProcessor( context, eglManager.nativeContext, "instant_motion_tracking.binarypb", "input_video", "output_video" ).apply { videoSurfaceOutput.setFlipY(true) setInputSidePackets( mapOf( "gif_asset_name" to packetCreator.createString("gif.obj.uuu"), "vertical_fov_radians" to packetCreator.createFloat32(fovRadians), "aspect_ratio" to packetCreator.createFloat32(resolution.width.toFloat() / resolution.height.toFloat()), "width" to packetCreator.createInt32(resolution.width), "height" to packetCreator.createInt32(resolution.height), "gif_texture" to packetCreator.createRgbaImageFrame(createBitmap(1, 1)) ) ) } EglManager : OpenGL ES の EGL コンテキストを作成・管理。MediaPipe のグラフ内 GPU Calculator（ GlAnimationOverlayCalculator 等）が OpenGL で描画するために必要 FrameProcessor : EGL コンテキストを受け取り、グラフの読み込み・入出力ストリームの管理・フレームごとのグラフ実行を行う instant_motion_tracking.binarypb : .pbtxt を Bazel でコンパイルしたグラフ定義バイナリ input_video : MediaPipe グラフへカメラフレームを入力 output_video : グラフで処理（AR 描画など）された映像を出力 videoSurfaceOutput.setFlipY(true) : OpenGL とカメラの Y 軸方向が逆のため、出力映像を上下反転して正しい向きにする setInputSidePackets : グラフの input_side_packet に対応する定数をまとめて設定。カメラの FOV・アスペクト比・解像度など、グラフ実行中に変化しない値を初期化時に一度だけ渡す gif_asset_name は AR テクスチャを描画するための ポリゴンメッシュ（頂点データ） 、ここでは公式サンプルの gif.obj.uuu を利用 2.2 カメラ映像の変換パイプライン構築 val externalTextureConverter = ExternalTextureConverter(eglManager.context, 2).apply { setFlipY(true) setConsumer(frameProcessor) setDestinationSize(resolution.width, resolution.height) } val cameraHelper = object : CameraXPreviewHelper() { override fun getCameraCharacteristics(context: Context?, lensFacing: Int?) = cameraCharacteristics }.apply { setOnCameraStartedListener(onCameraStartedListener) startCamera( context, lifecycleOwner, CameraHelper.CameraFacing.BACK, externalTextureConverter.surfaceTexture, Size(resolution.height, resolution.width) ) } ExternalTextureConverter : カメラの GL_EXTERNAL_OES テクスチャを MediaPipe が処理できる標準テクスチャに変換 setFlipY(true) : カメラ映像の上下反転を補正 setDestinationSize(resolution.width, resolution.height) : パイプラインの処理サイズはポートレート座標（例: 960×1280 ）で指定 CameraXPreviewHelper : CameraX でバックカメラを起動し、Converter の SurfaceTexture に出力 startCamera(targetSize = Size(resolution.height, resolution.width)) : CameraX はセンサー座標（ランドスケープ）を期待するため、width と height を入れ替えて渡す 公式サンプルでは CameraXPreviewHelper をそのまま使用し、内部で CameraManager からカメラ特性を取得します。 https://github.com/google-ai-edge/mediapipe/blob/v0.10.32/mediapipe/java/com/google/mediapipe/components/CameraXPreviewHelper.java#L558-L560 本実装では getCameraCharacteristics をオーバーライドし、事前に取得済みの CameraCharacteristics を直接渡します。これにより FOV やアスペクト比の算出に使うカメラ情報を、アプリ側で一元管理できます。 2.3 出力先SurfaceViewの設定 SurfaceView(context).apply { holder.addCallback(object : SurfaceHolder.Callback { override fun surfaceCreated(holder: SurfaceHolder) { frameProcessor.videoSurfaceOutput.setSurface(holder.surface) } override fun surfaceChanged(holder: SurfaceHolder, format: Int, width: Int, height: Int) { val displaySize = cameraHelper.computeDisplaySizeFromViewSize(Size(width, height)) val (displayWidth, displayHeight) = if (cameraHelper.isCameraRotated) { displaySize.height to displaySize.width } else { displaySize.width to displaySize.height } externalTextureConverter.setDestinationSize(displayWidth, displayHeight) } override fun surfaceDestroyed(holder: SurfaceHolder) { frameProcessor.videoSurfaceOutput.setSurface(null) } }) } SurfaceHolder.Callback : SurfaceView のライフサイクルに応じて FrameProcessor の出力先を管理 surfaceCreated : FrameProcessor の出力先として Surface を設定 surfaceChanged : 画面回転・サイズ変更時に出力解像度を調整 surfaceDestroyed : リソース解放 3. 検出座標をグラフに渡す 物体検出（TensorFlow Lite 等）で得られた座標を MediaPipe グラフに渡し、AR コンテンツを配置します。 3.1 グラフから変換済み画像を取得 MediaPipe グラフ内で ImageTransformationCalculator と GpuBufferToImageFrameCalculator によって変換された画像を addPacketCallback で受け取り、物体検出に使用します。 frameProcessor.addPacketCallback("transformed_input_video_cpu") { packet -> packet ?: return@addPacketCallback // 変換済み画像を物体検出（TensorFlow Lite）に渡す val bitmap = PacketGetter.getBitmapFromRgba(packet) objectDetector.detect(bitmap) { detections -> // 検出結果を処理 } } transformed_input_video_cpu : 変換後の画像を出力するストリーム名 3.2 座標の正規化 物体検出結果のピクセル座標を、MediaPipe が期待する正規化座標に変換します。 // ピクセル座標 → 正規化座標 (0.0〜1.0) val normalizedX = pixelX / imageWidth.toFloat() val normalizedY = pixelY / imageHeight.toFloat() 3.3 Sticker Proto の構造 Instant Motion Tracking では、AR オブジェクトの位置情報を Protocol Buffers 形式で定義します。 message Sticker { int32 id = 1; // ユニークID float x = 2; // 正規化X座標 (0.0〜1.0) float y = 3; // 正規化Y座標 (0.0〜1.0) float rotation = 4; // 回転角度 float scale = 5; // スケール int32 render_id = 6; // レンダリングID } message StickerRoll { repeated Sticker sticker = 1; } 3.4 フレームごとにパケットを送信 setOnWillAddFrameListener を使用して、各フレーム処理前に検出座標をグラフへ送信します。 frameProcessor.setOnWillAddFrameListener { timestamp -> with(frameProcessor.graph) { // 検出された物体の座標情報をパケットとして送信 val stickerRoll = StickerRoll.newBuilder() .addAllSticker(detectedObjects.map { detection -> Sticker.newBuilder() .setId(detection.id) .setX(detection.normalizedX) // 0.0〜1.0 .setY(detection.normalizedY) // 0.0〜1.0 .setScale(detection.scale) .build() }) .build() val stickersPacket = packetCreator.createSerializedProto(stickerRoll) addPacketToInputStream("sticker_proto_string", stickersPacket, timestamp) } } FrameProcessor.setOnWillAddFrameListener : 各フレームがグラフに送られる直前に呼ばれるコールバック FrameProcessor.graph.addPacketToInputStream : 入力ストリームにパケットを追加 sticker_proto_string : グラフ定義で指定された入力ストリーム名 4. テクスチャ（Bitmap）の描画と送信 位置情報と同時に、AR コンテンツとして描画する Bitmap テクスチャもグラフに渡します。 4.1 Bitmap テクスチャの生成 検出された各スイッチに対して、丸アイコンとラベルテキストを含む Bitmap を生成します。 val bitmap = createBitmap(width.toInt(), height.toInt()).apply { with(Canvas(this)) { concat(Matrix().apply { preScale(-1.0f, 1.0f, width / 2f, height / 2f) // X軸を反転して描画 }) drawCircle(circleX, circleY, CIRCLE_RADIUS, circlePaint) drawRect(rectLeft, rectTop, rectRight, rectBottom, backgroundPaint) } } Matrix().preScale(-1.0f, 1.0f) で Bitmap を左右反転しています。以下の IMU 行列に合わせるためです。 float imu_matrix[9] = { -1.0f, 0.0f, 0.0f, // X軸 → 反転(-X) 0.0f, 0.0f, 1.0f, // Y軸 → Z軸へ 0.0f, 1.0f, 0.0f // Z軸 → Y軸へ }; この行列は OpenGL モデル行列（4x4）の回転成分として使われ、Y/Z 軸の入れ替えと X 軸反転でテクスチャをカメラ平面に平行に固定します。 本来はデバイスの IMU センサーから回転行列を受け取り、端末の傾きに追従させます。 https://github.com/google-ai-edge/mediapipe/blob/0.10.32/mediapipe/examples/android/src/java/com/google/mediapipe/apps/instantmotiontracking/MainActivity.java#L218-L220 本実装では固定値にすることで 常にカメラ正面を向く （ビルボード効果）ようにし、 (0,0) の -1.0 による X 軸反転を Bitmap 側の preScale(-1.0f, 1.0f) で打ち消します。 4.2 テクスチャの送信 // テクスチャ画像（Bitmap配列） val texturesPacket = packetCreator.createRgbaImageFrameVector( renderStickers.map { it.bitmap }.toTypedArray() ) addPacketToInputStream("gif_textures", texturesPacket, timestamp) // アスペクト比（テクスチャの縦横比） val aspectRatiosPacket = packetCreator.createFloat32Vector( renderStickers.map { it.aspectRatio }.toFloatArray() ) addPacketToInputStream("gif_aspect_ratios", aspectRatiosPacket, timestamp) PacketCreator.createRgbaImageFrameVector : 複数の Bitmap を RGBA 形式のパケットに変換 gif_textures : テクスチャ画像の入力ストリーム gif_aspect_ratios : 各テクスチャのアスペクト比（正しいスケーリングに必要） 公式サンプルでは createRgbaImageFrame を使用して 単一のテクスチャ をグラフに渡します。 https://github.com/google-ai-edge/mediapipe/blob/0.10.32/mediapipe/examples/android/src/java/com/google/mediapipe/apps/instantmotiontracking/MainActivity.java#L608-L610 本実装では、複数の検出オブジェクトに対応するため createRgbaImageFrameVector で 複数テクスチャを同時に送信 し、 gif_aspect_ratios も createFloat32Vector で 各テクスチャに対応するアスペクト比の配列 を渡すよう拡張します。これにより、検出された各スイッチに異なるラベル（テキスト付きBitmap）を正しい縦横比で表示できます。 ここまでで AR コンテンツをカメラ上に表示できました。 5. 座標の更新 トラッキング中のステッカー座標を更新するには、新しい座標を持つ sticker_proto_string と、リセット対象の ID を含む sticker_sentinels を同一 timestamp で送信します。 TrackedAnchorManagerCalculator が該当 ID のトラッキングボックスを破棄し、新しい座標でトラッキングを再開します。 // 更新した座標で Sticker Proto を再構築 val stickersPacket = packetCreator.createSerializedProto(stickerRoll) addPacketToInputStream("sticker_proto_string", stickersPacket, timestamp) // リセット対象のステッカー ID を送信 val stickerSentinels = packetCreator.createInt32Vector(updateIds) addPacketToInputStream("sticker_sentinels", stickerSentinels, timestamp) 公式サンプルでは sticker_sentinel で 単一のステッカー ID を送信します。 https://github.com/google-ai-edge/mediapipe/blob/0.10.32/mediapipe/examples/android/src/java/com/google/mediapipe/apps/instantmotiontracking/MainActivity.java#L342-L344 本実装では sticker_sentinels として createInt32Vector で 複数のステッカー ID を配列 で渡すよう拡張し、物体検出で座標が更新された複数のステッカーを同時にリセットできるようにします。 最後に 以上が MediaPipe Instant Motion Tracking を用いた技術的な実装解説でした。決して容易に導入できる手法ではありませんが、本機能の要件に対して Android に最も適した解決策だと考えています。 以前に ARCore の検証も行いましたが、ARCore は SLAM 技術による事前の 3D マッピングに時間を要し、 素早くかつ安定した AR エフェクトの実現には適さなかったため、検証を断念しました。 両フレームワークの違いを以下にまとめます。AR 技術の検討で参考になれば幸いです。 項目 Instant Motion Tracking ARCore 仕組み 2D ボックストラッキング + OpenGL 描画 環境マッピング + 平面検出（SLAM） デバイス要件 OpenGL ES 対応であれば動作 ARCore 対応デバイスのみ（Google 認定必須） 安定性 検出座標に依存するため補正が必要 空間認識が高精度で安定 導入コスト Bazel ビルド・C++ Calculator のカスタマイズが必要 SDK 導入のみで比較的容易 オープンソース あり（Apache 2.0） なし（プロプライエタリ） カスタマイズ性 Calculator の追加・変更で柔軟に拡張可能 SDK の API 範囲内に限定 パフォーマンス 軽量（2D トラッキングベースのため CPU/GPU 負荷が低い） 高負荷（環境の 3D 空間マッピングを常時実行） 学習コスト 高い（Bazel・C++・OpenGL・Protocol Buffers の知識が必要） 低い（Android SDK の知見で導入可能）

G-gen の片岩です。当記事では Vertex AI Custom Training において カスタムコンテナ を使用し、標準では提供されていない LightGBM モデルの学習から 寄与度（SHAP）の出力 まで実行する方法を紹介します。 はじめに ビルド済みコンテナとカスタムコンテナの使い分け カスタムコンテナの利点 構成図 初期設定 データの準備と分割 カスタムコンテナの準備 ディレクトリとリポジトリの準備 学習スクリプトの作成 Dockerfile の作成 コンテナのビルドとプッシュ 学習ジョブの実行 推論と評価指標の確認 分析レポートの解釈 はじめに ビルド済みコンテナとカスタムコンテナの使い分け Vertex AI Custom Training には学習ジョブを実行するためのコンテナイメージとして、大きく 2 つの選択肢があります。 コンテナの種類 特徴 向いているケース ビルド済みコンテナ Google Cloud が用意したイメージ XGBoost や TensorFlow など、標準的なフレームワークをすぐに使いたい時 カスタムコンテナ 自分で Dockerfile を書いて作成するイメージ LightGBM など未提供のライブラリを使いたい時や、独自の処理を組み込みたい時 カスタムコンテナの利点 ビルド済みコンテナでもジョブ実行時の引数に requirements=["lightgbm", "shap"] のように指定することでライブラリを追加できます。ビルド済みコンテナについては以下の記事を参照してください。 blog.g-gen.co.jp しかし実務の本番運用において、実行時にライブラリを動的にインストールすることは、以下のデメリットがあります。 1 点目は、 環境の再現性が低下する ことです。 ジョブを実行するたびにインターネットから最新のパッケージを取得するため、依存ライブラリのバージョンが上がったために突然ジョブが落ちたり、学習結果が変わってしまうといった、本番運用で避けたいリスクを招きます。 2 点目は、 実行のたびにオーバーヘッドが発生する ことです。 毎回ライブラリをダウンロードしてインストールする処理が走るため、余計な待ち時間が発生します。 カスタムコンテナを利用することにより、上記のデメリットを回避できます。 参考 : カスタム コンテナの概要 構成図 当記事で紹介する手順に関する構成図は以下のとおりです。環境構築の負荷を軽減するため、ソースコードの作成や Python 実行環境に Colab Enterprise を使用します。 初期設定 はじめにライブラリのインストールと環境変数の設定を行います。今回は可視化や解釈のためのライブラリ（ seaborn 、 shap ）も追加します。 # 必要なライブラリのインストール !pip install google-cloud-aiplatform lightgbm shap scikit-learn pandas seaborn matplotlib -q # プロジェクトとリージョンの設定 # ※ ご自身の環境に合わせて書き換えてください PROJECT_ID = "your-project-id" LOCATION = "asia-northeast1" # バケットとフォルダの定義 ROOT_BUCKET = "gs://your-bucket" EXPERIMENT_NAME = "diamonds-lgbm-v1" WORK_DIR = f "{ROOT_BUCKET}/{EXPERIMENT_NAME}" # Vertex AI SDK の初期化 from google.cloud import aiplatform aiplatform.init(project=PROJECT_ID, location=LOCATION, staging_bucket=WORK_DIR) # バケットが存在しない場合のみ作成 !gsutil mb -l {LOCATION} {ROOT_BUCKET} データの準備と分割 データは機械学習デモで使用されるダイヤモンドの価格データを使用します。このデータはカラットなどの数値データや、カットや色といったカテゴリ変数を含みます。 学習データと推論データに分割して Cloud Storage に保存します。 import seaborn as sns from sklearn.model_selection import train_test_split import pandas as pd # データのロード (~54,000行) df = sns.load_dataset( 'diamonds' ) # 文字列カラムを 'category' 型に変換 cat_cols = [ 'cut' , 'color' , 'clarity' ] for col in cat_cols: df[col] = df[col].astype( 'category' ) # 学習データと推論データに 90:10 の割合で分割 train_full_df, test_df = train_test_split(df, test_size= 0.1 , random_state= 42 ) # データの保存 train_filename = "train.csv" train_full_df.to_csv(train_filename, index= False ) test_filename = "test.csv" test_df.to_csv(test_filename, index= False ) # GCS へアップロード !gsutil cp {train_filename} {WORK_DIR}/data/{train_filename} !gsutil cp {test_filename} {WORK_DIR}/data/{test_filename} print (f "学習データ: {WORK_DIR}/data/{train_filename}" ) print (f "推論データ: {WORK_DIR}/data/{test_filename}" ) カスタムコンテナの準備 ディレクトリとリポジトリの準備 Colab Enterprise 上に作業ディレクトリを用意し、Google Cloud 上に完成したコンテナの保存先となる Artifact Registry のリポジトリを作成します。 # 作業用ディレクトリの作成 !mkdir -p custom_container # Artifact Registry にリポジトリを作成 (初回のみ) !gcloud artifacts repositories create custom-training-repo \ --repository- format =docker \ --location={LOCATION} \ --description= "Custom Training Repository" || true 学習スクリプトの作成 コンテナ内で実行される task.py を作成します。 今回はモデルの学習だけでなく、過学習を確認するための学習曲線と、予測の根拠を説明するための寄与度の画像を生成し、モデルと一緒に Cloud Storage へアップロードする処理を組み込みます。 %%writefile custom_container/task.py import argparse import os import pandas as pd import lightgbm as lgb import shap import matplotlib.pyplot as plt from sklearn.model_selection import train_test_split from google.cloud import storage from urllib.parse import urlparse import warnings warnings.filterwarnings( 'ignore' ) parser = argparse.ArgumentParser() parser.add_argument( '--train-data-uri' , dest= 'train_data_uri' , type = str , required= True ) args = parser.parse_args() # --- GCS ダウンロード / アップロード用の関数 --- def download_from_gcs (gcs_uri, local_file): parsed_url = urlparse(gcs_uri) client = storage.Client() bucket = client.bucket(parsed_url.netloc) blob = bucket.blob(parsed_url.path.lstrip( "/" )) blob.download_to_filename(local_file) def upload_to_gcs (local_file, gcs_dir): parsed_url = urlparse(gcs_dir) client = storage.Client() bucket = client.bucket(parsed_url.netloc) blob_path = f "{parsed_url.path.lstrip('/').rstrip('/')}/{local_file}" bucket.blob(blob_path).upload_from_filename(local_file) # --- 1. データの準備 --- print (f "Downloading data from {args.train_data_uri}..." , flush= True ) local_train_file = "train.csv" download_from_gcs(args.train_data_uri, local_train_file) df = pd.read_csv(local_train_file) cat_cols = [ 'cut' , 'color' , 'clarity' ] for col in cat_cols: df[col] = df[col].astype( 'category' ) X = df.drop(columns=[ "price" ]) y = df[ "price" ] # スクリプト内で学習用と検証用に分割 (データリーク防止) X_train, X_val, y_train, y_val = train_test_split(X, y, test_size= 0.1 , random_state= 42 ) # --- 2. モデルの学習 --- print ( "Training LightGBM model..." , flush= True ) model = lgb.LGBMRegressor(n_estimators= 100 , random_state= 42 ) # 学習過程を記録するために eval_set を渡す model.fit( X_train, y_train, eval_set=[(X_train, y_train), (X_val, y_val)], eval_names=[ 'train' , 'valid' ] ) # --- 3. 分析画像の生成と保存 --- # ① 学習曲線の描画 lgb.plot_metric(model, metric= 'l2' ) plt.title( 'Learning Curve (MSE)' ) plt.tight_layout() plt.savefig( "learning_curve.png" ) plt.close() # ② SHAP値（寄与度）の描画 print ( "Calculating SHAP values..." , flush= True ) explainer = shap.TreeExplainer(model) shap_values = explainer(X_val.sample( min ( 1000 , len (X_val)), random_state= 42 )) plt.figure() shap.plots.beeswarm(shap_values, show= False ) plt.title( "SHAP Feature Importance" ) plt.tight_layout() plt.savefig( "shap_importance.png" ) plt.close() # --- 4. 成果物のアップロード --- aip_model_dir = os.getenv( "AIP_MODEL_DIR" ) if aip_model_dir: print (f "Uploading artifacts to: {aip_model_dir}" , flush= True ) model.booster_.save_model( "model.txt" ) upload_to_gcs( "model.txt" , aip_model_dir) upload_to_gcs( "learning_curve.png" , aip_model_dir) upload_to_gcs( "shap_importance.png" , aip_model_dir) print ( "Upload completed." , flush= True ) Dockerfile の作成 Dockerfile を記述します。ベースイメージには Python 3.12 を指定し、LightGBM に必要な libgomp1 をインストールします。 %%writefile custom_container/Dockerfile FROM python: 3.12 -slim # LightGBM に必須の OS ライブラリをインストール RUN apt-get update && apt-get install -y --no-install-recommends \ libgomp1 \ && rm -rf /var/lib/apt/lists/* # 必要な Python ライブラリのインストール RUN pip install --no-cache- dir \ pandas scikit-learn lightgbm shap matplotlib google-cloud-storage WORKDIR /app COPY task.py /app/task.py ENTRYPOINT [ "python" , "task.py" ] コンテナのビルドとプッシュ Cloud Build を使用してコンテナをビルドし、プッシュします。 # Cloud Build でビルドとプッシュを実行 REPO_NAME = "custom-training-repo" IMAGE_URI = f "{LOCATION}-docker.pkg.dev/{PROJECT_ID}/{REPO_NAME}/lgbm-shap-trainer:latest" !gcloud builds submit --tag {IMAGE_URI} ./custom_container 学習ジョブの実行 作成した自作コンテナ ( IMAGE_URI ) を指定して、学習ジョブを送信します。引数 base_output_dir を指定することで、指定した Cloud Storage のパス配下にモデルや画像を保存できます。 # ジョブの定義 job = aiplatform.CustomContainerTrainingJob( display_name= "diamonds-lgbm-shap-job" , container_uri=IMAGE_URI, ) # ジョブの実行 print ( "ジョブを送信しました。完了までお待ちください..." ) job.run( machine_type= "n1-standard-4" , replica_count= 1 , args=[ f "--train-data-uri={WORK_DIR}/data/train.csv" ], # 成果物の保存先フォルダを指定 base_output_dir=f "{WORK_DIR}/model_output" ) 推論と評価指標の確認 ジョブ完了後、Cloud Storage から学習済みモデルをダウンロードし、Colab Enterprise 上でテストデータに対する精度評価を行います。 import numpy as np import lightgbm as lgb from sklearn.metrics import r2_score, mean_squared_error import pandas as pd # 1. 学習の成果物のダウンロード MODEL_DIR = f "{WORK_DIR}/model_output/model" print ( "学習済みモデルと分析画像をダウンロードします..." ) !gsutil cp {MODEL_DIR}/model.txt . !gsutil cp {MODEL_DIR}/learning_curve.png . !gsutil cp {MODEL_DIR}/shap_importance.png . # 2. テストデータの読み込み df_test = pd.read_csv(f "{WORK_DIR}/data/test.csv" ) cat_cols = [ 'cut' , 'color' , 'clarity' ] for col in cat_cols: df_test[col] = df_test[col].astype( 'category' ) X_test = df_test.drop(columns=[ "price" ]) y_true = df_test[ "price" ] # 3. ローカル推論の実行 local_model = lgb.Booster(model_file= "model.txt" ) predictions = local_model.predict(X_test) # 4. 評価指標の計算と表示 r2 = r2_score(y_true, predictions) rmse = np.sqrt(mean_squared_error(y_true, predictions)) print ( "-" * 30 ) print (f "評価結果 (データ数: {len(y_true)}件)" ) print (f "R2 Score (決定係数): {r2:.4f}" ) print (f "RMSE (誤差の大きさ): {rmse:.4f}" ) print ( "-" * 30 ) 以下は筆者の環境における実行結果です。R2スコアが 0.98 を超える精度の高いモデルが作成できました。 ------------------------------ 評価結果 (データ数: 5394件) R2 Score (決定係数): 0.9817 RMSE (誤差の大きさ): 543.6218 ------------------------------ 分析レポートの解釈 単に予測精度を出すだけでなく、AI が なぜその予測をしたのか を解釈することは実務において重要です。コンテナ内で生成した学習曲線の画像と SHAP を用いた個別データの分析結果を確認します。 import shap from IPython.display import Image, display print ( "=== 学習曲線 (過学習の確認) ===" ) display(Image( "learning_curve.png" )) print ( " \n === 全体の寄与度 (SHAP Beeswarm) ===" ) display(Image( "shap_importance.png" )) # --- 個別データに対するSHAP（表形式）--- print ( " \n === 特定のデータ（1件目）の予測の根拠 ===" ) explainer = shap.TreeExplainer(local_model) single_instance = X_test.iloc[[ 0 ]] shap_values_single = explainer(single_instance) shap_df = pd.DataFrame({ "特徴量 (Feature)" : single_instance.columns, "実際の値 (Value)" : single_instance.values[ 0 ], "価格への影響 (SHAP値)" : shap_values_single.values[ 0 ] }) shap_df = shap_df.reindex(shap_df[ "価格への影響 (SHAP値)" ].abs().sort_values(ascending= False ).index) base_value = explainer.expected_value predicted_price = predictions[ 0 ] print (f "【ベースライン価格 (平均)】: {base_value:.2f}" ) display(shap_df.style.format({ "価格への影響 (SHAP値)" : "{:+.2f}" }).hide(axis= "index" )) print (f "【最終予測価格】: {predicted_price:.2f}" ) 学習曲線（Learning Curve） を確認すると、学習データと検証データの誤差（MSE）が共に右肩下がりで収束しています。 これは、未知のデータである検証データに対しても過学習を起こすことなく学習ができている証拠です。 全体の寄与度 では、上にある特徴量ほど予測への影響力が大きいことを示しています。横軸の 0 を基準に、右側が 価格を上げる要因 、左側が 価格を下げる要因 です。 プロットの赤色は数値が大きいデータであり、青色は数値の小さいデータを表しています。例えば carat は右側に赤色でプロットされているため、 carat が大きいほど高価になる ことが分かります。 最後に 特定の1件に対する予測の根拠 を表形式で出力しました。 全体の平均価格（ベースライン）を基準として、「重さが0.24カラットと小さいためマイナス評価」「透明度（clarity）がVVS1と高品質であるためプラス評価」といったように、最終的な予測価格に至るまでの内部の計算ロジックをビジネス部門に説明できます。 片岩 裕貴 (記事一覧) クラウドソリューション部 クラウドディベロッパー課 和歌山県在住のエンジニア。興味分野はAI/ML。Google Cloud Partner Top Engineer に選出（2025 / 2026）。

G-gen の片岩です。当記事では Vertex AI Custom Training を使用して、機械学習モデルのトレーニングから推論まで実行する方法を紹介します。Vertex AI Custom Training を使うことで、クラウド上のフルマネージド環境で機械学習モデルをトレーニングできます。 はじめに 機械学習モデルのトレーニング手法 Vertex AI Custom Training とは 構成図 初期設定 データの準備と分割 学習スクリプトの作成 学習ジョブの実行 推論の実行 推論結果の確認 はじめに 機械学習モデルのトレーニング手法 Google Cloud のマネージドサービスを使って機械学習モデルをトレーニングする方法は、大きく分けて3つあります。 手法 向いているケース Vertex AI AutoML データを用意するだけで、少ない手順でモデルをトレーニングしたい時 BigQuery ML BigQuery のデータを使用して SQL でモデルをトレーニングしたい時 Vertex AI Custom Training モデルの開発環境や工程を細かく制御したい時 当記事では、最も柔軟性の高い Vertex AI Custom Training に焦点を当て、モデルのトレーニングおよび推論を実行する方法をご紹介します。 Vertex AI についての解説、Vertex AI AutoML や BigQuery ML についての詳細は以下の記事を参照してください。 blog.g-gen.co.jp blog.g-gen.co.jp blog.g-gen.co.jp Vertex AI Custom Training とは Vertex AI Custom Training は、PyTorch、TensorFlow、Scikit-learn や XGBoost といったフレームワークを実行できるフルマネージドサービスです。 ローカルで開発した Python コードを Docker コンテナとして実行するため、サーバーの起動や停止といったインフラ管理を意識せず、 コードの自由度 と クラウドのスケーラビリティ を両立できます。 AutoML では対応しきれない細かいチューニングや、独自のアルゴリズムを実装したいエンジニアにとって、有用な選択肢です。 参考 : カスタム トレーニングの初心者向けガイド 構成図 当記事で紹介する手順に関する構成図は、以下のとおりです。環境構築の負荷を軽減するため、ソースコードの作成や Python 実行環境に Colab Enterprise を使用します。 Colab Enterprise については以下の記事を参照してください。 blog.g-gen.co.jp 初期設定 はじめに、ライブラリのインストールと環境変数の設定を行います。 今後、別のモデルを作成する時にバケットを使い回せるようにするため、Cloud Storage はバケットの中にディレクトリを作成します。 # 必要なライブラリのインストール !pip install google-cloud-aiplatform xgboost scikit-learn pandas -q # プロジェクトとリージョンの設定 # ※ ご自身の環境に合わせて書き換えてください PROJECT_ID = "your-project-id" LOCATION = "asia-northeast1" # バケットとフォルダの定義 ROOT_BUCKET = "gs://your-bucket" EXPERIMENT_NAME = "california-housing-xgb-v1" WORK_DIR = f "{ROOT_BUCKET}/{EXPERIMENT_NAME}" # Vertex AI SDK の初期化 from google.cloud import aiplatform # staging_bucket を指定すると、自動生成されるファイルもこのフォルダに整理されます aiplatform.init(project=PROJECT_ID, location=LOCATION, staging_bucket=WORK_DIR) # バケットが存在しない場合のみ作成 !gsutil mb -l {LOCATION} {ROOT_BUCKET} データの準備と分割 データは機械学習デモにしばしば使用される California Housing（カリフォルニアの住宅価格）を使用します。このデータセットは、scikit-learn に含まれています。 モデルの作成に使用する学習データと推論に使用する推論データに分割し、Cloud Storage に保存します。 from sklearn.datasets import fetch_california_housing from sklearn.model_selection import train_test_split import pandas as pd # データのロード data = fetch_california_housing(as_frame= True ) df = data.frame # 学習データと推論データに 90:10 の割合で分割 train_full_df, test_df = train_test_split(df, test_size= 0.1 , random_state= 42 ) # 学習データの保存 train_filename = "train.csv" train_full_df.to_csv(train_filename, index= False ) # 推論データの保存 test_filename = "test.csv" test_df.to_csv(test_filename, index= False ) # GCS へアップロード (フォルダ /data 配下へ) !gsutil cp {train_filename} {WORK_DIR}/data/{train_filename} !gsutil cp {test_filename} {WORK_DIR}/data/{test_filename} print (f "学習データ: {WORK_DIR}/data/{train_filename}" ) print (f "推論データ: {WORK_DIR}/data/{test_filename}" ) 学習スクリプトの作成 Vertex AI Custom Training として実行する Python スクリプトを作成します。今回は機械学習アルゴリズムに XGBoost を採用します。 環境変数 AIP_MODEL_DIR で与えられた Cloud Storage に作成したモデルを保存すると、自動的に Vertex AI Model Registry にアップロードされます。 %%writefile task.py import argparse import pandas as pd import xgboost as xgb from sklearn.model_selection import train_test_split from sklearn.metrics import mean_squared_error import os from google.cloud import storage from urllib.parse import urlparse # 引数の受け取り parser = argparse.ArgumentParser() parser.add_argument( '--train-data-uri' , dest= 'train_data_uri' , type = str , required= True ) args = parser.parse_args() # GCS から学習用データセットをダウンロード parsed_url = urlparse(args.train_data_uri) bucket_name = parsed_url.netloc blob_path = parsed_url.path.lstrip( "/" ) local_filename = "downloaded_train.csv" client = storage.Client() bucket = client.bucket(bucket_name) blob = bucket.blob(blob_path) print (f "Downloading training data from: {args.train_data_uri}" , flush= True ) blob.download_to_filename(local_filename) # 学習用データセットを読み込む df = pd.read_csv(args.train_data_uri) # 特徴量とターゲットに分離し、Numpy Array に変換 target_col = "MedHouseVal" X = df.drop(columns=[target_col]).values y = df[target_col].values # トレーニングに必要な学習データと評価データに分割する X_train, X_val, y_train, y_val = train_test_split(X, y, test_size= 0.1 , random_state= 42 ) # XGBoost モデルの構築と学習 model = xgb.XGBRegressor( n_estimators= 100 , learning_rate= 0.1 , max_depth= 5 , random_state= 42 , objective= 'reg:squarederror' ) model.fit( X_train, y_train, eval_set=[(X_val, y_val)], verbose= True ) # 評価スコアを Cloud Logging に出力する score = model.score(X_val, y_val) mse = mean_squared_error(y_val, model.predict(X_val)) print (f "Validation Score (R^2): {score:.4f}" , flush= True ) print (f "Validation MSE: {mse:.4f}" , flush= True ) # モデルの保存 model_filename = "model.bst" model.save_model(model_filename) aip_model_dir = os.getenv( "AIP_MODEL_DIR" ) if aip_model_dir: print (f "Uploading model to: {aip_model_dir}" , flush= True ) parsed_url = urlparse(aip_model_dir) bucket_name = parsed_url.netloc blob_path = parsed_url.path.lstrip( "/" ).rstrip( "/" ) + "/" + model_filename # Cloud Storage にアップロード bucket = client.bucket(bucket_name) blob = bucket.blob(blob_path) blob.upload_from_filename(model_filename) print ( "Model upload completed." , flush= True ) 学習ジョブの実行 作成したスクリプトをコンテナで実行します。今回は Google Cloud が 提供するビルド済みコンテナを使用するため、Dockerfile の作成は不要です。 また、軽量なモデルのため n1-standard-4 マシンタイプを使用しますが、巨大なモデルをトレーニングしたい場合は GPU を使用することで高速に処理できます。 # ジョブの定義 job = aiplatform.CustomTrainingJob( display_name= "california-housing-xgb-job" , script_path= "task.py" , container_uri= "asia-docker.pkg.dev/vertex-ai/training/xgboost-cpu.2-1:latest" , requirements=[ "pandas" , "google-cloud-storage" ], staging_bucket=WORK_DIR, model_serving_container_image_uri= "asia-docker.pkg.dev/vertex-ai/prediction/xgboost-cpu.2-1:latest" ) # ジョブの実行 print ( "ジョブを送信しました。完了まで5分ほどお待ちください..." ) model = job.run( machine_type= "n1-standard-4" , replica_count= 1 , model_display_name= "california-housing-xgb-model" , # GCS上の学習データパスを引数で渡す args=[f "--train-data-uri={WORK_DIR}/data/train.csv" ] ) ジョブ終了後に Cloud Storage にモデルが保存されていることを確認できます。 Google Cloud コンソールの モデルレジストリの画面でモデルが作成されていることを確認できます。 推論の実行 本番運用では Vertex AI に備わっているオンライン推論やバッチ推論を使用して大量のリクエストを捌きますが、ここでは Colab Enterprise の実行環境上で推論を実行します。 # --------------------------------------- import xgboost as xgb import pandas as pd import matplotlib.pyplot as plt # --- モデルの場所 --- model_gcs_uri = model.uri # GCSのパスを直接指定する場合 # model_gcs_uri = WORK_DIR + "/aiplatform-custom-training-xxxxx/model" print (f "参照モデル: {model_gcs_uri}" ) # 推論データのダウンロード !gsutil cp {model_gcs_uri}/model.bst . !gsutil cp {WORK_DIR}/data/test.csv . print ( "ダウンロード完了。推論を開始します..." ) # モデルの読み込み local_model = xgb.XGBRegressor() local_model.load_model( "model.bst" ) # データの読み込み df_test = pd.read_csv( "test.csv" ) # 推論用に「正解列」を削除したデータフレームを作ります target_col = "MedHouseVal" X_test = df_test.drop(columns=[target_col]) # 推論の実行 # .values をつけて数値行列として渡します predictions = local_model.predict(X_test.values) # --- 結果の表示 --- print ( " \n === 推論結果 (最初の10件) ===" ) for i, pred in enumerate (predictions[: 10 ]): print (f "データ{i}: 予測価格 {pred:.4f}" ) 推論結果の確認 最後に予測結果と正解データを照合して評価指標を計算し散布図を描画します。 from sklearn.metrics import r2_score, mean_squared_error import numpy as np # 正解データの抽出 y_true = df_test[ "MedHouseVal" ] # 精度評価 (答え合わせ) r2 = r2_score(y_true, predictions) rmse = np.sqrt(mean_squared_error(y_true, predictions)) print ( "-" * 30 ) print (f "評価結果 (データ数: {len(y_true)}件)" ) print (f "R2 Score (決定係数): {r2:.4f}" ) print (f "RMSE (誤差の大きさ): {rmse:.4f}" ) print ( "-" * 30 ) # 結果の可視化 (散布図) plt.figure(figsize=( 8 , 8 )) plt.scatter(y_true, predictions, alpha= 0.5 , color= 'blue' , label= 'Predictions' ) # 理想線 (y=x) min_val = min (y_true.min(), predictions.min()) max_val = max (y_true.max(), predictions.max()) plt.plot([min_val, max_val], [min_val, max_val], 'r--' , label= 'Ideal Fit' ) plt.title(f "Actual vs Predicted (R2: {r2:.3f})" ) plt.xlabel( "Actual Price" ) plt.ylabel( "Predicted Price" ) plt.legend() plt.grid( True ) plt.show() 片岩 裕貴 (記事一覧) クラウドソリューション部 クラウドディベロッパー課 和歌山県在住のエンジニア。興味分野はAI/ML。Google Cloud Partner Top Engineer に選出（2025 / 2026）。