グラフニューラルネットワーク完全ガイド: GCNからGraph Transformerまで

import numpy as np
import networkx as nx

# グラフの作成
G = nx.karate_club_graph()

# 隣接行列の取得
A = nx.adjacency_matrix(G).toarray().astype(float)

# 次数行列の計算
degrees = np.array(A.sum(axis=1)).flatten()
D = np.diag(degrees)

# ラプラシアン行列
L = D - A
print(f"ラプラシアン行列の形状: {L.shape}")
print(f"最小固有値: {np.linalg.eigvalsh(L).min():.4f}")
print(f"最大固有値: {np.linalg.eigvalsh(L).max():.4f}")

import torch
import torch.nn.functional as F
from torch_geometric.nn import GCNConv
from torch_geometric.datasets import Planetoid
from torch_geometric.transforms import NormalizeFeatures

# データセットの読み込み
dataset = Planetoid(root='/tmp/Cora', name='Cora', transform=NormalizeFeatures())
data = dataset[0]

print(f"ノード数: {data.num_nodes}")
print(f"エッジ数: {data.num_edges}")
print(f"ノード特徴量の次元: {data.num_node_features}")
print(f"クラス数: {dataset.num_classes}")

class GCN(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = GCNConv(dataset.num_features, 16)
        self.conv2 = GCNConv(16, dataset.num_classes)

    def forward(self, x, edge_index):
        x = F.relu(self.conv1(x, edge_index))
        x = F.dropout(x, training=self.training)
        x = self.conv2(x, edge_index)
        return F.log_softmax(x, dim=1)

model = GCN()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=5e-4)

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

# 学習の実行
for epoch in range(200):
    loss = train()
    if epoch % 20 == 0:
        print(f'Epoch {epoch}, Loss: {loss:.4f}')

from torch_geometric.nn import GATConv

