← 中文资源

简体中文版本

拓扑排序构建依赖并检测循环

用 Kahn 算法从依赖图生成构建顺序,遇到循环依赖时明确报错。

阅读英文原文

功能说明

这段代码把“任务到前置依赖”的映射转换成可执行的先后顺序。它适合构建系统、迁移脚本或初始化流程:只有当一个节点的所有前置依赖都完成后,该节点才会进入就绪队列。如果图中存在循环,函数不会静默返回残缺顺序,而是抛出包含阻塞节点的错误。

实现使用 Kahn 算法,并用 dependents 反向索引记录“某个依赖被哪些节点需要”,从而避免反复扫描整张图。

关键代码

build_order 先为每个节点统计入度:一条 node -> dep 的记录会让 dep 的入度保持零,同时让 node 的入度加一。所有入度为零的节点按名称排序后放入 deque。

取出节点后,把它加入结果,并遍历其 dependents,将子节点入度减一;减到零时追加到队列。循环结束后,如果结果长度小于节点总数,说明仍有节点入度大于零,代码会收集这些节点并抛出 ValueError。

使用注意

依赖列表中的重复项会让入度重复累加,但 dependents 也会重复记录,因此最终仍能抵消;不过最好在输入阶段去重。检测到的“阻塞节点”可能包含环下游的节点,不一定只是环本身。函数不校验节点名是否存在,未知依赖会被当作独立节点加入。

排序结果只保证拓扑正确,不保证全局字典序最小;如需稳定输出,可改用优先队列。图很大时应关注内存占用,因为需要保存完整邻接表。

验证方式

示例图会输出 db -> docs -> auth -> api -> app,代码断言 db 在 api 之前、api 在 app 之前。随后构造 a 依赖 b、b 又依赖 a 的环,捕获异常并打印 dependency cycle among: a, b。可以再加入自依赖 x: ["x"] 验证同样会被拒绝。

代码示例

from __future__ import annotations

from collections import defaultdict, deque


def build_order(prerequisites: dict[str, list[str]]) -> list[str]:
    indegree = {node: 0 for node in prerequisites}
    dependents: dict[str, list[str]] = defaultdict(list)
    for node, deps in prerequisites.items():
        for dep in deps:
            indegree.setdefault(dep, 0)
            dependents[dep].append(node)
            indegree[node] += 1

    ready = deque(sorted(node for node, degree in indegree.items() if degree == 0))
    order: list[str] = []
    while ready:
        node = ready.popleft()
        order.append(node)
        for child in sorted(dependents[node]):
            indegree[child] -= 1
            if indegree[child] == 0:
                ready.append(child)

    if len(order) != len(indegree):
        blocked = sorted(node for node, degree in indegree.items() if degree > 0)
        raise ValueError(f"dependency cycle among: {', '.join(blocked)}")
    return order


def main() -> None:
    graph = {
        "app": ["api", "db"],
        "api": ["db", "auth"],
        "auth": ["db"],
        "db": [],
        "docs": [],
    }
    order = build_order(graph)
    print(" -> ".join(order))
    assert order.index("db") < order.index("api") < order.index("app")
    try:
        build_order({"a": ["b"], "b": ["a"]})
    except ValueError as exc:
        print(f"cycle: {exc}")


if __name__ == "__main__":
    main()