1. 问题背景与挑战
在数据分析和处理中，我们经常需要计算不同数据项之间的相似性。例如，给定一个包含多个数据项及其属性的字典，我们可能需要计算任意两个数据项之间的余弦相似度。然而，当这些相似性结果被存储时，往往会出现冗余：例如，('a', 'd') 的相似度与 ('d', 'a') 的相似度是相同的，并且我们可能希望将所有相互之间具有相同相似度（例如都为1.0）的条目 ('a', 'd', 'c') 聚合到一起，而不是分别列出所有两两比较的结果。
原始的相似度计算方法通常会生成如下的冗余结果：
{
('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', 'C'): 1.0,
('O', 'L', 'S', 'N', 'P'): 0.412
}这种聚合能够显著减少冗余，并更清晰地展示数据项之间的内在关联。
2. 传统方法及其局限性
一种直观的尝试是使用多层循环和条件判断来构建一个“缓冲区”列表，根据相似度分数逐步添加和合并条目。然而，这种方法很快就会变得复杂，导致代码难以维护，并且在处理大量数据时效率低下，容易陷入嵌套循环和条件判断的“泥潭”。
3. 基于图论的解决方案：最大团问题
更优雅且高效的解决方案是将此问题建模为图论中的最大团（Maximal Clique）问题。
核心思想：
- 构建图： 将字典中的每个数据项视为图中的一个节点（顶点）。
- 定义边： 如果两个数据项之间的相似度达到某个特定值，则在它们之间添加一条边。
- 分离图： 对于每一个不同的相似度分数，我们构建一个独立的图。例如，所有相似度为1.0的对构成一个图，所有相似度为0.412的对构成另一个图。
- 寻找团： 在每个独立的图中，找到所有的最大团。一个团（clique）是一个子图，其中任意两个节点之间都存在一条边。最大团是指不能再通过添加更多节点来扩展的团。在本场景中，一个团内的所有节点都相互之间具有相同的相似度。
通过这种方式，每个最大团就代表了我们想要聚合的一个组，其值就是该团内所有节点之间共同的相似度分数。
4. 使用 networkx 库实现
Python的 networkx 库提供了强大的图论功能，包括查找图中的团。以下是详细的实现步骤。
4.1 准备数据和相似度计算函数
首先，我们需要原始的数据字典和计算余弦相似度的函数。
import math
from itertools import combinations
from collections import defaultdict
import networkx as nx
# 原始数据字典
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
},
# 假设还有其他类似'L', 'S', 'N', 'P'的条目，为了演示，我们只用已有的
'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
},
}
# Cosine similarity function
def square_root(x):
return round(math.sqrt(sum([a * a for a in x])), 3)
def cosine_similarity(a, b):
input1 = {}
input2 = {}
if len(a) > len(b):
input1 = a
input2 = b
else:
input1 = b
input2 = a
vector1 = list(input1.values())
vector2 = []
for k in input1.keys():
if k in input2:
vector2.append(float(input2[k]))
else:
vector2.append(float(0))
numerator = sum(x * y for x, y in zip(vector2, vector1))
denominator = square_root(vector1) * square_root(vector2)
if denominator == 0: # 避免除以零
return 0.0
return round(numerator / float(denominator), 3)
4.2 计算所有唯一对的相似度
使用 itertools.combinations 来生成所有不重复的键对，并计算它们的相似度。
# 获取所有字典键
keys = list(my_dict.keys())
all_pair_similarities = {}
# 计算所有唯一键对的相似度
for k1, k2 in combinations(keys, 2):
sim_score = cosine_similarity(my_dict[k1], my_dict[k2])
all_pair_similarities[(k1, k2)] = sim_score
print("所有唯一键对的相似度:")
print(all_pair_similarities)
# 示例输出:
# {('A', 'D'): 1.0, ('A', 'T'): 1.0, ('A', 'O'): 0.0, ('A', 'L'): 0.0, ('A', 'S'): 0.0,
# ('D', 'T'): 1.0, ('D', 'O'): 0.0, ('D', 'L'): 0.0, ('D', 'S'): 0.0,
# ('T', 'O'): 0.0, ('T', 'L'): 0.0, ('T', 'S'): 0.0,
# ('O', 'L'): 1.0, ('O', 'S'): 1.0, ('L', 'S'): 1.0}4.3 构建基于相似度分数的图
对于每个不同的相似度分数，创建一个 networkx.Graph 对象，并将具有该相似度分数的键对作为边添加到图中。
# 为每个不同的相似度分数构建一个图
graphs_by_similarity = defaultdict(nx.Graph)
for (p, q), s in all_pair_similarities.items():
# 注意：浮点数比较可能存在精度问题，可以考虑对s进行适当的四舍五入
# 例如：graphs_by_similarity[round(s, 5)].add_edge(p, q)
graphs_by_similarity[s].add_edge(p, q)
print("\n按相似度分数构建的图:")
for s, G in graphs_by_similarity.items():
print(f"相似度 {s}: 节点 {G.nodes}, 边 {G.edges}")
# 示例输出:
# 相似度 1.0: 节点 ['A', 'D', 'T', 'O', 'L', 'S'], 边 [('A', 'D'), ('A', 'T'), ('D', 'T'), ('O', 'L'), ('O', 'S'), ('L', 'S')]
# 相似度 0.0: 节点 ['A', 'O', 'L', 'S', 'D', 'T'], 边 [('A', 'O'), ('A', 'L'), ('A', 'S'), ('D', 'O'), ('D', 'L'), ('D', 'S'), ('T', 'O'), ('T', 'L'), ('T', 'S')]4.4 查找最大团并格式化输出
遍历每个相似度图，使用 nx.find_cliques(G) 找到所有的最大团。然后将这些团及其对应的相似度分数收集到最终的输出字典中。
# 查找最大团
grouped_results = {}
processed_nodes = set() # 用于跟踪已经处理过的节点，避免重复输出
for s, G in graphs_by_similarity.items():
# find_cliques返回一个迭代器，生成图中的所有最大团
for clique in nx.find_cliques(G):
# 将团转换为元组并排序，以确保一致性
sorted_clique = tuple(sorted(clique))
# 检查这个团是否已经完全包含在其他团中，或者是否已经处理过
# 这里的逻辑需要根据具体需求调整。
# 如果我们只关心最大的不重叠团，需要更复杂的处理。
# 对于本例，直接添加即可，因为find_cliques找到的是“最大”团。
# 但如果存在 ('A','D') 1.0 和 ('A','D','C') 1.0，find_cliques会给出 ('A','D','C')。
# 确保输出的键是唯一的且代表一个聚合组
# 简单处理：如果团的长度大于1，并且其中的所有节点尚未被其他已记录的团完全覆盖，则记录。
# 这是一个简化逻辑，实际应用中可能需要更精细的去重和合并策略
# 为了避免重复或子集问题，我们只保留长度大于1的团，并且如果一个团是另一个团的子集，我们倾向于保留更大的团。
# networkx.find_cliques 已经确保了找到的是“最大”团，即不能再通过添加一个节点来扩展的团。
# 因此，直接将它们添加到结果中即可，但要确保键的唯一性。
# 最终输出的键是元组，值是相似度
if len(sorted_clique) > 1: # 只考虑包含两个或更多元素的团
grouped_results[sorted_clique] = s
# 清理结果，确保没有重复的团或子团作为独立的键出现
# 由于nx.find_cliques返回的是最大团，通常不需要额外清理子团。
# 但为了确保最终输出的键是唯一的，且符合预期的聚合格式，我们可以进一步处理。
final_grouped_results = {}
for clique_tuple, score in grouped_results.items():
# 转换为集合便于比较
current_clique_set = set(clique_tuple)
# 检查当前团是否是已存在某个更大团的子集
is_subset_of_existing = False
for existing_clique_tuple in list(final_grouped_results.keys()):
existing_clique_set = set(existing_clique_tuple)
if current_clique_set.issubset(existing_clique_set) and current_clique_set != existing_clique_set:
is_subset_of_existing = True
break
# 检查当前团是否包含已存在某个更小团
removed_smaller_cliques = []
for existing_clique_tuple in list(final_grouped_results.keys()):
existing_clique_set = set(existing_clique_tuple)
if existing_clique_set.issubset(current_clique_set) and current_clique_set != existing_clique_set:
removed_smaller_cliques.append(existing_clique_tuple)
for smaller_clique in removed_smaller_cliques:
del final_grouped_results[smaller_clique]
if not is_subset_of_existing:
final_grouped_results[clique_tuple] = score
print("\n最终分组结果:")
print(final_grouped_results)
# 预期输出示例（基于提供的my_dict和cosine_similarity，并假设'O', 'L', 'S'是相似的）：
# {('A', 'D', 'T'): 1.0, ('O', 'L', 'S'): 1.0}完整代码示例：
import math
from itertools import combinations
from collections import defaultdict
import networkx as nx
# 原始数据字典
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
},
'L': { # 假设'L'与'O','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
},
'S': { # 假设'S'与'O','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
},
# 增加一个不完全相似的组，以展示0.412之类的分数
'X': {
'HUE_SAT': 1,
'GROUP_INPUT': 2, # 略有不同
'GROUP_OUTPUT': 1
},
'Y': {
'HUE_SAT': 1,
'GROUP_INPUT': 2, # 略有不同
'GROUP_OUTPUT': 1
},
'Z': {
'HUE_SAT': 1,
'GROUP_INPUT': 2, # 略有不同
'GROUP_OUTPUT': 1
},
}
# Cosine similarity function
def square_root(x):
return round(math.sqrt(sum([a * a for a in x])), 3)
def cosine_similarity(a, b):
input1 = {}
input2 = {}
if len(a) > len(b):
input1 = a
input2 = b
else:
input1 = b
input2 = a
vector1 = list(input1.values())
vector2 = []
for k in input1.keys():
if k in input2:
vector2.append(float(input2[k]))
else:
vector2.append(float(0))
numerator = sum(x * y for x, y in zip(vector2, vector1))
denominator = square_root(vector1) * square_root(vector2)
if denominator == 0:
return 0.0
return round(numerator / float(denominator), 3)
# 1. 计算所有唯一键对的相似度
keys = list(my_dict.keys())
all_pair_similarities = {}
for k1, k2 in combinations(keys, 2):
sim_score = cosine_similarity(my_dict[k1], my_dict[k2])
all_pair_similarities[(k1, k2)] = sim_score
# 2. 为每个不同的相似度分数构建一个图
graphs_by_similarity = defaultdict(nx.Graph)
for (p, q), s in all_pair_similarities.items():
# 浮点数比较可能存在精度问题，建议进行四舍五入
# 这样可以确保例如 0.9999999999999999 和 1.0 被视为相同的相似度
s_rounded = round(s, 5)
graphs_by_similarity[s_rounded].add_edge(p, q)
# 3. 查找最大团并格式化输出
final_grouped_results = {}
for s, G in graphs_by_similarity.items():
for clique in nx.find_cliques(G):
# 团必须至少包含两个元素才构成一个“组”
if len(clique) > 1:
sorted_clique = tuple(sorted(clique))
final_grouped_results[sorted_clique] = s
print("\n最终分组结果:")
# 为了更好的可读性，可以按相似度分数或团大小排序
sorted_results = sorted(final_grouped_results.items(), key=lambda item: (item[1], len(item[0])), reverse=True)
for clique, score in sorted_results:
print(f"{clique}: {score}")
5. 注意事项与最佳实践
- 浮点数精度： 在
