2.5.1 微调数据准备 · 标注 / 清洗 / 增强

2026-07-06· 编程工程· ai-llm·Fine-tuning·Instruction Tuning·SFT·数据标注·数据清洗·Self-Instruct· 约 1 分钟 · 统计中…

微调 LLM 前 80% 的工作都在准备数据 —— Alpaca / Self-Instruct / Evol-Instruct 三大范式 + 完整数据 pipeline

1. 为什么这个专题重要

微调 LLM 前 80% 的工作都在准备数据,这绝不是夸张。Stanford Alpaca 团队 2023 年的实验显示,在同一基座 LLaMA-7B 上,用 52K 清洗后的 Alpaca 数据微调,MT-Bench 评分 4.3;而用未清洗的原始 Self-Instruct 输出微调,评分只有 2.1——清洗带来的提升(2.2 分)远大于模型架构或超参调整(<0.3 分)。

“Garbage In, Garbage Out”在 LLM 时代被放大了十倍:模型越强,坏数据的反噬越明显。

1.1 微调失败的 7 类案例分布

根据社区公开复盘(HuggingFace Alignment Handbook / 内部 MLOps 报告综合),微调失败的根因分布大致如下:

失败类型	占比	表现
数据脏(重复 / 幻觉 / 格式乱)	32%	loss 震荡 / 输出复读 / 格式崩坏
数据偏(单一来源 / 领域失衡)	14%	风格单一 / 某些题型彻底崩
数据漏(测试集泄露)	9%	训练 loss 低,测试 loss 高 / 评估虚高
标注错(Schema 模糊 / 标错)	5%	模型学到错的关联
数据少 + 硬上微调(<1K 跑全量)	11%	灾难性遗忘 / 反而不如基座
超参 / 训练本身问题	19%	lr 错 / epoch 过多 / 量化损失
其它(环境 / 评测脚本 bug)	10%	误以为模型坏了

加粗前四项,纯数据问题占 60%。这个数字告诉我们:把数据 pipeline 写扎实,等于解决了大部分微调失败。

1.2 数据准备的标准产出物

一个合格的 SFT(Supervised Fine-Tuning)数据 pipeline,最后应交付:

一份 train.jsonl —— 每行一个 JSON,字段确定,UTF-8 无 BOM
一份 data_card.yaml —— 数据来源 / 规模 / 标注规范 / 已知问题
一份 quality_report.html —— 长度分布 / 重复率 / PII 命中 / LLM-as-Judge 抽样评分
一份 split_manifest.json —— train / dev / test 划分的 hash,防止后续误改

后续小节按”范式 → 标注 → 清洗 → 增强 → 格式 → 案例 → 决策 → 踩坑”展开。

2. 微调数据三大范式全景对比

当前主流的 SFT 数据生成范式可归纳为三类:指令输出对、对话历史、指令进化。三者并非互斥,实际项目常组合使用。

2.1 三大范式横向对比

维度	Alpaca 范式	Self-Instruct 范式	Evol-Instruct 范式
数据形式	单轮 instruction-output	单轮 instruction-output	单轮 instruction-output(进化版)
生成方式	人工写 seed + LLM 扩写	LLM 从 100 seed 自举	LLM 把简单指令改写成复杂指令
代表数据集	Alpaca-52K	Self-Instruct-52K	WizardLM-Evol-Instruct-70K
数据规模	5 万量级	5 万量级	7 万量级
质量	中(GPT-3.5 生成)	中偏低(早期 GPT-3 生成)	较高(过滤 + 进化)
多样性	中(受限于 175 seed)	高	高(5 个方向进化)
典型训练	LLaMA-1 时代主流	早期 SFT baseline	WizardLM / WizardMath
主要缺陷	instruction 偏短 / 单一	幻觉 / 重复多	可能改写出无效指令

2.2 三大范式纵向演进

timeline
    title 三大 SFT 数据范式纵向演进
    2022.12 : Self-Instruct (Wang 2022) — LLM 自举 GPT-3 — 52K
    2023.03 : Alpaca (Stanford) — 175 seed 扩写 GPT-3.5 — 52K
    2023.04 : ShareGPT — 用户分享 GPT-4 对话 — ~90K
    2023.05 : Evol-Instruct (WizardLM) — 5 方向进化 GPT-3.5/4 — 70K+

2.3 对话历史类数据集

除单轮范式,另一大类是直接抓取 ChatGPT / Claude 的真实多轮对话:

数据集	来源	规模	多轮	特点
ShareGPT	ShareGPT 站	90K	是	多轮 GPT-4 对话,质量高,版权争议
OpenHermes	Teknium	1M+	混合	汇集 Alpaca / Wizard / Orca 多源
UltraChat	开源	1.5M	是	围绕世界知识 / 创作 / 编程三类
LIMA	Meta	1K	混合	“Less Is More”,1K 高质量样本击败 1M

LIMA 论文 2023 的核心结论:65% 来自 Stack Exchange + WikiHow + Reddit 等高质量人工筛选,35% 来自 ShareGPT 风格人工改写,只用 1K 条就把 LLaMA-2-65B 的 win-rate 拉到了 45%(vs GPT-4 的 35%)。

启示:数据质量 > 数据数量,但 1K 太少不具普适性,工业落地多在 5K-50K 区间。

3. 数据标注详解

