如何高效从Python列表中移除同构实例?以NetworkX图为例
通用重复项移除优化问题(以NetworkX同构图为例)
给定一个可判断某类两个实例是否“相同”的布尔函数,如何高效从列表中移除重复项?
我的实际场景是处理NetworkX图,判断函数为networkx.is_isomorphic——同构图指结构相同但networkx.Graph实例属性可不同,目标是从列表li中仅保留结构不同的图。
我自行实现了基础解法:
def check_isomorphisms(G, new_li): for H in new_li: if are_isomorphic(G, H): return True return False def remove_isomorphisms(li): if li == []: return li new_li = [li[0]] for G in li: if check_isomorphisms(G, new_li) is True: continue else: new_li.append(G) return new_li
注:上述代码中are_isomorphic对应networkx.is_isomorphic,但问题具有通用性,只要存在判断两实例是否“相同”的布尔函数即可。
这是常规实例去重问题的变体,但常规场景中“相同”指属性一致,可通过哈希快速实现;而此场景的“相同”定义更通用,无法直接用哈希。
当前解法可行,但列表规模扩大后性能极差:1000个元素需约1分钟,4000个元素则需半小时,而生成原列表仅需数秒。请问是否存在更高效的实现方式?
高效优化方案
1. 预计算“不变量”分组,减少同构检查次数
同构图必然拥有相同的不变量(即结构属性不会随同构变换而改变的特征),先通过这些不变量对列表中的图做分组,仅在同一组内进行同构检查,可大幅减少需要比较的次数。常用的图不变量包括:
- 节点总数
- 边总数
- 排序后的度序列
- 子图计数(比如三角形数量)
- 邻接矩阵的特征值图谱
示例代码思路:
from collections import defaultdict import networkx as nx def get_graph_invariant(G): # 生成可哈希的不变量键,组合节点数、边数、排序后的度序列 node_count = G.number_of_nodes() edge_count = G.number_of_edges() degree_seq = tuple(sorted(d for n, d in G.degree())) return (node_count, edge_count, degree_seq) def remove_isomorphisms_optimized(li): if not li: return [] # 按不变量分组 groups = defaultdict(list) for G in li: key = get_graph_invariant(G) groups[key].append(G) # 每组内做去重 unique_graphs = [] for group in groups.values(): group_unique = [group[0]] for G in group[1:]: if not any(nx.is_isomorphic(G, H) for H in group_unique): group_unique.append(G) unique_graphs.extend(group_unique) return unique_graphs
2. 优化同构检查顺序
- 优先处理小规模图:在分组去重时,先将组内规模最小的图加入结果集,后续新图先与小图比较——小图的同构检查速度更快,能更早终止无效比较。
- 提前过滤:结合更多细粒度不变量(如节点度的频率分布),进一步缩小需要做同构检查的范围。
3. 替换更高效的同构检测算法
NetworkX默认的VF2算法在大规模图场景下性能有限,可尝试:
- 使用
nx.vf2pp_is_isomorphic(VF2++算法),针对无向图有性能优化; - 若图带有可用于区分的节点/边标签,给
is_isomorphic传入node_match/edge_match参数,提前过滤不符合条件的图; - 改用第三方库如
graph-tool,其同构检测实现性能远超NetworkX,适合超大规模图处理。
4. 并行化处理
同构检查是CPU密集型且相互独立的任务,可通过多进程并行处理分组去重:
from collections import defaultdict from multiprocessing import Pool import networkx as nx def get_graph_invariant(G): node_count = G.number_of_nodes() edge_count = G.number_of_edges() degree_seq = tuple(sorted(d for n, d in G.degree())) return (node_count, edge_count, degree_seq) def process_group(group): group_unique = [group[0]] for G in group[1:]: if not any(nx.is_isomorphic(G, H) for H in group_unique): group_unique.append(G) return group_unique def remove_isomorphisms_parallel(li): if not li: return [] groups = defaultdict(list) for G in li: key = get_graph_invariant(G) groups[key].append(G) with Pool() as pool: unique_groups = pool.map(process_group, groups.values()) return [g for group in unique_groups for g in group]
内容的提问来源于stack exchange,提问作者Bulkilol
相关产品推荐
相关产品推荐

