图神经网络（GNN）技术详解：从消息传递到现代图学习

摘要

图神经网络（Graph Neural Networks, GNN）是专门用于处理图结构数据的深度学习架构，通过消息传递机制学习节点和图的表示。本文深入解析GNN的核心原理、消息传递范式、主要变体以及从基础理论到现代应用的发展历程，帮助读者全面理解这一重要技术。

关键词： 图神经网络、GNN、消息传递、图学习、节点嵌入

---

1. 引言

图神经网络（GNN）是一类专门设计用于处理图结构数据的神经网络架构。与传统的深度学习模型不同，GNN能够直接处理非欧几里得空间中的图数据，在社交网络分析、推荐系统、分子性质预测、知识图谱等领域取得了显著成果。

1.1 GNN的发展历程

• 2005年: 图神经网络的概念首次提出
• 2009年: 消息传递神经网络的理论基础建立
• 2016年: GCN的提出标志着现代GNN的兴起
• 2017年至今: 各种GNN变体的快速发展

---

2. 图神经网络的基础概念

2.1 图的基本定义

图 $G=(V,E)$ 由节点集合 $V$ 和边集合 $E$ 组成：

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
import networkx as nx
import matplotlib.pyplot as plt

class Graph:
    def __init__(self, num_nodes, edges=None, node_features=None):
        self.num_nodes = num_nodes
        self.adjacency_matrix = torch.zeros(num_nodes, num_nodes)
        self.node_features = node_features if node_features is not None else torch.randn(num_nodes, 1)
        
        if edges is not None:
            for edge in edges:
                self.adjacency_matrix[edge[0], edge[1]] = 1
                self.adjacency_matrix[edge[1], edge[0]] = 1  # 无向图
    
    def get_neighbors(self, node_id):
        """获取节点的邻居"""
        return torch.nonzero(self.adjacency_matrix[node_id]).flatten()
    
    def get_degree(self, node_id):
        """获取节点的度"""
        return torch.sum(self.adjacency_matrix[node_id]).item()

# 创建示例图
edges = [(0, 1), (1, 2), (2, 3), (3, 0), (0, 2), (1, 3)]
graph = Graph(4, edges)

print("邻接矩阵:")
print(graph.adjacency_matrix)
print("\n节点特征:")
print(graph.node_features)
print(f"\n节点0的邻居: {graph.get_neighbors(0)}")
print(f"节点0的度: {graph.get_degree(0)}")

2.2 消息传递范式

GNN的核心是消息传递范式，包括三个步骤：

1. 消息（Message）: 节点向其邻居发送消息
2. 聚合（Aggregate）: 节点聚合来自邻居的消息
3. 更新（Update）: 节点根据聚合的消息更新自身状态

class MessagePassing(nn.Module):
    def __init__(self, in_channels, out_channels, aggr='add'):
        super().__init__()
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.aggr = aggr
        
        # 消息函数
        self.message_mlp = nn.Linear(in_channels, out_channels)
        
        # 更新函数
        self.update_mlp = nn.Linear(in_channels + out_channels, out_channels)
    
    def message(self, x_j):
        """消息函数：从邻居节点j发送消息"""
        return self.message_mlp(x_j)
    
    def aggregate(self, messages, index, num_nodes):
        """聚合函数：聚合来自邻居的消息"""
        if self.aggr == 'add':
            return torch.zeros(num_nodes, messages.size(1)).scatter_add_(0, index.unsqueeze(1).expand_as(messages), messages)
        elif self.aggr == 'mean':
            return torch.zeros(num_nodes, messages.size(1)).scatter_add_(0, index.unsqueeze(1).expand_as(messages), messages) / torch.bincount(index, minlength=num_nodes).float().unsqueeze(1)
        elif self.aggr == 'max':
            return torch.zeros(num_nodes, messages.size(1)).scatter_reduce_(0, index.unsqueeze(1).expand_as(messages), messages, reduce='amax')
    
    def update(self, x, aggregated_messages):
        """更新函数：根据聚合的消息更新节点状态"""
        combined = torch.cat([x, aggregated_messages], dim=1)
        return self.update_mlp(combined)
    
    def forward(self, x, edge_index):
        """
        x: 节点特征矩阵 (N, in_channels)
        edge_index: 边索引 (2, E)
        """
        row, col = edge_index
        
        # 消息传递
        messages = self.message(x[col])
        
        # 聚合消息
        aggregated = self.aggregate(messages, row, x.size(0))
        
        # 更新节点状态
        updated = self.update(x, aggregated)
        
        return updated

