Iterative Alignment › Implementation
Language AI HandbookAll chapters Iterative Alignment › ImplementationOne section · Sections in this chapter7 / 12 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
Let's implement an iterative DPO training loop that demonstrates these concepts. We'll simulate preference data generation and show how to structure multiple training rounds.
First, we'll create a simple language model for demonstration purposes. In practice, you'd use a full transformer, but this captures the key dynamics.
import torch
import torch.nn as nn
import torch.nn.functional as F
class SimpleLM(nn.Module):
"""A simple language model for showing iterative DPO."""
def __init__(self, vocab_size=1000, hidden_dim=256, num_layers=2):
super().__init__()
self.embedding = nn.Embedding(vocab_size, hidden_dim)
self.layers = nn.ModuleList(
[nn.Linear(hidden_dim, hidden_dim) for _ in range(num_layers)]
)
self.output = nn.Linear(hidden_dim, vocab_size)
self.vocab_size = vocab_size
def forward(self, x):
# x shape: (batch, seq_len)
h = self.embedding(x) # (batch, seq_len, hidden)
for layer in self.layers:
h = F.relu(layer(h))
logits = self.output(h) # (batch, seq_len, vocab)
return logits
def log_prob(self, sequences):
"""Compute log probability of sequences."""
logits = self.forward(sequences[:, :-1])
targets = sequences[:, 1:]
log_probs = F.log_softmax(logits, dim=-1)
token_log_probs = torch.gather(
log_probs, -1, targets.unsqueeze(-1)
).squeeze(-1)
return token_log_probs.sum(dim=-1)
def generate(self, prompt, max_length=20, temperature=1.0):
"""Generate a sequence given a prompt."""
self.eval()
with torch.no_grad():
current = prompt.clone()
for _ in range(max_length):
logits = self.forward(current)
next_logits = logits[:, -1, :] / temperature
probs = F.softmax(next_logits, dim=-1)
next_token = torch.multinomial(probs, 1)
current = torch.cat([current, next_token], dim=1)
return currentNow let's define the DPO training components.
def compute_dpo_loss(policy, reference, chosen, rejected, beta=0.1):
"""
Compute the DPO loss for a batch of preference pairs.
Args:
policy: Current policy model
reference: Reference model (frozen)
chosen: Preferred sequences (batch, seq_len)
rejected: Dispreferred sequences (batch, seq_len)
beta: Temperature parameter
Returns:
Scalar loss value
"""
# Compute log probabilities under policy
policy_chosen_logp = policy.log_prob(chosen)
policy_rejected_logp = policy.log_prob(rejected)
# Compute log probabilities under reference (no gradients)
with torch.no_grad():
ref_chosen_logp = reference.log_prob(chosen)
ref_rejected_logp = reference.log_prob(rejected)
# Compute log ratios
chosen_ratio = policy_chosen_logp - ref_chosen_logp
rejected_ratio = policy_rejected_logp - ref_rejected_logp
# DPO loss: -log(sigmoid(beta * (chosen_ratio - rejected_ratio)))
logits = beta * (chosen_ratio - rejected_ratio)
loss = -F.logsigmoid(logits).mean()
return lossWe need a simulated preference oracle. In practice, this would be human annotators or a reward model. Our oracle prefers sequences with certain patterns.
class PreferenceOracle:
"""
Simulated preference oracle that prefers certain patterns.
In real applications, this would be human feedback or a reward model.
"""
def __init__(self, preferred_tokens, penalty_tokens, seed=42):
self.preferred = set(preferred_tokens)
self.penalty = set(penalty_tokens)
self.rng = np.random.RandomState(seed)
def score(self, sequence):
"""Score a sequence based on preferred/penalty tokens."""
seq_list = sequence.cpu().numpy().tolist()
score = 0
for token in seq_list:
if token in self.preferred:
score += 1
elif token in self.penalty:
score -= 1
return score
def compare(self, seq1, seq2):
"""
Compare two sequences, return (chosen, rejected).
Adds some noise to simulate annotator disagreement.
"""
score1 = self.score(seq1) + self.rng.normal(0, 0.5)
score2 = self.score(seq2) + self.rng.normal(0, 0.5)
if score1 >= score2:
return seq1, seq2
else:
return seq2, seq1Now let's implement the iterative DPO training loop with both fixed and rolling reference options.
import copy
from collections import defaultdict
import numpy as np
class IterativeDPOTrainer:
"""Trainer for iterative DPO with configurable reference updating."""
def __init__(
self, model, oracle, beta=0.1, lr=1e-4, rolling_reference=True
):
self.model = model
self.oracle = oracle
self.beta = beta
self.lr = lr
self.rolling_reference = rolling_reference
# Initialize reference as copy of initial model
self.reference = copy.deepcopy(model)
self.reference.eval()
for param in self.reference.parameters():
param.requires_grad = False
# Keep original reference for KL monitoring
self.original_reference = copy.deepcopy(model)
self.original_reference.eval()
for param in self.original_reference.parameters():
param.requires_grad = False
self.optimizer = torch.optim.Adam(model.parameters(), lr=lr)
self.metrics = defaultdict(list)
def generate_preference_data(self, prompts, num_pairs=2, temperature=1.0):
"""
Generate preference pairs by sampling from current policy.
"""
self.model.eval()
chosen_list, rejected_list = [], []
for prompt in prompts:
# Generate multiple responses
responses = []
for _ in range(num_pairs * 2):
response = self.model.generate(
prompt.unsqueeze(0), max_length=15, temperature=temperature
)
responses.append(response.squeeze(0))
# Create preference pairs
for i in range(0, len(responses) - 1, 2):
chosen, rejected = self.oracle.compare(
responses[i], responses[i + 1]
)
chosen_list.append(chosen)
rejected_list.append(rejected)
self.model.train()
return chosen_list, rejected_list
def train_iteration(self, prompts, num_pairs=2, num_steps=50):
"""Run one iteration of DPO training."""
# Generate preference data from current policy
chosen_list, rejected_list = self.generate_preference_data(
prompts, num_pairs
)
# Training steps on this data
losses = []
for step in range(num_steps):
# Sample a batch
batch_size = min(16, len(chosen_list))
indices = np.random.choice(
len(chosen_list), batch_size, replace=False
)
# Pad sequences to same length
max_len = max(
max(chosen_list[i].size(0) for i in indices),
max(rejected_list[i].size(0) for i in indices),
)
chosen_batch = torch.zeros(batch_size, max_len, dtype=torch.long)
rejected_batch = torch.zeros(batch_size, max_len, dtype=torch.long)
for j, idx in enumerate(indices):
c, r = chosen_list[idx], rejected_list[idx]
chosen_batch[j, : c.size(0)] = c
rejected_batch[j, : r.size(0)] = r
# Compute loss and update
self.optimizer.zero_grad()
loss = compute_dpo_loss(
self.model,
self.reference,
chosen_batch,
rejected_batch,
self.beta,
)
loss.backward()
self.optimizer.step()
losses.append(loss.item())
return np.mean(losses)
def compute_kl_divergence(self, prompts):
"""Compute mean next-token KL divergence from the original model."""
self.model.eval()
divergences = []
with torch.no_grad():
for prompt in prompts:
policy_logits = self.model(prompt.unsqueeze(0))[:, -1, :]
reference_logits = self.original_reference(prompt.unsqueeze(0))[
:, -1, :
]
policy_logp = F.log_softmax(policy_logits, dim=-1)
reference_logp = F.log_softmax(reference_logits, dim=-1)
policy_p = policy_logp.exp()
kl = (policy_p * (policy_logp - reference_logp)).sum(dim=-1)
divergences.append(kl.item())
self.model.train()
return float(np.mean(divergences)) if divergences else 0.0
def update_reference(self):
"""Update reference model to current policy (for rolling reference)."""
if self.rolling_reference:
self.reference.load_state_dict(self.model.state_dict())
self.reference.eval()
def run_iterations(self, prompts, num_iterations=5, num_steps_per_iter=50):
"""Run multiple iterations of DPO training."""
for iteration in range(num_iterations):
# Train for one iteration
avg_loss = self.train_iteration(
prompts, num_pairs=2, num_steps=num_steps_per_iter
)
# Compute metrics
kl_div = self.compute_kl_divergence(prompts)
self.metrics["iteration"].append(iteration)
self.metrics["loss"].append(avg_loss)
self.metrics["kl_divergence"].append(kl_div)
# Update reference for next iteration
self.update_reference()
print(
f"Iteration {iteration + 1}: Loss={avg_loss:.4f}, KL={kl_div:.4f}"
)
return self.metricsLet's run the iterative training and visualize the results.
# Set up the experiment
torch.manual_seed(42)
# Create model and oracle
vocab_size = 100
model = SimpleLM(vocab_size=vocab_size, hidden_dim=64, num_layers=2)
# Oracle prefers tokens 10-20, penalizes tokens 80-90
oracle = PreferenceOracle(
preferred_tokens=list(range(10, 20)), penalty_tokens=list(range(80, 90))
)
# Create sample prompts (random token sequences)
num_prompts = 20
prompts = [torch.randint(0, vocab_size, (5,)) for _ in range(num_prompts)]
# Run iterative DPO with rolling reference
trainer_rolling = IterativeDPOTrainer(
model=SimpleLM(vocab_size=vocab_size, hidden_dim=64, num_layers=2),
oracle=oracle,
beta=0.1,
rolling_reference=True,
)
metrics_rolling = trainer_rolling.run_iterations(
prompts, num_iterations=5, num_steps_per_iter=30
)# Run with fixed reference for comparison
trainer_fixed = IterativeDPOTrainer(
model=SimpleLM(vocab_size=vocab_size, hidden_dim=64, num_layers=2),
oracle=oracle,
beta=0.1,
rolling_reference=False,
)
metrics_fixed = trainer_fixed.run_iterations(
prompts, num_iterations=5, num_steps_per_iter=30
)import matplotlib.pyplot as plt
use_book_style(fixed_canvas=True)
# Loss comparison
fig, ax = plt.subplots(figsize=(3.0, 4.0))
fig.subplots_adjust(left=0.28, right=0.97, bottom=0.38, top=0.88)
ax.plot(
metrics_rolling["iteration"],
metrics_rolling["loss"],
color=PALETTE["primary"],
marker="o",
label="Rolling Reference",
linewidth=1.1,
markersize=4,
)
ax.plot(
metrics_fixed["iteration"],
metrics_fixed["loss"],
color=PALETTE["coral"],
marker="s",
label="Fixed Reference",
linewidth=1.1,
markersize=4,
)
ax.set_xlabel("Iteration")
ax.set_ylabel("DPO Loss")
ax.set_title("Training Loss Across Iterations")
polish_axes(ax)
ax.legend(
loc="upper center",
bbox_to_anchor=(0.5, -0.24),
ncol=1,
handlelength=1.5,
borderaxespad=0,
)
plt.show()
# KL divergence comparison
fig, ax = plt.subplots(figsize=(3.0, 4.0))
fig.subplots_adjust(left=0.28, right=0.97, bottom=0.38, top=0.88)
ax.plot(
metrics_rolling["iteration"],
metrics_rolling["kl_divergence"],
color=PALETTE["primary"],
marker="o",
label="Rolling Reference",
linewidth=1.1,
markersize=4,
)
ax.plot(
metrics_fixed["iteration"],
metrics_fixed["kl_divergence"],
color=PALETTE["coral"],
marker="s",
label="Fixed Reference",
linewidth=1.1,
markersize=4,
)
ax.set_xlabel("Iteration")
ax.set_ylabel("Mean Next-Token KL")
ax.set_title("Policy Drift Over Iterations")
ax.set_ylim(bottom=0)
polish_axes(ax)
ax.legend(
loc="upper center",
bbox_to_anchor=(0.5, -0.24),
ncol=1,
handlelength=1.5,
borderaxespad=0,
)
plt.show()

