本文介绍一种基于深度优先遍历的递归方法，用于从任务定义列表中构建每个任务版本的完整下游依赖链表，解决重复添加和递归路径中断问题，并提供可直接运行的健壮实现。
本文介绍一种基于深度优先遍历的递归方法，用于从任务定义列表中构建每个任务版本的完整下游依赖链表，解决重复添加和递归路径中断问题，并提供可直接运行的健壮实现。
在构建工作流调度系统、CI/CD 任务图或 DAG 解析器时，常需将“上游 → 下游”的单层依赖关系（如 {"version": "2", "dependencies": ["1"]} 表示任务2依赖任务1）反向展开为每个上游任务的所有可达下游任务列表。例如：任务 "1" 不仅直接被 "2" 依赖，还经由 "2"→"3"→"4" 间接影响 "3" 和 "4"，因此其下游依赖应为 ["2", "3", "4"]。
原始代码的核心问题在于：递归调用 add_dependencies(i) 时未传递当前正在追溯的根任务版本（即“谁在依赖我”），导致每次递归都错误地将子任务的 version（如 "2"）作为下游节点追加到其直接依赖（如 "1"）的列表中，而未能继续向更深层传播该根节点（如 "1" 对 "4" 的间接影响）。简言之：缺少“源头标识”，递归无法跨层级传递依赖路径。
以下是修正后的专业实现，采用显式 root_version 参数确保依赖传播一致性：
from pprint import pprint
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 generate_downstream_map(tasks):
"""
构建下游依赖映射：{task_version -> [all_direct_and_indirect_downstreams]}
Args:
tasks: List[dict]，每个 dict 含 'version' (str) 和 'dependencies' (List[str])
Returns:
Dict[str, List[str]]：键为任务版本，值为按拓扑序排列的下游任务列表（无重复）
"""
# 预处理：构建 O(1) 版本查找表，避免每次递归线性扫描
task_map = {task["version"]: task for task in tasks}
result = {}
def dfs(current_version, root_version):
"""深度优先搜索，将 root_version 添加到 current_version 的所有上游依赖项中"""
if current_version not in result:
result[current_version] = []
# 将根任务（发起本次追溯的任务）加入当前节点的下游列表
if root_version != current_version: # 避免自依赖
result[current_version].append(root_version)
# 递归处理 current_version 的所有直接依赖（即它的上游任务）
for upstream in task_map.get(current_version, {}).get("dependencies", []):
dfs(upstream, root_version)
# 对每个任务，以它为根发起一次 DFS，收集其所有下游
for task in tasks:
dfs(task["version"], task["version"])
# 可选：对每个列表去重并保持拓扑顺序（此处输入为链式，已天然有序）
for version in result:
result[version] = list(dict.fromkeys(result[version])) # 去重保序
return result
# 执行并验证
downstream_map = generate_downstream_map(list_of_task_to_generate)
pprint(downstream_map)
输出结果：
{
'1': ['2', '3', '4'],
'2': ['3', '4'],
'3': ['4'],
'4': [],
'5': ['6'],
'6': []
}
✅ 关键改进点说明：
-
明确递归上下文：
dfs(current_version, root_version)中root_version恒定标识“最初被依赖的任务”，确保"1"能正确传播至"4"； -
O(1) 查找优化：通过
task_map预构建字典索引，取代原代码中for i in list_of_tasks的低效遍历； -
鲁棒性增强：添加
get(..., {})防御缺失字段，root_version != current_version避免自引用； -
去重保序：使用
dict.fromkeys()在必要时消除因图环（虽本例无环）或多路径导致的重复。
⚠️ 注意事项：
- 此算法假设依赖关系为有向无环图（DAG）。若存在循环依赖（如
"1"依赖"2"且"2"依赖"1"），需额外引入visited集合防止无限递归； - 若任务量极大（>10⁴），建议改用迭代 DFS 或 BFS 避免 Python 默认递归深度限制；
- 生产环境推荐将输入结构化为
dict[version, Task]而非列表，进一步提升可维护性与性能。
该方案简洁、可扩展，是解析任务依赖拓扑的通用范式。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
