0. はじめに

本記事では，PyTorchを用いた基本的な実装を書き纏めておきます(備忘録も兼ねて)．CIFAR10(カラー画像の分類セット)の分類を例に．

更新：
2021/06/11　「モジュールに別のネットワークを設定する」節の追加．備忘録として．
2022/06/13　利用しやすいよう，githubにコードをアップロードしました．以降の更新はgithubメインで行います．
https://github.com/Hsat-ppp/PyTorch_classifier

1. 掲載しているプログラムについて

1.1. MITライセンス

MIT License
Copyright (c) 2022 Hsat-ppp

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

1.2. プログラム全景

プログラムの全景は以下の通り．

.py

import torch
import torchvision
import torchvision.transforms as transforms
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import numpy as np

ave = 0.5               # 正規化平均
std = 0.5               # 正規化標準偏差
batch_size_train = 256  # 学習バッチサイズ
batch_size_test = 16    # テストバッチサイズ
val_ratio = 0.2         # データ全体に対する検証データの割合
epoch_num = 30          # 学習エポック数

class Net(nn.Module):
    # ネットワーク構造の定義
    def __init__(self):
        super(Net, self).__init__()
        self.init_conv = nn.Conv2d(3,16,3,padding=1)
        self.conv1 = nn.ModuleList([nn.Conv2d(16,16,3,padding=1) for _ in range(3)])
        self.bn1 = nn.ModuleList([nn.BatchNorm2d(16) for _ in range(3)])
        self.pool = nn.MaxPool2d(2, stride=2)
        self.fc1 = nn.ModuleList([nn.Linear(16*16*16, 128), nn.Linear(128, 32)])
        self.output_fc = nn.Linear(32, 10)

    # 順方向計算
    def forward(self, x):
        x = F.relu(self.init_conv(x))
        for l,bn in zip(self.conv1, self.bn1):
            x = F.relu(bn(l(x)))
        x = self.pool(x)
        x = x.view(-1,16*16*16) # flatten
        for l in self.fc1:
            x = F.relu(l(x))
        x = self.output_fc(x)
        return x

def set_GPU():
    # GPUの設定
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    print(device)
    return device

def load_data():
    # データのロード
    transform = transforms.Compose([transforms.ToTensor(),transforms.Normalize((ave,),(std,))])
    train_set = torchvision.datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
    test_set = torchvision.datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)

    # 検証データのsplit
    n_samples = len(train_set)
    val_size = int(n_samples * val_ratio)
    train_set, val_set = torch.utils.data.random_split(train_set, [(n_samples-val_size), val_size])

    # DataLoaderの定義
    train_loader = torch.utils.data.DataLoader(train_set, batch_size=batch_size_train, shuffle=True, num_workers=2)
    val_loader = torch.utils.data.DataLoader(val_set, batch_size=batch_size_train, shuffle=False, num_workers=2)
    test_loader = torch.utils.data.DataLoader(test_set, batch_size=batch_size_test, shuffle=False, num_workers=2)

    return train_loader, test_loader, val_loader

