PyTorch：線形回帰から学ぶ機械学習¶

import torch
from torch import nn
import matplotlib.pyplot as plt

# 使用可能なデバイスを確認（Apple SiliconのMPSまたはCPU）
device = "mps" if torch.backends.mps.is_available() else "cpu"
print(f"PyTorchバージョン: {torch.__version__}")
print(f"使用デバイス: {device}")

# 既知のパラメータを設定（モデルが学習すべき真の値）
weight = 0.7  # 重み（傾き）
bias = 0.3    # バイアス（y切片）

# データの範囲を設定
start = 0
end = 1
step = 0.02

# 線形関係 y = weight * x + bias に従ってデータを生成
X = torch.arange(start, end, step).unsqueeze(dim=1)  # 入力特徴量
y = weight * X + bias  # 出力ラベル

print(f"データ形状: X={X.shape}, y={y.shape}")
print(f"Xの最初の10個の値:\n{X[:10].flatten()}")
print(f"yの最初の10個の値:\n{y[:10].flatten()}")

# 80%を訓練用、20%をテスト用に分割
train_split = int(0.8 * len(X))
X_train, y_train = X[:train_split], y[:train_split]
X_test, y_test = X[train_split:], y[train_split:]

print(f"訓練データ: {len(X_train)}サンプル")
print(f"テストデータ: {len(X_test)}サンプル")

def plot_predictions(train_data=X_train,
                     train_labels=y_train,
                     test_data=X_test,
                     test_labels=y_test,
                     predictions=None):
    """
    訓練データ、テストデータ、予測結果を可視化する関数
    """
    plt.figure(figsize=(10, 7))

    # 訓練データとテストデータをプロット
    plt.scatter(train_data, train_labels, c="b", s=4, label="訓練データ")
    plt.scatter(test_data, test_labels, c="g", s=4, label="テストデータ")

    # 予測結果があればプロット
    if predictions is not None:
        plt.scatter(test_data, predictions, c="r", s=4, label="予測結果")

    plt.legend(prop={"size": 14})
    plt.xlabel("X")
    plt.ylabel("y")
    plt.title("線形回帰データの可視化")

# 初期データの可視化
plot_predictions()
plt.show()

