RAG文档去重，重复文档自动识别不重复入库
RAG知识库一旦混入重复文档，检索结果会出现大量冗余内容，还会白白消耗token和存储空间。
本文提供一套零基础也能照做的去重方案：先用内容哈希识别完全相同的文件，再用向量相似度拦截语义重复的改写内容，两个条件都通过才写入向量库，真正做到重复文档自动识别、不重复入库。
先分清两种“重复”
去重之前，要区分两类情况：
- 完全重复：同一个文件被上传多次，或者文件名不同但内容完全一致。这类问题用 MD5 或 SHA256 哈希即可解决。
- 语义重复：内容经过改写、换行、翻译、摘要加工，意思基本一样但字符不同。这类问题必须靠向量相似度判断。
正确的策略是先用哈希粗筛，再用向量细筛，兼顾速度和准确率。
准备去重环境
本文以 Python 为例，需要准备：
- Python 3.8 及以上环境
- 向量数据库客户端（以 pymilvus 为例）
- 嵌入模型库 sentence-transformers
安装依赖：
pip install pymilvus sentence-transformers hashlib嵌入模型建议先使用轻量级的 all-MiniLM-L6-v2，对中文也能处理，只是精度一般。
如果知识库以中文为主，可以换成 BAAI/bge-small-zh-v1.5：
pip install sentence-transformers同时准备好向量库中已有的文档 ID 和向量列表，这一步需要你提前建好 collection，并确保有 id 字段可以查重。
去重脚本：哈希判重 + 向量判重
下面是一段可执行的核心逻辑，重点在于“先查哈希表，再算相似度”。
import hashlib
import os
from sentence_transformers import SentenceTransformer
from pymilvus import Collection
# 加载模型，连接集合
model = SentenceTransformer('BAAI/bge-small-zh-v1.5')
col = Collection('your_kb') # 替换为实际集合名
col.load()
# 从向量库读取已有向量（假设最多10万条，按需调整）
existing_vectors = col.query(expr='id >= 0', output_fields=['id', 'text'], limit=100000)
existing_embeddings = [model.encode(item['text']) for item in existing_vectors]
# 哈希库：记录已经入库的文件哈希（可从单独表或文件读取）
hashed_records = set()
def text_hash(text: str) -> str:
return hashlib.md5(text.encode('utf-8')).hexdigest()
def is_duplicate(text: str, threshold: float = 0.92) -> bool:
# 第一步：哈希判断
h = text_hash(text)
if h in hashed_records:
return True
# 第二步：向量相似度判断
if existing_embeddings:
import numpy as np
vec = model.encode(text)
for existing in existing_embeddings:
sim = np.dot(vec, existing) / (np.linalg.norm(vec) * np.linalg.norm(existing))
if sim >= threshold:
return True
return False
def add_to_vector_db(file_path):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
if is_duplicate(content):
print(f'跳过重复文档: {file_path}')
return
# 向量化并写入向量库，同时记录哈希
vec = model.encode(content)
col.insert([[max_id + 1], [content], [vec]]) # 实际插入需要按你的 schema 调整
hashed_records.add(text_hash(content))
print(f'入库成功: {file_path}')注意：max_id 需要从集合中获取当前最大 id，插入字段顺序要和 schema 一致。
这段代码只是演示，生产环境建议把哈希记录存成 SQLite 或单独表，避免每次重启都重新加载。
增量导入和更新场景的处理
文档不会一成不变，增量导入需要单独处理：
- 新文件：先算哈希，若在哈希记录中则跳过；否则计算向量相似度，再决定是否入库。
- 更新文件：内容变化后哈希会变，此时不能简单跳过。建议在记录哈希时同时保存文件路径或版本号，当检测到同一路径的哈希变化时，先删除旧向量，再重新插入新内容。
- 目录扫描：用
os.walk递归遍历文件夹，对每个文档文件执行上述逻辑，即可实现批量去重入库。
for root, dirs, files in os.walk('./docs'):
for name in files:
if name.endswith('.md') or name.endswith('.txt'):
add_to_vector_db(os.path.join(root, name))避坑指南和高频疑问
阈值设多少合适？ 阈值太高（如 0.99）会漏掉改写内容，太低（如 0.8）会把相似但不相同的文档误判为重复。
建议先用 0.92 起步，再用一批人工标注的数据测试，观察误判率。
PDF 和 Word 里的内容怎么去重？ 先去提取文本，再使用本文的哈希和向量逻辑。
特别注意：同一份文档导出的 PDF 和 TXT，哈希结果不一样，但向量相似度通常很高，所以两步判断缺一不可。
向量库已有大量数据，首次去重要全量比对吗？ 如果已有向量超过几百万条，逐条计算相似度会非常慢。
建议先用哈希筛掉完全重复项，再对剩余文档用向量索引的 search 接口查 top1 相似度，而不是用 Python 循环比对。
哈希表存哪里？ 不要只放在内存里，重启会丢失。
推荐写入 SQLite，字段包含文件路径、内容哈希、入库时间，这样既支持增量去重，也能做溯源。
验证去重生效
运行脚本后，可以从三个角度确认效果：
- 看日志：确认有“跳过重复文档”输出，且数量符合预期。
- 查向量库总量：多次导入同一批文件，集合中的 document 数量不增长。
- 抽查相似对：随机找一篇替换几个同义词的文档，确认它能被拦截；再找一篇主题相近但内容不同的文档，确认它不会被误杀。
如果验证中发现误判，优先调整相似度阈值或更换更适配中文的嵌入模型。
完成上述步骤后，你的 RAG 导入流程就具备了自动去重能力。
以后只要把新文档放进扫描目录，重复内容会被自动识别并跳过，向量库会保持干净、紧凑，检索质量也会更稳定。
建议先把本文的逻辑改造成脚本，再逐步接入定时任务或文件上传接口。
