フリート全体のコンピュータビジョンモデルのライフサイクル管理は、品質検査のPoCが成功した瞬間に顕在化する問題です。1台のカメラと1つのモデルなら問題なく動作します。しかし工場が同じ結果を40のライン、それぞれに複数のカメラを備えた環境で求めると、カメラごとに見える世界が異なります。カメラごとに角度が異なり、同じステンレス表面での反射の具合が異なり、どのベイのドアが開いているかによって環境光も異なります。私たちが関わった本番デプロイでは、プールした映像でトレーニングした単一のモデルは、個々のカメラに対して十分な性能を発揮できず、結局各カメラに固有のファインチューニング済みモデルを持たせることになりました。すると本当の問題は、1つのモデルをうまくトレーニングすることではなく、人間を各モデルの関与に介入させることなく、何千ものモデルをトレーニング、評価、バージョン管理、デプロイすることになります。
本記事は、2023年シリーズで紹介したパターンを更新するものです。 エッジでのビジュアル品質検査のためのエンドツーエンドのMLOpsパイプラインを構築する を現行の Amazon SageMaker Python SDK v3向けに更新します。以前非推奨となったAmazon SageMaker Edge Managerに代わり、エッジ最適化にはAmazon SageMaker Neoを使用します。私は、カメラごとに異常検出器と欠陥分類器をトレーニングする二分岐のAmazon SageMaker Pipelineを解説します。品質ゲートにより、性能の低いモデルはブロックされます。パイプライン内のSageMaker Neoコンパイルステップにより、登録されるすべてのモデルにデバイス最適化済みアーティファクトが付属し、AWS Step Functionsオーケストレーションレイヤーがフリート全体のトレーニングのペースを制御します。すべてAWS Cloud Development Kit (AWS CDK)でデプロイし、合成データでのパイプラインの完全な実行にかかるコンピュートコストは約0.07ドルです。
完全なサンプルは以下で入手できます。 aws-samples GitHubリポジトリ.
ソリューションの概要
次の図は、サンプルのフリート全体モデルライフサイクルを示しています。オーケストレーションレイヤーが再トレーニングリクエストをキューイングし、同時実行数を制限します。カメラごとのトレーニングパイプラインがデータを準備し、異常検出器と分類器をトレーニング・評価し、エッジデバイス向けのバージョン管理されたモデルアーティファクトを生成します。
図1: ソリューションアーキテクチャ
トレーニング側は、カメラをパラメータとする単一のSageMaker Pipelineで、実行ごとに13ステップを実行します。3つの入口ステップが並列に実行されます: Preprocess、Fetch Latest Anomaly、Fetch Latest Classifierです。パイプラインはそのカメラの画像を前処理し、各ブランチについてそのカメラの直近の承認済みモデルを取得してから、2つのモデルをトレーニングします。1つ目は、正常な製品の姿を学習し、再構成誤差により異常を検出する小型の畳み込みオートエンコーダです。2つ目は、欠陥を確認するためにファインチューニングされたResNet18分類器です。各ブランチはモデルをONNXにエクスポートし、ホールドアウトデータに対してエクスポートされたアーティファクトを評価し、品質ゲートを通過します。ゲートを通過したモデルは、設定されたエッジターゲット向けにSageMaker Neoでコンパイルされ、その後にSageMaker Model Registryに登録されます。その結果、承認済みのすべてのパッケージにはデバイス対応のコンパイル済みアーティファクトが含まれます。パッケージはカメラIDをメタデータとして持ち、各ブランチのウォームスタート取得はそれでフィルタするため、再トレーニングは常にそのカメラ自身の履歴から再開されます。
オーケストレーション側はフリートのペースを制御します。再トレーニングリクエストはAmazon Simple Queue Service (Amazon SQS) FIFOキューに届き、AWS Lambda関数がリクエストごとにStep Functions実行を開始します。各実行は、パイプラインを開始する前にDynamoDBセマフォでスロットを取得するため、リクエストの急増も一斉に起動するのではなく一定の割合で処理されます。完了はAmazon Simple Notification Service (Amazon SNS)にパブリッシュされ、運用上の障害は元のエラーを保持したままAmazon EventBridgeルールを通じて同じトピックに届きます。
このシステムの出力は、バージョン管理され品質ゲートを通過したカメラごとのモデルアーティファクトです。Model RegistryはポータブルなONNXを保持し、S3はそのモデルのNeoコンパイル済みビルドを保持します。これはONNXではなく、DLRランタイムがロードするデバイス最適化された共有ライブラリです。どちらも、貴社の工場が使用する任意のデリバリー機構に対応できます。
前提条件
サンプルをデプロイするには以下が必要です:
- AWS Identity and Access Management (IAM)ロールを作成する権限を持つAWSアカウント。
- Amazon Simple Storage Service (Amazon S3)バケット。
- Amazon Elastic Container Registry (Amazon ECR)リポジトリ。
- スタック内のSageMaker AI、AWS Step Functions、AWS Lambda、Amazon Simple Queue Service (Amazon SQS)、Amazon Simple Notification Service (Amazon SNS)、Amazon EventBridge、Amazon DynamoDBのリソース。
- Python 3.10、3.11、または3.12、Docker。
- AWS CDK CLIバージョン2.1131.0以降。
それ以前のCLIバージョンは、最新バージョンの aws-cdk-libに対してクラウドアセンブリスキーマの不一致で失敗します。サービスクォータの引き上げは不要です。オーケストレーションレイヤーはデフォルトで同時に最大5つのパイプライン実行を許可し、これにより同時実行のSageMakerトレーニングジョブが標準のアカウントクォータ内に収まります。
CLIのデフォルト以外のAWSリージョンにデプロイする場合は、次の3つすべてを設定してください。 AWS_REGION, AWS_DEFAULT_REGION、さらに CDK_DEFAULT_REGION deploy コマンド自体に対して設定します。シェルのプロファイルではなく、たとえば、 AWS_REGION=us-east-2 AWS_DEFAULT_REGION=us-east-2 CDK_DEFAULT_REGION=us-east-2 cdk deploy --all。CDK CLI、Boto3、そして AWS Command Line Interface (AWS CLI) は、重なりはするが同一ではない変数セットからリージョンを解決するため、不一致が発生すると混乱を招く部分的なデプロイが起こります。コマンドの前にプレフィックスを付ければ、その実行時のみ3つすべてに適用され、ご自身のデフォルトリージョンは変更されずに済みます。
SDK v3で学習パイプラインを定義する
このセクションのコードは、スタンドアロンのスクリプトではなく、リポジトリ内のCDKアプリケーションから抜粋したものであり、スタックがパイプライン定義を合成する際のcdk deployの実行時に動作します。環境のセットアップとデプロイコマンドについては、後の「サンプルのデプロイと実行」で説明します。SDK v3ではSageMaker Python SDKがモジュール型のパッケージに再編成され、インポート文にはライブラリの新しい構造が示されています:
from sagemaker.train import ModelTrainer
from sagemaker.train.configs import InputData, Compute, OutputDataConfig
from sagemaker.core.training.configs import MetricDefinition
from sagemaker.serve.model_builder import ModelBuilder
from sagemaker.mlops.workflow.pipeline import Pipeline
from sagemaker.mlops.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.mlops.workflow.model_step import ModelStep
from sagemaker.mlops.workflow.condition_step import ConditionStep
from sagemaker.core.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.core.workflow.functions import JsonGetインストールする sagemaker>=3.0.0,<4 引き寄せる sagemaker-core, sagemaker-train, sagemaker-serve、そして sagemaker-mlops 個別のパッケージとして。v2があちこちに分散させていたトレーニング設定は Estimator コンストラクタ引数は現在、渡される型付き設定オブジェクトの中にあります。 ModelTrainerこれは異常検知ブランチのトレーナーで、サンプルからの抜粋をそのまま示します:
anomalib_trainer = ModelTrainer(
training_image=custom_training_image_uri,
compute=Compute(
instance_type=training_instance_type.to_string(),
instance_count=1,
),
base_job_name="cv-pipeline-anomalib-train",
sagemaker_session=pipeline_session,
role=role_arn,
hyperparameters={
"program": "train_anomalib.py",
"model_name": "autoencoder",
"image_size": "256",
"num_epochs": "3",
"batch_size": "16",
"learning_rate": "0.001",
},
input_data_config=anomalib_input_data,
output_data_config=OutputDataConfig(
s3_output_path=run_prefix("anomalib-training")
),
).with_metric_definitions(
[
MetricDefinition(
name="image_auroc",
regex=r"validation:image_auroc=([0-9]*\.?[0-9]+)",
),
]
)anomalib_input_data 4チャンネルを配線し（train, validation, test, pretrained_model）を上流ステップのプロパティから取得し、 run_prefix すべてのS3パスをパイプライン実行IDにスコープすることで、同時に実行されるフリートの実行同士が衝突しないようにします。両方ともリポジトリに含まれています。
v3の動作のうち2つは、パイプラインがCDKアプリケーション内部に存在する場合に問題となりますが、どちらも変更履歴には破壊的変更として記載されていません。第一に、SDK v3はパイプライン定義の構築時に、パイプラインの実行ロールをライブのIAMに対して検証します。この検証は実行時ではなく合成時に発生します。CDKアプリでは、ロールとパイプラインを1回のデプロイで作成することはできません。SDKが検証する時点でロールがまだ存在しないためです。これはcdk bootstrapとは別のもので、cdk bootstrapはアカウントとリージョンごとに1回CDKツールキットのリソースをプロビジョニングするものであり、すでに実行済みのアカウントもあるかもしれません。サンプル独自の2フェーズデプロイは、たまたま同じ名前を共有するコンテキストフラグによって制御されています。まず の初期パスでは -c bootstrap=true そのロール、バケット、ECRリポジトリを作成し、2回目のパスで既に存在するロールを前提としてパイプラインを合成します。パイプライン定義に埋め込まれるロールのAmazon Resource Name（ARN）も、未解決のCDKトークンではなく具体的な文字列でなければなりません。
第二に、v3のロールバリデーターはリソースARNのないECRプル権限をシミュレートするため、単一のECRリポジトリにスコープされた実行ロールは、実行時には動作するのに検証に失敗します。このサンプルのロールが、より広範なECR権限を持っているのはこのためです。
最後に、サンプルが用いている v3 の流儀をひとつ挙げます。パイプラインを次のように構築しています: PipelineSession(default_bucket=your_bucket) ステップコードなど SDK 管理のアップロードをパイプライン独自のバケットにルーティングするため、すべてのアーティファクトが 1 か所に集まり、実行ロールの権限も単一のバケットをカバーするだけで済みます。(ロールに SDK のデフォルトバケットへのアクセスを付与する方法でも動作しますが、この方法のほうが整然としています。)
欠陥データの振る舞いに合致する品質ゲート
学習済みモデルはすべて登録前に評価されますが、その評価はメモリ上のPyTorchモデルではなく、エクスポートされたONNXアーティファクトに対して実行されます。この区別は重要です。ゲートを通すアーティファクトがそのまま出荷されるアーティファクトであるため、エクスポートの問題が、一見健全な学習実行の裏で壊れたモデルをすり抜けさせることは決してありません。
異常検知ブランチはAUROC 0.8以上でゲートされます。分類ブランチは2つの指標で同時にゲートを行い、その指標の選択は欠陥データの本質的な特性である深刻なクラス不均衡によって形作られています。このサンプルでは約90パーセントの画像が良品であり、これは生産ラインとしては現実的な数値です。その分布を考えると、単純な正解率は無意味です。すべての画像を「良品」と予測するモデルは、欠陥を1つも検出できないにもかかわらず約0.9のスコアを得てしまうからです。クラスごとの再現率の平均であるbalanced accuracyはこれを補正しますが、それ自身にも盲点があります。良品部品の再現率が完璧で、欠陥の再現率がわずか20パーセントのモデルでも、(1.0 + 0.2) / 2 = 0.6というスコアを出します。そこでゲートはbalanced accuracy 0.6以上と欠陥再現率0.5以上の両方を要求します:
step_check_classifier = ConditionStep(
name="CheckClassifierQuality",
conditions=[
ConditionGreaterThanOrEqualTo(
left=JsonGet(
step_name=step_eval_classifier.name,
property_file="evaluation",
json_path="balanced_accuracy",
),
right=min_accuracy,
),
ConditionGreaterThanOrEqualTo(
left=JsonGet(
step_name=step_eval_classifier.name,
property_file="evaluation",
json_path="defect_recall",
),
right=min_defect_recall,
),
],
if_steps=[step_compile_classifier, step_register_classifier],
else_steps=[step_fail_classifier],
depends_on=[step_eval_classifier],
)A ConditionStep 条件をANDで結合するため、両方の条件を満たす必要があります。しきい値はパイプラインパラメータであり、再デプロイなしに実行ごとに調整できます。ゲートが失敗すると、 FailStep はその実行を失敗としてマークし、そのブランチのモデルはレジストリに到達しません(2つのブランチは独立にゲートされるため、もう一方のブランチのモデルは引き続き登録される可能性があります)。フリートは、そのカメラに対して以前承認されたモデルを提供し続けます。
訓練コードは、同じ不均衡に別の側面から対処しています。分類器はファインチューニングされた resnet18 ImageNetの重みと2クラスのヘッドを備えたものを使用します。カメラごとのデータセットサイズでは、事前学習済みの重みをファインチューニングすることこそが、モデルが多数派クラスに崩壊するのを防ぐ鍵となります。学習コードは、逆クラス頻度を用いて学習バッチのバランスを WeightedRandomSampler で取ります。一方、損失関数は重み付けされないままにしています。重み付き損失と重み付きサンプリングを組み合わせると、少数派クラスへの過剰な補正が起こります。
パイプライン内でSageMaker Neoを使ってエッジ向けにコンパイルする
門の if_steps 上記のコードでは:ゲートを通過すると、まずコンパイルが実行され、登録はそれに依存します。両方のトレーニングスクリプトは動的バッチ軸でONNXをエクスポートします。品質ゲートはONNX Runtime下でそれらのエクスポートされた成果物をスコアリングし、その後 CompileAnomalibNeo または CompileClassifierNeo ステップはゲート付きアーティファクトを引き渡す SageMaker Neo。不変条件は、ゲートを通過してからコンパイルし、その後に登録するという順序です。モデルは、フリートのエッジターゲット向けにコンパイルもされて初めて Approved パッケージとなるため、レジストリがデバイスで実行できないモデルをアドバタイズすることは決してありません。
コンパイルステップは、間違いやすい3つの詳細をカプセル化しています。Neoでは入力のtarballにモデルファイルのみが含まれている必要がありますが、トレーニングジョブは出力として model.tar.gz チェックポイントとメトリクスを ONNX モデルと併せて扱います。このステップはまず ONNX のみの tarball に再パッケージします。また、Neo は具体的な入力形状を必要とする一方、エクスポートでは動的なバッチ軸が宣言されているため、バッチサイズ 1 がコンパイルリクエストで固定されます。さらに、パイプラインのステップは再試行される可能性があるため、コンパイルジョブ名は実行 ID から決定論的に導出され、再試行時には新たに孤立したジョブを作るのではなく、すでに開始したジョブのポーリングを再開します。
ターゲットはパイプラインのパラメータです: NeoTargetDevice などの名前付きデバイスを選択する jetson_xavier または rasp4b、そのデフォルト設定では次のためにコンパイルします。 NeoTargetOs/NeoTargetArch プラットフォーム、デフォルトでは LINUX/X86_64 です。コンパイルされたパッケージには最適化された compiled.so さらに DLRランタイム それをロードするためのものです。デフォルトの出力は任意のx86_64 Linuxホスト上で実行されるため、デバイス固有のターゲットにコミットする前に検証できます。コンパイル自体には課金は発生しません。課金対象は、それを駆動する処理インスタンスと出力のS3ストレージです。再トレーニングなしで、既に登録済みのモデルを後から追加のデバイスタイプ向けにコンパイルする場合、リポジトリにはModel Registryに対して同じフローを実行するアドホックスクリプトが含まれています。
Model Registry におけるカメラ別モデル
各ブランチはそれぞれ独自のモデルパッケージグループに登録され、すべてのパッケージにはカメラIDとその評価メトリクスが含まれます。サンプルでは、メトリクスの接続部分を省略しています:
classifier_model_builder = ModelBuilder(
s3_model_data_url=classifier_model_data,
image_uri=custom_training_image_uri,
sagemaker_session=pipeline_session,
role_arn=role_arn,
)
step_register_classifier = ModelStep(
name="RegisterClassifierModel",
depends_on=[step_compile_classifier],
step_args=classifier_model_builder.register(
model_package_group_name="sm1k-classifier-model-group",
approval_status="Approved",
model_metrics=classifier_metrics,
customer_metadata_properties={"CameraId": camera_id},
),
)カメラの識別情報は customer_metadata_properties カメラごとのモデルを名前空間ごとに整理する仕組みこそが、2つの共有パッケージグループを1つのフリートカタログへと変えるものです。全カメラのパッケージが同じグループ内に存在し、カメラごとの履歴はメタデータでのフィルタリングによって復元されます。各ブランチにはフェッチステップがあり、要求元カメラの最新モデルが見つかるまで、承認済みパッケージを新しい順にページ送りしながら検索します。なぜなら ListModelPackages 顧客のメタデータをサーバー側でフィルタリングすることはできません。カメラの初回実行時にはフェッチで何も見つからず、トレーニングはコールドスタートの初期化にフォールバックします(分類器にはImageNetの重み、オートエンコーダにはランダム初期化)。それ以降の実行では、そのカメラ自身の以前のモデルから再開されます。これこそが、フリート全体で頻繁かつ低コストなインクリメンタル再トレーニングを可能にしている点です。
これらのレジストリエントリの性質を正しく理解してください。それらはバージョン管理され品質ゲートを通過したアーティファクトをカタログ化するものですが、そのコンテナ参照は推論エンドポイントを一切提供しないトレーニングイメージです。これらは、下流のシステムがアーティファクトを取得してデバイスに配布するフリートワークフロー向けのものであり、 CreateEndpoint.
フロート訓練リクエストの調整
すべてのAWSアカウントには、同時に実行できるSageMakerトレーニングジョブに対するサービスクォータが設定されており、ここでの各パイプライン実行は複数のジョブを起動するため、フリートの再トレーニングにはペース配分が必要です。500件のリクエストの急増が500件の同時パイプライン実行になるのを防ぐ仕組みがどこかに必要です。このサンプルではセマフォを使用しています。これは利用可能なスロットのカウンターで、5台分の駐車スペースを持つ駐車場のように機能します。Step Functionsのステートマシンがパイプラインを開始する前に、DynamoDB内のカウンターをアトミックにインクリメントしてスロットを確保します。上限に達した後はインクリメントを拒否する条件が付けられています:
acquire_semaphore = tasks.DynamoUpdateItem(
self, "AcquireSemaphore",
table=semaphore_table,
key={"LockName": tasks.DynamoAttributeValue.from_string("pipeline-concurrency")},
update_expression="SET ActiveCount = if_not_exists(ActiveCount, :zero) + :one",
condition_expression="attribute_not_exists(ActiveCount) OR ActiveCount < :limit",
)すべてのスロットが埋まっている場合、実行は待機してリトライを行います。パイプラインが完了すると、カウンターをデクリメントしてスロットを解放します。自分の設計に持ち込む価値のある詳細が一つあります。DynamoDBの条件式は、ステートマシンがカウンターを読み取って判断するのではなく、単一のアトミック操作で上限を強制します。読み取りと書き込みの間に競合の窓が生じるのを防ぐためです。リポジトリには、残りの運用面の堅牢化が実装されており、すべての失敗パスでのスロット解放や、Amazon EventBridgeを通じた失敗した実行のAmazon SNSへの転送が含まれます。待機境界のサイジングについてはREADMEが説明しています。
5という上限はスタックパラメータであり、クォータ増加の申請をせずに標準的なアカウントでサンプルをデプロイして実行できるよう、低めに設定されています。ご自身のアカウントのトレーニングジョブのクォータが許容する値に調整してください。1つのアカウントのクォータを超えてフリート規模のトレーニングを拡張するのは、それ自体が独立したトピックです。標準的な手段は、サービスクォータの引き上げ、インスタンスタイプやリージョンへのトレーニングの分散、複数アカウントへのフリートの分散です。SQSでバックログを保持し、キャパシティが空いた分だけ作業を受け入れるという方法もあります。これらの選択は、フリートの規模と構成によって異なります。このサンプルは、拡張のベースとなる動作するスケルトンを提供します。
サンプルのデプロイと実行
リポジトリのREADMEに記載のウォークスルーが正式な手順です。以下の簡易版は、クリーンなチェックアウトから始めて、初回デプロイ前に環境をセットアップするものです:
git clone https://github.com/aws-samples/sample-sagemaker-fleet-cv-pipeline
cd sample-sagemaker-fleet-cv-pipeline && python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cdk bootstrap # once per account/region
cdk deploy SageMakerCVPipelineStack -c bootstrap=true # roles, bucket, ECR
cdk deploy --all # pipeline + orchestration
./scripts/push_ecr_image.sh # build and push training imageデフォルトでは、パイプラインは合成の生産ライン画像を生成するため、データセットやデータ承認なしでシステム全体をエンドツーエンドで検証できます。完全な実行では、両方のNeoコンパイルを含むすべての13ステップが約15分で実行され、課金対象の時間として約$0.07がかかります(米国東部 (オハイオ) リージョン (us-east-2)、2026年7月、オンデマンド料金)。 ml.m5.xlarge time (US East (Ohio) Region (us-east-2)、2026年7月、オンデマンド)。オーケストレーションリソースはサンプル規模では合計で月額$0.10を大きく下回ります。
パイプラインを実際の画像で試すために、リポジトリには BCCDデータセット用のダウンローダーが含まれています。このデータセットは、もともとGitHubでオープンソース化され、MITライセンスの下でPascal VOC形式に再編成された、アノテーション付きの364枚の血液塗抹標本顕微鏡画像のコレクションです。同じデータセットは Roboflow によってパブリックドメインとしても配布されています。スクリプトは血小板の存在を欠陥クラスにマッピングし、オプションで画像をカメラごとに分割してフリートをシミュレートし、トレーニング用のtarballをパイプラインのバケットにアップロードします。付属のノートブックは実際のデータで完全なパイプラインを実行し、正常サンプルと異常サンプルの両方でONNX推論を並べて実行します。血小板を欠陥として扱うことは、製造欠陥分布の説明用の代用であり、臨床上の主張ではありません。
クリーンアップ
モデルパッケージはパイプラインの実行時に作成されるため、AWS CloudFormationはそれらを削除できません。まずAWS CLIで削除してから、 cdk destroy --all -c bootstrap=true --forceでスタックを破棄してください。 README には正確なコマンドが記載されています。
まとめ
この記事では、フリート規模のコンピュータビジョンライフサイクルのための、動作するSageMaker SDK v3リファレンスアーキテクチャを紹介しました。このアーキテクチャには、すべてのモデルがエクスポートされたONNXアーティファクトで品質ゲートを通過し、SageMaker Neoでコンパイルされ、その後初めて登録される、デュアルブランチのトレーニングパイプラインが含まれています。Model Registryのレイアウトでは、カメラのメタデータを使用して共有パッケージグループをカメラごとのウォームスタート可能な履歴に変換し、Step Functionsのオーケストレーションレイヤーがアカウントのクォータ内でフリートのトレーニングのペースを制御します。Amazon Lookout for Visionのようなマネージドオプションでニーズが満たされるチームにとっては、そちらがよりシンプルな道です。このパターンは、マネージドオプションではカバーされないカメラごとのフリート規模の制約を対象としています。この設計原則はコンピュータビジョンを超えて応用できます。精度ではなく、バランス精度と少数クラスのリコールでゲートを設けること、出荷するアーティファクトを評価およびコンパイルすること、そして運用上の障害が発生した場合にはサイレントに終了するのではなく通知することです。
サンプルは数分でデプロイでき、完全な検証実行には約7セントのコンピュートコストしかかからないため、低コストの次のステップとして リポジトリ をクローンして実行することが挙げられます。そこから、BCCDダウンローダーを実際のバケットに向けたり、ハードウェアに合わせて NeoTargetDevice を設定したり、フリートに合わせてオーケストレーションの上限を調整したりできます。SDK v3への移行自体については、 SageMaker Python SDKドキュメント と、新しいSDKの発表を紹介する2部構成のシリーズが、私が常に開いていた参考資料です。
このパターンを本番環境に適用する際にサポートが必要な場合は、AWSのアカウントチームにお問い合わせください。