标注是数据 pipeline 的第一站,决定后续一切的天花板。

3.1 标注方式三分天下

flowchart TD
    A["数据标注"]:::root
    A --> B["人工标注"]
    A --> C["LLM 辅助"]
    A --> D["半自动<br/>(LLM + 人工 spot check)"]

    B --> B1["内训"]
    B --> B2["众包"]
    B1 --- B1a["Label Studio"]
    B1 --- B1b["Prodigy"]
    B2 --- B2a["Amazon Mechanical Turk"]

    C --> C1["GPT-4"]
    C --> C2["Claude"]

    classDef root fill:#f9f,stroke:#333,stroke-width:2px;

3.1.1 人工标注方案选型

工具	部署	协作	适用规模	短板
Label Studio	自部署	多用户	1K-100K	模板学习曲线
Prodigy	本地	单人/小团队	<10K	商业授权
Labelbox	SaaS	大团队	>100K	贵 / 锁定生态
众包(Amazon Mechanical Turk)	远程	大	>10K	质量波动大
Argilla	自部署	中	1K-50K	较新,生态薄

3.1.2 Label Studio 启动 + 上传

pip install label-studio
label-studio start --port 8080
# 浏览器打开 http://localhost:8080,新建 Project → 上传 task.jsonl

task.jsonl 格式示例:

{"id": 1, "instruction": "解释什么是糖尿病", "output": ""}
{"id": 2, "instruction": "列出三种降压药", "output": ""}

3.2 LLM 辅助标注

LLM 辅助标注不是”用 GPT-4 全量替代人”,而是 GPT-4 标 + 人工 spot check + 主动学习。OpenAI Fine-tuning Guide(2024)推荐:全量 LLM 标注 + 5%-10% 人工复核。

# llm_label.py — 用 GPT-4 批量标注,然后 spot check
import json
from openai import OpenAI

client = OpenAI()

PROMPT = """你是一个标注员。请根据输入指令,生成高质量回答。
要求:
1. 准确(无幻觉)
2. 完整(覆盖问题)
3. 简洁(< 300 字)
4. 中文输出

指令: {instruction}
回答:"""

def label_one(instruction: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": PROMPT.format(instruction=instruction)}],
        temperature=0.3,
        max_tokens=600,
    )
    return resp.choices[0].message.content.strip()

def label_batch(input_path: str, output_path: str):
    with open(input_path, encoding="utf-8") as fin, \
         open(output_path, "w", encoding="utf-8") as fout:
        for line in fin:
            row = json.loads(line)
            row["output"] = label_one(row["instruction"])
            fout.write(json.dumps(row, ensure_ascii=False) + "\n")

3.3 标注一致性:Kappa 系数

多人标注同一批样本时,必须用 Cohen’s Kappa(两人)或 Fleiss’ Kappa(多人)评估一致性。Kappa > 0.8 才算合格;0.6-0.8 需讨论;< 0.6 必须重写 Schema。

# kappa.py — 标注一致性计算
from sklearn.metrics import cohen_kappa_score

def calc_kappa(annotator_a: list, annotator_b: list) -> float:
    """
    annotator_a, annotator_b: 同长度整数列表(0/1/2/...)
    返回 Cohen's Kappa
    """
    return cohen_kappa_score(annotator_a, annotator_b)

# 示例:两位标注员对 100 条医疗问答的"是否合格"判定
a = [1]*60 + [0]*25 + [1]*10 + [0]*5
b = [1]*58 + [0]*22 + [1]*12 + [0]*8
print(f"Kappa = {calc_kappa(a, b):.3f}")
# 输出 Kappa = 0.781 → 良好,可接受

# fleiss_kappa.py — 3 人以上
import numpy as np
from statsmodels.stats.inter_rater import cohens_kappa, fleiss_kappa

def calc_fleiss(table: np.ndarray) -> float:
    """
    table: shape (n_subjects, n_categories)
    每行是一个样本被各标注员分到各类别的计数
    """
    return fleiss_kappa(table)

3.4 标注 Schema 设计

Schema 是标注的”宪法”,必须封闭、无歧义、可枚举。反面教材:

# 反例 ❌
fields:
  quality: {type: string, description: "好的或不好的"}

正例:

# 正例 ✅
fields:
  - name: relevance
    type: int
    enum: [0, 1, 2]
    desc: |
      0=答非所问
      1=部分相关
      2=完全相关
  - name: factuality
    type: int
    enum: [0, 1, 2]
    desc: |
      0=含明确幻觉
      1=事实无误但表述模糊
      2=事实精确
  - name: tone
    type: enum
    values: [formal, casual, professional]
  - name: is_pii_leak
    type: bool
    desc: 输出是否含个人隐私(电话/身份证/姓名等)

3.5 完整标注 pipeline

# pipeline.py — 标注主流程
import json
import random

def main(input_path: str, output_path: str, sample_ratio: float = 0.05):
    """
    1) LLM 全量标注
    2) 随机抽 sample_ratio 比例走人工 spot check
    3) 不合格的回炉重标
    """
    # Step 1: LLM 标注
    label_batch(input_path, output_path + ".llm")

    # Step 2: 抽样
    with open(output_path + ".llm", encoding="utf-8") as f:
        rows = [json.loads(line) for line in f]
    n_spot = int(len(rows) * sample_ratio)
    spot = random.sample(rows, n_spot)

    with open(output_path + ".spot.jsonl", "w", encoding="utf-8") as f:
        for r in spot:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")

    # Step 3: 人工 spot check 在 Label Studio 完成,输出 pass/fail 标签
    # Step 4: pass 率 < 0.9 则回炉重标(脚本略)

