数据集准备：格式、清洗与质量

主流数据格式

微调数据本质上是"输入-输出"对，但不同格式适合不同场景：

Alpaca 格式

Stanford Alpaca 提出的经典指令格式，适合单轮指令跟随任务：

{
  "instruction": "将以下中文翻译成英文",
  "input": "人工智能正在改变世界",
  "output": "Artificial intelligence is changing the world"
}

// input 可以为空（纯指令场景）：
{
  "instruction": "写一首关于秋天的五言绝句",
  "input": "",
  "output": "秋风吹落叶，红霜染山林。\n..."
}

ShareGPT / 对话格式

多轮对话格式，适合聊天机器人和对话系统微调：

{
  "conversations": [
    {"from": "system", "value": "你是一个专业的法律顾问..."},
    {"from": "human",  "value": "劳动合同试用期最长多久？"},
    {"from": "gpt",   "value": "根据《劳动合同法》第19条..."},
    {"from": "human",  "value": "试用期可以不缴社保吗？"},
    {"from": "gpt",   "value": "不可以。试用期属于劳动关系存续期间..."}
  ]
}

Hugging Face Chat Template

现代模型普遍使用的格式，由 tokenizer 模板渲染：

# Llama 3 / ChatML 格式
messages = [
    {"role": "system",    "content": "你是专业的代码审查员..."},
    {"role": "user",      "content": "请审查这段 Python 代码"},
    {"role": "assistant", "content": "我发现以下几个问题：..."}
]
# 转化为训练文本（由 tokenizer.apply_chat_template 完成）
text = tokenizer.apply_chat_template(messages, tokenize=False)

数据质量的五个维度

数据清洗流水线

from datasets import Dataset
import re

def clean_dataset(examples):
    cleaned = []
    for item in examples:
        # 1. 过滤过短的输出（少于 20 字符）
        if len(item['output']) < 20:
            continue

        # 2. 过滤明显乱码（非 UTF-8 合法字符比例）
        if has_garbled_text(item['output']):
            continue

        # 3. 过滤包含敏感词的数据
        if contains_pii(item['output']):
            continue

        # 4. 去除首尾空白和异常字符
        item['output'] = item['output'].strip()
        item['output'] = re.sub(r'\n{3,}', '\n\n', item['output'])

        # 5. 截断超长数据（根据模型 max_length 设置）
        total_len = len(item['instruction']) + len(item['output'])
        if total_len > 4096:
            item['output'] = item['output'][:3000]

        cleaned.append(item)
    return cleaned

数据去重

重复数据会让模型过度拟合某些输出模式：

from datasketch import MinHash, MinHashLSH

def dedup_with_minhash(texts, threshold=0.8):
    """使用 MinHash LSH 进行模糊去重"""
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    minhashes = {}
    unique_indices = []

    for i, text in enumerate(texts):
        m = MinHash(num_perm=128)
        for word in text.split():
            m.update(word.encode('utf-8'))

        result = lsh.query(m)
        if not result:  # 没有重复
            lsh.insert(str(i), m)
            unique_indices.append(i)

    return unique_indices

使用 LLM 生成合成数据

当真实数据不足时，可以用强大的 LLM（GPT-4、Claude）生成高质量合成数据：

import anthropic

client = anthropic.Anthropic()

def generate_training_sample(topic: str, style: str) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""生成一条高质量的训练样本，格式为 JSON：
主题：{topic}
风格：{style}

要求：
- instruction 字段：一个真实用户会问的问题
- output 字段：专业、准确、格式规范的回答
- 输出纯 JSON，不要额外说明"""
        }]
    )
    return json.loads(response.content[0].text)

数据集拆分与检验

from datasets import Dataset
import json

# 加载并拆分数据集
data = json.load(open("training_data.json"))
dataset = Dataset.from_list(data)

# 90% 训练 / 10% 验证
split = dataset.train_test_split(test_size=0.1, seed=42)
train_ds = split['train']
eval_ds  = split['test']

# 基本统计检验
print(f"训练集：{len(train_ds)} 条")
print(f"验证集：{len(eval_ds)} 条")

# 输出长度分布
lengths = [len(x['output']) for x in train_ds]
print(f"平均长度：{sum(lengths)/len(lengths):.0f}")
print(f"最大长度：{max(lengths)}")