1. 问题背景与目标
在处理复杂数据结构时，我们常会遇到需要按层级或深度遍历的情况。考虑以下场景：给定一个字典my_dict，它表示一个有向图的邻接列表，其中键是节点，值是其直接邻居的列表。我们还拥有一个起始节点列表source_list和一个目标节点列表target_list。我们的目标是从source_list中的节点开始，逐层遍历my_dict，收集每个层级的节点及其邻居，直到我们遇到的邻居节点包含target_list中的元素。最终输出应是一个字典，其键为遍历的层级（迭代次数），值为该层级中所有被访问节点及其邻居的子字典。
例如，给定以下数据：
source_list = ['a', 'b']
target_list = ['x', 'y', 'z']
my_dict = {
'a': ['e'],
'b': ['f', 'd'],
'e': ['g'],
'f': ['t', 'h'],
'd': ['x'],
'g': ['x'],
't': ['y'],
'h': ['z']
}期望的输出是：
{0: {'a': ['e'], 'b': ['f', 'd']},
1: {'e': ['g'], 'f': ['t', 'h'], 'd': ['x']},
2: {'g': ['x'], 't': ['y'], 'h': ['z']}}2. 核心概念：广度优先搜索 (BFS)
广度优先搜索（BFS）是一种用于遍历或搜索树或图的算法。它从图的根（或任意源节点）开始，首先访问其所有邻居节点，然后访问这些邻居的邻居，依此类推。这种“层层推进”的特性使其非常适合解决按层级遍历的问题。
BFS通常使用队列（Queue）数据结构来实现。基本步骤如下：
立即学习“Python免费学习笔记（深入）”；
- 将起始节点加入队列。
- 标记起始节点为已访问。
- 当队列非空时：
- 从队列中取出一个节点。
- 处理该节点（例如，记录其信息）。
- 将其所有未访问过的邻居加入队列，并标记为已访问。
在我们的问题中，我们需要扩展BFS以记录每个节点的层级，并在遇到目标节点时停止进一步探索。
3. BFS 实现方案
我们将构建一个bfs函数来解决这个问题。该函数将接收source（起始节点列表）、target（目标节点列表）和graph（表示图的字典）作为参数。
from collections import deque
def bfs(source, target, graph):
"""
使用广度优先搜索（BFS）按层级提取字典数据。
Args:
source (list): 起始节点列表。
target (list): 目标节点列表。当邻居节点中包含目标节点时，停止进一步探索。
graph (dict): 表示图的字典，键是节点，值是其邻居列表。
Returns:
dict: 结构化输出，键为层级（迭代次数），值为该层级中所有被访问节点及其邻居的子字典。
"""
# 初始化队列，存储 (层级, 节点) 对
queue = deque((0, node) for node in source)
# 将目标列表转换为集合，以便进行O(1)的快速查找
target_set = set(target)
# 记录已访问过的节点，防止循环和重复处理
seen = set(source) # 初始时，source_list中的节点已被视为“已访问”
result = {} # 存储最终结果
while queue:
level, node = queue.popleft() # 取出当前层级和节点
# 确保当前节点在图中存在，避免KeyError
if node not in graph:
continue
neighbors = graph[node] # 获取当前节点的邻居
# 将当前节点及其邻居添加到结果字典中对应层级
# setdefault确保如果层级不存在，则创建一个空字典
result.setdefault(level, {})[node] = neighbors.copy()
# 遍历当前节点的邻居
for neighbor in neighbors:
# 如果邻居节点已访问过，或者邻居节点是目标节点，则不将其加入队列
# 这样可以在达到目标节点时停止进一步探索，并避免循环
if neighbor in seen or neighbor in target_set:
continue
# 标记邻居节点为已访问
seen.add(neighbor)
# 将邻居节点及其下一层级加入队列
queue.append((level + 1, neighbor))
return result
# 示例数据
source_list = ['a', 'b']
target_list = ['x', 'y', 'z']
my_dict = {
'a': ['e'],
'b': ['f', 'd'],
'e': ['g'],
'f': ['t', 'h'],
'd': ['x'],
'g': ['x'],
't': ['y'],
'h': ['z']
}
# 运行BFS函数
output = bfs(source_list, target_list, my_dict)
print(output)输出：
{0: {'a': ['e'], 'b': ['f', 'd']}, 1: {'e': ['g'], 'f': ['t', 'h'], 'd': ['x']}, 2: {'g': ['x'], 't': ['y'], 'h': ['z']}}4. 优化方案：按层级构建结果
上述BFS实现每次从队列中取出一个节点就处理。对于某些场景，如果希望在处理完一个完整层级的所有节点后再统一构建该层级的结果，可以采用一种略微不同的方法。这种方法通过在一个内部循环中处理当前层级的所有节点，从而更明确地按层级组织数据。
from collections import deque
def solution(source, target, graph):
"""
优化版BFS，按层级构建结果。
Args:
source (list): 起始节点列表。
target (list): 目标节点列表。
graph (dict): 表示图的字典。
Returns:
dict: 结构化输出，键为层级，值为该层级中所有被访问节点及其邻居的子字典。
"""
target_set = set(target)
result = {}
# seen集合在初始化时就包含所有source节点，避免重复添加到队列
seen = set(source)
# 队列初始化为所有source节点，不带层级信息，层级在外部循环中管理
queue = deque(source)
level = 0
while queue:
# 调用辅助函数构建当前层级的结果
result[level] = build_level_dict(graph, queue, seen, target_set)
level += 1
return result
def build_level_dict(graph, queue, seen, target_set):
"""
辅助函数，用于构建当前层级的字典。
在内部循环中处理队列中当前层级的所有节点。
"""
# 记录当前层级队列的末尾节点，用于判断何时结束当前层级的处理
# 注意：如果queue为空，此操作会报错。在solution函数中已确保queue非空。
tail = queue[-1]
level_dict = {} # 存储当前层级的节点及其邻居
while True:
node = queue.popleft() # 取出当前节点
# 确保当前节点在图中存在
if node not in graph:
# 如果节点不存在，但它在当前层级被处理，我们仍然需要记录它为空
# 或者选择跳过，取决于具体需求。这里选择跳过。
if node == tail: # 如果是当前层级的最后一个节点，需要跳出循环
return level_dict
continue # 跳过不存在的节点
neighbors = graph[node] # 获取邻居
level_dict[node] = neighbors.copy() # 添加到当前层级结果
for neighbor in neighbors:
# 如果邻居已访问或为目标节点，则不入队
if neighbor in seen or neighbor in target_set:
continue
seen.add(neighbor) # 标记为已访问
queue.append(neighbor) # 加入队列等待下一层级处理
# 如果当前节点是本层级的最后一个节点，则完成本层级处理
if node == tail:
return level_dict
# 示例数据 (与之前相同)
source_list = ['a', 'b']
target_list = ['x', 'y', 'z']
my_dict = {
'a': ['e'],
'b': ['f', 'd'],
'e': ['g'],
'f': ['t', 'h'],
'd': ['x'],
'g': ['x'],
't': ['y'],
'h': ['z']
}
# 运行优化版函数
output_optimized = solution(source_list, target_list, my_dict)
print(output_optimized)输出：
{0: {'a': ['e'], 'b': ['f', 'd']}, 1: {'e': ['g'], 'f': ['t', 'h'], 'd': ['x']}, 2: {'g': ['x'], 't': ['y'], 'h': ['z']}}5. 注意事项与总结
-
seen 集合的重要性：seen集合用于跟踪所有已访问过的节点。它有两个主要作用：
- 防止无限循环：如果图中存在环（循环），没有seen集合会导致BFS无限遍历。
- 避免重复处理：确保每个节点只被处理一次，提高效率。在我们的问题中，source列表中的节点在开始时就被添加到seen中。
- target_set 的作用：将target_list转换为target_set是为了在检查邻居时实现O(1)的平均时间复杂度查找，而不是O(N)的列表查找，这在目标列表较大时能显著提升性能。
- 停止条件：当一个节点的邻居是target_set中的元素时，该邻居不会被添加到队列中。这意味着我们只收集到target节点“前”一层的结构，target节点本身不会作为新的起始节点被进一步探索。这符合问题中“直到我达到目标列表中的值”的要求。
- 图的表示：my_dict作为邻接列表表示有向图。如果图中存在键但没有值（例如'k': []），或者键不存在（例如尝试访问graph['non_existent_node']），需要进行适当的错误处理或检查（例如使用graph.get(node, [])或if node in graph:）。上述代码中已添加了if node not in graph: continue的检查。
- deque 的使用：collections.deque（双端队列）比标准Python列表在两端添加和删除元素时效率更高，因此是实现队列的理想选择。
- 优化方案的适用性：第二种优化方案通过在内部循环中一次性处理一个层级的所有节点，可能在某些性能测试中略快，因为它减少了外部循环的迭代次数。然而，两种方案在功能和渐近时间复杂度上是等效的。
通过以上两种广度优先搜索的实现，我们可以高效地从复杂的字典结构中，按照指定的层级和停止条件提取所需的数据，这在图遍历、网络分析等领域具有广泛的应用价值。
