别再死记硬背了！用Python手把手教你实现DAG拓扑排序（附LeetCode真题）

用Python实战DAG拓扑排序：从原理到LeetCode高频考题解析

拓扑排序是算法面试中绕不开的核心考点，尤其在处理任务调度、课程安排等依赖关系问题时。第一次在LeetCode上遇到「课程表II」这道题时，我盯着那个不断报错的BFS实现百思不得其解——直到发现漏判了环路情况。本文将用Python带你从零实现DAG拓扑排序，并解决三个关键问题：如何检测有向图环路？DFS和BFS哪种实现更适合面试场景？如何处理多解情况？

1. 理解DAG的本质特性

有向无环图（DAG）就像一本没有循环引用的技术书籍目录。想象你在学习机器学习课程，必须按"线性代数→概率论→统计学习→深度学习"的顺序学习，这就是典型的DAG结构。其核心特征表现在：

• 方向性约束：边具有单向性，如A→B表示B依赖A
• 无环路保证：不存在A→B→C→A这样的循环依赖链
• 偏序关系：节点间存在先后关系但不必全序

# 典型DAG的邻接表表示 dag = { '线性代数': ['概率论'], '概率论': ['统计学习'], '统计学习': ['深度学习'], 'Python基础': ['统计学习'] }

在工业级应用中，DAG常用于：

• 编译器中的指令调度（LLVM）
• 数据管道依赖管理（Apache Airflow）
• 软件包版本依赖解析（pip/conda）

2. 拓扑排序的双实现策略

2.1 Kahn算法（BFS版）

这是最符合直觉的拓扑排序方法，其核心是不断移除入度为0的节点。我们通过一个实际例子来理解：

假设有课程依赖关系：

• 算法导论依赖C++基础
• 机器学习依赖算法导论和概率论
• 深度学习依赖机器学习和线性代数

from collections import deque def topological_sort_bfs(graph): in_degree = {u: 0 for u in graph} for u in graph: for v in graph[u]: in_degree[v] += 1 queue = deque([u for u in in_degree if in_degree[u] == 0]) topo_order = [] while queue: u = queue.popleft() topo_order.append(u) for v in graph.get(u, []): in_degree[v] -= 1 if in_degree[v] == 0: queue.append(v) if len(topo_order) != len(in_degree): raise ValueError("图中存在环路！") return topo_order

时间复杂度分析：

• 初始化入度表：O(V+E)
• 队列操作：O(V)
• 总复杂度：O(V+E)

2.2 基于DFS的递归实现

DFS版本更适合需要所有可能排序结果的场景，其核心在于后序遍历和状态标记：

def topological_sort_dfs(graph): visited = set() recursion_stack = set() result = [] def dfs(node): if node in recursion_stack: raise ValueError("检测到环路！") if node in visited: return visited.add(node) recursion_stack.add(node) for neighbor in graph.get(node, []): dfs(neighbor) recursion_stack.remove(node) result.append(node) for node in graph: if node not in visited: dfs(node) return result[::-1]

两种实现的对比：

提示：在技术面试中，建议优先实现Kahn算法，因为其非递归特性更不容易出错，且能直接给出环路检测结果。

3. LeetCode实战精讲

3.1 课程表问题（LeetCode 207/210）

这是拓扑排序最经典的考题，我们以210题为例进行深度解析：

def findOrder(numCourses, prerequisites): adj = [[] for _ in range(numCourses)] in_degree = [0] * numCourses for dest, src in prerequisites: adj[src].append(dest) in_degree[dest] += 1 queue = deque([i for i in range(numCourses) if in_degree[i] == 0]) order = [] while queue: course = queue.popleft() order.append(course) for neighbor in adj[course]: in_degree[neighbor] -= 1 if in_degree[neighbor] == 0: queue.append(neighbor) return order if len(order) == numCourses else []

常见陷阱：

1. 未初始化邻接表导致IndexError
2. 忘记判断最终结果长度（环路检测）
3. 混淆边的前后关系（prerequisites数组的解读）

3.2 外星文字典（LeetCode 269）

这道题将拓扑排序应用到字典序问题，需要从字符顺序关系构建DAG：

def alienOrder(words): adj = {c: set() for word in words for c in word} for i in range(len(words) - 1): w1, w2 = words[i], words[i+1] min_len = min(len(w1), len(w2)) if w1[:min_len] == w2[:min_len] and len(w1) > len(w2): return "" for j in range(min_len): if w1[j] != w2[j]: adj[w1[j]].add(w2[j]) break in_degree = {c: 0 for c in adj} for u in adj: for v in adj[u]: in_degree[v] += 1 queue = deque([c for c in in_degree if in_degree[c] == 0]) order = [] while queue: u = queue.popleft() order.append(u) for v in adj[u]: in_degree[v] -= 1 if in_degree[v] == 0: queue.append(v) return "".join(order) if len(order) == len(adj) else ""

特殊处理：

• 前缀相同但前词更长的情况直接返回空
• 需要先构建完整的字符集合
• 字典序的边只由第一对不同字符决定

4. 工程实践中的优化技巧

4.1 并行任务调度

当DAG用于任务调度时，可以扩展基础算法支持并行执行。关键修改点：

def parallel_topological_sort(graph, k): in_degree = {u: 0 for u in graph} for u in graph: for v in graph[u]: in_degree[v] += 1 available = deque([u for u in in_degree if in_degree[u] == 0]) result = [] time = 0 while available: batch = [] for _ in range(min(k, len(available))): batch.append(available.popleft()) result.append((time, batch)) time += 1 for u in batch: for v in graph.get(u, []): in_degree[v] -= 1 if in_degree[v] == 0: available.append(v) if len([u for batch in result for u in batch]) != len(in_degree): raise ValueError("存在环路依赖") return result

4.2 动态DAG处理

对于频繁变化的DAG（如CI/CD流水线），需要优化增量更新：

class DynamicDAG: def __init__(self): self.graph = defaultdict(set) self.in_degree = defaultdict(int) def add_edge(self, u, v): if v not in self.graph[u]: self.graph[u].add(v) self.in_degree[v] += 1 def remove_edge(self, u, v): if v in self.graph[u]: self.graph[u].remove(v) self.in_degree[v] -= 1 def get_ready_nodes(self): return [u for u in self.graph if self.in_degree[u] == 0]

4.3 性能对比测试

用随机生成的DAG测试不同实现的性能表现：

注意：当处理超大规模DAG（>1M节点）时，建议使用磁盘支持的图数据库或分布式处理框架。