if __name__ == "__main__":
    main("raw_instructions.jsonl", "labeled.jsonl")

4. 数据清洗详解

清洗三板斧:去重 / 质量过滤 / PII 脱敏。顺序很重要——先去重,再过滤,最后脱敏。

4.1 去重:从精确到近似

4.1.1 精确去重(MD5)

# dedup_exact.py — 基于 MD5 的精确去重
import hashlib
import json

def dedup_exact(input_path: str, output_path: str):
    seen = set()
    with open(input_path, encoding="utf-8") as fin, \
         open(output_path, "w", encoding="utf-8") as fout:
        for line in fin:
            row = json.loads(line)
            key = hashlib.md5(
                (row["instruction"] + row.get("input", "") + row["output"]).encode()
            ).hexdigest()
            if key in seen:
                continue
            seen.add(key)
            fout.write(json.dumps(row, ensure_ascii=False) + "\n")

精确去重只能干掉完全相同的样本,而 LLM 生成的数据大量是近似重复(改几个字、换种说法),必须上近似去重。

4.1.2 MinHash 近似去重

MinHash 是 1997 年 Broder 提出的概率算法,可高效估计两个集合的 Jaccard 相似度。datasketch 是 Python 最常用的实现。

# dedup_minhash.py — MinHash 近似去重
from datasketch import MinHash, MinHashLSH

def shingle(text: str, k: int = 5) -> set:
    """把字符串切成 k-shingle(字符 n-gram)"""
    return {text[i:i+k] for i in range(len(text) - k + 1)}

def build_minhash(text: str, num_perm: int = 128) -> MinHash:
    m = MinHash(num_perm=num_perm)
    for s in shingle(text):
        m.update(s.encode("utf-8"))
    return m

def dedup_minhash(rows: list, threshold: float = 0.85):
    """
    threshold: Jaccard 相似度阈值
    典型:0.7-0.85,过高(>0.9)漏检,过低(<0.6)误杀
    """
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    keep = []
    for i, row in enumerate(rows):
        text = row["instruction"] + " " + row["output"]
        m = build_minhash(text)
        if lsh.query(m):
            continue  # 与已有样本相似,丢弃
        lsh.insert(str(i), m)
        keep.append(row)
    return keep

4.1.3 SimHash 海量去重

SimHash(Charikar 2002)是 Google 用来在万亿级网页中找近似重复的算法,快、内存小,特别适合百万级以上数据。

# dedup_simhash.py
import jieba
from simhash import Simhash

def tokenize(text: str) -> list:
    return [w for w in jieba.cut(text) if len(w) > 1]

def get_simhash(text: str) -> int:
    return Simhash(tokenize(text), f=64).value

def hamming(a: int, b: int) -> int:
    return bin(a ^ b).count("1")

def dedup_simhash(rows: list, max_hamming: int = 3):
    """海明距离 ≤ 3 视为近似重复"""
    seen = []
    keep = []
    for row in rows:
        h = get_simhash(row["instruction"] + " " + row["output"])
        if any(hamming(h, s) <= max_hamming for s in seen):
            continue
        seen.append(h)
        keep.append(row)
    return keep

4.1.4 Embedding 相似度去重(最贵但最准)

# dedup_embedding.py
import numpy as np
import torch
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-small-zh-v1.5")

def embed_batch(texts: list) -> np.ndarray:
    return model.encode(texts, normalize_embeddings=True, batch_size=64)

def dedup_embedding(rows: list, threshold: float = 0.92):
    """
    threshold: 余弦相似度,典型 0.9-0.95
    比 MinHash 准,但贵 10-100 倍
    """
    texts = [r["instruction"] + " " + r["output"] for r in rows]
    embs = embed_batch(texts)
    keep_idx = []
    for i, e in enumerate(embs):
        if not keep_idx:
            keep_idx.append(i)
            continue
        sims = embs[keep_idx] @ e
        if sims.max() < threshold:
            keep_idx.append(i)
    return [rows[i] for i in keep_idx]

4.2 质量过滤

4.2.1 启发式规则

# quality_heuristic.py
import re

def is_valid(row: dict) -> tuple[bool, str]:
    """返回 (是否合格, 原因)"""
    inst, out = row.get("instruction", ""), row.get("output", "")

    # 长度
    if len(inst) < 5:   return False, "instruction too short"
    if len(out) < 20:   return False, "output too short"
    if len(out) > 4000: return False, "output too long"

    # 中英比(根据任务调)
    if len(re.findall(r"[\u4e00-\u9fa5]", out)) / max(len(out), 1) < 0.3:
        return False, "low chinese ratio"

    # 重复行(模型复读)
    lines = [l for l in out.split("\n") if l.strip()]
    if len(lines) > 3 and len(set(lines)) / len(lines) < 0.3:
        return False, "high line repetition"

    # 常见幻觉前缀
    bad_starts = ["作为 AI", "作为一个 AI 模型", "I cannot", "I'm sorry"]
    if any(out.strip().startswith(b) for b in bad_starts):
        return False, "ai refusal prefix"

    # 模板污染
    if "<<USER>>" in out or "<|im_start|>" in out:
        return False, "template leakage"

    return True, "ok"

