有勇气的牛排博客
☰

拓扑排序 使用DAG思想推理任务执行顺序

进群口令：博客

1 前言

在多任务系统、工作流编排、Agent 执行链（如 OCR → 清洗 → 翻译）、数据处理流程中，都需要解决一个核心问题：

多个任务之间存在依赖关系，该按什么顺序执行？

这正是 拓扑排序（Topological Sort） 的应用场景。

2 题目

任务处理的依赖关系如下：
task_dag = {‘A’: [‘F’], ‘B’: [], ‘C’: [‘A’, ‘B’], ‘D’: [], ‘E’: [‘A’, ‘C’], ‘F’: [‘B’]}
请输出任务的执行顺序？输入为dict，输出为list

给定一组任务及其依赖关系：

task_dag = {
    'A': ['F'],
    'B': [],
    'C': ['A', 'B'],
    'D': [],
    'E': ['A', 'C'],
    'F': ['B']
}

含义说明：

  • A 需要在 F 完成之后才能执行
  • C 需要依赖 A 和 B
  • F 依赖 B
  • B、D 无依赖，最先可执行
  • E 需要 A 和 C

我们的目标是：
输出所有任务的可执行顺序，如 ['B', 'D', 'F', 'A', 'C', 'E']。

这个顺序唯一吗？
不是，D 和 B 先后顺序可变，但必须满足所有依赖关系。

3 解决思路：拓扑排序

拓扑排序用于在 DAG（Directed Acyclic Graph，有向无环图） 中，找出节点的合规执行顺序。

核心逻辑：

  1. 找出所有 没有依赖 的任务（deps == []）→ 可立即执行
  2. 执行这些任务，并从 DAG 中移除它们
  3. 从所有剩余任务中，删除对已执行任务的依赖
  4. 若还存在任务，重复步骤 1
  5. 若某一步找不到可执行任务，则说明图中存在循环依赖（无法执行）

4 完整代码

import copy
from typing import List, Dict


def task_execution_order(task_dag: Dict[str, List[str]]) -> List[str]:
    # 拷贝字典，防止修改原数据
    dag = copy.deepcopy(task_dag)
    # dag = {k: v[:] for k, v in task_dag.items()}
    print(dag)
    run_order = []

    print(task_dag)

    while dag:
        print("====== start ======")

        # 找出依赖为空的任务
        # ready = [task for task, deps in dag.items() if not deps]

        ready = []
        for task, dependencies in dag.items():
            print("查空任务：", task, dependencies)
            if not dependencies:  # 空列表表示没有依赖
                ready.append(task)

        if not ready:
            raise ValueError("存在循环依赖，无法完成拓扑排序")

        # 执行任务
        for task in ready:
            print(f"执行任务：{task}")
            run_order.append(task)  # 记录到顺序
            dag.pop(task)  # 删除已执行任务

        # 删除已执行任务 在其他任务中的依赖
        for deps in dag.values():
            """
            deps 不是副本，是指向 dag[key] 的引用。
            修改 deps 就是在修改 dag[key] 本身。
            """
            for task in ready:
                if task in deps:
                    deps.remove(task)

    return run_order


if __name__ == '__main__':
    task_dag = {
        'A': ['F'],
        'B': [],
        'C': ['A', 'B'],
        'D': [],
        'E': ['A', 'C'],
        'F': ['B']
    }
    run_order = task_execution_order(task_dag)
    print(f"run_order: {run_order}")

DAG任务处理

5 总结：拓扑排序在 DAG 编排中的价值

拓扑排序是多任务系统中最重要的算法之一，广泛用于：

  • 工作流引擎（Workflow）
  • 多模型 Agent 执行链
  • 编译器任务依赖
  • 数据处理 Pipeline
  • 微服务启动顺序
  • 异步任务调度

它能确保：

  • 每个任务都在依赖完成后才执行
  • 不会陷入循环依赖
  • 得到一个稳定可靠的执行顺序

评论区

×
×