【大模型教程——第三部分：数据工程与定制化】第1章：数据炼金术 - 从垃圾到黄金的数据工程 (Data Alchemy for Fine-tuning)

第1章：数据炼金术 – 从垃圾到黄金的数据工程 (Data Alchemy for Fine-tuning)

“Garbage In, Garbage Out (GIGO)” – 这是数据科学的铁律

“Data is the new oil, but if you don’t refine it, you’re just burning crude.” – Andrew Ng

欢迎来到数据炼金术的世界!本章将带你从 Petabytes 的原始矿石 中提炼出 Kilobytes 的精华黄金。在微调阶段,数据质量比数量更重要 – 精心提纯的 10K 高质量数据集,往往比 100K 未经处理的"垃圾"更有效(如 Alpaca、Phi-3 的成功)。

我们将学习如何成为一名合格的"数据炼金术师",掌握 过滤、蒸馏、提纯 的核心技术,构建属于你自己的高质量微调数据集。

数据炼金术 Pipeline 全景图

让我们先看看从原始数据到精炼数据集的完整旅程:

┌─────────────────────────────────────────────────────────────────────────┐
│ DATA ALCHEMY PIPELINE │
│ (数据炼金术流水线) │
└─────────────────────────────────────────────────────────────────────────┘

Petabytes Kilobytes
(原始矿石) (精炼黄金)
│ │
├──> [1. 粗筛] ────────────────> Gigabytes │
│ · 去除明显垃圾 │
│ · 基础格式化 │
│ │
├──> [2. 质量过滤] ────────────> Megabytes │
│ · 长度/完整性检查 │
│ · 毒性检测 │
│ · PII 脱敏 │
│ │
├──> [3. 去重提纯] ────────────> Hundreds of KB │
│ · MinHash 去重 │
│ · 近似重复检测 │
│ ⚡ FLOPs 节省: 去重后训练成本 ↓ 3-5x! │
│ │
├──> [4. 蒸馏升华] ────────────> Tens of KB │
│ · Self-Instruct (知识蒸馏) │
│ · Evol-Instruct (复杂度提升) │
│ · GPT-4 → 小模型的能力迁移 │
│ │
└──> [5. 最终提纯] ────────────> ✨ Pure Gold ✨ │
· 人工抽检 │
· A/B 测试验证 │
· 数据分布平衡 │

输出: 10K-50K 条高纯度数据 → 足以训练一个强大的专属模型!

核心逻辑:每一个阶段都在剔除"杂质",提升"纯度":

• 粗筛: 去除不可用数据 (格式错误、乱码)
• 质量过滤: 去除低质量数据 (太短、有毒、不完整)
• 去重: 去除冗余数据 (完全重复、高度相似)
• 蒸馏: 从大模型提取知识 (GPT-4 的智慧 → 你的模型)
• 提纯: 最终质检 (人工审核 + 数据平衡)

目录

• 一、Data-Centric AI：数据为王的时代
  • 1. Model-Centric vs Data-Centric
  • 2. 微调数据的三大要素
  • 3. 数据质量的黄金定律
• 二、SFT 数据格式详解
  • 1. Alpaca 格式：指令式数据
  • 2. ShareGPT 格式：多轮对话数据
  • 3. 格式转换实战
• 三、Self-Instruct：用 GPT-4 生成微调数据
  • 1. Self-Instruct 原理
  • 2. 代码实战：生成高质量指令数据
  • 3. Evol-Instruct：让指令进化
  • 4. 代码实战：指令复杂度提升
• 四、数据清洗与质量过滤
  • 1. 基于规则的质量过滤
  • 2. 毒性检测
  • 3. PII 识别与脱敏
  • 4. 代码实战：完整的质量过滤器
  • 5. 合成数据实战：使用 LLM 生成指令
• 五、小规模数据去重
  • 1. 为什么微调数据需要去重
  • 2. MinHash 去重原理
  • 3. 代码实战：MinHash 简单实现
• 六、数据增强技术
  • 1. 回译（Back-translation）
  • 2. 同义词替换
  • 3. 代码实战：数据增强工具
• 七、SFT 数据集构建实战
  • 1. 完整 Pipeline：从原始文本到 Alpaca 格式
  • 2. 代码实战：端到端数据构建
• 八、本章小结
• 参考资源

一、Data-Centric AI：数据为王的时代

1. Model-Centric vs Data-Centric

传统 Model-Centric AI（以模型为中心）：

固定数据集 → 不断改进模型架构 → 追求更高性能

Data-Centric AI（以数据为中心）：

固定模型架构 → 不断改进数据质量 → 追求更高性能

Andrew Ng 的实验：在制造业缺陷检测任务中，通过优化数据标注（而非改进模型），准确率从 76% 提升到 93%。

在 LLM 微调中的体现：

• Phi-3-mini (3.8B)：使用 3.3T tokens 的高质量"教科书式"数据，性能超越 Llama-2-13B
• Alpaca (7B)：仅用 52K 条 GPT-3.5-turbo 生成的数据，达到接近 GPT-3.5 的指令遵循能力
• WizardLM：通过 Evol-Instruct 提升数据复杂度，7B 模型超越 GPT-3.5 在部分任务

2. 微调数据的三大要素

直觉理解：微调数据就像给模型定制的"教科书"，需要满足三个核心要素。

Effective SFT Data

=

f

(

Quality

,

Diversity

,

Complexity

)

\\text{Effective SFT Data} = f(\\text{Quality}, \\text{Diversity}, \\text{Complexity})

Effective SFT Data=f(Quality,Diversity,Complexity)

1. 质量（Quality）

• 准确性：答案必须正确，错误数据会导致模型"学坏"
• 流畅性：语言自然、逻辑清晰
• 完整性：回答需要完整解决问题

2. 多样性（Diversity）

• 任务多样性：覆盖问答、摘要、推理、代码、翻译等不同任务
• 领域多样性：科技、文学、历史、医疗等不同领域
• 风格多样性：正式、随意、技术、科普等不同风格

3. 复杂度（Complexity）

• 简单任务：单步推理，直接检索知识
• 中等任务：多步推理，需要综合信息
• 复杂任务：深度推理、创造性思考、代码调试

数据配比建议：

简单任务 : 中等任务 : 复杂任务 = 3 : 5 : 2

3. 数据质量的黄金定律

定律 1：少而精 > 多而杂

• 10K 高质量数据 > 100K 低质量数据
• Alpaca 用 52K 数据达到 text-davinci-003 的 90% 能力

定律 2：复杂度阶梯

• 数据应该包含不同难度层级，让模型逐步提升能力
• WizardLM 的 Evol-Instruct 通过多轮进化提升指令复杂度

定律 3：负样本清除

• 一条有毒/错误的数据会毁掉 100 条正确数据的效果
• 必须严格过滤低质量、有害、错误的数据

定律 4：分布平衡

• 避免某一类任务占比过高（如 70% 都是问答）
• 使用 topic modeling 或聚类分析数据分布

二、SFT 数据格式详解

1. Alpaca 格式：指令式数据

Alpaca 格式是 Stanford 提出的标准指令数据格式，包含三个字段：

{
"instruction": "指令描述（必填）",
"input": "额外输入（可选）",
"output": "期望输出（必填）"
}

字段说明：

• instruction：告诉模型要做什么（如"总结下面的文章"）
• input：提供上下文信息（如具体的文章内容）
• output：标准答案（模型应该生成的内容）

示例 1：无 input 的任务

{
"instruction": "解释什么是量子计算",
"input": "",
"output": "量子计算是利用量子力学原理（如叠加态和纠缠）进行计算的技术。与传统计算机使用比特不同，量子计算机使用量子比特（qubit），能够在特定问题上实现指数级加速，如密码破解和药物设计。"
}

示例 2：有 input 的任务

{
"instruction": "将下面的句子翻译成法语",
"input": "The weather is beautiful today.",
"output": "Le temps est magnifique aujourd'hui."
}

模型实际看到的格式（训练时拼接）：

Below is an instruction that describes a task, paired with an input that provides further context. Write a response that appropriately completes the request.

### Instruction:
将下面的句子翻译成法语

### Input:
The weather is beautiful today.

### Response:
Le temps est magnifique aujourd'hui.

