PyTorch を使用したモデルのトレーニング

7 分

PyTorch は、ディープラーニングモデルのトレーニングによく使用される機械学習フレームワークです。 Azure Databricks には、PyTorch が ML クラスターにプレインストールされています。

Note

このユニットのコードスニペットは、要点を強調するための例として提供されています。完全に機能する例のコードは、このモジュールの後の演習で実行する機会があります。

PyTorch ネットワークを定義する

PyTorch のモデルは、定義するネットワークに基づきます。ネットワークは、指定した入力と出力をそれぞれ持つ複数のレイヤーで構成されます。さらに、この作業で、データがネットワークを通過する際に各レイヤーに関数を適用する forward 関数を定義します。

ネットワークを定義するコード例を次に示します。

import torch
import torch.nn as nn
import torch.nn.functional as F

class MyNet(nn.Module):
    def __init__(self):
        super(MyNet, self).__init__()
        self.layer1 = nn.Linear(4, 5)
        self.layer2 = nn.Linear(5, 5)
        self.layer3 = nn.Linear(5, 3)

    def forward(self, x):
        layer1_output = torch.relu(self.layer1(x))
        layer2_output = torch.relu(self.layer2(layer1_output))
        y = self.layer3(layer2_output)
        return y

コードは最初は複雑に見えるかもしれませんが、このクラスでは、レイヤーが 3 つの比較的シンプルなネットワークを定義しています。

4 つの入力値を受け入れ、次のレイヤーに対して 5 つの出力値を生成する入力レイヤー。
5 つの入力を受け入れ、5 つの出力を生成するレイヤー。
5 つの入力を受け入れ、3 つの出力を生成する最終レイヤー。

forward 関数は、これらのレイヤーを入力データ (x) に適用し、各レイヤーの出力を次のレイヤーに渡し、最後に最終レイヤー (ラベル予測ベクトル y を含む) から出力を返します。レイヤー 1 と 2 の出力に "正規化線形ユニット" (ReLU) 活性化関数が適用され、出力値が正の数に制限されています。

Note

使用する損失条件の種類に応じて、戻り値に log_softmax などの活性化関数を適用して、0 から 1 の範囲に強制するように選択できます。ただし、一部の損失条件 (マルチクラス分類によく使用される CrossEntropyLoss など) では、適切な関数が自動的に適用されます。

トレーニング用のモデルの作成に必要なのは、次のようにネットワーククラスのインスタンスを作成することだけです。

myModel = MyNet()

モデリング用にデータを準備する

PyTorch レイヤーは、"テンソル" (行列のような構造) としてフォーマットされたデータで機能します。その他の一般的なデータ形式をテンソルに変換する関数が各種あり、PyTorch "データローダー" を定義して、トレーニングまたは推論用のモデルにデータテンソルを読み取ることができます。

ほとんどの教師あり機械学習手法と同様に、トレーニングと検証用に個別のデータセットを定義する必要があります。このように分離することで、トレーニングされていないデータが提示されたとき、モデルで正確に予測されることを検証できます。

2 つのデータローダーを定義するコードを次に示します。1 つはトレーニング用、もう 1 つはテスト用です。この例の各ローダーのソースデータは、特徴値の Numpy 配列と、対応するラベル値の Numpy 配列が想定されています。

# Create a dataset and loader for the training data and labels
train_x = torch.Tensor(x_train).float()
train_y = torch.Tensor(y_train).long()
train_ds = td.TensorDataset(train_x,train_y)
train_loader = td.DataLoader(train_ds, batch_size=20,
    shuffle=False, num_workers=1)

# Create a dataset and loader for the test data and labels
test_x = torch.Tensor(x_test).float()
test_y = torch.Tensor(y_test).long()
test_ds = td.TensorDataset(test_x,test_y)
test_loader = td.DataLoader(test_ds, batch_size=20,
    shuffle=False, num_workers=1)

この例では、ローダーによって、データが 30 個のバッチに分割され、トレーニングまたは推論中に forward 関数に渡されます。

損失条件とオプティマイザーアルゴリズムを選択する

モデルは、トレーニングデータをネットワークにフィードし、損失 (予測値と実測値の差の合計) を測定し、損失が最小になるように重みとバランスを調整してネットワークを最適化することでトレーニングされます。損失の計算方法と最小化方法の具体的な詳細は、選択する損失条件とオプティマイザーアルゴリズムによって制御されます。

損失条件

PyTorch では、以下を始め、複数の損失条件関数がサポートされています。

cross_entropy: 複数の変数の予測値と実測値の差の合計を測定する関数 (通常は、マルチクラス分類のクラス確率の損失を測定するために使用されます)。
binary_cross_entropy: 予測確率と実際の確率の差を測定する関数 (通常は、二項分類のクラス確率の損失を測定するために使用されます)。
mse_loss: 予測数値と実際の数値の平均二乗誤差損失を測定する関数 (通常は、回帰に使用されます)。

モデルのトレーニング時に使用する損失条件を指定するには、次のように、適切な関数のインスタンスを作成します。

import torch.nn as nn

loss_criteria = nn.CrossEntropyLoss

ヒント

PyTorch で使用できる損失条件の詳細については、PyTorch ドキュメントの「損失関数」を参照してください。

オプティマイザーアルゴリズム

