数据集准备与处理
"Garbage in, garbage out." 微调的效果上限由数据决定，而不是模型或方法。
1. 数据质量比数量重要
很多新手以为"数据越多越好"，但研究和实践都表明：
- 1000 条高质量数据 > 10000 条噪声数据
- LIMA（Meta 论文）：仅 1000 条精挑细选的样本就能训出效果出色的对话模型
- Alpaca：52K 条由 GPT 生成的指令数据，但有较多重复和低质内容
判断数据质量的几个标准：
| 维度 | 说明 |
|---|---|
| 多样性 | 覆盖不同主题、风格、难度 |
| 正确性 | 答案准确无误，不能误导 |
| 一致性 | 输出格式、风格统一 |
| 去重 | 避免完全相同或语义重复的样本 |
| 长度合理 | 太短信息少，太长浪费上下文 |
2. 常见数据格式
Alpaca 格式
最经典的 SFT 数据格式：
instruction：任务描述input：输入内容（可空）output：期望输出
适合单轮指令任务。
ShareGPT / OpenAI 多轮对话格式
{
"conversations": [
{"from": "system", "value": "你是一位友善的助手。"},
{"from": "human", "value": "Python 怎么读取文件？"},
{"from": "gpt", "value": "可以使用 open() 函数..."},
{"from": "human", "value": "如果文件不存在呢？"},
{"from": "gpt", "value": "需要捕获 FileNotFoundError..."}
]
}
或 OpenAI Messages 格式：
{
"messages": [
{"role": "system", "content": "你是一位友善的助手。"},
{"role": "user", "content": "你好"},
{"role": "assistant", "content": "你好！有什么可以帮你？"}
]
}
适合多轮对话场景。
DPO 偏好格式
chosen：好的回答rejected：差的回答
用于 DPO 偏好优化训练。
3. 数据来源
公开数据集
到 HuggingFace Datasets 浏览，常用中文数据集：
| 数据集 | 类型 | 规模 |
|---|---|---|
BelleGroup/train_3.5M_CN |
中文指令 | 350 万 |
silk-road/alpaca-data-gpt4-chinese |
GPT-4 生成中文指令 | 52K |
shareAI/CodeChat |
代码相关对话 | - |
wenge-research/yayi-train-data |
雅意通用指令 | - |
m-a-p/COIG-CQIA |
高质量中文指令 | 48K |
加载示例：
from datasets import load_dataset
dataset = load_dataset("m-a-p/COIG-CQIA", split="train")
print(dataset[0])
print(f"共 {len(dataset)} 条")
自建数据集
自建是私有任务最常见的方式，主要途径：
- 人工编写：质量最高，成本最高
- 业务日志：从客服、问答系统中抽取真实对话
- GPT 蒸馏：用 GPT-4/Claude 生成数据，成本低、规模可控
- 数据增强：基于已有样本生成相似变体
用 GPT 生成数据示例
from openai import OpenAI
import json
client = OpenAI()
def generate_qa(topic: str, n: int = 5):
prompt = f"""请围绕"{topic}"这个主题，生成 {n} 条高质量的问答对。
要求：
1. 问题多样化，覆盖不同角度
2. 回答准确、详细
3. 输出 JSON 数组格式：[{{"question": "...", "answer": "..."}}]
"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
)
return json.loads(response.choices[0].message.content)
data = generate_qa("Python 装饰器", n=10)
4. 数据清洗
去重
from datasets import load_dataset
dataset = load_dataset("json", data_files="raw_data.jsonl", split="train")
# 简单去重：基于完全相同的指令
seen = set()
def dedup(example):
key = example["instruction"]
if key in seen:
return False
seen.add(key)
return True
dataset = dataset.filter(dedup)
print(f"去重后剩余 {len(dataset)} 条")
更严格的方法是用 MinHash + LSH 做近似去重，参考 datasketch 库。
过滤低质量样本
def is_valid(example):
# 太短
if len(example["output"]) < 10:
return False
# 太长
if len(example["output"]) > 4000:
return False
# 包含敏感词
if any(word in example["output"] for word in ["违禁词1", "违禁词2"]):
return False
# 输出和输入完全一样
if example["output"] == example["input"]:
return False
return True
dataset = dataset.filter(is_valid)
长度统计
训练前看一下分布，决定 max_length：
import numpy as np
lengths = [len(x["instruction"]) + len(x.get("input", "")) + len(x["output"])
for x in dataset]
print(f"平均长度: {np.mean(lengths):.0f}")
print(f"中位数: {np.median(lengths):.0f}")
print(f"95%分位: {np.percentile(lengths, 95):.0f}")
print(f"最大长度: {max(lengths)}")
5. 应用对话模板（Chat Template）
不同模型有不同的对话模板，用错模板效果会大打折扣。HuggingFace tokenizer 内置了模板：
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
messages = [
{"role": "system", "content": "你是一位友善的助手。"},
{"role": "user", "content": "Python 是什么？"},
{"role": "assistant", "content": "Python 是一种解释型编程语言。"},
]
text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=False,
)
print(text)
输出（Qwen 模板）：
<|im_start|>system
你是一位友善的助手。<|im_end|>
<|im_start|>user
Python 是什么？<|im_end|>
<|im_start|>assistant
Python 是一种解释型编程语言。<|im_end|>
6. Tokenization
把文本变成模型能理解的 token id：
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
text = "Python 是一种编程语言。"
encoded = tokenizer(
text,
truncation=True,
max_length=512,
padding="max_length",
return_tensors="pt",
)
print(encoded["input_ids"].shape) # [1, 512]
print(tokenizer.decode(encoded["input_ids"][0][:10])) # 解码看看
7. 数据预处理完整脚本
下面是 SFT 训练前的标准数据处理流程：
from datasets import load_dataset
from transformers import AutoTokenizer
MODEL_NAME = "Qwen/Qwen2.5-7B-Instruct"
MAX_LENGTH = 1024
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
def format_alpaca(example):
"""将 Alpaca 格式转换为对话格式并应用模板"""
user_content = example["instruction"]
if example.get("input"):
user_content += "\n" + example["input"]
messages = [
{"role": "user", "content": user_content},
{"role": "assistant", "content": example["output"]},
]
text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=False,
)
return {"text": text}
def tokenize(example):
"""分词，保留 labels（用于计算 loss）"""
encoded = tokenizer(
example["text"],
truncation=True,
max_length=MAX_LENGTH,
padding=False,
)
encoded["labels"] = encoded["input_ids"].copy()
return encoded
# 加载原始数据
dataset = load_dataset("json", data_files="alpaca_data.jsonl", split="train")
print(f"原始数据: {len(dataset)} 条")
# 应用模板
dataset = dataset.map(format_alpaca, remove_columns=dataset.column_names)
# 分词
dataset = dataset.map(tokenize, remove_columns=["text"])
# 过滤超长样本
dataset = dataset.filter(lambda x: len(x["input_ids"]) <= MAX_LENGTH)
# 切分训练集和验证集
dataset = dataset.train_test_split(test_size=0.05, seed=42)
print(f"训练集: {len(dataset['train'])}, 验证集: {len(dataset['test'])}")
# 保存到磁盘，下次直接加载，不用重复处理
dataset.save_to_disk("./processed_data")
下次直接加载：
8. 一些实用经验
经验 1：先用小数据集跑通流程
不要一开始就用 100 万条数据，先用 100 条调通整个 pipeline，再扩大规模。
经验 2：保留多样性
如果数据全是同一类型（比如都是翻译），模型会失去其他能力（灾难性遗忘）。可以混入一些通用对话数据保持平衡。
经验 3：检查 labels 设置
SFT 中通常只对回答部分计算 loss，问题部分的 labels 设为 -100。详细做法在 SFT 章节会讲。
经验 4：动态 padding
不要把所有样本 pad 到同一长度，浪费显存。用 DataCollatorForLanguageModeling 按 batch 内最长样本动态 pad。
总结
- 数据质量 > 数据数量，1000 条高质量样本胜过 10K 噪声数据
- Alpaca 适合单轮指令，ShareGPT/Messages 适合多轮对话，DPO 用 chosen/rejected
- 数据清洗：去重、长度过滤、敏感词过滤
- 必须使用模型自带的
apply_chat_template，不能自己拼字符串 - 处理后的数据用
save_to_disk缓存，避免重复处理