第4章：実世界への応用

1. 分子特性予測（創薬への応用）

創薬（Drug Discovery）分野では、候補化合物の物性（溶解度、毒性、生体膜透過性など）を実験前に予測できると、開発コストと時間を大幅に削減できます。分子は原子をノード、化学結合をエッジとする自然なグラフ構造を持つため、GNNは分子特性予測（Molecular Property Prediction）と非常に相性が良いタスクです。

分子グラフの表現

分子をグラフとして扱う場合、典型的には以下のように対応付けます。

本節では、MoleculeNetベンチマークに含まれるESOLデータセットを使い、分子の水溶解度（対数溶解度）を回帰予測するGNNモデルを実装します。ESOLは1,128個の小分子から構成され、SMILES記法（分子構造を文字列で表す記法）から原子・結合特徴量が自動生成されます。

pip install rdkit

ESOLデータセットの読み込み

import torch
from torch_geometric.datasets import MoleculeNet

dataset = MoleculeNet(root='/tmp/ESOL', name='ESOL')

print(f"Dataset: {dataset}")
print(f"Number of molecules: {len(dataset)}")
print(f"Number of node features: {dataset.num_node_features}")
print(f"Number of edge features: {dataset.num_edge_features}")

sample = dataset[0]
print(f"\nSample molecule: {sample}")
print(f"SMILES: {sample.smiles}")
print(f"Target (log solubility): {sample.y.item():.4f}")

出力例：

Dataset: ESOL(1128)
Number of molecules: 1128
Number of node features: 9
Number of edge features: 3