損失を計算したら、オプティマイザーを使用して、損失を最小限に抑えるために重みとバランスを最適に調整する方法を決定します。オプティマイザーとは、関数を最小化するための "勾配降下" 法を具体的に実装したものです。 PyTorch で使用できるオプティマイザーには、以下のものがあります。

Adadelta: 適応学習率アルゴリズムに基づくオプティマイザー。
Adam: Adam アルゴリズムに基づく計算効率の高いオプティマイザー。
SGD: 確率的勾配降下アルゴリズムに基づくオプティマイザー。

これらのアルゴリズムのいずれかを使用してモデルをトレーニングするには、オプティマイザーのインスタンスを作成し、必須パラメーターを設定する必要があります。特定のパラメーターは選択したオプティマイザーによって異なりますが、ほとんどの場合、最適化ごとに行われる調整のサイズを制御する学習率を指定する必要があります。

Adam オプティマイザーのインスタンスを作成するコードを次に示します。

import torch.optim as opt

learning_rate = 0.001
optimizer = opt.Adam(model.parameters(), lr=learning_rate)

ヒント

PyTorch で使用可能なオプティマイザーの詳細については、PyTorch ドキュメントの「アルゴリズム」を参照してください。

トレーニング関数とテスト関数を作成する

ネットワークを定義し、そのためのデータを準備したら、トレーニングデータをネットワーク経由で渡し、損失を計算し、ネットワークの重みと偏りを最適化する、テストデータを使用してネットワークのパフォーマンスを検証することで、モデルのトレーニングとテストにデータを使用できます。データをネットワーク経由で渡し、トレーニングデータを使用してモデルを "トレーニング" する関数と、テストデータを使用してモデルを "テスト" する別の関数を定義するのが一般的です。

トレーニング関数を作成する

モデルをトレーニングする関数の例を次に示します。

def train(model, data_loader, optimizer):

    # Use GPU if available, otherwise CPU
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    
    # Set the model to training mode (to enable backpropagation)
    model.train()
    train_loss = 0
    
    # Feed the batches of data forward through the network
    for batch, tensor in enumerate(data_loader):
        data, target = tensor # Specify features and labels in a tensor
        optimizer.zero_grad() # Reset optimizer state
        out = model(data) # Pass the data through the network
        loss = loss_criteria(out, target) # Calculate the loss
        train_loss += loss.item() # Keep a running total of loss for each batch

        # backpropagate adjustments to weights/bias
        loss.backward()
        optimizer.step()

    #Return average loss for all batches
    avg_loss = train_loss / (batch+1)
    print('Training set: Average loss: {:.6f}'.format(avg_loss))
    return avg_loss

モデルをテストする関数の例を次に示します。

def test(model, data_loader):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    # Switch the model to evaluation mode (so we don't backpropagate)
    model.eval()
    test_loss = 0
    correct = 0

    # Pass the data through with no gradient computation
    with torch.no_grad():
        batch_count = 0
        for batch, tensor in enumerate(data_loader):
            batch_count += 1
            data, target = tensor
            # Get the predictions
            out = model(data)

            # calculate the loss
            test_loss += loss_criteria(out, target).item()

            # Calculate the accuracy
            _, predicted = torch.max(out.data, 1)
            correct += torch.sum(target==predicted).item()
            
    # Calculate the average loss and total accuracy for all batches
    avg_loss = test_loss/batch_count
    print('Validation set: Average loss: {:.6f}, Accuracy: {}/{} ({:.0f}%)\n'.format(
        avg_loss, correct, len(data_loader.dataset),
        100. * correct / len(data_loader.dataset)))
    return avg_loss

モデルのトレーニングを複数エポック実行する

ディープラーニングモデルをトレーニングするには、通常、トレーニング関数を複数回 ("エポック" と呼ばれます) 実行します。エポックごとにトレーニングデータから計算される損失を減らすことが目的です。テスト関数を使用して、(モデルがトレーニングされていない) テストデータの損失もトレーニングの損失に合わせて減少している、つまり、モデルトレーニングによって、トレーニングデータに "オーバーフィット" したモデルが生成されていないことを検証できます。

ヒント

エポックごとにテスト関数を実行する必要はありません。 1 エポックおきに実行することも、最後に 1 回実行することも選択できます。ただし、トレーニング中にモデルをテストすると、何エポック後にモデルがオーバーフィットになり始めるかを判断するのに役立つことがあります。

モデルのトレーニングを 50 エポック実行するコードを次に示します。

epochs = 50
for epoch in range(1, epochs + 1):

    # print the epoch number
    print('Epoch: {}'.format(epoch))
    
    # Feed training data into the model to optimize the weights
    train_loss = train(model, train_loader, optimizer)
    print(train_loss)
    
    # Feed the test data into the model to check its performance
    test_loss = test(model, test_loader)
    print(test_loss)

トレーニング済みモデルの状態を保存する

モデルのトレーニングが正常に完了したら、重みと偏りを次のように保存できます。

model_file = '/dbfs/my_model.pkl'
torch.save(model.state_dict(), model_file)

後でモデルを読み込んで使用するには、モデルの基になるネットワーククラスのインスタンスを作成し、保存した重みと偏りを読み込みます。

model = myNet()
model.load_state_dict(torch.load(model_file))