Como a IA aprende suas preferências? Desvendando o viés.

Aqui está o conteúdo reescrito no estilo do Lucas Tech, seguindo todas as suas instruções:

DPO Desvendado: Como Eu Ensinei uma IA a Escolher as MELHORES Respostas e Otimizei Modelos de Linguagem!

Olá, pessoal! Aqui é o Lucas Tech, e hoje a gente vai mergulhar de cabeça em um dos tópicos mais quentes do universo da Inteligência Artificial: o Direct Preference Optimization, ou DPO! Já pensou em como as IAs aprendem a ‘preferir’ uma resposta em vez de outra? Como a gente ensina elas a terem um ‘bom gosto’ e a serem realmente úteis pra gente?

Pois é, o DPO é a chave! Prepare-se para uma jornada incrível onde vamos construir um fluxo completo de aprendizado de preferências, usando um dataset superimportante da Anthropic. Vamos nessa?

Preparando o Terreno: Nosso Laboratório de DPO

A primeira coisa que fazemos em qualquer projeto sério de IA é preparar o ambiente, né? Imagine que estamos montando nosso laboratório de DPO aqui no Colab! Garanto que todas as bibliotecas necessárias estão instaladas – e olha que às vezes dá um trabalhinho com as dependências, mas a gente resolve! O importante é ter um lugar estável e configurado, com todas as ferramentas à mão, pra garantir que a gente possa focar no que realmente interessa: treinar nossa IA sem dores de cabeça.

A gente ainda dá uma olhada no hardware disponível (GPU, CPU), nos modos de precisão (bf16, fp16) e nas interfaces do TRL pra ter certeza que está tudo pronto.

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

Dissecando o Dataset Anthropic HH-RLHF

Agora que o ambiente está tinindo, é hora de apresentar a estrela da nossa base de dados: o Anthropic HH-RLHF! Ele é superimportante porque contém milhares de pares de respostas "escolhidas" e "rejeitadas". Basicamente, é como se tivéssemos várias conversas onde um humano disse: "Gostei mais dessa resposta do que daquela".

A gente carrega essas amostras, cria conjuntos de treino e teste bem equilibrados e, o mais importante, organiza cada conversa em mensagens estruturadas de usuário e assistente. É crucial garantir que a parte "escolhida" e a "rejeitada" tenham o mesmo início de conversa, sabe? Assim, a gente filtra qualquer par ‘problemático’ e trabalha só com exemplos de preferência bem alinhados. É como ter certeza que estamos comparando maçãs com maçãs!

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

Auditando as Preferências: Evitando ‘Trapaças’ da IA

Antes de treinar, a gente precisa ser um bom detetive! Vamos auditar esses pares de preferência para ver se não tem nenhuma ‘armadilha’. A gente analisa coisas como a diferença no tamanho das respostas (será que o modelo prefere só respostas mais longas?), a profundidade da conversa e até comportamentos específicos de cada parte do dataset.

Além disso, aplicamos um diagnóstico lexical usando TF-IDF e regressão logística. Isso é pra descobrir se a IA não está sendo ‘preguiçosa’, sabe? Tipo, aprendendo padrões linguísticos simples (como certas palavras ou frases) em vez de entender a verdadeira preferência por trás. Nosso objetivo é que ela aprenda o sinal de preferência real, e não um ‘atalho’ linguístico bobo!

php
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, targetnames=["rejected", "chosen"], digits=3))
perm = rng.permutation(len(labels))
, aucperm, , , = 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(aucperm – 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.")

Preparando os Dados e Montando o Treinador DPO

Com os dados auditados, é hora de ‘fatiar’ e ‘temperar’ tudo para o nosso DPO! Primeiro, a gente prepara o tokenizer – ele é quem transforma nosso texto em números que a IA consegue entender. Aplicamos um template de chat específico pra ele saber onde começa e termina cada turno de conversa.

Calculamos o tamanho de cada resposta em tokens e filtramos aquelas que são muito longas (não queremos sobrecarregar o modelo, certo?). O mais legal é que configuramos os parâmetros do DPO de forma dinâmica, adaptando-nos à versão do TRL que estamos usando – é flexibilidade pura! Depois, carregamos nosso modelo base (o Qwen2.5-0.5B-Instruct, no caso), ativamos o LoRA (que é uma forma super eficiente de treinar modelos gigantes sem precisar de uma supermáquina!) e montamos o nosso ‘treinador’ DPO, pronto para a ação!

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

A Hora da Verdade: Treinamento e Avaliação da Preferência

Agora é o momento mais esperado: o treinamento! A gente coloca o modelo para ‘aprender’ as preferências usando o DPO, com os parâmetros que configuramos. É aqui que ele ajusta seus neurônios para entender por que uma resposta é melhor que a outra.

Depois que ele treina, a gente não só cruza os dedos, mas avalia a ‘política’ de preferência que ele aprendeu em dados que ele nunca viu antes. Olhamos métricas como a ‘loss’ (o quão ‘errado’ ele está), as margens de recompensa e a precisão da recompensa. E claro, sempre bom ter um gráfico pra ver como o aprendizado evoluiu ao longo do tempo. É fascinante ver a IA se tornar mais ‘sábia’!

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

Análise Detalhada e Gerações Iniciais

Depois do treinamento, a gente não para por aí. Mergulhamos mais fundo para entender se o modelo realmente está preferindo as respostas certas, ou se ele está pegando algum ‘macete’. Calculamos a precisão da recompensa por cada ‘fonte’ do dataset e comparamos as probabilidades do nosso modelo com um modelo de referência.

Isso nos ajuda a ver se o que ele aprendeu é uma preferência genuína ou se ele está apenas optando por respostas mais longas, por exemplo. Por último, mas não menos importante, pedimos para a nossa IA gerar algumas respostas para ver, na prática, como ela se comporta depois de todo o treinamento. É sempre um momento de ‘uau’! E claro, salvamos tudo: o modelo treinado e o tokenizer, pra poder usar essa inteligência em projetos futuros.

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

Minha Visão

Olha, pessoal, essa jornada pelo DPO é mais do que um simples tutorial técnico. Pra mim, ela representa um passo GIGANTE na forma como construímos IAs. Não é só sobre fazer um modelo responder; é sobre fazer um modelo entender o que é uma boa resposta para nós. Auditar os dados, procurar vieses, garantir que a IA não pegue atalhos… tudo isso nos leva a modelos mais robustos, mais éticos e, acima de tudo, mais úteis. Chega de IAs que só dão respostas genéricas! Com DPO, estamos ensinando IAs a ter nuances, a ter ‘senso crítico’. É o futuro da interação humano-IA acontecendo agora!

E aí, o que vocês acham? Qual é o próximo passo para IAs que realmente entendem e nos servem melhor? Compartilhem suas ideias nos comentários!

Referência: Matéria Original