大模型预训练数据工程：Common Crawl 海量清洗、MinHash LSH 模糊去重与合成数据 Pipeline 实战

大模型预训练数据工程：Common Crawl 海量清洗、MinHash LSH 模糊去重与合成数据 Pipeline 实战

在大语言模型（LLM）与多模态大模型的技术演进中，业内达成了一个不可动摇的共识——“Data is the New Oil，数据质量决定模型能力的绝对上限（Garbage In, Garbage Out）”。

Meta 的 Llama-3、HuggingFace 的 FineWeb 以及深度求索（DeepSeek）的预训练成果表明：模型在数学推理、代码生成与复杂逻辑上的突破，70% 以上归功于极其严苛、精细化的大规模预训练数据工程流水线。

然而，直接从公开互联网抓取的原始数据（如Common Crawl PB 级网页抓取池）中充斥着毁灭性的**“数据垃圾与毒性噪声”**：

• 超过75% ~ 85%的网页是 SEO 垃圾站、机器翻译乱码、博彩引流与模板化广告；
• 海量高度相似或完全重复的文本（如版权声明、导航栏模板、转载新闻）如果不经清洗直接喂给模型，会导致大模型在训练时发生严重的**“记忆过拟合（Memorization Overfitting）”**并在生成时频繁陷入重复句式的死循环；
• 敏感个人身份信息（PII: 身份证、手机号、密码）与低俗有害内容若混入语料库，会直接触发安全合规红线。

如何从数十 TB 级的原始语料中淘洗出千亿级别的黄金 Token？

本文深入剖析现代大模型预训练数据清洗五阶段架构、MinHash + LSH（局部敏感哈希）海量模糊去重算法，并给出生产级 Python 预训练数据清洗与去重流水线实战代码。

一、大模型预训练数据清洗流水线五阶段全景对比矩阵

二、MinHash + LSH（局部敏感哈希）大规模模糊去重底层数学机理

对于拥有数亿篇文档的数据集，两两计算 Jaccard 相似度需要 $O(N^2)$ 的天文级算力，这在物理上是完全不可行的。

MinHash + LSH 算法将计算复杂度从 $O(N^2)$ 奇迹般压缩至 $O(N)$：

1. K-Shingle 集合化：将文本切分为连续的 $k$ 元字符组（如 $k=5$ 的 5-gram 集合）；
2. MinHash 降维签名：使用 $M$ 个独立的哈希函数 $h_1, h_2, \dots, h_M$，取集合中哈希值最小的元素构成一个固定长度为 $M$（如 128 维）的 MinHash 签名向量；
   $$\Pr\left(\min(h_i(A)) = \min(h_i(B))\right) = \text{Jaccard}(A, B)$$
3. LSH（Locality-Sensitive Hashing）分带哈希分桶：
   将 128 维签名切分为 $b$ 个带（Bands），每个带包含 $r$ 个哈希值（$b \times r = M$）。只有当两个文档在至少一个 Band 中具有完全相同的哈希签名时，它们才会被放入同一个桶中比对，彻底免去了 99.99% 的无效全局两两计算！

[文档 A 与文档 B 生成 128 维 MinHash 签名] +-------------------------------------------------------------------------------+ | Band 1 (r=4): [h1, h2, h3, h4] ====> Hash(Band 1) -> 写入 Bucket #1024 | | Band 2 (r=4): [h5, h6, h7, h8] ====> Hash(Band 2) -> 写入 Bucket #8899 | | ... | | Band b (r=4): [..., h128] ====> Hash(Band b) -> 写入 Bucket #5501 | +-------------------------------------------------------------------------------+ | v [只有落入同一个 Bucket 的文档才判定为疑似重复候选 -> 执行最终 Jaccard 校验去重!]

三、生产级 Python MinHash + LSH 大规模预训练去重流水线实现

下面的 Python 实现结合了启发式低质规则清洗、敏感 PII 自动脱敏、K-Shingle 切词、MinHash 签名生成与基于 LSH 桶聚合的高并发模糊去重引擎。

