如何在Python中实现DAG依赖拆分：多依赖节点复制

需求说明 #

需要拆分DAG依赖结构：将每个被多个节点依赖的节点复制为多个独立副本，每个副本仅对应一个依赖方，同时同步复制副本的所有子依赖链，最终确保每个节点仅被一个节点依赖。

示例输入 #

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

期望输出 #

expected_dag_dependencies = {
    'A': ['B', 'C'],
    'B': ['D1'],
    'C': ['D2'],
    'D1': ['E1', 'F1'],
    'D2': ['E2', 'F2'],
    'E1': [],
    'F1': [],
    'E2': [],
    'F2': [],
}

解决方案实现 #

BFS算法完全适用这个场景，结合递归子树复制可以高效完成节点拆分。以下是完整代码：

from collections import deque

def calculate_indegree(dag):
    """计算DAG中每个节点的入度"""
    indegree = {node: 0 for node in dag}
    for node in dag:
        for dep in dag[node]:
            indegree[dep] += 1
    return indegree

def get_root_nodes(dag):
    """找出DAG的根节点（入度为0的节点）"""
    indegree = calculate_indegree(dag)
    return [node for node in indegree if indegree[node] == 0]

def copy_subtree(original_node, copy_id, dag, result, copy_counter):
    """递归复制原始节点的整个子树，生成带编号的副本"""
    copy_name = f"{original_node}{copy_id}"
    # 初始化副本的子依赖列表
    copy_children = []
    for child in dag[original_node]:
        # 为子节点生成新的副本编号
        copy_counter[child] += 1
        child_copy_name = f"{child}{copy_counter[child]}"
        copy_children.append(child_copy_name)
        # 递归复制子节点的子树
        copy_subtree(child, copy_counter[child], dag, result, copy_counter)
    # 将副本添加到结果字典
    result[copy_name] = copy_children

def clean_unused_nodes(result, root_nodes):
    """移除结果中未被依赖的原始节点（根节点除外）"""
    indegree = {node: 0 for node in result}
    for node in result:
        for dep in result[node]:
            indegree[dep] += 1
    # 找出需要删除的节点：入度为0且不是根节点
    to_remove = [node for node in result if indegree[node] == 0 and node not in root_nodes]
    for node in to_remove:
        del result[node]
    return result

def split_dag(dag):
    """主函数：拆分DAG的多依赖节点"""
    # 计算每个节点的入度，确定需要复制的节点
    indegree = calculate_indegree(dag)
    # 初始化结果字典，复制原始DAG结构
    result = {node: dag[node].copy() for node in dag}
    # 记录每个原始节点的副本编号计数器
    copy_counter = {node: 0 for node in dag}
    # 队列存储需要处理的（父节点，依赖节点）对
    queue = deque()
    # 记录已处理过的依赖关系，避免重复操作
    processed = set()

    # 收集所有需要替换的依赖关系：依赖节点入度>1的情况
    for parent_node in dag:
        for dep_node in dag[parent_node]:
            if indegree[dep_node] > 1:
                queue.append((parent_node, dep_node))

    # 处理队列中的依赖替换请求
    while queue:
        parent, original_dep = queue.popleft()
        if (parent, original_dep) in processed:
            continue
        processed.add((parent, original_dep))

        # 生成新的副本编号和名称
        copy_counter[original_dep] += 1
        copy_id = copy_counter[original_dep]
        copy_name = f"{original_dep}{copy_id}"

        # 复制原始依赖节点的整个子树
        copy_subtree(original_dep, copy_id, dag, result, copy_counter)

        # 将父节点的依赖替换为新副本
        result[parent] = [copy_name if n == original_dep else n for n in result[parent]]

    # 清理无用的原始节点
    root_nodes = get_root_nodes(dag)
    result = clean_unused_nodes(result, root_nodes)

    return result

# 测试示例
if __name__ == "__main__":
    dag_dependencies = {
        'A': ['B', 'C'],
        'B': ['D'],
        'C': ['D'],
        'D': ['E', 'F'],
        'E': [],
        'F': [],
    }
    split_result = split_dag(dag_dependencies)
    # 格式化输出结果
    print("{")
    for key, value in split_result.items():
        print(f"    '{key}': {value},")
    print("}")

代码说明 #

1. 入度计算：通过calculate_indegree函数统计每个节点被多少个节点依赖，入度>1的节点就是需要复制的目标。
2. 子树复制：copy_subtree递归生成原始节点的所有子节点副本，确保整个依赖链同步复制。
3. 依赖替换：使用BFS队列处理所有需要替换的依赖关系，为每个父节点生成专属的依赖副本，并替换原依赖。
4. 清理节点：clean_unused_nodes移除结果中没有被任何节点依赖的原始节点，保证最终结构简洁。

运行代码后，输出结果将与期望的DAG结构完全一致。

内容的提问来源于stack exchange，提问作者ADAM King