3天搞定分类图手写实现，拒绝文档焦虑

官方文档太长，翻两页就忘，根本抓不住重点。与其对着枯燥的 API 列表发呆，不如直接上手，用 20 行代码跑通一个极简分类图原型。这里不堆砌术语，我们直接切入核心，通过手写实现的方式，把“分类图”这个听起来很高大上的数据结构，拆解成你能看懂的 Python 代码。

项目目标：我们要造一个什么样的轮子

很多开发者听到“分类图”或者“有向无环图（DAG）”就头大，觉得这是编译器或者复杂调度系统才需要关心的东西。其实不然。在数据清洗、任务依赖管理、甚至前端组件树渲染中，分类图的思想无处不在。

我们的目标很明确：从零搭建一个轻量级的分类图工具。它不需要支持百万级节点，不需要复杂的并发锁机制，但必须具备以下三个核心能力：

节点增删：能动态添加任务节点和依赖关系。
拓扑排序：能输出合法的执行顺序，这是分类图最核心的价值。
环检测：如果依赖关系形成了死循环，必须能准确报错，而不是让程序卡死。

为什么强调手写实现？因为库（如 networkx）虽然强大，但当你需要嵌入到特定业务逻辑中，或者面试被问到“请简述拓扑排序的底层原理”时，依赖库的黑盒会让你哑口无言。自己写一遍，才能把内存占用、时间复杂度这些指标刻在脑子里。

目录结构：极简主义，拒绝过度设计

作为一个实战项目，我们要保持工程化的整洁，但不搞形式主义。整个项目只需要三个文件，放在同一个文件夹下即可运行。

category_graph/
├── graph_core.py   # 核心算法实现：节点、边、拓扑排序
├── demo.py         # 演示脚本：构建具体业务场景
└── tests.py        # 单元测试：验证边界情况

这种结构的好处是，你可以随时复制 graph_core.py 到任何项目中复用，而不需要安装任何第三方依赖。这就是纯 Python 标准库的威力。

核心代码实现：逐行拆解拓扑排序

分类图的核心在于拓扑排序。通俗点说，就是“先完成前置任务，再执行后续任务”。最常用的算法是 Kahn 算法，它基于 BFS（广度优先搜索），利用入度（In-degree）来判断节点是否可以被处理。

下面是 graph_core.py 的完整代码。我会把关键逻辑拆解开，告诉你每一行代码背后的意图。

import collections
from typing import List, Dict, Set, Optionalclass CategoryGraph:"""一个基于有向无环图（DAG）的分类图实现用于管理任务依赖和执行顺序"""def __init__(self):# 邻接表：存储每个节点指向哪些后继节点# 例如：A -> [B, C] 表示 A 完成后，B 和 C 可以开始self.graph: Dict[str, List[str]] = {}# 入度表：记录每个节点有多少个前驱节点# 入度为 0 的节点，就是可以立即执行的节点self.in_degree: Dict[str, int] = {}def add_node(self, node: str):"""添加单个节点，初始化入度为0"""if node not in self.graph:self.graph[node] = []self.in_degree[node] = 0def add_edge(self, source: str, target: str):"""添加依赖关系：source -> target意味着 target 依赖于 source，source 必须先执行"""# 确保源节点和目标节点都已初始化self.add_node(source)self.add_node(target)# 检查是否已存在这条边，防止重复添加导致入度错误if target not in self.graph[source]:self.graph[source].append(target)self.in_degree[target] += 1# 如果形成了环，这里暂时不检测，留到拓扑排序时处理def topological_sort(self) -> Optional[List[str]]:"""执行拓扑排序返回：合法的任务执行顺序列表，如果存在环则返回 None"""# 1. 找出所有入度为 0 的节点，放入队列queue = collections.deque()for node, degree in self.in_degree.items():if degree == 0:queue.append(node)result = []processed_count = 0# 2. BFS 遍历过程while queue:current = queue.popleft()result.append(current)processed_count += 1# 遍历当前节点的所有后继节点for neighbor in self.graph[current]:# 后继节点的入度减 1，因为前驱节点已经处理完毕self.in_degree[neighbor] -= 1# 如果后继节点入度变为 0，说明它的所有前置依赖都满足了if self.in_degree[neighbor] == 0:queue.append(neighbor)# 3. 判断是否存在环# 如果处理的节点数少于总节点数，说明有节点永远无法入队（入度不为0），即存在环if processed_count < len(self.in_degree):return Nonereturn result

代码深度解析

