如何高效检测重复文件内容？优化余弦相似度查重代码问询

当前代码的核心问题是双重嵌套循环导致时间复杂度极高（O(n*m)，n为待处理文件数，m为已审核文件数），同时频繁的数据库save()操作也会拖慢整体性能。以下是针对性的优化方案：

1. 终止无效循环，减少冗余计算 #

一旦找到与待处理文件相似度超过阈值的已审核文件，立刻终止该文件的后续对比，避免无用计算；同时统一更新状态，减少数据库IO次数：

documents_processing = Document.objects.filter(status='processing')
documents_accepted = Document.objects.filter(status='accepted')

for processing in documents_processing:
    is_duplicate = False
    # 遍历已审核文件，找到相似项就终止循环
    for document in documents_accepted:
        cosine_sim = get_cosine_similary(document, processing)
        if cosine_sim > 0.9:
            is_duplicate = True
            break
    
    # 统一设置状态后再保存，避免多次IO
    processing.status = 'denied' if is_duplicate else 'accepted'
    processing.comment = 'duplicate' if is_duplicate else ''
    processing.save()

2. 批量计算+批量更新，降低数据库开销 #

预加载已审核文件的特征数据，完成所有相似度检查后批量更新数据库，彻底减少数据库交互次数：

# 预加载已审核文件的特征（假设数据库有存储特征向量的字段feature_vector）
accepted_docs = list(Document.objects.filter(status='accepted').values('feature_vector'))
accepted_vectors = [doc['feature_vector'] for doc in accepted_docs]

processing_docs = Document.objects.filter(status='processing')
update_batch = []

for doc in processing_docs:
    doc_vector = doc.feature_vector
    # 用any()快速判断是否存在相似项
    is_duplicate = any(get_cosine_similary(vec, doc_vector) > 0.9 for vec in accepted_vectors)
    
    doc.status = 'denied' if is_duplicate else 'accepted'
    doc.comment = 'duplicate' if is_duplicate else ''
    update_batch.append(doc)

# 批量更新数据库，仅一次IO操作
Document.objects.bulk_update(update_batch, ['status', 'comment'])

3. 引入近似最近邻(ANN)算法，彻底优化检索效率 #

当已审核文件数量达到数千级以上时，双重循环的复杂度会彻底拖垮系统。此时应该使用向量检索工具（如FAISS）将相似度检索的复杂度降到O(n log m)：

import faiss
import numpy as np

# 1. 构建已审核文件的向量索引（用内积直接对应余弦相似度）
accepted_vectors = np.array(
    [doc.feature_vector for doc in Document.objects.filter(status='accepted')]
).astype('float32')
index = faiss.IndexFlatIP(accepted_vectors.shape[1])  # 内积索引，值越大相似度越高
index.add(accepted_vectors)

# 2. 批量处理待审核文件
processing_docs = Document.objects.filter(status='processing')
update_batch = []

for doc in processing_docs:
    vec = np.array([doc.feature_vector]).astype('float32')
    # 搜索最相似的1个结果
    similarities, _ = index.search(vec, 1)
    max_similarity = similarities[0][0]
    
    if max_similarity > 0.9:
        doc.status = 'denied'
        doc.comment = 'duplicate'
    else:
        doc.status = 'accepted'
        doc.comment = ''
    update_batch.append(doc)

# 批量更新
Document.objects.bulk_update(update_batch, ['status', 'comment'])

4. 预处理特征向量，避免重复计算 #

如果get_cosine_similary每次都需要从原始文本重新提取特征（如TF-IDF、Word2Vec），会浪费大量计算资源。建议在文件上传时就预计算特征向量，存储到数据库的专用字段中，后续相似度计算直接调用预存向量即可。

5. 异步处理，不阻塞主流程 #

若文件审核无需实时完成，可将相似度检查逻辑放到异步任务队列（如Celery）中执行，避免阻塞用户上传或系统主流程：

# 定义异步任务
from celery import shared_task

@shared_task
def check_duplicates_task():
    documents_processing = Document.objects.filter(status='processing')
    # 此处放入上述优化后的批量检查逻辑
    # ...

# 文件上传完成后触发任务
document = Document.objects.create(..., status='processing')
check_duplicates_task.delay()

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