Planification de tâches de surveillance de dérive de biais

Après avoir créé votre référence, vous pouvez appeler la méthode create_monitoring_schedule() de votre instance de classe ModelBiasModelMonitor pour planifier une surveillance horaire de la dérive de biais. Les sections suivantes expliquent comment créer une surveillance de la dérive de biais pour un modèle déployé sur un point de terminaison en temps réel ainsi que pour une tâche de transformation par lots.

Important

Vous pouvez spécifier une entrée de transformation par lots ou une entrée de point de terminaison, mais pas les deux, lorsque vous créez votre planification de surveillance.

Contrairement à la surveillance de la qualité des données, vous devez fournir des étiquettes Ground Truth si vous souhaitez contrôler la qualité des modèles. Cependant, les étiquettes Ground Truth pourraient être retardées. Pour résoudre ce problème, spécifiez les décalages lorsque vous créez votre programme de surveillance. Pour plus d'informations sur comment créer des décalages temporels, consultez Décalages de Model Monitor.

Si vous avez envoyé une tâche de baselining, le moniteur récupère automatiquement la configuration d'analyse à partir de la tâche de baselining. Si vous ignorez l'étape de baselining ou si la nature du jeu de données de capture est différente de celle du jeu de données d'entraînement, vous devez fournir la configuration d'analyse.

Surveillance de la dérive de biais pour les modèles déployés sur des points de terminaison en temps réel

Pour planifier une surveillance de la dérive de biais pour un point de terminaison en temps réel, transmettez votre instance EndpointInput à l'argument endpoint_input de votre instance ModelBiasModelMonitor, comme indiqué dans l'exemple de code suivant :

from sagemaker.model_monitor import CronExpressionGenerator
            
model_bias_monitor = ModelBiasModelMonitor(
    role=sagemaker.get_execution_role(),
    ...
)

model_bias_analysis_config = None
if not model_bias_monitor.latest_baselining_job:
    model_bias_analysis_config = BiasAnalysisConfig(
        model_bias_config,
        headers=all_headers,
        label=label_header,
    )

model_bias_monitor.create_monitoring_schedule(
    monitor_schedule_name=schedule_name,
    post_analytics_processor_script=s3_code_postprocessor_uri,
    output_s3_uri=s3_report_path,
    statistics=model_bias_monitor.baseline_statistics(),
    constraints=model_bias_monitor.suggested_constraints(),
    schedule_cron_expression=CronExpressionGenerator.hourly(),
    enable_cloudwatch_metrics=True,
    analysis_config=model_bias_analysis_config,
    endpoint_input=EndpointInput(
        endpoint_name=endpoint_name,
        destination="/opt/ml/processing/input/endpoint",
        start_time_offset="-PT1H",
        end_time_offset="-PT0H",
        probability_threshold_attribute=0.8,
    ),
)

Surveillance de la dérive des biais pour les tâches de transformation par lots

Pour planifier une surveillance de la dérive de biais pour une tâche de transformation par lots, transmettez votre instance BatchTransformInput à l'argument batch_transform_input de votre instance ModelBiasModelMonitor, comme indiqué dans l'exemple de code suivant :

from sagemaker.model_monitor import CronExpressionGenerator
                
model_bias_monitor = ModelBiasModelMonitor(
    role=sagemaker.get_execution_role(),
    ...
)

model_bias_analysis_config = None
if not model_bias_monitor.latest_baselining_job:
    model_bias_analysis_config = BiasAnalysisConfig(
        model_bias_config,
        headers=all_headers,
        label=label_header,
    )
    
schedule = model_bias_monitor.create_monitoring_schedule(
   monitor_schedule_name=schedule_name,
   post_analytics_processor_script=s3_code_postprocessor_uri,
   output_s3_uri=s3_report_path,
   statistics=model_bias_monitor.baseline_statistics(),
   constraints=model_bias_monitor.suggested_constraints(),
   schedule_cron_expression=CronExpressionGenerator.hourly(),
   enable_cloudwatch_metrics=True,
   analysis_config=model_bias_analysis_config,
   batch_transform_input=BatchTransformInput(
        destination="opt/ml/processing/input",
        data_captured_destination_s3_uri=s3_capture_path,
        start_time_offset="-PT1H",
        end_time_offset="-PT0H",
        probability_threshold_attribute=0.8
   ),
)