引言

在自然语言处理（NLP）任务中，数据质量是模型性能的关键因素之一。重复或冗余的数据会导致模型过度拟合或浪费计算资源，特别是在大语言模型（如 BERT、GPT 系列等）训练和推理阶段。传统的基于字符匹配的去重方法（如字符串哈希或编辑距离）在面对语义相似的文本时表现有限，而语义相似度算法则能更好地捕获文本之间的深层语义关系。

本文将介绍一种基于语义表示的去重方法，通过大语言模型生成的嵌入向量结合高效的相似度计算工具（如 FAISS），对大规模文本数据进行去重。此方法不仅适用于数据清洗，还可以应用在搜索引擎、推荐系统等需要衡量语义相似度的场景。

原理与方法

1. 传统去重方法的局限性

在 NLP 任务中，传统的去重方法包括：

• 字符串哈希：
  基于文本的哈希值进行判重，适合完全重复的文本，但无法处理语义相似但表达不同的情况，例如：
  • 文本 A：我喜欢吃苹果。
  • 文本 B：苹果是我最喜欢的水果。

虽然两者语义相近，但哈希值完全不同。

• 编辑距离（Levenshtein Distance）：
  衡量两个字符串的编辑代价，适合处理少量字符差异的文本，但无法捕捉深层语义关系。

上述方法对文本的语义相似性缺乏鲁棒性，特别是在短文本或同义表达常见的场景下。例如，问答生成、文档去重、语料清洗等任务中，语义相似的重复数据可能会严重影响模型性能。

2. 基于语义嵌入的去重

语义嵌入（Semantic Embedding）是一种将文本映射到高维向量空间的技术，向量的物理距离或角度可以反映文本语义的相似程度。常见的嵌入生成模型包括：

• BERT、RoBERTa、GPT 等大语言模型：能够生成上下文相关的语义表示。
• Sentence-BERT（SBERT）：专为语义相似度任务设计，提升了嵌入的语义表达能力。

基本流程：

• 1. 文本嵌入生成：
     使用大语言模型将文本转化为固定维度的向量表示（如 768 维）。
• 2. 相似度计算：
     通过数学距离（如余弦相似度或内积）衡量文本向量之间的相似性。
• 3. 去重判断：
     基于相似度阈值判断文本是否为重复内容。

3. 相似度计算方法对比

在语义嵌入的基础上，常用的相似度计算方法包括：

3.1. 余弦相似度（Cosine Similarity）

余弦相似度衡量两个向量的夹角余弦值，范围为 $[-1,1]$，归一化后范围为 $[0,1]$。公式如下：
$$\text{Cosine Similarity}(A,B)=\frac{A\cdot B}{\Vert A\Vert \Vert B\Vert }$$

• 优点：消除向量模长的影响，只关注向量方向。
• 缺点：计算开销稍高。

3.2. 内积相似度（Inner Product Similarity）

内积相似度直接计算两向量的点积值：
$$\text{Inner Product}(A,B)=A\cdot B$$

• 优点：计算简单，速度快。
• 缺点：受向量模长影响，需要确保输入向量已归一化（模长为 1），否则结果不等价于余弦相似度。

欧几里得距离（Euclidean Distance）

衡量两个向量在高维空间中的直线距离：
$$\text{Euclidean Distance}(A,B)=\sqrt{\sum _{i=1}^n(A_i-B_i{)}^2}$$

• 优点：适合绝对位置相关的任务。
• 缺点：不适合捕获方向性的语义相似度。

4. 高效的大规模相似度计算

直接比较所有嵌入向量的相似度在大规模数据中效率低下（复杂度为 $O(n^2)$）。为此，我们借助 FAISS（Facebook AI Similarity Search）工具，能够在百万级甚至亿级数据中高效实现近似最近邻搜索。

4.1. FAISS 简介

FAISS 是一个高效的相似度搜索库，专为高维向量的最近邻搜索设计，支持以下特性：

• 多种索引结构：
  • Flat：暴力搜索，适合中小规模数据。
  • IVF（倒排文件索引）：适合大规模数据。
  • PQ（分组量化）：进一步压缩内存占用。
• GPU 加速：支持 GPU 版本，在大规模数据上极大提升搜索速度。
• 灵活的距离度量：支持内积、余弦、欧几里得距离等。

4.2. 使用 FAISS 的语义去重流程

1. 初始化 FAISS 索引：选择适合任务的数据结构（如 IndexFlatIP）。
2. 添加向量：将嵌入向量添加到索引。
3. 查询相似度：对每个新向量，查找与索引中最近的向量，判断是否重复。

代码实现

