数据：大模型能力的源头
“Garbage in, garbage out"在LLM时代被放大了1000倍。一个7B模型用高质量数据训练可以超越用低质量数据训练的70B模型。数据治理是决定模型能力上限的第一道关卡。
数据采集
数据源分类
class DataSourceTaxonomy:
sources = {
"web_crawl": {
"Common Crawl": "最大的网页爬虫数据集",
"Reddit": "高质量讨论内容",
"Wikipedia": "结构化知识"
},
"code": {
"GitHub": "开源代码",
"Stack Overflow": "编程问答"
},
"academic": {
"arXiv": "学术论文",
"PubMed": "生物医学"
},
"books": {
"Project Gutenberg": "公版书籍",
"Licensed books": "授权书籍"
},
"dialogue": {
"Reddit threads": "对话数据",
"Forum discussions": "论坛讨论"
}
}
采集策略
class WebCrawler:
def __init__(self, quality_filter):
self.filter = quality_filter
def crawl(self, url):
# 1. 抓取页面
html = self._fetch(url)
# 2. 正文提取（去除导航、广告等）
content = self._extract_main_content(html)
# 3. 质量初筛
quality_score = self.filter.assess(content)
if quality_score < 0.3:
return None # 质量太低，跳过
# 4. 语言检测
lang = detect_language(content)
# 5. 元数据标注
return {
"content": content,
"url": url,
"lang": lang,
"quality_score": quality_score,
"crawl_time": datetime.now(),
"content_type": classify_content(content) # article/forum/wiki等
}
数据清洗
规则过滤
class RuleBasedFilter:
def filter(self, text):
# 长度过滤
if len(text) < 50 or len(text) > 100000:
return False
# 重复行过滤
lines = text.split('\n')
unique_ratio = len(set(lines)) / len(lines)
if unique_ratio < 0.5:
return False
# 特殊字符比例
special_ratio = sum(1 for c in text if not c.isalnum() and c not in ' \n.,!?;:\'"-()') / len(text)
if special_ratio > 0.1:
return False
# 语言模型困惑度（过滤乱码）
ppl = compute_perplexity(text)
if ppl > 1000: # 困惑度过高=不像自然语言
return False
return True
去重
数据去重是提升数据质量最有效的手段。研究表明，去重可以将模型性能提升3-5个点。
精确去重：
def exact_dedup(documents):
seen_hashes = set()
unique = []
for doc in documents:
h = hash(doc.content)
if h not in seen_hashes:
seen_hashes.add(h)
unique.append(doc)
return unique
模糊去重（MinHash）：
from datasketch import MinHash, MinHashLSH
def fuzzy_dedup(documents, threshold=0.8, num_perm=128):
lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
unique_docs = []
for i, doc in enumerate(documents):
# 创建MinHash
mh = MinHash(num_perm=num_perm)
for word in doc.content.split():
mh.update(word.encode('utf-8'))
# 检查相似文档
similar = lsh.query(mh)
if not similar:
lsh.insert(i, mh)
unique_docs.append(doc)
return unique_docs
安全过滤
class SafetyFilter:
def __init__(self):
self.pii_detector = PIIDetector()
self.toxic_classifier = ToxicClassifier()
self.adult_classifier = AdultContentClassifier()
def filter(self, text):
# PII脱敏
text = self.pii_detector.redact(text)
# 邮箱、手机号、身份证号等替换为占位符
# 有害内容过滤
if self.toxic_classifier.is_toxic(text):
return None
# 成人内容过滤
if self.adult_classifier.is_adult(text):
return None
return text
数据质量评估
多维度质量评分
class DataQualityAssessor:
def __init__(self):
self.metrics = {
"fluency": FluencyMetric(), # 流畅度
"coherence": CoherenceMetric(), # 连贯性
"informativeness": InfoMetric(), # 信息量
"accuracy": AccuracyMetric(), # 准确性
"diversity": DiversityMetric(), # 多样性
}
def assess(self, document):
scores = {}
for name, metric in self.metrics.items():
scores[name] = metric.score(document)
# 加权综合分
weights = {
"fluency": 0.25,
"coherence": 0.20,
"informativeness": 0.25,
"accuracy": 0.20,
"diversity": 0.10
}
overall = sum(scores[k] * weights[k] for k in scores)
return {"overall": overall, "details": scores}
基于模型的质量评估
用强模型评估训练数据质量：
def llm_quality_score(text, judge_model="gpt-4o"):
prompt = f"""评估以下文本作为大模型训练数据的质量。
文本：{text[:1000]}
评分维度（1-5分）：
1. 信息密度：是否包含有价值的信息
2. 表达质量：语言是否规范、清晰
3. 事实准确性：内容是否准确
4. 教育价值：对学习是否有帮助
输出JSON格式。
"""
result = llm.generate(prompt, model=judge_model)
return parse_json(result)
数据配比优化
预训练数据配比
不同类型数据的配比显著影响模型能力：
class DataMixer:
def __init__(self):
self.default_ratio = {
"web": 0.50, # 网页数据
"code": 0.15, # 代码
"academic": 0.10, # 学术论文
"books": 0.10, # 书籍
"dialogue": 0.10, # 对话
"math": 0.05, # 数学
}
def optimize_ratio(self, target_capabilities):
"""根据目标能力优化数据配比"""
if "code" in target_capabilities:
self.default_ratio["code"] = 0.25
self.default_ratio["web"] = 0.40
if "math" in target_capabilities:
self.default_ratio["math"] = 0.10
self.default_ratio["web"] = 0.40
if "chinese" in target_capabilities:
# 增加中文数据比例
self._adjust_language_ratio(chinese=0.40, english=0.55, other=0.05)
return self.default_ratio
动态配比调度
训练不同阶段使用不同配比：
class CurriculumScheduler:
def get_ratio(self, training_step, total_steps):
progress = training_step / total_steps
if progress < 0.3:
# 早期：高质量数据为主
return {"high_quality": 0.7, "web": 0.3}
elif progress < 0.7:
# 中期：均衡配比
return {"high_quality": 0.4, "web": 0.4, "code": 0.2}
else:
# 后期：增加困难样本
return {"high_quality": 0.3, "web": 0.3, "code": 0.2, "hard": 0.2}
数据版本管理
class DataVersionControl:
def __init__(self, storage):
self.storage = storage
def create_version(self, dataset, version_tag):
# 计算数据集指纹
fingerprint = self._compute_fingerprint(dataset)
# 存储版本信息
version_info = {
"tag": version_tag,
"fingerprint": fingerprint,
"size": len(dataset),
"created_at": datetime.now(),
"stats": self._compute_stats(dataset),
"manifest": self._create_manifest(dataset)
}
self.storage.save(version_info)
return version_info
def diff_versions(self, v1_tag, v2_tag):
"""比较两个版本的数据差异"""
v1 = self.storage.load(v1_tag)
v2 = self.storage.load(v2_tag)
return {
"added": set(v2.fingerprints) - set(v1.fingerprints),
"removed": set(v1.fingerprints) - set(v2.fingerprints),
"stats_change": diff_stats(v1.stats, v2.stats)
}
实践建议
数据质量>数据量
实验表明，用10T高质量数据训练的模型效果优于用15T混合质量数据训练的模型。在数据量达到一定规模后（约10T tokens），质量提升的收益远大于数量增加。
持续数据迭代
class DataIterationPipeline:
def iterate(self, model, eval_results):
# 1. 分析模型弱点
weaknesses = analyze_weaknesses(eval_results)
# 例：数学能力弱 → 需要更多数学数据
# 2. 针对性收集数据
new_data = collect_targeted_data(weaknesses)
# 3. 质量过滤
filtered = quality_filter(new_data)
# 4. 去重
deduped = dedup(filtered, existing_data)
# 5. 配比调整
new_ratio = adjust_ratio(weaknesses)
# 6. 继续训练
return continue_training(model, deduped, new_ratio)
结语
数据是模型的"食物”，数据治理就是"营养管理"。在大模型能力趋同的今天，数据质量正在成为真正的差异化优势。那些建立了完善数据治理体系、持续优化数据质量的团队，将在模型能力的长期竞争中占据优势。记住：没有最好的模型架构，只有最好的训练数据。