はじめに

前回は、PoTorchでニューラルネットワークを実装し、MNIST を使って多クラス分類をしました。今回は、畳み込みNNを実装し、CIFAR-10の分類を行います。

できたこと

データセット: CIFAR-10
- 3x32x32
- 10クラス
- 50,000枚の訓練画像, 10,000枚の検証用画像
機械学習モデル: CNN
- 畳み込み
  - 位置の移動に無関係な特徴量を抽出
- プーリング
  - Max Pooling
  - 物体の大きさによらない普遍的な特徴量を抽出
活性化関数: ReLU
損失関数: 交差エントロピー関数
最適化手法: SGD

分類結果のイメージ

学習曲線

できていないこと

チューニングはほとんどやっていないので、きっちり過学習しています。伸びしろですね。

ライブラリ

%matplotlib inline
import numpy as np
import matplotlib.pyplot as plt
import japanize_matplotlib
from IPython.display import display

# PyTorch関連ライブラリ
import torch
import torch.nn as nn
import torch.optim as optim
from torchinfo import summary
from torchviz import make_dot
import torchvision.datasets as datasets # データセット
import torchvision.transforms as transforms # 前処理
from torch.utils.data import DataLoader # データローダー
from tqdm.notebook import tqdm # プログレスバー

# warning表示off
import warnings
warnings.simplefilter('ignore')
# デフォルトフォントサイズ変更
plt.rcParams['font.size'] = 14
# デフォルトグラフサイズ変更
plt.rcParams['figure.figsize'] = (6,6)
# デフォルトで方眼表示ON
plt.rcParams['axes.grid'] = True
# numpyの表示桁数設定
np.set_printoptions(suppress=True, precision=5)

GPUチェック

# デバイスの割り当て
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(device)

前処理の設定

一階テンソル化（一階化関数: flatten)ありバージョンとなしバージョンがありますが、本記事では前者しか使ってないはずです。

# transform1: 一階テンソル化有り
transform1 = transforms.Compose([
    transforms.ToTensor(), # テンソル化
    transforms.Normalize(0.5, 0.5), # 正規化
    transforms.Lambda(lambda x: x.view(-1)), # 一階テンソル化
])

# transform2: 一階テンソル化無し
transform2 = transforms.Compose([
    transforms.ToTensor(), # テンソル化
    transforms.Normalize(0.5, 0.5), # 正規化
])

データ準備

CIFAR-10をダウンロードします。

# ダウンロード先ディレクトリ名
data_root = './ignore_dir/data'

# 訓練データセット 1階テンソル版
train_set1 = datasets.CIFAR10(
    root = data_root, train = True, 
    download = True, transform = transform1)

# 検証データセット 1階テンソル版
test_set1 = datasets.CIFAR10(
    root = data_root, train = False, 
    download = True, transform = transform1)

# 訓練データセット 3階テンソル版
train_set2 = datasets.CIFAR10(
    root =  data_root, train = True, 
    download = True, transform = transform2)

# 検証データセット 3階テンソル版
test_set2 = datasets.CIFAR10(
    root = data_root, train = False, 
    download = True, transform = transform2)

訓練データが5万件、検証データが1万件です。確認は割愛します。

データローダー

訓練用のデータローダーはシャッフルありですが、検証用のデータローダーはシャッフルなしです。

# ミニバッチのサイズ指定
batch_size = 100
# 訓練用データローダー
train_loader1 = DataLoader(train_set1, batch_size=batch_size, shuffle=True) # 訓練用なので、シャッフルをかける
# 検証用データローダー
test_loader1 = DataLoader(test_set1,  batch_size=batch_size, shuffle=False) # 検証時にシャッフルは不要
# 訓練用データローダー
train_loader2 = DataLoader(train_set2, batch_size=batch_size, shuffle=True) # 訓練用なので、シャッフルをかける
# 検証用データローダー
test_loader2 = DataLoader(test_set2,  batch_size=batch_size, shuffle=False)  # 検証時にシャッフルは不要

イメージ確認

現物確認はしておいた方がよいと思うので、実装します。まずはデータローダーを使って、最初の1セット（50個）を取得します。

# train_loader1から1セット取得
for images1, labels1 in train_loader1:
    break

