Anthropic HH-RLHF データの選好バイアス監査と DPO 微調整

In this tutorial, we design an end-to-end preference-learning workflow using the Anthropic HH-RLHF dataset and Direct Preference Optimization (DPO). We begin by preparing a robust Colab environment, loading and parsing chosen–rejected response pairs, and auditing the dataset for structural and length-based preference biases. We then run lexical shortcut diagnostics to determine whether surface-level linguistic patterns can separate preferred from rejected responses, prepare conversational data with tokenizer-aware length filtering, and construct a version-robust DPO training pipeline with TRL and optional LoRA adaptation. Finally, we fine-tune a Qwen2.5-0.5B-Instruct model, evaluate reward accuracy and training behavior, analyze performance across individual HH-RLHF subsets, inspect potential length bias, generate sample responses, and save the resulting policy for further experimentation.

Copy CodeCopiedUse a different Browser

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():

"""Install in ONE pip call so the resolver picks a mutually compatible set."""

try:

import trl

import transformers

return False

except ImportError:

print("Installing dependencies...")

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "-U", *REQUIRED])

return True

def drop_broken_torchao():

"""Colab ships torchao 0.10.0; peft demands >0.16 and raises rather than skipping.

Nothing here uses torchao, so removing it is safer than upgrading (an upgrade can

drag in a torch build that does not match this runtime)."""

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("Removing incompatible torchao (unused, but peft raises on it)...")

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("\nEnvironment changed. RESTART THE RUNTIME (Runtime > Restart session), "

"then run this cell again.")

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():

"""Belt and braces: if torchao survived the uninstall, stop peft raising on it."""

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" neutralising peft's torchao check ({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: {device} | bf16={bf16} | fp16={fp16}")

if not cuda:

print("CPU fallback is enabled; training is intentionally shortened.")

cfg_fields = {f.name for f in dataclasses.fields(DPOConfig)}

trainer_params = set(inspect.signature(DPOTrainer.__init__).parameters)

print(f"DPOConfig subclasses TrainingArguments : {issubclass(DPOConfig, TrainingArguments)}")

print(f"DPOConfig fields : {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 'NOT ACCEPTED ANYWHERE'}")

if not issubclass(DPOConfig, TrainingArguments) or "per_device_train_batch_size" not in cfg_fields:

print("\n!! DPOConfig looks broken. Reinstall in one command, then restart:")

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()

We set up the required libraries, handle dependency compatibility issues, and configure the main parameters used throughout the tutorial. We also initialize reproducibility settings and inspect the available hardware, precision modes, and installed TRL interfaces. This gives us a stable environment before we process the HH-RLHF dataset and train the preference model.

Copy CodeCopiedUse a different Browser

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}")

We load samples from the different Anthropic HH-RLHF subsets and create balanced training and testing datasets. We parse each conversation into structured user and assistant messages while ensuring that chosen and rejected responses share the same conversational prefix. We then filter invalid pairs so that we work only with properly aligned preference examples.

Copy CodeCopiedUse a different Browser

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.")

We analyze the preference pairs to measure differences in response length, conversation depth, and source-specific behavior. We also train a TF-IDF and logistic regression diagnostic to test whether simple lexical patterns can distinguish chosen responses from rejected ones. This helps us detect shortcuts that the language model could potentially exploit instead of learning the intended preference signal.

Copy CodeCopiedUse a different Browser

print("\nPreparing conversational DPO data...")

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("Tokenizer had no chat template; installed a ChatML fallback.")

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"DPO-ready rows after {MAX_LENGTH}-token filter -> train={len(dpo_train)}, test={len(dpo_test)}")

print("DPO schema:", 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 unsupported here -> converted to warmup_steps={steps}")

forwarded, truly_dropped = split_kwargs(dropped, TRAINER_PARAMS)

if forwarded:

print(" forwarded to DPOTrainer:", sorted(forwarded))

if truly_dropped:

print(" dropped (accepted nowhere in this build):", sorted(truly_dropped))

if "max_prompt_length" in truly_dropped:

print(" -> harmless: the token filter in section 7 already caps prompts")

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("\nBuilding DPOConfig for the installed TRL...")

args, forwarded_to_trainer = build_dpo_config(wanted_args)

print(" DPOConfig built OK")

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 enabled (the frozen base doubles as the reference model)")

except ImportError:

print(" peft not installed -> full fine-tune with an explicit reference model")

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("\nBuilding DPOTrainer...")

patch_peft_torchao()

model = build_model()

trainer = build_trainer(model, args, dpo_train, dpo_test, tok, peft_config, forwarded_to_trainer)

print(" DPOTrainer built OK")

We prepare the tokenizer, apply the conversational chat template, and calculate token lengths for every preference pair. We filter examples that exceed our prompt or total sequence limits and dynamically construct DPO configuration arguments based on the installed TRL version. We then load the base model, configure LoRA when available, and build the DPO trainer that we use for fine-tuning.

Copy CodeCopiedUse a different Browser

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()

We train the model using Direct Preference Optimization with the configured batch size, gradient accumulation, learning rate, and optimization steps. We evaluate the resulting policy on held-out preference pairs and inspect metrics such