RAG 入库前怎样按内容指纹去除重复片段

RAG 知识库出现“同一段话被检索多次”，通常不是向量模型先出了问题，而是重复片段已经在入库前混在一起。更稳妥的做法是在切片完成、向量化之前增加一层内容指纹：先规范化文本并计算精确指纹，完全相同的片段只保留一份；疑似近重复的片段进入候选集合，由版本、来源和业务规则决定是否合并。

官方资料地址：https://huggingface.co/docs/transformers/main/en/tasks/retrieval

本文只讨论文档指纹、重复片段和入库边界。Hugging Face 的检索文档把文档内容、文档向量和索引看作检索资产，因此去重应发生在建立向量索引之前，而不是等检索结果重复后再补救。

本文要解决三件事：统一“相同”的判断口径；避免重复片段重复占用向量索引；保留来源和版本信息，让后续删除、更新和追溯仍然可做。

先把“重复”拆成三种情况

不要一上来就用一个相似度阈值删除数据。RAG 入库至少要区分三种关系：

• 精确重复：正文内容相同，只是换行、全角空格或 Unicode 表示不同。这类可以在规范化后用同一个指纹稳定拦截。
• 近重复：两段文本只有标题、日期、免责声明或少量句子不同。它们可能是同一版本，也可能是需要并存的历史版本，不能只靠哈希相等判断。
• 业务上不同：正文相似，但来源、版本、权限或生效日期不同。它们可能必须保留，否则问答会引用错误版本。

因此，指纹字段负责回答“规范化后的文本是否相同”，元数据负责回答“这份内容是否应该保留”。

规范化文本后再计算精确指纹

同一段文本可能混有不同换行符、连续空格、全角标点或组合字符。Python 的 unicodedata.normalize() 可以把 Unicode 文本转换成指定规范形式，hashlib.sha256() 则能对统一后的字节串计算稳定摘要。这里的关键是：规范化规则必须固定，并且指纹中要加入规则版本，避免以后改规则却无法解释旧数据。

import hashlib
import re
import unicodedata

def canonicalize(text: str) -> str:
    # NFKC 统一兼容字符；连续空白收敛成一个空格，避免换行差异制造重复。
    text = unicodedata.normalize("NFKC", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text

def content_fingerprint(text: str, rule_version: str = "v1") -> str:
    # 把规则版本放进摘要输入，后续升级规范化逻辑时仍能区分旧指纹。
    canonical = canonicalize(text)
    payload = f"{rule_version}\0{canonical}".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()

这段代码只负责“相同文本的稳定标识”，不负责判断文本是否适合合并。比如把所有标点都删除，可能会把代码、版本号和金额表达式改成危险的同值；生产规则应只收敛明确无业务意义的差异。

按精确指纹拦截重复片段，同时保留来源

精确去重时不要直接删除重复记录。更可追溯的模型是“一份 canonical_chunk 对应多个 source_metadata”：向量索引只绑定规范化后的主片段，重复来源保留文档 ID、页码、版本、更新时间和权限标签。

from dataclasses import dataclass, field

@dataclass
class ChunkRecord:
    # exact_fingerprint 是去重键，source_ids 用于保留所有来源映射。
    exact_fingerprint: str
    canonical_text: str
    source_ids: list[str] = field(default_factory=list)

def merge_chunk(index: dict[str, ChunkRecord], source_id: str, text: str) -> ChunkRecord:
    # 相同指纹只保留一个向量对象，但把新的来源追加到主记录。
    fingerprint = content_fingerprint(text)
    record = index.get(fingerprint)
    if record is None:
        record = ChunkRecord(fingerprint, canonicalize(text), [source_id])
        index[fingerprint] = record
    elif source_id not in record.source_ids:
        record.source_ids.append(source_id)
    return record

真正接向量库时，可以把 exact_fingerprint 作为幂等键，把 canonical_text 和嵌入向量绑定到同一条 canonical 记录。删除某个来源时，只移除对应来源映射；当没有来源仍指向这条记录，再由清理任务决定是否删除向量。

这一层还应保留 rule_version、切片策略版本和访问控制标签。否则文本虽然去重了，权限过滤或后续增量更新却无法准确工作。

近重复只做候选，不要直接删除

当两段文本只改了标题或一个日期，SHA-256 会完全不同。这时可以把文本切成固定长度的字符或词片段，计算轻量的 SimHash，或者先用 MinHash/Jaccard 做候选召回，再由业务规则确认。它们的共同点都是“缩小人工或二次模型比较范围”，不是自动证明两段内容可以合并。

def keep_or_merge(candidate: dict, existing: dict) -> str:
    # 相似度阈值只是候选信号，版本和权限差异优先于文本相似。
    if candidate["permission_scope"] != existing["permission_scope"]:
        return "keep"
    if candidate["version"] != existing["version"]:
        return "keep"
    if candidate["similarity"] >= 0.96:
        return "review"
    return "keep"

阈值不能脱离语料直接照搬。政策、接口文档和产品说明常常只有一个版本号不同，技术上很像，业务上却不能合并。建议把近重复记录写入候选表，保存比较方法、阈值、两条来源和处理结论，必要时让维护者确认。

入库前的最小检查清单

1. 规范化规则是否固定，并且带有可追踪的规则版本。
2. 精确指纹是否在向量化前计算，是否能作为幂等键重复提交。
3. 重复来源是否仍保留文档、页码、版本和权限信息。
4. 近重复是否进入候选区，而不是因为一个阈值被静默删除。
5. 更新和删除是否能从来源元数据反查到 canonical_chunk 与 vector_index。

如果只需要先解决“同一切片重复入库”，从规范化文本加 SHA-256 开始就够了；等精确重复稳定后，再根据语料规模和误删成本引入近重复候选。这样既能减少向量索引中的冗余，也不会用一套无法解释的相似度规则覆盖所有业务。

常见问题

为什么不直接对原文做 SHA-256？

原文中的换行、空白和 Unicode 表示差异会让视觉上相同的片段得到不同摘要。先固定、可解释地规范化，再对规范化结果计算摘要，才适合做精确去重。

重复片段是否应该合并所有来源？

可以合并向量对象，但不应丢掉来源映射。来源决定权限、版本和删除范围，canonical_chunk 只是检索资产，不是来源事实的替代品。

相似度达到 0.96 就能自动删除吗？

不能。相似度只适合发现候选；版本、权限、生效时间和来源差异都可能要求两段文本并存。