# 使用示例
mp = MessagePassing(in_channels=4, out_channels=8, aggr='add')
x = torch.randn(4, 4)  # 4个节点，每个节点4维特征
edge_index = torch.tensor([[0, 1, 2, 3, 0, 2], [1, 2, 3, 0, 2, 0]])  # 边索引

output = mp(x, edge_index)
print(f"输入特征形状: {x.shape}")
print(f"输出特征形状: {output.shape}")

---

3. 经典GNN架构

3.1 图卷积网络（GCN）

GCN是最经典的GNN架构之一：

class GCNLayer(nn.Module):
    def __init__(self, in_channels, out_channels, bias=True):
        super().__init__()
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.weight = nn.Parameter(torch.FloatTensor(in_channels, out_channels))
        
        if bias:
            self.bias = nn.Parameter(torch.FloatTensor(out_channels))
        else:
            self.register_parameter('bias', None)
        
        self.reset_parameters()
    
    def reset_parameters(self):
        nn.init.xavier_uniform_(self.weight)
        if self.bias is not None:
            nn.init.zeros_(self.bias)
    
    def forward(self, x, adj):
        """
        x: 节点特征 (N, in_channels)
        adj: 归一化邻接矩阵 (N, N)
        """
        # 线性变换
        support = torch.mm(x, self.weight)
        
        # 图卷积
        output = torch.spmm(adj, support)
        
        if self.bias is not None:
            output += self.bias
        
        return output

def normalize_adjacency(adjacency_matrix):
    """归一化邻接矩阵"""
    # 添加自环
    adj = adjacency_matrix + torch.eye(adjacency_matrix.shape[0])
    
    # 计算度矩阵
    degree = torch.sum(adj, dim=1)
    degree_inv_sqrt = torch.pow(degree, -0.5)
    degree_inv_sqrt[torch.isinf(degree_inv_sqrt)] = 0.0
    
    # 归一化
    degree_matrix_inv_sqrt = torch.diag(degree_inv_sqrt)
    normalized_adj = degree_matrix_inv_sqrt @ adj @ degree_matrix_inv_sqrt
    
    return normalized_adj

class GCN(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, num_layers=2, dropout=0.5):
        super().__init__()
        self.layers = nn.ModuleList()
        
        # 输入层
        self.layers.append(GCNLayer(input_dim, hidden_dim))
        
        # 隐藏层
        for _ in range(num_layers - 2):
            self.layers.append(GCNLayer(hidden_dim, hidden_dim))
        
        # 输出层
        self.layers.append(GCNLayer(hidden_dim, output_dim))
        
        self.dropout = nn.Dropout(dropout)
        self.relu = nn.ReLU()
    
    def forward(self, x, adj):
        for i, layer in enumerate(self.layers[:-1]):
            x = layer(x, adj)
            x = self.relu(x)
            x = self.dropout(x)
        
        x = self.layers[-1](x, adj)
        return x

# 使用示例
gcn = GCN(input_dim=4, hidden_dim=16, output_dim=2)
x = torch.randn(4, 4)
adj = torch.tensor([
    [0, 1, 1, 0],
    [1, 0, 1, 1],
    [1, 1, 0, 1],
    [0, 1, 1, 0]
], dtype=torch.float32)
normalized_adj = normalize_adjacency(adj)

output = gcn(x, normalized_adj)
print(f"GCN输出形状: {output.shape}")

3.2 图注意力网络（GAT）

GAT引入了注意力机制来学习邻居节点的重要性：