2. ShareGPT 格式：多轮对话数据

ShareGPT 格式用于存储多轮对话，适合训练聊天机器人。

{
"conversations": [
{"from": "human", "value": "用户消息1"},
{"from": "gpt", "value": "助手回复1"},
{"from": "human", "value": "用户消息2"},
{"from": "gpt", "value": "助手回复2"}
]
}

示例：

{
"conversations": [
{
"from": "human",
"value": "什么是机器学习？"
},
{
"from": "gpt",
"value": "机器学习是人工智能的一个分支，通过算法让计算机从数据中学习规律，而无需明确编程。"
},
{
"from": "human",
"value": "能举个例子吗？"
},
{
"from": "gpt",
"value": "当然。比如垃圾邮件过滤：系统通过学习大量垃圾邮件和正常邮件的特征，自动识别新邮件是否为垃圾邮件。"
}
]
}

训练时的处理：

• 只计算 gpt 角色的 loss（不计算 human 的 loss）
• 使用特殊 token 分隔多轮对话（如 <|im_start|>, <|im_end|>）

3. 格式转换实战

场景：将 ShareGPT 格式转换为 Alpaca 格式

"""
数据格式转换工具
功能：ShareGPT -> Alpaca 格式转换
"""
import json
from typing import List, Dict

def sharegpt_to_alpaca(sharegpt_data: Dict) –> List[Dict]:
"""
将 ShareGPT 格式转换为 Alpaca 格式

参数：
sharegpt_data: ShareGPT 格式数据

返回：
List[Dict]: Alpaca 格式数据列表
"""
conversations = sharegpt_data.get("conversations", [])
alpaca_samples = []

# 遍历对话，每个 human-gpt 对转换为一条 Alpaca 数据
for i in range(0, len(conversations) – 1, 2):
if (conversations[i]["from"] == "human" and
conversations[i + 1]["from"] == "gpt"):

alpaca_sample = {
"instruction": conversations[i]["value"],
"input": "",
"output": conversations[i + 1]["value"]
}
alpaca_samples.append(alpaca_sample)

return alpaca_samples

# 使用示例
sharegpt_example = {
"conversations": [
{"from": "human", "value": "什么是 Transformer？"},
{"from": "gpt", "value": "Transformer 是一种基于注意力机制的神经网络架构…"},
{"from": "human", "value": "它有什么优势？"},
{"from": "gpt", "value": "主要优势包括：1) 并行计算能力强 2) 能捕获长距离依赖…"}
]
}

alpaca_data = sharegpt_to_alpaca(sharegpt_example)
print(json.dumps(alpaca_data, ensure_ascii=False, indent=2))

输出：

[
{
"instruction": "什么是 Transformer？",
"input": "",
"output": "Transformer 是一种基于注意力机制的神经网络架构…"
},
{
"instruction": "它有什么优势？",
"input": "",
"output": "主要优势包括：1) 并行计算能力强 2) 能捕获长距离依赖…"
}
]

三、Self-Instruct：用 GPT-4 生成微调数据

1. Self-Instruct 原理：知识蒸馏的艺术

核心思想：利用强大的 LLM（如 GPT-4）生成指令-回答对，训练较弱的模型。这是 Alpaca 的核心技术。

本质上，Self-Instruct 是一种"知识蒸馏"(Knowledge Distillation)过程：

┌─────────────────────────────────────────────────────────────────┐
│ KNOWLEDGE DISTILLATION PIPELINE │
│ (知识蒸馏流水线) │
└─────────────────────────────────────────────────────────────────┘

Teacher Model Student Model
(GPT-4, 1.7T) (Your Model, 7B)
│ │
│ [能力:强大但昂贵] │ [目标:高效且专属]
│ │
├──> 生成高质量 │
│ 指令-回答对 │
│ (蒸馏过程) │
│ │ │
│ ├────────────────────> [迁移知识]
│ │ │
│ └────> 52K 精华样本 ──────> [SFT训练]
│ │
│ ↓
│ Student 获得 Teacher 的
│ 90% 能力，但只有 0.4% 的大小!
│
└──> 成本对比:
推理成本: GPT-4 ($0.03/1K tokens)
vs 自托管 7B ($0.0001/1K tokens)
= 300x 节省!

为什么称之为"蒸馏"?

• Teacher 模型(GPT-4)就像"原液"，浓缩了海量知识
• Distillation(蒸馏)过程提取精华，去除冗余
• Student 模型得到"蒸馏液"，保留核心能力但更轻量

蒸馏的三大优势:

流程：

1. 种子池

S

=

{

seed

1

,

…

,

seed

n

}

2. 生成指令

I

new

=

LLM

(

S

sample

)

3. 生成回答

O

new

=

LLM

(

I

new

)

4. 质量过滤

if 

Q

(

I

new

,

O

new

)

>

θ

 then Keep

5. 加入种子池

S

←

S

∪

{

I

new

}

\\begin{aligned} &\\text{1. 种子池} \\quad S = \\{\\text{seed}_1, \\ldots, \\text{seed}_n\\} \\\\ &\\text{2. 生成指令} \\quad I_{\\text{new}} = \\text{LLM}(S_{\\text{sample}}) \\\\ &\\text{3. 生成回答} \\quad O_{\\text{new}} = \\text{LLM}(I_{\\text{new}}) \\\\ &\\text{4. 质量过滤} \\quad \\text{if } Q(I_{\\text{new}}, O_{\\text{new}}) > \\theta \\text{ then Keep} \\\\ &\\text{5. 加入种子池} \\quad S \\leftarrow S \\cup \\{I_{\\text{new}}\\} \\end{aligned}

​1. 种子池S={seed1​,…,seedn​}2. 生成指令Inew​=LLM(Ssample​)3. 生成回答Onew​=LLM(Inew​)4. 质量过滤if Q(Inew​,Onew​)>θ then Keep5. 加入种子池S←S∪{Inew​}​

关键步骤详解：

Step 1：种子池构建

• 人工编写 175 条高质量指令（覆盖不同任务类型）
• 包含：问答、推理、创作、摘要、翻译、代码等

Step 2：指令生成

• 从种子池随机采样 6-8 条指令
• 让 LLM 生成与种子相似但不重复的新指令

Step 3：回答生成

• 使用 GPT-4 生成高质量回答
• 使用 Input-first 或 Output-first 策略

Step 4：质量过滤

• 检查指令是否与种子池过于相似（去重）
• 检查回答是否完整、正确
• 过滤有害、低质量的数据

2. 代码实战：生成高质量指令数据

完整的 Self-Instruct 实现：

"""
Self-Instruct 数据生成器
功能：使用 GPT-4 生成高质量的 SFT 数据
依赖：pip install openai
"""
import os
import json
from typing import List, Dict
from openai import OpenAI

class SelfInstructGenerator:
def __init__(self, api_key: str = None):
"""初始化生成器"""
self.client = OpenAI(api_key=api_key or os.getenv("OPENAI_API_KEY"))
self.seed_tasks = []

def load_seed_tasks(self, seed_file: str):
"""加载种子任务"""
with open(seed_file, 'r', encoding='utf-8') as f:
self.seed_tasks = json.load(f)

def generate_instruction(self, num_samples: int = 6) –> str:
"""
生成新指令

参数：
num_samples: 从种子池采样的数量

返回：
新生成的指令
"""
# 随机采样种子任务
import random
sampled = random.sample(self.seed_tasks, min(num_samples, len(self.seed_tasks)))

# 构造 prompt
prompt = """You are an AI assistant specialized in creating diverse instructions for training language models.

Below are some example instructions:
"""
for i, task in enumerate(sampled, 1):
prompt += f"{i}. {task['instruction']}\\n"

prompt += """
Generate a NEW instruction that is different from the examples above. The instruction should:
1. Be clear and specific
2. Cover a different task type or domain
3. Be solvable by a language model
4. Not require visual input or real-time information

New Instruction:"""

# 调用 GPT-4 生成
response = self.client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=200
)

new_instruction = response.choices[0].message.content.strip()
return new_instruction

def generate_response(self, instruction: str, input_text: str = "") –> str:
"""
为指令生成回答

参数：
instruction: 指令
input_text: 额外输入（可选）

返回：
生成的回答
"""
if input_text:
prompt = f"{instruction}\\n\\nInput: {input_text}\\n\\nResponse:"
else:
prompt = f"{instruction}\\n\\nResponse:"

