32. Data Engineering for SFT | SFT 数据工程

难度： Medium | 环境： CPU-first | 标签： 训练微调, SFT, 数据清洗 | 目标人群： 训练机制学习者

🚀 云端运行环境

本章节的实战代码可以点击以下链接在免费 GPU 算力平台上直接运行：

(国内推荐：魔搭社区免费实例)

---

本节导读

SFT 训练经常不是输在优化器或学习率，而是输在数据没有被稳定组织成训练样本。字段缺失、空回答、模板不统一、超长样本混进来，这些问题都会直接污染 loss 和评测。09 先回答最小训练循环怎么跑，30 再把上下文长度拉长，而 32 往更上游退一步，先回答“在开始训练之前，数据本身是否已经整理到足够可靠”。

这一节不做复杂数据平台，也不追求完整的数据治理系统，而是先把 SFT 数据工程收敛成三个最小动作：原始记录怎样清洗成稳定字段、最小数据审计要看哪些异常、清洗结果怎样落成训练样本。它在训练微调路线里承担的是项目收口前补链的“数据入口”环节：09 先把 input_ids / attention_mask / labels 和 loss 对齐讲清楚，32 再往上游退一步，解释为什么脏样本会直接破坏 09 的训练闭环，后面才接 33 和 60。学完后，你应该能看清“原始记录 -> 数据审计 -> 训练样本 -> SFT 闭环”这条链路，而不是把数据问题留到训练 loss 异常时再回头排查。

关键词： dataset audit, prompt template, record cleaning

---

前置阅读

导语： 这一节同时承接数据组织和最小训练闭环两条线：先知道训练接口期待什么输入，再回来看原始指令数据为什么必须先被整理成稳定格式。

• 04. Python Config and Data Entry | Python 配置与数据组织
• 09. SFT Training Loop | SFT 训练循环
• 12. PyTorch Minimal Training Interface | PyTorch 最小训练接口

相关阅读

导语： 学完 SFT 数据工程后，下一步可以沿两条线继续走：一条是训练项目线，去看这些样本怎样进入完整微调交付；另一条是数据质量线，去验证“样本能读”是否真的等于“样本可训”。

• 30. Long Context Fine-Tuning | 长上下文微调
• 62. Instruction Fine-Tuning Project | 指令微调项目
• 64. SFT Data Quality Project | SFT 数据质量项目
• 2.3

---

Step 1: 先把原始记录清洗成稳定结构

• 统一保留 instruction / input / response 三个核心字段。
• 去掉多余空格和无意义空串，避免相同样本因为格式差异被误判成不同记录。
• 若字段缺失，要尽早补空值或剔除，而不是把异常留到训练阶段。

Step 2: 做最小数据审计

• 统计总样本数、空回答数、重复样本数和超长样本数。
• 长度问题要同时看 prompt 和 response，而不是只看单字段。
• 如果异常比例已经很高，优先返工数据，而不是继续调训练超参。

Step 3: 把清洗结果变成训练样本

• 把 instruction 和 input 拼成 prompt。
• 保留 response 作为监督目标。
• 给后续训练保留 metadata，例如 prompt 长度和原始索引，方便排查异常样本。

Step 4: 动手实战

1. 补全 clean_sft_records，统一字段并去掉首尾空白。
2. 补全 audit_sft_dataset，统计空回答、重复和超长样本。
3. 补全 build_sft_training_examples，把清洗后的记录转成训练样本。

提示

• TODO 1 先把每条原始记录统一成 instruction / input / response 三个字段，并去掉首尾空白。
• TODO 2 先统计总量，再依次统计空回答、重复样本、超长样本和平均长度。
• TODO 3 先把 instruction 和可选 input 拼成 prompt，再补训练所需的 metadata。

from typing import Dict, List

def clean_sft_records(records: List[Dict[str, str]]) -> List[Dict[str, str]]:
    """
    TODO 1: 统一字段并去掉首尾空白。
    """
    # 提示：先创建 cleaned，再逐条把 instruction / input / response 转成字符串并 strip。
    # cleaned = ???
    raise NotImplementedError


def audit_sft_dataset(records: List[Dict[str, str]], max_total_chars: int) -> Dict[str, float]:
    """
    TODO 2: 统计空回答、重复和超长样本。
    """
    # 提示：先建立 seen 和各项计数器，再逐条统计 total_chars、duplicate、empty_response 和 over_length。
    # total_samples = ???
    # duplicate_count = ???
    # avg_total_chars = ???
    raise NotImplementedError


def build_sft_training_examples(records: List[Dict[str, str]], template_prefix: str) -> List[Dict[str, object]]:
    """
    TODO 3: 把清洗后的记录转成训练样本。
    """
    # 提示：先创建 examples，再拼 prompt；如果有 input，再追加 Input 段。
    # examples = ???
    # prompt = ???
    raise NotImplementedError

