pytorchの概要と環境設定及び実装例

機械学習技術人工知能技術デジタルトランスフォーメーション自然言語処理技術画像処理技術オンライン学習深層学習技術確率生成モデル強化学習技術グラフニューラルネットワーク python Python 本ブログのナビ

Pytorchについて

PyTorchは、Facebookが開発しオープンソースで提供されている深層学習のライブラリであり、柔軟性、動的計算グラフ、GPU加速などの特徴を持ち、様々な機械学習タスクを実装を可能としてくれるものとなる。以下に、PyTorchの特徴と利点について述べる。

1. 動的計算グラフ (Dynamic Computational Graphs): PyTorchは動的計算グラフを使用しており、これは、計算グラフが実行時に構築され、それによって制御フローの変更や条件付きの実行が容易になる。また、TensorFlowなどの静的計算グラフを持つフレームワークと比較して、PyTorchはデバッグが容易であり、より自然なPythonの文法に近い。

2. 柔軟性と簡潔さ: PyTorchはPythonicであり、シンプルなAPIを提供している。これにより、ユーザーは直感的にモデルを構築し、実験することができる。また、高レベルの抽象化と低レベルのコントロールが両方とも可能であり、ニーズに合わせて使い分けることが可能なフレームワークとなる。

3. GPUサポートと高速な演算: PyTorchはGPUをサポートしており、モデルの学習や推論を高速化することができま、CUDAを使用してGPUを利用することが可能なライブラリとなる。さらに、自動微分（Automatic differentiation）により、勾配計算が自動的に行われ、高速で効率的な学習が可能となる。

4. 豊富なコミュニティと資源: PyTorchは非常にアクティブで大規模なコミュニティを持ち、ドキュメント、チュートリアル、サンプルコードなどの資源が豊富である。さらに、PyTorch HubやTorchVisionなどの拡張ライブラリも利用でき、事前学習済みモデルやデータセットが提供されている。

5. 様々な応用: 画像処理、自然言語処理、音声処理、強化学習など、幅広い機械学習タスクにPyTorchを適用することができる。

6. 転移学習 (Transfer Learning) のサポート: 事前学習済みモデルを利用して、新しいタスクに適用する転移学習は、PyTorchで容易に実装することができる。

7. オープンソースとコミュニティの拡大: PyTorchはオープンソースプロジェクトであり、ユーザーがフィードバックを提供し、改善に貢献することができる。

pytorchの環境設定

PyTorch を使用するための基本的な環境設定について述べる。PyTorchは以下の手順に従って、インストールし、環境を設定することができる。

1. Python のインストール: まず、Python をインストールする。PyTorch は Python で動作するため、Python のバージョン 3.6 以上が推奨されている。Python 公式サイトからインストーラーをダウンロードし、インストールする。

2. PyTorch のインストール: PyTorch をインストールする方法はいくつかあるが、通常は pip を使用して以下の様にインストールする。

CPU バージョンのインストール:

pip install torch torchvision torchaudio

CUDA を利用した GPU バージョンのインストール: GPU を利用する場合は、CUDA Toolkit がインストールされていることを確認し、次のコマンドで PyTorch をインストールする。

pip install torch torchvision torchaudio cudatoolkit=xx.x -c pytorch

ここで、xx.x には自分の環境に対応する CUDA バージョンが入る。たとえば、CUDA 11.1 を使用する場合は、以下のようになる。

pip install torch torchvision torchaudio cudatoolkit=11.1 -c pytorch

3. バージョンの確認: PyTorch が正しくインストールされたかどうかを確認するために、Python インタラクティブシェルやスクリプトで次のコードを実行する。

import torch

print(torch.__version__)
print(torch.cuda.is_available())  # GPU が利用可能かどうか

これで、PyTorch の基本的なインストールと設定が完了する。また、PyTorch Geometric やその他のライブラリを使用する場合は、追加のライブラリもインストールする必要がある。

4. PyTorch Geometric のインストール(option): PyTorch Geometric を使用する場合は、次の手順でインストールする。

pip install torch-geometric

5. Jupyter Notebook のインストール (オプション): Jupyter Notebook を使用すると、インタラクティブな Python プログラムを実行し、結果を視覚化することができる。インストールする場合は、次のコマンドを実行する。

pip install jupyterlab

6. ライブラリのインポート: 最後に、インストールしたライブラリを Python スクリプトや Jupyter Notebook で使用するために、必要なライブラリをインポートする。

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import DataLoader