4.2.2 LLM-as-Judge

# llm_judge.py — 用 GPT-4 当裁判,5 分制
from openai import OpenAI
client = OpenAI()

JUDGE_PROMPT = """你是质检员,严格评分(1-5):
1=胡言乱语 / 含幻觉
2=答非所问
3=基本正确但缺信息
4=正确完整
5=正确完整且有洞见

指令: {instruction}
回答: {output}

只输出一个整数(1-5)。"""

def judge(inst: str, out: str) -> int:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user",
                   "content": JUDGE_PROMPT.format(instruction=inst, output=out)}],
        temperature=0,
    )
    try:
        return int(resp.choices[0].message.content.strip()[0])
    except (ValueError, IndexError):
        return 3

# 抽样 5% 做 LLM-as-Judge,过滤掉 < 3 分的

4.3 PII 脱敏

PII(Personally Identifiable Information)是数据合规的红线。医疗 / 法律 / 金融场景尤其敏感。

4.3.1 正则方案(中文 + 英文)

# pii_regex.py
import re

PII_PATTERNS = {
    "id_card_cn":  r"\d{17}[\dXx]",
    "phone_cn":    r"1[3-9]\d{9}",
    "email":       r"[\w.+-]+@[\w-]+\.[a-zA-Z]{2,}",
    "ipv4":        r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}",
    "credit_card": r"\d{4}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}",
    "name_cn":     r"[\u4e00-\u9fa5]{2,4}(先生|女士|医生|老师|主任)",
}

def scrub_text(text: str) -> str:
    for label, pat in PII_PATTERNS.items():
        text = re.sub(pat, f"[{label.upper()}_REDACTED]", text)
    return text

4.3.2 Presidio(工业级)

# pii_presidio.py — Microsoft Presidio 工业级方案
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def scrub_presidio(text: str, lang: str = "zh") -> str:
    results = analyzer.analyze(text=text, language=lang)
    return anonymizer.anonymize(text=text, analyzer_results=results).text

# 示例
print(scrub_presidio("张三先生电话 13800138000,身份证 110101199003078811"))
# 张三[REDACTED]电话 [REDACTED],身份证 [REDACTED]

4.4 完整清洗 pipeline

# clean_pipeline.py
import json

def clean(input_path: str, output_path: str):
    with open(input_path, encoding="utf-8") as f:
        rows = [json.loads(l) for l in f]
    print(f"[1/4] load: {len(rows)}")

    # 1) PII 脱敏(先做,避免后续去重 hash 含 PII)
    for r in rows:
        r["instruction"] = scrub_presidio(r["instruction"])
        r["output"]      = scrub_presidio(r["output"])

    # 2) 启发式过滤
    before = len(rows)
    rows = [r for r in rows if is_valid(r)[0]]
    print(f"[2/4] heuristic filter: {before} -> {len(rows)}")

    # 3) MinHash 去重
    before = len(rows)
    rows = dedup_minhash(rows, threshold=0.85)
    print(f"[3/4] dedup: {before} -> {len(rows)}")

    # 4) LLM-as-Judge(抽样,标记低分)
    # 实际只对 5% 抽样打分,过滤掉 < 3 分
    # rows = [r for r in rows if judge(r["instruction"], r["output"]) >= 3]

    with open(output_path, "w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print(f"[4/4] save: {len(rows)} -> {output_path}")

5. 数据增强详解

数据增强在数据少时是救命稻草。三个核心武器:Self-Instruct / Evol-Instruct / Back-translation。

5.1 Self-Instruct 完整实现

Self-Instruct 是 Wang 2022 提出的自举法,核心四步:seed pool → LLM 生成新指令 → 分类过滤 → 去重入池。

5.1.1 Seed Task Pool 准备

# seed_pool.py — 种子任务池
SEED_TASKS = [
    {"instruction": "解释相对论的核心思想", "category": "Science"},
    {"instruction": "给一段 Python 代码,实现快速排序", "category": "Coding"},
    {"instruction": "把这句话翻译成英文:今天天气很好", "category": "Translation"},
    {"instruction": "列出 5 个适合周末和朋友一起做的活动", "category": "Brainstorming"},
    {"instruction": "总结这段新闻的核心内容", "category": "Summarization"},
    # ... 至少 100 条,覆盖目标任务的各类题型
]

5.1.2 Self-Instruct 生成

# self_instruct.py
import json
import random
from openai import OpenAI
client = OpenAI()

GEN_PROMPT = """你是指令生成器。基于下面的种子指令,生成 {n} 条新的、不同的指令。
要求:
1. 不要与种子重复
2. 覆盖尽可能多的任务类型
3. 中文输出

种子指令:
{seed_block}

新指令(每行一条,前面加 - ):"""

def generate_new_tasks(seed: list, n: int = 8) -> list:
    seed_block = "\n".join(f"- {s['instruction']}" for s in seed[:8])
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user",
                   "content": GEN_PROMPT.format(n=n, seed_block=seed_block)}],
        temperature=0.9,  # 高温,鼓励多样性
        max_tokens=800,
    )
    text = resp.choices[0].message.content
    return [line[2:].strip() for line in text.split("\n")
            if line.strip().startswith("-")]