class GATLayer(nn.Module):
    def __init__(self, in_channels, out_channels, dropout=0.6, alpha=0.2):
        super().__init__()
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.dropout = dropout
        self.alpha = alpha
        
        # 线性变换
        self.W = nn.Linear(in_channels, out_channels, bias=False)
        
        # 注意力机制
        self.a = nn.Linear(2 * out_channels, 1, bias=False)
        
        self.leakyrelu = nn.LeakyReLU(self.alpha)
        self.dropout_layer = nn.Dropout(dropout)
    
    def forward(self, x, adj):
        """
        x: 节点特征 (N, in_channels)
        adj: 邻接矩阵 (N, N)
        """
        h = self.W(x)  # (N, out_channels)
        N = h.size(0)
        
        # 计算注意力系数
        a_input = self._prepare_attentional_mechanism_input(h)
        e = self.leakyrelu(self.a(a_input).squeeze(2))  # (N, N)
        
        # 应用邻接矩阵掩码
        zero_vec = -9e15 * torch.ones_like(e)
        attention = torch.where(adj > 0, e, zero_vec)
        attention = F.softmax(attention, dim=1)
        attention = self.dropout_layer(attention)
        
        # 应用注意力权重
        h_prime = torch.matmul(attention, h)
        
        return h_prime
    
    def _prepare_attentional_mechanism_input(self, h):
        N = h.size(0)
        h_repeated_in_chunks = h.repeat_interleave(N, dim=0)
        h_repeated_alternating = h.repeat(N, 1)
        all_combinations_matrix = torch.cat([h_repeated_in_chunks, h_repeated_alternating], dim=1)
        return all_combinations_matrix.view(N, N, 2 * self.out_channels)

class GAT(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, num_heads=8, dropout=0.6):
        super().__init__()
        self.dropout = dropout
        
        # 多头注意力层
        self.attention_layers = nn.ModuleList([
            GATLayer(input_dim, hidden_dim, dropout=dropout)
            for _ in range(num_heads)
        ])
        
        # 输出层
        self.out_attention = GATLayer(hidden_dim * num_heads, output_dim, dropout=dropout)
    
    def forward(self, x, adj):
        # 多头注意力
        x = torch.cat([att(x, adj) for att in self.attention_layers], dim=1)
        x = F.dropout(x, self.dropout, training=self.training)
        
        # 输出层
        x = self.out_attention(x, adj)
        return x

# 使用示例
gat = GAT(input_dim=4, hidden_dim=8, output_dim=2, num_heads=4)
output = gat(x, adj)
print(f"GAT输出形状: {output.shape}")

3.3 GraphSAGE

GraphSAGE使用采样和聚合策略：

class GraphSAGELayer(nn.Module):
    def __init__(self, in_channels, out_channels, aggr='mean'):
        super().__init__()
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.aggr = aggr
        
        # 邻居聚合器
        self.neighbor_aggregator = nn.Linear(in_channels, out_channels)
        
        # 自身变换
        self.self_transform = nn.Linear(in_channels, out_channels)
        
        # 组合函数
        self.combine = nn.Linear(2 * out_channels, out_channels)
    
    def forward(self, x, adj):
        # 聚合邻居信息
        neighbor_info = torch.spmm(adj, x)
        neighbor_emb = self.neighbor_aggregator(neighbor_info)
        
        # 自身变换
        self_emb = self.self_transform(x)
        
        # 组合
        combined = torch.cat([self_emb, neighbor_emb], dim=1)
        output = self.combine(combined)
        
        return output

class GraphSAGE(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, num_layers=2, dropout=0.5):
        super().__init__()
        self.layers = nn.ModuleList()
        
        # 输入层
        self.layers.append(GraphSAGELayer(input_dim, hidden_dim))
        
        # 隐藏层
        for _ in range(num_layers - 2):
            self.layers.append(GraphSAGELayer(hidden_dim, hidden_dim))
        
        # 输出层
        self.layers.append(GraphSAGELayer(hidden_dim, output_dim))
        
        self.dropout = nn.Dropout(dropout)
        self.relu = nn.ReLU()
    
    def forward(self, x, adj):
        for i, layer in enumerate(self.layers[:-1]):
            x = layer(x, adj)
            x = self.relu(x)
            x = self.dropout(x)
        
        x = self.layers[-1](x, adj)
        return x

# 使用示例
sage = GraphSAGE(input_dim=4, hidden_dim=16, output_dim=2)
output = sage(x, adj)
print(f"GraphSAGE输出形状: {output.shape}")

---

4. 图神经网络的应用

4.1 节点分类