これで、PyTorch の環境設定が完了する。これらを用いて実装を行うには、必要に応じて、仮想環境や別の Python バージョンにインストールすることもできる。

pytorchのサンプルコード

以下にPyTorch を使用した簡単なサンプルコードを示す。この例では、線形回帰モデルを使用して、簡単な線形関数のフィッティングを行うものとなる。

<線形回帰モデルの実装例>

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import matplotlib.pyplot as plt

# データの準備
x = np.random.rand(100, 1)  # ランダムな入力データ
y = 3 * x + 2 + 0.2 * np.random.randn(100, 1)  # 真の関数にノイズを加えた出力データ

x_tensor = torch.from_numpy(x).float()
y_tensor = torch.from_numpy(y).float()

# 線形回帰モデルの定義
class LinearRegression(nn.Module):
    def __init__(self):
        super(LinearRegression, self).__init__()
        self.linear = nn.Linear(1, 1)  # 入力次元: 1, 出力次元: 1

    def forward(self, x):
        return self.linear(x)

model = LinearRegression()

# 損失関数と最適化手法の定義
criterion = nn.MSELoss()
optimizer = optim.SGD(model.parameters(), lr=0.01)

# モデルの学習
num_epochs = 100
for epoch in range(num_epochs):
    optimizer.zero_grad()
    outputs = model(x_tensor)
    loss = criterion(outputs, y_tensor)
    loss.backward()
    optimizer.step()

    if (epoch+1) % 10 == 0:
        print('Epoch [{}/{}], Loss: {:.4f}'.format(epoch+1, num_epochs, loss.item()))

# 学習したモデルの評価
model.eval()
predicted = model(x_tensor).detach().numpy()

# 元のデータと学習したモデルのフィッティング結果をプロット
plt.scatter(x, y, label='Original data')
plt.plot(x, predicted, label='Fitted line', color='r')
plt.legend()
plt.xlabel('x')
plt.ylabel('y')
plt.title('Linear Regression Example')
plt.show()

このコードでは、次の手順で線形回帰モデルを定義し、学習、評価、結果のプロットを行っている。

1. データの準備: ランダムな入力データ x を生成し、それに対応する出力データ y を生成する。

2. 線形回帰モデルの定義: LinearRegression クラスを定義し、nn.Linear モジュールを使用して線形モデルを定義する。

3. 損失関数と最適化手法の定義: 平均二乗誤差（MSE）を損失関数として定義し、確率的勾配降下法（SGD）を最適化手法として定義する。

4. モデルの学習: 100 エポックでモデルを学習し、各エポックで損失を計算する。

5. 学習したモデルの評価: モデルを評価モードに設定し、学習したモデルで入力データ x_tensor を推論し、予測結果を取得する。

6. プロット: 元のデータと学習したモデルの予測結果をプロットして、フィッティングの結果を可視化する。

<深層学習の実装例>

以下に、PyTorchを用いた深層学習の実装例を示す。深層学習技術の詳細は”深層学習について“を参照のこと。

1. 画像分類 (Image Classification): 画像データセット（例: CIFAR-10、MNISTなど）を使用して、画像を異なるクラスに分類するモデルを構築する。

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision.transforms as transforms
from torchvision.datasets import CIFAR10
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))
])

train_dataset = CIFAR10(root='./data', train=True, download=True, transform=transform)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)

# ネットワークの定義
class SimpleCNN(nn.Module):
    def __init__(self):
        super(SimpleCNN, self).__init__()
        self.conv1 = nn.Conv2d(3, 16, 3, 1, 1)
        self.conv2 = nn.Conv2d(16, 32, 3, 1, 1)
        self.fc1 = nn.Linear(32 * 8 * 8, 128)
        self.fc2 = nn.Linear(128, 10)

    def forward(self, x):
        x = torch.relu(self.conv1(x))
        x = torch.relu(self.conv2(x))
        x = x.view(-1, 32 * 8 * 8)
        x = torch.relu(self.fc1(x))
        x = self.fc2(x)
        return x

model = SimpleCNN()

# 損失関数と最適化手法
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 学習ループ
for epoch in range(5):
    running_loss = 0.0
    for inputs, labels in train_loader:
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        running_loss += loss.item()
    print(f"Epoch {epoch+1}, Loss: {running_loss/len(train_loader)}")

# テスト
test_dataset = CIFAR10(root='./data', train=False, download=True, transform=transform)
test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)

