Applicare SageMaker Smart Sifting allo script PyTorch

Configura l’interfaccia di SageMaker Smart Sifting.

La libreria SageMaker Smart Sifting implementa una tecnica di campionamento basata sulla perdita a soglia relativa che aiuta a filtrare i campioni con un impatto minore sulla riduzione del valore di perdita. L’algoritmo di SageMaker Smart Sifting calcola il valore di perdita di ogni campione di dati di input utilizzando un passaggio avanti e calcola il relativo percentile rispetto ai valori di perdita dei dati precedenti.

È necessario specificare i due parametri seguenti per la classe RelativeProbabilisticSiftConfig per creare un oggetto di configurazione di sifting.

Il seguente esempio di codice mostra l’impostazione di un oggetto della classe RelativeProbabilisticSiftConfig.

from smart_sifting.sift_config.sift_configs import (
    RelativeProbabilisticSiftConfig
    LossConfig
    SiftingBaseConfig
)

sift_config=RelativeProbabilisticSiftConfig(
    beta_value=0.5,
    loss_history_length=500,
    loss_based_sift_config=LossConfig(
         sift_config=SiftingBaseConfig(sift_delay=0)
    )
)

Per ulteriori informazioni sul parametro loss_based_sift_config e sulle classi correlate, consulta Moduli di configurazione di SageMaker Smart Sifting nella sezione del riferimento di SageMaker Smart Sifting Python SDK.

L’oggetto sift_config nell’esempio di codice precedente viene utilizzato nella fase 4 per configurare la classe SiftingDataloader.

(Facoltativo) Configura una classe di trasformazione in batch di SageMaker Smart Sifting.

Casi d’uso diversi per l’addestramento richiedono formati di dati di addestramento differenti. Data la varietà di formati di dati, l’algoritmo di SageMaker Smart Sifting deve identificare come eseguire il sifting su un determinato batch. Per risolvere questo problema, SageMaker Smart Sifting fornisce un modulo di trasformazione in batch che aiuta a convertire i batch in formati standardizzati che possono essere filtrati in modo efficiente.

1. SageMaker Smart Sifting gestisce la trasformazione in batch dei dati di addestramento nei seguenti formati: elenchi Python, dizionari, tuple e tensori. Per questi formati di dati, SageMaker Smart Sifting gestisce automaticamente la conversione del formato di dati in batch ed è possibile ignorare il resto della procedura. Se salti questa fase, nella fase 4 per la configurazione di SiftingDataloader, mantieni il valore predefinito del parametro batch_transforms di SiftingDataloader, ovvero None.