def test_sft_data_engineering_template():
    try:
        records = [
            {'instruction': ' 解释 LoRA ', 'input': '', 'response': ' 一种低秩适配方法。 '},
            {'instruction': '解释 LoRA', 'input': '', 'response': '一种低秩适配方法。'},
            {'instruction': '给出总结', 'input': '结合训练成本', 'response': ''},
        ]
        cleaned = clean_sft_records(records)
        assert cleaned[0]['instruction'] == '解释 LoRA'
        assert cleaned[0]['response'] == '一种低秩适配方法。'
        audit = audit_sft_dataset(cleaned, max_total_chars=30)
        assert audit['total_samples'] == 3
        assert audit['empty_response_count'] == 1
        assert audit['duplicate_count'] == 1
        examples = build_sft_training_examples(cleaned[:1], template_prefix='### Instruction\n')
        assert examples[0]['has_input'] is False
        assert examples[0]['prompt'].startswith('### Instruction')
        print('测试通过：SFT 数据工程模板可以工作。')
    except NotImplementedError:
        raise
    except (AttributeError, NameError, TypeError, ValueError, AssertionError, IndexError) as e:
        raise NotImplementedError('请先完成 TODO 代码！') from e


test_sft_data_engineering_template()

---

🛑 STOP HERE 🛑

请先尝试自己完成代码并跑通测试。
如果你正在 Colab 中运行，并且遇到困难没有思路，可以向下滚动查看参考答案。

---

参考代码与解析

代码

def clean_sft_records(records: List[Dict[str, str]]) -> List[Dict[str, str]]:
    """
    TODO 1: 统一字段并去掉首尾空白。
    """
    # 提示：先创建 cleaned，再逐条把 instruction / input / response 转成字符串并 strip。
    # cleaned = ???
    cleaned: List[Dict[str, str]] = []
    for record in records:
        cleaned.append({
            'instruction': str(record.get('instruction', '')).strip(),
            'input': str(record.get('input', '')).strip(),
            'response': str(record.get('response', '')).strip(),
        })
    return cleaned


def audit_sft_dataset(records: List[Dict[str, str]], max_total_chars: int) -> Dict[str, float]:
    """
    TODO 2: 统计空回答、重复和超长样本。
    """
    # 提示：先建立 seen 和各项计数器，再逐条统计 total_chars、duplicate、empty_response 和 over_length。
    # total_samples = ???
    # duplicate_count = ???
    # avg_total_chars = ???
    seen = set()
    empty_response_count = 0
    duplicate_count = 0
    over_length_count = 0
    total_chars = 0
    for record in records:
        total_text = record.get('instruction', '') + record.get('input', '') + record.get('response', '')
        total_chars += len(total_text)
        key = (record.get('instruction', ''), record.get('input', ''), record.get('response', ''))
        if not record.get('response', '').strip():
            empty_response_count += 1
        if key in seen:
            duplicate_count += 1
        else:
            seen.add(key)
        if len(total_text) > max_total_chars:
            over_length_count += 1
    total_samples = len(records)
    return {
        'total_samples': total_samples,
        'empty_response_count': empty_response_count,
        'duplicate_count': duplicate_count,
        'over_length_count': over_length_count,
        'avg_total_chars': total_chars / total_samples if total_samples else 0.0,
    }


def build_sft_training_examples(records: List[Dict[str, str]], template_prefix: str) -> List[Dict[str, object]]:
    """
    TODO 3: 把清洗后的记录转成训练样本。
    """
    # 提示：先创建 examples，再拼 prompt；如果有 input，再追加 Input 段。
    # examples = ???
    # prompt = ???
    examples = []
    for record in records:
        instruction = record.get('instruction', '')
        input_text = record.get('input', '')
        prompt = f"{template_prefix}{instruction}"
        if input_text:
            prompt += f"\n### Input\n{input_text}"
        examples.append({
            'prompt': prompt,
            'response': record.get('response', ''),
            'prompt_chars': len(prompt),
            'has_input': bool(input_text),
        })
    return examples

解析

1. TODO 1：统一字段并去掉首尾空白

• 先把原始记录统一成 instruction / input / response 三个核心字段，再对每个字段做字符串化和 strip()。
• 数据工程的第一步不是上 tokenizer，而是先把结构收紧，避免同一条样本因为格式噪声被当成不同记录。

2. TODO 2：统计空回答、重复和超长样本

• 先建立 seen 和各项计数器，再逐条统计 empty_response_count、duplicate_count、over_length_count 和平均长度。
• 审计的意义是先确认数据问题究竟出在空回答、重复样本，还是整体长度失控，而不是盲目进入训练。

3. TODO 3：把清洗后的记录转成训练样本

• 先把 instruction 和可选 input 拼成统一 prompt，再补 response、prompt_chars 和 has_input 等训练所需 metadata。
• 训练循环真正消费的是统一模板下的 prompt + response，不是原始记录本身。

4. 这页的定位

• 数据工程的起点是把结构收紧，而不是直接进入 tokenizer。
• 审计至少要覆盖空回答、重复和超长样本三类高频问题。
• 训练循环真正消费的是统一模板下的 prompt + response，不是原始记录本身。