def train():
    device = set_GPU()
    train_loader, test_loader, val_loader = load_data()
    model = Net()
    model.to(device)

    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, verbose=True)

    min_loss = 999999999
    print("training start")
    for epoch in range(epoch_num):
        train_loss = 0.0
        val_loss = 0.0
        train_batches = 0
        val_batches = 0
        model.train()   # 訓練モード
        for i, data in enumerate(train_loader):   # バッチ毎に読み込む
            inputs, labels = data[0].to(device), data[1].to(device) # data は [inputs, labels] のリスト

            # 勾配のリセット
            optimizer.zero_grad()

            outputs = model(inputs)    # 順方向計算
            loss = criterion(outputs, labels)   # 損失の計算
            loss.backward()                     # 逆方向計算(勾配計算)
            optimizer.step()                    # パラメータの更新

            # 履歴の累積
            train_loss += loss.item()
            train_batches += 1

        # validation_lossの計算
        model.eval()    # 推論モード
        with torch.no_grad():
            for i, data in enumerate(val_loader):   # バッチ毎に読み込む
                inputs, labels = data[0].to(device), data[1].to(device) # data は [inputs, labels] のリスト
                outputs = model(inputs)               # 順方向計算
                loss = criterion(outputs, labels)   # 損失の計算

                # 履歴の累積
                val_loss += loss.item()
                val_batches += 1

        # 履歴の出力
        print('epoch %d train_loss: %.10f' %
              (epoch + 1,  train_loss/train_batches))
        print('epoch %d val_loss: %.10f' %
              (epoch + 1,  val_loss/val_batches))

        with open("history.csv",'a') as f:
            print(str(epoch+1) + ',' + str(train_loss/train_batches) + ',' + str(val_loss/val_batches),file=f)

        # 最良モデルの保存
        if min_loss > val_loss/val_batches:
            min_loss = val_loss/val_batches
            PATH = "best.pth"
            torch.save(model.state_dict(), PATH)

        # 学習率の動的変更
        scheduler.step(val_loss/val_batches)

    # 最終エポックのモデル保存
    print("training finished")
    PATH = "lastepoch.pth"
    torch.save(model.state_dict(), PATH)

if __name__ == "__main__":
    train()

2. 各実装の解説

2.0. ライブラリのインポートと定数の定義

本記事で使うライブラリは大体以下の通り．

.py

import torch
import torchvision
import torchvision.transforms as transforms
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import numpy as np

後々使う定数も定義しておく．

.py

ave = 0.5               # 正規化平均
std = 0.5               # 正規化標準偏差
batch_size_train = 256  # 学習バッチサイズ
batch_size_test = 16    # テストバッチサイズ
val_ratio = 0.2         # データ全体に対する検証データの割合
epoch_num = 30          # 学習エポック数

2.1. GPUの設定

各種設定に先立って，GPUを使うならdeviceの指定が必要．

.py

def set_GPU():
    # GPUの設定
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
    print(device)
    return device

このdeviceというオブジェクトを通じてGPUを使用する．例えば，

.py

data.to(device)
model.to(device)

とすることで，データやニューラルネットワークモデルをGPUに載せることができる．

2.2 データの準備

PyTorchにはいくつかのデータセットが既に用意されている．例えば，CIFAR10なら，以下のように準備できる．

.py

def load_data():
    # データのロード
    transform = transforms.Compose([transforms.ToTensor(),transforms.Normalize((ave,),(std,))])
    train_set = torchvision.datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
    test_set = torchvision.datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)

    # 検証データのsplit
    n_samples = len(train_set)
    val_size = int(n_samples * val_ratio)
    train_set, val_set = torch.utils.data.random_split(train_set, [(n_samples-val_size), val_size])

    # DataLoaderの定義
    train_loader = torch.utils.data.DataLoader(train_set, batch_size=batch_size_train, shuffle=True, num_workers=2)
    val_loader = torch.utils.data.DataLoader(val_set, batch_size=batch_size_train, shuffle=False, num_workers=2)
    test_loader = torch.utils.data.DataLoader(test_set, batch_size=batch_size_test, shuffle=False, num_workers=2)

    return train_loader, test_loader, val_loader

順番に説明する．まず，transformはデータを変換する機能を持った一連の処理を表す．

.py

    transform = transforms.Compose([transforms.ToTensor(),transforms.Normalize((ave,),(std,))])

上の例では，transforms.ToTensor()でデータをTensor(PyTorchのデータ型)に変えたのち，transforms.Normalize((ave,),(std,))で平均ave,標準偏差stdの正規化を行っている．なお，Composeは一連の処理をまとめる役割をしている．
他にもデータ変換にはいくつかの種類がある．ドキュメントを参照のこと．

