待翻译：Auditing Preference Biases and Fine-Tuning Language Models with Direct Preference Optimization on Anthropic HH-RLHF Using TRL and LoRA

AI 服务暂时不可用，以下为来源正文，待恢复后补全翻译。

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: {', '.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) 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) observed AUC is within permutation noise: no detectable lexical shortcut.") elif auc 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 %}" "{{ '' + m['role'] + '\n' + m['content'] + '\n' }}" "{% endfor %}" "{% if add_generation_prompt %}{{ 'assistant\n' }}{% endif %}" ) if getattr(tok, "chat_template", None) is None: tok.chat_tem [truncated for AI cost control]