5.1.3 分类 + 过滤

# filter.py — Self-Instruct 的过滤关键步骤
def is_classification_task(instruction: str) -> bool:
    """检测分类题(只生成 output,不该用作 SFT)"""
    keywords = ["分类", "判断", "是不是", "属于哪个", "以下哪个"]
    return any(k in instruction for k in keywords)

def too_similar_to_seed(new_inst: str, seed: list, threshold: float = 0.85) -> bool:
    """用 MinHash 检查与种子是否近似"""
    new_m = build_minhash(new_inst)
    for s in seed:
        s_m = build_minhash(s["instruction"])
        if new_m.jaccard(s_m) > threshold:
            return True
    return False

def filter_tasks(new_tasks: list, seed: list) -> list:
    keep = []
    for t in new_tasks:
        if len(t) < 5 or len(t) > 350:        continue  # 长度合理
        if is_classification_task(t):         continue  # 丢弃纯分类
        if too_similar_to_seed(t, seed):      continue  # 丢弃近似
        if any(t == s["instruction"] for s in seed): continue
        keep.append(t)
    return keep

5.1.4 主循环

# self_instruct_main.py
def self_instruct_pipeline(seed: list, target_size: int = 50000) -> list:
    pool = list(seed)
    seen = {s["instruction"] for s in seed}
    iteration = 0
    while len(pool) < target_size:
        iteration += 1
        # 随机抽 8 个种子当 context
        ctx = random.sample(pool, min(8, len(pool)))
        new = generate_new_tasks(ctx, n=8)
        new = filter_tasks(new, ctx)
        for t in new:
            if t not in seen:
                seen.add(t)
                pool.append({"instruction": t, "category": "Generated"})
        print(f"[iter {iteration}] pool size = {len(pool)}")
    return pool

5.2 Evol-Instruct 完整实现

WizardLM 团队 2023 年提出的指令进化法,把简单指令改写成复杂指令,分五个进化方向:

方向	含义	例子
add constraints	加约束	“写诗” → “写七言绝句,押韵’ang’,主题’边塞’”
deepening	加深度	“解释 X” → “对比 X 与 Y 的 3 个本质区别”
concretizing	具体化	“写代码” → “写 Python 爬虫,目标豆瓣电影 Top250”
increased reasoning	增推理	“1+1=?” → “甲有 5 元买 2 个包子 1.5 元/个,剩多少?”
widening	扩广度	“介绍猫” → “介绍猫科 5 种动物的栖息地和习性”

# evol_instruct.py
from openai import OpenAI
client = OpenAI()

EVOL_PROMPTS = {
    "add_constraints": """对原指令加 2-3 条约束(格式 / 长度 / 风格 / 假设),
不要改变原意。原指令:{inst}
进化后指令:""",
    "deepening": """对原指令加 1 阶深度(对比 / 因果 / 类比)。
原指令:{inst}
进化后指令:""",
    "concretizing": """把原指令具体化到可执行场景。
原指令:{inst}
进化后指令:""",
    "increased_reasoning": """加入 2-3 步推理,需要列式计算或逻辑推导。
原指令:{inst}
进化后指令:""",
    "widening": """扩展主题到相关邻域,要求覆盖 3-5 个相关方面。
原指令:{inst}
进化后指令:""",
}

def evolve(instruction: str, direction: str = "deepening") -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user",
                   "content": EVOL_PROMPTS[direction].format(inst=instruction)}],
        temperature=0.7,
    )
    return resp.choices[0].message.content.strip()

5.3 Back-translation(回译增强)

# back_translation.py — 中 → 英 → 中,自然改写
from openai import OpenAI
client = OpenAI()

def back_translate(text: str) -> str:
    en = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user",
                   "content": f"翻译成英文,保持原意:\n{text}"}],
    ).choices[0].message.content.strip()
    zh = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user",
                   "content": f"把以下英文翻译回中文,要求通顺自然:\n{en}"}],
    ).choices[0].message.content.strip()
    return zh

6. 格式转换

不同框架吃不同格式,这是 SFT 入门第一道坎。

6.1 五大格式横向对比

格式	文件结构	多轮	主流框架
Alpaca	`{instruction, input, output}`	否	LLaMA-Factory / axolotl
ShareGPT	`conversations: [{from, value}]`	是	FastChat / LLaMA-Factory
OpenAI ChatML	`messages: [{role, content}]`	是	OpenAI / vLLM
Llama-3 chat	`<\\|begin_of_text\\|>...<\\|eot_id\\|>`	是	Llama-3 原生
Mistral Instruct	`[INST] {q} [/INST] {a}`	否	Mistral 原生

6.2 Alpaca 格式样例

{
  "instruction": "把以下中文翻译成英文",
  "input": "今天天气很好",
  "output": "The weather is nice today."
}

6.3 ShareGPT 格式样例

{
  "conversations": [
    {"from": "human", "value": "你好"},
    {"from": "gpt", "value": "你好!有什么可以帮你?"},
    {"from": "human", "value": "讲个笑话"},
    {"from": "gpt", "value": "为什么程序员总穿黑衣?因为他们怕 bug!"}
  ]
}

6.4 互转脚本

