提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  其他

当前栏目

【算法优化】文本相似性去重算法优化实践【终结篇】

算法 索引
2023-03-20 14:52:10 时间

1、前一版本算法分析

前一版本的算法主要时间是消耗在合并有交集的类别上，代码结构如：

for i, cls in enumerate(clses):
    for cls_j in clses[i+1:]:
        # ......

显然这里是一个两层循环，平方的时间复杂度，如果列表有1万个元素，这就变成了1亿次循环（平方就是这么可怕）！

那这里优化的方向：

  1. 是否减少输入的列表的长度？
  2. 循环能否转成矩阵计算（并行计算）？

开始时，只要是从这两方面进行考虑。第二点实现不了，因为循环时对前一个状态有依赖，没法并行，那就只剩下第一点了。减少列表的输入长度是可行的，不过这只是量变，产生不了质变，只能缓解一下问题，可能从耗时可以从二十几秒降到十几秒，只要数据稍微增加一点，时间马上就回去了。

2、相似文章的特点

我们再回头理解理解相似文章，前面我们已经假设将两个文章的simhash码均分为4段之后，如果它们是相似的，那至少有一段是完全相同的。

一段simhash码的子串有16个二进制位，该子串的表达空间最大有2的16次方（共65536），就算有一亿篇文章，如果均分成65536份，那也变得很小（当然实际的分布并不是均匀的），如果从这点出发，那计算的时间复杂度将会大大降低。

重新梳理一下这算法流程，假设有三篇文章，每篇文章有4个simhash子串：

文章1: A1, B1, C1, D1
文章2: A2, B2, C2, D3
文章3: A3, B3, C3, D3

按照我们的假设，如果文章1和文章2相似，则下面4个条件必须至少有一个成立：

A1 == A2
B1 == B2
C1 == C2
D1 == D2

也就是说，也就是只要通过文章1的4个simhash子串就能找到该文章的所有相似文章（当然召回率不会是100%，但是可以忽略，本身simhash就会有误差）。而且，这恰好能满足我们的需求。

那就三个步骤：

  1. 使用文章的simhash子串建立索引；
  2. 通过索引找到相似文章的候选集；
  3. 从候选集里选择真正的相似文章。

很清晰。

3、第四个优化版本

建立simhash子串的索引：

# 建立索引
print(len(data))
start = time.time()
child_key_dict = {}
for item in data:
    simhash, article_id = item['simhash'], item['id']
    simhash = '0' * (64-len(simhash)) + simhash
    item['simhash'] = simhash   # 格式simhash
    # 不足64位的前面补0，文章id放到第一个值
    item = article_id + simhash
    for i in range(4):
        child_key_dict.setdefault(str(i)+simhash[i*16:i*16+16], []).append(item)
# 文章id到hash值的索引
aid_simhash_dict = {item['id']: item['simhash'] for item in data}
print('time: ', time.time()-start, len(child_key_dict))

筛选候选集及从候选集中过滤出真正的相似文章：

# 循环找相似文章
sim_articles = []
checked_aid_dict = {item['id']: False for item in data}
for item in data:
    if checked_aid_dict[item['id']]:
        continue
    simhash, article_id = item['simhash'], item['id']
    checked_aid_dict[article_id] = True
    # 可能相似的文章
    sim_articles_maybe = []
    for i in range(4): 
        key = str(i) + simhash[i*16:i*16+16]
        sim_articles_maybe += child_key_dict[key]
    # 去重文章及已经被归类的文章
    sim_articles_maybe = set(sim_articles_maybe)
    sim_articles_maybe.remove(article_id+simhash)
    sim_articles_maybe = list(sim_articles_maybe)
    sim_articles_maybe = [aid for aid in sim_articles_maybe if not checked_aid_dict[aid[:-64]]]
    if len(sim_articles_maybe) == 0:
        sim_articles.append([article_id])
        continue

    # 找相似的文章
    sim_articles_maybe = [[aid[:-64]] + list(aid[-64:]) for aid in sim_articles_maybe]
    sim_articles_maybe = np.array(sim_articles_maybe)
    sim_aids = find_sim_aids(np.array([article_id]+list(simhash)), sim_articles_maybe)
    if len(sim_aids) == 0:
        sim_articles.append([article_id])
        continue
    sim_articles.append([article_id]+sim_aids)
    for aid in sim_aids:
        checked_aid_dict[aid] = True

print('cluster time: ', time.time()-start, len(sim_articles), len(data))

这基本就是线性的时间复杂度，相对于第三个版本的平方时间复杂度，那是快多了，1.7万的文章进行相似性过滤时间能降到一秒以下。

软烟罗

4、小结

通过索引快速过滤出候选集（有点类似ES的意思），牺牲一点点精度，这就能大大减少无意义的计算。

相关文章