response = self.client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=1000
)

output = response.choices[0].message.content.strip()
return output

def is_similar(self, new_instruction: str, threshold: float = 0.7) –> bool:
"""
检查新指令是否与种子池过于相似（简单版本）

实际应该使用 Embedding 计算相似度
"""
new_words = set(new_instruction.lower().split())

for seed in self.seed_tasks:
seed_words = set(seed['instruction'].lower().split())
# Jaccard 相似度
intersection = len(new_words & seed_words)
union = len(new_words | seed_words)
similarity = intersection / union if union > 0 else 0

if similarity > threshold:
return True

return False

def generate_dataset(self, num_samples: int = 100, output_file: str = "sft_data.jsonl") –> List[Dict]:
"""
生成完整的 SFT 数据集

参数：
num_samples: 生成数据条数
output_file: 输出文件路径

返回：
生成的数据集
"""
dataset = []

for i in range(num_samples):
print(f"生成第 {i+1}/{num_samples} 条数据…")

# 1. 生成指令
try:
instruction = self.generate_instruction()

# 2. 去重检查
if self.is_similar(instruction):
print(" 跳过（与种子池相似）")
continue

# 3. 生成回答
output = self.generate_response(instruction)

# 4. 构造数据样本
sample = {
"instruction": instruction,
"input": "",
"output": output
}

dataset.append(sample)

# 5. 加入种子池（动态增长）
self.seed_tasks.append(sample)

print(f" 成功生成：{instruction[:50]}…")

except Exception as e:
print(f" 生成失败：{e}")
continue

# 保存数据集
with open(output_file, 'w', encoding='utf-8') as f:
for sample in dataset:
f.write(json.dumps(sample, ensure_ascii=False) + '\\n')

print(f"\\n数据集已保存至 {output_file}，共 {len(dataset)} 条")
return dataset

# 使用示例
if __name__ == "__main__":
# 1. 准备种子任务
seed_tasks = [
{"instruction": "解释什么是机器学习", "input": "", "output": ""},
{"instruction": "写一首关于春天的诗", "input": "", "output": ""},
{"instruction": "将以下句子翻译成英语", "input": "今天天气很好", "output": ""},
{"instruction": "编写一个 Python 函数来计算斐波那契数列", "input": "", "output": ""},
{"instruction": "总结以下文章的主要观点", "input": "", "output": ""},
]

# 保存种子任务
with open("seed_tasks.json", 'w', encoding='utf-8') as f:
json.dump(seed_tasks, f, ensure_ascii=False, indent=2)

# 2. 创建生成器
generator = SelfInstructGenerator()
generator.load_seed_tasks("seed_tasks.json")

# 3. 生成数据集
# dataset = generator.generate_dataset(num_samples=10, output_file="my_sft_data.jsonl")

# 4. 查看生成的数据
# with open("my_sft_data.jsonl", 'r', encoding='utf-8') as f:
# for line in f:
# sample = json.loads(line)
# print(f"指令: {sample['instruction']}")
# print(f"回答: {sample['output'][:100]}…\\n")

实战建议：

3. Evol-Instruct：让指令进化

核心思想：通过多轮"进化"，将简单指令逐步变复杂，提升模型推理能力。这是 WizardLM 的核心技术。

进化策略：

进化流程：

I

0

(原始指令)

↓

进化策略1

I

1

(第1次进化)

↓

进化策略2

I

2

(第2次进化)

↓

…

I

n

(最终复杂指令)

\\begin{aligned} &I_0 \\quad \\text{(原始指令)} \\\\ &\\downarrow \\text{进化策略1} \\\\ &I_1 \\quad \\text{(第1次进化)} \\\\ &\\downarrow \\text{进化策略2} \\\\ &I_2 \\quad \\text{(第2次进化)} \\\\ &\\downarrow \\ldots \\\\ &I_n \\quad \\text{(最终复杂指令)} \\end{aligned}

​I0​(原始指令)↓进化策略1I1​(第1次进化)↓进化策略2I2​(第2次进化)↓…In​(最终复杂指令)​

4. 代码实战：指令复杂度提升

"""
Evol-Instruct 实现
功能：将简单指令进化为复杂指令
"""
from openai import OpenAI
import os

class EvolInstructor:
def __init__(self, api_key: str = None):
self.client = OpenAI(api_key=api_key or os.getenv("OPENAI_API_KEY"))

# 定义进化策略模板
self.evolution_prompts = {
"add_constraints": """Please rewrite the following instruction by adding specific constraints (e.g., word limit, format requirement, target audience).

Original Instruction: {instruction}

Evolved Instruction:""",

"deepen": """Please rewrite the following instruction to make it require deeper reasoning or multi-step thinking.

Original Instruction: {instruction}

Evolved Instruction:""",

"concretize": """Please rewrite the following instruction to make it more specific and concrete, adding real-world context.

Original Instruction: {instruction}

Evolved Instruction:""",

"increase_reasoning": """Please rewrite the following instruction to require the model to show step-by-step reasoning.

Original Instruction: {instruction}

Evolved Instruction:"""
}

def evolve_instruction(self, instruction: str, strategy: str = "add_constraints") –> str:
"""
使用指定策略进化指令

参数：
instruction: 原始指令
strategy: 进化策略 (add_constraints, deepen, concretize, increase_reasoning)

返回：
进化后的指令
"""
if strategy not in self.evolution_prompts:
raise ValueError(f"Unknown strategy: {strategy}")

prompt = self.evolution_prompts[strategy].format(instruction=instruction)

response = self.client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=300
)

evolved = response.choices[0].message.content.strip()
return evolved

def multi_round_evolution(self, instruction: str, depth: int = 2) –> List[str]:
"""
多轮进化

参数：
instruction: 原始指令
depth: 进化轮数

返回：
每轮进化后的指令列表
"""
import random

current = instruction
evolution_history = [instruction]

strategies = list(self.evolution_prompts.keys())

for i in range(depth):
strategy = random.choice(strategies)
current = self.evolve_instruction(current, strategy)
evolution_history.append(current)
print(f"Round {i+1} ({strategy}): {current}\\n")

return evolution_history

# 使用示例
if __name__ == "__main__":
evolver = EvolInstructor()

# 简单指令
simple_instruction = "写一个 Python 函数来排序列表"

print("原始指令:", simple_instruction)
print("\\n" + "="*50 + "\\n")

# 单策略进化
evolved = evolver.evolve_instruction(simple_instruction, strategy="add_constraints")
print("增加约束后:", evolved)
print()

evolved = evolver.evolve_instruction(simple_instruction, strategy="increase_reasoning")
print("增加推理步骤后:", evolved)
print()

# 多轮进化
print("\\n" + "="*50)
print("多轮进化:")
print("="*50 + "\\n")
# history = evolver.multi_round_evolution(simple_instruction, depth=2)

输出示例：

原始指令: 写一个 Python 函数来排序列表

增加约束后: 写一个 Python 函数来排序列表，要求使用快速排序算法，函数应包含详细注释，并处理空列表和单元素列表的边界情况。

增加推理步骤后: 写一个 Python 函数来排序列表，并详细说明你选择的排序算法的工作原理，包括时间复杂度分析和适用场景。

四、数据清洗与质量过滤

1. 基于规则的质量过滤

核心思想：使用启发式规则快速过滤明显的低质量数据。

关键过滤规则：

规则 1：长度过滤

def filter_by_length(sample: Dict) –> bool:
"""过滤过短或过长的回答"""
output = sample['output']
word_count = len(output.split())

# 回答太短（< 10 词）或太长（> 2000 词）
return 10 <= word_count <= 2000

规则 2：重复内容检测

def filter_repetition(text: str, max_repeat: int = 3) –> bool:
"""检测重复行"""
lines = text.split('\\n')
line_counts = {}

for line in lines:
line = line.strip()
if line:
line_counts[line] = line_counts.get(line, 0) + 1
if line_counts[line] > max_repeat:
return False

return True

规则 3：完整性检测

