准备高质量训练数据集
收集、清理并按照 Alpaca 和 ShareGPT 格式整理指令遵循数据，执行数据去重，并将数据拆分为训练集和验证集。
准备高质量训练数据集 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
数据是微调中最重要的因素
在微调中，训练数据的质量比任何超参数、架构选择或训练技术都更加重要。100 个高质量示例胜过 10000 个平庸示例。输入什么，输出就是什么——微调模型会忠实地复现数据中的各种模式，包括错误、偏见和格式不一致。投资于数据质量，是任何微调项目中投入产出比最高的行动。
遵循指令的格式
大多数用于遵循指令任务的微调，都会采用包含系统、用户和助手角色的对话消息格式。OpenAI 的微调 API 使用 JSONL 文件，其中每一行都是一个完整的对话示例。Alpaca 格式（指令/输入/输出）和 ShareGPT 格式（对话列表）也被广泛使用。请选择与您计划使用的微调框架相匹配的格式。
import json
# OpenAI fine-tuning format (JSONL)
# Each line is one training example
openai_example = {
'messages': [
{'role': 'system', 'content': 'You are a JSON extraction agent.'},
{'role': 'user', 'content': 'Extract: "John Smith, age 32, from Seattle, joined 2023-01-15"'},
{'role': 'assistant', 'content': '{"name": "John Smith", "age": 32, "city": "Seattle", "join_date": "2023-01-15"}'}
]
}
# Alpaca format
alpaca_example = {
'instruction': 'Extract structured data from the following text.',
'input': 'John Smith, age 32, from Seattle, joined 2023-01-15',
'output': '{"name": "John Smith", "age": 32, "city": "Seattle", "join_date": "2023-01-15"}'
}
# Write as JSONL
with open('train.jsonl', 'w') as f:
f.write(json.dumps(openai_example) + '\n')
# Add more examples here...收集训练数据：三种策略
构建微调数据集主要有三种策略。人工生成：由专家手动编写理想示例，质量最高，但速度慢且成本高。LLM 生成：由强大的模型（GPT-4o）生成示例，再由人工进行验证，速度更快、成本更低；只要经过验证，质量也很好。从日志中挖掘：从现有生产日志中提取输入—输出对，并使用用户评分或 LLM 评审打分等质量信号筛选高质量示例。
from openai import OpenAI
client = OpenAI()
def generate_training_example_with_gpt4o(task_description: str, example_input: str) -> dict:
'''Use GPT-4o to generate a training example for a smaller model.'''
prompt = f'''You are creating a training example for fine-tuning a smaller model.
Task: {task_description}
Given this input:
{example_input}
Write the ideal assistant response that demonstrates the correct behavior for this task.
Be specific, accurate, and follow the expected format precisely.'''
response = client.chat.completions.create(
model='gpt-4o', # teacher model
messages=[{'role': 'user', 'content': prompt}]
)
return {
'messages': [
{'role': 'system', 'content': task_description},
{'role': 'user', 'content': example_input},
{'role': 'assistant', 'content': response.choices[0].message.content}
]
}质量筛选与验证
每个训练示例在纳入数据集之前，都应通过质量验证步骤。请验证：响应格式是否正确；对于事实性任务，响应是否准确；响应是否包含幻觉或有害内容；指令与响应是否连贯；以及示例是否能代表生产使用场景。使用自动验证检查格式，使用 LLM 评审判断内容质量，并抽样进行人工审查。
import json
def validate_training_example(example: dict, schema_validator=None) -> dict:
issues = []
# Format check
if 'messages' not in example:
issues.append('Missing messages field')
return {'valid': False, 'issues': issues}
messages = example['messages']
if not any(m['role'] == 'assistant' for m in messages):
issues.append('No assistant message found')
# Check assistant message quality
assistant_content = next((m['content'] for m in messages if m['role'] == 'assistant'), '')
if len(assistant_content) < 5:
issues.append('Assistant response too short')
# Schema validation for JSON output tasks
if schema_validator:
try:
parsed = json.loads(assistant_content)
schema_validator(parsed) # raises if invalid
except json.JSONDecodeError:
issues.append('Assistant response is not valid JSON')
except Exception as e:
issues.append(f'Schema validation failed: {str(e)}')
return {'valid': len(issues) == 0, 'issues': issues}
# Run validation on all examples before training
examples = load_training_examples('raw_dataset.jsonl')
valid_examples = [e for e in examples if validate_training_example(e)['valid']]
print(f'Valid examples: {len(valid_examples)}/{len(examples)}')数据去重
训练数据中的重复或近似重复示例会造成危害。它们会使模型过度拟合这些特定示例，浪费本可用于学习多样模式的训练容量。在确定数据集之前，请先执行去重。完全去重使用哈希查找完全相同的示例。近似去重使用 MinHash 或嵌入相似度查找仅在细微方面有所不同的示例。
import hashlib
from datasketch import MinHash, MinHashLSH
def exact_deduplicate(examples: list[dict]) -> list[dict]:
seen_hashes = set()
unique = []
for ex in examples:
# Hash the user and assistant messages
content = str(ex['messages'])
h = hashlib.md5(content.encode()).hexdigest()
if h not in seen_hashes:
seen_hashes.add(h)
unique.append(ex)
print(f'Exact dedup: {len(examples)} -> {len(unique)} ({len(examples)-len(unique)} removed)')
return unique
def near_deduplicate_by_input(examples: list[dict], similarity_threshold=0.85) -> list[dict]:
# Build index of input texts
inputs = [next((m['content'] for m in ex['messages'] if m['role'] == 'user'), '') for ex in examples]
lsh = MinHashLSH(threshold=similarity_threshold, num_perm=128)
unique_indices = set()
for i, text in enumerate(inputs):
m = MinHash(num_perm=128)
for word in text.lower().split():
m.update(word.encode('utf-8'))
if not lsh.query(m): # no similar items found
lsh.insert(str(i), m)
unique_indices.add(i)
return [examples[i] for i in sorted(unique_indices)]训练集与验证集划分
在开始任何微调训练之前，请将数据集划分为训练集和验证集。验证集用于监控训练过程中的过拟合情况（如果验证损失上升而训练损失下降，说明模型正在过拟合）。典型划分方式是 90% 训练集 / 10% 验证集。请确保划分是随机的；如果数据集包含有意义的类别，还应进行分层划分（例如，让两组中都均匀包含所有意图类型）。
import random
import json
def split_dataset(examples: list[dict], val_fraction=0.1, seed=42) -> tuple[list, list]:
random.seed(seed) # reproducible split
shuffled = examples.copy()
random.shuffle(shuffled)
n_val = max(1, int(len(shuffled) * val_fraction))
val_set = shuffled[:n_val]
train_set = shuffled[n_val:]
print(f'Train: {len(train_set)} examples, Validation: {len(val_set)} examples')
return train_set, val_set
def save_jsonl(examples: list[dict], path: str):
with open(path, 'w') as f:
for ex in examples:
f.write(json.dumps(ex) + '\n')
# Split and save
examples = load_training_examples('clean_dataset.jsonl')
train, val = split_dataset(examples, val_fraction=0.1)
save_jsonl(train, 'train.jsonl')
save_jsonl(val, 'validation.jsonl')
print(f'Saved train.jsonl ({len(train)}) and validation.jsonl ({len(val)})')平衡数据集
不平衡数据集会使微调模型过度专注于常见情况，却无法处理少见但重要的情况。如果您的数据集中有 900 个类别 A 示例和 100 个类别 B 示例，模型可能会学会始终预测 A。您可以通过以下方式平衡数据集：对少数类别过采样（复制少见示例）、对多数类别欠采样，或使用 GPT-4o 为代表性不足的情况生成合成示例。
from collections import Counter
import random
def analyze_distribution(examples: list[dict], category_extractor) -> dict:
categories = [category_extractor(ex) for ex in examples]
counts = Counter(categories)
print('Category distribution:')
for cat, count in counts.most_common():
print(f' {cat}: {count} ({100*count/len(examples):.1f}%)')
return counts
def oversample_minority(examples: list[dict], category_extractor, target_count: int) -> list[dict]:
by_category = {}
for ex in examples:
cat = category_extractor(ex)
by_category.setdefault(cat, []).append(ex)
balanced = []
for cat, cat_examples in by_category.items():
if len(cat_examples) < target_count:
# Oversample with replacement
oversampled = random.choices(cat_examples, k=target_count)
balanced.extend(oversampled)
else:
# Undersample to target_count
balanced.extend(random.sample(cat_examples, target_count))
random.shuffle(balanced)
return balanced数据清理与规范化
训练数据通常包含会损害微调效果的不一致之处：输出字段中的大小写混用、末尾空格、引号使用不一致、数字格式混用，或 JSON 键命名方式不统一。请在训练前将这些内容规范化。微调模型会学习数据中实际存在的精确格式——如果数据存在不一致，模型也会复现这些问题。
import json
import re
def normalize_json_output_example(example: dict) -> dict:
'''Normalize JSON output in assistant messages for consistency.'''
messages = example.get('messages', [])
normalized = []
for msg in messages:
if msg['role'] == 'assistant':
content = msg['content'].strip()
# Try to parse and re-serialize JSON for consistent formatting
try:
parsed = json.loads(content)
# Normalize: sort keys, consistent spacing
content = json.dumps(parsed, ensure_ascii=False, sort_keys=True)
except json.JSONDecodeError:
pass # Not JSON output - leave as is
normalized.append({'role': 'assistant', 'content': content})
else:
normalized.append(msg)
return {'messages': normalized}
def normalize_dataset(examples: list[dict]) -> list[dict]:
return [normalize_json_output_example(ex) for ex in examples]衡量数据集质量指标
在提交数据进行微调之前，请计算整个数据集的质量指标。检查以下内容：每个示例的平均和最大令牌数（较长的示例训练成本更高，而且可能被截断）；词汇覆盖率（数据集是否覆盖生产输入的全部多样性）；以及一致性分数（相似输入是否得到相似输出）。大多数微调服务提供商都有数据验证接口，会在因训练失败向您收费之前检查格式错误。
import tiktoken
def analyze_dataset_quality(examples: list[dict], model='gpt-4o-mini') -> dict:
encoder = tiktoken.encoding_for_model(model)
token_counts = []
for ex in examples:
total_tokens = sum(
len(encoder.encode(m['content']))
for m in ex['messages']
)
token_counts.append(total_tokens)
report = {
'total_examples': len(examples),
'avg_tokens_per_example': sum(token_counts) / len(token_counts),
'max_tokens': max(token_counts),
'min_tokens': min(token_counts),
'examples_over_4k_tokens': sum(1 for t in token_counts if t > 4096),
'estimated_training_tokens': sum(token_counts),
'estimated_cost': sum(token_counts) / 1_000_000 * 8.0 # ~$8/1M tokens for gpt-4o-mini
}
for key, value in report.items():
print(f'{key}: {value}')
return report迭代改进数据集
数据集准备是一个迭代过程。使用初始数据集微调一个小型模型，在留出的示例上评估模型，找出失败模式，并追溯这些问题是由数据集缺口还是质量问题造成的。然后修复数据并重新训练。这种错误驱动的数据改进循环是生产环境中微调的标准做法，远比试图一次性收集完美数据集有效。
跨示例保持系统提示一致
如果您的微调模型在生产环境中始终使用相同的系统提示词，请在每个训练示例中加入完全一致的系统提示词。如果您希望模型在没有系统提示词的情况下也能工作，请在训练时不要使用系统提示词。训练条件与推理条件不一致，是微调效果令人失望的主要原因之一。模型会根据训练期间看到的确切提示词结构，学习相应的行为。
快速检查
请测试您对本课所讲的微调训练数据集准备方法的理解。
课程回顾
本课中您学习了：在微调中，数据质量胜过数据数量——100 个优秀示例的效果优于 10,000 个普通示例；在开始任何训练之前，去重、验证、平衡和规范化是数据清理的四个关键步骤；此外，训练集/验证集划分对于在训练期间发现过拟合、避免其损害真实场景中的性能至关重要。接下来，我们将使用 Hugging Face PEFT 进行 LoRA 微调。
常见问题解答
「准备高质量训练数据集」课时是免费的吗？
是的 — 「准备高质量训练数据集」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「准备高质量训练数据集」这节课中我会学到什么？
收集、清理并按照 Alpaca 和 ShareGPT 格式整理指令遵循数据，执行数据去重，并将数据拆分为训练集和验证集。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「准备高质量训练数据集」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。