""" pretrain_data_pipeline.py 生产级大模型预训练数据清洗与 MinHash LSH 模糊去重流水线实战 """ import hashlib import re from dataclasses import dataclass from typing import Dict, List, Set, Tuple @dataclass class CleanedDocument: doc_id: str text: str is_valid: bool filter_reason: str = "" class HeuristicTextCleaner: """阶段 1 & 2: 启发式质量规则过滤与 PII 隐私脱敏""" @staticmethod def clean_and_validate(doc_id: str, raw_text: str) -> CleanedDocument: # 1. 过滤 HTML/JS 常见残留 text = re.sub(r"<script.*?>.*?</script>", "", raw_text, flags=re.DOTALL | re.IGNORECASE) text = re.sub(r"<style.*?>.*?</style>", "", text, flags=re.DOTALL | re.IGNORECASE) text = re.sub(r"<.*?>", "", text) text = text.strip() # 2. 启发式规则 A: 过滤过短文本 (少于 50 个字符) if len(text) < 50: return CleanedDocument(doc_id, text, False, "TEXT_TOO_SHORT") # 3. 启发式规则 B: 标点符号与特殊字符比例异常过滤 (乱码检测) special_char_count = len(re.findall(r"[\#\$\%\^\&\*\@\_\+\=\|\~\`]", text)) if special_char_count / max(1, len(text)) > 0.20: return CleanedDocument(doc_id, text, False, "TOO_MANY_SPECIAL_CHARS") # 4. 敏感 PII 隐私脱敏 (手机号与邮箱脱敏) text = re.sub(r"(1[3-9]\d)(\d{4})(\d{4})", r"\1****\3", text) text = re.sub(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+", "[EMAIL_MASK]", text) return CleanedDocument(doc_id, text, True, "PASSED") class MinHashLSHDeduplicator: """阶段 3: 基于 MinHash 与 LSH 的大规模模糊去重器""" def __init__(self, num_perm: int = 64, num_bands: int = 16, threshold: float = 0.8): self.num_perm = num_perm self.num_bands = num_bands self.rows_per_band = num_perm // num_bands self.threshold = threshold # LSH 桶哈希表: {band_idx: {bucket_hash: [doc_ids]}} self.buckets: Dict[int, Dict[str, List[str]]] = {i: {} for i in range(num_bands)} self.seen_doc_signatures: Dict[str, List[int]] = {} def _get_shingles(self, text: str, k: int = 5) -> Set[str]: """提取 5-gram 连续字符 Shingles""" cleaned = re.sub(r"\s+", " ", text).lower() if len(cleaned) < k: return {cleaned} return {cleaned[i:i+k] for i in range(len(cleaned) - k + 1)} def _compute_minhash_signature(self, shingles: Set[str]) -> List[int]: """计算固定维度的 MinHash 签名向量""" signature = [] for seed in range(self.num_perm): min_val = float('inf') for shingle in shingles: # 简单模拟不同的哈希函数 h = int(hashlib.md5(f"{seed}_{shingle}".encode()).hexdigest(), 16) if h < min_val: min_val = h signature.append(min_val) return signature def is_duplicate_and_insert(self, doc_id: str, text: str) -> bool: """检查文档是否与已有语料库发生近重复 (Near-Duplicate)，若不重复则入库""" shingles = self._get_shingles(text, k=5) signature = self._compute_minhash_signature(shingles) candidate_duplicates: Set[str] = set() # 1. 遍历每个 Band，在 LSH 桶中快速查找候选碰撞 for b in range(self.num_bands): start = b * self.rows_per_band end = start + self.rows_per_band band_tuple = tuple(signature[start:end]) bucket_key = hashlib.md5(str(band_tuple).encode()).hexdigest() if bucket_key in self.buckets[b]: for candidate_id in self.buckets[b][bucket_key]: candidate_duplicates.add(candidate_id) # 2. 对候选碰撞计算精确的 MinHash Jaccard 相似度 for cand_id in candidate_duplicates: cand_sig = self.seen_doc_signatures[cand_id] # 计算签名中相同哈希值的比例 matches = sum(1 for i in range(self.num_perm) if signature[i] == cand_sig[i]) similarity = matches / float(self.num_perm) if similarity >= self.threshold: # 判定为重复文档！ return True # 3. 确认为全新优质文档，注册进 LSH 桶 self.seen_doc_signatures[doc_id] = signature for b in range(self.num_bands): start = b * self.rows_per_band end = start + self.rows_per_band band_tuple = tuple(signature[start:end]) bucket_key = hashlib.md5(str(band_tuple).encode()).hexdigest() self.buckets[b].setdefault(bucket_key, []).append(doc_id) return False