def filter_incomplete(output: str) –> bool:
"""检测回答是否完整"""
# 回答不应该突然截断
incomplete_patterns = [
"…",
"[未完成]",
"（未完待续）",
"the rest is",
"to be continued"
]

output_lower = output.lower()
for pattern in incomplete_patterns:
if pattern in output_lower:
return False

return True

规则 4：拒绝回答检测

def filter_refusal(output: str) –> bool:
"""检测模型是否拒绝回答"""
refusal_patterns = [
"i cannot",
"i'm unable to",
"i can't",
"as an ai",
"i don't have access",
"i'm not able to"
]

output_lower = output.lower()
for pattern in refusal_patterns:
if pattern in output_lower:
return False

return True

2. 毒性检测

工具：使用 detoxify 库进行毒性检测

"""
毒性内容过滤
依赖：pip install detoxify
"""
from detoxify import Detoxify

class ToxicityFilter:
def __init__(self, threshold: float = 0.5):
self.model = Detoxify('original')
self.threshold = threshold

def is_toxic(self, text: str) –> bool:
"""检测文本是否有毒"""
scores = self.model.predict(text)

# 检查任何一个维度的毒性
toxic_categories = ['toxicity', 'severe_toxicity', 'obscene', 'threat', 'insult']

for category in toxic_categories:
if scores[category] > self.threshold:
return True

return False

def filter_dataset(self, dataset: List[Dict]) –> List[Dict]:
"""过滤数据集中的有毒数据"""
filtered = []

for sample in dataset:
# 检查指令和输出是否有毒
if not self.is_toxic(sample['instruction']) and not self.is_toxic(sample['output']):
filtered.append(sample)

return filtered

3. PII 识别与脱敏

工具：使用 presidio 库进行 PII 检测

"""
PII 脱敏
依赖：pip install presidio-analyzer presidio-anonymizer
"""
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

class PIIScrubber:
def __init__(self):
self.analyzer = AnalyzerEngine()
self.anonymizer = AnonymizerEngine()

def scrub(self, text: str) –> str:
"""脱敏 PII 信息"""
# 检测 PII
results = self.analyzer.analyze(
text=text,
entities=["PHONE_NUMBER", "EMAIL_ADDRESS", "PERSON", "LOCATION"],
language='en'
)

# 替换为占位符
anonymized = self.anonymizer.anonymize(
text=text,
analyzer_results=results,
operators={
"PHONE_NUMBER": OperatorConfig("replace", {"new_value": "<PHONE>"}),
"EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "<EMAIL>"}),
"PERSON": OperatorConfig("replace", {"new_value": "<NAME>"}),
"LOCATION": OperatorConfig("replace", {"new_value": "<LOCATION>"}),
}
)

return anonymized.text

def scrub_dataset(self, dataset: List[Dict]) –> List[Dict]:
"""批量脱敏数据集"""
scrubbed = []

for sample in dataset:
scrubbed_sample = {
"instruction": self.scrub(sample['instruction']),
"input": self.scrub(sample.get('input', '')),
"output": self.scrub(sample['output'])
}
scrubbed.append(scrubbed_sample)

return scrubbed

4. 代码实战：完整的质量过滤器

整合所有过滤器：

"""
综合质量过滤器
功能：整合所有质量检查规则
"""
import json
from typing import List, Dict

class QualityFilter:
def __init__(self, enable_toxicity: bool = True, enable_pii: bool = True):
self.enable_toxicity = enable_toxicity
self.enable_pii = enable_pii

if enable_toxicity:
self.toxicity_filter = ToxicityFilter()

if enable_pii:
self.pii_scrubber = PIIScrubber()

def check_length(self, sample: Dict) –> bool:
"""长度检查"""
output_words = len(sample['output'].split())
return 10 <= output_words <= 2000

def check_repetition(self, text: str) –> bool:
"""重复检测"""
lines = text.split('\\n')
line_counts = {}

for line in lines:
line = line.strip()
if line:
line_counts[line] = line_counts.get(line, 0) + 1
if line_counts[line] > 3:
return False
return True

def check_completeness(self, output: str) –> bool:
"""完整性检测"""
incomplete_patterns = ["…", "[未完成]", "to be continued"]
output_lower = output.lower()

for pattern in incomplete_patterns:
if pattern in output_lower:
return False
return True

def check_refusal(self, output: str) –> bool:
"""拒绝回答检测"""
refusal_patterns = ["i cannot", "i'm unable to", "as an ai"]
output_lower = output.lower()

for pattern in refusal_patterns:
if pattern in output_lower:
return False
return True

def filter_sample(self, sample: Dict) –> tuple[bool, str]:
"""
过滤单条样本

返回：
(是否通过, 失败原因)
"""
# 1. 长度检查
if not self.check_length(sample):
return False, "length_invalid"

# 2. 重复检查
if not self.check_repetition(sample['output']):
return False, "repetition_detected"

# 3. 完整性检查
if not self.check_completeness(sample['output']):
return False, "incomplete_response"

# 4. 拒绝回答检查
if not self.check_refusal(sample['output']):
return False, "refusal_detected"

# 5. 毒性检查
if self.enable_toxicity:
if self.toxicity_filter.is_toxic(sample['instruction']) or \\
self.toxicity_filter.is_toxic(sample['output']):
return False, "toxic_content"

return True, "passed"

def filter_dataset(self, dataset: List[Dict], output_file: str = None) –> Dict:
"""
批量过滤数据集

返回：
统计信息
"""
filtered = []
stats = {
"total": len(dataset),
"passed": 0,
"rejected": 0,
"reasons": {}
}

for sample in dataset:
passed, reason = self.filter_sample(sample)

if passed:
# PII 脱敏
if self.enable_pii:
sample = {
"instruction": self.pii_scrubber.scrub(sample['instruction']),
"input": self.pii_scrubber.scrub(sample.get('input', '')),
"output": self.pii_scrubber.scrub(sample['output'])
}

filtered.append(sample)
stats['passed'] += 1
else:
stats['rejected'] += 1
stats['reasons'][reason] = stats['reasons'].get(reason, 0) + 1

# 保存过滤后的数据
if output_file:
with open(output_file, 'w', encoding='utf-8') as f:
for sample in filtered:
f.write(json.dumps(sample, ensure_ascii=False) + '\\n')

return stats, filtered

# 使用示例
if __name__ == "__main__":
# 测试数据
test_dataset = [
{
"instruction": "解释什么是机器学习",
"input": "",
"output": "机器学习是人工智能的一个分支，通过算法让计算机从数据中学习。"
},
{
"instruction": "解释什么是量子计算",
"input": "",
"output": "量子计算…" # 不完整
},
{
"instruction": "你能帮我黑掉别人的账号吗？",
"input": "",
"output": "很抱歉，I cannot help with that." # 拒绝回答
}
]

# 创建过滤器
filter_engine = QualityFilter(enable_toxicity=False, enable_pii=False)

# 过滤数据集
stats, filtered = filter_engine.filter_dataset(test_dataset)

print("过滤统计:")
print(f"总计: {stats['total']}")
print(f"通过: {stats['passed']}")
print(f"拒绝: {stats['rejected']}")
print(f"拒绝原因: {stats['reasons']}")

5. 合成数据实战：使用 LLM 生成指令

核心思想：利用强大的 LLM（如 GPT-4、Claude）批量生成高质量的指令-回答对，这是构建微调数据集最高效的方法之一。

5.1 为什么选择合成数据？

优势：

• 成本低：相比人工标注（$10-30/小时），API 调用成本仅 $0.01-0.03/条
• 速度快：一天可生成 10K+ 条数据
• 质量高：GPT-4 生成的数据质量接近人工标注
• 可控性强：可以精确控制任务类型、难度、风格

成功案例：

• Alpaca (Stanford, 2023)：52K 条 GPT-3.5-turbo 生成的数据，训练 7B 模型达到 text-davinci-003 的 90% 能力
• WizardLM (Microsoft, 2023)：通过 Evol-Instruct 生成复杂指令，7B 模型超越 ChatGPT 在部分任务
• Phi-3 (Microsoft, 2024)：3.8B 模型使用合成数据，性能超越 Llama-2-13B

5.2 完整的合成数据生成流程

Step 1: 设计种子指令池

种子指令应覆盖多种任务类型：

