PyTorchによる多クラス分類の実装

1. 概要

PyTorchはKerasのようにネットワーク層を定義して、fitさせるだけで学習できるものではなく、いろいろとやることがあるので、最初は取っ掛かりにくいと思います。

私もだいぶ苦戦した記憶があります。

なので、irisデータを使って基本的な多クラス分類を実際にやってみて、PyTorchのモデル化の流れを紹介したいと思います。

最後に全体のコードの載せていますので、流れだけでも理解してもらえると嬉しいです。

PyTorch Lightningを使用して多クラス分類を実装した記事もあるので、良ければ参考にしながら見ていただければと思います。

venoda.hatenablog.com

2. モデル化の流れ

PyTorchは次の流れでモデル化していけば大きく間違えることはないかと思います。

1. 準備
   1. データ準備と前処理
   2. Datasetの作成
   3. DataLoaderの作成
2. ネットワークの作成
   1. ネットワークの定義
   2. 損失関数の定義
   3. 最適化手法の定義
3. 学習と予測
   1. 学習の実行
   2. 予測の実行

複雑なステップに感じるかもしれませんが、それぞれ難しいところはPyTorchがやってくれるので

やることといえばそれぞれの処理を定義するだけで事足りてしまいます。

では流れに沿って実装していきたいと思います。

3. 準備

3.1. データ準備と前処理

まずはSklearnのirisデータを読み込み、学習データと検証データに分けます。

その後、PyTorchに入力できるようにTensor型にデータを変換します。

import pandas as pd
from sklearn import datasets
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import torch

# データ読み込み
iris = datasets.load_iris()
data = iris['data']
target = iris['target']

# 学習データと検証データに分割
x_train, x_valid, y_train, y_valid = train_test_split(data, target, shuffle=True)

# 特徴量の標準化
scaler = StandardScaler()
scaler.fit(x_train)

x_train = scaler.transform(x_train)
x_valid = scaler.transform(x_valid)

# Tensor型に変換
# 学習に入れるときはfloat型 or long型になっている必要があるのここで変換してしまう
x_train = torch.from_numpy(x_train).float()
y_train = torch.from_numpy(y_train).long()
x_valid = torch.from_numpy(x_valid).float()
y_valid = torch.from_numpy(y_valid).long()

print('x_train : ', x_train.shape)
print('y_train : ', y_train.shape)
print('x_valid : ', x_valid.shape)
print('y_valid : ', y_valid.shape)

# Output
x_train :  torch.Size([112, 4])
y_train :  torch.Size([112])
x_valid :  torch.Size([38, 4])
y_valid :  torch.Size([38])

3.2. Datasetの作成

PyTorchのTensorDatasetを使って説明変数と目的変数をワンセットにしたDatasetを作成します。

from torch.utils.data import TensorDataset

train_dataset = TensorDataset(x_train, y_train)
valid_dataset = TensorDataset(x_valid, y_valid)

# 動作確認
# indexを指定すればデータを取り出すことができます。
index = 0
print(train_dataset.__getitem__(index)[0].size())
print(train_dataset.__getitem__(index)[1])

# Output
torch.Size([4])
tensor([1, 0, 0], dtype=torch.uint8)

3.3. DataLoaderの作成

バッチ処理を適用するためにDataLoaderを作成して、データセットをバッチ単位で取り出せるようにします。

from torch.utils.data import DataLoader

batch_size = 32
train_dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
valid_dataloader = DataLoader(valid_dataset, batch_size=batch_size, shuffle=False)

# 動作確認
# こんな感じでバッチ単位で取り出す子ができます。
# イテレータに変換
batch_iterator = iter(train_dataloader)
# 1番目の要素を取り出す
inputs, labels = next(batch_iterator)
print(inputs.size())
print(labels.size())

# Output
torch.Size([32, 4])
torch.Size([32])

4. ネットワークの作成

4.1. ネットワークの定義

ネットワークを定義するために、nn.Moduleを継承したクラスを作成します。

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

class Net(nn.Module):    
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(4, 50)
        self.fc2 = nn.Linear(50, 3)
    
    def forward(self, x):
        x = self.fc1(x)
        x = F.relu(x)
        x = self.fc2(x)
        x = F.softmax(x, dim=1)
        return x

