关键词
偏好数据、人工标注、自动化标注、拒绝采样、合成数据、数据质量、标注指南、LLM-as-Judge、Elo评分、课程学习
概述
偏好数据是对齐训练的核心燃料。无论是DPO、KTO、ORPO还是传统的PPO方法，都需要高质量的偏好数据来指导模型学习人类价值观。构建高质量偏好数据是一个系统性的工程问题，涉及数据收集、标注规范、质量控制、清洗增强等多个环节。本文档将详细介绍偏好数据构建的完整流程和最佳实践。
偏好数据收集方法
数据收集策略概览
偏好数据的来源可以分为三大类：
| 来源 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 人工标注 | 质量高、一致性好 | 成本高、速度慢 | 核心数据、高价值场景 |
| 模型生成 | 成本低、规模大 | 质量参差、需要验证 | 辅助数据、边缘案例 |
| 混合策略 | 平衡质量和成本 | 复杂度高 | 大规模生产 |
人工标注流程
1. 标注任务设计
标注任务设计原则
- 明确标注目标：帮助性、安全性、简洁性等
- 简化决策过程：尽量减少标注者的认知负担
- 提供足够的上下文：让标注者能做出明智判断
# 标注任务设计示例
annotation_task = {
"task_name": "助手响应偏好标注",
"task_description": "对AI助手的响应进行偏好排序",
"instructions": """
请对以下AI助手对同一问题的多个响应进行偏好排序。
评分标准：
1. 帮助性（40%）：响应是否解决了用户的问题
2. 准确性（30%）：响应中的信息是否正确
3. 安全性（20%）：响应是否包含有害内容
4. 清晰度（10%）：响应是否易于理解
标注规则：
- 每个响应必须给出 A、B、C 等级
- 必须为每个响应提供简要理由
- 遇到不确定时，标注"需要专家审核"
""",
"ui_components": [
{"type": "prompt_display", "label": "用户问题"},
{"type": "response_list", "label": "候选响应"},
{"type": "ranking", "label": "偏好排序"},
{"type": "reason_input", "label": "标注理由"},
{"type": "confidence_slider", "label": "标注置信度"},
]
}2. 标注员培训
# 标注员培训模块
training_modules = [
{
"title": "基础标注规范",
"content": """
1. 理解任务背景
- 这是为了训练AI助手
- 目标是让AI更有帮助、更安全
2. 标注标准
- A级：优秀，全面满足用户需求
- B级：良好，满足大部分需求
- C级：一般，存在明显不足
- D级：较差，有严重问题
3. 常见陷阱
- 长度偏见：不要偏好过长的响应
- 位置偏见：注意响应的随机顺序
- 一致性：保持标注标准的一致
""",
"examples": [
{
"input": "如何学习编程？",
"responses": [
"多看视频教程。",
"学习编程的建议：1. 选择语言 2. 实践项目 3. 参与社区 4. 持续学习",
],
"correct_label": "B > A",
"explanation": "响应2更具体、可操作",
},
# 更多示例...
],
"test_questions": [
# 测试题...
]
},
{
"title": "安全性标注",
"content": """
安全性是最高优先级。
必须拒绝的情况：
1. 提供危险物品制作方法
2. 协助非法活动
3. 侵犯隐私的建议
4. 歧视性内容
需要谨慎的情况：
1. 医疗建议（建议咨询专业人士）
2. 法律问题（明确局限性）
3. 敏感历史事件（客观中立）
""",
"test_questions": [
# 安全相关测试题...
]
}
]3. 标注系统实现
# 标注系统核心组件
class AnnotationSystem:
def __init__(self, config):
self.config = config
self.current_task = None
def load_batch(self, prompts):
"""加载一批待标注数据"""
self.current_batch = []
for prompt in prompts:
# 为每个prompt生成多个响应
responses = self.generate_candidates(prompt)
self.current_batch.append({
'prompt': prompt,
'responses': responses,
'annotations': []
})
return self.current_batch
def get_next_item(self, annotator_id):
"""获取下一个待标注项目"""
# 分配任务（考虑负载均衡）
item = self.assign_task(annotator_id)
return item
def submit_annotation(self, annotator_id, item_id, annotation):
"""提交标注"""
# 验证标注格式
validated = self.validate_annotation(annotation)
# 存储标注
self.store_annotation(item_id, annotator_id, validated)
# 更新统计
self.update_stats(annotator_id, validated)
return {"status": "success", "quality_score": validated.quality}
def validate_annotation(self, annotation):
"""验证标注质量"""
# 检查必填项
# 检查格式
# 检查一致性
# 计算质量分数
return validated_annotation自动化偏好数据生成
LLM-as-Judge范式
使用强大的LLM作为评判者生成偏好数据：
class LLMJudge:
def __init__(self, judge_model):
self.judge = judge_model
def generate_preference(self, prompt, response_a, response_b):
"""生成偏好判断"""
judge_prompt = f"""
你是一个专业的AI助手评估员。请判断以下两个响应哪个更好。
用户问题：{prompt}
响应A：{response_a}
响应B：{response_b}
请给出：
1. 偏好选择：A、B或平局
2. 置信度：低、中、高
3. 判断理由
4. 评分差异（0-10分）
"""
result = self.judge.generate(judge_prompt)
return self.parse_judgment(result)
def batch_generate_preferences(self, pairs):
"""批量生成偏好"""
preferences = []
for prompt, resp_a, resp_b in pairs:
pref = self.generate_preference(prompt, resp_a, resp_b)
preferences.append({
'prompt': prompt,
'chosen': pref.chosen,
'rejected': pref.rejected,
'confidence': pref.confidence,
'reasoning': pref.reasoning
})
return preferences自动偏好生成的Prompt模板
JUDGE_PROMPTS = {
"helpfulness": """
请评估以下AI响应的帮助性。
问题：{prompt}
响应：{response}
帮助性评估标准：
- 是否直接回答了用户的问题
- 是否提供了有用的信息或建议
- 是否有清晰的逻辑结构
- 是否考虑了用户的潜在需求
请给出1-10的帮助性评分。
""",
"safety": """
请评估以下AI响应的安全性。
问题：{prompt}
响应：{response}
安全性评估标准：
- 是否包含有害内容
- 是否可能造成危险
- 是否尊重用户隐私
- 是否避免了不当内容
请判断：安全/需要修改/不安全，并说明理由。
""",
"comparison": """
请比较以下两个AI响应的优劣。
问题：{prompt}
响应A：{response_a}
响应B：{response_b}
请从以下维度进行比较：
1. 准确性
2. 帮助性
3. 完整性
4. 清晰度
哪个响应整体更好？请给出详细理由。
"""
}拒绝采样（Rejection Sampling）
拒绝采样的基本原理
拒绝采样是一种通过筛选高奖励样本进行训练的策略：
┌──────────────────────────────────────────┐
│ 拒绝采样流程 │
├──────────────────────────────────────────┤
│ │
│ Prompt → Model → Response → RM Score │
│ │ │ │ │
│ │ ↓ ↓ │
│ │ 响应集合 分数集合 │
│ │ │ │ │
│ │ └──────┬───────┘ │
│ │ ↓ │
│ │ Score > Threshold? │
│ │ │ │
│ │ ┌─────────┴─────────┐ │
│ │ ↓ ↓ │
│ │ 接受 拒绝 │
│ │ │ │ │
│ │ ↓ ↓ │
│ │ 训练数据 丢弃 │
│ │
└──────────────────────────────────────────┘
实现代码
def rejection_sampling(
model,
reward_model,
prompts,
num_samples_per_prompt=10,
threshold_percentile=70,
):
"""
拒绝采样主函数
Args:
model: 待训练的模型
reward_model: 奖励模型
prompts: 提示列表
num_samples_per_prompt: 每个提示采样的数量
threshold_percentile: 接受阈值（百分位）
"""
all_samples = []
# 1. 批量生成响应
for prompt in tqdm(prompts):
samples = []
for _ in range(num_samples_per_prompt):
response = model.generate(
prompt,
temperature=0.8,
do_sample=True,
)
score = reward_model(prompt, response)
samples.append({
'prompt': prompt,
'response': response,
'score': score
})
all_samples.extend(samples)
# 2. 计算阈值
scores = [s['score'] for s in all_samples]
threshold = np.percentile(scores, threshold_percentile)
# 3. 筛选高质量样本
accepted = [s for s in all_samples if s['score'] >= threshold]
rejected = [s for s in all_samples if s['score'] < threshold]
return accepted, rejected, threshold
def adaptive_rejection_sampling(
model,
reward_model,
prompts,
min_samples=5,
max_samples=20,
target_accept_rate=0.3,
):
"""
自适应拒绝采样
根据响应质量动态调整采样数量
"""
results = []
for prompt in prompts:
samples = []
current_samples = 0
while current_samples < max_samples:
response = model.generate(
prompt,
temperature=0.7 + 0.3 * random.random(),
)
score = reward_model(prompt, response)
samples.append({
'prompt': prompt,
'response': response,
'score': score
})
current_samples += 1
# 自适应停止
if current_samples >= min_samples:
sorted_scores = sorted([s['score'] for s in samples])
top_ratio = sorted_scores[-max(1, len(sorted_scores)//3)]
bottom_ratio = sorted_scores[len(sorted_scores)//3]
if (top_ratio - bottom_ratio) > 0.5:
break
results.extend(samples)
return results合成数据技术
合成数据的必要性
当真实偏好数据不足时，合成数据是一种有效的补充手段。
合成数据的局限性
- 可能存在分布偏差
- 需要人工验证子集
- 可能放大模型偏见
合成数据生成策略
1. 基于规则生成
def rule_based_generation():
"""基于规则的合成数据生成"""
rules = [
{
"pattern": "解释.*概念",
"good_response_template": "{concept}是指{definition}。它具有以下特点：{features}",
"bad_response_template": "{concept}是一个东西。",
"goodness_criteria": {
"min_length": 100,
"has_structure": True,
"has_examples": True
}
},
# 更多规则...
]
generated = []
for rule in rules:
for concept in get_concepts():
good = fill_template(rule["good_response_template"], concept)
bad = fill_template(rule["bad_response_template"], concept)
generated.append({
"prompt": f"解释{concept['name']}",
"chosen": good,
"rejected": bad,
"source": "rule_based"
})
return generated2. 基于模型生成
class SyntheticPreferenceGenerator:
def __init__(self, base_model, ref_model):
self.base = base_model
self.ref = ref_model
def generate_preference_pairs(
self,
prompts,
strategy="self-improvement",
):
"""使用模型生成偏好对"""
if strategy == "self-improvement":
return self.self_improvement_strategy(prompts)
elif strategy == "diversity":
return self.diversity_strategy(prompts)
elif strategy == "adversarial":
return self.adversarial_strategy(prompts)
def self_improvement_strategy(self, prompts):
"""自我改进策略：生成v1和v2，v2是v1的改进版"""
pairs = []
for prompt in prompts:
# 生成初始响应
v1 = self.base.generate(prompt, temperature=0.7)
# 让模型改进自己的响应
improve_prompt = f"""
请改进以下响应，使其更有帮助、更清晰：
原始响应：{v1}
改进后的响应：
"""
v2 = self.base.generate(improve_prompt, temperature=0.5)
pairs.append({
"prompt": prompt,
"chosen": v2,
"rejected": v1,
"source": "self_improvement"
})
return pairs
def diversity_strategy(self, prompts):
"""多样性策略：生成多个不同风格的响应"""
pairs = []
for prompt in prompts:
# 生成多个不同风格的响应
responses = []
for style in ["formal", "casual", "technical", "simple"]:
style_prompt = f"用{style}风格回答：{prompt}"
response = self.base.generate(style_prompt, temperature=0.9)
responses.append(response)
# 选择最佳和最差的
scored = [(r, self.ref(prompt, r)) for r in responses]
scored.sort(key=lambda x: x[1])
pairs.append({
"prompt": prompt,
"chosen": scored[-1][0],
"rejected": scored[0][0],
"source": "diversity"
})
return pairs
def adversarial_strategy(self, prompts):
"""对抗策略：生成试图欺骗的响应"""
pairs = []
for prompt in prompts:
# 正常响应
normal = self.base.generate(prompt, temperature=0.7)
# 对抗性响应（可能包含偏见/错误）
adversarial = self.base.generate(
f"提供一个有偏见但看起来合理的回答：{prompt}",
temperature=1.0
)
pairs.append({
"prompt": prompt,
"chosen": normal,
"rejected": adversarial,
"source": "adversarial"
})
return pairs数据质量控制
质量评估指标
| 指标 | 计算方法 | 目标值 | 说明 |
|---|---|---|---|
| 标注一致性 | Kappa系数 | >0.6 | 标注者间一致性 |
| 偏好分布 | 偏好比例 | 0.3-0.7 | 避免极端分布 |
| 响应多样性 | Unique-gram比例 | >0.6 | 多样性足够 |
| 长度分布 | 变异系数 | <1.0 | 长度合理 |
| 质量得分 | 奖励模型评分 | 符合预期 | 整体质量 |
质量控制流程
class QualityController:
def __init__(self, config):
self.thresholds = config.quality_thresholds
def check_batch_quality(self, batch):
"""检查批次质量"""
results = {
'consistency': self.check_consistency(batch),
'distribution': self.check_distribution(batch),
'diversity': self.check_diversity(batch),
'cleanliness': self.check_cleanliness(batch),
}
# 决定是否接受
results['accepted'] = all([
results['consistency'] > self.thresholds['consistency'],
results['distribution'] > self.thresholds['distribution'],
results['diversity'] > self.thresholds['diversity'],
])
return results
def check_consistency(self, batch):
"""检查标注一致性"""
# 计算交叉标注的一致性
kappa_scores = []
for item in batch:
if item.get('cross_annotations'):
kappa = compute_fleiss_kappa(item['cross_annotations'])
kappa_scores.append(kappa)
return np.mean(kappa_scores) if kappa_scores else 0.0
def check_distribution(self, batch):
"""检查偏好分布"""
chosen_count = sum(1 for item in batch if item['label'] == 'chosen')
total = len(batch)
ratio = chosen_count / total
# 理想比例在0.3-0.7之间
if 0.3 <= ratio <= 0.7:
return 1.0
else:
return min(ratio, 1 - ratio) / 0.3
def check_diversity(self, batch):
"""检查响应多样性"""
all_responses = [item['response'] for item in batch]
unique_ratio = len(set(all_responses)) / len(all_responses)
return unique_ratio
def auto_filter(self, batch):
"""自动过滤低质量样本"""
filtered = []
for item in batch:
score = self.item_quality_score(item)
if score >= self.thresholds['item_score']:
filtered.append(item)
else:
self.log_filtered(item, score)
return filtered
def item_quality_score(self, item):
"""计算单个样本质量分数"""
scores = {
'length': self.length_score(item['response']),
'format': self.format_score(item['response']),
'coherence': self.coherence_score(item),
}
# 加权平均
return 0.3 * scores['length'] + 0.2 * scores['format'] + 0.5 * scores['coherence']数据清洗实战
def clean_preference_data(raw_data):
"""清洗偏好数据"""
cleaned = []
for item in tqdm(raw_data):
# 1. 格式检查
if not validate_format(item):
continue
# 2. 长度过滤
if not (10 < len(item['chosen']) < 4000 and
10 < len(item['rejected']) < 4000):
continue
# 3. 内容过滤
if contains_nonsense(item['chosen']) or contains_nonsense(item['rejected']):
continue
# 4. 重复检测
if is_duplicate(item):
continue
# 5. 长度比检查（避免悬殊差异）
length_ratio = len(item['chosen']) / len(item['rejected'])
if not (0.5 < length_ratio < 2.0):
continue
cleaned.append(item)
return cleaned
def validate_format(item):
"""验证数据格式"""
required_fields = ['prompt', 'chosen', 'rejected']
return all(field in item for field in required_fields)
def contains_nonsense(text):
"""检测无意义内容"""
# 检查重复字符
if re.search(r'(.)\1{5,}', text):
return True
# 检查乱码
if '詁' in text or 'ﾧ' in text: # 日文乱码特征
return True
# 检查单词重复
words = text.split()
if len(words) > 10:
unique_ratio = len(set(words)) / len(words)
if unique_ratio < 0.3:
return True
return False数据格式与存储
标准JSON格式
{
"version": "1.0",
"created_at": "2024-01-15T10:30:00Z",
"metadata": {
"source": "human_annotation",
"annotators": ["A001", "A002", "A003"],
"agreement_score": 0.72,
"total_samples": 10000
},
"data": [
{
"id": "sample_001",
"prompt": "如何学习一门新语言？",
"chosen": "学习新语言的建议：1. 设定明确目标 2. 每日练习...",
"rejected": "多练习就好了。",
"preference": {
"primary": "A",
"confidence": "high",
"reasoning": "chosen响应更详细、有条理",
"annotator_id": "A001"
},
"quality_scores": {
"helpfulness": 0.9,
"safety": 1.0,
"accuracy": 0.85
}
}
]
}字段说明
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| id | string | 是 | 唯一标识符 |
| prompt | string | 是 | 用户输入 |
| chosen | string | 是 | 被选中的响应 |
| rejected | string | 是 | 被拒绝的响应 |
| preference | object | 是 | 偏好标注信息 |
| quality_scores | object | 否 | 各维度质量分 |
| metadata | object | 否 | 元信息 |
数据增强策略
1. 语义等价增强
def paraphrase_augmentation(item, model):
"""生成语义等价的改写"""
chosen_paraphrased = model.generate(
f"用不同方式表达：{item['chosen']}",
temperature=0.8
)
return {
**item,
'chosen': chosen_paraphrased,
'augmentation': 'paraphrase'
}2. 对抗增强
def adversarial_augmentation(item, model):
"""生成对抗性负样本"""
# 尝试让模型生成一个看起来不错但有害的响应
adversarial = model.generate(
f"生成一个看似合理但有问题的回答：{item['prompt']}",
temperature=1.0
)
return {
**item,
'rejected': adversarial,
'augmentation': 'adversarial'
}3. 难度递进增强
def curriculum_augmentation(prompts, model, reward_model):
"""课程学习式的难度递进增强"""
easy_pairs = generate_easy_pairs(prompts, model)
medium_pairs = generate_medium_pairs(prompts, model)
hard_pairs = generate_hard_pairs(prompts, model)
return {
'easy': easy_pairs,
'medium': medium_pairs,
'hard': hard_pairs
}相关文档
- DPO深度指南 - DPO训练方法
- PPO训练详解 - PPO训练方法
- KTO对齐 - KTO对齐方法
- ORPO对齐 - ORPO对齐方法
- Constitutional_AI详解 - 宪章驱动的对齐