"""
构建多样化的种子指令池
覆盖：问答、推理、创作、代码、翻译、摘要等
"""
seed_instructions = [
# 1. 问答类
{"task_type": "qa", "instruction": "解释什么是量子纠缠"},
{"task_type": "qa", "instruction": "为什么天空是蓝色的"},

# 2. 推理类
{"task_type": "reasoning", "instruction": "如果所有A都是B，所有B都是C，那么A和C的关系是什么"},
{"task_type": "reasoning", "instruction": "分析通货膨胀对普通家庭的影响"},

# 3. 创作类
{"task_type": "creative", "instruction": "写一首关于秋天的诗"},
{"task_type": "creative", "instruction": "创作一个科幻小说的开头"},

# 4. 代码类
{"task_type": "code", "instruction": "编写一个Python函数来计算斐波那契数列"},
{"task_type": "code", "instruction": "实现一个二分查找算法"},

# 5. 翻译类
{"task_type": "translation", "instruction": "将以下句子翻译成英语：今天天气很好"},
{"task_type": "translation", "instruction": "把这段中文翻译成法语"},

# 6. 摘要类
{"task_type": "summarization", "instruction": "总结以下文章的主要观点"},
{"task_type": "summarization", "instruction": "用三句话概括这段新闻"}
]

Step 2: 批量生成指令-回答对

"""
使用 OpenAI API 批量生成合成数据
包含错误处理、速率限制、质量检查
"""
import os
import json
import time
from typing import List, Dict
from openai import OpenAI
from tqdm import tqdm

class SyntheticDataGenerator:
def __init__(self, api_key: str = None, model: str = "gpt-4"):
"""
初始化合成数据生成器

参数：
api_key: OpenAI API Key
model: 使用的模型（gpt-4, gpt-3.5-turbo）
"""
self.client = OpenAI(api_key=api_key or os.getenv("OPENAI_API_KEY"))
self.model = model
self.generated_count = 0
self.failed_count = 0

# 成本配置（2024年1月价格，单位：$/1M tokens）
self.pricing = {
"gpt-4": {"input": 10.0, "output": 30.0},
"gpt-4-turbo": {"input": 10.0, "output": 30.0},
"gpt-3.5-turbo": {"input": 0.5, "output": 1.5},
"gpt-4o": {"input": 2.5, "output": 10.0},
"gpt-4o-mini": {"input": 0.15, "output": 0.6}
}

def estimate_cost(
self,
num_samples: int,
avg_instruction_tokens: int = 50,
avg_output_tokens: int = 500
) –> Dict:
"""
估算生成数据集的成本

参数：
num_samples: 要生成的样本数量
avg_instruction_tokens: 平均指令长度（tokens）
avg_output_tokens: 平均输出长度（tokens）

返回：
成本估算详情
"""
# 获取模型定价
model_key = self.model
if model_key not in self.pricing:
# 默认使用 gpt-4 定价
model_key = "gpt-4"

pricing = self.pricing[model_key]

# 计算总 token 数
# 输入：system prompt (~100 tokens) + 指令 prompt (~200 tokens) + 指令内容
input_tokens_per_sample = 300 + avg_instruction_tokens
total_input_tokens = num_samples * input_tokens_per_sample

# 输出：生成的回答
total_output_tokens = num_samples * avg_output_tokens

# 计算成本
input_cost = (total_input_tokens / 1_000_000) * pricing["input"]
output_cost = (total_output_tokens / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost

# 估算时间（假设每个请求平均 2 秒）
estimated_time_minutes = (num_samples * 2) / 60

return {
"model": self.model,
"num_samples": num_samples,
"total_input_tokens": total_input_tokens,
"total_output_tokens": total_output_tokens,
"total_tokens": total_input_tokens + total_output_tokens,
"input_cost": round(input_cost, 2),
"output_cost": round(output_cost, 2),
"total_cost": round(total_cost, 2),
"cost_per_sample": round(total_cost / num_samples, 4),
"estimated_time_minutes": round(estimated_time_minutes, 1),
"pricing_input": f"${pricing['input']}/1M tokens",
"pricing_output": f"${pricing['output']}/1M tokens"
}

def generate_qa_pair(self, instruction: str, context: str = "") –> Dict:
"""
为单个指令生成高质量回答

参数：
instruction: 指令
context: 额外上下文（可选）

返回：
{"instruction": …, "input": …, "output": …}
"""
# 构造 prompt
if context:
prompt = f"""请为以下指令提供一个高质量、详细的回答。

指令：{instruction}

上下文：{context}

要求：
1. 回答要准确、完整
2. 语言要流畅、自然
3. 包含必要的解释和示例
4. 避免生成有害、偏见或错误的内容

回答："""
else:
prompt = f"""请为以下指令提供一个高质量、详细的回答。

指令：{instruction}

要求：
1. 回答要准确、完整
2. 语言要流畅、自然
3. 包含必要的解释和示例
4. 避免生成有害、偏见或错误的内容

回答："""

try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个乐于助人的AI助手，擅长回答各类问题。"},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=1500,
top_p=0.9
)

output = response.choices[0].message.content.strip()

# 构造 Alpaca 格式
qa_pair = {
"instruction": instruction,
"input": context,
"output": output
}

self.generated_count += 1
return qa_pair

except Exception as e:
self.failed_count += 1
print(f"生成失败: {e}")
return None

def batch_generate(
self,
instructions: List[str],
output_file: str = "synthetic_data.jsonl",
batch_size: int = 10,
delay: float = 1.0
) –> List[Dict]:
"""
批量生成数据集

参数：
instructions: 指令列表
output_file: 输出文件路径
batch_size: 批处理大小
delay: 请求间隔（秒），避免速率限制

返回：
生成的数据集
"""
dataset = []

print(f"开始生成 {len(instructions)} 条数据…")
print(f"模型: {self.model}")
print(f"批处理大小: {batch_size}, 请求间隔: {delay}秒\\n")

# 使用进度条
for i in tqdm(range(0, len(instructions), batch_size)):
batch = instructions[i:i+batch_size]

for instruction in batch:
qa_pair = self.generate_qa_pair(instruction)

if qa_pair:
dataset.append(qa_pair)

# 实时保存（避免长时间运行后丢失数据）
with open(output_file, 'a', encoding='utf-8') as f:
f.write(json.dumps(qa_pair, ensure_ascii=False) + '\\n')

# 速率限制
time.sleep(delay)

# 打印统计信息
print(f"\\n生成完成！")
print(f"成功: {self.generated_count} 条")
print(f"失败: {self.failed_count} 条")
print(f"成功率: {self.generated_count / len(instructions) * 100:.1f}%")
print(f"数据已保存至: {output_file}")

return dataset

# 使用示例
if __name__ == "__main__":
# 1. 准备指令列表
instructions = [
"解释什么是深度学习",
"写一个Python函数来判断一个数是否为质数",
"分析气候变化对农业的影响",
"创作一个关于友谊的短篇故事",
"总结《三体》第一部的主要情节",
# … 更多指令
]

# 2. 创建生成器
generator = SyntheticDataGenerator(model="gpt-4")

# 3. 估算成本（在生成前）
cost_estimate = generator.estimate_cost(
num_samples=len(instructions),
avg_instruction_tokens=50,
avg_output_tokens=500
)
print("\\n📊 成本估算:")
print(f" 模型: {cost_estimate['model']}")
print(f" 样本数: {cost_estimate['num_samples']}")
print(f" 总 tokens: {cost_estimate['total_tokens']:,}")
print(f" 输入成本: ${cost_estimate['input_cost']}")
print(f" 输出成本: ${cost_estimate['output_cost']}")
print(f" 总成本: ${cost_estimate['total_cost']}")
print(f" 每条成本: ${cost_estimate['cost_per_sample']}")
print(f" 预计耗时: {cost_estimate['estimated_time_minutes']} 分钟\\n")

# 4. 批量生成
# dataset = generator.batch_generate(
# instructions=instructions,
# output_file="my_synthetic_data.jsonl",
# batch_size=5,
# delay=1.0
# )

# 4. 查看生成的数据
# with open("my_synthetic_data.jsonl", 'r', encoding='utf-8') as f:
# for line in f:
# sample = json.loads(line)
# print(f"指令: {sample['instruction']}")
# print(f"回答: {sample['output'][:100]}…\\n")

