MarkTechPost新闻
使用TRL和LoRA在Anthropic HH-RLHF上审计偏好偏差并通过直接偏好优化微调语言模型
marktechpost.com作者：Sana Hassan教程AI评分：50/100
本教程提供了一个端到端的工作流程，用于使用直接偏好优化（DPO）微调语言模型。我们演示如何审计Anthropic HH-RLHF数据集中的结构和长度偏差，使用TRL和LoRA实现稳健的训练流程，并评估模型性能以确保真正的偏好学习而非依赖词汇捷径。
在本教程中，我们使用Anthropic HH-RLHF数据集和直接偏好优化（DPO）设计了一个端到端的偏好学习工作流。我们首先准备一个稳健的Colab环境，加载并解析“选定—拒绝”响应对，并审计数据集中存在的结构性和基于长度的偏好偏差。随后，我们运行词汇捷径诊断，以判断表面层面的语言模式是否能区分优选响应与拒绝响应，使用考虑分词器的长度过滤准备对话数据，并构建一个版本稳健的、基于TRL并可选LoRA适配的DPO训练流程。最后，我们对Qwen2.5-0.5B-Instruct模型进行微调，评估奖励准确性和训练行为，分析模型在各HH-RLHF子集上的表现，检查潜在的长度偏差，生成示例响应，并保存最终策略以供进一步实验使用。
import dataclasses
import importlib.util
import inspect
import os
import re
import subprocess
import sys
import warnings
warnings.filterwarnings("ignore", category=UserWarning)
REQUIRED = ["trl>=0.12", "transformers>=4.45", "accelerate", "datasets", "peft", "scikit-learn"]
def ensure_deps():
"""在单次pip调用中安装，以便解析器选择一组相互兼容的包。"""
try:
import trl
import transformers
return False
except ImportError:
print("正在安装依赖项...")
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "-U", *REQUIRED])
return True
def drop_broken_torchao():
"""Colab自带torchao 0.10.0；peft要求版本>0.16，否则会直接报错而不是跳过。
此处未使用torchao，因此移除它比升级更安全（升级可能会引入与此运行时不匹配的torch构建）。"""
if importlib.util.find_spec("torchao") is None:
return False
try:
from peft.import_utils import is_torchao_available
is_torchao_available()
return False
except ImportError:
print("正在移除不兼容的torchao（未使用，但peft会因其报错）...")
subprocess.call([sys.executable, "-m", "pip", "uninstall", "-y", "-q", "torchao"])
return True
except Exception:
return False
_installed = ensure_deps()
_removed = drop_broken_torchao() if not _installed else False
if _installed or _removed:
print("\n环境已更改。请重启运行时（运行时 > 重启会话），然后重新运行此单元格。")
raise SystemExit(0)
import numpy as np
import pandas as pd
import torch
import matplotlib.pyplot as plt
from datasets import load_dataset, concatenate_datasets
from transformers import AutoModelForCausalLM, AutoTokenizer, set_seed
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, classification_report, roc_auc_score
import transformers
import trl
from trl import DPOConfig, DPOTrainer
def patch_peft_torchao():
"""双保险：如果torchao在卸载后仍然存在，阻止peft因其报错。"""
try:
from peft import import_utils
from peft.tuners.lora import torchao as lora_torchao
except ImportError:
return
try:
import_utils.is_torchao_available()
except ImportError as exc:
print(f" 正在中和peft的torchao检查（{exc}）")
import_utils.is_torchao_available = lambda: False
lora_torchao.is_torchao_available = lambda: False
MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"
SUBSETS = ["helpful-base", "helpful-rejection-sampled", "helpful-online", "harmless-base"]
N_TRAIN_PER_SUBSET = 120
N_TEST_PER_SUBSET = 30
MAX_LENGTH = 512
MAX_PROMPT_LENGTH = 256
BETA = 0.1
MAX_STEPS = 30
BATCH_SIZE = 1
GRAD_ACCUM = 8
LEARNING_RATE = 5e-6
WARMUP_RATIO = 0.1
LOGGING_STEPS = 5
USE_LORA = True
N_REWARD_EVAL = 40
SEED = 17
OUTPUT_DIR = "/content/dpo-hh" if os.path.isdir("/content") else "./dpo-hh"
set_seed(SEED)
rng = np.random.default_rng(SEED)
def report_environment():
from transformers import TrainingArguments
cuda = torch.cuda.is_available()
bf16 = bool(cuda and torch.cuda.is_bf16_supported())
fp16 = bool(cuda and not bf16)
device = "cuda" if cuda else "cpu"
print(f"python : {sys.version.split()[0]}")
print(f"torch : {torch.__version__}")
print(f"transformers : {transformers.__version__}")
print(f"trl : {trl.__version__}")
print(f"设备: {device} | bf16={bf16} | fp16={fp16}")
if not cuda:
print("已启用CPU回退；训练时长已相应缩短。")
cfg_fields = {f.name for f in dataclasses.fields(DPOConfig)}
trainer_params = set(inspect.signature(DPOTrainer.__init__).parameters)
print(f"DPOConfig继承自TrainingArguments : {issubclass(DPOConfig, TrainingArguments)}")
print(f"DPOConfig字段数量 : {len(cfg_fields)}")
for probe in ("warmup_ratio", "warmup_steps", "beta", "max_length", "max_prompt_length"):
where = [c for c, s in (("DPOConfig", cfg_fields), ("DPOTrainer", trainer_params))
if probe in s]
print(f" {probe:<20} -> {', '.join(where) if where else '任何位置均不接受'}")
if not issubclass(DPOConfig, TrainingArguments) or "per_device_train_batch_size" not in cfg_fields:
print("\n!! DPOConfig似乎有问题。请用一条命令重新安装，然后重启：")
print(" pip install -U trl transformers accelerate datasets peft")
return device, bf16, fp16, cfg_fields, trainer_params
DEVICE, BF16, FP16, CFG_FIELDS, TRAINER_PARAMS = report_environment()我们配置所需的库，处理依赖兼容性问题，并设置本教程中使用的主要参数。同时，我们初始化可复现性设置，并检查可用的硬件、精度模式以及已安装的TRL接口。这样，在处理HH-RLHF数据集和训练偏好模型之前，我们就拥有了一个稳定的环境。
def sample_split(ds, n, seed):
return ds.shuffle(seed=seed).select(range(min(n, len(ds)))).flatten_indices()
def load_hh():
train_parts, test_parts = [], []
for i, subset in enumerate(SUBSETS):
ds = load_dataset("Anthropic/hh-rlhf", data_dir=subset)
tr = sample_split(ds["train"], N_TRAIN_PER_SUBSET, SEED + i)
te = sample_split(ds["test"], N_TEST_PER_SUBSET, SEED + i)
train_parts.append(tr.add_column("source", [subset] * len(tr)))
test_parts.append(te.add_column("source", [subset] * len(te)))
return concatenate_datasets(train_parts), concatenate_datasets(test_parts)
raw_train, raw_test = load_hh()
print(f"\nRaw sampled rows -> train={len(raw_train)}, test={len(raw_test)}")
print(pd.Series(raw_train["source"]).value_counts().sort_index().to_string())
TURN_RE = re.compile(r"\n\n(Human|Assistant):[ ]?")
def parse_transcript(text):
if not isinstance(text, str) or not text.strip():
return None
parts = TURN_RE.split(text)
if parts[0].strip():
return None
roles, contents = parts[1::2], parts[2::2]
if len(roles) != len(contents) or len(roles) < 2:
return None
msgs = [{"role": "user" if r == "Human" else "assistant", "content": c.strip()}
for r, c in zip(roles, contents)]
if msgs[0]["role"] != "user" or msgs[-1]["role"] != "assistant":
return None
if any(a["role"] == b["role"] for a, b in zip(msgs, msgs[1:])):
return None
if any(not m["content"] for m in msgs):
return None
return msgs
def to_pair(row):
ch = parse_transcript(row["chosen"])
rj = parse_transcript(row["rejected"])
ok = ch is not None and rj is not None and ch[:-1] == rj[:-1]
return {
"ok": bool(ok),
"prompt": ch[:-1] if ok else [],
"chosen": [ch[-1]] if ok else [],
"rejected": [rj[-1]] if ok else [],
"prompt_turns": len(ch) - 1 if ok else 0,
"source": row["source"],
}
parsed_train = raw_train.map(to_pair, remove_columns=raw_train.column_names).filter(lambda r: r["ok"])
parsed_test = raw_test.map(to_pair, remove_columns=raw_test.column_names).filter(lambda r: r["ok"])
print(f"\nValid parsed rows -> train={len(parsed_train)}, test={len(parsed_test)}")
identical = sum(1 for c, r in zip(parsed_train["chosen"], parsed_train["rejected"])
if c[0]["content"] == r[0]["content"])
print(f"Identical completion pairs in sampled train: {identical}")我们从Anthropic HH-RLHF的不同子集中加载样本，并创建平衡的训练和测试数据集。我们将每个对话解析为结构化的用户和助手消息，同时确保被选中的和被拒绝的回复共享相同的前置对话内容。然后，我们过滤掉无效的配对，以便只使用正确对齐的偏好示例。
audit = pd.DataFrame({
"source": parsed_train["source"],
"prompt_turns": parsed_train["prompt_turns"],
"chosen_words": [len(c[0]["content"].split()) for c in parsed_train["chosen"]],
"rejected_words": [len(r[0]["content"].split()) for r in parsed_train["rejected"]],
})
audit["length_delta"] = audit["chosen_words"] - audit["rejected_words"]
summary = audit.groupby("source").agg(
pairs=("chosen_words", "size"),
chosen_words_mean=("chosen_words", "mean"),
rejected_words_mean=("rejected_words", "mean"),
median_turns=("prompt_turns", "median"),
mean_length_delta=("length_delta", "mean"),
).round(2)
print("\nPreference-pair audit:")
print(summary.to_string())
fig, axes = plt.subplots(1, 2, figsize=(11, 4))
summary["mean_length_delta"].plot(kind="barh", ax=axes[0], color="#4c72b0")
axes[0].axvline(0, color="0.3", lw=1)
axes[0].set_title("mean(chosen − rejected) words")
axes[0].set_ylabel("")
for src, grp in audit.groupby("source"):
axes[1].hist(grp["length_delta"], bins=30, histtype="step", lw=1.6, label=src)
axes[1].axvline(0, color="0.3", lw=1)
axes[1].set_title("per-pair length delta")
axes[1].legend(fontsize=7)
plt.tight_layout()
plt.show()
print("\nSanitized structural preview (user text is not printed):")
for i in range(min(3, len(audit))):
r = audit.iloc[i]
print({"source": r["source"], "prompt_turns": int(r["prompt_turns"]),
"chosen_words": int(r["chosen_words"]), "rejected_words": int(r["rejected_words"])})
def build_lexical_dataset(ds):
chosen_txt = [c[0]["content"] for c in ds["chosen"]]
rejected_txt = [r[0]["content"] for r in ds["rejected"]]
texts = chosen_txt + rejected_txt
labels = np.concatenate([np.ones(len(chosen_txt), int), np.zeros(len(rejected_txt), int)])
pair_id = np.concatenate([np.arange(len(chosen_txt)), np.arange(len(rejected_txt))])
assert texts[: len(chosen_txt)] == chosen_txt and labels[: len(chosen_txt)].all()
assert not labels[len(chosen_txt):].any()
return np.array(texts, dtype=object), labels, pair_id
def run_lexical_diagnostic(texts, labels, pair_id, tag="observed"):
pairs = np.unique(pair_id)
shuffled = rng.permutation(pairs)
test_pairs = set(shuffled[: len(shuffled) // 2].tolist())
is_test = np.array([p in test_pairs for p in pair_id])
vec = TfidfVectorizer(ngram_range=(1, 2), min_df=2, max_features=20000, sublinear_tf=True)
Xtr = vec.fit_transform(texts[~is_test])
Xte = vec.transform(texts[is_test])
clf = LogisticRegression(max_iter=2000).fit(Xtr, labels[~is_test])
pred = clf.predict(Xte)
prob = clf.predict_proba(Xte)[:, 1]
acc = accuracy_score(labels[is_test], pred)
auc = roc_auc_score(labels[is_test], prob)
print(f"Lexical diagnostic ({tag}) accuracy: {acc:.3f}")
print(f"Lexical diagnostic ({tag}) ROC-AUC: {auc:.3f}")
return acc, auc, clf, labels[is_test], pred
print("\nTraining a lexical diagnostic to detect easy preference shortcuts...")
texts, labels, pair_id = build_lexical_dataset(parsed_train)
acc, auc, clf, y_true, y_pred = run_lexical_diagnostic(texts, labels, pair_id)
print(classification_report(y_true, y_pred, target_names=["rejected", "chosen"], digits=3))
perm = rng.permutation(len(labels))
_, auc_perm, _, _, _ = run_lexical_diagnostic(texts, labels[perm], pair_id, tag="permuted labels")
print(f"Chance baseline from permuted labels: AUC {auc_perm:.3f}")
if abs(auc - 0.5) <= abs(auc_perm - 0.5) + 0.02:
print("-> observed AUC is within permutation noise: no detectable lexical shortcut.")
elif auc < 0.5:
print("-> observed AUC is BELOW chance beyond noise: inspect label ordering upstream.")
else:
print("-> observed AUC is ABOVE chance: a real lexical shortcut exists in this sample.")
coefs = np.sort(np.abs(clf.coef_.ravel()))[-20:]
print(f"Top-20 absolute lexical coefficient range: {coefs[0]:.3f} to {coefs[-1]:.3f}")
print("Feature strings are intentionally not printed because the source corpus may contain offensive text.")我们分析偏好对，以衡量回复长度、对话深度和来源特定行为之间的差异。我们还训练了一个TF-IDF和逻辑回归诊断模型，以测试简单的词汇模式能否区分被选中的回复和被拒绝的回复。这有助于我们发现语言模型可能利用的捷径，而不是学习预期的偏好信号。
print("\n准备对话式DPO数据...")
tok = AutoTokenizer.from_pretrained(MODEL_ID)
if tok.pad_token is None:
tok.pad_token = tok.eos_token
CHATML = (
"{% for m in messages %}"
"{{ '<|im_start|>' + m['role'] + '\n' + m['content'] + '<|im_end|>\n' }}"
"{% endfor %}"
"{% if add_generation_prompt %}{{ '<|im_start|>assistant\n' }}{% endif %}"
)
if getattr(tok, "chat_template", None) is None:
tok.chat_template = CHATML
print("分词器没有聊天模板；已安装ChatML回退方案。")
def add_lengths(row):
prompt_txt = tok.apply_chat_template(row["prompt"], tokenize=False, add_generation_prompt=True)
n_prompt = len(tok(prompt_txt, add_special_tokens=False)["input_ids"])
n_ch = len(tok(row["chosen"][0]["content"], add_special_tokens=False)["input_ids"])
n_rj = len(tok(row["rejected"][0]["content"], add_special_tokens=False)["input_ids"])
return {"n_prompt": n_prompt, "n_total": n_prompt + max(n_ch, n_rj)}
def fits(row):
return row["n_prompt"] <= MAX_PROMPT_LENGTH and row["n_total"] <= MAX_LENGTH
dpo_train_full = parsed_train.map(add_lengths).filter(fits)
dpo_test_full = parsed_test.map(add_lengths).filter(fits)
test_sources = list(dpo_test_full["source"])
test_prompts = list(dpo_test_full["prompt"])
test_chosen = list(dpo_test_full["chosen"])
test_rejected = list(dpo_test_full["rejected"])
DPO_COLS = ["prompt", "chosen", "rejected"]
dpo_train = dpo_train_full.remove_columns([c for c in dpo_train_full.column_names if c not in DPO_COLS])
dpo_test = dpo_test_full.remove_columns([c for c in dpo_test_full.column_names if c not in DPO_COLS])
print(f"经过{MAX_LENGTH}-token过滤后的DPO就绪行数 -> train={len(dpo_train)}, test={len(dpo_test)}")
print("DPO模式:", dict(dpo_train.features))
def split_kwargs(wanted, valid):
return ({k: v for k, v in wanted.items() if k in valid},
{k: v for k, v in wanted.items() if k not in valid})
def build_dpo_config(wanted):
kept, dropped = split_kwargs(wanted, CFG_FIELDS)
if "warmup_ratio" in dropped and "warmup_steps" in CFG_FIELDS:
steps = max(1, int(dropped.pop("warmup_ratio") * wanted.get("max_steps", 100)))
kept["warmup_steps"] = steps
print(f" 此处不支持warmup_ratio -> 已转换为warmup_steps={steps}")
forwarded, truly_dropped = split_kwargs(dropped, TRAINER_PARAMS)
if forwarded:
print(" 已转发给DPOTrainer:", sorted(forwarded))
if truly_dropped:
print(" 已丢弃（此构建中任何位置均不接受）:", sorted(truly_dropped))
if "max_prompt_length" in truly_dropped:
print(" -> 无影响：第7节中的token过滤器已限制提示长度")
return DPOConfig(**kept), forwarded
wanted_args = dict(
output_dir=OUTPUT_DIR,
max_steps=MAX_STEPS,
per_device_train_batch_size=BATCH_SIZE,
per_device_eval_batch_size=BATCH_SIZE,
gradient_accumulation_steps=GRAD_ACCUM,
learning_rate=LEARNING_RATE,
warmup_ratio=WARMUP_RATIO,
logging_steps=LOGGING_STEPS,
save_strategy="no",
report_to=[],
remove_unused_columns=False,
bf16=BF16,
fp16=FP16,
seed=SEED,
beta=BETA,
max_length=MAX_LENGTH,
max_prompt_length=MAX_PROMPT_LENGTH,
)
print("\n为已安装的TRL构建DPOConfig...")
args, forwarded_to_trainer = build_dpo_config(wanted_args)
print(" DPOConfig构建成功")
def build_model():
dtype = torch.bfloat16 if BF16 else torch.float32
try:
model = AutoModelForCausalLM.from_pretrained(MODEL_ID, dtype=dtype)
except TypeError:
model = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=dtype)
model.config.use_cache = False
return model
peft_config = None
if USE_LORA:
try:
from peft import LoraConfig
peft_config = LoraConfig(
r=16, lora_alpha=32, lora_dropout=0.05, bias="none",
task_type="CAUSAL_LM",
)
print(" 已启用LoRA（冻结的基础模型兼作参考模型）")
except ImportError:
print(" 未安装peft -> 使用显式参考模型进行全量微调")
def build_trainer(model, args, train_ds, eval_ds, tokenizer, peft_config, extra):
kwargs = dict(model=model, args=args, train_dataset=train_ds, eval_dataset=eval_ds)
if "processing_class" in TRAINER_PARAMS:
kwargs["processing_class"] = tokenizer
elif "tokenizer" in TRAINER_PARAMS:
kwargs["tokenizer"] = tokenizer
if peft_config is not None and "peft_config" in TRAINER_PARAMS:
kwargs["peft_config"] = peft_config
elif peft_config is None and "ref_model" in TRAINER_PARAMS:
kwargs["ref_model"] = None
kwargs.update(extra)
print(" DPOTrainer kwargs:", sorted(kwargs))
return DPOTrainer(**kwargs)
print("\n构建DPOTrainer...")
patch_peft_torchao()
model = build_model()
trainer = build_trainer(model, args, dpo_train, dpo_test, tok, peft_config, forwarded_to_trainer)
print(" DPOTrainer构建成功")我们准备分词器，应用对话式聊天模板，并计算每个偏好对的词元长度。我们过滤掉超出提示或总序列限制的样本，并根据已安装的 TRL 版本动态构建 DPO 配置参数。然后加载基础模型，在可用时配置 LoRA，并构建用于微调的 DPO 训练器。
print(f"\nTraining for {MAX_STEPS} steps on {DEVICE} "
f"(effective batch {BATCH_SIZE * GRAD_ACCUM})...")
train_result = trainer.train()
print("\nTraining metrics:")
for k, v in sorted(train_result.metrics.items()):
print(f" {k:<28} {v}")
print("\nEvaluating on held-out pairs...")
eval_metrics = trainer.evaluate()
for k, v in sorted(eval_metrics.items()):
if any(t in k for t in ("accuracies", "margins", "rewards", "loss")):
print(f" {k:<34} {v:.4f}" if isinstance(v, float) else f" {k:<34} {v}")
log_df = pd.DataFrame(trainer.state.log_history)
if "loss" in log_df.columns:
fig, ax = plt.subplots(figsize=(7, 3.5))
d = log_df.dropna(subset=["loss"])
ax.plot(d["step"], d["loss"], marker="o", ms=3, label="train loss")
acc_col = next((c for c in log_df.columns if c.endswith("rewards/accuracies")), None)
if acc_col:
d2 = log_df.dropna(subset=[acc_col])
ax.plot(d2["step"], d2[acc_col], marker="s", ms=3, label="reward accuracy")
ax.axhline(0.5, color="0.6", lw=0.8, ls="--")
ax.set_xlabel("step")
ax.legend(fontsize=8)
ax.set_title("DPO training")
plt.tight_layout()
plt.show()我们使用直接偏好优化，以配置的批量大小、梯度累积、学习率和优化步数来训练模型。我们在保留的偏好对上评估所得策略，并检查损失、奖励边际和奖励准确率等指标。我们还可视化训练历史，以观察偏好学习性能在整个优化过程中的变化。
@torch.no_grad()
def completion_logprob(policy, messages_prompt, message_completion, use_ref=False):
prompt_txt = tok.apply_chat_template(messages_prompt, tokenize=False, add_generation_prompt=True)
full_txt = prompt_txt + message_completion["content"] + tok.eos_token
p_ids = tok(prompt_txt, add_special_tokens=False, return_tensors="pt")["input_ids"]
f_ids = tok(full_txt, add_special_tokens=False, return_tensors="pt",
truncation=True, max_length=MAX_LENGTH)["input_ids"].to(policy.device)
start = min(p_ids.shape[1], f_ids.shape[1] - 1)
ctx = policy.disable_adapter() if (use_ref and hasattr(policy, "disable_adapter")) else None
if ctx is not None:
with ctx:
logits = policy(f_ids).logits
else:
logits = policy(f_ids).logits
logprobs = torch.log_softmax(logits[:, :-1].float(), dim=-1)
targets = f_ids[:, 1:]
picked = logprobs.gather(-1, targets.unsqueeze(-1)).squeeze(-1)
return picked[:, start:].sum().item()
def per_source_reward_accuracy(n=N_REWARD_EVAL):
policy = trainer.model
policy.eval()
if not hasattr(policy, "disable_adapter") and getattr(trainer, "ref_model", None) is None:
print(" no reference model reachable; skipping per-source analysis")
return None
idx = rng.permutation(len(test_sources))[:min(n, len(test_sources))]
rows = []
for i in idx:
i = int(i)
rc = completion_logprob(policy, test_prompts[i], test_chosen[i][0])
rr = completion_logprob(policy, test_prompts[i], test_rejected[i][0])
refc = completion_logprob(policy, test_prompts[i], test_chosen[i][0], use_ref=True)
refr = completion_logprob(policy, test_prompts[i], test_rejected[i][0], use_ref=True)
rows.append({
"source": test_sources[i],
"margin": BETA * ((rc - refc) - (rr - refr)),
"correct": BETA * ((rc - refc) - (rr - refr)) > 0,
"len_delta": len(test_chosen[i][0]["content"].split())
- len(test_rejected[i][0]["content"].split()),
})
df = pd.DataFrame(rows)
out = df.groupby("source").agg(
n=("correct", "size"),
reward_accuracy=("correct", "mean"),
mean_margin=("margin", "mean"),
mean_len_delta=("len_delta", "mean"),
).round(3)
print(out.to_string())
longer_wins = (df["correct"] == (df["len_delta"] > 0)).mean()
print(f"\n agreement between 'model prefers chosen' and 'chosen is longer': {longer_wins:.3f}")
print(" (near 0.5 = no length shortcut; near 1.0 = the policy is mostly ranking by length)")
return out
print(f"\nPer-source reward accuracy on {N_REWARD_EVAL} held-out pairs:")
try:
per_source = per_source_reward_accuracy()
except Exception as exc:
print(f" skipped: {type(exc).__name__}: {exc}")
per_source = None
def generate(messages, max_new_tokens=96):
text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
enc = tok(text, return_tensors="pt").to(trainer.model.device)
with torch.no_grad():
out = trainer.model.generate(**enc, max_new_tokens=max_new_tokens, do_sample=False,
pad_token_id=tok.pad_token_id)
return tok.decode(out[0][enc["input_ids"].shape[1]:], skip_special_tokens=True).strip()
probes = [
[{"role": "user", "content": "My laptop fan is suddenly very loud. What should I check first?"}],
[{"role": "user", "content": "Explain in two sentences why DPO does not need a separate reward model."}],
]
print("\nSample generations from the tuned policy:")
for p in probes:
print(f"\n user : {p[0]['content']}")
print(f" assistant : {generate(p)}")
trainer.save_model(OUTPUT_DIR)
tok.save_pretrained(OUTPUT_DIR)
print(f"\nSaved to {OUTPUT_DIR}")
print("""
Reading the results
* At MAX_STEPS=30 on CPU this is a smoke test, not a trained model. Reward accuracy
near 0.5 is the expected outcome; raise MAX_STEPS on a GPU before concluding anything.
* The number to watch is the per-source table, not the aggregate. If harmless-base
reward accuracy drops while the helpful subsets rise, the policy is learning the
length asymmetry visible in the section 5 audit rather than the intended preference.
* If a TRL call fails after an upgrade, the environment report at the top names which
class accepts which argument in your build; extend `wanted_args` from that list.
""")我们计算每个来源的奖励准确率，并比较策略模型与参考模型的对数概率，以检验模型是否真正偏好被选中的回答。我们考察偏好决策与回答长度差异之间的关系，然后从调优后的策略中生成示例回答，以定性检查其行为。最后，我们保存训练好的模型和分词器，以便在后续实验中可以复用得到的 DPO 策略。
总之，我们开发了一套完整的基于DPO的偏好学习流程，其意义超越了简单地在选定和拒绝的响应上对语言模型进行微调。我们检查了HH-RLHF数据中的长度不对称性和词汇捷径，强制执行了一致的对话格式和令牌限制，并采用了能够适应TRL和Transformers版本差异的灵活训练设置。我们还分别在总体层面和按来源层面评估了调整后的策略，从而能够判断改进是反映了真正的偏好学习，还是依赖于诸如偏好较长答案之类的不良捷径。通过结合数据集审计、诊断分析、基于LoRA的高效DPO训练、奖励评估和生成测试，我们建立了一个研究和改进语言模型偏好对齐的框架。
需要与我们合作推广您的GitHub仓库、Hugging Face页面、产品发布或网络研讨会吗？请联系我们。
Sana Hassan是Marktechpost的咨询实习生，也是印度马德拉斯理工学院（IIT Madras）的双学位学生，热衷于将技术和人工智能应用于解决现实世界的挑战。他对解决实际问题有着浓厚的兴趣，为人工智能与现实解决方案的交汇带来了全新的视角。