net = Net()
print(net)

# Output
Net(
  (fc1): Linear(in_features=4, out_features=50, bias=True)
  (fc2): Linear(in_features=50, out_features=3, bias=True)
)

4.2. 損失関数の定義

損失関数を定義します。

今回はクロスエントロピー誤差を使用します。

import torch.nn as nn

criterion = nn.CrossEntropyLoss()

4.3. 最適化手法の定義

最適化手法を定義します。

最適化していくパラメータとハイパーパラメータを指定していきます。

最適化していくパラメータはnet.parameters()で渡します。

import torch.optim as optim

optimizer = optim.SGD(net.parameters(), lr=0.01)

5. 学習と予測

5.1. 学習の実行

PyTorchでは学習時推論時でネットワークのモードを分ける必要があります。

「net.train()」「net.eval()」でそれぞれのモードを分ける処理を書いています。

# エポック数
num_epochs = 50

# 学習時と検証時で分けるためディクショナリを用意
dataloaders_dict = {
    'train': train_dataloader,
    'val': valid_dataloader
}

for epoch in range(num_epochs):
    print('Epoch {}/{}'.format(epoch+1, num_epochs))
    print('-------------')
    
    for phase in ['train', 'val']:
        
        if phase == 'train':
            # モデルを訓練モードに設定
            net.train()
        else:
            # モデルを推論モードに設定
            net.eval()
        
        # 損失和
        epoch_loss = 0.0
        # 正解数
        epoch_corrects = 0
        
        # DataLoaderからデータをバッチごとに取り出す
        for inputs, labels in dataloaders_dict[phase]:
            
            # optimizerの初期化
            optimizer.zero_grad()
            
            # 学習時のみ勾配を計算させる設定にする
            with torch.set_grad_enabled(phase == 'train'):
                outputs = net(inputs)
                
                # 損失を計算
                loss = criterion(outputs, labels)
                
                # ラベルを予測
                _, preds = torch.max(outputs, 1)
                
                # 訓練時はバックプロパゲーション
                if phase == 'train':
                    # 逆伝搬の計算
                    loss.backward()
                    # パラメータの更新
                    optimizer.step()
                
                # イテレーション結果の計算
                # lossの合計を更新
                # PyTorchの仕様上各バッチ内での平均のlossが計算される。
                # データ数を掛けることで平均から合計に変換をしている。
                # 損失和は「全データの損失/データ数」で計算されるため、
                # 平均のままだと損失和を求めることができないため。
                epoch_loss += loss.item() * inputs.size(0)
                
                # 正解数の合計を更新
                epoch_corrects += torch.sum(preds == labels.data)

        # epochごとのlossと正解率を表示
        epoch_loss = epoch_loss / len(dataloaders_dict[phase].dataset)
        epoch_acc = epoch_corrects.double() / len(dataloaders_dict[phase].dataset)

        print('{} Loss: {:.4f} Acc: {:.4f}'.format(phase, epoch_loss, epoch_acc))

# Output
Epoch 1/50
-------------
train Loss: 1.0259 Acc: 0.6696
val Loss: 1.0283 Acc: 0.6316
Epoch 2/50
-------------
train Loss: 1.0217 Acc: 0.6696
val Loss: 1.0244 Acc: 0.6316

～～～～～～～～

Epoch 49/50
-------------
train Loss: 0.8726 Acc: 0.8839
val Loss: 0.8803 Acc: 0.8947
Epoch 50/50
-------------
train Loss: 0.8705 Acc: 0.8839
val Loss: 0.8780 Acc: 0.8947

以上でPyTorchでのモデル化の流れが完了です。

5.2. 予測の実行

予測の実行は、学習したモデルに入れてやれば予測結果を取得することができます。

# 予測用のダミーデータ
x = torch.randn([3, 4])
preds = net(x)

リンク

おすすめ記事 venoda.hatenablog.com

6. 全体のコード

最後に全体のコードをのせておきます。

import numpy as np
from sklearn import datasets
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

import torch
import torch.nn.functional as F
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

# 1. データ準備、前処理
# データ読み込み
iris = datasets.load_iris()
data = iris['data']
target = iris['target']

