ニューラルネットを使った分類の体験

今回の目標

データサイエンスの学習を進めている人の中には、「理論については理解している一方でどのようにプログラムを組むのかわからない」「とりあえず、どんなものか動かしてみたい」という人が少なくないと思います。

ここでは私の復習も兼ねつつ、ニューラルネットを用いたデータの分類を体験することを目的とし、コピペして実行するだけで処理を行えるサンプルコードを共有したいと思います。

実行環境

Google Colaboratory

サンプルコード

パッケージなど

!pip install mplcyberpunk
!pip install watermark

import mplcyberpunk
import copy
import numpy as np
import pandas as pd
from matplotlib import cm
import matplotlib.pyplot as plt
import torch
from torch import nn, optim
import torch.nn.functional as F
from torch.utils.data import TensorDataset, DataLoader
from functorch import make_functional, vmap, vjp, jvp, jacrev
from sklearn.model_selection import train_test_split
import warnings
warnings.simplefilter('ignore')

分類用の人工データ作成

from sklearn.datasets import make_circles

# サンプル数を1000に設定
num_samples = 1000

# Create circles
X, y = make_circles(num_samples,
                    noise=0.03,
                    random_state=42)

circles = pd.DataFrame({"X0":X[:, 0], "X1":X[:, 1], "label":y})

circles.head()

作成したデータの様子

fig, ax = plt.subplots(1, 1, figsize=(5, 5))
ax.scatter(X[:, 0], X[:, 1], c=y)
ax.set_xlabel('$X1$')
ax.set_ylabel('$X2$')

識別境界をプロットするためのプログラム（最後に使います）

def plot_decision_boundary(model, X, y):
  # Define the axis boundaries of the plot and create a meshgrid
  x_min, x_max = X[:, 0].min() - 0.1, X[:, 0].max() + 0.1
  y_min, y_max = X[:, 1].min() - 0.1, X[:, 1].max() + 0.1
  xx, yy = np.meshgrid(np.linspace(x_min, x_max, 1000),
                       np.linspace(y_min, y_max, 1000))

  x_in = np.c_[xx.ravel(), yy.ravel()]
  x_in = torch.tensor(x_in).to(torch.float32)

  # Make predictions using the trained model
  y_hat= model(x_in)
  y_hat = torch.round(torch.max(y_hat, axis=1)[0]).reshape(xx.shape)

  # Plot decision boundary
  fig, ax = plt.subplots(1, 1, figsize=(5, 5))
  ax.pcolormesh(xx, yy, y_hat.detach().numpy(), alpha=0.7)
  ax.scatter(X[:, 0], X[:, 1], c=y, s=40)
  ax.set_xlim(xx.min(), xx.max())
  ax.set_ylim(yy.min(), yy.max())

分類プログラム（今回のメイン）

# 中間層の幅
width = 3 * 10**3
# エポック数
epochs = 500
# 学習率
learning_rate = 1/100
# バッチサイズ
train_batch_size = int(0.8 * num_samples/2)
# バッチサイズ
val_batch_size = int(0.2 * num_samples/2)

class NeuralNetwork(nn.Module):
    def __init__(self):
        super(NeuralNetwork, self).__init__()
        self.classifier = nn.Sequential(nn.Linear(2, width),
                                       nn.ReLU(inplace=True),
                                       nn.Linear(width, width),
                                       nn.ReLU(inplace=True),
                                       nn.Linear(width, 1),
                                       nn.Sigmoid())           #分類器では出力層の活性化関数をsigmoid関数にする必要
    def forward(self, X):
        y_hat = self.classifier(X)
        return y_hat

model2 = NeuralNetwork()

criterion = nn.BCELoss()
optimizer = optim.SGD(model2.parameters(), lr=learning_rate)

X = torch.tensor(X).to(torch.float32)
y = torch.tensor(y).to(torch.float32)

X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2)

train_loader = DataLoader(TensorDataset(X_train.unsqueeze(1), y_train.unsqueeze(1)), batch_size=train_batch_size,
                          pin_memory=True, shuffle=True)
val_loader = DataLoader(TensorDataset(X_val.unsqueeze(1), y_val.unsqueeze(1)), batch_size=val_batch_size,
                        pin_memory=True, shuffle=True)

train_loss = []
val_loss = []
for epoch in np.arange(epochs):
    # エポック毎にモデルの訓練と評価を繰り返す
    # モデルの訓練
    model2.train()
    temp_train_loss = []
    for X_train, y_train in train_loader:             # データローダーから訓練データ（バッチ）を抽出
        optimizer.zero_grad()                         # 勾配情報を初期化
        y_hat = model2(X_train).squeeze(1)            # 出力の予測値を生成
        loss = criterion(input=y_hat, target=y_train) # 損失関数に出力の予測値と正解の出力を代入し訓練誤差を計算
        loss.backward()                               # パラメータ毎に訓練誤差の勾配を計算
        optimizer.step()                              # 勾配降下法によるパラメータ更新
        temp_train_loss.append(loss.detach())
    train_loss.append(np.average(temp_train_loss))    # 訓練誤差をリストに保存
    # モデルの評価
    model2.eval()
    temp_val_loss = []
    for X_val, y_val in val_loader:
        y_hat = model2(X_val).squeeze(1)              # データローダーからテストデータ（バッチ）を抽出
        loss = criterion(input=y_hat, target=y_val)   # 損失関数に出力の予測値と正解の出力を代入して汎化誤差の近似値を計算
        temp_val_loss.append(loss.detach())
    val_loss.append(np.average(temp_val_loss))        # 汎化誤差の近似値をリストに保存

plt.style.use("cyberpunk")
plt.plot(train_loss, label='Train')
plt.plot(val_loss, c='magenta', label='Validation')
plt.ylabel('loss')
plt.xlabel('epoch')
plt.title("Learning rate %f"%(learning_rate))
plt.legend()
plt.show()

初めて見ると何をやっているのかさっぱりなコードですが、画像にあるように「分類の誤差を小さくするようにたくさんパラメータ更新をしている」と理解できれば最初はOKです。

最後に学習結果としての識別境界を見てみましょう。

plot_decision_boundary(model2, X, y)

どのようになったでしょうか？


中間層の幅やエポック数そして学習率は学習の精度に大きく影響します。数値を小さくしたり大きくしたりして試してみると面白いです。


ただ今回のコードを実行した際、私の環境では約9分かかりました。時間には余裕を持っておくといいでしょう。