フレームワーク

PyTorch

概要

PyTorchは、Facebookが開発した動的計算グラフを特徴とする機械学習フレームワークです。

詳細

PyTorch（パイトーチ）は2016年にFacebook（現Meta）によって開発された、動的ニューラルネットワークと自動微分を特徴とするオープンソース機械学習ライブラリです。Pythonライクな直感的なAPIと動的計算グラフ（Define-by-Run）により、研究開発に適した柔軟性を提供します。NumPyとの高い親和性、GPUアクセラレーション、分散学習サポートなどが主要な特徴です。学術界や研究機関で特に人気が高く、プロトタイピングから本格的な研究まで幅広く使用されています。torchvision、torchaudio、torchtextなどの専門ライブラリエコシステム、TorchScriptによる本番環境デプロイメント、PyTorch Lightningによる高レベル抽象化などにより、研究から実用まで対応可能です。直感的なPythonic APIとデバッグのしやすさから、機械学習の学習・実験・研究開発において多くの開発者に選ばれています。

メリット・デメリット

メリット

• 直感的なAPI: Pythonらしい自然な記述でコードが書きやすい
• 動的計算グラフ: 実行時にグラフを構築し、柔軟なモデル設計が可能
• デバッグしやすさ: 標準的なPythonデバッガーでステップ実行が可能
• 研究向け: 実験的なアーキテクチャの実装に適している
• NumPy互換: NumPyとの相互変換が簡単で学習コストが低い
• 豊富なコミュニティ: 学術界での採用により、最新研究の実装が豊富
• GPU最適化: CUDAとの統合による効率的なGPU利用

デメリット

• プロダクション展開: TensorFlowと比較して本番環境のツールが限定的
• モバイル対応: モバイルデバイスでの実行サポートが TensorFlow Lite より限定的
• パフォーマンス: 動的な性質により、静的グラフより実行速度が遅い場合がある
• バージョン互換性: 新しいバージョンでAPIが変更されることがある
• 学習リソース: TensorFlowと比較して入門向け教材が少ない

主要リンク

• PyTorch公式サイト
• PyTorch公式ドキュメント
• PyTorch GitHub リポジトリ
• PyTorch チュートリアル
• PyTorch Lightning
• PyTorch Hub

書き方の例

Hello World

import torch
import torch.nn as nn
import torch.nn.functional as F

# PyTorchのバージョン確認
print(f"PyTorch version: {torch.__version__}")

# 基本的なテンソル操作
x = torch.tensor([[1., 2., 3.], [4., 5., 6.]])
print("テンソル x:")
print(x)

# GPU利用可能かチェック
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用デバイス: {device}")

# テンソルをGPUに移動（GPUが利用可能な場合）
if torch.cuda.is_available():
    x = x.to(device)
    print("テンソルをGPUに移動しました")

# 基本的な演算
y = x * 2
z = torch.matmul(x, x.T)
print(f"x * 2:\n{y}")
print(f"行列積:\n{z}")

簡単なニューラルネットワーク

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

# デバイスの設定
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# シンプルなニューラルネットワークの定義
class SimpleNet(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(SimpleNet, self).__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.fc2 = nn.Linear(hidden_size, hidden_size)
        self.fc3 = nn.Linear(hidden_size, output_size)
        self.dropout = nn.Dropout(0.2)
    
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = self.dropout(x)
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x

# モデルの初期化
model = SimpleNet(10, 64, 2).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# サンプルデータの作成
X = torch.randn(1000, 10)
y = torch.randint(0, 2, (1000,))
dataset = TensorDataset(X, y)
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)

# 学習ループ
model.train()
for epoch in range(100):
    total_loss = 0
    for batch_X, batch_y in dataloader:
        batch_X, batch_y = batch_X.to(device), batch_y.to(device)
        
        optimizer.zero_grad()
        outputs = model(batch_X)
        loss = criterion(outputs, batch_y)
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
    
    if epoch % 20 == 0:
        print(f"エポック {epoch}: 損失 = {total_loss:.4f}")

CNNによる画像分類

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
from torch.utils.data import DataLoader

# データの前処理
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
])

# CIFAR-10データセットの読み込み
trainset = torchvision.datasets.CIFAR10(
    root='./data', train=True, download=True, transform=transform
)
trainloader = DataLoader(trainset, batch_size=32, shuffle=True)

# CNNモデルの定義
class CNN(nn.Module):
    def __init__(self):
        super(CNN, self).__init__()
        self.conv1 = nn.Conv2d(3, 32, 3, padding=1)
        self.conv2 = nn.Conv2d(32, 64, 3, padding=1)
        self.conv3 = nn.Conv2d(64, 64, 3, padding=1)
        self.pool = nn.MaxPool2d(2, 2)
        self.fc1 = nn.Linear(64 * 4 * 4, 512)
        self.fc2 = nn.Linear(512, 10)
        self.dropout = nn.Dropout(0.5)
    
    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))
        x = self.pool(F.relu(self.conv2(x)))
        x = self.pool(F.relu(self.conv3(x)))
        x = x.view(-1, 64 * 4 * 4)
        x = F.relu(self.fc1(x))
        x = self.dropout(x)
        x = self.fc2(x)
        return x

