生产变体

在生产 ML 工作流中，数据科学家和工程师经常尝试以各种方式提高性能，例如使用 SageMaker 执行自动模型调优、通过额外或较新的数据进行训练、改进功能选择、使用更新过的更好实例和提供容器。您可以使用生产变体来比较您的模型、实例和容器，并选择性能最佳的候选变体来响应推理请求。

使用 SageMaker 多变体端点，您可以为每个生产变体提供流量分配以在多个变体之间分配端点调用请求，也可以为每个请求直接调用特定的变体。在本主题中，我们介绍了两种测试 ML 模型的方法。

主题

指定流量分配以测试模型
调用特定的变体以测试模型
模型 A/B 测试示例

指定流量分配以测试模型

要在多个模型之间分配流量以测试这些模型，请在端点配置中为每个生产变体指定权重以指定路由到每个模型的流量百分比。有关信息，请参阅 CreateEndpointConfig。下图更详细地说明了它的工作方式。

调用特定的变体以测试模型

要通过为每个请求调用特定模型来测试多个模型，请在调用 InvokeEndpoint 时，通过提供 TargetVariant 参数值来指定要调用的模型的特定版本。SageMaker 会确保由您指定的生产变体处理该请求。如果您已提供流量分配并指定 TargetVariant 参数值，目标路由将覆盖随机流量分配。下图更详细地说明了它的工作方式。

模型 A/B 测试示例

对于新模型的验证过程，在新模型和旧模型之间使用生产流量执行 A/B 测试可能是有效的最后一步。在 A/B 测试中，您测试模型的不同变体，并比较每个变体的性能。如果模型的较新版本提供的性能优于以前存在的版本，请在生产中使用模型的新版本替换旧版本。

以下示例说明了如何执行 A/B 模型测试。有关实施该示例的示例笔记本，请参阅在生产中对 ML 模型进行 A/B 测试。

步骤 1：创建并部署模型

首先，我们定义模型在 Amazon S3 中的位置。在后续步骤中部署模型时，将使用这些位置：


model_url = f"s3://{path_to_model_1}"
model_url2 = f"s3://{path_to_model_2}"

接下来，我们使用图像和模型数据创建模型对象。这些模型对象用于在端点上部署生产变体。这些模型是通过不同数据集、不同算法或 ML 框架以及不同超参数训练 ML 模型而开发的：


from sagemaker.amazon.amazon_estimator import get_image_uri

model_name = f"DEMO-xgb-churn-pred-{datetime.now():%Y-%m-%d-%H-%M-%S}"
model_name2 = f"DEMO-xgb-churn-pred2-{datetime.now():%Y-%m-%d-%H-%M-%S}"
image_uri = get_image_uri(boto3.Session().region_name, 'xgboost', '0.90-1')
image_uri2 = get_image_uri(boto3.Session().region_name, 'xgboost', '0.90-2')

sm_session.create_model(
    name=model_name,
    role=role,
    container_defs={
        'Image': image_uri,
        'ModelDataUrl': model_url
    }
)

sm_session.create_model(
    name=model_name2,
    role=role,
    container_defs={
        'Image': image_uri2,
        'ModelDataUrl': model_url2
    }
)

现在，我们创建两个生产变体，每个变体具有自己不同的模型和资源要求（实例类型和数量）。这样，您还可以在不同的实例类型上测试模型。

我们将两个变体的 initial_weight 设置为 1。这意味着，50% 的请求发送到 Variant1，其余 50% 的请求发送到 Variant2。两个变体的权重总和为 2，每个变体的权重分配为 1。这意味着，每个变体接收总流量的 1/2 或 50%。


from sagemaker.session import production_variant

variant1 = production_variant(
               model_name=model_name,
               instance_type="ml.m5.xlarge",
               initial_instance_count=1,
               variant_name='Variant1',
               initial_weight=1,
           )

variant2 = production_variant(
               model_name=model_name2,
               instance_type="ml.m5.xlarge",
               initial_instance_count=1,
               variant_name='Variant2',
               initial_weight=1,
           )

最后，我们准备在 SageMaker 端点上部署这些生产变体。


endpoint_name = f"DEMO-xgb-churn-pred-{datetime.now():%Y-%m-%d-%H-%M-%S}"
print(f"EndpointName={endpoint_name}")

sm_session.endpoint_from_production_variants(
    name=endpoint_name,
    production_variants=[variant1, variant2]
)

步骤 2：调用部署的模型

现在，我们将请求发送到该端点以实时获得推理结果。我们同时使用流量分配和直接定位。