Step 3: 质量控制与后处理

"""
合成数据的质量控制
包含：长度过滤、重复检测、毒性检测
"""
class SyntheticDataQualityControl:
def __init__(self):
self.min_output_length = 50 # 最短回答长度（字符）
self.max_output_length = 2000 # 最长回答长度

def check_quality(self, sample: Dict) –> tuple[bool, str]:
"""
检查单条样本的质量

返回：
(是否通过, 失败原因)
"""
output = sample['output']

# 1. 长度检查
if len(output) < self.min_output_length:
return False, "output_too_short"
if len(output) > self.max_output_length:
return False, "output_too_long"

# 2. 检查是否包含拒绝回答的模式
refusal_patterns = [
"i cannot", "i can't", "i'm unable to",
"as an ai", "i don't have access",
"抱歉", "对不起，我无法", "我不能"
]
output_lower = output.lower()
for pattern in refusal_patterns:
if pattern in output_lower:
return False, "refusal_detected"

# 3. 检查是否过于简短
if len(output.split()) < 10:
return False, "too_few_words"

return True, "passed"

def filter_dataset(self, input_file: str, output_file: str) –> Dict:
"""
批量过滤数据集

返回：
统计信息
"""
filtered = []
stats = {
"total": 0,
"passed": 0,
"rejected": 0,
"reasons": {}
}

with open(input_file, 'r', encoding='utf-8') as f:
for line in f:
stats['total'] += 1
sample = json.loads(line)

passed, reason = self.check_quality(sample)

if passed:
filtered.append(sample)
stats['passed'] += 1
else:
stats['rejected'] += 1
stats['reasons'][reason] = stats['reasons'].get(reason, 0) + 1

# 保存过滤后的数据
with open(output_file, 'w', encoding='utf-8') as f:
for sample in filtered:
f.write(json.dumps(sample, ensure_ascii=False) + '\\n')

print(f"\\n质量过滤完成！")
print(f"总计: {stats['total']}")
print(f"通过: {stats['passed']}")
print(f"拒绝: {stats['rejected']}")
print(f"通过率: {stats['passed'] / stats['total'] * 100:.1f}%")
print(f"拒绝原因: {stats['reasons']}")

return stats

# 使用示例
# qc = SyntheticDataQualityControl()
# stats = qc.filter_dataset("my_synthetic_data.jsonl", "filtered_data.jsonl")

5.3 成本估算与优化

成本估算表：

使用 estimate_cost() 方法进行精确预估：

上面的 SyntheticDataGenerator 类已内置 estimate_cost() 方法，可以在生成前精确计算成本。

使用示例：

# 创建生成器
generator = SyntheticDataGenerator(model="gpt-4")

# 估算生成 10K 条数据的成本
cost_info = generator.estimate_cost(
num_samples=10000,
avg_instruction_tokens=50, # 平均指令长度
avg_output_tokens=500 # 平均回答长度
)

print(f"模型: {cost_info['model']}")
print(f"样本数: {cost_info['num_samples']:,}")
print(f"总 tokens: {cost_info['total_tokens']:,}")
print(f"总成本: ${cost_info['total_cost']}")
print(f"每条成本: ${cost_info['cost_per_sample']}")
print(f"预计耗时: {cost_info['estimated_time_minutes']} 分钟")

输出示例：

模型: gpt-4
样本数: 10,000
总 tokens: 8,500,000
总成本: $185.0
每条成本: $0.0185
预计耗时: 333.3 分钟

不同模型的成本对比：

# 对比不同模型的成本
models = ["gpt-4", "gpt-3.5-turbo", "gpt-4o-mini"]
num_samples = 10000

print("\\n📊 成本对比（10K 条数据）:\\n")
print(f"{'模型':<20} {'总成本':>10} {'每条成本':>12} {'预计耗时':>12}")
print("-" * 60)

for model in models:
gen = SyntheticDataGenerator(model=model)
cost = gen.estimate_cost(num_samples=num_samples)
print(f"{model:<20} ${cost['total_cost']:>9.2f} ${cost['cost_per_sample']:>11.4f} {cost['estimated_time_minutes']:>9.1f}分钟")

输出示例：

📊 成本对比（10K 条数据）:

模型 总成本 每条成本 预计耗时
————————————————————
gpt-4 $185.00 $0.0185 333.3分钟
gpt-3.5-turbo $9.25 $0.0009 333.3分钟
gpt-4o-mini $3.88 $0.0004 333.3分钟

优化策略：

5.4 实战建议

数据量建议：

• 小规模任务（客服、FAQ）：1K – 5K 条
• 中等任务（通用对话）：10K – 50K 条
• 大规模任务（多领域）：50K – 500K 条

质量保证：

常见陷阱：

• ❌ 种子指令过于单一，导致生成数据缺乏多样性
• ❌ 不做质量过滤，直接使用所有生成数据
• ❌ 全部使用 GPT-4，成本过高
• ✅ 结合人工标注 + 合成数据，取长补短

五、小规模数据去重

1. 为什么微调数据需要去重：Scalability 的关键

问题：重复数据会导致四大问题

问题 1：过拟合

• 模型死记硬背重复样本，泛化能力下降
• 在验证集上表现好，实际应用效果差

问题 2：训练偏差

• 重复样本占比过高，模型偏向这些样本
• 例如：如果 30% 的数据都是"解释 XXX"，模型会倾向于生成解释类回答

问题 3：计算浪费 ⚡ 这是最严重的问题!

• 重复数据不提供新信息，却消耗同样的计算资源
• 去重后可用更少的 FLOPs 达到同等效果

问题 4：内存浪费

• 重复数据占用存储和显存，限制batch size

去重对 Scalability 的巨大影响

根据 Lee et al. (2022) 的论文 “Deduplicating Training Data Makes Language Models Better”：

┌─────────────────────────────────────────────────────────────────┐
│ DEDUPLICATION → MASSIVE EFFICIENCY GAINS │
│ (去重 → 效率爆炸式提升) │
└─────────────────────────────────────────────────────────────────┘

实验设置:
– 模型: GPT-3 规模(175B)
– 数据集: C4 (750GB)
– 去重比例: 移除 ~25% 的近似重复数据

结果对比:

训练前 去重后
───── ──────
数据量: 750GB 560GB (-25%)
训练时间: 1000 GPU-days 750 GPU-days (-25%)
FLOPs: 2e23 1.5e23 (-25%)

性能: Perplexity: 15.2 Perplexity: 14.8 (↑2.6%)
↑ 更少的数据 → 更好的效果!

关键发现:
✅ 去重后，只需 75% 的 FLOPs 就能达到更好的效果
✅ 等效性能下，训练成本降低 3-5x
✅ 下游任务准确率平均提升 1-3%

为什么去重后效果反而更好?

原因1: 信息密度提升

去重前: 100K样本，30%重复 → 有效信息只有70K
去重后: 70K样本，0%重复 → 有效信息就是70K
信息密度: 70% → 100%

原因2: 梯度更新质量提升

• 重复样本导致梯度被同一信息主导
• 去重后每个batch包含更多样化的信息
• 模型学习更均衡，泛化能力更强

去重策略：

• 精确去重：移除完全相同的样本（基于哈希）
• 模糊去重：移除高度相似的样本（基于 MinHash）

实践建议:

• 小规模数据(< 10K): 精确去重即可，成本可忽略
• 中等规模(10K-100K): 使用 MinHash，去重率通常 15-30%
• 大规模数据(> 100K): 必须去重，否则训练效率极低

2. MinHash 去重原理：文档"指纹识别"技术

核心思想：MinHash 为每个文档生成固定长度的"签名"(指纹)，使得相似文档的签名也相似。

形象化理解：文档指纹化

想象你要识别海量文档中的重复内容。直接逐字比较太慢了！MinHash 就像给每个文档"按指纹":

原始文档 A: MinHash 签名:
"机器学习是人工智能的分支…" [0x3F, 0xA2, 0x1B, …]
│ │
├─> Shingling (切词) │ 只需比较短签名
│ {"机器", "学习", "是人", …} │ (128维) 而非
│ │ 完整文档!
├─> Hash 映射 │
│ {h1(机器)=0x3F, …} │
│ │
└─> 取最小值 ──────────────────────> [指纹]

