🧭 第14课：图遍历算法

from collections import defaultdict, deque import heapq class GraphAlgorithms: """图遍历与图算法集合""" def __init__(self): self.adj = defaultdict(dict) # {node: {neighbor: weight}} def add_edge(self, u, v, weight=1, directed=True): self.adj[u][v] = weight if not directed: self.adj[v][u] = weight def bfs(self, start): """广度优先搜索""" visited = {start} queue = deque([start]) order = [] while queue: node = queue.popleft() order.append(node) for neighbor in sorted(self.adj[node]): if neighbor not in visited: visited.add(neighbor) queue.append(neighbor) return order def dfs(self, start): """深度优先搜索""" visited = set() order = [] def _dfs(node): visited.add(node) order.append(node) for neighbor in sorted(self.adj[node]): if neighbor not in visited: _dfs(neighbor) _dfs(start) return order def dijkstra(self, start, end): """Dijkstra最短路径""" dist = {start: 0} prev = {} pq = [(0, start)] visited = set() while pq: d, node = heapq.heappop(pq) if node in visited: continue visited.add(node) if node == end: path = [] current = end while current in prev: path.append(current) current = prev[current] path.append(start) return path[::-1], dist[end] for neighbor, w in self.adj[node].items(): new_dist = d + w if neighbor not in dist or new_dist < dist[neighbor]: dist[neighbor] = new_dist prev[neighbor] = node heapq.heappush(pq, (new_dist, neighbor)) return None, float('inf') def pagerank(self, iterations=100, damping=0.85): """PageRank算法""" nodes = set(self.adj.keys()) for neighbors in self.adj.values(): nodes.update(neighbors.keys()) N = len(nodes) pr = {n: 1.0 / N for n in nodes} for _ in range(iterations): new_pr = {} for node in nodes: rank = (1 - damping) / N for src in nodes: if node in self.adj[src]: out_degree = len(self.adj[src]) rank += damping * pr[src] / out_degree new_pr[node] = rank pr = new_pr return sorted(pr.items(), key=lambda x: -x[1]) def connected_components(self): """弱连通分量""" nodes = set(self.adj.keys()) for neighbors in self.adj.values(): nodes.update(neighbors.keys()) visited = set() components = [] for node in nodes: if node not in visited: comp = [] queue = deque([node]) visited.add(node) while queue: n = queue.popleft() comp.append(n) for nb in self.adj[n]: if nb not in visited: visited.add(nb) queue.append(nb) for src in self.adj: if n in self.adj[src] and src not in visited: visited.add(src) queue.append(src) components.append(comp) return components # ========== 构建知识图谱图结构 ========== ga = GraphAlgorithms() ga.add_edge("鲁迅", "呐喊") ga.add_edge("鲁迅", "彷徨") ga.add_edge("鲁迅", "绍兴") ga.add_edge("老舍", "骆驼祥子") ga.add_edge("老舍", "北京") ga.add_edge("绍兴", "浙江省") ga.add_edge("浙江省", "中国") ga.add_edge("北京", "中国") ga.add_edge("徐志摩", "再别康桥") ga.add_edge("徐志摩", "海宁") print("=== BFS遍历（从鲁迅出发） ===") print(" → ".join(ga.bfs("鲁迅"))) print(" === DFS遍历（从鲁迅出发） ===") print(" → ".join(ga.dfs("鲁迅"))) print(" === PageRank ===") for node, score in ga.pagerank(iterations=50)[:5]: print(f" {node}: {score:.4f}") print(" === 连通分量 ===") for i, comp in enumerate(ga.connected_components()): print(f" 分量{i+1}: {comp}")

import numpy as np class SimpleGNN: """简易图神经网络：消息传递机制""" def __init__(self, node_features, adj_list, hidden_dim=4): self.features = node_features # {node: vector} self.adj = adj_list # {node: [neighbors]} self.hidden_dim = hidden_dim # 随机初始化权重 np.random.seed(42) self.W_self = np.random.randn(len(next(iter(node_features.values()))), hidden_dim) * 0.1 self.W_neigh = np.random.randn(len(next(iter(node_features.values()))), hidden_dim) * 0.1 def message_passing(self, num_hops=2): """多跳消息传递""" current_features = dict(self.features) for hop in range(num_hops): new_features = {} for node in self.adj: # 聚合邻居特征（均值聚合） neighbors = self.adj[node] if neighbors: neigh_agg = np.mean([current_features.get(n, np.zeros(len(current_features[node]))) for n in neighbors], axis=0) else: neigh_agg = np.zeros(len(current_features[node])) # 组合自身和邻居信息 self_feat = current_features[node] @ self.W_self neigh_feat = neigh_agg @ self.W_neigh new_features[node] = np.tanh(self_feat + neigh_feat) current_features = new_features return current_features # 构建示例图 features = { "鲁迅": np.array([1.0, 0.0, 0.0]), "老舍": np.array([0.8, 0.2, 0.0]), "呐喊": np.array([0.0, 1.0, 0.0]), "绍兴": np.array([0.0, 0.0, 1.0]), "北京": np.array([0.0, 0.0, 0.9]), } adj = { "鲁迅": ["呐喊", "绍兴"], "老舍": ["北京"], "呐喊": ["鲁迅"], "绍兴": [], "北京": [], } gnn = SimpleGNN(features, adj, hidden_dim=4) updated = gnn.message_passing(num_hops=2) print("=== GNN消息传递后的节点特征 ===") for node, feat in updated.items(): print(f" {node}: {np.round(feat, 3)}") # 计算相似度 print(" === GNN特征余弦相似度 ===") for n1 in ["鲁迅", "老舍"]: for n2 in ["呐喊", "绍兴", "北京"]: v1, v2 = updated[n1], updated[n2] cos = np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2) + 1e-8) print(f" sim({n1}, {n2}) = {cos:.4f}")