# 学習データと検証データに分割
x_train, x_valid, y_train, y_valid = train_test_split(data, target, shuffle=True)

# 特徴量の標準化
scaler = StandardScaler()
scaler.fit(x_train)

x_train = scaler.transform(x_train)
x_valid = scaler.transform(x_valid)

# Tensor型に変換
# 学習に入れるときはfloat型になっている必要があるのここで変換してしまう
x_train = torch.from_numpy(x_train).float()
y_train = torch.from_numpy(y_train).long()
x_valid = torch.from_numpy(x_valid).float()
y_valid = torch.from_numpy(y_valid).long()

print('x_train : ', x_train.shape)
print('y_train : ', y_train.shape)
print('x_valid : ', x_valid.shape)
print('y_valid : ', y_valid.shape)

# 2. Datasetの作成
train_dataset = TensorDataset(x_train, y_train)
valid_dataset = TensorDataset(x_valid, y_valid)

# 動作確認
# indexを指定すればデータを取り出すことができます。
index = 0
print(train_dataset.__getitem__(index)[0].size())
print(train_dataset.__getitem__(index)[1])

# 3. DataLoaderの作成
batch_size = 32
train_dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
valid_dataloader = DataLoader(valid_dataset, batch_size=batch_size, shuffle=False)

# 動作確認
# こんな感じでバッチ単位で取り出す子ができます。
# イテレータに変換
batch_iterator = iter(train_dataloader)
# 1番目の要素を取り出す
inputs, labels = next(batch_iterator)
print(inputs.size())
print(labels.size())

# 4. ネットワークの定義
class Net(nn.Module):    
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(4, 50)
        self.fc2 = nn.Linear(50, 3)
    
    def forward(self, x):
        x = self.fc1(x)
        x = F.relu(x)
        x = self.fc2(x)
        x = F.softmax(x, dim=1)
        return x

net = Net()
print(net)

# 5. 損失関数の定義
criterion = nn.CrossEntropyLoss()

# 6. 最適化手法の定義
optimizer = optim.SGD(net.parameters(), lr=0.01)

# 7. 学習・評価
# エポック数
num_epochs = 50

# 学習時と検証時で分けるためディクショナリを用意
dataloaders_dict = {
    'train': train_dataloader,
    'val': valid_dataloader
}

for epoch in range(num_epochs):
    print('Epoch {}/{}'.format(epoch+1, num_epochs))
    print('-------------')
    
    for phase in ['train', 'val']:
        
        if phase == 'train':
            # モデルを訓練モードに設定
            net.train()
        else:
            # モデルを推論モードに設定
            net.eval()
        
        # 損失和
        epoch_loss = 0.0
        # 正解数
        epoch_corrects = 0
        
        # DataLoaderからデータをバッチごとに取り出す
        for inputs, labels in dataloaders_dict[phase]:
            
            # optimizerの初期化
            optimizer.zero_grad()
            
            # 学習時のみ勾配を計算させる設定にする
            with torch.set_grad_enabled(phase == 'train'):
                outputs = net(inputs)
                
                # 損失を計算
                loss = criterion(outputs, labels)
                
                # ラベルを予測
                _, preds = torch.max(outputs, 1)
                
                # 訓練時はバックプロパゲーション
                if phase == 'train':
                    # 逆伝搬の計算
                    loss.backward()
                    # パラメータの更新
                    optimizer.step()
                
                # イテレーション結果の計算
                # lossの合計を更新
                # PyTorchの仕様上各バッチ内での平均のlossが計算される。
                # データ数を掛けることで平均から合計に変換をしている。
                # 損失和は「全データの損失/データ数」で計算されるため、
                # 平均のままだと損失和を求めることができないため。
                epoch_loss += loss.item() * inputs.size(0)
                
                # 正解数の合計を更新
                epoch_corrects += torch.sum(preds == labels.data)

        # epochごとのlossと正解率を表示
        epoch_loss = epoch_loss / len(dataloaders_dict[phase].dataset)
        epoch_acc = epoch_corrects.double() / len(dataloaders_dict[phase].dataset)

        print('{} Loss: {:.4f} Acc: {:.4f}'.format(phase, epoch_loss, epoch_acc))