2. Se il set di dati non è in questi formati, procedi con il resto di questa fase per creare una trasformazione in batch personalizzata mediante SiftingBatchTransform.

   Nei casi in cui il set di dati non è in uno dei formati supportati da SageMaker Smart Sifting, potrebbero verificarsi errori. Tali errori di formato dei dati possono essere risolti aggiungendo il parametro batch_format_index oppure batch_transforms alla classe SiftingDataloader, configurata nella fase 4. Di seguito sono riportati alcuni esempi di errori dovuti a un formato di dati incompatibile e le relative risoluzioni.

   Messaggio di errore	Risoluzione
   Batches of type {type(batch)} are not supported by default.	Questo errore indica che il formato del batch non è supportato per impostazione predefinita. È necessario implementare una classe di trasformazione in batch personalizzata e utilizzarla specificandola nel parametro batch_transforms della classe SiftingDataloader.
   Unable to index the batch of type {type(batch)}	Questo errore indica che l’oggetto di batch non può essere indicizzato normalmente. L’utente deve implementare una trasformazione in batch personalizzata e passarla utilizzando il parametro batch_transforms.
   Batch size {batch_size} does not match dimension 0 or dimension 1 sizes	Questo errore si verifica quando la dimensione del batch fornita non corrisponde alla dimensione 0 oppure 1 del batch. L’utente deve implementare una trasformazione in batch personalizzata e passarla utilizzando il parametro batch_transforms.
   Both dimension 0 and dimension 1 match batch size	Questo errore indica che, poiché più dimensioni corrispondono alla dimensione del batch fornita, sono necessarie ulteriori informazioni per filtrare il batch. L’utente può fornire il parametro batch_format_index per indicare se il batch è indicizzabile per campione o funzionalità. Gli utenti possono anche implementare una trasformazione in batch personalizzata, che tuttavia richiede più lavoro del necessario.

   Per risolvere i problemi sopra menzionati, è necessario creare una classe di trasformazione in batch personalizzata utilizzando il modulo SiftingBatchTransform. Una classe di trasformazione in batch deve essere costituita da una coppia di funzioni di trasformazione diretta e trasformazione inversa. La coppia di funzioni converte il formato dei dati in un formato elaborabile dall’algoritmo di SageMaker Smart Sifting. Dopo aver creato una classe di trasformazione in batch, questa restituisce un oggetto SiftingBatch da passare alla classe SiftingDataloader nella fase 4.

   Di seguito sono riportati alcuni esempi di classi di trasformazione in batch personalizzate del modulo SiftingBatchTransform.

   • Un esempio di implementazione di trasformazione in batch di elenchi personalizzata con la funzionalità di smart sifting di SageMaker per i casi in cui il blocco dello strumento di caricamento dei dati contiene input, maschere ed etichette.

     from typing import Any
     
     import torch
     
     from smart_sifting.data_model.data_model_interface import SiftingBatchTransform
     from smart_sifting.data_model.list_batch import ListBatch
     
     class ListBatchTransform(SiftingBatchTransform):
         def transform(self, batch: Any):
             inputs = batch[0].tolist()
             labels = batch[-1].tolist()  # assume the last one is the list of labels
             return ListBatch(inputs, labels)
     
         def reverse_transform(self, list_batch: ListBatch):
             a_batch = [torch.tensor(list_batch.inputs), torch.tensor(list_batch.labels)]
             return a_batch

   • Un esempio di implementazione di trasformazione in batch di elenchi personalizzata con SageMaker Smart Sifting per i casi in cui non sono necessarie etichette per la trasformazione inversa.

     class ListBatchTransformNoLabels(SiftingBatchTransform):
         def transform(self, batch: Any):
             return ListBatch(batch[0].tolist())
     
         def reverse_transform(self, list_batch: ListBatch):
             a_batch = [torch.tensor(list_batch.inputs)]
             return a_batch

   • Un esempio di implementazione di batch di tensori personalizzata con la funzionalità di smart sifting di SageMaker per i casi in cui il blocco dello strumento di caricamento dei dati contiene input, maschere ed etichette.

     from typing import Any
     
     from smart_sifting.data_model.data_model_interface import SiftingBatchTransform
     from smart_sifting.data_model.tensor_batch import TensorBatch
     
     class TensorBatchTransform(SiftingBatchTransform):
         def transform(self, batch: Any):
             a_tensor_batch = TensorBatch(
                 batch[0], batch[-1]
             )  # assume the last one is the list of labels
             return a_tensor_batch
     
         def reverse_transform(self, tensor_batch: TensorBatch):
             a_batch = [tensor_batch.inputs, tensor_batch.labels]
             return a_batch

   Dopo aver creato una classe di trasformazione in batch implementata da SiftingBatchTransform, utilizza questa classe nella fase 4 per configurare la classe SiftingDataloader. Il resto di questa guida presuppone la creazione di una classe ListBatchTransform. Nella fase 4, questa classe viene passata a batch_transforms.

Crea una classe per implementare l’interfaccia Loss di SageMaker Smart Sifting. Questo tutorial presuppone che la classe sia denominata SiftingImplementedLoss. Durante la configurazione di questa classe, è consigliabile utilizzare la stessa funzione di perdita nel ciclo di addestramento dei modelli. Segui le seguenti fasi secondarie per creare una classe implementata da Loss per SageMaker Smart Sifting.

