SageMaker スマートふるい分けを PyTorch スクリプトに適用する

SageMaker スマートふるい分けインターフェイスを設定します。

SageMaker スマートふるい分けライブラリは、相対的しきい値の損失ベースのサンプリング手法を実装しており、損失値の減少への影響が少ないサンプルを除外するのに役立ちます。 SageMaker スマートふるい分けアルゴリズムは、フォワードパスを使用してすべての入力データサンプルの損失値を計算し、前のデータの損失値に対する相対パーセンタイルを計算します。

次の 2 つのパラメータは、ふるい分け設定オブジェクトを作成するためにRelativeProbabilisticSiftConfigクラスに指定する必要があります。

次のコード例は、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)
    )
)

loss_based_sift_config パラメータと関連クラスの詳細については、 SageMaker 「スマートふるい分け Python SDKリファレンス」セクションSageMaker スマートふるい分け設定モジュールの「」を参照してください。

前述のコード例のsift_configオブジェクトは、SiftingDataloaderクラスを設定するステップ 4 で使用されます。

（オプション) SageMaker スマートふるい分けバッチ変換クラスを設定します。

トレーニングのユースケースごとに、トレーニングデータ形式が異なります。さまざまなデータ形式を考慮すると、 SageMaker スマートふるい分けアルゴリズムは、特定のバッチでふるい分けを実行する方法を特定する必要があります。これに対処するために、 SageMaker スマートふるい分けはバッチを効率的にふるい分けできる標準化された形式に変換するのに役立つバッチ変換モジュールを提供します。

1. SageMaker スマートふるいは、Python リスト、ディクショナリ、タプル、テンソルの形式でトレーニングデータのバッチ変換を処理します。これらのデータ形式の場合、 SageMaker スマートふるいはバッチデータ形式の変換を自動的に処理し、このステップの残りの部分はスキップできます。このステップをスキップする場合は、 の設定手順 4 でSiftingDataloader、 の batch_transforms パラメータをデフォルト値 SiftingDataloaderのままにしますNone。

2. データセットがこれらの形式でない場合は、このステップの残りの部分に移動して、 を使用してカスタムバッチ変換を作成する必要がありますSiftingBatchTransform。

   SageMaker スマートふるいでサポートされている形式にデータセットがない場合、エラーが発生する可能性があります。このようなデータ形式のエラーは、ステップ 4 で設定した batch_format_indexまたは batch_transformsパラメータをSiftingDataloaderクラスに追加することで解決できます。互換性のないデータ形式と解決方法によるエラーの例を次に示します。

   エラーメッセージ	解決方法
   タイプのバッチ {type(batch)} はデフォルトではサポートされていません。	このエラーは、バッチ形式がデフォルトでサポートされていないことを示します。カスタムバッチ変換クラスを実装し、SiftingDataloaderクラスの batch_transformsパラメータに指定してこれを使用する必要があります。
   タイプのバッチをインデックス化できません {type(batch)}	このエラーは、バッチオブジェクトを正常にインデックス化できないことを示します。ユーザーはカスタムバッチ変換を実装し、 batch_transformsパラメータを使用してこれを渡す必要があります。
   バッチサイズ {batch_size} はディメンション 0 またはディメンション 1 のサイズと一致しません	このエラーは、指定されたバッチサイズがバッチの 0 番目または 1 番目のディメンションと一致しない場合に発生します。ユーザーはカスタムバッチ変換を実装し、 batch_transformsパラメータを使用してこれを渡す必要があります。
   ディメンション 0 とディメンション 1 の両方がバッチサイズと一致する	このエラーは、複数のディメンションが指定されたバッチサイズと一致するため、バッチをふるい分けるためにより多くの情報が必要であることを示します。ユーザーは batch_format_indexパラメータを指定して、バッチがサンプルまたは機能によってインデックス化可能かどうかを指定できます。ユーザーはカスタムバッチ変換を実装することもできますが、これは必要以上の作業です。

   上記の問題を解決するには、 SiftingBatchTransformモジュールを使用してカスタムバッチ変換クラスを作成する必要があります。バッチ変換クラスは、変換関数と逆変換関数のペアで構成される必要があります。関数ペアは、データ形式を SageMaker スマートふるい分けアルゴリズムが処理できる形式に変換します。バッチ変換クラスを作成すると、クラスはステップ 4 でSiftingDataloaderクラスに渡されるSiftingBatchオブジェクトを返します。

   以下は、SiftingBatchTransformモジュールのカスタムバッチ変換クラスの例です。

   • データローダーチャンクに入力、マスク、ラベルがある場合の SageMaker スマートふるいによるカスタムリストバッチ変換実装の例。

     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

   • 逆変換にラベルが不要な場合に SageMaker スマートふるい分けを使用するカスタムリストバッチ変換実装の例。

     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

   • データローダーチャンクに入力、マスク、ラベルがある場合の SageMaker スマートふるいによるカスタムテンソルバッチ実装の例。

     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

   SiftingBatchTransform実装されたバッチ変換クラスを作成したら、ステップ 4 でこのクラスを使用してSiftingDataloaderクラスを設定します。このガイドの残りの部分では、ListBatchTransformクラスが作成されていることを前提としています。ステップ 4 では、このクラスは に渡されますbatch_transforms。

SageMaker スマートふるい分けLossインターフェイスを実装するためのクラスを作成します。このチュートリアルでは、クラスの名前が であることを前提としていますSiftingImplementedLoss。このクラスを設定するときは、モデルトレーニングループで同じ損失関数を使用することをお勧めします。以下のサブステップに従って、 SageMaker スマートふるいにLoss実装されたクラスを作成します。

1. SageMaker スマートふるいは、バッチの単一の損失値を計算するのではなく、トレーニングデータサンプルごとに損失値を計算します。 SageMaker スマートふるいで同じ損失計算ロジックを使用するようにするには、 smart-sifting-implemented損失関数を使用し、トレーニングサンプルごとに損失を計算する SageMaker スマートふるいLossモジュールを使用して損失関数を作成します。

   ヒント

   SageMaker スマートふるい分けアルゴリズムはバッチ全体ではなくすべてのデータサンプルで実行されるため、初期化関数を追加して、削減戦略なしで PyTorch 損失関数を設定する必要があります。

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

   これは次のコード例にも示されています。

2. とモデルを受け入れる損失関数を定義します original_batch (または、ステップ 2 でバッチ変換を設定transformed_batchしている場合） PyTorch。指定された損失関数をリダクションなしで使用して、 SageMaker スマートふるいは各データサンプルに対してフォワードパスを実行して損失値を評価します。

次のコードは、 という名前のLossインターフェイスの例です smart-sifting-implementedSiftingImplementedLoss。

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])

トレーニングループが実際のフォワードパスに達する前に、このふるい損失計算は、反復ごとにバッチを取得するデータロードフェーズ中に行われます。次に、個々の損失値が以前の損失値と比較され、その相対パーセンタイルはステップ 1 で設定したオブジェクトごとに推定RelativeProbabilisticSiftConfigされます。

クラス別に PyTroch データローダーを SageMaker SiftingDataloaderラップします。

最後に、前のステップで設定したすべての SageMaker スマートふるい分け実装クラスを設定 SageMaker SiftingDataloderクラスに使用します。このクラスは のラッパーです PyTorch DataLoader。 SageMaker をラップ PyTorchすることでDataLoader、スマートふるいは PyTorch トレーニングジョブの各反復でデータロードの一部として実行されるように登録されます。次のコード例は、 への SageMaker 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
)