PyTorchによるニューラルネットワーク分類¶

import sklearn
from sklearn.datasets import make_circles

# 1000個のサンプルを作成
n_samples = 1000

# 円形データの生成
X, y = make_circles(n_samples = n_samples,
                    noise = 0.03,      # ノイズを追加してより現実的に
                    random_state=42)   # 再現性のため乱数シードを固定
print(type(X))
print(f"最初の5つのX特徴量:\n{X[:5]}")
print(f"\n最初の5つのyラベル:\n{y[:5]}")

# 円形データのDataFrameを作成
import pandas as pd
circles = pd.DataFrame({"X1":X[:,0],
                        "X2":X[:,1],
                        "label":y})
print(circles.head(10))

# データの可視化
import matplotlib.pyplot as plt

plt.scatter(x=X[:,0],
            y=X[:,1],
            c=y,
            cmap=plt.cm.RdYlBu)

print(X.shape, y.shape)
print(type(X))

# 1つのサンプルの特徴量とラベルを確認
X_sample = X[0]
y_sample = y[0]
print(f"1つのサンプルのX値: {X_sample} と対応するy値: {y_sample}")
print(f"1つのサンプルのX形状: {X_sample.shape} と対応するy形状: {y_sample.shape}")

# データをテンソルに変換
import torch

X = torch.from_numpy(X).type(torch.float)
y = torch.from_numpy(y).type(torch.float)

print(X[0],y[0])

# データを訓練セットとテストセットに分割
from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(X,y,test_size=0.2)
print(X_test.shape, y_test.shape)
print(X_train[:5], y_train[:5])
len(X_train),len(X_test),len(y_train),len(y_test)

from torch import nn

# デバイスの設定（M1/M2 MacではMPS、それ以外はCPU）
device = "mps" if torch.mps.is_available() else "cpu"

# 1. nn.Moduleを継承したモデルクラスを構築
class CircleModelV0(nn.Module):
    def __init__(self):
        super().__init__()
        # 2. データの形状を扱える2つのnn.Linear層を作成
        # 2つの特徴量を入力、5つの特徴量を出力
        self.layer_1 = nn.Linear(in_features=2, out_features=5)
        # 5つの特徴量を入力、1つの特徴量（y）を出力
        self.layer_2 = nn.Linear(in_features=5, out_features=1)

    # 3. 順伝播を定義するforward()メソッド
    def forward(self, x):
        return self.layer_2(self.layer_1(x))

# 4. モデルクラスのインスタンスを作成し、ターゲットデバイスに送信
model_0 = CircleModelV0().to(device=device)
next(model_0.parameters()).device

# nn.Sequential()を使用して上記のモデルを複製
model_0 = nn.Sequential(
    nn.Linear(in_features=2, out_features=5),
    nn.ReLU(),
    nn.Linear(in_features=5, out_features=1)
).to(device=device)
model_0.state_dict()

# 予測を行う
with torch.inference_mode():
    untrained_preds = model_0(X_test.to(device))
print(f"予測の長さ: {len(untrained_preds)}, 形状: {untrained_preds.shape}")
print(f"テストサンプルの長さ: {len(X_test)}, 形状: {X_test.shape}")
print(f"最初の10個の予測: {untrained_preds[:10]}")
print(f"最初の10個のラベル: {y_test[:10]}")

# 損失関数を作成
loss_fn = nn.BCEWithLogitsLoss()

# オプティマイザを作成（学習率0.1のSGD）
optimizer = torch.optim.SGD(params=model_0.parameters(),
                            lr=0.1)

# 精度を計算する関数
def accuracy_fn(y_true, y_pred):
    correct = torch.eq(y_true, y_pred).sum().item()
    acc = (correct/len(y_pred)) * 100
    return acc

# テストデータに対する順伝播の最初の5つの出力を表示
y_logits = model_0(X_test.to(device))[:5]
y_logits

# モデルのロジットにシグモイド関数を適用
y_pred_probs = torch.sigmoid(y_logits)
print(y_test[:5], torch.round(y_pred_probs))

# 予測ラベルを見つける（0.5を閾値として丸める）
y_preds = torch.round(y_pred_probs)

# 完全な処理
y_pred_labels = torch.round(torch.sigmoid(model_0(X_test.to(device))[:5]))

# 等価性を確認
print(torch.eq(y_preds.squeeze(), y_pred_labels.squeeze()))

# 余分な次元を削除
y_preds.squeeze()

torch.mps.manual_seed(42)

