车队级计算机视觉模型生命周期管理的问题，会在质量检测概念验证成功的瞬间出现。一个摄像头配一个模型工作得很好。然后工厂要求在 40 条生产线上实现同样的效果，每条线有几个摄像头，而每个摄像头看到的画面都不一样。它们有不同的角度、同一不锈钢表面上不同的反光，以及随哪个车间大门打开而不同的环境光。在我们参与过的生产部署中，用汇总素材训练的单一模型，在足够多的单个摄像头上的表现都不佳，以至于每个摄像头最终都需要自己的微调模型。于是真正的问题不再是训练好一个模型，而是无需人工逐一介入地训练、评估、版本管理和发布数千个模型。
本文更新了 2023 年系列中引入的模式 构建边缘端视觉质量检测的端到端 MLOps 流水线 以适配当前的 Amazon SageMaker Python SDK v3。它用 Amazon SageMaker Neo 取代了已被弃用的 Amazon SageMaker Edge Manager 来进行边缘优化。我将介绍一个双分支的 Amazon SageMaker 流水线，它为每个摄像头训练一个异常检测器和一个缺陷分类器，并配备阻止表现不佳模型的质量门控。流水线内的 SageMaker Neo 编译步骤确保每个注册模型都附带设备优化产物，AWS Step Functions 编排层则控制车队训练的节奏。所有内容都通过 AWS Cloud Development Kit (AWS CDK) 部署，在合成数据上完整运行一次流水线的计算成本约为 0.07 美元。
完整示例可在以下位置获取： aws-samples GitHub 仓库.
解决方案概览
下图展示了示例的车队级模型生命周期。编排层将再训练请求排队并限制并发工作量。每个摄像头的训练流水线准备数据、训练并评估一个异常检测器和一个分类器，并为边缘设备生成带版本的模型产物。
图 1：解决方案架构
训练部分是一个单一的 SageMaker 流水线，以摄像头为参数，每次执行运行 13 个步骤。三个入口步骤并行运行：Preprocess、Fetch Latest Anomaly 和 Fetch Latest Classifier。流水线对该摄像头的图像进行预处理，并为每个分支获取该摄像头最近一个已批准的模型，然后训练两个模型。第一个是一个小型卷积自编码器，它学习正常产品的样子，并通过重建误差来标记异常。第二个是经过微调以确认缺陷的 ResNet18 分类器。每个分支将其模型导出为 ONNX，用留出数据评估导出的产物，并通过质量门控。通过门控后，模型会使用 SageMaker Neo 为配置的边缘目标进行编译，然后才注册到 SageMaker Model Registry。这样，每个已批准的包都具有设备就绪的编译产物。包以摄像头 ID 作为元数据，每个分支的 warm-start 获取会基于它进行过滤，因此再训练总是从该摄像头自己的历史记录恢复。
编排部分控制车队节奏。再训练请求进入 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 编译版本，这是一个由 DLR 运行时加载的设备优化共享库，而不是 ONNX。两者都已准备好适配您工厂使用的任何交付机制。
前提条件
部署该示例需要以下内容：
- 一个有权创建 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 版本在与以下组件的最新版本交互时会因 cloud assembly 架构不匹配而失败： aws-cdk-lib。无需提升任何服务配额。编排层默认最多同时允许五个流水线执行，这使并发 SageMaker 训练作业保持在标准账户配额之内。
如果您部署到的 AWS 区域不是 CLI 的默认区域，请设置以下全部三项 AWS_REGION, AWS_DEFAULT_REGION，以及 CDK_DEFAULT_REGION 直接写在部署命令本身上,而不是写在你的 shell 配置文件中,例如, 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) 从重叠但不完全相同的变量集合中解析区域，不一致时会产生令人困惑的部分部署。为命令添加前缀仅为该次调用设置这三者，因此您自己的默认区域保持不变。
使用 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 从上游步骤属性连接四个通道（train, validation, test, pretrained_model），并且 run_prefix 将每个 S3 路径限定在流水线执行 ID 下，因此并发的车队运行不会相互冲突。两者都在仓库中。
当流水线位于 CDK 应用内时，有两个 v3 行为需要注意，而它们在变更日志中都没有被列为破坏性变更。首先，SDK v3 在构建流水线定义时会针对实时 IAM 验证流水线的执行角色。这种验证发生在合成时而非执行时。CDK 应用无法在一次部署中同时创建角色和流水线，因为 SDK 验证时角色尚不存在。这与 cdk bootstrap 是分开的，后者为每个账户和区域一次性预置 CDK 工具包资源，您的账户中可能已经完成。示例自身的两阶段部署由一个恰好与该名称相同的上下文标志驱动。第一遍使用 -c bootstrap=true 创建角色、存储桶和 ECR 仓库，第二遍则针对现已存在的角色合成流水线。嵌入在流水线定义中的角色 Amazon Resource Name (ARN) 也必须是具体的字符串，而不是未解析的 CDK 令牌。
其次，v3 的角色验证器在不带资源 ARN 的情况下模拟 ECR 拉取权限，因此限定在单个 ECR 仓库的执行角色即使在实际运行时可行，也无法通过验证。正是出于这个原因，示例中的角色拥有更广泛的 ECR 授权。
最后，示例使用的一个 v3 习惯做法：使用 PipelineSession(default_bucket=your_bucket) 构造流水线，这会将 SDK 管理的上传（如步骤代码）路由到流水线自己的存储桶中，因此所有工件都集中在一处，执行角色的权限也只需覆盖单个存储桶。（授予角色访问 SDK 默认存储桶的权限也可以。这种方式更整洁。）
与缺陷数据行为相匹配的质量门控
每个训练好的模型在注册之前都会被评估，而且评估针对的是导出的 ONNX 工件而非内存中的 PyTorch 模型。这一区别很重要：您所门控的工件就是您所发布的工件，因此导出问题绝不可能让一个损坏的模型在看似健康的训练运行之后蒙混过关。
异常分支以至少 0.8 的 AUROC 作为门控。分类器分支同时以两个指标进行门控，指标的选择取决于缺陷数据的本质特征：严重的类别不平衡。在此示例中，大约 90% 的图像是良品，这在生产线上是符合实际的。在这种分布下，普通准确率毫无意义，因为一个对所有图像都预测“良品”的模型得分约为 0.9，却抓不到任何缺陷。平衡准确率（各类别召回率的均值）可以纠正这一点，但它也有自己的盲区。一个对良品具有完美召回率、而对缺陷只有 20% 召回率的模型仍能得到 (1.0 + 0.2) / 2 = 0.6 的分数。因此门控要求平衡准确率至少为 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 会将该执行标记为失败，该分支的模型永远无法到达注册表（两个分支独立门控，因此另一分支的模型仍可能注册）。车队将继续为该摄像头提供之前已批准的模型。
训练代码从另一方面处理同样的不平衡问题。分类器是一个经过微调的 resnet18 ，带有 ImageNet 权重和一个两类输出头。在每个摄像头的数据集规模下，微调预训练权重正是防止模型坍缩到多数类的关键。训练代码使用基于逆类别频率的 WeightedRandomSampler 来平衡训练批次，而损失函数保持不加权。将加权损失与加权采样结合使用会向少数类过度矫正。
在流水线内使用 SageMaker Neo 面向边缘进行编译
注意前述代码中的 if_steps 门控：过了门控，编译先运行，注册依赖它。两个训练脚本都导出带有动态批次轴的 ONNX。质量门控在 ONNX Runtime 下对导出的工件进行评分，然后一个 CompileAnomalibNeo 或 CompileClassifierNeo 步骤将已通过门控的工件交给 SageMaker Neo。其不变式是：先门控，再编译，再注册。一个模型只有在还为机群的边缘目标成功编译后，才会成为 Approved 包，因此注册表绝不会宣传设备无法运行的模型。
编译步骤封装了三个容易出错的细节。Neo 要求输入 tar 包只包含模型文件，但训练作业输出的 model.tar.gz 会将 checkpoint 和指标与 ONNX 模型放在一起。该步骤首先重新打包一个仅含 ONNX 的 tar 包。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 中，摄像机标识将两个共享包组变成了一个机群目录。所有摄像机的包都存在于相同的组中，按摄像机的历史记录通过过滤元数据来恢复。每个分支都有一个获取步骤，按从新到旧的顺序分页遍历已批准的包，直到找到请求摄像机的最新模型，因为 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 是一个堆栈参数，设置得较低，以便示例无需申请配额提升即可在任何标准账户中部署和运行。请根据您账户的训练作业配额进行调整。超出单个账户配额之后，扩展机群训练本身就是一个独立话题。标准手段包括提高服务配额、将训练分散到不同实例类型或区域，以及将机群分布到多个账户。您还可以将积压工作保存在 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默认情况下,该管道生成合成的生产线图像,因此无需数据集和任何数据审批即可端到端验证整个系统。完整运行会执行全部 13 个步骤,包括两次 Neo 编译,大约需要 15 分钟,计费成本约为 0.07 美元 ml.m5.xlarge 时间(美国东部(俄亥俄)区域 (us-east-2),2026 年 7 月,按需价格)。在样本规模下,编排资源合计每月成本远低于 0.10 美元。
为了在真实图像上测试该管道,代码仓库包含一个针对 BCCD 数据集的下载器。该数据集包含 364 张带标注的血液涂片显微图像,最初在 GitHub 上开源,并以 MIT 许可证重新整理为 Pascal VOC 格式。同一数据集也由 Roboflow 以公有领域形式分发。该脚本将血小板的存在映射为缺陷类别,可选地按相机拆分图像以模拟设备群,并将训练压缩包上传到管道存储桶。随附的 notebook 在真实数据上运行完整管道,并对正常样本和异常样本进行并排的 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 等托管选项满足的团队,那是更简单的路径。本模式针对的是托管选项未覆盖的按相机、设备群规模的约束。这些设计规则可以推广到计算机视觉之外:以平衡准确率加少数类召回率而非准确率作为门控标准,对你要交付的工件进行评估和编译,并让操作性故障发出通知而不是默默终止。
该示例几分钟即可部署,一次完整的验证运行大约只需七美分的计算成本,因此一个低成本的下一步是克隆该 代码仓库 并运行它。之后,将 BCCD 下载器指向真实的存储桶,为你的硬件设置 NeoTargetDevice 容量,或根据你的设备群调整编排边界。对于 SDK v3 迁移本身, SageMaker Python SDK 文档 以及宣布新 SDK 的两部分系列文章是我全程保持打开的参考资料。
如果你希望在将此模式适配到你的生产环境时获得帮助,请联系你的 AWS 客户团队。