# train_loader2から1セット取得
for images2, labels2 in train_loader2:
    break

# それぞれのshape確認
print(images1.shape)
print(images2.shape)

show_images_labelsという関数名でまとめておきます。

# イメージとラベル表示のための関数
def show_images_labels(loader, classes, net, device):

    # データローダーから最初の1セットを取得する
    for images, labels in loader:
        break
    # 表示数は50個とバッチサイズのうち小さい方
    n_size = min(len(images), 50)

    if net is not None:
      # デバイスの割り当て
      inputs = images.to(device)
      labels = labels.to(device)

      # 予測計算
      outputs = net(inputs)
      predicted = torch.max(outputs,1)[1]
      #images = images.to('cpu')

    # 最初のn_size個の表示
    plt.figure(figsize=(20, 15))
    for i in range(n_size):
        ax = plt.subplot(5, 10, i + 1)
        label_name = classes[labels[i]]
        # netがNoneでない場合は、予測結果もタイトルに表示する
        if net is not None:
          predicted_name = classes[predicted[i]]
          # 正解かどうかで色分けをする
          if label_name == predicted_name:
            c = 'k'
          else:
            c = 'b'
          ax.set_title(label_name + ':' + predicted_name, c=c, fontsize=20)
        # netがNoneの場合は、正解ラベルのみ表示
        else:
          ax.set_title(label_name, fontsize=20)
        # TensorをNumPyに変換
        image_np = images[i].numpy().copy()
        # 軸の順番変更 (channel, row, column) -> (row, column, channel)
        img = np.transpose(image_np, (1, 2, 0))
        # 値の範囲を[-1, 1] -> [0, 1]に戻す
        img = (img + 1)/2
        # 結果表示
        plt.imshow(img)
        ax.set_axis_off()
    plt.show()

正解ラベルを定義し、イメージを表示します。

# 正解ラベル定義
classes = ('plane', 'car', 'bird', 'cat',
           'deer', 'dog', 'frog', 'horse', 'ship', 'truck')

# 検証データ最初の50個の表示
show_images_labels(test_loader2, classes, None, None)

1000画素程度なので、だいぶ粗いです。鹿と馬とか見分けるのは大変そう。

モデル定義

冒頭で書いた通り、畳み込みニューラルネットワークを実装します。畳み込み層は2つで、全結合層は1つです。プーリング層はmax pooling、活性関数はReLUです。

class CNN(nn.Module):
  def __init__(self, n_output, n_hidden):
    super().__init__()
    self.conv1 = nn.Conv2d(3, 32, 3)
    self.conv2 = nn.Conv2d(32, 32, 3)
    self.relu = nn.ReLU(inplace=True)
    self.maxpool = nn.MaxPool2d((2,2))
    self.flatten = nn.Flatten()
    self.l1 = nn.Linear(6272, n_hidden)
    self.l2 = nn.Linear(n_hidden, n_output)

    self.features = nn.Sequential(
        self.conv1,
        self.relu,
        self.conv2,
        self.relu,
        self.maxpool)
    
    self.classifier = nn.Sequential(
       self.l1,
       self.relu,
       self.l2)

  def forward(self, x):
    x1 = self.features(x)
    x2 = self.flatten(x1)
    x3 = self.classifier(x2)
    return x3

パラメータ設定＆モデル概要確認

その他のハイパーパラメータを設定し、モデルの概要を確認しておきます。まずは乱数固定から。

# PyTorch乱数固定用

def torch_seed(seed=123):
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.backends.cudnn.deterministic = True
    torch.use_deterministic_algorithms = True

# 乱数初期化
torch_seed()

次は入出力の次元数確認と、モデルインスタンスの確認を行います。

# 出力次元数
# 分類先クラス数　今回は10になる
n_output = len(classes)

# 隠れ層のノード数
n_hidden = 128

# 結果確認
print(f'n_hidden: {n_hidden} n_output: {n_output}')

# モデルインスタンス生成
net = CNN(n_output, n_hidden).to(device)

学習率なども設定します。

# 学習率
lr = 0.01

# 損失関数： 交差エントロピー関数
criterion = nn.CrossEntropyLoss()


# 最適化関数: 勾配降下法
optimizer = torch.optim.SGD(net.parameters(), lr=lr)