# エポック数の設定
epochs = 1000

# データをターゲットデバイスに配置
X_train, X_test = X_train.to(device), X_test.to(device)
y_train, y_test = y_train.to(device), y_test.to(device)

# 訓練と評価のループを構築
for epoch in range(epochs):
    # 訓練モード
    model_0.train()

    # 1. 順伝播
    y_logits = model_0(X_train).squeeze()
    y_preds = torch.round(torch.sigmoid(y_logits))

    # 2. 損失と精度の計算
    loss = loss_fn(y_logits, y_train)
    acc = accuracy_fn(y_true=y_train,
                      y_pred=y_preds)

    # 3. オプティマイザの勾配をゼロに
    optimizer.zero_grad()

    # 4. 逆伝播
    loss.backward()

    # 5. オプティマイザのステップ
    optimizer.step()

    # 評価モード
    model_0.eval()
    with torch.inference_mode():
        test_logits = model_0(X_test).squeeze()
        test_preds = torch.round(torch.sigmoid(test_logits))
        test_loss = loss_fn(test_logits, y_test)
        test_acc = accuracy_fn(y_true=y_test,
                               y_pred=test_preds)

    # 進捗を出力
    if epoch % 100 == 0:
        print(f"エポック: {epoch} | 損失: {loss:.5f}, 精度: {acc:.2f}% | テスト損失: {test_loss:.5f}, テスト精度: {test_acc:.2f}%")

from helper_functions import plot_predictions, plot_decision_boundary

# モデルの決定境界をプロット
plt.figure(figsize=(12, 6))
plt.subplot(1, 2, 1)
plt.title("訓練データ")
plot_decision_boundary(model_0, X_train, y_train)
plt.subplot(1, 2, 2)
plt.title("テストデータ")
plot_decision_boundary(model_0, X_test, y_test)

class CircleModelV1(nn.Module):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.layer_1 = nn.Linear(in_features=2, out_features=10)
        self.layer_2 = nn.Linear(in_features=10, out_features=10)
        self.layer_3 = nn.Linear(in_features=10, out_features=1)

    def forward(self, x):
        return self.layer_3(self.layer_2(self.layer_1(x)))

model_1 = CircleModelV1().to(device)
model_1

loss_fn = nn.BCEWithLogitsLoss()
optimizer = torch.optim.SGD(model_1.parameters(), 0.1)
torch.manual_seed(42)
torch.mps.manual_seed(42)

# エポック数の設定
epochs = 1000

# データをターゲットデバイスに配置
X_train, X_test = X_train.to(device), X_test.to(device)
y_train, y_test = y_train.to(device), y_test.to(device)

# 訓練と評価のループを構築
for epoch in range(epochs):
    # 訓練
    model_1.train()
    # 1. 順伝播
    y_logits = model_1(X_train).squeeze()
    y_preds = torch.round(torch.sigmoid(y_logits))
    # 2. 損失/精度の計算
    loss = loss_fn(y_logits, y_train)
    acc = accuracy_fn(y_true=y_train,
                      y_pred=y_preds)
    # 3. オプティマイザの勾配をゼロに
    optimizer.zero_grad()
    # 4. 逆伝播
    loss.backward()
    # 5. オプティマイザのステップ
    optimizer.step()

    # テスト
    model_1.eval()
    with torch.inference_mode():
        test_logits = model_1(X_test).squeeze()
        test_preds = torch.round(torch.sigmoid(test_logits))
        test_loss = loss_fn(test_logits, y_test)
        test_acc = accuracy_fn(y_true=y_test,
                               y_pred=test_preds)

    # 出力
    if epoch % 100 == 0:
        print(f"エポック: {epoch} | 損失: {loss:.5f}, 精度: {acc:.2f}% | テスト損失: {test_loss:.5f}, テスト精度: {test_acc:.2f}%")

# データを作成（ノートブック01と同じ）
weight = 0.7
bias = 0.3
start = 0
end = 1
step = 0.01

# データを作成
X_regression = torch.arange(start, end, step).unsqueeze(dim=1)
y_regression = weight * X_regression + bias # 線形回帰の式

# データを確認
print(len(X_regression))
X_regression[:5], y_regression[:5]

# 訓練とテストセットを作成
train_split = int(0.8 * len(X_regression)) # データの80%を訓練用に使用
X_train_regression, y_train_regression = X_regression[:train_split], y_regression[:train_split]
X_test_regression, y_test_regression = X_regression[train_split:], y_regression[train_split:]