Sample molecule: Data(x=[32, 9], edge_index=[2, 68], edge_attr=[68, 3], y=[1, 1], smiles='OCC3OC(OCC2OC(OC(C#N)c1ccccc1)C(O)C(O)C2O)C(O)C(O)C3O')
SMILES: OCC3OC(OCC2OC(OC(C#N)c1ccccc1)C(O)C(O)C2O)C(O)C(O)C3O
Target (log solubility): -0.7700

GINを使った分子特性予測モデル

分子特性予測では、GIN（Graph Isomorphism Network）という層がよく使われます。GINは理論的にWeisfeiler-Lehmanグラフ同型性テストと同等の識別力を持つように設計されており、分子のような小さく密な構造の違いを捉えるのに適しています。

$$\mathbf{x}_i^{(k+1)} = \text{MLP}^{(k)}\left((1 + \epsilon^{(k)}) \cdot \mathbf{x}_i^{(k)} + \sum_{j \in \mathcal{N}(i)} \mathbf{x}_j^{(k)}\right)$$

import torch
import torch.nn.functional as F
from torch.nn import Linear, Sequential, ReLU, BatchNorm1d
from torch_geometric.nn import GINConv, global_mean_pool

class MolecularGIN(torch.nn.Module):
    def __init__(self, num_features, hidden_channels):
        super().__init__()
        nn1 = Sequential(Linear(num_features, hidden_channels), ReLU(),
                          Linear(hidden_channels, hidden_channels), BatchNorm1d(hidden_channels))
        self.conv1 = GINConv(nn1)

        nn2 = Sequential(Linear(hidden_channels, hidden_channels), ReLU(),
                          Linear(hidden_channels, hidden_channels), BatchNorm1d(hidden_channels))
        self.conv2 = GINConv(nn2)

        self.lin1 = Linear(hidden_channels, hidden_channels)
        self.lin2 = Linear(hidden_channels, 1)  # 回帰タスクなので出力は1次元

    def forward(self, x, edge_index, batch):
        x = self.conv1(x, edge_index).relu()
        x = self.conv2(x, edge_index).relu()

        # グラフ全体を1つのベクトルに集約(グラフレベルのプーリング)
        x = global_mean_pool(x, batch)

        x = self.lin1(x).relu()
        x = F.dropout(x, p=0.2, training=self.training)
        x = self.lin2(x)
        return x

学習と評価

import torch
from torch_geometric.datasets import MoleculeNet
from torch_geometric.loader import DataLoader

dataset = MoleculeNet(root='/tmp/ESOL', name='ESOL')
dataset = dataset.shuffle()

# 特徴量をfloatに統一(データセットにより整数型で格納されている場合があるため)
for data in dataset:
    data.x = data.x.float()

train_size = int(len(dataset) * 0.8)
train_dataset = dataset[:train_size]
test_dataset = dataset[train_size:]

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

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = MolecularGIN(num_features=dataset.num_node_features, hidden_channels=64).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = torch.nn.MSELoss()

def train():
    model.train()
    total_loss = 0
    for batch in train_loader:
        batch = batch.to(device)
        optimizer.zero_grad()
        out = model(batch.x, batch.edge_index, batch.batch)
        loss = criterion(out.view(-1), batch.y.view(-1))
        loss.backward()
        optimizer.step()
        total_loss += loss.item() * batch.num_graphs
    return total_loss / len(train_loader.dataset)

@torch.no_grad()
def evaluate(loader):
    model.eval()
    total_mae = 0
    for batch in loader:
        batch = batch.to(device)
        out = model(batch.x, batch.edge_index, batch.batch)
        total_mae += (out.view(-1) - batch.y.view(-1)).abs().sum().item()
    return total_mae / len(loader.dataset)

for epoch in range(1, 101):
    train_loss = train()
    if epoch % 20 == 0:
        test_mae = evaluate(test_loader)
        print(f'Epoch {epoch:03d}, Train MSE: {train_loss:.4f}, Test MAE: {test_mae:.4f}')

出力例：

Epoch 020, Train MSE: 1.1832, Test MAE: 0.8214
Epoch 040, Train MSE: 0.7145, Test MAE: 0.6839
Epoch 060, Train MSE: 0.5203, Test MAE: 0.6215
Epoch 080, Train MSE: 0.4109, Test MAE: 0.5981
Epoch 100, Train MSE: 0.3542, Test MAE: 0.5820

2. 引用ネットワーク分析

第1章ではCoraデータセットを使ってノード分類を行いました。本節では引用ネットワーク分析をさらに一歩進め、（1）ノード埋め込みの可視化による定性的な分析と、（2）まだ存在しない引用関係を予測するリンク予測（Link Prediction）を実装します。リンク予測は「この論文とあの論文は将来引用関係を持つ可能性が高い」といった関連論文推薦にそのまま応用できます。

リンク予測タスクの設計

リンク予測では、既知のエッジの一部を隠し、ノード埋め込みからエッジの存在確率を予測できるようにモデルを学習します。PyGのRandomLinkSplitを使うと、訓練・検証・テスト用にエッジを分割し、負例（存在しないエッジ）も自動生成できます。

import torch
import torch.nn.functional as F
from sklearn.metrics import roc_auc_score
from torch_geometric.datasets import Planetoid
from torch_geometric.transforms import RandomLinkSplit, NormalizeFeatures
from torch_geometric.nn import GCNConv
from torch_geometric.utils import negative_sampling

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

split = RandomLinkSplit(num_val=0.05, num_test=0.1, is_undirected=True,
                         add_negative_train_samples=False)
train_data, val_data, test_data = split(data)

print(f"Train edges: {train_data.edge_label_index.size(1)}")
print(f"Val edges (pos+neg): {val_data.edge_label_index.size(1)}")
print(f"Test edges (pos+neg): {test_data.edge_label_index.size(1)}")

エンコーダーとリンク予測モデル

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

    def encode(self, x, edge_index):
        x = self.conv1(x, edge_index).relu()
        x = self.conv2(x, edge_index)
        return x

    def decode(self, z, edge_label_index):
        src, dst = edge_label_index
        return (z[src] * z[dst]).sum(dim=-1)  # 内積でエッジスコアを計算

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = LinkPredictionGCN(dataset.num_features, 64, 32).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

train_data = train_data.to(device)
val_data = val_data.to(device)
test_data = test_data.to(device)

def train_step():
    model.train()
    optimizer.zero_grad()
    z = model.encode(train_data.x, train_data.edge_index)

    # 訓練エッジと同数の負例(存在しないエッジ)をサンプリング
    neg_edge_index = negative_sampling(
        edge_index=train_data.edge_index,
        num_nodes=train_data.num_nodes,
        num_neg_samples=train_data.edge_label_index.size(1),
        method='sparse')

    edge_label_index = torch.cat([train_data.edge_label_index, neg_edge_index], dim=-1)
    edge_label = torch.cat([
        train_data.edge_label,
        train_data.edge_label.new_zeros(neg_edge_index.size(1))
    ], dim=0)

    out = model.decode(z, edge_label_index)
    loss = F.binary_cross_entropy_with_logits(out, edge_label)
    loss.backward()
    optimizer.step()
    return loss.item()

@torch.no_grad()
def eval_step(eval_data):
    model.eval()
    z = model.encode(eval_data.x, eval_data.edge_index)
    out = model.decode(z, eval_data.edge_label_index).sigmoid()
    return roc_auc_score(eval_data.edge_label.cpu().numpy(), out.cpu().numpy())

for epoch in range(1, 101):
    loss = train_step()
    if epoch % 20 == 0:
        val_auc = eval_step(val_data)
        print(f'Epoch {epoch:03d}, Loss: {loss:.4f}, Val AUC: {val_auc:.4f}')

test_auc = eval_step(test_data)
print(f'\nTest AUC: {test_auc:.4f}')

出力例：

Epoch 020, Loss: 0.5842, Val AUC: 0.8215
Epoch 040, Loss: 0.4931, Val AUC: 0.8603
Epoch 060, Loss: 0.4520, Val AUC: 0.8791
Epoch 080, Loss: 0.4288, Val AUC: 0.8874
Epoch 100, Loss: 0.4109, Val AUC: 0.8902

Test AUC: 0.8856

AUC(Area Under the Curve、ROC曲線下の面積)が0.89前後であれば、モデルは正例(実在する引用)と負例(存在しない引用)をかなり良く区別できていると言えます。この仕組みは「この論文を読んでいる人へのおすすめ論文」のような機能にそのまま活用できます。

ノード埋め込みの可視化

学習済みモデルが獲得した埋め込み空間を2次元に圧縮して可視化すると、同じ分野の論文が近くに集まっているかを定性的に確認できます。

import matplotlib
matplotlib.use('Agg')  # 画面表示なしでファイル保存するためのバックエンド
import matplotlib.pyplot as plt
from sklearn.decomposition import PCA

model.eval()
with torch.no_grad():
    z = model.encode(data.x.to(device), data.edge_index.to(device)).cpu().numpy()

pca = PCA(n_components=2)
z_2d = pca.fit_transform(z)

plt.figure(figsize=(8, 6))
scatter = plt.scatter(z_2d[:, 0], z_2d[:, 1], c=data.y.numpy(), cmap='tab10', s=15)
plt.legend(*scatter.legend_elements(), title="クラス", loc='best')
plt.title('Coraノード埋め込みのPCA可視化')
plt.xlabel('主成分1')
plt.ylabel('主成分2')
plt.tight_layout()
plt.savefig('cora_embeddings.png', dpi=150)
print("埋め込み可視化を cora_embeddings.png に保存しました")

3. レコメンデーションシステム

レコメンデーションシステム（Recommendation System）は、ユーザーとアイテムの相互作用（購入、視聴、評価など）を二部グラフ（Bipartite Graph）として表現することで、GNNの得意分野になります。ユーザーノードとアイテムノードの間にメッセージパッシングを行うことで、「似た嗜好を持つユーザーが好んだアイテム」の情報を自然に伝播できます。

二部グラフの構築

ここでは学習用に、ユーザーとアイテムの相互作用を模した合成データを作成します。ユーザーノードとアイテムノードを同じインデックス空間に配置し(アイテムのインデックスはユーザー数だけオフセット)、1つの同種グラフとして扱います。

import torch

torch.manual_seed(0)

num_users = 300
num_items = 150
num_raw_interactions = 3000

user_ids = torch.randint(0, num_users, (num_raw_interactions,))
item_ids = torch.randint(0, num_items, (num_raw_interactions,))

# 重複する(user, item)ペアを除去
interactions = torch.unique(torch.stack([user_ids, item_ids], dim=1), dim=0)
num_interactions = interactions.size(0)
print(f"ユニークな相互作用数: {num_interactions}")

item_node_offset = num_users
edge_user = interactions[:, 0]
edge_item = interactions[:, 1] + item_node_offset
num_nodes = num_users + num_items

# 訓練/テストに8:2で分割
perm = torch.randperm(num_interactions)
train_size = int(num_interactions * 0.8)
train_idx, test_idx = perm[:train_size], perm[train_size:]

train_user, train_item = edge_user[train_idx], edge_item[train_idx]
test_user, test_item = edge_user[test_idx], edge_item[test_idx]

# メッセージパッシング用グラフは訓練用相互作用のみから構築(双方向)
train_edge_index = torch.cat([
    torch.stack([train_user, train_item], dim=0),
    torch.stack([train_item, train_user], dim=0),
], dim=1)

print(f"訓練エッジ数(双方向込み): {train_edge_index.size(1)}")

グラフベースレコメンダーの実装

ユーザー・アイテムそれぞれに学習可能な初期埋め込み(Embedding)を持たせ、GraphSAGE層で近傍情報を集約したのち、内積でスコアを計算します。学習にはBPR損失（Bayesian Personalized Ranking Loss）を使い、「実際に相互作用したアイテム」のスコアが「ランダムに選んだ未相互作用アイテム」より高くなるように最適化します。

$$\mathcal{L}_{\text{BPR}} = -\sum_{(u, i, j)} \log \sigma(\hat{y}_{ui} - \hat{y}_{uj})$$

import torch.nn.functional as F
from torch_geometric.nn import SAGEConv

class GraphRecommender(torch.nn.Module):
    def __init__(self, num_nodes, embedding_dim, hidden_dim):
        super().__init__()
        self.embedding = torch.nn.Embedding(num_nodes, embedding_dim)
        self.conv1 = SAGEConv(embedding_dim, hidden_dim)
        self.conv2 = SAGEConv(hidden_dim, hidden_dim)

    def forward(self, edge_index):
        x = self.embedding.weight
        x = self.conv1(x, edge_index).relu()
        x = self.conv2(x, edge_index)
        return x

    def score(self, z, users, items):
        return (z[users] * z[items]).sum(dim=-1)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = GraphRecommender(num_nodes, embedding_dim=32, hidden_dim=32).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

train_edge_index_d = train_edge_index.to(device)
train_user_d = train_user.to(device)
train_item_d = train_item.to(device)

def sample_negative_items(users):
    # ユーザー数分、ランダムなアイテムを負例としてサンプリング
    return torch.randint(0, num_items, (users.size(0),), device=users.device) + item_node_offset

model.train()
for epoch in range(1, 51):
    optimizer.zero_grad()
    z = model(train_edge_index_d)

    pos_scores = model.score(z, train_user_d, train_item_d)
    neg_items = sample_negative_items(train_user_d)
    neg_scores = model.score(z, train_user_d, neg_items)

    loss = -F.logsigmoid(pos_scores - neg_scores).mean()
    loss.backward()
    optimizer.step()

    if epoch % 10 == 0:
        print(f'Epoch {epoch:02d}, BPR Loss: {loss.item():.4f}')

Recall@Kによる評価

@torch.no_grad()
def recall_at_k(z, test_user, test_item, train_user, train_item, k=10, num_eval_users=50):
    unique_test_users = torch.unique(test_user)[:num_eval_users]
    hits, total = 0, 0

    for u in unique_test_users:
        true_items = test_item[test_user == u]
        if true_items.numel() == 0:
            continue

        known_items = train_item[train_user == u] - item_node_offset
        item_embeddings = z[item_node_offset: item_node_offset + num_items]
        scores = (z[u] * item_embeddings).sum(dim=-1)
        scores[known_items] = -1e9  # 既知の相互作用は推薦候補から除外

        topk_items = torch.topk(scores, k).indices + item_node_offset
        hit = any(t.item() in topk_items.tolist() for t in true_items)
        hits += int(hit)
        total += 1

    return hits / total if total > 0 else 0.0

model.eval()
z = model(train_edge_index_d)
recall = recall_at_k(z, test_user.to(device), test_item.to(device),
                      train_user_d, train_item_d, k=10)
print(f'Recall@10: {recall:.4f}')

4. 大規模グラフの効率的処理

Coraのような数千ノードのグラフであれば全ノードを一度にGPUメモリへ載せて学習するフルバッチ学習（Full-batch Training）が可能です。しかし、実務で扱うグラフはノード数が数百万〜数億に達することも珍しくありません。このような大規模グラフ（Large-scale Graph）では、全ノードとその特徴量をメモリに載せることが物理的に不可能になります。

ミニバッチ学習と近傍サンプリング

この問題を解決するのが近傍サンプリング（Neighbor Sampling）に基づくミニバッチ学習です。各ミニバッチでは、対象ノード（シードノード）を選び、そこから一定数の近傍ノードだけを層ごとにサンプリングして、必要な部分グラフのみを計算に使います。PyGではNeighborLoaderがこれを実現します。

ここでは大規模グラフの雰囲気をつかむため、5,000ノードの合成グラフを生成し、NeighborLoaderによるミニバッチ学習を実装します。同じコードパターンは、ノード数が数百万に達する実際の大規模グラフ(例: ogbn-productsやReddit)にもそのまま適用できます。

import torch
import torch.nn.functional as F
from torch_geometric.data import Data
from torch_geometric.utils import erdos_renyi_graph
from torch_geometric.loader import NeighborLoader
from torch_geometric.nn import SAGEConv

torch.manual_seed(42)

num_nodes = 5000
num_features = 32
num_classes = 5

edge_index = erdos_renyi_graph(num_nodes, edge_prob=0.002, directed=False)
x = torch.randn(num_nodes, num_features)
y = torch.randint(0, num_classes, (num_nodes,))

perm = torch.randperm(num_nodes)
train_mask = torch.zeros(num_nodes, dtype=torch.bool)
val_mask = torch.zeros(num_nodes, dtype=torch.bool)
test_mask = torch.zeros(num_nodes, dtype=torch.bool)
train_mask[perm[:3000]] = True
val_mask[perm[3000:4000]] = True
test_mask[perm[4000:]] = True

data = Data(x=x, edge_index=edge_index, y=y,
            train_mask=train_mask, val_mask=val_mask, test_mask=test_mask)

print(data)
print(f"平均次数: {data.num_edges / data.num_nodes:.2f}")

NeighborLoaderによるミニバッチ学習

train_loader = NeighborLoader(
    data,
    num_neighbors=[10, 10],  # 各層でサンプリングする近傍数(1層目10個、2層目10個)
    batch_size=256,
    input_nodes=data.train_mask,
    shuffle=True,
)

class SAGE(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 = self.conv1(x, edge_index).relu()
        x = F.dropout(x, p=0.5, training=self.training)
        x = self.conv2(x, edge_index)
        return x

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = SAGE(num_features, hidden_channels=64, out_channels=num_classes).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

model.train()
for epoch in range(1, 6):
    total_loss = 0
    for batch in train_loader:
        batch = batch.to(device)
        optimizer.zero_grad()
        out = model(batch.x, batch.edge_index)

        # NeighborLoaderのバッチでは、先頭batch.batch_size個がシードノード
        loss = F.cross_entropy(out[:batch.batch_size], batch.y[:batch.batch_size])
        loss.backward()
        optimizer.step()
        total_loss += loss.item() * batch.batch_size

    avg_loss = total_loss / int(data.train_mask.sum())
    print(f'Epoch {epoch:02d}, Loss: {avg_loss:.4f}')

出力例：

Epoch 01, Loss: 1.6132
Epoch 02, Loss: 1.5890
Epoch 03, Loss: 1.5701
Epoch 04, Loss: 1.5523
Epoch 05, Loss: 1.5388

5. モデルの解釈性と可視化

GNNは高い予測精度を達成できる一方で、「なぜその予測をしたのか」を説明することが難しいブラックボックスモデルになりがちです。特に創薬や医療などの意思決定に関わる領域では、モデルの解釈性（Interpretability）が実用上の必須要件になります。PyGはtorch_geometric.explainモジュールで、代表的な説明手法であるGNNExplainerを提供しています。

GNNExplainerによる予測根拠の可視化

GNNExplainerは、特定ノードの予測に対して「どのエッジ」と「どの特徴量」が重要だったかを、小さなマスク(重み)を学習することで明らかにします。

import torch
import torch.nn.functional as F
from torch_geometric.datasets import Planetoid
from torch_geometric.nn import GCNConv
from torch_geometric.explain import Explainer, GNNExplainer

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

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

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

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = GCN(dataset.num_features, dataset.num_classes).to(device)
data = data.to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=5e-4)

# 簡易学習(解釈性デモのため短めのエポック数)
model.train()
for epoch in range(100):
    optimizer.zero_grad()
    out = model(data.x, data.edge_index)
    loss = F.cross_entropy(out[data.train_mask], data.y[data.train_mask])
    loss.backward()
    optimizer.step()

print(f"Final train loss: {loss.item():.4f}")

explainer = Explainer(
    model=model,
    algorithm=GNNExplainer(epochs=200),
    explanation_type='model',
    node_mask_type='attributes',
    edge_mask_type='object',
    model_config=dict(
        mode='multiclass_classification',
        task_level='node',
        return_type='raw',
    ),
)

# ノード10番の予測根拠を説明
node_index = 10
explanation = explainer(data.x, data.edge_index, index=node_index)

print(f"説明対象ノード: {node_index}")
print(f"重要な特徴量マスクの形状: {explanation.node_mask.shape}")
print(f"重要なエッジマスクの形状: {explanation.edge_mask.shape}")

# 重要度の高い特徴量トップ5
top_features = explanation.node_mask[node_index].topk(5)
print(f"\n重要度の高い特徴量インデックス: {top_features.indices.tolist()}")
print(f"重要度スコア: {[round(v, 4) for v in top_features.values.tolist()]}")

# 重要度の高いエッジトップ5
top_edges = explanation.edge_mask.topk(5)
print(f"\n重要度の高いエッジのインデックス: {top_edges.indices.tolist()}")

出力例：

説明対象ノード: 10
重要な特徴量マスクの形状: torch.Size([2708, 1433])
重要なエッジマスクの形状: torch.Size([10556])

重要度の高い特徴量インデックス: [19, 142, 501, 88, 973]
重要度スコア: [0.8123, 0.7654, 0.7211, 0.6987, 0.6544]

重要度の高いエッジのインデックス: [3421, 128, 5502, 891, 2077]

説明サブグラフの可視化

import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import networkx as nx
from torch_geometric.utils import to_networkx

# 重要度上位のエッジだけを抜き出したサブグラフを可視化
threshold = explanation.edge_mask.topk(15).values.min()
important_edge_mask = explanation.edge_mask >= threshold
important_edges = data.edge_index[:, important_edge_mask].cpu()

involved_nodes = torch.unique(important_edges).tolist()
if node_index not in involved_nodes:
    involved_nodes.append(node_index)

subgraph = nx.Graph()
subgraph.add_nodes_from(involved_nodes)
for src, dst in important_edges.t().tolist():
    subgraph.add_edge(src, dst)

colors = ['#e74c3c' if n == node_index else '#3498db' for n in subgraph.nodes()]

plt.figure(figsize=(7, 6))
nx.draw(subgraph, node_color=colors, with_labels=True, node_size=400,
        font_size=8, font_color='white')
plt.title(f'ノード{node_index}の予測に重要なサブグラフ')
plt.savefig('gnn_explanation.png', dpi=150, bbox_inches='tight')
print("説明サブグラフを gnn_explanation.png に保存しました")

6. 本番環境へのデプロイ

研究・実験段階で高精度なGNNモデルが完成しても、それを本番環境(Production Environment)で安定的に稼働させるには、学習時とは異なる考慮事項が数多くあります。ここでは分子特性予測モデル(第1節のESOLモデル)を例に、実務でのデプロイパターンを解説します。

推論用モデルの保存とロード

import torch

# モデルの重みだけを保存(推奨される保存方法)
torch.save(model.state_dict(), 'molecular_gin_esol.pt')

# ロード時はモデル構造を再定義してから重みを読み込む
loaded_model = MolecularGIN(num_features=9, hidden_channels=64)
loaded_model.load_state_dict(torch.load('molecular_gin_esol.pt', map_location='cpu'))
loaded_model.eval()

print("モデルのロードが完了しました")

SMILESから予測までの推論パイプライン

本番環境では、学習時と同じ前処理(SMILES文字列から分子グラフへの変換)を推論時にも再現する必要があります。前処理と後処理を1つの関数にまとめておくと、デプロイ時の事故を防げます。

import torch
from torch_geometric.datasets import MoleculeNet
from torch_geometric.loader import DataLoader

# MoleculeNetの特徴量抽出ロジックを再利用して、
# 学習時と全く同じ前処理を1分子だけに適用する例
def predict_solubility(model, molecule_data, device='cpu'):
    """1分子分の推論を行うヘルパー関数

    Args:
        model: 学習済みのMolecularGINモデル(eval済み)
        molecule_data: torch_geometric.data.Data (x, edge_index を含む)
        device: 推論に使うデバイス

    Returns:
        float: 予測された対数溶解度
    """
    model = model.to(device)
    molecule_data = molecule_data.to(device)

    with torch.no_grad():
        x = molecule_data.x.float()
        # 単一グラフの場合、全ノードが同じグラフIDに属することを示すバッチベクトルが必要
        batch = torch.zeros(x.size(0), dtype=torch.long, device=device)
        pred = model(x, molecule_data.edge_index, batch)

    return pred.item()

# 検証用データセットの分子で推論パイプラインを確認
dataset = MoleculeNet(root='/tmp/ESOL', name='ESOL')
sample_molecule = dataset[0]

prediction = predict_solubility(loaded_model, sample_molecule)
print(f"SMILES: {sample_molecule.smiles}")
print(f"予測された対数溶解度: {prediction:.4f}")
print(f"実測値: {sample_molecule.y.item():.4f}")

本番デプロイのチェックリスト

演習問題

import torch
from torch.nn import Linear, Sequential, ReLU
from torch_geometric.nn import GINEConv, global_mean_pool

class MolecularGINE(torch.nn.Module):
    def __init__(self, num_features, num_edge_features, hidden_channels):
        super().__init__()
        nn1 = Sequential(Linear(hidden_channels, hidden_channels), ReLU(),
                          Linear(hidden_channels, hidden_channels))
        # GINEConvは入力特徴量とエッジ特徴量の次元を揃える必要があるため、
        # 事前に線形層で次元をそろえる
        self.node_proj = Linear(num_features, hidden_channels)
        self.edge_proj = Linear(num_edge_features, hidden_channels)
        self.conv1 = GINEConv(nn1)

        self.lin1 = Linear(hidden_channels, hidden_channels)
        self.lin2 = Linear(hidden_channels, 1)

    def forward(self, x, edge_index, edge_attr, batch):
        x = self.node_proj(x)
        edge_attr = self.edge_proj(edge_attr)
        x = self.conv1(x, edge_index, edge_attr).relu()
        x = global_mean_pool(x, batch)
        x = self.lin1(x).relu()
        x = self.lin2(x)
        return x

# 使用例:
# model = MolecularGINE(num_features=9, num_edge_features=3, hidden_channels=64)
# out = model(batch.x.float(), batch.edge_index, batch.edge_attr.float(), batch.batch)

import time
import torch.nn.functional as F
from torch_geometric.loader import NeighborLoader

def run_experiment(data, num_neighbors, num_features, num_classes, device):
    loader = NeighborLoader(
        data, num_neighbors=num_neighbors, batch_size=256,
        input_nodes=data.train_mask, shuffle=True,
    )
    model = SAGE(num_features, 64, num_classes).to(device)
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

    start = time.time()
    model.train()
    total_loss = 0
    for batch in loader:
        batch = batch.to(device)
        optimizer.zero_grad()
        out = model(batch.x, batch.edge_index)
        loss = F.cross_entropy(out[:batch.batch_size], batch.y[:batch.batch_size])
        loss.backward()
        optimizer.step()
        total_loss += loss.item() * batch.batch_size
    elapsed = time.time() - start
    avg_loss = total_loss / int(data.train_mask.sum())
    return elapsed, avg_loss

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
for neighbors in ([5, 5], [10, 10], [25, 25]):
    elapsed, avg_loss = run_experiment(data, neighbors, num_features, num_classes, device)
    print(f"num_neighbors={neighbors}: time={elapsed:.2f}s, loss={avg_loss:.4f}")

import torch

model.eval()
with torch.no_grad():
    pred = model(data.x, data.edge_index).argmax(dim=1)

correct_mask = (pred == data.y) & data.test_mask
incorrect_mask = (pred != data.y) & data.test_mask

correct_node = correct_mask.nonzero(as_tuple=True)[0][0].item()
incorrect_candidates = incorrect_mask.nonzero(as_tuple=True)[0]

if incorrect_candidates.numel() > 0:
    incorrect_node = incorrect_candidates[0].item()

    for label, node_idx in [("正しく分類", correct_node), ("誤分類", incorrect_node)]:
        explanation = explainer(data.x, data.edge_index, index=node_idx)
        top_edges = explanation.edge_mask.topk(5)
        print(f"\n[{label}] ノード{node_idx}")
        print(f"  真のラベル: {data.y[node_idx].item()}, 予測: {pred[node_idx].item()}")
        print(f"  重要エッジのインデックス: {top_edges.indices.tolist()}")
else:
    print("テストセット中に誤分類ノードが見つかりませんでした(モデル精度が非常に高い場合に発生)")

まとめ

この章では、GNNを実世界の問題に応用するための実践的な手法を学びました。

• ✅ MoleculeNet(ESOL)とGINConvを使った分子特性予測(回帰タスク)
• ✅ RandomLinkSplitとGCNエンコーダーによる引用ネットワークのリンク予測、およびPCAによる埋め込み可視化
• ✅ 二部グラフとGraphSAGE、BPR損失を用いたグラフベースレコメンデーション
• ✅ NeighborLoaderによる近傍サンプリングを使った大規模グラフのミニバッチ学習
• ✅ GNNExplainerによる予測根拠の可視化とモデル解釈性の向上
• ✅ モデル保存・推論パイプライン構築、本番デプロイ時の実践的な注意点

参考リソース

• PyTorch Geometric公式ドキュメント
• PyG Explainability公式チュートリアル
• MoleculeNet: ベンチマークデータセット
• GIN論文: How Powerful are Graph Neural Networks?
• GraphSAGE論文: Inductive Representation Learning on Large Graphs