图神经网络入门实战：消息传递、GCN 与 GAT

import numpy as np
# 3 个节点：0-1 相连，1-2 相连
A = np.array([[0, 1, 0], [1, 0, 1], [0, 1, 0]], dtype=float)
print(A.sum(axis=1))    # 每个节点的度：[1, 2, 1]

import torch

# edge_index 形状 (2, E)，第 0 行源、第 1 行目标（无向图两方向都存）
edge_index = torch.tensor([[0, 1, 1, 2],
                           [1, 0, 2, 1]], dtype=torch.long)

D^-1 A           行归一化，每行和为 1
D^-1/2 A D^-1/2  对称归一化，GCN 用的就是这个

对每一层 k：
  1. 消息生成  m_ij = M(h_i, h_j, e_ij)        邻居 j 发来的消息
  2. 消息聚合  m_i  = AGG({m_ij : j 属于 N(i)})  求和/均值/最大/注意力
  3. 节点更新  h_i' = U(h_i, m_i)              通常接一个线性层+激活

import torch

def simple_message_passing(h, edge_index):
    """最朴素的均值聚合：h_new[i] = mean(h[j] for j in N(i))"""
    src, dst = edge_index[0], edge_index[1]
    out = torch.zeros_like(h)
    out.index_add_(0, dst, h[src])          # 把邻居特征累加到目标节点
    deg = torch.zeros(h.size(0), device=h.device)
    deg.index_add_(0, dst, torch.ones(src.size(0), device=h.device))
    return out / deg.clamp(min=1).unsqueeze(-1)

def add_self_loops(edge_index, num_nodes):
    loop = torch.arange(num_nodes, device=edge_index.device)
    return torch.cat([edge_index, loop.unsqueeze(0).repeat(2, 1)], dim=1)

H' = sigma( A_hat H W )
A_hat = D_hat^-1/2 (A + I) D_hat^-1/2     ← 加自环后对称归一化

import torch.nn as nn
from torch_geometric.nn import GCNConv

class GCN(nn.Module):
    def __init__(self, in_dim, hidden, out_dim):
        super().__init__()
        self.conv1 = GCNConv(in_dim, hidden)
        self.conv2 = GCNConv(hidden, out_dim)
    def forward(self, x, edge_index):
        x = self.conv1(x, edge_index).relu()
        x = nn.functional.dropout(x, p=0.5, training=self.training)
        return self.conv2(x, edge_index)

e_ij = LeakyReLU( a^T [ W h_i || W h_j ] )      拼接后过单层网络
alpha_ij = softmax_j(e_ij)                       在同一节点的邻居间归一化
h_i' = sigma( sum_j alpha_ij W h_j )             加权求和

from torch_geometric.nn import GATConv

class GAT(nn.Module):
    def __init__(self, in_dim, hidden, out_dim, heads=8):
        super().__init__()
        self.conv1 = GATConv(in_dim, hidden, heads=heads, dropout=0.6)
        # 输出层用单头并平均，避免维度爆炸
        self.conv2 = GATConv(hidden * heads, out_dim, heads=1,
                             concat=False, dropout=0.6)
    def forward(self, x, edge_index):
        x = self.conv1(x, edge_index).elu()
        return self.conv2(x, edge_index)

criterion = nn.CrossEntropyLoss()

def train_step(model, data, optimizer):
    model.train(); optimizer.zero_grad()
    out = model(data.x, data.edge_index)
    loss = criterion(out[data.train_mask], data.y[data.train_mask])
    loss.backward(); optimizer.step()
    return loss.item()

全局池化：h_G = mean / sum / max({h_i})
层次池化：DiffPool / TopKPool，边池化边学结构

from torch_geometric.nn import global_mean_pool

class GraphClassifier(nn.Module):
    def __init__(self, in_dim, hidden, num_classes):
        super().__init__()
        self.conv1 = GCNConv(in_dim, hidden)
        self.conv2 = GCNConv(hidden, hidden)
        self.head = nn.Linear(hidden, num_classes)
    def forward(self, x, edge_index, batch):
        x = self.conv1(x, edge_index).relu()
        x = self.conv2(x, edge_index).relu()
        x = global_mean_pool(x, batch)     # batch 指明每个节点属于哪张图
        return self.head(x)

import torch
from torch_geometric.datasets import Planetoid
from torch_geometric.transforms import NormalizeFeatures

dataset = Planetoid(root="/tmp/Cora", name="Cora",
                    transform=NormalizeFeatures())
data = dataset[0]
# Data(x=[2708, 1433], edge_index=[2, 10556], y=[2708],
#      train_mask=[2708], val_mask=[2708], test_mask=[2708])

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = GCN(dataset.num_features, 16, dataset.num_classes).to(device)
data = data.to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=5e-4)
def evaluate(mask):
    model.eval()
    with torch.no_grad():
        pred = model(data.x, data.edge_index).argmax(dim=1)
    return (pred[mask] == data.y[mask]).float().mean().item()

best_val = 0
for epoch in range(200):
    loss = train_step(model, data, optimizer)
    val_acc = evaluate(data.val_mask)
    if val_acc > best_val:
        best_val = val_acc
        torch.save(model.state_dict(), "best.pt")
    if epoch % 20 == 0:
        print(f"epoch {epoch:3d} loss {loss:.4f} val {val_acc:.4f}")
model.load_state_dict(torch.load("best.pt"))
print("test acc:", round(evaluate(data.test_mask), 4))
# 两层 GCN 在 Cora 上通常能到 0.80 左右

GCN（2 层）      Cora 测试准确率 约 0.80
GAT（2 层 8 头） 约 0.82~0.83
MLP（忽略结构）  约 0.55~0.60

hidden_dim 16~256（Cora 上 16 就够）  layers 2~3  dropout 0.5~0.6
lr 0.005~0.01  weight_decay 5e-4  epochs 200（早停看验证集）

图的表示 → 消息传递范式（消息-聚合-更新）
  → GCN（归一化 + 自环，2 层甜点）→ GAT（注意力加权）
  → 节点分类（半监督掩码）/ 图分类（全局池化）
  → 大图采样（NeighborLoader）