# 各分割の長さを確認
print(len(X_train_regression), 
    len(y_train_regression), 
    len(X_test_regression), 
    len(y_test_regression))

plot_predictions(train_data=X_train_regression,
    train_labels=y_train_regression,
    test_data=X_test_regression,
    test_labels=y_test_regression
)

# model_1と同じアーキテクチャ（ただしnn.Sequentialを使用）
model_2 = nn.Sequential(
    nn.Linear(in_features=1, out_features=10),
    nn.Linear(in_features=10, out_features=10),
    nn.Linear(in_features=10, out_features=1)
).to(device)

model_2

# 損失関数とオプティマイザ
loss_fn = nn.L1Loss()
optimizer = torch.optim.SGD(model_2.parameters(), lr=0.1)

# モデルを訓練
torch.manual_seed(42)

# エポック数を設定
epochs = 1000

# データをターゲットデバイスに配置
X_train_regression, y_train_regression = X_train_regression.to(device), y_train_regression.to(device)
X_test_regression, y_test_regression = X_test_regression.to(device), y_test_regression.to(device)

for epoch in range(epochs):
    ### 訓練 
    # 1. 順伝播
    y_pred = model_2(X_train_regression)

    # 2. 損失を計算（回帰問題なので精度は計算しない）
    loss = loss_fn(y_pred, y_train_regression)

    # 3. オプティマイザの勾配をゼロに
    optimizer.zero_grad()

    # 4. 逆伝播
    loss.backward()

    # 5. オプティマイザのステップ
    optimizer.step()

    ### テスト
    model_2.eval()
    with torch.inference_mode():
      # 1. 順伝播
      test_pred = model_2(X_test_regression)
      # 2. 損失を計算 
      test_loss = loss_fn(test_pred, y_test_regression)

    # 進捗を出力
    if epoch % 100 == 0: 
        print(f"エポック: {epoch} | 訓練損失: {loss:.5f}, テスト損失: {test_loss:.5f}")

# 評価モードに切り替え
model_2.eval()

# 予測を実行（推論）
with torch.inference_mode():
    y_preds = model_2(X_test_regression)

# データと予測をプロット（matplotlibはGPU上のデータを扱えないため、CPUに移動）
plot_predictions(train_data=X_train_regression.cpu(),
                 train_labels=y_train_regression.cpu(),
                 test_data=X_test_regression.cpu(),
                 test_labels=y_test_regression.cpu(),
                 predictions=y_preds.cpu())

# データを作成してプロット
import matplotlib.pyplot as plt
from sklearn.datasets import make_circles

n_sample = 1000
X, y = make_circles(n_samples=1000,
                    noise=0.03,
                    random_state=42)

plt.scatter(X[:,0], X[:,1], c=y, cmap=plt.cm.RdBu)

import torch
from sklearn.model_selection import train_test_split

X = torch.from_numpy(X).type(torch.float)
y = torch.from_numpy(y).type(torch.float)

# 訓練とテストセットに分割
X_train, X_test, y_train, y_test = train_test_split(X, 
                                                    y, 
                                                    test_size=0.2,
                                                    random_state=42
)
X_train[:5], y_train[:5]

from torch import nn
class CircleModelV2(nn.Module):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.layer_1 = nn.Linear(in_features=2, out_features=10)
        self.layer_2 = nn.Linear(in_features=10, out_features=10)
        self.relu = nn.ReLU()  # 非線形活性化関数
        self.layer_3 = nn.Linear(in_features=10, out_features=1)

    def forward(self, x):
        # ReLU活性化関数を層の間に適用
        return self.layer_3(self.relu(self.layer_2(self.relu(self.layer_1(x)))))

model_3 = CircleModelV2().to(device)
model_3

# 損失関数とオプティマイザを設定
loss_fn = nn.BCEWithLogitsLoss()
optimizer = torch.optim.SGD(model_3.parameters(), 0.1)

# モデルを訓練
torch.manual_seed(42)
torch.mps.manual_seed(42)
epochs = 1000

X_train, y_train = X_train.to(device), y_train.to(device)
X_test, y_test = X_test.to(device), y_test.to(device)