# 繰り返し回数
num_epochs = 50

# 評価結果記録用
history2 = np.zeros((0,5))

モデルの概要を確認します。

# モデルの概要表示
print(net)

CNN(
  (conv1): Conv2d(3, 32, kernel_size=(3, 3), stride=(1, 1))
  (conv2): Conv2d(32, 32, kernel_size=(3, 3), stride=(1, 1))
  (relu): ReLU(inplace=True)
  (maxpool): MaxPool2d(kernel_size=(2, 2), stride=(2, 2), padding=0, dilation=1, ceil_mode=False)
  (flatten): Flatten(start_dim=1, end_dim=-1)
  (l1): Linear(in_features=6272, out_features=128, bias=True)
  (l2): Linear(in_features=128, out_features=10, bias=True)
  (features): Sequential(
    (0): Conv2d(3, 32, kernel_size=(3, 3), stride=(1, 1))
    (1): ReLU(inplace=True)
    (2): Conv2d(32, 32, kernel_size=(3, 3), stride=(1, 1))
    (3): ReLU(inplace=True)
    (4): MaxPool2d(kernel_size=(2, 2), stride=(2, 2), padding=0, dilation=1, ceil_mode=False)
  )
  (classifier): Sequential(
    (0): Linear(in_features=6272, out_features=128, bias=True)
    (1): ReLU(inplace=True)
    (2): Linear(in_features=128, out_features=10, bias=True)
  )
)

# モデルのサマリー表示
summary(net,(100,3,32,32),depth=1)

==========================================================================================
Layer (type:depth-idx)                   Output Shape              Param #
==========================================================================================
CNN                                      [100, 10]                 --
├─Sequential: 1-1                        [100, 32, 14, 14]         10,144
├─Sequential: 1-4                        --                        (recursive)
├─Sequential: 1-5                        --                        (recursive)
├─Sequential: 1-4                        --                        (recursive)
├─Sequential: 1-5                        --                        (recursive)
├─Flatten: 1-6                           [100, 6272]               --
├─Sequential: 1-7                        [100, 10]                 804,234
==========================================================================================
Total params: 823,626
Trainable params: 823,626
Non-trainable params: 0
Total mult-adds (Units.MEGABYTES): 886.11
==========================================================================================
Input size (MB): 1.23
Forward/backward pass size (MB): 43.22
Params size (MB): 3.26
Estimated Total Size (MB): 47.71
==========================================================================================

パラメータ数は82万個。単回帰から始めたことを考えると、ずいぶん感慨深いですが、近年のLLMと比べると軽いです。

訓練

訓練用のコードをfitという関数名でまとめます。

# 学習用関数
def fit(net, optimizer, criterion, num_epochs, train_loader, test_loader, device, history):

    # tqdmライブラリのインポート
    from tqdm.notebook import tqdm

    base_epochs = len(history)
  
    for epoch in range(base_epochs, num_epochs+base_epochs):
        # 1エポックあたりの正解数(精度計算用)
        n_train_acc, n_val_acc = 0, 0
        # 1エポックあたりの累積損失(平均化前)
        train_loss, val_loss = 0, 0
        # 1エポックあたりのデータ累積件数
        n_train, n_test = 0, 0

        #訓練フェーズ
        net.train()

        for inputs, labels in tqdm(train_loader):
            # 1バッチあたりのデータ件数
            train_batch_size = len(labels)
            # 1エポックあたりのデータ累積件数
            n_train += train_batch_size
    
            # GPUヘ転送
            inputs = inputs.to(device)
            labels = labels.to(device)

            # 勾配の初期化
            optimizer.zero_grad()

            # 予測計算
            outputs = net(inputs)

            # 損失計算
            loss = criterion(outputs, labels)

            # 勾配計算
            loss.backward()

            # パラメータ修正
            optimizer.step()

            # 予測ラベル導出
            predicted = torch.max(outputs, 1)[1]

            # 平均前の損失と正解数の計算
            # lossは平均計算が行われているので平均前の損失に戻して加算
            train_loss += loss.item() * train_batch_size 
            n_train_acc += (predicted == labels).sum().item() 

        #予測フェーズ
        net.eval()

        for inputs_test, labels_test in test_loader:
            # 1バッチあたりのデータ件数
            test_batch_size = len(labels_test)
            # 1エポックあたりのデータ累積件数
            n_test += test_batch_size

            # GPUヘ転送
            inputs_test = inputs_test.to(device)
            labels_test = labels_test.to(device)

            # 予測計算
            outputs_test = net(inputs_test)

            # 損失計算
            loss_test = criterion(outputs_test, labels_test)
 
            # 予測ラベル導出
            predicted_test = torch.max(outputs_test, 1)[1]

            #  平均前の損失と正解数の計算
            # lossは平均計算が行われているので平均前の損失に戻して加算
            val_loss +=  loss_test.item() * test_batch_size
            n_val_acc +=  (predicted_test == labels_test).sum().item()

        # 精度計算
        train_acc = n_train_acc / n_train
        val_acc = n_val_acc / n_test
        # 損失計算
        avg_train_loss = train_loss / n_train
        avg_val_loss = val_loss / n_test
        # 結果表示
        print (f'Epoch [{(epoch+1)}/{num_epochs+base_epochs}], loss: {avg_train_loss:.5f} acc: {train_acc:.5f} val_loss: {avg_val_loss:.5f}, val_acc: {val_acc:.5f}')
        # 記録
        item = np.array([epoch+1, avg_train_loss, train_acc, avg_val_loss, val_acc])
        history = np.vstack((history, item))
    return history