class GAT(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = GATConv(dataset.num_features, 8, heads=8, dropout=0.6)
        self.conv2 = GATConv(8 * 8, dataset.num_classes, heads=1, dropout=0.6)

    def forward(self, x, edge_index):
        x = F.dropout(x, p=0.6, training=self.training)
        x = F.elu(self.conv1(x, edge_index))
        x = F.dropout(x, p=0.6, training=self.training)
        x = self.conv2(x, edge_index)
        return F.log_softmax(x, dim=1)

model = GAT()
optimizer = torch.optim.Adam(model.parameters(), lr=0.005, weight_decay=5e-4)

from torch_geometric.nn import SAGEConv
from torch_geometric.loader import NeighborLoader

class GraphSAGE(torch.nn.Module):
    def __init__(self, in_channels, hidden_channels, out_channels):
        super().__init__()
        self.conv1 = SAGEConv(in_channels, hidden_channels)
        self.conv2 = SAGEConv(hidden_channels, out_channels)

    def forward(self, x, edge_index):
        x = F.relu(self.conv1(x, edge_index))
        x = F.dropout(x, p=0.5, training=self.training)
        x = self.conv2(x, edge_index)
        return F.log_softmax(x, dim=1)

# 近傍サンプリングによるミニバッチ学習
train_loader = NeighborLoader(
    data,
    num_neighbors=[25, 10],
    batch_size=256,
    input_nodes=data.train_mask,
    shuffle=True
)

from torch_geometric.nn import MessagePassing
from torch_geometric.utils import add_self_loops, degree
import torch

class CustomGNNConv(MessagePassing):
    def __init__(self, in_channels, out_channels):
        super().__init__(aggr='add')
        self.lin = torch.nn.Linear(in_channels, out_channels)

    def forward(self, x, edge_index):
        # 自己ループの追加
        edge_index, _ = add_self_loops(edge_index, num_nodes=x.size(0))

        # 正規化係数の計算
        row, col = edge_index
        deg = degree(col, x.size(0), dtype=x.dtype)
        deg_inv_sqrt = deg.pow(-0.5)
        norm = deg_inv_sqrt[row] * deg_inv_sqrt[col]

        return self.propagate(edge_index, x=x, norm=norm)

    def message(self, x_j, norm):
        # メッセージの計算: 正規化された近傍特徴量
        return norm.view(-1, 1) * x_j

    def update(self, aggr_out):
        # 更新: 線形変換
        return self.lin(aggr_out)

import torch
import torch.nn as nn
from torch_geometric.nn import TransformerConv

class GraphTransformer(nn.Module):
    def __init__(self, in_channels, hidden_channels, out_channels, heads=4):
        super().__init__()
        self.conv1 = TransformerConv(
            in_channels, hidden_channels, heads=heads, dropout=0.1
        )
        self.conv2 = TransformerConv(
            hidden_channels * heads, out_channels, heads=1, concat=False
        )
        self.norm1 = nn.LayerNorm(hidden_channels * heads)
        self.norm2 = nn.LayerNorm(out_channels)

    def forward(self, x, edge_index):
        x = self.conv1(x, edge_index)
        x = self.norm1(x)
        x = F.elu(x)
        x = F.dropout(x, p=0.1, training=self.training)
        x = self.conv2(x, edge_index)
        x = self.norm2(x)
        return F.log_softmax(x, dim=1)

from rdkit import Chem
from torch_geometric.utils import from_smiles

# SMILESをグラフに変換
mol = from_smiles('CC(=O)Oc1ccccc1C(=O)O')  # アスピリン
print(f"ノード(原子)数: {mol.num_nodes}")
print(f"エッジ(結合)数: {mol.num_edges}")
print(f"ノード特徴量の次元: {mol.x.shape}")

import torch
from torch_geometric.nn import TransE

# TransEモデルの初期化
model = TransE(
    num_nodes=1000,
    num_relations=50,
    hidden_channels=50,
    margin=1.0,
    p_norm=1
)

import torch
import torch.nn.functional as F
from torch_geometric.datasets import Reddit
from torch_geometric.loader import ClusterData, ClusterLoader
from torch_geometric.nn import SAGEConv

# データ読み込み
dataset = Reddit(root='/tmp/Reddit')
data = dataset[0]

# クラスター分割
cluster_data = ClusterData(data, num_parts=1500, recursive=False)
train_loader = ClusterLoader(
    cluster_data, batch_size=20, shuffle=True, num_workers=2
)

class ClusterGCN(torch.nn.Module):
    def __init__(self, in_channels, hidden_channels, out_channels):
        super().__init__()
        self.conv1 = SAGEConv(in_channels, hidden_channels)
        self.conv2 = SAGEConv(hidden_channels, hidden_channels)
        self.conv3 = SAGEConv(hidden_channels, out_channels)

    def forward(self, x, edge_index):
        x = F.relu(self.conv1(x, edge_index))
        x = F.dropout(x, p=0.5, training=self.training)
        x = F.relu(self.conv2(x, edge_index))
        x = F.dropout(x, p=0.5, training=self.training)
        x = self.conv3(x, edge_index)
        return F.log_softmax(x, dim=1)

model = ClusterGCN(
    in_channels=dataset.num_features,
    hidden_channels=256,
    out_channels=dataset.num_classes
)

optimizer = torch.optim.Adam(model.parameters(), lr=0.005)

def train():
    model.train()
    total_loss = total_nodes = 0
    for batch in train_loader:
        optimizer.zero_grad()
        out = model(batch.x, batch.edge_index)
        loss = F.nll_loss(out[batch.train_mask], batch.y[batch.train_mask])
        loss.backward()
        optimizer.step()
        nodes = batch.train_mask.sum().item()
        total_loss += loss.item() * nodes
        total_nodes += nodes
    return total_loss / total_nodes

from torch_geometric.datasets import TUDataset
from torch_geometric.loader import DataLoader
from torch_geometric.nn import GINConv, global_add_pool

# MUTAGデータセット
dataset = TUDataset(root='/tmp/TUDataset', name='MUTAG')
dataset = dataset.shuffle()
n = len(dataset)
train_dataset = dataset[:int(0.8 * n)]
test_dataset = dataset[int(0.8 * n):]

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=32)

class GIN(torch.nn.Module):
    """Graph Isomorphism Network - 最大表現力を持つGNN"""

    def __init__(self, in_channels, hidden_channels, out_channels, num_layers=5):
        super().__init__()
        self.convs = torch.nn.ModuleList()
        self.bns = torch.nn.ModuleList()

        for i in range(num_layers):
            in_ch = in_channels if i == 0 else hidden_channels
            mlp = torch.nn.Sequential(
                torch.nn.Linear(in_ch, hidden_channels),
                torch.nn.BatchNorm1d(hidden_channels),
                torch.nn.ReLU(),
                torch.nn.Linear(hidden_channels, hidden_channels)
            )
            self.convs.append(GINConv(mlp, train_eps=True))
            self.bns.append(torch.nn.BatchNorm1d(hidden_channels))

        self.classifier = torch.nn.Sequential(
            torch.nn.Linear(hidden_channels * num_layers, hidden_channels),
            torch.nn.ReLU(),
            torch.nn.Dropout(0.5),
            torch.nn.Linear(hidden_channels, out_channels)
        )

    def forward(self, x, edge_index, batch):
        xs = []
        for conv, bn in zip(self.convs, self.bns):
            x = conv(x, edge_index)
            x = bn(x)
            x = F.relu(x)
            xs.append(global_add_pool(x, batch))

        out = torch.cat(xs, dim=1)
        return self.classifier(out)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model_gin = GIN(
    in_channels=dataset.num_features,
    hidden_channels=64,
    out_channels=dataset.num_classes
).to(device)