class NodeClassifier(nn.Module):
    def __init__(self, input_dim, hidden_dim, num_classes, model_type='GCN'):
        super().__init__()
        self.model_type = model_type
        
        if model_type == 'GCN':
            self.gnn = GCN(input_dim, hidden_dim, num_classes)
        elif model_type == 'GAT':
            self.gnn = GAT(input_dim, hidden_dim, num_classes)
        elif model_type == 'GraphSAGE':
            self.gnn = GraphSAGE(input_dim, hidden_dim, num_classes)
    
    def forward(self, x, adj):
        return self.gnn(x, adj)
    
    def train_model(self, x, adj, labels, train_mask, val_mask, epochs=200):
        optimizer = torch.optim.Adam(self.parameters(), lr=0.01, weight_decay=5e-4)
        criterion = nn.CrossEntropyLoss()
        
        best_val_acc = 0
        for epoch in range(epochs):
            self.train()
            optimizer.zero_grad()
            
            output = self.forward(x, adj)
            loss = criterion(output[train_mask], labels[train_mask])
            
            loss.backward()
            optimizer.step()
            
            if epoch % 20 == 0:
                self.eval()
                with torch.no_grad():
                    val_output = self.forward(x, adj)
                    val_pred = val_output[val_mask].argmax(dim=1)
                    val_acc = (val_pred == labels[val_mask]).float().mean().item()
                    
                    if val_acc > best_val_acc:
                        best_val_acc = val_acc
                    
                    print(f'Epoch {epoch}, Loss: {loss.item():.4f}, Val Acc: {val_acc:.4f}')
        
        return best_val_acc

# 创建模拟数据
def create_synthetic_data(num_nodes=100, num_features=10, num_classes=3):
    # 生成随机图
    adj_matrix = torch.rand(num_nodes, num_nodes)
    adj_matrix = (adj_matrix + adj_matrix.T) / 2  # 对称化
    adj_matrix = (adj_matrix > 0.1).float()  # 二值化
    adj_matrix.fill_diagonal_(0)  # 移除自环
    
    # 生成节点特征
    x = torch.randn(num_nodes, num_features)
    
    # 生成标签（基于图结构）
    labels = torch.randint(0, num_classes, (num_nodes,))
    
    # 创建训练/验证掩码
    train_mask = torch.zeros(num_nodes, dtype=torch.bool)
    val_mask = torch.zeros(num_nodes, dtype=torch.bool)
    
    indices = torch.randperm(num_nodes)
    train_mask[indices[:70]] = True
    val_mask[indices[70:90]] = True
    
    return x, adj_matrix, labels, train_mask, val_mask

# 比较不同GNN模型
def compare_gnn_models():
    x, adj, labels, train_mask, val_mask = create_synthetic_data()
    normalized_adj = normalize_adjacency(adj)
    
    models = ['GCN', 'GAT', 'GraphSAGE']
    results = {}
    
    for model_type in models:
        print(f"\n训练 {model_type} 模型...")
        model = NodeClassifier(input_dim=10, hidden_dim=16, num_classes=3, model_type=model_type)
        best_acc = model.train_model(x, normalized_adj, labels, train_mask, val_mask, epochs=100)
        results[model_type] = best_acc
    
    print("\n模型比较结果:")
    for model_type, acc in results.items():
        print(f"{model_type}: {acc:.4f}")

compare_gnn_models()

4.2 图分类

class GraphClassifier(nn.Module):
    def __init__(self, input_dim, hidden_dim, num_classes, model_type='GCN'):
        super().__init__()
        self.model_type = model_type
        
        if model_type == 'GCN':
            self.gnn = GCN(input_dim, hidden_dim, hidden_dim)
        elif model_type == 'GAT':
            self.gnn = GAT(input_dim, hidden_dim, hidden_dim)
        elif model_type == 'GraphSAGE':
            self.gnn = GraphSAGE(input_dim, hidden_dim, hidden_dim)
        
        # 图级别分类器
        self.classifier = nn.Linear(hidden_dim, num_classes)
        self.pooling = nn.AdaptiveAvgPool1d(1)
    
    def forward(self, x, adj, batch_size):
        # 节点级别表示
        node_embeddings = self.gnn(x, adj)
        
        # 图级别表示（全局平均池化）
        graph_embedding = torch.mean(node_embeddings, dim=0, keepdim=True)
        graph_embedding = graph_embedding.repeat(batch_size, 1)
        
        # 分类
        output = self.classifier(graph_embedding)
        return output

