引言：处理字典条目的相似度分组挑战
在数据处理和分析中，我们经常需要识别数据中的相似性并进行分组。当数据以字典形式存在，并且需要根据它们之间的相似度得分来聚合条目时，传统的迭代方法可能会导致代码复杂且难以维护，尤其是在相似度计算结果中存在大量冗余信息时。本文将介绍一种基于图论的优雅解决方案，利用python的networkx库来高效地解决这类问题。
问题背景：冗余相似度数据的困境
假设我们有一个嵌套字典，其中每个键代表一个实体，其值是包含多个属性的子字典：
my_dict = {
'A': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'D': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'T': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'O': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1},
# ... 更多条目
}我们需要计算这些条目两两之间的相似度（例如，使用余弦相似度），并将结果存储在一个字典中。初始的相似度计算可能会产生如下形式的冗余结果：
{
('A', 'D'): 1.0,
('A', 'C'): 1.0,
('D', 'A'): 1.0,
('D', 'C'): 1.0,
('C', 'A'): 1.0,
('C', 'D'): 1.0,
# ...
}其中 ('A', 'D') 和 ('D', 'A') 表示相同的比较，且具有相同的相似度得分。更进一步，如果 A, D, C 三者之间两两相似度都为 1.0，我们希望将它们分组为 ('A', 'D', 'C'): 1.0，而不是列出所有两两组合。手动通过嵌套循环和条件判断来处理这种分组逻辑，会迅速变得复杂且难以管理。
解决方案：图论与最大团（Clique）
将此问题建模为图论中的“最大团”问题，可以提供一个简洁而强大的解决方案。
-
构建图模型：
- 将字典中的每个实体（例如 'A', 'D', 'T', 'O'）视为图中的一个节点（Vertex）。
- 如果两个实体之间的相似度得分达到某个特定值，则在它们之间建立一条边（Edge）。
-
为每个独特的相似度值构建一个图：
- 由于我们希望将所有相互之间具有相同相似度得分的条目分组，因此不能只构建一个图。
- 我们应该为每一个独特的相似度得分（例如 1.0, 0.412 等）构建一个独立的图。
- 在某个特定相似度值 S 的图中，只有当两个节点之间的相似度恰好为 S 时，它们之间才存在一条边。
-
查找最大团（Maximal Cliques）：
- 在图论中，一个团（Clique）是图的一个子集，其中任意两个节点之间都存在边。
- 一个最大团（Maximal Clique）是一个团，它不能通过添加图中的任何其他节点来扩展。换句话说，它是包含尽可能多相互连接节点的子图。
- 通过在为每个相似度值构建的图中查找最大团，我们就能找到所有相互之间具有该特定相似度得分的实体组。
Python的networkx库是一个功能强大的图论库，它提供了高效的算法来构建、操作和分析图，包括查找最大团。
实现步骤与代码示例
1. 计算所有条目间的两两相似度
首先，我们需要一个函数来计算两个子字典之间的相似度。这里使用余弦相似度作为示例。
from math import sqrt
from itertools import combinations
from collections import defaultdict
import networkx as nx
# 示例数据 (请替换为您的实际数据 my_dict)
my_dict = {
'A': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'D': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'T': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'O': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1},
'C': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1}, # 添加C以演示分组
'L': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1} # 添加L以演示分组
}
def square_root(x):
"""计算向量平方和的平方根，并四舍五入到3位小数。"""
return round(sqrt(sum([a * a for a in x])), 3)
def cosine_similarity(a, b):
"""
计算两个字典的余弦相似度。
将字典转换为向量，对于缺失的键，值设为0。
"""
all_keys = sorted(list(set(a.keys()) | set(b.keys())))
vector1 = [a.get(k, 0) for k in all_keys]
vector2 = [b.get(k, 0) for k in all_keys]
numerator = sum(v1 * v2 for v1, v2 in zip(vector1, vector2))
denominator = square_root(vector1) * square_root(vector2)
if denominator == 0:
return 0.0 # 避免除以零
return round(numerator / float(denominator), 3)
# 计算所有条目之间的两两相似度
pairwise_similarities = {}
keys = list(my_dict.keys())
for k1, k2 in combinations(keys, 2): # 使用combinations避免重复和自比较
pairwise_similarities[(k1, k2)] = cosine_similarity(my_dict[k1], my_dict[k2])
print("原始两两相似度示例:")
for (k1, k2), sim in list(pairwise_similarities.items())[:5]: # 打印前5个
print(f" ({k1}, {k2}): {sim}")
print("-" * 30)2. 为每个独特的相似度值构建图
接下来，我们将遍历计算出的两两相似度结果。对于每个独特的相似度得分，我们创建一个networkx.Graph对象，并将具有该相似度得分的条目作为边添加到相应的图中。
# 为每个独特的相似度值构建图
# 使用defaultdict可以自动创建新的图对象
graphs_by_similarity = defaultdict(nx.Graph)
for (p, q), s in pairwise_similarities.items():
# 浮点数精度问题：建议对相似度值进行适当的四舍五入或量化
# 这里使用round(s, 5)来确保相似度值作为字典键时的稳定性
rounded_s = round(s, 5)
graphs_by_similarity[rounded_s].add_edge(p, q)
print("按相似度值构建的图数量:", len(graphs_by_similarity))
print("-" * 30)3. 查找图中的最大团
现在，我们遍历为每个相似度值构建的图，并使用nx.find_cliques()函数来查找每个图中的所有最大团。find_cliques()会返回一个生成器，其中包含每个最大团的节点列表。
# 查找最大团并整合结果
grouped_results = {}
for s_value, G in graphs_by_similarity.items():
# nx.find_cliques(G) 返回图中所有最大团的生成器
for clique in nx.find_cliques(G):
# 将团的节点列表转换为元组，并将其作为键，相似度值作为值
# 只有当团的成员数量大于1时才记录，因为单个节点不是一个“组”
if len(clique) > 1:
grouped_results[tuple(sorted(clique))] = s_value
# 打印最终分组结果
print("最终分组结果:")
# 对结果进行排序以便更好地展示 (可选)
sorted_grouped_results = dict(sorted(grouped_results.items(), key=lambda item: (len(item[0]), item[1]), reverse=True))
for group, sim in sorted_grouped_results.items():
print(f" {group}: {sim}")完整代码示例
将上述所有步骤整合在一起，形成一个完整的可运行脚本：
from math import sqrt
from itertools import combinations
from collections import defaultdict
import networkx as nx
# 1. 原始数据字典
my_dict = {
'A': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'D': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'T': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'O': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1},
'C': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1},
'L': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1},
'S': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1},
'N': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1},
'P': {'GROUP_INPUT': 3, 'MAPPING': 2, 'TEX_NOISE': 2, 'UVMAP': 2, 'VALTORGB': 3,
'GROUP_OUTPUT': 1, 'AMBIENT_OCCLUSION': 1, 'MIX': 4, 'REROUTE': 1,
'NEW_GEOMETRY': 1, 'VECT_MATH': 1},
'E': {'HUE_SAT': 1, 'GROUP_INPUT': 1, 'GROUP_OUTPUT': 1} # 更多相似条目
}
# 2. 余弦相似度计算函数
def square_root(x):
"""计算向量平方和的平方根，并四舍五入到3位小数。"""
return round(sqrt(sum([a * a for a in x])), 3)
def cosine_similarity(a, b):
"""
计算两个字典的余弦相似度。
将字典转换为向量，对于缺失的键，值设为0。
"""
all_keys = sorted(list(set(a.keys()) | set(b.keys())))
vector1 = [a.get(k, 0) for k in all_keys]
vector2 = [b.get(k, 0) for k in all_keys]
numerator = sum(v1 * v2 for v1, v2 in zip(vector1, vector2))
denominator = square_root(vector1) * square_root(vector2)
if denominator == 0:
return 0.0 # 避免除以零
return round(numerator / float(denominator), 3)
# 3. 计算所有条目间的两两相似度
pairwise_similarities = {}
keys = list(my_dict.keys())
for k1, k2 in combinations(keys, 2):
pairwise_similarities[(k1, k2)] = cosine_similarity(my_dict[k1], my_dict[k2])
# 4. 为每个独特的相似度值构建图
graphs_by_similarity = defaultdict(nx.Graph)
for (p, q), s in pairwise_similarities.items():
# 建议对相似度值进行适当的四舍五入或量化，以处理浮点数精度问题
rounded_s = round(s, 5)
graphs_by_similarity[rounded_s].add_edge(p, q)
# 5. 查找最大团并整合结果
grouped_results = {}
for s_value, G in graphs_by_similarity.items():
for clique in nx.find_cliques(G):
# 只有当团的成员数量大于1时才记录，因为单个节点不是一个“组”
if len(clique) > 1:
# 将团的节点列表转换为元组，并进行排序，确保键的唯一性
grouped_results[tuple(sorted(clique))] = s_value
# 6. 打印最终分组结果
print("最终分组结果:")
# 对结果进行排序以便更好地展示 (可选：按组大小降序，然后按相似度降序)
sorted_grouped_results = dict(sorted(grouped_results.items(), key=lambda item: (len(item[0]), item[1]), reverse=True))
for group, sim in sorted_grouped_results.items():
print(f" {group}: {sim}")
运行上述代码，你将得到类似以下输出：
最终分组结果:
('L', 'N', 'O', 'P', 'S'): 1.0
('A', 'C', 'D', 'E', 'T'): 1.0这正是我们期望的结果，它将所有相互之间相似度为 1.0 的实体高效地聚合到一起，避免了冗余。
注意事项
- 浮点数精度： 浮点数比较可能存在精度问题。在将相似度值作为字典键或进行比较时，建议对其进行适当的四舍五入（如 round(s, 5)）或使用容差范围进行比较，以确保相同或非常接近的相似度值被视为相等。
- 最大团算法的计算复杂度： 查找最大团是一个NP-hard问题，这意味着对于非常大的图，其计算时间可能会呈指数级增长。networkx库中的find_cliques函数使用了优化的算法（Bron-Kerbosch算法），但在处理包含数百万节点和边的超大型图时，仍需考虑性能。
- 结果的解读： nx.find_cliques返回的是最大团，这意味着每个团都是一个不能再扩展的完全子图。如果存在多个重叠的团，它们会分别作为独立的结果出现。例如，如果 (A,B,C) 是一个团，(A,B,D) 也是一个团，它们会分别列出。如果需要更高级的聚类（如非重叠聚类），可能需要进一步处理这些最大团，例如使用社区发现算法。
- 适用场景： 这种基于图论和最大团的方法特别适用于需要将相互之间具有特定关系（本例中是特定相似度）的所有实体进行聚合的场景。它提供了一种比手动迭代和条件判断更清晰、更易于维护的解决方案。
- 代码复用： cosine_similarity函数是通用的，可以应用于任何具有数值属性的字典。networkx的图构建和团查找逻辑也是高度可复用的。
总结
通过将字典条目分组问题建模为图论中的最大团问题，并结合networkx库的强大功能，我们能够以一种优雅且高效的方式解决数据中的冗余相似度分组挑战。这种方法不仅避免了复杂的嵌套循环，还提供了清晰的逻辑结构，使得代码更具可读性和可维护性。理解图论的基本概念并熟练运用像networkx这样的库，能够为处理复杂的数据关系提供强大的工具。