文档 B (高度相似): MinHash 签名:
"机器学习是AI的一个分支…" [0x3F, 0xA2, 0x1C, …]
↑ 非常接近!

文档 C (完全不同): MinHash 签名:
"今天天气很好…" [0x8D, 0x5F, 0xC2, …]
↑ 完全不同!

MinHash 签名的神奇性质:

如果两个文档相似度为 80%，
那么它们的 MinHash 签名有 80% 的位相同！

Jaccard相似度(A,B) = MinHash签名相似度(A,B)

这使得我们可以：

数学原理：

给定两个文档

A

A

A 和

B

B

B，它们的 Jaccard 相似度定义为：

J

(

A

,

B

)

=

∣

A

∩

B

∣

∣

A

∪

B

∣

J(A, B) = \\frac{|A \\cap B|}{|A \\cup B|}

J(A,B)=∣A∪B∣∣A∩B∣​

MinHash 的核心性质：

P

(

MinHash

(

A

)

i

=

MinHash

(

B

)

i

)

=

J

(

A

,

B

)

P(\\text{MinHash}(A)_i = \\text{MinHash}(B)_i) = J(A, B)

P(MinHash(A)i​=MinHash(B)i​)=J(A,B)

即：MinHash 签名的某一位相同的概率，等于 Jaccard 相似度。

算法步骤：

3. 代码实战：MinHash 简单实现

适用于小规模微调数据（< 100K）：

"""
MinHash 去重（简化版）
功能：小规模数据集的模糊去重
依赖：pip install datasketch
"""
from datasketch import MinHash, MinHashLSH
from typing import List, Dict, Set
import json

class SimpleDeduplicator:
def __init__(self, threshold: float = 0.8, num_perm: int = 128):
"""
初始化去重器

参数：
threshold: 相似度阈值（0.8 表示 80% 相似即视为重复）
num_perm: MinHash 签名长度
"""
self.threshold = threshold
self.num_perm = num_perm

def _tokenize(self, text: str) –> Set[str]:
"""分词（简单版：按空格分词）"""
return set(text.lower().split())

def _compute_minhash(self, text: str) –> MinHash:
"""计算 MinHash 签名"""
m = MinHash(num_perm=self.num_perm)
tokens = self._tokenize(text)

for token in tokens:
m.update(token.encode('utf-8'))

return m

def deduplicate(self, dataset: List[Dict]) –> List[Dict]:
"""
去重数据集

参数：
dataset: Alpaca 格式数据集

返回：
去重后的数据集
"""
lsh = MinHashLSH(threshold=self.threshold, num_perm=self.num_perm)

unique_data = []
seen_indices = set()

# 第一轮：建立 LSH 索引
print("建立 LSH 索引…")
minhashes = []
for idx, sample in enumerate(dataset):
# 拼接 instruction 和 output 作为去重依据
combined_text = f"{sample['instruction']} {sample['output']}"
mh = self._compute_minhash(combined_text)
minhashes.append(mh)

# 第二轮：查找重复
print("查找重复样本…")
for idx, mh in enumerate(minhashes):
if idx in seen_indices:
continue

doc_id = f"doc_{idx}"

# 查询相似文档
candidates = lsh.query(mh)

if len(candidates) == 0:
# 第一次见到该文档
lsh.insert(doc_id, mh)
unique_data.append(dataset[idx])
else:
# 已有相似文档，跳过
seen_indices.add(idx)

return unique_data

# 使用示例
if __name__ == "__main__":
# 测试数据（包含重复）
test_data = [
{
"instruction": "解释什么是机器学习",
"input": "",
"output": "机器学习是人工智能的分支，通过算法从数据中学习。"
},
{
"instruction": "解释机器学习是什么",
"input": "",
"output": "机器学习是 AI 的一个分支，让计算机从数据中学习。" # 高度相似
},
{
"instruction": "什么是深度学习",
"input": "",
"output": "深度学习是机器学习的子集，使用多层神经网络。"
},
{
"instruction": "解释什么是机器学习", # 完全重复
"input": "",
"output": "机器学习是人工智能的分支,通过算法从数据中学习。"
}
]

# 去重
deduper = SimpleDeduplicator(threshold=0.8)
unique_data = deduper.deduplicate(test_data)

print(f"\\n原始数据: {len(test_data)} 条")
print(f"去重后: {len(unique_data)} 条")
print(f"移除: {len(test_data) – len(unique_data)} 条重复\\n")

print("去重后的数据:")
for i, sample in enumerate(unique_data, 1):
print(f"{i}. {sample['instruction']}")

性能建议：

• 对于 < 10K 数据：直接两两比较即可
• 对于 10K – 100K 数据：使用 MinHash + LSH
• 对于 > 100K 数据：参考 [Part 7 第6章] 的大规模去重方案

六、数据增强技术

1. 回译（Back-translation）

核心思想：将文本翻译成另一种语言，再翻译回来，得到语义相同但表述不同的数据。

"""
回译数据增强
依赖：pip install googletrans==4.0.0-rc1
"""
from googletrans import Translator

class BackTranslator:
def __init__(self):
self.translator = Translator()

def augment(self, text: str, intermediate_lang: str = 'fr') –> str:
"""
回译增强

参数：
text: 原始文本
intermediate_lang: 中间语言（fr=法语, de=德语, es=西班牙语）

返回：
回译后的文本
"""
# 英语 -> 中间语言
translated = self.translator.translate(text, dest=intermediate_lang)

# 中间语言 -> 英语
back_translated = self.translator.translate(translated.text, dest='en')

return back_translated.text

# 使用示例
# translator = BackTranslator()
# original = "Machine learning is a subset of artificial intelligence."
# augmented = translator.augment(original, intermediate_lang='fr')
# print(f"原始: {original}")
# print(f"增强: {augmented}")

2. 同义词替换

核心思想：随机替换文本中的词汇为同义词。

"""
同义词替换
依赖：pip install nltk
"""
import nltk
from nltk.corpus import wordnet
import random

# 下载 WordNet
# nltk.download('wordnet')
# nltk.download('omw-1.4')

class SynonymReplacer:
def __init__(self, replace_ratio: float = 0.1):
"""
参数：
replace_ratio: 替换比例（0.1 表示替换 10% 的词）
"""
self.replace_ratio = replace_ratio

def get_synonyms(self, word: str) –> List[str]:
"""获取同义词"""
synonyms = set()

for syn in wordnet.synsets(word):
for lemma in syn.lemmas():
synonym = lemma.name().replace('_', ' ')
if synonym != word:
synonyms.add(synonym)

return list(synonyms)

def augment(self, text: str) –> str:
"""同义词替换增强"""
words = text.split()
num_replace = max(1, int(len(words) * self.replace_ratio))

# 随机选择要替换的位置
replace_indices = random.sample(range(len(words)), min(num_replace, len(words)))

for idx in replace_indices:
word = words[idx]
synonyms = self.get_synonyms(word.lower())

if synonyms:
words[idx] = random.choice(synonyms)

return ' '.join(words)

# 使用示例
# replacer = SynonymReplacer(replace_ratio=0.2)
# original = "Machine learning is a powerful technique for data analysis."
# augmented = replacer.augment(original)
# print(f"原始: {original}")
# print(f"增强: {augmented}")

3. 代码实战：数据增强工具

整合多种增强策略：

"""
综合数据增强工具
"""
class DataAugmenter:
def __init__(self):
self.back_translator = BackTranslator()
self.synonym_replacer = SynonymReplacer()

def augment_dataset(self, dataset: List[Dict], target_size: int) –> List[Dict]:
"""
扩充数据集到目标大小

参数：
dataset: 原始数据集
target_size: 目标数据集大小

返回：
扩充后的数据集
"""
augmented = list(dataset) # 复制原始数据

while len(augmented) < target_size:
# 随机选择一条样本
sample = random.choice(dataset)

# 随机选择增强策略
strategy = random.choice(['back_translation', 'synonym'])

try:
if strategy == 'back_translation':
aug_instruction = self.back_translator.augment(sample['instruction'])
else:
aug_instruction = self.synonym_replacer.augment(sample['instruction'])