The visualization highlights two practical points about this toy run. The DPO losses stay in a narrow band rather than separating monotonically, so loss alone does not reveal how far either policy has moved. The mean next-token KL remains nonnegative and exposes the different drift trajectories created by rolling and fixed anchors. In a real experiment, these curves should be interpreted across repeated runs rather than treated as guaranteed monotonic behavior.
Now let's implement a self-improvement loop using the SPIN concept.
class SPINTrainer:
"""
Self-Play Fine-Tuning trainer.
Uses ground truth responses as positive examples and
model generations as negative examples.
"""
def __init__(self, model, beta=0.1, lr=1e-4):
self.model = model
self.beta = beta
self.optimizer = torch.optim.Adam(model.parameters(), lr=lr)
# Reference is always the previous iteration's model
self.reference = copy.deepcopy(model)
self.reference.eval()
for param in self.reference.parameters():
param.requires_grad = False
self.metrics = defaultdict(list)
def train_iteration(self, prompts, ground_truth_responses, num_steps=50):
"""
One SPIN iteration: model outputs are rejected,
ground truth is chosen.
"""
self.model.train()
losses = []
for step in range(num_steps):
# Sample batch
batch_size = min(8, len(prompts))
indices = np.random.choice(len(prompts), batch_size, replace=False)
# Ground truth responses are "chosen"
chosen_batch = ground_truth_responses[indices]
# Generate responses from current model as "rejected"
self.model.eval()
rejected_list = []
for idx in indices:
prompt = prompts[idx : idx + 1]
response = self.model.generate(prompt, max_length=15)
rejected_list.append(response.squeeze(0))
self.model.train()
# Pad rejected sequences
max_len = max(r.size(0) for r in rejected_list)
max_len = max(max_len, chosen_batch.size(1))
rejected_batch = torch.zeros(batch_size, max_len, dtype=torch.long)
for j, r in enumerate(rejected_list):
rejected_batch[j, : r.size(0)] = r
# Pad chosen if needed
if chosen_batch.size(1) < max_len:
chosen_padded = torch.zeros(
batch_size, max_len, dtype=torch.long
)
chosen_padded[:, : chosen_batch.size(1)] = chosen_batch
chosen_batch = chosen_padded
# Compute SPIN loss (same as DPO but with different data)
self.optimizer.zero_grad()
loss = compute_dpo_loss(
self.model,
self.reference,
chosen_batch,
rejected_batch,
self.beta,
)
loss.backward()
self.optimizer.step()
losses.append(loss.item())
return np.mean(losses)
def update_reference(self):
"""Update reference to current model for next iteration."""
self.reference.load_state_dict(self.model.state_dict())
def run_iterations(
self, prompts, ground_truth, num_iterations=5, num_steps=50
):
"""Run multiple SPIN iterations."""
for iteration in range(num_iterations):
avg_loss = self.train_iteration(prompts, ground_truth, num_steps)
self.metrics["iteration"].append(iteration)
self.metrics["loss"].append(avg_loss)
# Update reference for next iteration
self.update_reference()
print(f"SPIN Iteration {iteration + 1}: Loss={avg_loss:.4f}")
return self.metrics# Create synthetic "ground truth" responses
# In practice, these would be human-written high-quality responses
torch.manual_seed(123)
prompts_tensor = torch.stack([p for p in prompts]) # (num_prompts, prompt_len)
# Ground truth: sequences with many preferred tokens
ground_truth = torch.zeros(len(prompts), 20, dtype=torch.long)
for i in range(len(prompts)):
ground_truth[i, :5] = prompts[i] # Copy prompt
# Fill rest with preferred tokens (10-20)
ground_truth[i, 5:] = torch.randint(10, 20, (15,))
# Run SPIN training
spin_model = SimpleLM(vocab_size=vocab_size, hidden_dim=64, num_layers=2)
spin_trainer = SPINTrainer(spin_model, beta=0.1)
spin_metrics = spin_trainer.run_iterations(
prompts_tensor, ground_truth, num_iterations=5, num_steps=30
)Let's evaluate how the model's outputs change over SPIN iterations.
def evaluate_generation_quality(model, prompts, oracle, num_samples=5):
"""Evaluate average oracle score of model generations."""
model.eval()
total_score = 0
count = 0
with torch.no_grad():
for prompt in prompts[:10]:
for _ in range(num_samples):
response = model.generate(prompt.unsqueeze(0), max_length=15)
score = oracle.score(response.squeeze(0))
total_score += score
count += 1
return total_score / count# Compare initial vs trained model
initial_model = SimpleLM(vocab_size=vocab_size, hidden_dim=64, num_layers=2)
initial_score = evaluate_generation_quality(initial_model, prompts, oracle)
spin_score = evaluate_generation_quality(spin_trainer.model, prompts, oracle)Generation Quality Evaluation: ---------------------------------------- Initial model average score: 0.60 SPIN-trained model average score: 1.24
The SPIN-trained model achieves a significantly higher score, confirming that self-play fine-tuning effectively improved alignment with the oracle's preferences.
import matplotlib.pyplot as plt
use_book_style()
fig, ax = plt.subplots(figsize=(6.0, 4.0))
ax.plot(
spin_metrics["iteration"],
spin_metrics["loss"],
color=PALETTE["teal"],
marker="o",
linewidth=1.1,
markersize=5,
)
ax.set_xlabel("Iteration")
ax.set_ylabel("SPIN Loss")
ax.set_title("Self-Play Fine-Tuning Loss")
polish_axes(ax)
plt.show()
The rapid decrease in loss indicates the model is successfully optimizing the SPIN objective, learning to distinguish ground truth responses from its own generations. Finally, let's implement monitoring for alignment stability.
class AlignmentMonitor:
"""Monitor alignment stability across iterations."""
def __init__(self, original_model, oracle):
self.original_model = copy.deepcopy(original_model)
self.original_model.eval()
self.oracle = oracle
self.history = defaultdict(list)
def compute_diversity(self, model, prompts, num_samples=10):
"""Measure output diversity using unique token ratio."""
model.eval()
all_tokens = []
with torch.no_grad():
for prompt in prompts[:5]:
for _ in range(num_samples):
response = model.generate(
prompt.unsqueeze(0), max_length=15
)
tokens = response.squeeze(0)[5:].tolist() # Exclude prompt
all_tokens.extend(tokens)
unique_ratio = (
len(set(all_tokens)) / len(all_tokens) if all_tokens else 0
)
return unique_ratio
def compute_length_stats(self, model, prompts, num_samples=10):
"""Measure response length statistics."""
model.eval()
lengths = []
with torch.no_grad():
for prompt in prompts[:5]:
for _ in range(num_samples):
response = model.generate(
prompt.unsqueeze(0), max_length=20
)
lengths.append(response.size(1))
return np.mean(lengths), np.std(lengths)
def snapshot(self, model, prompts, iteration):
"""Record metrics for current iteration."""
diversity = self.compute_diversity(model, prompts)
mean_len, std_len = self.compute_length_stats(model, prompts)
oracle_score = evaluate_generation_quality(
model, prompts, self.oracle, 5
)
self.history["iteration"].append(iteration)
self.history["diversity"].append(diversity)
self.history["mean_length"].append(mean_len)
self.history["std_length"].append(std_len)
self.history["oracle_score"].append(oracle_score)
return {
"diversity": diversity,
"mean_length": mean_len,
"oracle_score": oracle_score,
}# Run iterative training with monitoring
monitored_model = SimpleLM(vocab_size=vocab_size, hidden_dim=64, num_layers=2)
monitor = AlignmentMonitor(monitored_model, oracle)
trainer_monitored = IterativeDPOTrainer(
model=monitored_model, oracle=oracle, beta=0.1, rolling_reference=True
)
stability_metrics = []
for iteration in range(6):
# Record metrics before training
metrics = monitor.snapshot(trainer_monitored.model, prompts, iteration)
stability_metrics.append(metrics)
# Train one iteration (except on last loop)
if iteration < 5:
trainer_monitored.train_iteration(prompts, num_pairs=2, num_steps=30)
trainer_monitored.update_reference()Alignment Stability Metrics: Iter 0: Diversity=0.133, Score=0.52 Iter 1: Diversity=0.133, Score=0.48 Iter 2: Diversity=0.133, Score=0.92 Iter 3: Diversity=0.133, Score=0.58 Iter 4: Diversity=0.133, Score=0.88 Iter 5: Diversity=0.133, Score=1.14
In this deterministic toy run, the corpus-level unique-token ratio stays nearly flat while the oracle score fluctuates and finishes above its starting value. That is itself useful monitoring evidence: a short run need not exhibit a clean monotonic diversity-quality trade-off, so both metrics should be tracked rather than inferred from the training objective.
import matplotlib.pyplot as plt
use_book_style(fixed_canvas=True)
iterations = list(range(len(stability_metrics)))
diversities = [m["diversity"] for m in stability_metrics]
scores = [m["oracle_score"] for m in stability_metrics]
# Left plot: Diversity
fig, ax = plt.subplots(figsize=(3.0, 3.0))
fig.subplots_adjust(left=0.22, right=0.97, bottom=0.18, top=0.87)
ax.plot(
iterations,
diversities,
color=PALETTE["purple"],
marker="o",
linewidth=1.1,
markersize=4,
)
ax.set_xlabel("Iteration")
ax.set_ylabel("Unique Token Ratio")
ax.set_title("Output Diversity Over Iterations")
ax.set_ylim(0, 1)
polish_axes(ax)
plt.show()
# Right plot: Score
fig, ax = plt.subplots(figsize=(3.0, 3.0))
fig.subplots_adjust(left=0.22, right=0.97, bottom=0.18, top=0.87)
ax.plot(
iterations,
scores,
color=PALETTE["teal"],
marker="s",
linewidth=1.1,
markersize=4,
)
ax.set_xlabel("Iteration")
ax.set_ylabel("Oracle Preference Score")
ax.set_title("Alignment Quality Over Iterations")
polish_axes(ax)
plt.show()

The stability metrics reveal the trade-offs inherent in iterative alignment. As the model learns to satisfy the oracle's preferences, it may sacrifice diversity, a phenomenon that warrants careful monitoring in production systems.