Python中5万+推文列表相似度比对的快速优化方法咨询

哇，5万条推文用itertools.combinations遍历所有对？那可是超过10亿次比对，慢到天荒地老太正常了😂。咱们得换个思路，把O(n²)的复杂度砍下来，给你几个实用的优化方向，从粗到细帮你提速：

两条推文如果长度差太大，根本不可能达到75%的相似度。比如假设推文A长度是L，那推文B的长度必须在0.75*L到L/0.75（约1.33倍L）之间，否则最长公共子序列的占比肯定到不了75%。我们可以先按长度分组，只在长度相近的组里做比对：

from collections import defaultdict
import itertools

# 按长度区间分组，比如每5个字符为一个区间，减少分组数量
length_groups = defaultdict(list)
for idx, tweet in enumerate(tweets):
    length_key = len(tweet) // 5 * 5
    length_groups[length_key].append((idx, tweet))

# 只在当前组和相邻组（避免跨区间的漏判）生成候选对
candidate_pairs = []
sorted_lengths = sorted(length_groups.keys())
for i, length in enumerate(sorted_lengths):
    current_group = length_groups[length]
    # 加入前一个相邻长度组的内容
    if i > 0:
        current_group.extend(length_groups[sorted_lengths[i-1]])
    # 只在组内生成组合，比对量会大幅减少
    for (idx1, t1), (idx2, t2) in itertools.combinations(current_group, 2):
        candidate_pairs.append((idx1, t1, idx2, t2))

SimHash是专门针对海量文本去重的算法，核心是把每个文本转换成64位哈希值，相似文本的哈希值汉明距离极小（比如≤3）。我们可以先计算所有推文的SimHash，快速把候选对从10亿级压缩到几万级，再做精确验证：

import simhash
from collections import defaultdict
import itertools
from difflib import SequenceMatcher

def compute_simhash(text, ngram=3):
    # 把文本拆成3-gram特征，计算SimHash值
    tokens = [text[i:i+ngram] for i in range(len(text)-ngram+1)] if len(text)>=3 else [text]
    return simhash.SimHash(tokens).value

# 计算所有推文的SimHash和索引
hash_records = [(compute_simhash(tweet), idx, tweet) for idx, tweet in enumerate(tweets)]

# 按哈希的高6位分桶，汉明距离近的文本大概率在同一个桶里
bucket_bits = 6
buckets = defaultdict(list)
for h_val, idx, tweet in hash_records:
    bucket_key = h_val >> (64 - bucket_bits)
    buckets[bucket_key].append((h_val, idx, tweet))

# 只在桶内比对汉明距离，再做精确相似度验证
duplicate_pairs = set()
for bucket in buckets.values():
    for (h1, idx1, t1), (h2, idx2, t2) in itertools.combinations(bucket, 2):
        # 计算汉明距离
        hamming_dist = bin(h1 ^ h2).count('1')
        if hamming_dist <= 3:
            # 用difflib做精确验证
            if SequenceMatcher(None, t1, t2).ratio() >= 0.75:
                # 存索引对，避免重复（比如只存idx1<idx2的）
                if idx1 < idx2:
                    duplicate_pairs.add((idx1, idx2))
                else:
                    duplicate_pairs.add((idx2, idx1))

这个方法的时间复杂度接近O(n)，5万条推文的处理时间会从几天降到几十分钟。

如果你的需求是语义相似而非仅仅字面重复，可以把推文转换成向量，用FAISS/Annoy这样的库快速找近似最近邻，只比对每个文本的Top-N候选：

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
from difflib import SequenceMatcher

# 加载轻量级预训练模型，生成语义向量
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(tweets, show_progress_bar=True)

# 构建FAISS索引，用于快速搜索近似最近邻
index = faiss.IndexFlatL2(embeddings.shape[1])
index.add(embeddings)

# 每个推文找Top-50相似候选（可根据需要调整）
k = 50
distances, indices = index.search(embeddings, k)

# 遍历候选对，做精确验证
duplicate_pairs = set()
for i in range(len(tweets)):
    # 跳过自己，只处理i<j的对避免重复
    for j in indices[i][1:]:
        if i >= j:
            continue
        if SequenceMatcher(None, tweets[i], tweets[j]).ratio() >= 0.75:
            duplicate_pairs.add((i,j))

如果一定要用字面相似度计算，difflib.SequenceMatcher是纯Python实现，速度较慢。可以换成rapidfuzz（C++实现，快几十倍）：

from rapidfuzz import fuzz

# 替换原来的ratio计算
similarity = fuzz.ratio(tweet1, tweet2) / 100  # 返回0-100的数值，转成0-1
if similarity >= 0.75:
    # 处理重复逻辑

内容的提问来源于stack exchange，提问作者Jack Alan