# モデル、損失関数、オプティマイザーの設定
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = CNN().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 学習
model.train()
for epoch in range(10):
    running_loss = 0.0
    for i, (inputs, labels) in enumerate(trainloader):
        inputs, labels = inputs.to(device), labels.to(device)
        
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        
        running_loss += loss.item()
        
        if i % 200 == 199:
            print(f'[{epoch + 1}, {i + 1:5d}] loss: {running_loss / 200:.3f}')
            running_loss = 0.0

カスタムデータセット

import torch
from torch.utils.data import Dataset, DataLoader
import pandas as pd
import numpy as np

class CustomDataset(Dataset):
    def __init__(self, csv_file, transform=None):
        self.data = pd.read_csv(csv_file)
        self.transform = transform
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        # データの取得（例：最初の列をラベル、残りを特徴量とする）
        label = self.data.iloc[idx, 0]
        features = self.data.iloc[idx, 1:].values.astype(np.float32)
        
        sample = {'features': features, 'label': label}
        
        if self.transform:
            sample = self.transform(sample)
        
        return sample

# カスタムトランスフォーム
class ToTensor:
    def __call__(self, sample):
        features, label = sample['features'], sample['label']
        return {
            'features': torch.from_tensor(features),
            'label': torch.tensor(label, dtype=torch.long)
        }

class Normalize:
    def __init__(self, mean, std):
        self.mean = mean
        self.std = std
    
    def __call__(self, sample):
        features = sample['features']
        features = (features - self.mean) / self.std
        return {'features': features, 'label': sample['label']}

# データセットとデータローダーの作成
transforms_list = [ToTensor(), Normalize(0.5, 0.5)]
composed = transforms.Compose(transforms_list)

# カスタムデータセットの使用例
# dataset = CustomDataset('data.csv', transform=composed)
# dataloader = DataLoader(dataset, batch_size=32, shuffle=True)

print("カスタムデータセットクラスを定義しました")

転移学習

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision.models as models
import torchvision.transforms as transforms

# 事前学習済みモデルの読み込み
model = models.resnet50(pretrained=True)

# 最終層のみ学習可能にする（特徴抽出）
for param in model.parameters():
    param.requires_grad = False

# 最終層の置き換え（新しいタスク用）
num_classes = 10  # 新しいタスクのクラス数
model.fc = nn.Linear(model.fc.in_features, num_classes)

# 最終層のみ学習可能にする
for param in model.fc.parameters():
    param.requires_grad = True

# デバイスに移動
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)

# 損失関数とオプティマイザー
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.fc.parameters(), lr=0.001)

# ファインチューニングの場合（全層を学習）
def enable_fine_tuning(model, lr=0.0001):
    for param in model.parameters():
        param.requires_grad = True
    
    # レイヤー別学習率の設定
    optimizer = optim.Adam([
        {'params': model.conv1.parameters(), 'lr': lr * 0.1},
        {'params': model.layer1.parameters(), 'lr': lr * 0.1},
        {'params': model.layer2.parameters(), 'lr': lr * 0.5},
        {'params': model.layer3.parameters(), 'lr': lr * 0.5},
        {'params': model.layer4.parameters(), 'lr': lr},
        {'params': model.fc.parameters(), 'lr': lr * 10}
    ])
    
    return optimizer

print("転移学習用のモデルを設定しました")
print(f"モデルの最終層: {model.fc}")

モデルの保存と読み込み

import torch
import torch.nn as nn

# モデルの定義
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(10, 50)
        self.fc2 = nn.Linear(50, 10)
        self.fc3 = nn.Linear(10, 1)
    
    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

# モデルのインスタンス化と学習（簡単な例）
model = Net()
optimizer = torch.optim.Adam(model.parameters())

# サンプル学習
x = torch.randn(100, 10)
y = torch.randn(100, 1)

for epoch in range(100):
    optimizer.zero_grad()
    output = model(x)
    loss = nn.MSELoss()(output, y)
    loss.backward()
    optimizer.step()

# 1. モデル全体の保存・読み込み
torch.save(model, 'complete_model.pth')
loaded_model = torch.load('complete_model.pth')

# 2. 重みのみ保存・読み込み（推奨）
torch.save(model.state_dict(), 'model_weights.pth')

# 読み込み時はモデル構造を再定義してから重みを読み込む
new_model = Net()
new_model.load_state_dict(torch.load('model_weights.pth'))

# 3. チェックポイント（モデル + オプティマイザー状態）
checkpoint = {
    'epoch': 100,
    'model_state_dict': model.state_dict(),
    'optimizer_state_dict': optimizer.state_dict(),
    'loss': loss.item()
}
torch.save(checkpoint, 'checkpoint.pth')

# チェックポイントの読み込み
checkpoint = torch.load('checkpoint.pth')
model.load_state_dict(checkpoint['model_state_dict'])
optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
epoch = checkpoint['epoch']
loss = checkpoint['loss']

# 推論モードに設定
model.eval()

print("モデルの保存・読み込みが完了しました")
print(f"最終エポック: {epoch}, 最終損失: {loss:.4f}")