# format_convert.py
def alpaca_to_sharegpt(row: dict) -> dict:
    return {
        "conversations": [
            {"from": "human",
             "value": row["instruction"] + ("\n" + row.get("input", "")
                                            if row.get("input") else "")},
            {"from": "gpt", "value": row["output"]},
        ]
    }

def sharegpt_to_chatml(row: dict) -> dict:
    role_map = {"human": "user", "gpt": "assistant", "system": "system"}
    return {"messages": [
        {"role": role_map[c["from"]], "content": c["value"]}
        for c in row["conversations"]
    ]}

def chatml_to_llama3(row: dict) -> str:
    """Llama-3 chat template: <|begin_of_text|><|start_header_id|>...<|eot_id|>"""
    parts = ["<|begin_of_text|>"]
    for m in row["messages"]:
        parts.append(f"<|start_header_id|>{m['role']}<|end_header_id|>\n\n{m['content']}<|eot_id|>")
    parts.append("<|start_header_id|>assistant<|end_header_id|>")
    return "\n".join(parts)

def chatml_to_mistral(row: dict) -> str:
    """Mistral 单轮模板 [INST]...[/INST]"""
    msgs = row["messages"]
    # 单轮简化
    user = next((m["content"] for m in msgs if m["role"] == "user"), "")
    asst = next((m["content"] for m in msgs if m["role"] == "assistant"), "")
    return f"<s>[INST] {user} [/INST] {asst}</s>"

6.5 训练前一定要验证 token 化

# verify_template.py
from transformers import AutoTokenizer
tok = AutoTokenizer.from_pretrained("meta-llama/Meta-Llama-3-8B-Instruct")

sample = "<|begin_of_text|><|start_header_id|>user<|end_header_id|>\n\n你好<|eot_id|>"
ids = tok.encode(sample, add_special_tokens=False)
print(tok.decode(ids))
# 应原样还原

7. 实战案例 4 个

7.1 案例 1:从 0 构建中文医疗问答 SFT 数据集

场景:某医疗 SaaS 公司要做”医生助手”,需要让基座模型理解中文医学术语 + 给出合规回答。

流程(约 3 周,2 人):

种子指令 200 条:由 3 名三甲医生写,覆盖内科 / 外科 / 妇儿 / 药理 4 类
LLM 扩写:GPT-4 对每个 seed 生成 50 条变体 → 10000 条候选
医生 spot check:随机抽 1000 条由医生打分(relevance / factuality / 安全性),Kappa=0.82
质量过滤:
- 启发式去掉 < 30 字 / 重复行 / 含 PII
- Presidio 脱敏所有患者信息
- LLM-as-Judge 抽 5%,过滤掉 < 3 分
MinHash 去重:阈值 0.85,10000 → 5800
领域增强:Evol-Instruct 加深度,把”什么是高血压”进化为”对比原发性与继发性高血压的鉴别诊断”
格式:转 Alpaca + 注入 system prompt:”你是医生助手,回答需附循证等级”
最终:5800 条 train + 200 条 dev + 200 条 test,没有任何样本重叠(MD5 校验)

质量评估:在 200 条 test 上 GPT-4 judge 平均 4.21/5,医生 spot check 合格率 92%。

7.2 案例 2:Self-Instruct 5W → 3.5W

场景:客户只有 100 条高质量中文客服 seed,要快速扩到可用规模。

实测:

100 条 seed 启动,跑 Self-Instruct 50 轮
原始生成 52341 条
过滤:启发式 -12%(弃太短/太长/分类题)
MinHash 阈值 0.85 去重 -28%,剩 35120 条
LLM-as-Judge 抽样 5%,3 分以下 -5%,剩 33364 条
最终 33364 条 SFT 数据,训练 Qwen-1.5B 单卡 4h,客服意图识别 F1 从 0.61 → 0.84

关键经验:去重比生成更重要——生成 5W 容易,但去重后剩 3.5W 才是有效数据。

7.3 案例 3:Evol-Instruct 进化对比

实验:同基座 LLaMA-3-8B,同超参,两组训练数据:

A 组:5000 条原始简单指令(平均 12 字)
B 组:5000 条 Evol-Instruct 进化后(平均 47 字)

MT-Bench 评分(GPT-4 judge):

维度	A 组	B 组	Δ
Reasoning	3.8	4.5	+0.7
Coding	3.5	3.9	+0.4
Writing	4.0	4.3	+0.3
Extraction	4.2	4.2	0
Average	3.88	4.23	+0.35

结论:推理 / 编码类任务受益最大(+0.7 / +0.4),纯抽取类(原文已在,无需复杂生成)提升为零,与 WizardLM 论文一致。

7.4 案例 4:医疗病历 PII 脱敏

场景:医院 12000 份电子病历,要做 SFT,但合规要求去除患者隐私。

Pipeline:

正则初筛:去除身份证 / 手机号 / 银行卡 → 命中 9.8%
Presidio 深度识别:姓名 / 地址 / 医院名 / 医生名 → 命中 6.3%
人工复核:随机 100 份,召回率 96.4%,误报率 2.1%
回填:漏掉的 3.6% 手工补 + 入”待脱敏字典”重新跑

效果:脱敏前后模型 F1 几乎无变化(0.89 → 0.88,差 0.01 在误差范围内),但合规通过率 100%。脱敏的代价微乎其微,但合规价值千金。

