| name | rhlf-preference |
| description | Guide complet du Reinforcement Learning from Human Feedback et optimisation de préférences — RLHF, DPO, KTO, IPO, ORPO, SimPO, récompenses, alignment. En français. |
RLHF & Preference Optimization — Guide Complet
De RLHF à DPO, aligner les LLM avec les préférences humaines.
1. Le Problème de l'Alignment
Pipeline d'Alignment
1. Pré-entraînement (next token prediction)
↓
2. SFT (Supervised Fine-Tuning) — imiter des démonstrations
↓
3. RM (Reward Model) — apprendre les préférences humaines
↓
4. RL fine-tuning (PPO/GRPO) — optimiser vers RM
ou
DPO — alignment direct (sans RM explicite)
2. Collecte des Préférences
Format Bradley-Terry
Types de données de préférence
| Type | Description | Exemple |
|---|
| Paires (chosen/rejected) | Comparaison binaire | A > B |
| Ranking | Classement de N réponses | A > B > C > D |
| Ratings | Score sur échelle | 1-5 étoiles |
| Binaire | Accept/Reject | ✓ / ✗ |
| Langue naturelle | Feedback textuel | "Trop long, sois concis" |
3. Reward Model Training
class RewardModel(nn.Module):
"""Modèle de récompense : prédit un score pour (prompt, réponse)."""
def __init__(self, base_model, dropout=0.1):
super().__init__()
self.base_model = base_model
self.value_head = nn.Sequential(
nn.Linear(base_model.config.hidden_size, 1024),
nn.Dropout(dropout),
nn.ReLU(),
nn.Linear(1024, 1),
)
def forward(self, input_ids, attention_mask):
outputs = self.base_model(input_ids, attention_mask=attention_mask,
output_hidden_states=True)
hidden = outputs.hidden_states[-1]
last_indices = attention_mask.sum(dim=1) - 1
last_hidden = hidden[torch.arange(hidden.size(0)), last_indices]
reward = self.value_head(last_hidden).squeeze(-1)
return reward
def train_reward_model(reward_model, dataloader, optimizer):
"""Entraîne le réward model sur des paires (chosen, rejected).
Loss : -log σ( r(x, y_w) - r(x, y_l) )
"""
reward_model.train()
total_loss = 0
for batch in dataloader:
reward_chosen = reward_model(batch['chosen_ids'], batch[])
reward_rejected = reward_model(batch[], batch[])
loss = -F.logsigmoid(reward_chosen - reward_rejected).mean()
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
total_loss / (dataloader)
4. RLHF (PPO avec Reward Model)
class PPOWithRM(nn.Module):
"""RLHF : PPO avec un réward model pré-entraîné.
Composants :
- Policy (π_θ) : le LLM qu'on entraîne
- Value Model (V_φ) : estimateur de valeur
- Reward Model (r_ψ) : gelé, donne les récompenses
- Reference Model (π_ref) : gelé, KL divergence
"""
def __init__(self, policy, value_model, reward_model, ref_model,
kl_coef=0.1, clip_range=0.2):
super().__init__()
self.policy = policy
self.value_model = value_model
self.reward_model = reward_model
self.ref_model = ref_model
self.kl_coef = kl_coef
self.clip_range = clip_range
def compute_rewards(self, prompts, responses, masks):
"""Calcule la récompense totale = RM score - KL penalty."""
with torch.no_grad():
scores = self.reward_model(prompts, responses, masks)
log_probs = self.policy.get_log_probs(prompts, responses)
ref_log_probs = self.ref_model.get_log_probs(prompts, responses)
kl = log_probs - ref_log_probs
kl_penalty = self.kl_coef * kl
rewards = scores.unsqueeze(-) - kl_penalty
rewards
():
prompts, responses, masks = batch
log_probs, values = .policy(prompts, responses, masks)
rewards = .compute_rewards(prompts, responses, masks)
advantages, returns = .compute_gae(rewards, values, masks)
ratio = torch.exp(log_probs - old_log_probs)
surr1 = ratio * advantages
surr2 = torch.clamp(ratio, - .clip_range, + .clip_range) * advantages
policy_loss = -torch.(surr1, surr2).mean()
value_loss = F.mse_loss(values, returns)
policy_loss + * value_loss
Problèmes du RLHF
5. DPO — Direct Preference Optimization (Rafailov et al., 2023)
Formulation mathématique
Implémentation
class DPOTrainer:
"""Direct Preference Optimization — version complète."""
def __init__(self, model, ref_model, beta=0.1, lr=1e-6):
self.model = model
self.ref_model = ref_model
self.beta = beta
self.optimizer = torch.optim.AdamW(model.parameters(), lr=lr)
def get_log_probs(self, model, input_ids, attention_mask, labels):
"""Calcule les log-probabilités des tokens labels."""
outputs = model(input_ids, attention_mask=attention_mask)
logits = outputs.logits
shift_logits = logits[..., :-1, :].contiguous()
shift_labels = labels[..., 1:].contiguous()
shift_mask = attention_mask[..., 1:].contiguous()
log_probs = -F.cross_entropy(
shift_logits.view(-1, shift_logits.size(-1)),
shift_labels.view(-1),
reduction='none',
).view_as(shift_labels)
log_probs = log_probs * shift_mask
return log_probs.sum(dim=-1) / shift_mask.sum(dim=-1)
def dpo_loss(self, batch):
policy_chosen_logps = .get_log_probs(
.model, batch[], batch[], batch[])
policy_rejected_logps = .get_log_probs(
.model, batch[], batch[], batch[])
torch.no_grad():
ref_chosen_logps = .get_log_probs(
.ref_model, batch[], batch[], batch[])
ref_rejected_logps = .get_log_probs(
.ref_model, batch[], batch[], batch[])
pi_logratios = policy_chosen_logps - policy_rejected_logps
ref_logratios = ref_chosen_logps - ref_rejected_logps
beta_logratios = .beta * (pi_logratios - ref_logratios)
losses = -F.logsigmoid(beta_logratios)
loss = losses.mean()
chosen_reward = .beta * (policy_chosen_logps - ref_chosen_logps).detach()
rejected_reward = .beta * (policy_rejected_logps - ref_rejected_logps).detach()
accuracy = (chosen_reward > rejected_reward).().mean()
{
: loss,
: chosen_reward.mean().item(),
: rejected_reward.mean().item(),
: accuracy.item(),
}
():
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=)
epoch (num_epochs):
batch dataloader:
metrics = .dpo_loss(batch)
metrics[].backward()
torch.nn.utils.clip_grad_norm_(.model.parameters(), )
.optimizer.step()
.optimizer.zero_grad()
(
)
6. KTO — Kahneman-Tversky Optimization (2024)
def kto_loss(policy_logps, ref_logps, y_is_good, beta=0.1):
"""
KTO Loss :
L = 1 - σ(β · (log(π/π_ref) - z_0)) pour y_good
L = 1 - σ(-β · (log(π/π_ref) - z_0)) pour y_bad
Où z_0 = β · KL(π || π_ref) (bias de référence)
"""
log_ratio = policy_logps - ref_logps
kl = (log_ratio.exp() - 1 - log_ratio).mean()
z0 = beta * kl.detach()
loss_good = (1 - F.sigmoid(beta * (log_ratio - z0))) * y_is_good
loss_bad = (1 - F.sigmoid(-beta * (log_ratio - z0))) * (1 - y_is_good)
return (loss_good + loss_bad).mean()
7. IPO — Identity Preference Optimization (2023)
def ipo_loss(policy_logps, ref_logps, beta=0.1):
"""Identity Preference Optimization.
L = (log(π/π_ref) - 1/(2β))² pour la paire (chosen, rejected)
"""
log_ratio = policy_logps - ref_logps
tau = 0.5 * beta
loss = (log_ratio - 1/tau) ** 2
return loss.mean()
8. ORPO — Odds Ratio Preference Optimization (2024)
def orpo_loss(policy_logps_chosen, policy_logps_rejected, lambda_coef=0.1):
"""Odds Ratio Preference Optimization.
L = -log π(y_w|x) + λ · log(1 + odds_ratio)
Où odds = π(y|x) / (1 - π(y|x))
OR = odds(y_w) / odds(y_rejected)
"""
sft_loss = -policy_logps_chosen.mean()
odds_chosen = torch.exp(policy_logps_chosen)
odds_rejected = torch.exp(policy_logps_rejected)
log_odds_ratio = torch.log(odds_chosen / odds_rejected)
or_loss = -F.logsigmoid(log_odds_ratio).mean()
return sft_loss + lambda_coef * or_loss
9. SimPO — Simple Preference Optimization (2024)
def simpo_loss(policy_logps_chosen, policy_logps_rejected,
beta=2.0, gamma_beta_ratio=1.0):
"""Simple Preference Optimization.
L = -log σ(β · (avg_log_prob_chosen - avg_log_prob_rejected) - γ)
Où γ = reward margin (target gap between chosen and rejected)
"""
avg_chosen = policy_logps_chosen.mean(dim=-1)
avg_rejected = policy_logps_rejected.mean(dim=-1)
reward_margin = avg_chosen - avg_rejected - gamma_beta_ratio
loss = -F.logsigmoid(beta * reward_margin).mean()
return loss
10. GRPO — Group Relative Policy Optimization (DeepSeek-R1, 2025)
class GRPOTrainer:
"""Group Relative Policy Optimization.
Utilisé par DeepSeek-R1 pour le reasoning RL.
Pas de critic (value model) — utilise un groupe d'échantillons.
"""
def __init__(self, model, ref_model=None, group_size=8,
beta=0.04, clip_epsilon=0.2):
self.model = model
self.ref_model = ref_model
self.group_size = group_size
self.beta = beta
self.clip_epsilon = clip_epsilon
def compute_group_advantage(self, rewards):
"""Avantage = (reward - mean(G)) / std(G) normalisé intra-groupe."""
rewards = torch.tensor(rewards)
return (rewards - rewards.mean()) / (rewards.std() + 1e-8)
def train_step(self, prompts, group_size=8):
"""
Pour chaque prompt :
1. Génère G = 8 réponses différentes (température élevée)
2. Évalue chaque réponse (récompense connue : math, code)
3. Calcule avantage intra-groupe
4. PPO clip (sans critic)
"""
all_responses = []
for _ in range(group_size):
response = self.model.generate(prompts, temperature=0.7, top_p=0.95)
all_responses.append(response)
rewards = [evaluate(response) for response in all_responses]
advantages = self.compute_group_advantage(rewards)
total_loss =
response, adv (all_responses, advantages):
log_probs = .model.get_log_probs(prompts, response)
.ref_model:
torch.no_grad():
ref_log_probs = .ref_model.get_log_probs(prompts, response)
kl = F.kl_div(log_probs, ref_log_probs, reduction=)
:
kl =
ratio = torch.exp(log_probs - log_probs.detach())
clipped = torch.clamp(ratio, - .clip_epsilon, + .clip_epsilon)
loss = -torch.(ratio * adv, clipped * adv).mean()
loss += .beta * kl
total_loss += loss
total_loss / group_size
11. Tableau Comparatif
| Méthode | RM nécessaire | Modèle réf. | Données | Complexité | Stabilité | Année |
|---|
| RLHF (PPO) | ✓ | ✓ | Paires | ★★★★★ | ★★☆☆☆ | 2020 |
| DPO | ✗ | ✓ | Paires | ★★★☆☆ | ★★★★☆ | 2023 |
| KTO | ✗ | ✓ | Individ. | ★★★☆☆ | ★★★★☆ | 2024 |
| IPO | ✗ | ✓ | Paires | ★★★☆☆ | ★★★★★ | 2023 |
| ORPO | ✗ | ✗ | Paires | ★★☆☆☆ | ★★★☆☆ | 2024 |
| SimPO | ✗ | ✗ | Paires | ★★☆☆☆ | ★★★★☆ | 2024 |
| GRPO | ✗ | ✓ | Groupes | ★★★★☆ | ★★★☆☆ | 2025 |
12. Métriques d'Alignment
Références