数据结构选择：我们使用了两个字典：self.graph 和 self.in_degree。
- self.graph 是邻接表，Dict[str, List[str]]。为什么不用列表？因为节点名称可能是字符串，用字典查找邻居是 O(1) 的，而列表遍历是 O(N)。在大规模图中，这点差异会被放大。
- self.in_degree 记录依赖数。这是 Kahn 算法的灵魂。只有入度为 0，节点才是“自由”的，才能被调度。
为什么用 collections.deque？ 在 Python 中，list.pop(0) 的时间复杂度是 O(N)，因为它需要移动所有后续元素。而 deque.popleft() 是 O(1)。在拓扑排序中，队列操作频繁，使用 deque 是性能优化的关键细节。这一点在 MDN Web Docs 关于 JavaScript 数据结构的文章中也有提及，虽然语言不同，但底层逻辑在高性能计算中是通用的。
环检测逻辑：代码最后有一个判断：if processed_count < len(self.in_degree)。想象一下，如果有 A->B, B->C, C->A 这样的循环。A 的入度是 1（来自 C），B 是 1（来自 A），C 是 1（来自 B）。初始队列是空的！程序直接结束，processed_count 为 0，小于总节点数 3，于是返回 None。这就是最简单的环检测，不需要额外的 DFS 标记栈。

运行与测试：用业务场景验证逻辑

光有算法不够，得跑起来。我们在 demo.py 中模拟一个“网站部署流水线”的场景。

场景描述：

install_deps：安装依赖（无依赖）
lint_code：代码检查（依赖 install_deps）
unit_test：单元测试（依赖 install_deps）
build_docker：构建镜像（依赖 lint_code 和 unit_test）
deploy_prod：生产部署（依赖 build_docker）

from graph_core import CategoryGraphdef main():print("=== 开始构建部署流水线分类图 ===")g = CategoryGraph()# 添加节点和依赖关系g.add_edge("install_deps", "lint_code")g.add_edge("install_deps", "unit_test")g.add_edge("lint_code", "build_docker")g.add_edge("unit_test", "build_docker")g.add_edge("build_docker", "deploy_prod")# 执行拓扑排序order = g.topological_sort()if order:print("合法的执行顺序:")for i, step in enumerate(order, 1):print(f"  {i}. {step}")else:print("错误：检测到循环依赖！")print("\n=== 测试循环依赖 ===")g2 = CategoryGraph()g2.add_edge("A", "B")g2.add_edge("B", "C")g2.add_edge("C", "A") # 形成环 A->B->C->Aorder2 = g2.topological_sort()print(f"检测结果: {order2}") # 应该输出 Noneif __name__ == "__main__":main()

运行结果：

=== 开始构建部署流水线分类图 ===
合法的执行顺序:1. install_deps2. lint_code3. unit_test4. build_docker5. deploy_prod=== 测试循环依赖 ===
检测结果: None

注意看输出，lint_code 和 unit_test 的顺序可能互换，这取决于字典的遍历顺序。在 Python 3.7+ 中，字典是有序的，但在逻辑上，这两个任务是可以并行的。如果业务要求严格串行，你需要在应用层加锁；如果允许并行，这个顺序就是完美的。

优化扩展：从玩具到生产级

上面的代码能跑，但离“生产级”还有距离。以下是几个在实际项目中必须考虑的优化点：

并行执行支持：当前的 topological_sort 返回的是一个线性列表。但在实际部署中，lint_code 和 unit_test 可以同时进行。 改进方案：修改算法，返回“层级列表”（List of List）。每一层的节点可以并行执行。

# 伪代码思路
levels = []
current_level = [node for node in in_degree if in_degree[node] == 0]
while current_level:levels.append(current_level)next_level = []for node in current_level:for neighbor in graph[node]:in_degree[neighbor] -= 1if in_degree[neighbor] == 0:next_level.append(neighbor)current_level = next_level

内存优化：如果节点数量达到百万级，Dict[str, List[str]] 的开销会很大。 改进方案：将节点名称映射为整数 ID。使用 List[List[int]] 存储邻接表。整数的内存占用远小于字符串，且 CPU 缓存友好度更高。
持久化存储：分类图结构经常需要保存和加载。 改进方案：实现 to_dict 和 from_dict 方法，将图结构序列化为 JSON。
```
def to_dict(self):return {"graph": self.graph,"in_degree": self.in_degree}
```
异常处理增强：当前 add_edge 没有检查 source 或 target 是否为空。在生产环境中，输入验证是防止脏数据进入系统的最后一道防线。建议添加 assert source and target。