# 图分类示例
def graph_classification_example():
    # 创建多个图的数据
    num_graphs = 50
    graphs_data = []
    
    for i in range(num_graphs):
        num_nodes = np.random.randint(10, 20)
        adj = torch.rand(num_nodes, num_nodes)
        adj = (adj + adj.T) / 2
        adj = (adj > 0.3).float()
        adj.fill_diagonal_(0)
        
        x = torch.randn(num_nodes, 5)
        label = torch.randint(0, 2, (1,))
        
        graphs_data.append((x, adj, label))
    
    # 训练图分类器
    model = GraphClassifier(input_dim=5, hidden_dim=16, num_classes=2, model_type='GCN')
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    criterion = nn.CrossEntropyLoss()
    
    for epoch in range(100):
        total_loss = 0
        for x, adj, label in graphs_data:
            normalized_adj = normalize_adjacency(adj)
            output = model(x, normalized_adj, 1)
            loss = criterion(output, label)
            
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        if epoch % 20 == 0:
            print(f'Epoch {epoch}, Loss: {total_loss/len(graphs_data):.4f}')

graph_classification_example()

4.3 链接预测

class LinkPredictor(nn.Module):
    def __init__(self, input_dim, hidden_dim, model_type='GCN'):
        super().__init__()
        self.model_type = model_type
        
        if model_type == 'GCN':
            self.gnn = GCN(input_dim, hidden_dim, hidden_dim)
        elif model_type == 'GAT':
            self.gnn = GAT(input_dim, hidden_dim, hidden_dim)
        elif model_type == 'GraphSAGE':
            self.gnn = GraphSAGE(input_dim, hidden_dim, hidden_dim)
        
        # 链接预测头
        self.predictor = nn.Linear(2 * hidden_dim, 1)
    
    def forward(self, x, adj, edge_index):
        # 获取节点嵌入
        node_embeddings = self.gnn(x, adj)
        
        # 获取边的端点嵌入
        src_emb = node_embeddings[edge_index[0]]
        dst_emb = node_embeddings[edge_index[1]]
        
        # 组合端点嵌入
        edge_emb = torch.cat([src_emb, dst_emb], dim=1)
        
        # 预测链接概率
        link_prob = torch.sigmoid(self.predictor(edge_emb))
        
        return link_prob.squeeze()

# 链接预测示例
def link_prediction_example():
    # 创建图数据
    num_nodes = 50
    num_features = 10
    
    # 生成邻接矩阵
    adj = torch.rand(num_nodes, num_nodes)
    adj = (adj + adj.T) / 2
    adj = (adj > 0.1).float()
    adj.fill_diagonal_(0)
    
    # 生成节点特征
    x = torch.randn(num_nodes, num_features)
    
    # 创建正负样本
    positive_edges = torch.nonzero(adj).T
    negative_edges = torch.nonzero(1 - adj).T
    
    # 平衡正负样本
    num_positive = positive_edges.size(1)
    negative_indices = torch.randperm(negative_edges.size(1))[:num_positive]
    negative_edges = negative_edges[:, negative_indices]
    
    # 合并正负样本
    all_edges = torch.cat([positive_edges, negative_edges], dim=1)
    labels = torch.cat([torch.ones(num_positive), torch.zeros(num_positive)])
    
    # 训练模型
    model = LinkPredictor(input_dim=num_features, hidden_dim=16, model_type='GCN')
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    criterion = nn.BCELoss()
    
    normalized_adj = normalize_adjacency(adj)
    
    for epoch in range(100):
        optimizer.zero_grad()
        
        predictions = model(x, normalized_adj, all_edges)
        loss = criterion(predictions, labels)
        
        loss.backward()
        optimizer.step()
        
        if epoch % 20 == 0:
            print(f'Epoch {epoch}, Loss: {loss.item():.4f}')

link_prediction_example()

---

5. 图神经网络的高级技术

5.1 图采样技术

class GraphSampler:
    def __init__(self, adj_matrix, num_samples=10):
        self.adj_matrix = adj_matrix
        self.num_samples = num_samples
    
    def random_walk_sampling(self, start_node, walk_length=5):
        """随机游走采样"""
        current_node = start_node
        walk = [current_node]
        
        for _ in range(walk_length - 1):
            neighbors = torch.nonzero(self.adj_matrix[current_node]).flatten()
            if len(neighbors) > 0:
                current_node = neighbors[torch.randint(0, len(neighbors), (1,))].item()
                walk.append(current_node)
            else:
                break
        
        return walk
    
    def neighborhood_sampling(self, center_node, num_hops=2):
        """邻域采样"""
        sampled_nodes = {center_node}
        current_level = {center_node}
        
        for _ in range(num_hops):
            next_level = set()
            for node in current_level:
                neighbors = torch.nonzero(self.adj_matrix[node]).flatten()
                for neighbor in neighbors:
                    if neighbor.item() not in sampled_nodes:
                        next_level.add(neighbor.item())
                        sampled_nodes.add(neighbor.item())
            
            current_level = next_level
            if not current_level:
                break
        
        return list(sampled_nodes)

