模型监控器常见问题

有关 Amazon SageMaker 模型监控器的更多信息，请参阅以下常见问题。

问：模型监控器和 SageMaker Clarify 如何帮助客户监控模型行为？

客户可以通过 Amazon SageMaker 模型监控器和 SageMaker Clarify 从如下四个维度监控模型行为：数据质量、模型质量、偏差偏移和特征归因偏移。模型监控器会持续监控生产中的 Amazon SageMaker 机器学习模型的质量。这包括监控数据质量和模型质量指标（如准确性和 RMSE）的偏移。SageMaker Clarify 偏差监控可帮助数据科学家和机器学习工程师监控模型预测中的偏差和特征归因偏移。

问：启用 SageMaker 模型监控器后，后台会发生什么？

Amazon SageMaker 模型监控器可自动监控模型，从而无需手动监控模型或构建任何其他工具。为了自动执行该过程，模型监控器使您能够使用训练模型时所用的数据创建一组基准统计数据和约束，然后设置计划来监控对端点所做的预测。模型监控器使用规则来检测模型中的偏差，并在出现偏差时提醒您。以下步骤描述了启用模型监控时会发生的情况：

当您使用模型监控器监控批量转换作业而不是实时端点时，模型监控器将监控推理输入和输出，而不是接收对端点的请求并跟踪预测。在模型监控器计划中，客户提供要在处理作业中使用的实例的数量和类型。无论当前执行的状态如何，这些资源都会一直保留到计划被删除为止。

问：什么是数据捕获，为什么需要它，以及如何启用它？

要将模型端点的输入以及已部署模型的推理输出记录到 Amazon S3，可以启用名为数据捕获的特征。有关如何为实时端点和批量转换作业启用该特征的更多详细信息，请参阅从实时端点捕获数据和从批量转换作业捕获数据。

问：启用数据捕获会影响实时端点的性能吗？

数据捕获以异步方式进行，不会影响生产流量。启用数据捕获后，请求和响应负载将与其他一些元数据一起保存到您在 DataCaptureConfig 中指定的 Amazon S3 位置。请注意，将捕获的数据传播到 Amazon S3 时可能会有延迟。

还可以通过列出存储在 Amazon S3 中的数据捕获文件来查看捕获的数据。Amazon S3 路径格式为：s3:///{endpoint-name}/{variant-name}/yyyy/mm/dd/hh/filename.jsonl。Amazon S3 数据捕获应与模型监控器计划位于同一区域。您还应确保基准数据集的列名称仅使用小写字母，并使用下划线 (_) 作为唯一分隔符。

问：为什么模型监控需要 Ground Truth？

模型监控器的以下特征需要使用 Ground Truth 标签：

问：对于利用 Ground Truth 进行标注的客户，我可以采取哪些措施来监控模型的质量？

模型质量监控将您的模型所做的预测与 Ground Truth 标签进行比较，以衡量模型的质量。为此，您需要定期为端点或批量转换作业捕获的数据添加标签，然后将其上传到 Amazon S3。除了捕获之外，模型偏差监控的执行还需要 Ground Truth 数据。在实际使用案例中，应定期收集 Ground Truth 数据并将其上传到指定的 Amazon S3 位置。要将 Ground Truth 标签与捕获的预测数据进行匹配，数据集中的每条记录都必须有一个唯一的标识符。有关 Ground Truth 数据的每条记录的结构，请参阅摄取 Ground Truth 标签并将其与预测合并。

以下代码示例可用于为表格数据集生成人工 Ground Truth 数据。

import random

def ground_truth_with_id(inference_id):
    random.seed(inference_id)  # to get consistent results
    rand = random.random()
    # format required by the merge container
    return {
        "groundTruthData": {
            "data": "1" if rand < 0.7 else "0",  # randomly generate positive labels 70% of the time
            "encoding": "CSV",
        },
        "eventMetadata": {
            "eventId": str(inference_id),
        },
        "eventVersion": "0",
    }


def upload_ground_truth(upload_time):
    records = [ground_truth_with_id(i) for i in range(test_dataset_size)]
    fake_records = [json.dumps(r) for r in records]
    data_to_upload = "\n".join(fake_records)
    target_s3_uri = f"{ground_truth_upload_path}/{upload_time:%Y/%m/%d/%H/%M%S}.jsonl"
    print(f"Uploading {len(fake_records)} records to", target_s3_uri)
    S3Uploader.upload_string_as_file_body(data_to_upload, target_s3_uri)
