🧠 第17课：图神经网络

import numpy as np class GraphConvLayer: """图卷积层 (GCN)""" def __init__(self, in_dim, out_dim): np.random.seed(42) self.W = np.random.randn(in_dim, out_dim) * np.sqrt(2.0 / (in_dim + out_dim)) self.bias = np.zeros(out_dim) def forward(self, features, adj_matrix): """ features: (N, in_dim) 节点特征矩阵 adj_matrix: (N, N) 邻接矩阵（归一化后） """ # D^{-1/2} A D^{-1/2} X W + b aggregated = adj_matrix @ features output = aggregated @ self.W + self.bias return np.tanh(output) class RelationalGCNLayer: """关系图卷积层 (R-GCN) - 专为知识图谱设计""" def __init__(self, in_dim, out_dim, num_relations): np.random.seed(42) self.W_base = np.random.randn(in_dim, out_dim) * 0.1 # 自身权重 self.W_rel = {r: np.random.randn(in_dim, out_dim) * 0.1 for r in range(num_relations)} self.bias = np.zeros(out_dim) def forward(self, features, triplets, num_nodes): """ features: (N, in_dim) triplets: [(head, relation, tail), ...] """ output = features @ self.W_base # 自身变换 # 按关系类型聚合 for h, r, t in triplets: if r in self.W_rel: # 尾实体聚合头实体的信息（通过关系r） output[t] += features[h] @ self.W_rel[r] # 头实体也聚合尾实体的信息（逆关系） output[h] += features[t] @ self.W_rel[r] return np.tanh(output + self.bias) class SimpleGNNModel: """完整的GNN模型""" def __init__(self, num_nodes, num_relations, hidden_dim=16, num_classes=3): self.num_nodes = num_nodes self.num_relations = num_relations np.random.seed(42) self.node_embeddings = np.random.randn(num_nodes, hidden_dim) * 0.1 self.rgcn1 = RelationalGCNLayer(hidden_dim, hidden_dim, num_relations) self.rgcn2 = RelationalGCNLayer(hidden_dim, num_classes, num_relations) def forward(self, triplets): h1 = self.rgcn1.forward(self.node_embeddings, triplets, self.num_nodes) h2 = self.rgcn2.forward(h1, triplets, self.num_nodes) return h2 def predict_link(self, triplets, head, relation, tail): """链接预测评分""" h = self.forward(triplets) # 简单得分：头尾实体嵌入的点积 score = np.dot(h[head], h[tail]) return score # ========== 测试 ========== # 节点映射: 0=鲁迅, 1=老舍, 2=呐喊, 3=绍兴, 4=北京, 5=浙江省, 6=中国 # 关系映射: 0=创作, 1=出生地, 2=属于 triplets = [ (0, 0, 2), # 鲁迅-创作-呐喊 (0, 1, 3), # 鲁迅-出生地-绍兴 (1, 1, 4), # 老舍-出生地-北京 (3, 2, 5), # 绍兴-属于-浙江省 (5, 2, 6), # 浙江省-属于-中国 (4, 2, 6), # 北京-属于-中国 ] model = SimpleGNNModel(num_nodes=7, num_relations=3, hidden_dim=8, num_classes=4) output = model.forward(triplets) print("=== GNN输出 ===") print(f"输出形状: {output.shape}") names = ["鲁迅", "老舍", "呐喊", "绍兴", "北京", "浙江省", "中国"] for i, name in enumerate(names): print(f" {name}: {np.round(output[i], 3)}") # 链接预测 print(" === 链接预测 ===") print(f"鲁迅-创作-呐喊: {model.predict_link(triplets, 0, 0, 2):.4f}") print(f">>鲁迅-创作-北京: {model.predict_link(triplets, 0, 0, 4):.4f}") # 相似度 print(" === GNN嵌入相似度 ===") for i in range(7): for j in range(i+1, 7): sim = np.dot(output[i], output[j]) / (np.linalg.norm(output[i]) * np.linalg.norm(output[j]) + 1e-8) if abs(sim) > 0.3: print(f" sim({names[i]}, {names[j]}) = {sim:.4f}")

# GAT实现（图注意力网络） class GATLayer: """图注意力层""" def __init__(self, in_dim, out_dim, num_heads=4): np.random.seed(42) self.W = np.random.randn(in_dim, out_dim) * 0.1 self.a = np.random.randn(out_dim * 2) * 0.1 self.leaky_relu_slope = 0.2 def forward(self, features, adj_list): h = features @ self.W N = len(features) attention = np.zeros((N, N)) for i in range(N): for j in adj_list.get(i, []): e_ij = np.dot(self.a, np.concatenate([h[i], h[j]])) attention[i][j] = np.where(e_ij > 0, e_ij, self.leaky_relu_slope * e_ij) row = attention[i] exp_row = np.exp(row - np.max(row)) attention[i] = exp_row / (exp_row.sum() + 1e-8) return attention @ h print('GAT层实现完成')