生产演练与海量语料清洗去重看板展示

if __name__ == "__main__": print("=== 🚀 大模型预训练数据工程清洗与 MinHash LSH 去重演练 ===") cleaner = HeuristicTextCleaner() dedup = MinHashLSHDeduplicator(num_perm=32, num_bands=8, threshold=0.75) # 模拟从网页抓取的 4 篇原始语料 corpus = [ # 文档 1: 正常高质量长文本 ("DOC_001", "Apache Flink 是一个分布式流处理引擎，具备极高的吞吐与亚秒级低延迟特性，广泛应用于各大互联网企业的实时风控与数仓治理。"), # 文档 2: 垃圾乱码文本 ("DOC_002", "###$$$%%% 点击这里免费领红包 ￥￥￥ 广告链接广告链接广告链接"), # 文档 3: 🌟 针对文档 1 的轻微改动转载副本 (包含相同的核心段落，属于典型近重复) ("DOC_003", "Apache Flink 是一个分布式流处理引擎，具备极高的吞吐与亚秒级低延迟特性，广泛应用于各大互联网公司的实时风控与数仓建设。"), # 文档 4: 另一篇独立的高质量技术文本 ("DOC_004", "ClickHouse 采用现代列式存储与 SIMD 向量化执行引擎，在百亿级数据的多维聚合分析中展现出极致的查询性能。") ] retained_docs = [] for doc_id, raw_content in corpus: # 第一步：启发式清洗与隐私脱敏 clean_res = cleaner.clean_and_validate(doc_id, raw_content) if not clean_res.is_valid: print(f"❌ [{doc_id}] 被启发式规则拦截过滤: 原因={clean_res.filter_reason}") continue # 第二步：MinHash LSH 模糊去重 is_dup = dedup.is_duplicate_and_insert(doc_id, clean_res.text) if is_dup: print(f"⚠️ [{doc_id}] 命中 MinHash LSH 近重复语料 (Jaccard > 0.75)，已被安全剔除！") else: print(f"✅ [{doc_id}] 成功通过全部清洗与去重质检，已存入预训练黄金 Token 语料库！") retained_docs.append(clean_res) print("\n=======================================================") print("📊 🌟 预训练数据清洗流水线收割报告 🌟") print("=======================================================") print(f"【输入原始样本量】: {len(corpus)} 篇") print(f"【最终留存合格样本】: {len(retained_docs)} 篇 (清洗过滤损耗率: {(1-len(retained_docs)/len(corpus))*100:.1f}%)")

四、生产避坑与预训练数据工程落地红线

在构建 TB / PB 级大模型预训练数据中台时，必须坚守以下四项落地原则：

1. 严禁在分布式分片之间做全局内存排序去重：
   在 Spark / Ray 集群上处理数十亿篇文档时，利用 LSH 的BucketID将潜在重复候选分发到同一个 Partition 中进行局部去重，彻底消灭全表 Shuffle 带来的网络带宽耗尽。
2. 结合 FastText 训练领域分类器（Domain Classifier）过滤机器乱码：
   使用 Wikipedia、ArXiv 与高质量开源教科书训练一个二分类 FastText 模型，对每篇网页预测其知识密度置信度，直接斩断 80% 的低质垃圾站内容。
3. 保留原始文档元数据（URL、Domain、Timestamp）：
   清洗过程中必须保留来源 Domain。在最终组装 Token 训练配比时，实施按顶级域名抽样配额（Domain Upsampling/Downsampling），避免某个高频爬虫站点占据过大比例导致语料失衡。

通过将启发式规则过滤、PII 隐私动态脱敏与 MinHash LSH 高性能模糊去重有机结合，AI 基础架构团队能够从互联网海量泥沙中淘洗出知识密度极高、分布均衡纯净的黄金语料，为大模型注入真正坚韧、渊博的底层认知智能。