# Generate data for the last hour
upload_ground_truth(datetime.utcnow() - timedelta(hours=1))
# Generate data once a hour
def generate_fake_ground_truth(terminate_event):
    upload_ground_truth(datetime.utcnow())
    for _ in range(0, 60):
        time.sleep(60)
        if terminate_event.is_set():
            break


ground_truth_thread = WorkerThread(do_run=generate_fake_ground_truth)
ground_truth_thread.start()

以下代码示例演示了如何生成人工流量以发送到模型端点。请注意上面用于调用的 inferenceId 属性。如果存在该属性，则用它联接 Ground Truth 数据（否则使用 eventId）。

import threading

class WorkerThread(threading.Thread):
    def __init__(self, do_run, *args, **kwargs):
        super(WorkerThread, self).__init__(*args, **kwargs)
        self.__do_run = do_run
        self.__terminate_event = threading.Event()

    def terminate(self):
        self.__terminate_event.set()

    def run(self):
        while not self.__terminate_event.is_set():
            self.__do_run(self.__terminate_event)
def invoke_endpoint(terminate_event):
    with open(test_dataset, "r") as f:
        i = 0
        for row in f:
            payload = row.rstrip("\n")
            response = sagemaker_runtime_client.invoke_endpoint(
                EndpointName=endpoint_name,
                ContentType="text/csv",
                Body=payload,
                InferenceId=str(i),  # unique ID per row
            )
            i += 1
            response["Body"].read()
            time.sleep(1)
            if terminate_event.is_set():
                break


# Keep invoking the endpoint with test data
invoke_endpoint_thread = WorkerThread(do_run=invoke_endpoint)
invoke_endpoint_thread.start()

必须将 Ground Truth 数据上传到路径格式与已捕获数据的路径格式相同的 Amazon S3 存储桶，格式如下：s3://<bucket>/<prefix>/yyyy/mm/dd/hh

问：客户如何自定义监控计划？

除了使用内置监控机制之外，还可以使用预处理和后处理脚本，或通过使用或构建您自己的容器来创建您自己的自定义监控计划和过程。需要注意的是，预处理和后处理脚本仅适用于数据和模型质量作业。

利用 Amazon SageMaker，您可以监控和评估模型端点观察到的数据。为此，您必须创建一个用于比较实时流量的基准。基准准备就绪后，请设置一个计划，以持续评估并与基准进行比较。创建计划时，您可以提供预处理和后处理脚本。

以下示例说明如何使用预处理和后处理脚本自定义监控计划。

import boto3, osfrom sagemaker import get_execution_role, Sessionfrom sagemaker.model_monitor import CronExpressionGenerator, DefaultModelMonitor
# Upload pre and postprocessor scripts
session = Session()
bucket = boto3.Session().resource("s3").Bucket(session.default_bucket())
prefix = "demo-sagemaker-model-monitor"
pre_processor_script = bucket.Object(os.path.join(prefix, "preprocessor.py")).upload_file("preprocessor.py")
post_processor_script = bucket.Object(os.path.join(prefix, "postprocessor.py")).upload_file("postprocessor.py")
# Get execution role
role = get_execution_role() # can be an empty string
# Instance type
instance_type = "instance-type"
# instance_type = "ml.m5.xlarge" # Example
# Create a monitoring schedule with pre and post-processing
my_default_monitor = DefaultModelMonitor(
    role=role,
    instance_count=1,
    instance_type=instance_type,
    volume_size_in_gb=20,
    max_runtime_in_seconds=3600,
)

s3_report_path = "s3://{}/{}".format(bucket, "reports")
monitor_schedule_name = "monitor-schedule-name"
endpoint_name = "endpoint-name"
my_default_monitor.create_monitoring_schedule(
    post_analytics_processor_script=post_processor_script,
    record_preprocessor_script=pre_processor_script,
    monitor_schedule_name=monitor_schedule_name,
    # use endpoint_input for real-time endpoint
    endpoint_input=endpoint_name,
    # or use batch_transform_input for batch transform jobs
# batch_transform_input=batch_transform_name,
    output_s3_uri=s3_report_path,
    statistics=my_default_monitor.baseline_statistics(),
    constraints=my_default_monitor.suggested_constraints(),
    schedule_cron_expression=CronExpressionGenerator.hourly(),
    enable_cloudwatch_metrics=True,
)

问：我可以在哪些场景或使用案例中使用预处理脚本？

当需要将输入转换为模型监控器时，可以使用预处理脚本。考虑以下示例场景：

问：我什么时候可以利用后处理脚本？