1. SageMaker Smart Sifting calcola un valore di perdita per ogni campione di dati di addestramento, anziché calcolare un singolo valore di perdita per un batch. Per garantire che SageMaker Smart Sifting utilizzi la stessa logica di calcolo delle perdite, crea una funzione di perdita implementata con smart sifting utilizzando il relativo modulo Loss di SageMaker che utilizza la funzione di perdita e calcola la perdita per campione di addestramento.

   Suggerimento

   L’algoritmo di SageMaker Smart Sifting viene eseguito su ogni campione di dati, non sull’intero batch, quindi è necessario aggiungere una funzione di inizializzazione per impostare la funzione di perdita PyTorch senza alcuna strategia di riduzione.

   class SiftingImplementedLoss(Loss):  
       def __init__(self):
           self.loss = torch.nn.CrossEntropyLoss(reduction='none')

   Questo processo viene mostrato anche nell’esempio di codice seguente.

2. Definisci una funzione di perdita che accetti original_batch (o transformed_batch se hai impostato una trasformazione in batch nella fase 2) e il modello PyTorch. Utilizzando la funzione di perdita specificata senza alcuna riduzione, SageMaker Smart Sifting esegue un passaggio avanti per ogni campione di dati per valutarne il valore di perdita.

Il codice seguente è un esempio di interfaccia Loss implementata con smart sifting denominata SiftingImplementedLoss.

from typing import Any

import torch
import torch.nn as nn
from torch import Tensor

from smart_sifting.data_model.data_model_interface import SiftingBatch
from smart_sifting.loss.abstract_sift_loss_module import Loss

model=... # a PyTorch model based on torch.nn.Module

class SiftingImplementedLoss(Loss):   
    # You should add the following initializaztion function 
    # to calculate loss per sample, not per batch.
    def __init__(self):
        self.loss_no_reduction = torch.nn.CrossEntropyLoss(reduction='none')

    def loss(
        self,
        model: torch.nn.Module,
        transformed_batch: SiftingBatch,
        original_batch: Any = None,
    ) -> torch.Tensor:
        device = next(model.parameters()).device
        batch = [t.to(device) for t in original_batch] # use this if you use original batch and skipped step 2
        # batch = [t.to(device) for t in transformed_batch] # use this if you transformed batches in step 2

        # compute loss
        outputs = model(batch)
        return self.loss_no_reduction(outputs.logits, batch[2])

Prima che il ciclo di addestramento raggiunga l’effettivo passaggio avanti, questo calcolo della perdita di sifting viene eseguito durante la fase di caricamento dei dati di recupero di un batch in ogni iterazione. Il valore di perdita individuale viene quindi confrontato con i valori di perdita precedenti e il relativo percentile relativo viene stimato in base all’oggetto di RelativeProbabilisticSiftConfig impostato nella fase 1.

Esegui il wrapping dello strumento di caricamento dei dati di PyTroch dalla classe SiftingDataloader di SageMaker AI.

Infine, utilizza tutte le classi implementate con la funzionalità SageMaker Smart Sifting configurata nelle fasi precedenti per la classe di configurazione SiftingDataloder di SageMaker AI. Questa classe è un wrapper per PyTorch DataLoader. Eseguendo il wrapping di PyTorch DataLoader, SageMaker Smart Sifting viene registrato per essere eseguito come parte del caricamento dei dati in ogni iterazione di un job di addestramento PyTorch. Il seguente esempio di codice dimostra l’implementazione di SageMaker AI Smart Sifting su PyTorch DataLoader.

from smart_sifting.dataloader.sift_dataloader import SiftingDataloader
from torch.utils.data import DataLoader

train_dataloader = DataLoader(...) # PyTorch data loader

# Wrap the PyTorch data loader by SiftingDataloder
train_dataloader = SiftingDataloader(
    sift_config=sift_config, # config object of RelativeProbabilisticSiftConfig
    orig_dataloader=train_dataloader,
    batch_transforms=ListBatchTransform(), # Optional, this is the custom class from step 2
    loss_impl=SiftingImplementedLoss(), # PyTorch loss function wrapped by the Sifting Loss interface
    model=model,
    log_batch_data=False
)