首先，我们使用在上一步中配置的流量分配。每个推理响应包含处理请求的生产变体的名称，因此，我们可以看到发送到两个生产变体的流量大致相等。


# get a subset of test data for a quick test
!tail -120 test_data/test-dataset-input-cols.csv > test_data/test_sample_tail_input_cols.csv
print(f"Sending test traffic to the endpoint {endpoint_name}. \nPlease wait...")

with open('test_data/test_sample_tail_input_cols.csv', 'r') as f:
    for row in f:
        print(".", end="", flush=True)
        payload = row.rstrip('\n')
        sm_runtime.invoke_endpoint(
            EndpointName=endpoint_name,
            ContentType="text/csv",
            Body=payload
        )
        time.sleep(0.5)

print("Done!")

SageMaker 针对 Amazon CloudWatch 中的每个变体发出 Latency 和 Invocations 等指标。有关 SageMaker 发出的指标的完整列表，请参阅使用 Amazon CloudWatch 监控 Amazon SageMaker。让我们查询 CloudWatch 以获取每个变体的调用次数，以显示默认情况下如何在变体之间拆分调用：

现在，让我们在 invoke_endpoint 调用中将 Variant1 指定为 TargetVariant 以调用特定模型版本。


print(f"Sending test traffic to the endpoint {endpoint_name}. \nPlease wait...")
with open('test_data/test_sample_tail_input_cols.csv', 'r') as f:
    for row in f:
        print(".", end="", flush=True)
        payload = row.rstrip('\n')
        sm_runtime.invoke_endpoint(
            EndpointName=endpoint_name,
            ContentType="text/csv",
            Body=payload,
            TargetVariant="Variant1"
        ) 
        time.sleep(0.5)

为了确认 Variant1 已处理所有新的调用，我们可以查询 CloudWatch 以获取每个变量的调用次数。我们看到，对于最新的调用（最新的时间戳），Variant1 按我们指定的方式处理了所有请求。没有针对 Variant2 的调用。

步骤 3：评估模型性能

为了查看哪个模型版本的性能更好，让我们评估每个变体的准确性、精度、召回率、F1 分数以及受试者操作特征/曲线下面积。首先，让我们看一下 Variant1 的这些指标：

现在，让我们看一下 Variant2 的指标：

对于我们定义的大多数指标，Variant2 性能更好，因此，这是我们要在生产中使用的变体。

步骤 4：为最佳模型增加流量

既然我们已确定 Variant2 性能比 Variant1 好，我们将更多流量转移到该变体。我们可以继续使用 TargetVariant 调用特定的模型变体，但更简单的方法是调用 UpdateEndpointWeightsAndCapacities 以更新分配给每个变体的权重。这会更改生产变体的流量分配，而无需更新端点。回顾一下设置部分，我们设置变体权重以将流量拆分为 50/50。下面有关每个变体的总调用次数的 CloudWatch 指标向我们显示每个变体的调用方式：

现在，我们使用 UpdateEndpointWeightsAndCapacities 为每个变体分配新的权重，将 75% 的流量转移到 Variant2。现在，SageMaker 向 Variant2 发送 75% 的推理请求，其余 25% 的请求发送给 Variant1。


sm.update_endpoint_weights_and_capacities(
    EndpointName=endpoint_name,
    DesiredWeightsAndCapacities=[
        {
            "DesiredWeight": 25,
            "VariantName": variant1["VariantName"]
        },
        {
            "DesiredWeight": 75,
            "VariantName": variant2["VariantName"]
        }
    ]
)

有关每个变体的总调用次数的 CloudWatch 指标向我们显示 Variant2 的调用次数比 Variant1 高：

我们可以继续监控指标，在对某个变体的性能感到满意时，我们可以将 100% 的流量路由到该变体。我们使用 UpdateEndpointWeightsAndCapacities 更新变体的流量分配。Variant1 的权重设置为 0，Variant2 的权重设置为 1。现在，SageMaker 会将所有推理请求全部发送给 Variant2。


sm.update_endpoint_weights_and_capacities(
    EndpointName=endpoint_name,
    DesiredWeightsAndCapacities=[
        {
            "DesiredWeight": 0,
            "VariantName": variant1["VariantName"]
        },
        {
            "DesiredWeight": 1,
            "VariantName": variant2["VariantName"]
        }
    ]
)

有关每个变体的总调用次数的 CloudWatch 指标表明，所有推理请求由 Variant2 进行处理，而没有由 Variant1 处理的推理请求。

现在，您可以安全地更新端点，并将 Variant1 从端点中删除。您也可以在端点中添加新的变体，并执行步骤 2-4 以继续在生产中测试新模型。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

在生产中安全验证模型

阴影变体