correct = 0
total = 0
with torch.no_grad():
    for inputs, labels in test_loader:
        outputs = model(inputs)
        _, predicted = torch.max(outputs.data, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()

print(f"Accuracy on test set: {(correct/total)*100}%")

2. 転移学習 (Transfer Learning): 事前学習済みモデルを取得し、新しいデータセットに適用することで、高性能なモデルを構築する。

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

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

train_dataset = CIFAR10(root='./data', train=True, download=True, transform=transform)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)

# 事前学習済みのResNet18モデルを読み込み
model = models.resnet18(pretrained=True)
num_ftrs = model.fc.in_features
model.fc = nn.Linear(num_ftrs, 10)  # CIFAR-10のクラス数に合わせる

# 損失関数と最適化手法
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 学習ループ
for epoch in range(5):
    running_loss = 0.0
    for inputs, labels in train_loader:
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        running_loss += loss.item()
    print(f"Epoch {epoch+1}, Loss: {running_loss/len(train_loader)}")

# テスト
test_dataset = CIFAR10(root='./data', train=False, download=True, transform=transform)
test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)

correct = 0
total = 0
with torch.no_grad():
    for inputs, labels in test_loader:
        outputs = model(inputs)
        _, predicted = torch.max(outputs.data, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()

print(f"Accuracy on test set: {(correct/total)*100}%")

3. 自然言語処理 (NLP): LSTMによる感情分析: IMDb映画レビューデータセットを使用して、映画レビューの感情を分類するモデルを構築する。

import torch
import torch.nn as nn
import torch.optim as optim
from torchtext.legacy import data
from torchtext.legacy import datasets
import random

# データの準備
SEED = 1234

torch.manual_seed(SEED)
torch.backends.cudnn.deterministic = True

TEXT = data.Field(tokenize='spacy', tokenizer_language='en_core_web_sm')
LABEL = data.LabelField(dtype=torch.float)

train_data, test_data = datasets.IMDB.splits(TEXT, LABEL)

train_data, valid_data = train_data.split(random_state=random.seed(SEED))

MAX_VOCAB_SIZE = 25000

TEXT.build_vocab(train_data, max_size=MAX_VOCAB_SIZE)
LABEL.build_vocab(train_data)

BATCH_SIZE = 64

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

train_iterator, valid_iterator, test_iterator = data.BucketIterator.splits(
    (train_data, valid_data, test_data), 
    batch_size=BATCH_SIZE, 
    device=device)

# ネットワークの定義
class SentimentLSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, n_layers, bidirectional, dropout):
        super().__init__()

        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, num_layers=n_layers, bidirectional=bidirectional, dropout=dropout)
        self.fc = nn.Linear(hidden_dim * 2, output_dim)
        self.dropout = nn.Dropout(dropout)

    def forward(self, text):
        embedded = self.dropout(self.embedding(text))
        output, (hidden, cell) = self.lstm(embedded)
        hidden = self.dropout(torch.cat((hidden[-2,:,:], hidden[-1,:,:]), dim=1))
        return self.fc(hidden.squeeze(0))

INPUT_DIM = len(TEXT.vocab)
EMBEDDING_DIM = 100
HIDDEN_DIM = 256
OUTPUT_DIM = 1
N_LAYERS = 2
BIDIRECTIONAL = True
DROPOUT = 0.5

model = SentimentLSTM(INPUT_DIM, EMBEDDING_DIM, HIDDEN_DIM, OUTPUT_DIM, N_LAYERS, BIDIRECTIONAL, DROPOUT)

# 損失関数と最適化手法
criterion = nn.BCEWithLogitsLoss()
optimizer = optim.Adam(model.parameters())

model = model.to(device)
criterion = criterion.to(device)

# 学習ループ
def binary_accuracy(preds, y):
    rounded_preds = torch.round(torch.sigmoid(preds))
    correct = (rounded_preds == y).float()
    acc = correct.sum() / len(correct)
    return acc

def train(model, iterator, optimizer, criterion):
    epoch_loss = 0
    epoch_acc = 0

    model.train()

    for batch in iterator:
        optimizer.zero_grad()
        predictions = model(batch.text).squeeze(1)
        loss = criterion(predictions, batch.label)
        acc = binary_accuracy(predictions, batch.label)
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
        epoch_acc += acc.item()

    return epoch_loss / len(iterator), epoch_acc / len(iterator)