# 使用图采样
def graph_sampling_example():
    adj = torch.tensor([
        [0, 1, 1, 0, 0],
        [1, 0, 1, 1, 0],
        [1, 1, 0, 1, 1],
        [0, 1, 1, 0, 1],
        [0, 0, 1, 1, 0]
    ], dtype=torch.float32)
    
    sampler = GraphSampler(adj)
    
    # 随机游走采样
    walk = sampler.random_walk_sampling(0, walk_length=4)
    print(f"随机游走采样: {walk}")
    
    # 邻域采样
    neighborhood = sampler.neighborhood_sampling(0, num_hops=2)
    print(f"邻域采样: {neighborhood}")

graph_sampling_example()

5.2 图数据增强

class GraphAugmentation:
    def __init__(self, adj_matrix, node_features):
        self.adj_matrix = adj_matrix
        self.node_features = node_features
    
    def edge_dropout(self, dropout_rate=0.1):
        """边丢弃"""
        mask = torch.rand_like(self.adj_matrix) > dropout_rate
        augmented_adj = self.adj_matrix * mask.float()
        return augmented_adj
    
    def node_dropout(self, dropout_rate=0.1):
        """节点特征丢弃"""
        mask = torch.rand_like(self.node_features) > dropout_rate
        augmented_features = self.node_features * mask.float()
        return augmented_features
    
    def subgraph_sampling(self, num_nodes=None):
        """子图采样"""
        if num_nodes is None:
            num_nodes = self.adj_matrix.size(0) // 2
        
        # 随机选择节点
        selected_nodes = torch.randperm(self.adj_matrix.size(0))[:num_nodes]
        
        # 提取子图
        sub_adj = self.adj_matrix[selected_nodes][:, selected_nodes]
        sub_features = self.node_features[selected_nodes]
        
        return sub_adj, sub_features, selected_nodes

# 图数据增强示例
def graph_augmentation_example():
    adj = torch.rand(10, 10)
    adj = (adj + adj.T) / 2
    adj = (adj > 0.3).float()
    adj.fill_diagonal_(0)
    
    features = torch.randn(10, 5)
    
    aug = GraphAugmentation(adj, features)
    
    # 边丢弃
    edge_dropped_adj = aug.edge_dropout(dropout_rate=0.2)
    print(f"原始邻接矩阵边数: {torch.sum(adj)}")
    print(f"边丢弃后边数: {torch.sum(edge_dropped_adj)}")
    
    # 节点特征丢弃
    feature_dropped = aug.node_dropout(dropout_rate=0.2)
    print(f"原始特征非零数: {torch.count_nonzero(features)}")
    print(f"特征丢弃后非零数: {torch.count_nonzero(feature_dropped)}")
    
    # 子图采样
    sub_adj, sub_features, selected_nodes = aug.subgraph_sampling(num_nodes=5)
    print(f"子图节点数: {sub_adj.size(0)}")
    print(f"选中的节点: {selected_nodes}")

graph_augmentation_example()

---

6. 图神经网络的优化技术

6.1 图正则化

class GraphRegularization:
    def __init__(self, adj_matrix):
        self.adj_matrix = adj_matrix
    
    def smoothness_regularization(self, node_embeddings):
        """平滑性正则化"""
        # 计算相邻节点的嵌入差异
        diff = node_embeddings.unsqueeze(1) - node_embeddings.unsqueeze(0)
        squared_diff = torch.sum(diff ** 2, dim=2)
        
        # 只考虑有边的节点对
        smoothness_loss = torch.sum(self.adj_matrix * squared_diff)
        return smoothness_loss
    
    def degree_regularization(self, node_embeddings):
        """度正则化"""
        degrees = torch.sum(self.adj_matrix, dim=1)
        degree_weights = degrees / torch.sum(degrees)
        
        # 高度数节点应该有更大的嵌入范数
        embedding_norms = torch.norm(node_embeddings, dim=1)
        degree_loss = torch.sum(degree_weights * embedding_norms)
        
        return degree_loss