import json
from transformers import BertTokenizer, BertModel
import torch
from tqdm import tqdm
import faiss
from typing import List, Dict, Unionclass TextDeduplicatorWithFAISS:"""使用 FAISS 索引实现的文本去重类（基于余弦相似度）。"""def __init__(self, model_name: str = 'bert-base-chinese', device: str = None) -> None:"""初始化文本去重类。参数:- model_name: 使用的预训练模型名称，默认为 'bert-base-chinese'。- device: 指定运行设备（'cpu' 或 'cuda'），默认为自动检测。"""self.tokenizer = BertTokenizer.from_pretrained(model_name)self.model = BertModel.from_pretrained(model_name)self.device = device if device else ('cuda' if torch.cuda.is_available() else 'cpu')self.model = self.model.to(self.device)# 初始化 FAISS 索引self.embedding_dim = 768  # BERT 输出嵌入维度self.index = faiss.IndexFlatIP(self.embedding_dim)  # 使用内积（IP）作为相似度度量self.index_ids = []  # 存储对应嵌入的 ID，方便后续处理def get_embeddings(self, texts: List[str]) -> torch.Tensor:"""计算文本的嵌入表示，并进行归一化。参数:- texts: 要计算嵌入的一组文本列表。返回:- 归一化后的文本嵌入张量，形状为 (batch_size, hidden_size)。"""inputs = self.tokenizer(texts, return_tensors="pt", padding=True, truncation=True, max_length=512)inputs = inputs.to(self.device)  # 将输入张量移动到指定设备with torch.no_grad():  # 禁用梯度计算以节省内存outputs = self.model(**inputs)embeddings = outputs.last_hidden_state[:, 0, :].cpu()  # 获取 [CLS] 的嵌入并移动到 CPU# 对嵌入进行归一化处理（实现余弦相似度）embeddings = embeddings / torch.norm(embeddings, dim=1, keepdim=True)return embeddingsdef is_duplicate(self, embedding: torch.Tensor, threshold: float = 0.9) -> bool:"""检查一个嵌入是否与 FAISS 索引中的嵌入重复。参数:- embedding: 待检查的嵌入向量，形状为 (1, hidden_size)。- threshold: 相似度的阈值，默认为 0.9。返回:- 是否为重复项（True / False）。"""if self.index.ntotal == 0:  # 如果索引为空，肯定不是重复return False# 通过 FAISS 查找最近的向量及其相似度embedding_np = embedding.numpy()  # 转为 NumPy 格式distances, _ = self.index.search(embedding_np, k=1)  # 查找最近的 1 个向量# 检查最近向量的相似度是否高于阈值max_similarity = distances[0][0]  # FAISS 返回的是归一化向量的内积（等价于余弦相似度）return max_similarity >= thresholddef add_to_index(self, embedding: torch.Tensor, doc_id: int) -> None:"""将新的嵌入添加到 FAISS 索引中。参数:- embedding: 要添加的嵌入向量，形状为 (1, hidden_size)。- doc_id: 该嵌入对应的文档 ID。"""embedding_np = embedding.numpy()  # 转为 NumPy 格式self.index.add(embedding_np)  # 添加到索引中self.index_ids.append(doc_id)  # 保存对应的文档 IDdef process_and_save(self, input_path: str, output_path: str, threshold: float = 0.9) -> None:"""处理输入文件，去除相似文本并保存到输出文件。参数:- input_path: 输入 JSONL 文件路径。- output_path: 输出 JSONL 文件路径。- threshold: 去重的相似度阈值，默认值为 0.9。"""doc_id = 0  # 用于标记每条文档的唯一 IDwith open(input_path, 'r', encoding='utf-8') as infile, open(output_path, 'w', encoding='utf-8') as outfile:for line in tqdm(infile, desc="Processing lines"):item: Dict[str, Union[str, int, float]] = json.loads(line)  # 从 JSONL 文件中读取一条数据output_text: str = item['output']  # 获取文本内容# 获取当前文本的嵌入current_embedding = self.get_embeddings([output_text])# 检查是否为重复if not self.is_duplicate(current_embedding, threshold):# 如果不重复，保存文本，并将嵌入添加到索引outfile.write(json.dumps(item, ensure_ascii=False) + '\n')self.add_to_index(current_embedding, doc_id)doc_id += 1# 使用示例
if __name__ == "__main__":# 初始化去重器deduplicator = TextDeduplicatorWithFAISS(model_name='bert-base-chinese')# 去重并保存结果deduplicator.process_and_save(input_path='=./processed_unique_data-5.jsonl',output_path='=./processed_unique_data-6.jsonl',threshold=0.95)

数据示例：

{"id": 1, "output": "什么是人工智能？人工智能是指让机器具备人类智能的技术。"}
{"id": 2, "output": "人工智能的定义是什么？人工智能是赋予机器类似人类智能的能力。"}

总结

本文介绍了一种基于语义嵌入的大规模文本去重方法，通过结合大语言模型（如 BERT）和高效相似度搜索工具（FAISS），实现了对语料库的语义级去重。该方法具有以下优点：

• 高精度：捕捉语义相似性，避免遗漏同义表达的重复数据。
• 高扩展性：支持大规模数据处理，适用于百万级文本的去重任务。
• 通用性强：不仅适用于去重，还可扩展至相似文本检索、推荐系统等任务。