記録した学習ログ(history)から、最初と最後の損失＆精度を表示し、学習曲線を描画するためのコードを実装します。関数名はevaluate_historyとします。

# 学習ログ解析

def evaluate_history(history):
    #損失と精度の確認
    print(f'初期状態: 損失: {history[0,3]:.5f} 精度: {history[0,4]:.5f}') 
    print(f'最終状態: 損失: {history[-1,3]:.5f} 精度: {history[-1,4]:.5f}' )

    num_epochs = len(history)
    unit = num_epochs / 10

    # 学習曲線の表示 (損失)
    plt.figure(figsize=(9,8))
    plt.plot(history[:,0], history[:,1], 'b', label='訓練')
    plt.plot(history[:,0], history[:,3], 'k', label='検証')
    plt.xticks(np.arange(0,num_epochs+1, unit))
    plt.xlabel('繰り返し回数')
    plt.ylabel('損失')
    plt.title('学習曲線(損失)')
    plt.legend()
    plt.show()

    # 学習曲線の表示 (精度)
    plt.figure(figsize=(9,8))
    plt.plot(history[:,0], history[:,2], 'b', label='訓練')
    plt.plot(history[:,0], history[:,4], 'k', label='検証')
    plt.xticks(np.arange(0,num_epochs+1,unit))
    plt.xlabel('繰り返し回数')
    plt.ylabel('精度')
    plt.title('学習曲線(精度)')
    plt.legend()
    plt.show()

いよいよ訓練を開始します。

# 学習
history2 = fit(net, optimizer, criterion, num_epochs, train_loader2, test_loader2, device, history2)

プログレスバーが表示されます。

手元のPC（GPUなし）では30分程度かかったので、訓練済みのモデルを保存しておきます。

# 訓練済みモデル保存
net.to('cpu')
params = net.state_dict()
torch.save(params, 'ignore_dir/trained_model/model_ch09_cnn.prm')
net.to(device) # GPUに送る

結果確認

evaluate_history関数を使って、結果を確認します。

# 評価
evaluate_history(history2)

精度は65%まで到達したみたいです。

初期状態: 損失: 1.85420 精度: 0.35170
最終状態: 損失: 1.91180 精度: 0.65520

学習曲線も表示されるはずです。エポック数が25を過ぎたあたりから損失が増大しています。過学習しています。冒頭でも書いた通り、正則化を行っていないので、納得の結果です。

イメージも確認します。

# 最初の50個の表示
show_images_labels(test_loader2, classes, net, device)

おわりに

PyTorchを使って畳み込みニューラルネットワークを実装しました
過学習気味、20回程度で止めておくべきだったと考えられる
精度は65%程度まで到達
- 記事にはしていないが、全結合版だと精度は53%程度だったので、畳み込みによる効果は実感できた
まだまだ誤認識が多い
次はチューニングを行う

出典