Source: MarkTechPost
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.
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 = "https://www.marktechpost.com/content/dpo-hh" if os.path.isdir("https://www.marktechpost.com/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.
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"nn(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.
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.
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|>assistantn' }}{% 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.
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 as loss, reward margins, and reward accuracy. We also visualize the training history to observe how preference-learning performance changes throughout optimization.
@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. """) We calculate per-source reward accuracy and compare policy and reference-model log probabilities to examine whether the model genuinely prefers the chosen responses. We investigate the relationship between preference decisions and response-length differences, then generate sample answers from the tuned policy to inspect its behavior qualitatively. Finally, we save both the trained model and tokenizer so that we can reuse the resulting DPO policy in later experiments.
In conclusion, we developed a complete DPO-based preference-learning pipeline that goes beyond simply fine-tuning a language model on chosen and rejected responses. We examined the HH-RLHF data for length asymmetries and lexical shortcuts, enforced consistent conversational formatting and token limits, and used a flexible training setup that adapts to differences across TRL and Transformers versions. We also evaluated the tuned policy at both the aggregate and per-source levels, allowing us to identify whether improvements reflect genuine preference learning or undesirable shortcuts such as favoring longer answers. By combining dataset auditing, diagnostic analysis, efficient LoRA-based DPO training, reward evaluation, and generation testing, we established a framework for studying and improving preference alignment in language models.
Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
Sana Hassan
Sana Hassan, a consulting intern at Marktechpost and dual-degree student at IIT Madras, is passionate about applying technology and AI to address real-world challenges. With a keen interest in solving practical problems, he brings a fresh perspective to the intersection of AI and real-life solutions.