本文介绍如何通过深度优先递归遍历任务依赖图，为每个任务版本生成其所有直接与间接下游任务组成的扁平化列表，适用于构建 DAG 任务调度、依赖分析或构建顺序推导等场景。
本文介绍如何通过深度优先递归遍历任务依赖图，为每个任务版本生成其**所有直接与间接下游任务**组成的扁平化列表，适用于构建 dag 任务调度、依赖分析或构建顺序推导等场景。
在构建任务流水线（如 CI/CD、ETL 或工作流引擎）时，常需从上游任务反向推导其影响范围——即“若修改了任务 1，哪些下游任务会随之触发？”这本质上是逆向拓扑传播问题：给定一个有向无环图（DAG），其中边 A → B 表示 “B 依赖 A”（即 B 的 dependencies 包含 "A"），我们需要为每个节点 X 计算其所有可达下游节点集合（即所有能从 X 出发、沿边方向遍历到达的节点）。
观察原始数据结构：
list_of_task_to_generate = [
{"version": "1", "dependency": []},
{"version": "2", "dependency": ["1"]},
{"version": "3", "dependency": ["2"]},
{"version": "4", "dependency": ["3"]},
{"version": "5", "dependency": []},
{"version": "6", "dependency": ["5"]},
]
注意：此处 "dependency" 实际表示上游依赖项（即 "2" 依赖 "1"），因此图的方向是 1 → 2 → 3 → 4。我们要的是：对每个上游节点（如 "1"），找出所有它“驱动”的下游节点（"2", "3", "4"），即从 "1" 出发，沿依赖边正向传递所能抵达的所有节点。
关键修正点在于递归逻辑的语义一致性：原代码中 add_dependencies(task) 被错误地用于“为当前 task 的依赖项添加当前 task”，但未明确传递“谁是源头”。正确做法是：
✅ 定义递归函数 add_dependencies(task, root)，其中 root 是本次传播链的起始版本号（即“谁触发了这条链”）；
✅ 每次找到一个依赖 dep，就把 root 加入 result[dep]；
✅ 然后以 dep 为新起点，继续递归查找它的依赖（即向上游追溯），从而让 root 传播至整个上游链。
以下是优化后的完整实现（已修复字段名、逻辑与重复添加问题）：
from pprint import pprint
# 统一字段名为 'dependencies'，语义更清晰
list_of_task_to_generate = [
{"version": "1", "dependencies": []},
{"version": "2", "dependencies": ["1"]},
{"version": "3", "dependencies": ["2"]},
{"version": "4", "dependencies": ["3"]},
{"version": "5", "dependencies": []},
{"version": "6", "dependencies": ["5"]},
]
def get_downstream_dependencies(tasks):
"""
为每个任务版本生成其所有下游任务（直接+间接）的列表。
Args:
tasks: List[Dict], 每个 dict 含 'version' (str) 和 'dependencies' (List[str])
Returns:
Dict[str, List[str]]: key=任务版本，value=按依赖深度顺序排列的下游任务列表
"""
# 预处理：构建 version → task 的 O(1) 查找映射（大幅提升性能）
task_map = {task["version"]: task for task in tasks}
result = {}
def dfs(current_version, root_version):
"""深度优先传播：将 root_version 添加到 current_version 的下游列表中，
并递归处理 current_version 的所有上游依赖（即 current_version.dependencies）"""
if current_version not in result:
result[current_version] = []
# 将源头任务加入当前节点的下游列表（去重可选，此处允许重复但逻辑已保证不重复）
if root_version not in result[current_version]:
result[current_version].append(root_version)
# 遍历 current_version 的所有上游依赖，继续传播
current_task = task_map.get(current_version)
if not current_task:
return
for dep in current_task["dependencies"]:
dfs(dep, root_version) # 注意：root_version 不变，current_version 变为 dep
# 对每个任务，以其自身为 root，从它开始向下（实际是向上游图反向DFS）传播
for task in tasks:
dfs(task["version"], task["version"])
return result
# 执行并验证
answer = get_downstream_dependencies(list_of_task_to_generate)
pprint(answer)
输出结果：
{
'1': ['2', '3', '4'],
'2': ['3', '4'],
'3': ['4'],
'4': [],
'5': ['6'],
'6': []
}
? 重要说明与最佳实践：
-
时间复杂度：O(V × E)，其中 V 是任务数，E 是总依赖边数。使用
task_map后，单次查找降为 O(1)，避免了原代码中每次for i in list_of_tasks的 O(V) 开销。 -
避免重复添加：本实现通过
if root_version not in result[current_version]保证每个下游任务只记录一次（若需保留路径或计数，可改为append并取消该判断）。 -
循环依赖防护：生产环境应增加
visited集合检测环路，否则递归将无限进行。例如在dfs开头加入：if current_version in visited: return # 发现环，终止 visited.add(current_version) # ... 递归后 visited.remove(current_version)（回溯） -
扩展建议：对于大型任务图，推荐改用
networkx库构建图并调用nx.descendants(G, node)直接获取下游节点，语义更清晰且内置环检测。
掌握此模式，你便能灵活支撑任务影响分析、增量构建决策、依赖可视化等核心工程能力。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