class LinearRegressionModel(nn.Module):
    """
    線形回帰モデル: y = weight * x + bias
    """
    def __init__(self):
        super().__init__()
        # 学習可能なパラメータを定義
        self.weights = nn.Parameter(torch.randn(1, 
                                               requires_grad=True,
                                               dtype=torch.float))
        self.bias = nn.Parameter(torch.randn(1,
                                            requires_grad=True,
                                            dtype=torch.float))

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        フォワードパス: モデルの計算を定義
        """
        return self.weights * x + self.bias

# 再現性のためにランダムシードを設定
torch.manual_seed(42)

# モデルのインスタンスを作成
model_0 = LinearRegressionModel()

# モデルのパラメータを確認
print("モデルのパラメータ:")
for name, param in model_0.named_parameters():
    print(f"{name}: {param.data}")

print(f"\nモデルの状態辞書:\n{model_0.state_dict()}")

# 推論モードで予測を実行
model_0.eval()  # 評価モードに設定
with torch.inference_mode():  # 勾配計算を無効化して高速化
    y_preds = model_0(X_test)

print("訓練前の予測結果:")
print(y_preds[:5].flatten())
print("\n実際の値:")
print(y_test[:5].flatten())

# 予測結果を可視化
plot_predictions(predictions=y_preds)
plt.title("訓練前のモデル予測（ランダムパラメータ）")
plt.show()

# 損失関数：平均絶対誤差（L1Loss）
loss_fn = nn.L1Loss()

# オプティマイザー：確率的勾配降下法（SGD）
optimizer = torch.optim.SGD(params=model_0.parameters(),
                           lr=0.005)  # 学習率

# 訓練の設定
epochs = 500  # エポック数（全データを何回学習するか）

# 進捗追跡用のリスト
epoch_count = []
loss_values = []
test_loss_values = []

print("訓練開始...")
for epoch in range(epochs):
    ### 訓練フェーズ ###
    model_0.train()  # 訓練モードに設定

    # 1. フォワードパス
    y_pred = model_0(X_train)

    # 2. 損失計算
    loss = loss_fn(y_pred, y_train)

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

    # 4. バックプロパゲーション
    loss.backward()

    # 5. パラメータ更新
    optimizer.step()

    ### テストフェーズ ###
    model_0.eval()  # 評価モードに設定

    with torch.inference_mode():
        # テストデータで予測
        test_pred = model_0(X_test)
        # テスト損失を計算
        test_loss = loss_fn(test_pred, y_test)

    # 進捗を記録・表示
    if epoch % 50 == 0:
        epoch_count.append(epoch)
        loss_values.append(loss.detach())  # 勾配情報を切り離して保存
        test_loss_values.append(test_loss.detach())

        print(f"エポック: {epoch:3d} | 訓練損失: {loss:.6f} | テスト損失: {test_loss:.6f}")
        print(f"パラメータ - 重み: {model_0.state_dict()['weights'].item():.4f}, "
              f"バイアス: {model_0.state_dict()['bias'].item():.4f}")

print(f"\n訓練完了！最終パラメータ:")
print(f"重み: {model_0.state_dict()['weights'].item():.4f} (真の値: {weight})")
print(f"バイアス: {model_0.state_dict()['bias'].item():.4f} (真の値: {bias})")

import numpy as np

plt.figure(figsize=(10, 6))
plt.plot(epoch_count, np.array([t.numpy() for t in loss_values]), 
         label="訓練損失", marker='o')
plt.plot(epoch_count, np.array([t.numpy() for t in test_loss_values]), 
         label="テスト損失", marker='s')
plt.title("訓練・テスト損失の変化")
plt.ylabel("損失値")
plt.xlabel("エポック")
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

print(f"真のパラメータ - 重み: {weight}, バイアス: {bias}")

# 訓練済みモデルで予測を実行
model_0.eval()
with torch.inference_mode():
    y_pred_final = model_0(X_test)

print(f"予測結果:")
for i in range(len(y_pred_final)):
    print(f"X={X_test[i].item():.2f} -> 予測={y_pred_final[i].item():.4f}, "
          f"実際={y_test[i].item():.4f}, "
          f"誤差={abs(y_pred_final[i].item() - y_test[i].item()):.4f}")

# 最終的な予測結果を可視化
plot_predictions(predictions=y_pred_final)
plt.title("訓練後のモデル予測結果")
plt.show()

from pathlib import Path

# 1. モデル保存用ディレクトリの作成
MODEL_PATH = Path("models")
MODEL_PATH.mkdir(parents=True, exist_ok=True)

# 2. 保存ファイルパスの設定
MODEL_NAME = "01_pytorch_workflow_model_0.pth"
MODEL_SAVE_PATH = MODEL_PATH / MODEL_NAME

print(f"モデル保存パス: {MODEL_SAVE_PATH}")

# 3. state_dict()を保存（推奨方法）
print(f"モデルを保存中: {MODEL_SAVE_PATH}")
torch.save(obj=model_0.state_dict(), f=MODEL_SAVE_PATH)
print("保存完了！")

# 新しいモデルインスタンスを作成
loaded_model_0 = LinearRegressionModel()

# 保存されたstate_dict()を読み込み
loaded_model_0.load_state_dict(torch.load(f=MODEL_SAVE_PATH))

print("読み込み完了！")
print(f"読み込まれたモデルのパラメータ:\n{loaded_model_0.state_dict()}")

# 予測結果の一致を確認
loaded_model_0.eval()
with torch.inference_mode():
    loaded_model_preds = loaded_model_0(X_test)

# 元のモデルと予測結果が一致するかチェック
predictions_match = torch.allclose(loaded_model_preds, y_pred_final)
print(f"\n予測結果の一致: {predictions_match}")

class LinearRegressionModelV2(nn.Module):
    """
    nn.Linearを使用した線形回帰モデル
    """
    def __init__(self):
        super().__init__()
        # PyTorchの組み込み線形レイヤーを使用
        self.linear_layer = nn.Linear(in_features=1, out_features=1)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.linear_layer(x)

# モデルの作成とデバイス設定
torch.manual_seed(42)
model_1 = LinearRegressionModelV2()
model_1.to(device=device)

print(f"モデルのデバイス: {next(model_1.parameters()).device}")
print(f"初期パラメータ:\n{model_1.state_dict()}")

# デバイス対応の設定
device = "mps" if torch.backends.mps.is_available() else "cpu"
print(f"使用デバイス: {device}")

# データをデバイスに移動
X_train = X_train.to(device)
y_train = y_train.to(device)
X_test = X_test.to(device)
y_test = y_test.to(device)

# 損失関数とオプティマイザーの設定
loss_fn = nn.L1Loss()
optimizer = torch.optim.SGD(params=model_1.parameters(), lr=0.01)

# 訓練ループ
torch.manual_seed(42)
epochs = 200

print("改良版モデルの訓練開始...")
for epoch in range(epochs):
    # 訓練フェーズ
    model_1.train()
    y_pred = model_1(X_train)
    loss = loss_fn(y_pred, y_train)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

    # テストフェーズ
    model_1.eval()
    with torch.inference_mode():
        test_pred = model_1(X_test)
        test_loss = loss_fn(test_pred, y_test)

    # 進捗表示
    if epoch % 20 == 0:
        print(f"エポック: {epoch:3d} | 訓練損失: {loss:.6f} | テスト損失: {test_loss:.6f}")

print(f"\n最終パラメータ:\n{model_1.state_dict()}")

# 最終予測と可視化
model_1.eval()
with torch.inference_mode():
    y_preds_v2 = model_1(X_test)

plot_predictions(predictions=y_preds_v2.cpu())
plt.title("改良版モデル（nn.Linear使用）の予測結果")
plt.show()

# モデルの保存
MODEL_NAME_V2 = "01_pytorch_workflow_model_1.pth"
MODEL_SAVE_PATH_V2 = MODEL_PATH / MODEL_NAME_V2

print(f"改良版モデルを保存中: {MODEL_SAVE_PATH_V2}")
torch.save(obj=model_1.state_dict(), f=MODEL_SAVE_PATH_V2)

# モデルの読み込みと検証
loaded_model_1 = LinearRegressionModelV2()
loaded_model_1.load_state_dict(torch.load(MODEL_SAVE_PATH_V2))
loaded_model_1.to(device=device)

loaded_model_1.eval()
with torch.inference_mode():
    loaded_model_1_preds = loaded_model_1(X_test)

# 予測結果の一致確認
predictions_match_v2 = torch.allclose(y_preds_v2, loaded_model_1_preds)
print(f"改良版モデルの予測結果一致: {predictions_match_v2}")

import numpy as np
import matplotlib.pyplot as plt

# パラメータ設定
x_data = 2.0    # 固定入力値
true_target = 5.0  # 真の目標値
bias = 1.0      # 固定バイアス

# 重みの範囲
w_values = np.linspace(-1, 6, 100)

# MSE損失関数の計算
mse_loss_values = ((w_values * x_data + bias) - true_target)**2

# 勾配の計算
derivative_values = 2 * x_data * (w_values * x_data + bias - true_target)

# 最適な重みの値（解析解）
optimal_w = (true_target - bias) / x_data

plt.figure(figsize=(15, 5))

# MSE損失関数のプロット
plt.subplot(1, 3, 1)
plt.plot(w_values, mse_loss_values, 'b-', linewidth=2, label='MSE損失')
plt.axvline(optimal_w, color='r', linestyle='--', label=f'最適値 w={optimal_w:.1f}')
plt.xlabel('重み (w)')
plt.ylabel('MSE損失')
plt.title('MSE損失関数')
plt.grid(True, alpha=0.3)
plt.legend()

# 勾配のプロット
plt.subplot(1, 3, 2)
plt.plot(w_values, derivative_values, 'r-', linewidth=2, label='勾配 dL/dw')
plt.axhline(0, color='black', linewidth=0.8, linestyle='--', alpha=0.7)
plt.axvline(optimal_w, color='r', linestyle='--', label=f'勾配=0の点')
plt.xlabel('重み (w)')
plt.ylabel('勾配 dL/dw')
plt.title('損失関数の勾配')
plt.grid(True, alpha=0.3)
plt.legend()

# 勾配降下法のシミュレーション
plt.subplot(1, 3, 3)
w_current = 0.5  # 初期重み
learning_rate = 0.1
history_w = [w_current]
history_loss = [((w_current * x_data + bias) - true_target)**2]

for step in range(15):
    # 現在の勾配を計算
    gradient = 2 * x_data * (w_current * x_data + bias - true_target)
    # パラメータを更新
    w_current = w_current - learning_rate * gradient
    # 履歴を記録
    history_w.append(w_current)
    history_loss.append(((w_current * x_data + bias) - true_target)**2)

plt.plot(w_values, mse_loss_values, 'b-', alpha=0.6, label='MSE損失')
plt.plot(history_w, history_loss, 'ro-', markersize=6, label='勾配降下法の軌跡')
plt.axvline(optimal_w, color='g', linestyle='--', label=f'最適解')
plt.xlabel('重み (w)')
plt.ylabel('MSE損失')
plt.title('勾配降下法による最適化')
plt.grid(True, alpha=0.3)
plt.legend()

plt.tight_layout()
plt.show()

print(f"最適な重み（解析解）: {optimal_w:.4f}")
print(f"勾配降下法の最終結果: {history_w[-1]:.4f}")
print(f"収束誤差: {abs(optimal_w - history_w[-1]):.6f}")

def demonstrate_learning_rates():
    """異なる学習率での収束の様子を比較"""
    learning_rates = [0.001, 0.01, 0.1, 0.5]

    plt.figure(figsize=(15, 4))

    for i, lr in enumerate(learning_rates):
        plt.subplot(1, 4, i+1)

        # 簡単な最適化問題を設定
        w_current = 0.1
        target = 2.0
        history = [w_current]

        for step in range(20):
            gradient = 2 * (w_current - target)  # 簡単な二次関数の勾配
            w_current = w_current - lr * gradient
            history.append(w_current)

            if abs(w_current - target) < 1e-6:  # 収束判定
                break

        steps = range(len(history))
        plt.plot(steps, history, 'o-', markersize=4)
        plt.axhline(target, color='r', linestyle='--', alpha=0.7, label='目標値')
        plt.title(f'学習率: {lr}')
        plt.xlabel('ステップ')
        plt.ylabel('パラメータ値')
        plt.grid(True, alpha=0.3)
        plt.legend()

        # 収束性の評価
        if len(history) <= 20:
            convergence_status = f"収束（{len(history)-1}ステップ）"
        else:
            convergence_status = "収束せず"
        plt.text(0.5, 0.9, convergence_status, transform=plt.gca().transAxes, 
                bbox=dict(boxstyle="round,pad=0.3", facecolor="yellow", alpha=0.7))

    plt.tight_layout()
    plt.show()

demonstrate_learning_rates()