人工智能
使用 SageMaker SDK v3 在边缘实现车队级计算机视觉生命周期管理车队级计算机视觉模型生命周期管理是一个在质量检测概念验证成功后立刻就会显现的问题。一台摄像头和一个模型运行良好。随后工厂要求在 40 条产线上获得同样的结果，每条产线都有多台摄像头，而每台摄像头看到的世界都不相同。每台摄像头角度不同，同一不锈钢表面的反光不同，环境光也因哪扇车间门打开而不同。在我们参与的生产部署中，用汇总素材训练的单一模型在足够多的单台摄像头上表现不佳，以至于每台摄像头最终都需要自己的微调模型。现在真正的问题不再是训练好一个模型，而是训练、评估、版本管理并交付数千个模型，且无需为每个模型都安排人工介入。
本文针对当前的 Amazon SageMaker Python SDK v3 更新了 2023 年系列文章 Build an end-to-end MLOps pipeline for visual quality inspection at the edge 中介绍的模式。它用 Amazon SageMaker Neo 进行边缘优化，取代了此后已弃用的 Amazon SageMaker Edge Manager。我将逐步介绍一个双分支 Amazon SageMaker Pipeline，它为每台摄像头训练一个异常检测器和一个缺陷分类器，并设有质量门禁来拦截表现不佳的模型。流水线内的 SageMaker Neo 编译步骤确保每个注册模型都附带设备优化后的产物，而 AWS Step Functions 编排层则控制车队训练的节奏。你使用 AWS Cloud Development Kit（AWS CDK）部署所有内容，在合成数据上完整运行一次流水线的计算成本约为 $0.07。
完整示例可在 aws-samples GitHub 仓库中获取.
解决方案概览
下图展示了该示例的车队级模型生命周期。编排层将再训练请求排队，并限制并发工作量。每台摄像头的训练流水线负责准备数据、训练并评估异常检测器和分类器，并为边缘设备生成带版本的模型产物。
图 1：解决方案架构
训练部分是一个以摄像头为参数的单一 SageMaker Pipeline，每次执行运行 13 个步骤。三个入口步骤并行运行：Preprocess、Fetch Latest Anomaly 和 Fetch Latest Classifier。该流水线预处理该摄像头的图像，并为每个分支获取该摄像头最近一次获批的模型，然后训练两个模型。第一个是一个小型卷积自编码器，它学习正常产品的外观，并通过重建误差标记异常。第二个是经过微调的 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 编译构建，这是一个由 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 schema）时会出现不匹配错误。 aws-cdk-lib无需增加服务配额。默认情况下，编排层一次最多允许五个流水线执行，这使并发的 SageMaker 训练作业保持在标准账户配额之内。
如果你部署到非 CLI 默认区域的 AWS 区域，请在 deploy 命令本身上设置全部三个 AWS_REGION, AWS_DEFAULT_REGION，以及 CDK_DEFAULT_REGION 在 deploy 命令本身上设置，而不是在你的 shell 配置文件中，例如， AWS_REGION=us-east-2 AWS_DEFAULT_REGION=us-east-2 CDK_DEFAULT_REGION=us-east-2 cdk deploy --allCDK CLI、Boto3 和 AWS 命令行界面（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 角色验证器在模拟 ECR 拉取权限时没有指定资源 ARN，因此即使执行角色在运行时能够正常工作，只要它被限定到单个 ECR 仓库，就会无法通过验证。正因如此，示例中的角色携带了范围更广的 ECR 授权。
最后，示例采用的一个 v3 习惯做法是：用以下方式构建流水线： PipelineSession(default_bucket=your_bucket) 将 SDK 管理的上传（例如步骤代码）路由到流水线自己的存储桶中，这样每个产物都存放在同一个位置，执行角色的权限也只需覆盖单个存储桶。（授予该角色访问 SDK 默认存储桶的权限同样可行。这种做法更整洁。）
与缺陷数据实际表现相匹配的质量门禁
每个训练好的模型在注册前都会经过评估，而且评估针对的是导出的 ONNX 产物，而不是内存中的 PyTorch 模型。这一区别很重要：你设门禁的产物就是你交付的产物，因此导出问题绝不可能让一个损坏的模型绕过看似健康的训练运行。
异常分支的门禁要求 AUROC 至少达到 0.8。分类器分支则同时以两项指标作为门禁条件，而指标的选择取决于缺陷数据的决定性特征：严重的类别不平衡。在这个样本中，大约 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],
)一个 ConditionStep 并且它的各项条件，因此两者都必须通过。这些阈值是流水线参数，可按每次执行调整而无需重新部署。如果某个门禁失败，一个 FailStep 表示执行失败，该分支的模型永远不会进入注册表（两个分支独立门控，因此另一分支的模型仍可能注册）。车队会继续为该摄像头提供此前已批准的模型。
训练代码则从另一侧解决同样的不平衡问题。分类器是一个微调后的 resnet18 并带有 ImageNet 权重和双类别分类头。在每台摄像头的数据集规模下，微调预训练权重正是防止模型坍缩到多数类上的关键。训练代码通过一个 WeightedRandomSampler 使用逆类别频率，而损失保持不加权。将加权损失与加权采样结合会过度校正，偏向少数类。
在流水线内部使用 SageMaker Neo 为边缘编译
注意，门禁的 if_steps 在前面的代码中：通过门禁之后，编译先运行，注册依赖于它。两个训练脚本都导出具有动态批次轴的 ONNX。质量门禁在 ONNX Runtime 下对这些导出的产物进行评分，然后一个 CompileAnomalibNeo 或 CompileClassifierNeo 步骤将门禁通过的产物交给 SageMaker Neo。 SageMaker Neo不变量是：先门禁，再编译，然后注册。一个模型只有在也为车队的边缘目标编译成功时，才会成为已批准包，因此注册表绝不会公布设备无法运行的模型。
编译步骤封装了三个容易出错的细节。Neo 要求输入的 tarball 只包含模型文件，但训练作业会输出 model.tar.gz 以及检查点和指标。该步骤会先重新打包一个仅含 ONNX 的 tarball。Neo 还要求具体的输入形状，而导出产物声明的是动态批次轴，因此在编译请求中固定批次大小为 1。又因为流水线步骤可能重试，编译作业名称由执行 ID 确定性地派生而来，这样重试时会继续轮询它已经启动的作业，而不是将其遗弃。
目标是流水线参数： NeoTargetDevice 选择一个命名设备，例如 jetson_xavier ，其默认值为 / 平台编译，默认为 LINUX/X86_64。 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 个同时启动的管道。示例使用信号量，即一个可用槽位计数器，其工作方式就像一个只有五个车位的停车场。在 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 涵盖了等待边界的规模设定。
上限为五是一个堆栈参数，设置得较低，以便示例能在任何标准账户中部署和运行，而无需申请提高配额。请根据你账户的训练作业配额支持情况调整它。超出单个账户的配额后，扩展车队训练就是另一个独立话题了。标准手段包括提高服务配额、将训练分散到不同实例类型或区域，以及将车队分布到多个账户。你也可以将积压工作保留在 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（美国东部（俄亥俄）区域（us-east-2），2026 年 7 月，按需）。 ml.m5.xlarge （美国东部（俄亥俄）区域（us-east-2），2026 年 7 月，按需）。在示例规模下，这些编排资源合计每月成本远低于 $0.10。
为了在真实图像上运行该流水线，仓库中包含一个下载器，用于下载 BCCD 数据集该数据集包含364张带标注的血涂片显微镜图像，最初在GitHub上开源，并以MIT许可证重新整理为Pascal VOC格式。同一数据集也由 Roboflow 作为公共领域。该脚本将血小板的存在映射到缺陷类别，可选地按摄像头拆分图像以模拟一个车队，并将训练 tar 包上传到流水线存储桶。随附的 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 客户团队。