# 创建增强样本
aug_sample = {
"instruction": aug_instruction,
"input": sample['input'],
"output": sample['output'] # 输出保持不变
}

augmented.append(aug_sample)

except Exception as e:
continue # 增强失败，跳过

return augmented[:target_size]

注意事项：

七、SFT 数据集构建实战

1. 完整 Pipeline：从原始文本到 Alpaca 格式

场景：将一批技术文档转换为问答对，用于微调模型。

Pipeline：

原始文档

↓

1. 文档分块

文档段落

↓

2. 生成问答对（GPT-4）

(Question, Answer) 对

↓

3. 格式化为 Alpaca

Alpaca 数据集

↓

4. 质量过滤

高质量数据集

↓

5. 去重

最终数据集

\\begin{aligned} &\\text{原始文档} \\\\ &\\downarrow \\text{1. 文档分块} \\\\ &\\text{文档段落} \\\\ &\\downarrow \\text{2. 生成问答对（GPT-4）} \\\\ &\\text{(Question, Answer) 对} \\\\ &\\downarrow \\text{3. 格式化为 Alpaca} \\\\ &\\text{Alpaca 数据集} \\\\ &\\downarrow \\text{4. 质量过滤} \\\\ &\\text{高质量数据集} \\\\ &\\downarrow \\text{5. 去重} \\\\ &\\text{最终数据集} \\end{aligned}

​原始文档↓1. 文档分块文档段落↓2. 生成问答对（GPT-4）(Question, Answer) 对↓3. 格式化为 AlpacaAlpaca 数据集↓4. 质量过滤高质量数据集↓5. 去重最终数据集​

2. 代码实战：端到端数据构建

"""
端到端 SFT 数据构建 Pipeline
功能：从原始文档到 Alpaca 格式的完整流程
"""
import os
import json
from typing import List, Dict
from openai import OpenAI

class SFTDatasetBuilder:
def __init__(self, api_key: str = None):
self.client = OpenAI(api_key=api_key or os.getenv("OPENAI_API_KEY"))
self.quality_filter = QualityFilter(enable_toxicity=False, enable_pii=False)
self.deduplicator = SimpleDeduplicator(threshold=0.85)

def chunk_document(self, document: str, chunk_size: int = 500) –> List[str]:
"""
文档分块

参数：
document: 原始文档
chunk_size: 每块的词数

返回：
文档段落列表
"""
words = document.split()
chunks = []

for i in range(0, len(words), chunk_size):
chunk = ' '.join(words[i:i + chunk_size])
chunks.append(chunk)

return chunks

def generate_qa_from_chunk(self, chunk: str, num_qa: int = 3) –> List[Dict]:
"""
从文档段落生成问答对

参数：
chunk: 文档段落
num_qa: 生成问答对的数量

返回：
问答对列表
"""
prompt = f"""Based on the following text, generate {num_qa} diverse question-answer pairs. The questions should cover different aspects and difficulty levels.

Text:
{chunk}

Generate {num_qa} Q&A pairs in JSON format:
[
{{"question": "…", "answer": "…"}},
…
]

JSON:"""

try:
response = self.client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=1500
)

# 解析 JSON
qa_pairs = json.loads(response.choices[0].message.content)
return qa_pairs

except Exception as e:
print(f"生成失败: {e}")
return []

def convert_to_alpaca(self, qa_pairs: List[Dict]) –> List[Dict]:
"""
转换为 Alpaca 格式

参数：
qa_pairs: 问答对列表 [{"question": "…", "answer": "…"}]

返回：
Alpaca 格式数据
"""
alpaca_data = []

for qa in qa_pairs:
sample = {
"instruction": qa['question'],
"input": "",
"output": qa['answer']
}
alpaca_data.append(sample)

return alpaca_data

def build_from_document(self, document: str, output_file: str = "sft_dataset.jsonl"):
"""
从单个文档构建数据集

参数：
document: 原始文档
output_file: 输出文件路径
"""
print("Step 1: 文档分块…")
chunks = self.chunk_document(document)
print(f" 共 {len(chunks)} 个段落")

print("\\nStep 2: 生成问答对…")
all_qa_pairs = []
for i, chunk in enumerate(chunks):
print(f" 处理段落 {i+1}/{len(chunks)}…")
qa_pairs = self.generate_qa_from_chunk(chunk, num_qa=2)
all_qa_pairs.extend(qa_pairs)
print(f" 共生成 {len(all_qa_pairs)} 对问答")

print("\\nStep 3: 转换为 Alpaca 格式…")
alpaca_data = self.convert_to_alpaca(all_qa_pairs)

print("\\nStep 4: 质量过滤…")
stats, filtered_data = self.quality_filter.filter_dataset(alpaca_data)
print(f" 通过: {stats['passed']}/{stats['total']}")

print("\\nStep 5: 去重…")
final_data = self.deduplicator.deduplicate(filtered_data)
print(f" 最终数据: {len(final_data)} 条")

# 保存
with open(output_file, 'w', encoding='utf-8') as f:
for sample in final_data:
f.write(json.dumps(sample, ensure_ascii=False) + '\\n')

print(f"\\n数据集已保存至 {output_file}")
return final_data

# 使用示例
if __name__ == "__main__":
# 示例文档
sample_document = """
Transformer 是一种革命性的神经网络架构,由 Google 在 2017 年提出。
它完全基于注意力机制,摒弃了传统的循环神经网络(RNN)和卷积神经网络(CNN)。
Transformer 的核心创新在于 Self-Attention 机制,使得模型能够并行处理序列中的所有位置,
大大提升了训练效率。这一架构后来成为 BERT、GPT 等大语言模型的基础。

Transformer 包含编码器(Encoder)和解码器(Decoder)两个主要组件。
编码器负责理解输入序列,而解码器负责生成输出序列。
每个编码器层包含 Multi-Head Self-Attention 和 Feed-Forward Network 两个子层。
通过堆叠多层编码器和解码器,Transformer 能够学习复杂的语言模式。

自 2017 年以来,Transformer 已经主导了自然语言处理领域,并扩展到计算机视觉、
语音识别等多个领域。它的成功证明了注意力机制的强大能力,
也为后续的模型创新奠定了基础。
"""

# 构建数据集
builder = SFTDatasetBuilder()
# dataset = builder.build_from_document(sample_document, output_file="transformer_sft.jsonl")

# 查看生成的数据
# with open("transformer_sft.jsonl", 'r', encoding='utf-8') as f:
# for line in f:
# sample = json.loads(line)
# print(f"\\n指令: {sample['instruction']}")
# print(f"回答: {sample['output'][:100]}…")

输出示例：

Step 1: 文档分块…
共 1 个段落

Step 2: 生成问答对…
处理段落 1/1…
共生成 2 对问答

Step 3: 转换为 Alpaca 格式…

Step 4: 质量过滤…
通过: 2/2

Step 5: 去重…
最终数据: 2 条

数据集已保存至 transformer_sft.jsonl

实战建议：

八、本章小结

核心要点：

下一步：

• 有了高质量数据后，下一章将学习如何使用这些数据进行微调（详见 第2章_微调你的专属模型）

参考资源

论文：

• Self-Instruct: Aligning Language Models with Self-Generated Instructions (Stanford, 2023)
• Alpaca: Stanford Alpaca: An Instruction-following LLaMA Model
• WizardLM: WizardLM: Empowering Large Language Models to Follow Complex Instructions
• Phi-3: Phi-3 Technical Report (Microsoft, 2024)

工具库：

• OpenAI API: https://platform.openai.com/docs
• datasketch: https://github.com/ekzhu/datasketch (MinHash 去重)
• detoxify: https://github.com/unitaryai/detoxify (毒性检测)
• presidio: https://github.com/microsoft/presidio (PII 脱敏)

数据集：

• Alpaca-52K: https://github.com/tatsu-lab/stanford_alpaca/blob/main/alpaca_data.json
• ShareGPT: https://huggingface.co/datasets/anon8231489123/ShareGPT_Vicuna_unfiltered
• WizardLM Dataset: https://huggingface.co/datasets/WizardLM/WizardLM_evol_instruct_V2_196k

下一章：第2章_微调你的专属模型 将详细讲解 LoRA、QLoRA 等高效微调技术。