数据决定上限 在LLM微调领域有一句共识：模型架构决定下限，数据质量决定上限。 一个精心调优的7B模型配合高质量数据，可以超越用粗糙数据训练的70B模型。这不是夸张——实践已经反复证明这一点。
但"高质量数据"不是一个静态标签，而是一条工程管线的输出。本文将完整拆解从原始数据到训练就绪数据的全流程。
数据管线全景图 原始数据源 │ ├──→ 格式标准化 ├──→ 基础清洗（去噪、截断、编码修复） ├──→ 去重（精确去重 + 语义去重） ├──→ 质量评估（规则 + 模型打分） ├──→ 安全过滤（有害内容、PII脱敏） ├──→ 分布均衡（领域采样） └──→ 最终数据集 一、格式标准化 生产环境的数据来源五花八门：爬虫页面、API导出、用户日志、合成数据。第一步是统一到标准格式。
import json from dataclasses import dataclass from typing import List @dataclass class TrainingSample: """标准化训练样本结构""" instruction: str # 用户指令 input: str = "" # 附加输入（可选） output: str = "" # 期望输出 metadata: dict = None # 来源、质量分数等元信息 def normalize_to_alpaca(raw_samples: List[dict]) -> List[dict]: """将各种格式统一为Alpaca格式""" normalized = [] for sample in raw_samples: # 处理ShareGPT格式 if "conversations" in sample: convs = sample["conversations"] instruction = convs[0]["value"] if len(convs) > 0 else "" output = convs[1]["value"] if len(convs) > 1 else "" normalized.append({"instruction": instruction, "output": output}) # 处理Alpaca格式 elif "instruction" in sample: normalized.append(sample) # 处理OpenAI格式 elif "messages" in sample: msgs = sample["messages"] instruction = msgs[0]["content"] if msgs[0]["role"] == "user" else "" output = next((m["content"] for m in msgs if m["role"] == "assistant"), "") normalized.append({"instruction": instruction, "output": output}) return normalized 二、基础清洗 文本去噪 import re class TextCleaner: def __init__(self): self.patterns = { # HTML标签 "html": re.compile(r'<[^>]+>'), # 连续空白 "whitespace": re.compile(r'\s{3,}'), # URL（可选保留） "url": re.compile(r'https?://\S+'), # 乱码字符 "garbage": re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]'), # 重复标点（超过3个） "repeat_punct": re.compile(r'([！。，,.!?？]){4,}'), } def clean(self, text: str) -> str: text = self.patterns["garbage"].sub('', text) text = self.patterns["html"].sub('', text) text = self.patterns["whitespace"].sub('\n', text) text = self.patterns["repeat_punct"].sub(r'\1\1\1', text) return text.strip() def is_valid(self, text: str, min_len=10, max_len=4096) -> bool: """基础有效性检查""" if not text or len(text) < min_len or len(text) > max_len: return False # 数字与字母占比过高（可能是日志/代码误入） alpha_ratio = sum(c.isalpha() for c in text) / len(text) if alpha_ratio < 0.3: return False return True 编码修复 def fix_encoding(text: str) -> str: """修复常见的编码问题""" # 修复UTF-8被误解码为Latin-1的情况 try: return text.encode('latin-1').decode('utf-8') except (UnicodeDecodeError, UnicodeEncodeError): pass return text 三、去重：比你想的更重要 数据重复是微调的隐形杀手。重复数据会导致模型过拟合特定模式，降低泛化能力。去重分为两个层次：
...