# 使用图正则化
def graph_regularization_example():
    adj = torch.tensor([
        [0, 1, 1, 0],
        [1, 0, 1, 1],
        [1, 1, 0, 1],
        [0, 1, 1, 0]
    ], dtype=torch.float32)
    
    node_embeddings = torch.randn(4, 8)
    
    reg = GraphRegularization(adj)
    
    smoothness_loss = reg.smoothness_regularization(node_embeddings)
    degree_loss = reg.degree_regularization(node_embeddings)
    
    print(f"平滑性正则化损失: {smoothness_loss:.4f}")
    print(f"度正则化损失: {degree_loss:.4f}")

graph_regularization_example()

6.2 图对比学习

class GraphContrastiveLearning(nn.Module):
    def __init__(self, input_dim, hidden_dim, temperature=0.1):
        super().__init__()
        self.temperature = temperature
        self.encoder = GCN(input_dim, hidden_dim, hidden_dim)
        self.projection = nn.Linear(hidden_dim, hidden_dim)
    
    def forward(self, x, adj):
        # 编码
        embeddings = self.encoder(x, adj)
        
        # 投影
        projections = self.projection(embeddings)
        projections = F.normalize(projections, dim=1)
        
        return projections
    
    def contrastive_loss(self, x, adj, aug_adj):
        """对比学习损失"""
        # 原始图的表示
        z1 = self.forward(x, adj)
        
        # 增强图的表示
        z2 = self.forward(x, aug_adj)
        
        # 计算相似度矩阵
        sim_matrix = torch.mm(z1, z2.T) / self.temperature
        
        # 正样本对（对角线）
        positive_sim = torch.diag(sim_matrix)
        
        # 负样本对（非对角线）
        negative_sim = sim_matrix - torch.diag(positive_sim)
        
        # InfoNCE损失
        numerator = torch.exp(positive_sim)
        denominator = torch.sum(torch.exp(sim_matrix), dim=1)
        
        loss = -torch.mean(torch.log(numerator / denominator))
        
        return loss

# 图对比学习示例
def graph_contrastive_learning_example():
    # 创建数据
    num_nodes = 20
    num_features = 10
    
    adj = torch.rand(num_nodes, num_nodes)
    adj = (adj + adj.T) / 2
    adj = (adj > 0.2).float()
    adj.fill_diagonal_(0)
    
    x = torch.randn(num_nodes, num_features)
    
    # 创建增强图
    aug_adj = adj.clone()
    mask = torch.rand_like(adj) > 0.1
    aug_adj = aug_adj * mask.float()
    
    # 训练对比学习模型
    model = GraphContrastiveLearning(num_features, 16)
    optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    
    for epoch in range(50):
        optimizer.zero_grad()
        
        loss = model.contrastive_loss(x, adj, aug_adj)
        loss.backward()
        optimizer.step()
        
        if epoch % 10 == 0:
            print(f'Epoch {epoch}, Contrastive Loss: {loss.item():.4f}')

graph_contrastive_learning_example()

---

7. 相关论文与研究方向

7.1 经典论文

1. “The Graph Neural Network Model” (2009) - Scarselli et al.

   • GNN的经典论文
   • 建立了消息传递的理论基础

2. “Semi-Supervised Classification with Graph Convolutional Networks” (2016) - Kipf & Welling

   • GCN的经典论文
   • 提出了图卷积的简化形式

3. “Graph Attention Networks” (2018) - Veličković et al.

   • GAT的论文
   • 引入了注意力机制到图神经网络

7.2 现代发展

1. “How Powerful are Graph Neural Networks?” (2019) - Xu et al.

   • 分析了GNN的表达能力
   • 提出了GIN（Graph Isomorphism Network）

2. “Graph Neural Networks: A Review of Methods and Applications” (2020) - Wu et al.

   • GNN的全面综述
   • 总结了各种GNN变体

3. “Graph Transformer Networks” (2019) - Dwivedi & Bresson

   • 将Transformer应用到图数据
   • 开启了图Transformer的研究

参考文献

1. Scarselli, F., et al. (2009). The graph neural network model. IEEE transactions on neural networks, 20(1), 61-80.

2. Kipf, T. N., & Welling, M. (2016). Semi-supervised classification with graph convolutional networks. arXiv preprint arXiv:1609.02907.

3. Veličković, P., et al. (2018). Graph attention networks. International conference on learning representations.

4. Xu, K., et al. (2019). How powerful are graph neural networks?. International conference on learning representations.

5. Wu, Z., et al. (2020). Graph neural networks: A review of methods and applications. AI open, 1, 57-81.