SFT 数据清洗实战：从原始语料到高质量训练集一、一个真实踩坑故事去年我在做一个内部客服机器人的 SFT（Supervised Fine-Tuning）时，做了一件教科书式的蠢事：花了两周用 DeepSeek 生成了 5000 条对话数据，没做任何清洗，直接扔进 Llama-3-8B 微调。训练 loss 降得很漂亮，eval loss 也正常。但一上线，模型出现了三个诡异行为：答非所问：用户问"怎么退款"，模型回复"您好，感谢您的咨询，请稍等……"然后开始介绍公司历史幻觉满天飞：虚构了一个不存在的"退款VIP通道"，煞有介事地编造了操作步骤数值崩溃：loss 在 epoch 3 突然飙升，从 0.8 跳到 12.3排查一圈，罪魁祸首是数据质量问题：大量对话里 Assistant 重复输出相同模板话术、格式不一致（有的有### Assistant:前缀，有的没有）、prompt 和 response 之间逻辑断裂。这些问题在单独抽查时根本看不出来，但放一起训练就炸了。这让我深刻意识到：SFT 的核心不是模型架构，不是超参调优，而是数据质量。数据决定了模型的上限，而清洗决定了你能触达这个上限的几分。下面我会从实战角度，系统梳理 SFT 数据清洗的完整流程，附带可运行的 Python 代码。为什么 SFT 比预训练更"挑"数据？很多人不理解：预训练数据那么脏（网页抓取、PDF 提取、论坛帖子），模型不也好好的？为什么 SFT 阶段对数据质量如此敏感？答案在于训练方式的不同。预训练是无监督学习，模型看的是海量文本的统计规律——偶尔的噪声会被巨量数据平均掉，就像大海里的几滴墨水，根本看不出来。但 SFT 是有监督的 instruction tuning，每条数据都在直接教模型"遇到 X 该输出 Y"。一条错误数据相当于老师说了一句错误答案，学生（模型）会牢牢记在心里。用一个比喻：预训练是让模型读了整个图书馆的书——个别错别字不影响理解。SFT 则是让模型参加一对一辅导——老师说错的每一句话，学生都当真理吸收了。数据量越小、数据越精，每一条的权重就越大，质量要求就越高。这就是为什么 5000 条高质量 SFT 数据能胜过 50000 条中低质量数据的根本原因。二、SFT 数据质量问题全景图在动手清洗之前，我们需要知道自己面对的是什么。我把常见的 SFT 数据问题分为四层：层级问题类型典型表现影响程度L1 格式层编码错误、特殊字符乱码、截断\x00空字节、UTF-8 BOM、不配对的引号训练中断或 tokenization 异常L2 结构层缺失字段、对话轮次不完整assistant 回复为空、只有 user 没有 assistantLoss 计算错误，模型学到"沉默"L3 语义层逻辑断裂、答非所问、内容重复第3轮的回复在第7轮又出现一遍模型学会"复读机"行为L4 对齐层有害内容、格式偏好不一致Markdown/JSON 格式不统一、语气突变对齐效果差，用户体验割裂L1-L2 是门槛问题，脚本自动处理即可。L3-L4 是质量瓶颈，需要 LLM-as-Judge + 规则引擎的组合拳。一个有趣的数据：我统计过某开源数据集的 10 万条 SFT 数据，L1+L2 问题占了 8% 左右，L3 问题占了 15%，L4 占了约 5%。换句话说，如果你不做 L3 层清洗，有大约五分之一的数据在"污染"你的模型。常见的 SFT 数据格式在实际工作中，你会遇到多种数据格式。最常见的是 ShareGPT（OpenAI 对话格式）：{"messages":[{"role":"system","content":"你是一个有帮助的助手"},{"role":"user","content":"解释一下什么是过拟合"},{"role":"assistant","content":"过拟合是指模型在训练数据上表现很好，但在新数据上表现差的现象..."}]}另外还有 Alpaca 格式（instruction+input+output）、ChatML 格式（|im_start|token 分隔）等。我的建议是：无论原始数据是什么格式，在清洗前统一转成 ShareGPT 格式。这样所有下游工具只需要理解一种格式，Pipeline 的维护成本大幅降低。转换不难——核心就是确定"哪部分是用户说的，哪部分是模型该回复的"。三、L1 格式层：自动清洗流水线这层最容易自动化。下面是一套可以直接用的清洗 pipeline：importreimportjsonfromtypingimportList,Dict,OptionalclassFormatCleaner:"""SFT 数据 L1 格式层清洗器"""# 常见的乱码和特殊字符INVALID_CHARS=re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f]')# 控制字符（保留 \n \t）REPEATED_PUNCTUATION=re.compile(r'([!?。！？])\1{3,}')# 重复标点BOM_PATTERN='\ufeff'@staticmethoddefclean_text(text:str)-str:"""基础文本清洗"""# 1. 去除 BOMtext=text.replace(FormatCleaner.BOM_PATTERN,'')# 2. 去除无效控制字符text=FormatCleaner.INVALID_CHARS.sub('',text)# 3. Unicode 规范化（全角半角统一）text=text.replace('\u3000',' ')# 全角空格 → 半角# 4. 压缩重复标点（???? → ?）text=FormatCleaner.REPEATED_PUNCTUATION.sub(r'\1',text)# 5. 清理行首尾空白text='\n'.join(line.strip()forlineintext.split('\n'))returntext.strip()@staticmethoddeffix_encoding(text:str)-str:"""修复常见的双重编码问题（如 UTF-8 被当作 Latin-1 再编码）"""try:# 尝试检测并修复：文本被错误当作 latin-1 编码text.encode('latin-1').decode('utf-8')except(UnicodeDecodeError,UnicodeEncodeError):passreturntextclassStructureValidator:"""SFT 数据 L2 结构层验证器"""@staticmethoddefvalidate_conversation(conv:Dict)-Optional[str]:"""验证单条对话数据，返回错误信息或 None"""messages=conv.get('messages',conv.get('conversations',[]))ifnotmessages:return"空对话"# 检查每个消息的必要字段fori,msginenumerate(messages):ifnotmsg.get('content','').strip():returnf"第{i}条消息内容为空"role=msg.get('role','')ifrolenotin('user','assistant','system'):returnf"第{i}条消息角色异常:{role}"# 检查是否以 user 开头（SFT 训练通常要求 user → assistant 交替）ifmessages[0].get('role')!='user':return"对话不以 user 开头"# 检查最后一轮是否有 assistant 回复ifmessages[-1].get('role')!='assistant':return"对话不以 assistant 结尾（最后一轮无回复）"returnNone# 通过验证@staticmethoddeffix_truncation(messages:List[Dict])-List[Dict]:"""修复截断：如果最后一条 assistant 消息被截断，删除不完整的轮次"""ifnotmessages:returnmessages last=messages[-1]iflast.get('role')=='assistant':content=last.get('content','')# 检测截断：句子不完整、引号未闭合、代码块未闭合truncation_markers=[content.rstrip().endswith('...'),content.count('```')%2!=0,content.count('"')%2!=0,content.count('**')%2!=0,len(content)5# 过短]ifany(truncation_markers):# 删除最后的 user + assistant 轮次returnmessages[:-2]iflen(messages)=2