8. 选型决策树 + 数据规模建议

8.1 决策树

flowchart TD
    Q{"你要做 SFT?"}:::decision
    Q -->|"有标注数据"| S1
    Q -->|"无标注数据"| R["Prompt / RAG<br/>(不需要 SFT)"]

    S1{"数据规模?"}:::decision
    S1 -->|"数据 > 10K"| P1["清洗 + 去重"]:::action
    S1 -->|"数据 < 1K"| P2["Self-Instruct<br/>生成 5W+ 候选<br/>→ 清洗 → 35K"]:::action

    P1 --> S2{"任务需要复杂推理?"}:::decision
    S2 -->|"是"| F1["Evol-Instruct<br/>(进化 + 训练)"]:::final
    S2 -->|"否"| F2["简单指令即可<br/>(清洗 + 训练)"]:::final

    classDef decision fill:#ffd54f,stroke:#b8860b,color:#000
    classDef action fill:#bbdefb,stroke:#1976d2,color:#000
    classDef final fill:#a5d6a7,stroke:#2e7d32,color:#000

8.2 数据规模建议表

任务类型	数据量(最低 / 推荐)	训练 epoch	团队规模	预算(估算)
客服意图分类	500 / 2K	3-5	1-2 人	< $500
领域问答	2K / 10K	2-3	2-3 人	$500-$2K
复杂多轮对话	5K / 30K	1-2	3+ 人	$2K-$10K
代码生成(单语言)	10K / 50K	1-2	3+ 人	$5K-$20K
数学推理	20K / 100K	2-3	5+ 人	$10K-$50K
通用助手	50K / 500K	1-2	10+ 人	$50K-$500K

预算估算含 GPT-4 标注费 + 训练算力(云 GPU)。

8.3 选型口诀

1. 数据少别硬上,Prompt / RAG 先试一趟。 2. 数据脏毁一切,去重过滤比加量更重要。 3. 质量先于数量,1K 高质胜过 1M 灌水。

9. 踩坑 6 个

坑 1:Alpaca 格式 instruction 是空的

症状:训练后模型会”答非所问”,用户问啥它都自己接话,而非遵循 instruction。

原因:直接从用户 query 提取 instruction,但 query 是用户问的问题(如”什么是 AI”),没有真正告诉模型要”做什么”(如”用一段话解释什么是 AI,要求通俗易懂,200 字以内”)。

修法:把 instruction 当成”任务描述”而不是”用户原始问题”。必要时手动改写。

# fix_empty_instruction.py
def refine_instruction(q: str) -> str:
    """把用户原问题改写成清晰的指令"""
    return f"请用通俗易懂的语言回答以下问题,200 字以内:\n{q}"

row["instruction"] = refine_instruction(row["raw_query"])

坑 2:Self-Instruct 生成内容有幻觉

症状:训练数据中夹杂大量”看起来合理但事实错误”的内容,模型学到了”自信胡说”。

原因:Self-Instruct 用 LLM 生成时,LLM 会不可避免地产生幻觉,且 Self-Instruct 不做事实校验。

修法:多层过滤 + 关键事实必须外部校验。

# hallucination_filter.py
HALLUCINATION_KEYWORDS = [
    "据说", "据传", "有研究称",  # 模糊引用
    "1984 年苹果公司发布",          # 可校验事实,需查证
]

def likely_hallucination(output: str) -> bool:
    # 简化:模糊引用 + 数字断言 = 高风险
    has_vague = any(k in output for k in HALLUCINATION_KEYWORDS)
    has_number = bool(re.search(r"\d{4}\s*年", output))
    return has_vague and has_number

坑 3:多轮对话角色标签错乱

症状:训练后模型偶尔会把 USER 说的话当 ASSISTANT 输出,或者 SYSTEM 提示语出现在回答中。

原因:ShareGPT 转 ChatML 时角色映射错了,或 dataset 本身有脏数据(“human” 和 “user” 混用)。

修法:强制统一角色枚举 + 启动时校验。

# role_validate.py
ALLOWED_ROLES = {"system", "user", "assistant"}

def validate_and_fix(row: dict) -> dict:
    for m in row["messages"]:
        assert m["role"] in ALLOWED_ROLES, f"非法 role: {m['role']}"
        # 修复 ShareGPT 的 human/gpt
        if m["role"] == "human": m["role"] = "user"
        if m["role"] == "gpt":   m["role"] = "assistant"
    # 保证首条 user/system 前无 assistant
    assert row["messages"][0]["role"] != "assistant"
    # 保证 system 在最前(可一条)
    sys_msgs = [m for m in row["messages"] if m["role"] == "system"]
    assert len(sys_msgs) <= 1
    return row

坑 4:中英混杂 + 编码乱

症状:训练数据读入后报错 UnicodeDecodeError,或训练后模型输出”中文里嵌英文 tag”。

原因:文件是 UTF-8 BOM 或 GBK;字段里有残留 HTML tag、Markdown 控制符、全角半角混用。

修法:统一编码 + 规范化文本。

# normalize_text.py
import codecs
import re
import unicodedata

def read_file_safely(path: str) -> list[str]:
    # 自动剥 BOM
    with open(path, "rb") as f:
        raw = f.read()
    if raw.startswith(codecs.BOM_UTF8):
        raw = raw[3:]
    return raw.decode("utf-8", errors="replace").splitlines()