for epoch in range(epochs):
    model_3.train()
    y_logits = model_3(X_train).squeeze()
    y_preds = torch.round(torch.sigmoid(y_logits))

    loss = loss_fn(y_logits, y_train)
    acc = accuracy_fn(y_true=y_train,
                      y_pred=y_preds)

    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

    model_3.eval()
    with torch.inference_mode():
        test_logits = model_3(X_test).squeeze()
        test_preds = torch.round(torch.sigmoid(test_logits))
        test_loss = loss_fn(test_logits, y_test)
        test_acc = accuracy_fn(y_true=y_test,
                          y_pred=test_preds)
        if epoch % 100 == 0:
            print(f"エポック: {epoch} | 損失: {loss:.5f}, 精度: {acc:.2f}% | テスト損失: {test_loss:.5f}, テスト精度: {test_acc:.2f}%")

X_train, y_train = X_train.to(device), y_train.to(device)
X_test, y_test = X_test.to(device), y_test.to(device)
model_3.eval()
with torch.inference_mode():
    y_logits = model_3(X_test).squeeze()
    y_sigmoid = torch.sigmoid(y_logits)
    y_preds = torch.round(y_sigmoid).squeeze()

print("ロジット:", y_logits[:10])
print("シグモイド後:", y_sigmoid[:10])
print("予測:", y_preds[:10])
print("正解:", y_test[:10])

from helper_functions import plot_predictions, plot_decision_boundary

# モデルの決定境界をプロット
plt.figure(figsize=(12, 6))
plt.subplot(1, 2, 1)
plt.title("訓練データ")
plot_decision_boundary(model_3, X_train, y_train)
plt.subplot(1, 2, 2)
plt.title("テストデータ")
plot_decision_boundary(model_3, X_test, y_test)

import torch
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs
from sklearn.model_selection import train_test_split

# データ作成のハイパーパラメータを設定
NUM_CLASSES = 4
NUM_FEATURES = 2
RANDOM_SEED = 42

# 1. 多クラスデータを作成
X_blob, y_blob = make_blobs(n_samples=1000,
                            n_features=NUM_FEATURES,
                            centers=NUM_CLASSES,
                            cluster_std=1.5,  # クラスターに少し変動を加える
                            random_state=RANDOM_SEED)

# 2. データをテンソルに変換
X_blob = torch.from_numpy(X_blob).type(torch.float)
y_blob = torch.from_numpy(y_blob).type(torch.LongTensor)

# 3. 訓練とテストセットに分割
X_blob_train, X_blob_test, y_blob_train, y_blob_test = train_test_split(X_blob,
                                                                        y_blob,
                                                                        test_size=0.2,
                                                                        random_state=RANDOM_SEED)

print(X_blob_test[:5])
# データをプロット
plt.figure(figsize=(10, 7))
plt.scatter(X_blob[:, 0], X_blob[:, 1], c=y_blob, cmap=plt.cm.RdBu)

import torch

device = "mps" if torch.mps.is_available() else "cpu"

# モデルを構築
from torch import nn
class BlobModel(nn.Module):
    def __init__(self, input_features, output_features, hidden_units=8):
        """多クラス分類モデルのハイパーパラメータを初期化

        Args:
            input_features (int): モデルへの入力特徴量の数
            output_features (int): モデルの出力特徴量の数（クラス数）
            hidden_units (int): 層間の隠れユニット数、デフォルトは8
        """
        super().__init__()
        self.linear_layer_stack = nn.Sequential(
            nn.Linear(in_features=input_features, out_features=hidden_units),
            nn.ReLU(),
            nn.Linear(in_features=hidden_units, out_features=hidden_units),
            nn.ReLU(),
            nn.Linear(in_features=hidden_units, out_features=output_features)
        )
    def forward(self, x):
        return self.linear_layer_stack(x)

# BlobModelのインスタンスを作成し、ターゲットデバイスに送信
model_4 = BlobModel(input_features=NUM_FEATURES,
                    output_features=NUM_CLASSES,
                    hidden_units=8).to(device=device)
model_4

# 多クラス分類用の損失関数を作成
loss_fn = nn.CrossEntropyLoss()
# オプティマイザを作成
optimizer = torch.optim.SGD(params=model_4.parameters(),
                            lr=0.1)

model_4.eval()
with torch.inference_mode():
    y_logits = model_4(X_blob_test.to(device))[:5]

print("ロジット:", y_logits)
print("正解ラベル:", y_blob_test[:5])

y_pred_probs = torch.softmax(y_logits, dim=1)
print("ロジット:", y_logits[:5])
print("確率:", y_pred_probs[:5])

