代码实战项目10：图算法实现¶
项目概述¶
图算法是计算机科学的重要组成部分，在社交网络、路由协议、资源调度等领域有广泛应用。本项目将使用 Python 实现常见的图算法，帮助深入理解图论和算法设计。
学习目标¶
- 理解图的基本概念和表示方法
- 掌握图的遍历算法（DFS、BFS）
- 学会最短路径算法（Dijkstra、Floyd-Warshall）
- 理解最小生成树算法（Kruskal、Prim）
- 掌握拓扑排序和关键路径算法
图的基本概念¶
图的定义¶
图 G = (V, E) 由顶点集 V 和边集 E 组成。
图的类型¶
- 有向图：边有方向
- 无向图：边没有方向
- 带权图：边有权重
- 无权图：边没有权重
图的表示¶
- 邻接矩阵：二维数组表示
- 邻接表：链表或字典表示
代码实现¶
1. 图类定义（graph.py）¶
from collections import deque
import heapq
class Graph:
"""图类"""
def __init__(self, directed=False):
"""
初始化图
参数:
directed: 是否为有向图
"""
self.directed = directed
self.vertices = {} # 顶点集合
self.adj_list = {} # 邻接表
def add_vertex(self, vertex):
"""
添加顶点
参数:
vertex: 顶点标识
"""
if vertex not in self.vertices:
self.vertices[vertex] = True
self.adj_list[vertex] = []
def add_edge(self, from_vertex, to_vertex, weight=1):
"""
添加边
参数:
from_vertex: 起始顶点
to_vertex: 目标顶点
weight: 边的权重
"""
if from_vertex not in self.vertices:
self.add_vertex(from_vertex)
if to_vertex not in self.vertices:
self.add_vertex(to_vertex)
self.adj_list[from_vertex].append((to_vertex, weight))
if not self.directed:
self.adj_list[to_vertex].append((from_vertex, weight))
def get_vertices(self):
"""获取所有顶点"""
return list(self.vertices.keys())
def get_edges(self):
"""获取所有边"""
edges = []
for vertex in self.adj_list:
for neighbor, weight in self.adj_list[vertex]:
if self.directed or (neighbor, vertex, weight) not in edges:
edges.append((vertex, neighbor, weight))
return edges
def __str__(self):
"""字符串表示"""
result = ""
for vertex in self.adj_list:
neighbors = [f"{n}({w})" for n, w in self.adj_list[vertex]]
result += f"{vertex} -> {', '.join(neighbors)}\n"
return result
class DirectedGraph(Graph):
"""有向图类"""
def __init__(self):
super().__init__(directed=True)
class WeightedGraph(Graph):
"""带权图类"""
def __init__(self, directed=False):
super().__init__(directed=directed)
2. 深度优先搜索（DFS）¶
def dfs(graph, start, visited=None):
"""
深度优先搜索
参数:
graph: 图对象
start: 起始顶点
visited: 已访问集合
返回:
list: 遍历顺序
"""
if visited is None:
visited = set()
visited.add(start)
result = [start]
for neighbor, _ in graph.adj_list.get(start, []):
if neighbor not in visited:
result.extend(dfs(graph, neighbor, visited))
return result
def dfs_iterative(graph, start):
"""
深度优先搜索（迭代版本）
参数:
graph: 图对象
start: 起始顶点
返回:
list: 遍历顺序
"""
visited = set()
stack = [start]
result = []
while stack:
vertex = stack.pop()
if vertex not in visited:
visited.add(vertex)
result.append(vertex)
# 将邻居按相反顺序压入栈，保证正确的遍历顺序
neighbors = [n for n, _ in graph.adj_list.get(vertex, [])]
stack.extend(reversed(neighbors))
return result
def dfs_paths(graph, start, goal, path=None):
"""
查找从 start 到 goal 的所有路径（DFS）
参数:
graph: 图对象
start: 起始顶点
goal: 目标顶点
path: 当前路径
返回:
list: 所有路径
"""
if path is None:
path = []
path = path + [start]
if start == goal:
return [path]
if start not in graph.adj_list:
return []
paths = []
for vertex, _ in graph.adj_list[start]:
if vertex not in path: # 避免循环
new_paths = dfs_paths(graph, vertex, goal, path)
paths.extend(new_paths)
return paths
def shortest_path_dfs(graph, start, goal):
"""
查找最短路径（DFS版本）
参数:
graph: 图对象
start: 起始顶点
goal: 目标顶点
返回:
list: 最短路径，不存在返回 None
"""
paths = dfs_paths(graph, start, goal)
if not paths:
return None
return min(paths, key=len)
3. 广度优先搜索（BFS）¶
def bfs(graph, start):
"""
广度优先搜索
参数:
graph: 图对象
start: 起始顶点
返回:
list: 遍历顺序
"""
visited = set([start])
queue = deque([start])
result = []
while queue:
vertex = queue.popleft()
result.append(vertex)
for neighbor, _ in graph.adj_list.get(vertex, []):
if neighbor not in visited:
visited.add(neighbor)
queue.append(neighbor)
return result
def bfs_shortest_path(graph, start, goal):
"""
查找最短路径（BFS）
参数:
graph: 图对象
start: 起始顶点
goal: 目标顶点
返回:
list: 最短路径，不存在返回 None
"""
if start == goal:
return [start]
visited = {start}
queue = deque([(start, [start])])
while queue:
vertex, path = queue.popleft()
for neighbor, _ in graph.adj_list.get(vertex, []):
if neighbor == goal:
return path + [neighbor]
if neighbor not in visited:
visited.add(neighbor)
queue.append((neighbor, path + [neighbor]))
return None
def bfs_distance(graph, start, goal):
"""
计算两个顶点之间的最短距离（BFS）
参数:
graph: 图对象
start: 起始顶点
goal: 目标顶点
返回:
int: 最短距离，不可达返回 -1
"""
if start == goal:
return 0
visited = {start}
queue = deque([(start, 0)])
while queue:
vertex, distance = queue.popleft()
for neighbor, _ in graph.adj_list.get(vertex, []):
if neighbor == goal:
return distance + 1
if neighbor not in visited:
visited.add(neighbor)
queue.append((neighbor, distance + 1))
return -1
4. Dijkstra 最短路径算法¶
def dijkstra(graph, start):
"""
Dijkstra 最短路径算法
参数:
graph: 带权图对象
start: 起始顶点
返回:
dict: {顶点: (最短距离, 前驱顶点)}
"""
distances = {vertex: float('infinity') for vertex in graph.vertices}
previous = {vertex: None for vertex in graph.vertices}
distances[start] = 0
# 优先队列：(距离, 顶点)
pq = [(0, start)]
while pq:
current_distance, current_vertex = heapq.heappop(pq)
# 如果找到更长的路径，跳过
if current_distance > distances[current_vertex]:
continue
# 遍历邻居
for neighbor, weight in graph.adj_list.get(current_vertex, []):
distance = current_distance + weight
# 如果找到更短的路径
if distance < distances[neighbor]:
distances[neighbor] = distance
previous[neighbor] = current_vertex
heapq.heappush(pq, (distance, neighbor))
return {v: (distances[v], previous[v]) for v in graph.vertices}
def dijkstra_path(graph, start, goal):
"""
使用 Dijkstra 算法查找最短路径
参数:
graph: 带权图对象
start: 起始顶点
goal: 目标顶点
返回:
tuple: (路径, 总距离)
"""
if start not in graph.vertices or goal not in graph.vertices:
return None, float('infinity')
if start == goal:
return [start], 0
# 运行 Dijkstra
result = dijkstra(graph, start)
distances = {v: result[v][0] for v in result}
previous = {v: result[v][1] for v in result}
# 重建路径
if distances[goal] == float('infinity'):
return None, float('infinity')
path = []
current = goal
while current is not None:
path.append(current)
current = previous[current]
path.reverse()
return path, distances[goal]
5. Floyd-Warshall 算法¶
def floyd_warshall(graph):
"""
Floyd-Warshall 全源最短路径算法
参数:
graph: 带权图对象
返回:
tuple: (距离矩阵, 路径矩阵)
"""
vertices = graph.get_vertices()
n = len(vertices)
vertex_index = {v: i for i, v in enumerate(vertices)}
# 初始化距离矩阵和路径矩阵
dist = [[float('infinity')] * n for _ in range(n)]
next_node = [[None] * n for _ in range(n)]
for i in range(n):
dist[i][i] = 0
for from_v, to_v, weight in graph.get_edges():
i, j = vertex_index[from_v], vertex_index[to_v]
dist[i][j] = weight
next_node[i][j] = to_v
# Floyd-Warshall 算法
for k in range(n):
for i in range(n):
for j in range(n):
if dist[i][k] + dist[k][j] < dist[i][j]:
dist[i][j] = dist[i][k] + dist[k][j]
next_node[i][j] = next_node[i][k]
return dist, next_node
def floyd_warshall_path(graph, start, goal):
"""
使用 Floyd-Warshall 算法查找最短路径
参数:
graph: 带权图对象
start: 起始顶点
goal: 目标顶点
返回:
tuple: (路径, 总距离)
"""
vertices = graph.get_vertices()
if start not in vertices or goal not in vertices:
return None, float('infinity')
dist, next_node = floyd_warshall(graph)
vertex_index = {v: i for i, v in enumerate(vertices)}
i, j = vertex_index[start], vertex_index[goal]
if dist[i][j] == float('infinity'):
return None, float('infinity')
# 重建路径
path = []
current = start
while current != goal:
path.append(current)
current_i = vertex_index[current]
current = next_node[current_i][j]
if current is None:
return None, float('infinity')
path.append(goal)
return path, dist[i][j]
6. 最小生成树算法¶
Kruskal 算法¶
class UnionFind:
"""并查集数据结构"""
def __init__(self, elements):
self.parent = {e: e for e in elements}
self.rank = {e: 0 for e in elements}
def find(self, element):
"""查找根节点"""
if self.parent[element] != element:
self.parent[element] = self.find(self.parent[element])
return self.parent[element]
def union(self, element1, element2):
"""合并两个集合"""
root1 = self.find(element1)
root2 = self.find(element2)
if root1 != root2:
if self.rank[root1] < self.rank[root2]:
self.parent[root1] = root2
elif self.rank[root1] > self.rank[root2]:
self.parent[root2] = root1
else:
self.parent[root2] = root1
self.rank[root1] += 1
return True
return False
def kruskal(graph):
"""
Kruskal 最小生成树算法
参数:
graph: 带权图对象
返回:
tuple: (边列表, 总权重)
"""
# 获取所有边并按权重排序
edges = graph.get_edges()
edges.sort(key=lambda x: x[2])
# 初始化并查集
uf = UnionFind(graph.get_vertices())
mst_edges = []
total_weight = 0
for edge in edges:
from_v, to_v, weight = edge
# 如果不在同一集合，添加边
if uf.union(from_v, to_v):
mst_edges.append(edge)
total_weight += weight
# 如果已经找到 n-1 条边，结束
if len(mst_edges) == len(graph.vertices) - 1:
break
return mst_edges, total_weight
Prim 算法¶
def prim(graph, start=None):
"""
Prim 最小生成树算法
参数:
graph: 带权图对象
start: 起始顶点（可选）
返回:
tuple: (边列表, 总权重)
"""
vertices = graph.get_vertices()
if not vertices:
return [], 0
# 选择起始顶点
if start is None:
start = vertices[0]
# 初始化
visited = set([start])
mst_edges = []
total_weight = 0
# 优先队列：(权重, from_vertex, to_vertex)
edges = []
# 添加起始顶点的所有边
for neighbor, weight in graph.adj_list.get(start, []):
heapq.heappush(edges, (weight, start, neighbor))
while edges and len(visited) < len(vertices):
weight, from_v, to_v = heapq.heappop(edges)
if to_v in visited:
continue
# 添加边
visited.add(to_v)
mst_edges.append((from_v, to_v, weight))
total_weight += weight
# 添加新顶点的所有边
for neighbor, w in graph.adj_list.get(to_v, []):
if neighbor not in visited:
heapq.heappush(edges, (w, to_v, neighbor))
return mst_edges, total_weight
7. 拓扑排序¶
def topological_sort(graph):
"""
拓扑排序（Kahn 算法）
参数:
graph: 有向图对象
返回:
list: 拓扑排序结果，有环返回 None
"""
if not graph.directed:
raise ValueError("拓扑排序只适用于有向图")
# 计算入度
in_degree = {v: 0 for v in graph.vertices}
for vertex in graph.adj_list:
for neighbor, _ in graph.adj_list[vertex]:
in_degree[neighbor] += 1
# 找到所有入度为0的顶点
queue = deque([v for v in graph.vertices if in_degree[v] == 0])
result = []
while queue:
vertex = queue.popleft()
result.append(vertex)
# 减少邻居的入度
for neighbor, _ in graph.adj_list.get(vertex, []):
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
# 检查是否有环
if len(result) != len(graph.vertices):
return None # 有环
return result
def topological_sort_dfs(graph):
"""
拓扑排序（DFS 算法）
参数:
graph: 有向图对象
返回:
list: 拓扑排序结果，有环返回 None
"""
if not graph.directed:
raise ValueError("拓扑排序只适用于有向图")
visited = set()
temp_visited = set()
result = []
def dfs(vertex):
if vertex in temp_visited:
return False # 有环
if vertex in visited:
return True
temp_visited.add(vertex)
for neighbor, _ in graph.adj_list.get(vertex, []):
if not dfs(neighbor):
return False
temp_visited.remove(vertex)
visited.add(vertex)
result.append(vertex)
return True
for vertex in graph.vertices:
if vertex not in visited:
if not dfs(vertex):
return None # 有环
result.reverse()
return result
8. 测试和示例¶
def test_graph_algorithms():
"""测试图算法"""
print("="*60)
print("图算法测试")
print("="*60)
# 创建无向图
print("\n1. 创建无向图")
graph = Graph()
edges = [
('A', 'B'), ('A', 'C'), ('B', 'D'),
('C', 'D'), ('C', 'E'), ('D', 'E')
]
for u, v in edges:
graph.add_edge(u, v)
print(graph)
# DFS
print("\n2. 深度优先搜索 (DFS)")
dfs_result = dfs(graph, 'A')
print(f"DFS 遍历顺序: {dfs_result}")
# BFS
print("\n3. 广度优先搜索 (BFS)")
bfs_result = bfs(graph, 'A')
print(f"BFS 遍历顺序: {bfs_result}")
# 最短路径（无权）
print("\n4. 最短路径（无权图）")
path = bfs_shortest_path(graph, 'A', 'E')
print(f"A 到 E 的最短路径: {path}")
# 创建带权图
print("\n5. 创建带权图")
weighted_graph = WeightedGraph()
weighted_edges = [
('A', 'B', 4), ('A', 'C', 2), ('B', 'C', 1),
('B', 'D', 5), ('C', 'D', 8), ('C', 'E', 10),
('D', 'E', 2)
]
for u, v, w in weighted_edges:
weighted_graph.add_edge(u, v, w)
print(weighted_graph)
# Dijkstra
print("\n6. Dijkstra 最短路径")
dijkstra_result = dijkstra(weighted_graph, 'A')
for v in dijkstra_result:
dist, prev = dijkstra_result[v]
print(f"A -> {v}: 距离 = {dist}, 前驱 = {prev}")
path, distance = dijkstra_path(weighted_graph, 'A', 'E')
print(f"A 到 E 的最短路径: {path}, 距离 = {distance}")
# 最小生成树
print("\n7. 最小生成树")
kruskal_edges, kruskal_weight = kruskal(weighted_graph)
print("Kruskal 算法:")
for edge in kruskal_edges:
print(f" {edge[0]} - {edge[1]} (权重: {edge[2]})")
print(f"总权重: {kruskal_weight}")
prim_edges, prim_weight = prim(weighted_graph, 'A')
print("\nPrim 算法:")
for edge in prim_edges:
print(f" {edge[0]} - {edge[1]} (权重: {edge[2]})")
print(f"总权重: {prim_weight}")
# 拓扑排序
print("\n8. 拓扑排序")
dag = DirectedGraph()
dag_edges = [
('A', 'B'), ('A', 'C'), ('B', 'D'),
('C', 'D'), ('D', 'E')
]
for u, v in dag_edges:
dag.add_edge(u, v)
topo_result = topological_sort(dag)
print(f"拓扑排序结果: {topo_result}")
if __name__ == '__main__':
test_graph_algorithms()
算法复杂度分析¶
| 算法 | 时间复杂度 | 空间复杂度 | 适用场景 |
|---|---|---|---|
| DFS | O(V + E) | O(V) | 连通性检测、路径查找 |
| BFS | O(V + E) | O(V) | 最短路径（无权图）、层次遍历 |
| Dijkstra | O((V + E) log V) | O(V) | 单源最短路径（非负权） |
| Floyd-Warshall | O(V³) | O(V²) | 全源最短路径 |
| Kruskal | O(E log E) | O(V + E) | 稀疏图的最小生成树 |
| Prim | O((V + E) log V) | O(V + E) | 密集图的最小生成树 |
| 拓扑排序 | O(V + E) | O(V) | 任务调度、依赖关系 |
应用场景¶
- 社交网络
- 好友推荐（最短路径）
-
社区发现（连通分量）
-
路由协议
- 最短路由（Dijkstra）
-
网络拓扑（最小生成树）
-
任务调度
- 依赖关系（拓扑排序）
-
关键路径（最长路径）
-
地图导航
- 最短路径（Dijkstra、A*）
- 地图匹配（图匹配）
扩展算法¶
- A* 算法：启发式搜索
- Bellman-Ford：负权边的最短路径
- 强连通分量：Kosaraju、Tarjan 算法
- 二分图匹配：匈牙利算法
- 网络流：Ford-Fulkerson、Edmonds-Karp
项目总结¶
本项目实现了常见的图算法，涵盖了：
- 图的基本操作：创建、添加顶点和边
- 遍历算法：DFS、BFS
- 最短路径：Dijkstra、Floyd-Warshall
- 最小生成树：Kruskal、Prim
- 拓扑排序：Kahn、DFS
通过本项目，你将掌握： - 图的数据结构表示 - 经典图算法的实现 - 算法复杂度分析 - 实际问题的建模和求解
下一步¶
尝试实现更多高级算法： - A* 启发式搜索 - Bellman-Ford 算法 - 强连通分量算法 - 最大流最小割 - 二分图匹配 - 欧拉路径和哈密顿路径
祝你学习愉快！