def normalize_text(text: str) -> str:
    # 全角 → 半角(仅标点)
    text = text.translate(str.maketrans(
        "，。！？；：" " ,.!?;:"
    ))
    # 去控制字符
    text = "".join(c for c in text if unicodedata.category(c)[0] != "C" or c == "\n")
    # 规范化空白
    text = re.sub(r"[ \t]+", " ", text)
    return text.strip()

坑 5:训练数据泄露(测试集出现在训练集)

症状:train loss 正常,test loss 极低,以为模型学得神;但线上效果糟糕。

原因:清洗时只对 train 内部去重,没考虑 train ∩ test。或者 Evol-Instruct 进化出的样本恰好和原始 test 高度相似。

修法:去重阶段就要 union(train, dev, test)一起做,并记录每条样本的 hash 到 split_manifest.json。

# leak_check.py
def check_leak(train_path: str, test_path: str):
    train_hashes = set()
    with open(train_path, encoding="utf-8") as f:
        for line in f:
            row = json.loads(line)
            h = hashlib.md5(
                (row["instruction"] + row["output"]).encode()
            ).hexdigest()
            train_hashes.add(h)
    leaks = []
    with open(test_path, encoding="utf-8") as f:
        for line in f:
            row = json.loads(line)
            h = hashlib.md5(
                (row["instruction"] + row["output"]).encode()
            ).hexdigest()
            if h in train_hashes:
                leaks.append(row)
    print(f"leak count: {len(leaks)}")  # 应为 0
    return leaks

坑 6:数据量少硬上微调(< 1K)

症状:”我用 500 条数据微调了 Qwen-7B,效果反而变差了”。

原因:数据量小于 1K 时,LoRA 也容易过拟合;模型会”灾难性遗忘”原始能力。

修法:数据 < 1K,优先 Prompt / RAG / Few-shot;1K-5K 才考虑 LoRA;5K+ 再考虑全量。

# decision.py — 数据量 vs 方案
def recommend_strategy(n: int, has_gpu: bool) -> str:
    if n < 1000:
        return "Prompt + Few-shot + RAG(不微调)"
    if n < 5000:
        return "LoRA r=8-16, 1-2 epoch" if has_gpu else "Prompt + RAG"
    if n < 50000:
        return "LoRA r=32-64, 2-3 epoch 或全量"
    return "全量 SFT, 1-2 epoch"

10. 数据准备 Checklist(15 项)

数据来源

数据来源合法,有授权 / 公开协议 / 自有版权
标注规范 Schema 已评审,术语无歧义

质量保障

全部数据 UTF-8 无 BOM,JSONL 行尾换行 \n
启发式过滤(长度 / 重复行 / 模板污染 / 中文比例)
近似去重(MinHash 阈值 0.85 或 Embedding 0.92)
PII 脱敏(正则 + Presidio 双保险)
LLM-as-Judge 抽样 ≥ 5%,过滤 < 3 分
标注一致性 Kappa ≥ 0.7(2 人 ≥ 0.6)

规模与平衡

数据规模符合任务需求(参考第 8 节表)
各领域 / 类型比例合理(无单一来源 > 60%)
train / dev / test 划分,无 hash 交集

格式与可复现

转目标训练框架格式(Alpaca / ShareGPT / ChatML)
验证 token 化后能正确还原
data_card.yaml 写完(来源 / 规模 / 已知问题)
split_manifest.json 记录各 split 的 hash

11. 微调数据规模速查表

任务难度	数据量	epoch	推荐增强方法
简单分类	500-2K	3-5	无需增强
领域问答	2K-10K	2-3	Self-Instruct
多轮对话	10K-50K	1-2	ShareGPT + 过滤
复杂推理	50K-200K	2-3	Evol-Instruct + Self-Instruct
通用助手	200K-1M	1-2	多源融合

12. 选型口诀(再背一遍)

数据少别硬上,Prompt / RAG 先试一趟。
数据脏毁一切,去重过滤比加量更重要。
质量先于数量,1K 高质胜过 1M 灌水。

自检报告

文件大小:略(运行 ls -la 查看)
行数:略(运行 wc -l 查看)
字符数:略(运行 wc -c 查看)
代码块数:30+ 处(标注 3 / 清洗 6 / 增强 5 / 格式 5 / 评估 2 / 其它 ≥ 9)
实战案例:4 个
踩坑:6 个
关键词命中:
- SFT:出现于章节 1.2 / 2.1 / 12 等
- Alpaca:出现于 1 / 2.1 / 6.2 / 全文
- Self-Instruct:出现于 2.1 / 5.1 / 案例 2
- Evol-Instruct:出现于 2.1 / 5.2 / 案例 3
- ShareGPT:出现于 2.3 / 6.3
- PII:出现于 3.4 / 4.3 / 案例 4
- Presidio:出现于 4.3.2 / 案例 4
- MinHash:出现于 4.1.2 / 4.1.4 / 5.1.3 / 案例 2

说明 · 本站内容均为学习笔记与经验总结,所有菜谱与技法请结合实际食材、季节与个人口味灵活调整。涉及生食、营养与健康的内容仅供参考,特殊体质或疾病请咨询专业营养师/医生。