次に，CIFAR10のデータを読み込む．

.py

    train_set = torchvision.datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
    test_set = torchvision.datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)

rootにはダウンロード先を指定する．PyTorchの既存データセットはtrainとtestで分けられているらしい．trainというオプションで指定する．ここで，先ほど定義したtransformを引数に渡すことでデータの変換を行うように設定する．

読み込んだデータから検証用データを分離したい．そのために，torch.utils.data.random_splitを使う．

.py

    train_set, val_set = torch.utils.data.random_split(train_set, [(n_samples-val_size), val_size])

第二引数で指定した数にデータをランダムに分けてくれる．

PyTorchで学習を行う場合，DataLoaderを使うと学習時とても便利．以下のように作る．

.py

    train_loader = torch.utils.data.DataLoader(train_set, batch_size=batch_size_train, shuffle=True, num_workers=2)
    val_loader = torch.utils.data.DataLoader(val_set, batch_size=batch_size_train, shuffle=False, num_workers=2)
    test_loader = torch.utils.data.DataLoader(test_set, batch_size=batch_size_test, shuffle=False, num_workers=2)

第一引数にデータセットを渡す．batch_sizeはバッチサイズ，shuffleはデータのシャッフルの有無，num_workersは読み込み時のsubprocess数(並列数)を指定している．
ちなみに，DataLoaderの正体はイテレータだったりする．なので，学習時にはfor文でバッチ毎にデータを取り出すことになる．

自作のデータを使いたい場合