# 最初のサンプルのソフトマックス出力の合計
print("確率:", y_pred_probs[0])
print("合計:", torch.sum(y_pred_probs[0]))
print("最大値:", torch.max(y_pred_probs[0]))
print("最大値のインデックス（予測クラス）:", torch.argmax(y_pred_probs[0]))

torch.manual_seed(42)
torch.mps.manual_seed(42)

# 多クラス分類用の損失関数を作成
loss_fn = nn.CrossEntropyLoss()
# オプティマイザを作成
optimizer = torch.optim.SGD(params=model_4.parameters(),
                            lr=0.1)

# エポック数を設定
epochs = 100

# データをターゲットデバイスに配置
X_blob_train, y_blob_train = X_blob_train.to(device), y_blob_train.to(device)
X_blob_test, y_blob_test = X_blob_test.to(device), y_blob_test.to(device)

for epoch in range(epochs):
    model_4.train()

    y_logits = model_4(X_blob_train)
    y_pred = torch.softmax(y_logits, dim=1).argmax(dim=1)

    loss = loss_fn(y_logits, y_blob_train)
    acc = accuracy_fn(y_true=y_blob_train,
                      y_pred=y_pred)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

    model_4.eval()
    with torch.inference_mode():
        test_logits = model_4(X_blob_test)
        test_pred = torch.softmax(test_logits, dim=1).argmax(dim=1)

        test_loss = loss_fn(test_logits, y_blob_test)
        test_acc = accuracy_fn(y_true=y_blob_test,
                               y_pred=test_pred)
        if epoch % 10 == 0:
            print(f"エポック: {epoch} | 損失: {loss:.5f}, 精度: {acc:.2f}% | テスト損失: {test_loss:.5f}, テスト精度: {test_acc:.2f}%")

# 予測を実行
model_4.eval()

with torch.inference_mode():
    y_logits = model_4(X_blob_test)

y_pred_probs = torch.softmax(y_logits[:10], dim=1)
print("予測確率:")
print(y_pred_probs)
print("\n予測クラス:", y_pred_probs.argmax(dim=1))
print("正解クラス:", y_blob_test[:10])

plt.figure(figsize=(12, 6))
plt.subplot(1, 2, 1)
plt.title("Train")
plot_decision_boundary(model_4, X_blob_train, y_blob_train)
plt.subplot(1, 2, 2)
plt.title("test")
plot_decision_boundary(model_4, X_blob_test, y_blob_test)

# CrossEntropyLossを使用する際の注意点
# ラベルは長整数型（LongTensor）である必要がある
y_blob_train_long = y_blob_train.long()
y_blob_test_long = y_blob_test.long()

# BCEWithLogitsLossを使用する際は浮動小数点型
y_train_float = y_train.float()
y_test_float = y_test.float()

# モデルの保存
torch.save(model_4.state_dict(), "multi_class_model.pth")

# モデルの読み込み
loaded_model = BlobModel(input_features=NUM_FEATURES,
                        output_features=NUM_CLASSES,
                        hidden_units=8)
loaded_model.load_state_dict(torch.load("multi_class_model.pth"))
loaded_model.to(device)

# 学習率の調整例
learning_rates = [0.001, 0.01, 0.1, 1.0]
best_acc = 0
best_lr = 0

for lr in learning_rates:
    # 新しいモデルとオプティマイザを作成
    test_model = BlobModel(NUM_FEATURES, NUM_CLASSES, 8).to(device)
    test_optimizer = torch.optim.SGD(test_model.parameters(), lr=lr)

    # 簡単な訓練ループ（省略）
    # ...

    # 最良の学習率を記録
    if test_acc > best_acc:
        best_acc = test_acc
        best_lr = lr

print(f"最適な学習率: {best_lr}, 最高精度: {best_acc:.2f}%")

# エラー例: RuntimeError: Expected floating point type for target with class probabilities
# 原因: CrossEntropyLossでラベルがfloat型になっている
# 解決法: ラベルをlong型に変換
y_correct = y_blob_train.long()

# エラー例: RuntimeError: The size of tensor a (4) must match the size of tensor b (1)
# 原因: 出力層のニューロン数とクラス数が一致していない
# 解決法: モデル定義を確認し、output_featuresをクラス数に合わせる

# 学習率が高すぎる場合の対処
# 症状: 損失が振動する、NaNになる
# 解決法: 学習率を下げる
optimizer = torch.optim.SGD(model.parameters(), lr=0.001)  # 0.1から0.001に変更

# 勾配爆発の対処
# 症状: 損失が急激に増加する
# 解決法: 勾配クリッピングを使用
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)