def evaluate(model, iterator, criterion):
    epoch_loss = 0
    epoch_acc = 0

    model.eval()

    with torch.no_grad():
        for batch in iterator:
            predictions = model(batch.text).squeeze(1)
            loss = criterion(predictions, batch.label)
            acc = binary_accuracy(predictions, batch.label)
            epoch_loss += loss.item()
            epoch_acc += acc.item()

    return epoch_loss / len(iterator), epoch_acc / len(iterator)

N_EPOCHS = 5

for epoch in range(N_EPOCHS):
    train_loss, train_acc = train(model, train_iterator, optimizer, criterion)
    valid_loss, valid_acc = evaluate(model, valid_iterator, criterion)
    print(f'Epoch: {epoch+1:02}')
    print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc*100:.2f}%')
    print(f'\t Val. Loss: {valid_loss:.3f} |  Val. Acc: {valid_acc*100:.2f}%')

# テスト
test_loss, test_acc = evaluate(model, test_iterator, criterion)
print(f'Test Loss: {test_loss:.3f} | Test Acc: {test_acc*100:.2f}%')

<強化学習の実装例>

以下に強化学習を実装する例を示す。ここでは、CartPole環境を使用し、Q学習（Q-Learning）アルゴリズムを実装している。強化学習技術の詳細は”様々な強化学習技術の理論とアルゴリズムとpythonによる実装“を参照のこと。

まず、必要なライブラリと環境をインポートする。

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import gym

次に、Q学習のエージェントを定義する。

class QLearningAgent:
    def __init__(self, state_size, action_size, learning_rate=0.01, discount_rate=0.99, exploration_rate=1.0, min_exploration_rate=0.01, exploration_decay_rate=0.99):
        self.state_size = state_size
        self.action_size = action_size
        self.learning_rate = learning_rate
        self.discount_rate = discount_rate
        self.exploration_rate = exploration_rate
        self.min_exploration_rate = min_exploration_rate
        self.exploration_decay_rate = exploration_decay_rate
        self.q_table = np.zeros((state_size, action_size))

    def choose_action(self, state):
        if np.random.uniform(0, 1) < self.exploration_rate:
            return np.random.choice(self.action_size)
        else:
            return np.argmax(self.q_table[state, :])

    def update_q_table(self, state, action, reward, next_state):
        best_next_action = np.argmax(self.q_table[next_state, :])
        td_target = reward + self.discount_rate * self.q_table[next_state, best_next_action]
        td_error = td_target - self.q_table[state, action]
        self.q_table[state, action] += self.learning_rate * td_error

    def decay_exploration_rate(self):
        self.exploration_rate = max(self.min_exploration_rate, self.exploration_rate * self.exploration_decay_rate)

次に、CartPole環境を作成し、エージェントを初期化する。

env = gym.make('CartPole-v1')
state_size = env.observation_space.shape[0]
action_size = env.action_space.n

agent = QLearningAgent(state_size, action_size)

学習を行う。

num_episodes = 1000

for episode in range(num_episodes):
    state = env.reset()
    done = False
    total_reward = 0

    while not done:
        action = agent.choose_action(state)
        next_state, reward, done, _ = env.step(action)
        total_reward += reward
        agent.update_q_table(state, action, reward, next_state)
        state = next_state

    agent.decay_exploration_rate()

    if episode % 100 == 0:
        print(f"Episode {episode}, Total Reward: {total_reward}")

env.close()

このコードは、Q学習エージェントを使用してCartPole環境で学習を行っている。エージェントは環境内で行動し、報酬を受け取りながらQ値を更新していき、学習が進むにつれて、エージェントは最適な行動を選択するようになる。

<ガウス過程の実装例>

以下にガウス過程（Gaussian Processes, GP）を実装する例を示す。ここでは、ガウス過程回帰を行っている。ガウス過程の詳細は”ノンパラメトリックベイズとガウス過程について“を参照のこと。

まず、必要なライブラリをインポートする。

import torch
import gpytorch
from matplotlib import pyplot as plt

次に、サンプルデータを作成する。ここでは、sin関数にノイズを加えたものをサンプルとして使用している。

train_x = torch.linspace(0, 1, 100)
train_y = torch.sin(train_x * (2 * 3.1416)) + torch.randn(train_x.size()) * 0.2

次に、ガウス過程モデルを定義する。ここでは、ガウス過程回帰を行うためのモデルを定義している。

