如何遍历Python图结构字典并按不同深度获取叶子节点，同时实现并行处理？

刚好我之前处理过类似的图遍历和并行任务，咱们一步步来解决你的问题：

首先明确：你的图结构里，叶子节点就是children为空列表的节点对吧？我写两个方法，递归和迭代的，分别适合不同场景：

递归法（代码简洁，适合层级不深的图） #

这个函数会返回一个字典，key是节点深度（根节点算0层），value对应深度的所有叶子节点的Node值：

def get_all_leaf_nodes(graph, current_depth=0, depth_map=None):
    if depth_map is None:
        depth_map = {}
    # 判断当前节点是否是叶子
    if not graph['children']:
        if current_depth not in depth_map:
            depth_map[current_depth] = []
        depth_map[current_depth].append(graph['Node'])
        return depth_map
    # 递归遍历子节点，深度+1
    for child in graph['children']:
        get_all_leaf_nodes(child, current_depth + 1, depth_map)
    return depth_map

用你的示例图测试：

example_graph = {'Node': [0, 0, 0], 'children': [{'Node': [1, 0, 0], 'children': [{'Node': [2, 0, 0], 'children': []}, {'Node': [1, 1, 0], 'children': []}, {'Node': [1, 0, 1], 'children': []}]}, {'Node': [0, 1, 0], 'children': [{'Node': [1, 1, 0], 'children': []}, {'Node': [0, 2, 0], 'children': []}, {'Node': [0, 1, 1], 'children': []}]}, {'Node': [0, 0, 1], 'children': [{'Node': [1, 0, 1], 'children': []}, {'Node': [0, 1, 1], 'children': []}, {'Node': [0, 0, 2], 'children': []}]}]}

depth_map = get_all_leaf_nodes(example_graph)
print(depth_map[2])  # 输出所有深度为2的叶子节点：[[2,0,0], [1,1,0], [1,0,1], ...]

迭代法（适合超深层级的图，避免递归栈溢出） #

用栈模拟递归，逻辑和递归一致，但更稳定：

def get_leaf_nodes_by_depth_iterative(graph, target_depth=None):
    stack = [(graph, 0)]
    depth_map = {}
    while stack:
        node, depth = stack.pop()
        if not node['children']:
            if depth not in depth_map:
                depth_map[depth] = []
            depth_map[depth].append(node['Node'])
            continue
        # 反转子节点是为了保持和递归一样的遍历顺序，不需要的话可以去掉reverse
        for child in reversed(node['children']):
            stack.append((child, depth + 1))
    # 如果指定了目标深度，直接返回该深度的叶子，否则返回所有
    return depth_map.get(target_depth, []) if target_depth else depth_map

多进程的核心注意点：进程间内存隔离，所以不能直接在子进程里修改原字典，最好是把叶子节点的数据传递给子进程，处理完返回新的节点结构，再替换回原图。

先写你的添加子节点函数（这里模拟一个示例）：

def add_child_to_node(node_data):
    # 这里写你实际的添加子节点逻辑，比如生成新的子节点
    new_child = {'Node': [x + 1 for x in node_data], 'children': []}
    # 返回更新后的节点（原叶子现在有了子节点）
    return {'Node': node_data, 'children': [new_child]}

用multiprocessing实现并行 #

import multiprocessing

def process_leaves_parallel(leaf_nodes):
    # 用CPU核心数创建进程池，也可以手动指定processes=4这种
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
        # map函数自动把每个叶子节点分配给子进程处理
        updated_nodes = pool.map(add_child_to_node, leaf_nodes)
    return updated_nodes

用joblib实现并行（更简洁） #

joblib对numpy数组、字典这类数据的处理更友好，代码更短：

from joblib import Parallel, delayed

def process_leaves_joblib(leaf_nodes):
    # n_jobs=-1表示用所有CPU核心
    updated_nodes = Parallel(n_jobs=-1)(
        delayed(add_child_to_node)(node) for node in leaf_nodes
    )
    return updated_nodes

把处理后的节点替换回原图 #

最后需要一个辅助函数，把更新后的叶子节点替换到原结构图里：

def replace_leaf_node(graph, target_node_data, new_node):
    if not graph['children']:
        if graph['Node'] == target_node_data:
            return new_node
        return graph
    # 递归遍历子节点，替换目标节点
    new_children = [replace_leaf_node(child, target_node_data, new_node) for child in graph['children']]
    graph['children'] = new_children
    return graph

# 调用示例
leaf_nodes_data = get_leaf_nodes_by_depth_iterative(example_graph, target_depth=2)
updated_leaves = process_leaves_parallel(leaf_nodes_data)

# 逐个替换原叶子节点
for original_data, updated_node in zip(leaf_nodes_data, updated_leaves):
    example_graph = replace_leaf_node(example_graph, original_data, updated_node)

普通字典的叶子节点定义：值不是非空字典（或按需扩展到列表等容器）的键值对。这里写一个通用的遍历函数，能返回叶子节点的路径和值：

def get_plain_dict_leaves(d, parent_key='', sep='.'):
    leaves = []
    for k, v in d.items():
        # 生成当前节点的路径键，比如children[0].Node
        current_key = f"{parent_key}{sep}{k}" if parent_key else k
        # 如果是不为空的字典，继续递归
        if isinstance(v, dict) and v:
            leaves.extend(get_plain_dict_leaves(v, current_key, sep))
        # 如果是字典列表，遍历每个元素（可选，根据你的需求调整）
        elif isinstance(v, list) and all(isinstance(item, dict) for item in v):
            for i, item in enumerate(v):
                list_key = f"{current_key}[{i}]"
                leaves.extend(get_plain_dict_leaves(item, list_key, sep))
        # 否则就是叶子节点
        else:
            leaves.append((current_key, v))
    return leaves

用你的示例图测试，会返回类似这样的结果：

[('Node', [0, 0, 0]), ('children[0].Node', [1, 0, 0]), ('children[0].children[0].Node', [2, 0, 0]), ('children[0].children[0].children', []), ...]

如果不需要处理列表，直接把列表当作叶子值，可以去掉列表的判断分支，简化函数。

内容的提问来源于stack exchange，提问作者Nicola Alimonda