自作のデータを使いたい場面がよくある．この場合，以下のように自分でデータセットクラスを定義するとよい．(参考：https://qiita.com/mathlive/items/2a512831878b8018db02)

.py

class MyDataset(torch.utils.data.Dataset):
    def __init__(self, data, label, transform=None):
        self.transform = transform
        self.data = data
        self.data_num = len(data)
        self.label = label

    def __len__(self):
        return self.data_num

    def __getitem__(self, idx):
        out_data = self.data[idx]
        out_label =  self.label[idx]
        if self.transform:
            out_data = self.transform(out_data)

        return out_data, out_label

少なくとも，クラス内にはlen(データの大きさを返す関数)とgetitem(データを取得する関数)を定義する．これを使って，

.py

dataset = MyDataset(入力データ, 教師ラベル, transform=必要なデータ変換)

のようにデータセットを作ることができる．
余談だが，MyDatasetの中身を見るとPyTorchが中で何をやっているのかが何となくわかる．つまり，idxが指定されたとき，そのインデックスにあたるデータを返している．しかも，transformが何かしら指定されている場合はそれをデータに施している．
getitemをいじることで様々な動作をさせることもできる(例えば，2つのtransformを使うなど)が，割愛．

2.3. ニューラルネットワークの作成

PyTorchではクラスを使ってニューラルネットワークを定義するのが楽．CIFAR10の分類器なら，例えば以下のようになる．

.py

class Net(nn.Module):
    # ネットワーク構造の定義
    def __init__(self):
        super(Net, self).__init__()
        self.init_conv = nn.Conv2d(3,16,3,padding=1)
        self.conv1 = nn.ModuleList([nn.Conv2d(16,16,3,padding=1) for _ in range(3)])
        self.bn1 = nn.ModuleList([nn.BatchNorm2d(16) for _ in range(3)])
        self.pool = nn.MaxPool2d(2, stride=2)
        self.fc1 = nn.ModuleList([nn.Linear(16*16*16, 128), nn.Linear(128, 32)])
        self.output_fc = nn.Linear(32, 10)

    # 順方向計算
    def forward(self, x):
        x = F.relu(self.init_conv(x))
        for l,bn in zip(self.conv1, self.bn1):
            x = F.relu(bn(l(x)))
        x = self.pool(x)
        x = x.view(-1,16*16*16) # flatten
        for l in self.fc1:
            x = F.relu(l(x))
        x = self.output_fc(x)
        return x

色々解説してみる．
###ネットワークの構築
ネットワークはnn.Moduleを継承して作る．
init内に各層を自身のメンバとして定義する．

.py

    def __init__(self):
        super(Net, self).__init__()
        self.init_conv = nn.Conv2d(3,16,3,padding=1)
        self.conv1 = nn.ModuleList([nn.Conv2d(16,16,3,padding=1) for _ in range(3)])
        self.bn1 = nn.ModuleList([nn.BatchNorm2d(16) for _ in range(3)])
        self.pool = nn.MaxPool2d(2, stride=2)
        self.fc1 = nn.ModuleList([nn.Linear(16*16*16, 128), nn.Linear(128, 32)])
        self.output_fc = nn.Linear(32, 10)

例えば，nn.Conv2dは以下のような引数の渡し方をする．

.py

nn.Conv2d(入力のチャネル数，出力のチャネル数，カーネルサイズ，padding=パディングサイズ，stride=移動量)

詳細は公式ドキュメントを参照．(https://pytorch.org/docs/stable/nn.html)

ちょっとしたテクニック

ネットワークの層は，nn.ModuleListを用いると配列として定義することもできる．

.py

        self.conv1 = nn.ModuleList([nn.Conv2d(16,16,3,padding=1) for _ in range(3)])

大規模かつ繰り返し構造を持つネットワークを定義する際などに特に便利．
なお，nn.ModuleListを使わずに普通の配列にしてしまうと，パラメータの更新ができないらしい．(参考：https://qiita.com/perrying/items/857df46bb6cdc3047bd8)
ちゃんとnn.ModuleListを使うようにする．

モジュールに別のネットワークを設定する

ModuleListと同様に，自分で定義したネットワーククラスを別のネットワークに組み込むことができる．すなわち，

.py

class SubNet(nn.Module):
    """ネットワークの定義"""

class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.subnet = SubNet()
        self.subnet_to_output = nn.Linear(......)

という形．
さらに，これをModuleListと組み合わせることもでき，

.py

nn.ModuleList([SubNet() for _ in range(5)])

と書くことで，自作のネットワークもnn.Linearなどと同様に繰り返し組み込むことができる．これを活用することで記述量を削減できる．

順方向計算の定義

順方向計算はforwardとして定義する．

.py

    def forward(self, x):
        x = F.relu(self.init_conv(x))
        for l,bn in zip(self.conv1, self.bn1):
            x = F.relu(bn(l(x)))
        x = self.pool(x)
        x = x.view(-1,16*16*16) # flatten
        for l in self.fc1:
            x = F.relu(l(x))
        x = self.output_fc(x)
        return x

nn.ModuleListで配列化した層はfor文で取り出している．こういうところも便利．
また，途中で

.py

        x = x.view(-1,16*16*16) # flatten

とある．ここでは，画像状のデータを1次元ベクトルに変換している(第二引数に，チャネル数×画像の縦幅×画像の横幅を渡す)．第一引数を-1としているのは，バッチサイズに合わせて自動的に変換を行うため．

2.4. 損失関数と更新手法

損失関数はtorch.nnに，更新手法はtorch.optimにそれぞれ定義されており，これを呼び出して使う．今回は分類を行うため，損失関数にはCrossEntropyLossを使用する．また，更新手法にはAdamを使用する．

.py

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

動的な学習率の設定

学習を効率的に行うために，学習率を動的に設定(というか，削減)したいことがある．その場合，lr_schedulerというものを使う．例えば，

.py

scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, verbose=True)

という風にschedulerを定義する．これを用いると，検証データへの損失を計算した後に，

.py

scheduler.step(val_loss)

と記述することで，(patience)エポックの間に改善が起きなかった場合，学習率を自動的に減らしてくれる．これにより，学習の停滞を防ぐことができる．

自作損失関数

データセット同様，損失関数も自分で作りたいことがある．これについてはPyTorchのクラスを継承して作っても，単なる関数として定義してもよいらしい．(参考：https://kento1109.hatenablog.com/entry/2018/08/13/092939)
単純な回帰・分類のタスクであればMSELoss・CrossEntropyLossで上手くいくことが多いが，機械学習系の論文だと損失関数に工夫を加えて性能を上げていたりするため，結構大事な実装．

2.5. 実際の学習

ここまで準備してから，いよいよ学習を行う．
まず，今までに定義した各設定を適用する．

.py

    device = set_GPU()
    train_loader, test_loader, val_loader = load_data()
    model = Net()
    model.to(device)

    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5, verbose=True)

学習は以下のようになる．基本的にはコメントアウトしてある通り．

.py

    min_loss = 999999999
    print("training start")
    for epoch in range(epoch_num):
        train_loss = 0.0
        val_loss = 0.0
        train_batches = 0
        val_batches = 0
        model.train()   # 訓練モード
        for i, data in enumerate(train_loader):   # バッチ毎に読み込む
            inputs, labels = data[0].to(device), data[1].to(device) # data は [inputs, labels] のリスト

            # 勾配のリセット
            optimizer.zero_grad()

            outputs = model(inputs)    # 順方向計算
            loss = criterion(outputs, labels)   # 損失の計算
            loss.backward()                     # 逆方向計算(勾配計算)
            optimizer.step()                    # パラメータの更新

            # 履歴の累積
            train_loss += loss.item()
            train_batches += 1

        # validation_lossの計算
        model.eval()    # 推論モード
        with torch.no_grad():
            for i, data in enumerate(val_loader):   # バッチ毎に読み込む
                inputs, labels = data[0].to(device), data[1].to(device) # data は [inputs, labels] のリスト
                outputs = model(inputs)               # 順方向計算
                loss = criterion(outputs, labels)   # 損失の計算

                # 履歴の累積
                val_loss += loss.item()
                val_batches += 1

        # 履歴の出力
        print('epoch %d train_loss: %.10f' %
              (epoch + 1,  train_loss/train_batches))
        print('epoch %d val_loss: %.10f' %
              (epoch + 1,  val_loss/val_batches))

        with open("history.csv",'a') as f:
            print(str(epoch+1) + ',' + str(train_loss/train_batches) + ',' + str(val_loss/val_batches),file=f)

        # 最良モデルの保存
        if min_loss > val_loss/val_batches:
            min_loss = val_loss/val_batches
            PATH = "best.pth"
            torch.save(model.state_dict(), PATH)

        # 学習率の動的変更
        scheduler.step(val_loss/val_batches)

    # 最終エポックのモデル保存
    print("training finished")
    PATH = "lastepoch.pth"
    torch.save(model.state_dict(), PATH)

PyTorchでは損失関数の計算や誤差の逆伝搬なども明示的に記述する必要がある(これをラップアップするライブラリも存在する)．

学習したモデルのテストなどについては公式チュートリアル(https://pytorch.org/tutorials/beginner/blitz/cifar10_tutorial.html
)に載っている．このモデルの場合，各クラスのAccuracyは40~80%程度になる(結構ブレが大きい...)．

trainモードとevalモードについて

BatchNormalizationやDropoutに関しては学習時と推論時で振る舞いを切り替える必要がある。
そのため、学習時と推論時で、

.py

model.train()
model.eval()

という記述をそれぞれ加える必要があるらしい。
また、これとは別に、torch.no_grad()というモードも存在している。こちらはgradient(勾配)情報を保存しないモード。検証時にはbackward計算をしないためgradient情報は不要で、これを省略することで計算速度が上がったり、省メモリになったりする。

3. 終わりに

結構途中で力尽きた感があるので，そのうち書き足すかもしれません．
参考になれば幸いです．

4. 参考文献