class ExactGPModel(gpytorch.models.ExactGP):
    def __init__(self, train_x, train_y, likelihood):
        super(ExactGPModel, self).__init__(train_x, train_y, likelihood)
        self.mean_module = gpytorch.means.ConstantMean()
        self.covar_module = gpytorch.kernels.ScaleKernel(gpytorch.kernels.RBFKernel())

    def forward(self, x):
        mean_x = self.mean_module(x)
        covar_x = self.covar_module(x)
        return gpytorch.distributions.MultivariateNormal(mean_x, covar_x)

likelihood = gpytorch.likelihoods.GaussianLikelihood()
model = ExactGPModel(train_x, train_y, likelihood)

次に、モデルをトレーニングする。

model.train()
likelihood.train()

optimizer = torch.optim.Adam(model.parameters(), lr=0.1)
mll = gpytorch.mlls.ExactMarginalLogLikelihood(likelihood, model)

training_iterations = 50
for i in range(training_iterations):
    optimizer.zero_grad()
    output = model(train_x)
    loss = -mll(output, train_y)
    loss.backward()
    optimizer.step()

最後に、モデルをテストして結果をプロットする。

model.eval()
likelihood.eval()

test_x = torch.linspace(0, 1, 51)
with torch.no_grad(), gpytorch.settings.fast_pred_var():
    observed_pred = likelihood(model(test_x))

mean = observed_pred.mean
lower, upper = observed_pred.confidence_region()

plt.figure(figsize=(8, 6))
plt.plot(train_x.numpy(), train_y.numpy(), 'k.')
plt.plot(test_x.numpy(), mean.numpy(), 'b')
plt.fill_between(test_x.numpy(), lower.numpy(), upper.numpy(), alpha=0.5)
plt.xlabel('x')
plt.ylabel('y')
plt.title('Gaussian Process Regression')
plt.show()

これで、ガウス過程回帰モデルが作成され、サンプルデータに適合しているかどうかがプロットされている。上の信頼区間は、予測の不確かさを示している。この例では、PyTorchを使用してガウス過程回帰を実装し、簡単なサンプルデータに適合させている。

<GNNの実装例>

以下にPyTorch Geometricを使用して、グラフニューラルネットワーク（Graph Neural Network, GNN）を実装する例を示す。ここでは、グラフデータセットであるCoraを使用して、ノードの分類タスクを行っている。GNNの詳細は”グラフニューラルネットワーク“を参照のこと。

まず、必要なライブラリをインポートする。

import torch
import torch.nn.functional as F
from torch_geometric.datasets import Planetoid
import torch_geometric.nn as pyg_nn
import torch_geometric.utils as pyg_utils

次に、Coraデータセットを読み込む。

dataset = Planetoid(root='data/Cora', name='Cora')
data = dataset[0]

次に、グラフニューラルネットワーク（GNN）の定義を行う。

class GNN(torch.nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super(GNN, self).__init__()
        self.conv1 = pyg_nn.GCNConv(input_dim, hidden_dim)
        self.conv2 = pyg_nn.GCNConv(hidden_dim, output_dim)

    def forward(self, data):
        x, edge_index = data.x, data.edge_index

        x = self.conv1(x, edge_index)
        x = F.relu(x)
        x = F.dropout(x, training=self.training)

        x = self.conv2(x, edge_index)
        return F.log_softmax(x, dim=1)

このモデルでは、2つのGCNConvレイヤーを使用しており、それぞれのレイヤーの入力次元、隠れ層の次元、出力次元が指定されている。

次に、モデルのインスタンスを作成し、最適化手法を設定する。

model = GNN(dataset.num_node_features, hidden_dim=16, output_dim=dataset.num_classes)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=5e-4)

モデルの学習を行う。

def train():
    model.train()
    optimizer.zero_grad()
    out = model(data)
    loss = F.nll_loss(out[data.train_mask], data.y[data.train_mask])
    loss.backward()
    optimizer.step()
    return loss.item()

def test():
    model.eval()
    out = model(data)
    pred = out.argmax(dim=1)
    test_correct = pred[data.test_mask] == data.y[data.test_mask]
    test_acc = int(test_correct.sum()) / int(data.test_mask.sum())
    return test_acc

for epoch in range(200):
    loss = train()
    if epoch % 10 == 0:
        test_acc = test()
        print(f'Epoch: {epoch}, Loss: {loss:.4f}, Test Accuracy: {test_acc:.4f}')

このコードは、Coraデータセットを使用してGNNモデルをトレーニングし、ノードの分類タスクを実行している。学習のたびにテストデータセットでの精度が表示される。