成功运行监控后，您可以将后处理脚本用作扩展。以下是一个简单示例，但您可以执行或调用成功运行监控后需要执行的任何业务功能。

def postprocess_handler(): 
    print("Hello from the post-processing script!")

问：我什么时候应该考虑自带容器进行模型监控？

SageMaker 提供了一个预构建的容器，用于分析从端点捕获的数据或表格数据集的批量转换作业。但在某些情况下，您可能需要创建自己的容器。考虑以下场景：

问：我有 NLP 和 CV 模型。如何监控它们是否存在数据偏差？

Amazon SageMaker 的预构建容器支持表格数据集。如果您想监控 NLP 和 CV 模型，可以利用模型监控器提供的扩展点自带容器。有关要求的更多详细信息，请参阅自带容器。下面提供了更多示例：

问：我想删除已启用模型监控器的模型端点，但由于监控计划仍处于活动状态，因此无法删除。我应该怎么办？

如果要删除已启用模型监控器的 SageMaker 中托管的推理端点，必须先删除模型监控计划（通过 DeleteMonitoringSchedule CLI 或 API）。然后，删除该端点。

问：SageMaker 模型监控器是否会计算输入的指标和统计数据？

模型监控器计算输出而不是输入的指标和统计数据。

问：SageMaker 模型监控器是否支持多模型端点？

否，模型监控器仅支持托管单个模型的端点，不支持监控多模型端点。

问：SageMaker 模型监控器是否提供有关推理管道中各个容器的监控数据？

模型监控器支持监控推理管道，但捕获和分析数据是针对整个管道而不是针对管道中的各个容器完成的。

问：设置数据捕获后，我能做些什么来防止推理请求受到影响？

为了防止对推理请求产生影响，数据捕获功能会在磁盘利用率较高时停止捕获请求。建议将磁盘利用率保持在 75% 以下，以确保数据捕获功能继续捕获请求。

问：Amazon S3 数据捕获是否可以位于与设置监控计划的区域不同的 Amazon 区域？

否，Amazon S3 数据捕获必须与监控计划位于同一区域。

问：什么是基准，如何创建基准？ 我可以创建自定义基准吗？

基准用作比较模型的实时预测或批量预测的参考。它计算统计数据和指标以及对它们的约束。监控期间，将结合使用所有这些来识别违规行为。

要使用 Amazon SageMaker 模型监控器的默认解决方案，您可以利用 Amazon SageMaker Python SDK。具体而言，使用 ModelMonitor 或 ModelQualityMonitor 类的 suggest_baseline 方法来触发计算基准指标和约束的处理作业。

基准作业的结果是两个文件：statistics.json 和 constraints.json。统计数据的架构和约束的架构包含相应文件的架构。使用生成的约束进行监控之前，您可以查看这些约束并对其进行修改。根据您对域和业务问题的理解，您可以使约束更严格，也可以放宽约束以控制违规的数量和性质。

问：创建基准数据集的指导原则是什么？

任何监控的首要条件都是要有一个基准数据集，用于计算指标和约束。通常，这是模型使用的训练数据集，但在某些情况下，您可能会选择使用其他参考数据集。

基准数据集的列名称应与 Spark 兼容。为了保持 Spark、CSV、JSON 和 Parquet 之间的最大兼容性，建议仅使用小写字母，并且仅使用 _ 作为分隔符。包括 “ ” 在内的特殊字符可能会导致问题。

问：StartTimeOffset 和 EndTimeOffset 参数是什么？何时使用它们？

当模型质量等监控作业需要 Amazon SageMaker Ground Truth 时，您需要确保监控作业只使用 Ground Truth 可用的数据。EndpointInput 的 start_time_offset 和 end_time_offset 参数可用于选择监控作业使用的数据。监控作业使用由 start_time_offset 和 end_time_offset 定义的时段中的数据。这些参数需要以 ISO 8601 持续时间格式指定。下面是一些示例：

问：我是否可以运行“按需”监控作业？

是，您可以通过运行 SageMaker Processing 作业来运行“按需”监控作业。对于批量转换，SageMaker Pipelines 有一个 MonitorBatchTransformStep，您可以用它来创建按需运行监控作业的 SageMaker 管道。SageMaker 示例存储库包含用于按需运行数据质量和模型质量监控作业的代码示例。

问：如何设置模型监控器？

您可以使用以下方法设置模型监控器：

问：模型监控器如何与 SageMaker Model Dashboard 集成

SageMaker Model Dashboard 通过提供有关偏离预期行为的自动警报和问题排查，可以对所有模型进行统一监控，以检查模型并分析随时间推移影响模型性能的因素。