Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Hybrid-policy optimization combining multiple importance sampling and exploration-based advantage functions. Prevents capability boundary collapse while maintaining verification rewards, enabling LLMs to exceed baseline boundaries.
RL-PLUS: Countering Capability Boundary Collapse in RLVR
Core Concept
RL-PLUS addresses a fundamental limitation in Reinforcement Learning with Verifiable Reward (RLVR): capability boundary collapse, where models narrow their problem-solving abilities rather than expanding them. The approach combines importance sampling to handle external data distributions and exploration-based advantage functions that guide models toward high-value, unexplored reasoning paths, enabling breakthrough performance on difficult tasks.
Architecture Overview
Multiple Importance Sampling: Addresses distributional mismatch between on-policy RL data and external training sources
Exploration-Based Advantage Function: Quantifies value of unexplored reasoning paths
Hybrid Policy Optimization: Balances on-policy verification rewards with off-policy external data
Capability Frontier Detection: Identifies problems near model boundaries where exploration is most valuable
Adaptive Data Weighting: Prioritizes external data that pushes capability boundaries
Implementation Steps
Step 1: Design Importance Sampling for Distribution Mismatch
Implement multiple importance sampling (MIS) to weight external data appropriately.
import numpy as np
from scipy.special import softmax
defcompute_importance_weights(model, external_data, on_policy_data, batch_size=32):
"""
Compute importance weights for external data using MIS.
Args:
model: Current model to evaluate
external_data: Off-policy data from external source
on_policy_data: On-policy data from model rollouts
batch_size: Batch size for computation
Returns:
Importance weights for each external sample
"""
weights = []
for batch in create_batches(external_data, batch_size):
# Compute likelihood under current model
model_logprobs = model.get_logprobs(batch)
# Estimate data distribution (typically uniform or from original source)
data_logprobs = estimate_data_distribution(batch)
# Importance weight = p_model / p_data
log_weights = model_logprobs - data_logprobs
log_weights = np.clip(log_weights, -, )
weights.extend(np.exp(log_weights))
np.array(weights)
():
all_weights = []
estimator_idx (num_samples):
weights = compute_importance_weights(
model,
external_data,
on_policy_data
)
max_weight = np.percentile(weights, )
truncated_weights = np.minimum(weights, max_weight)
all_weights.append(truncated_weights)
final_weights = np.mean(all_weights, axis=)
final_weights / final_weights.()
# Clip to prevent extreme values (Huber-like truncation)
"""
Use multiple importance sampling estimators for robust weighting.
Args:
model: Current model
external_data: External training data
on_policy_data: Model-generated data
num_samples: Number of importance estimators
Returns:
Robust importance weights (averaged across estimators)
"""
for
in
range
# Different importance estimators
# Optionally apply different truncation for each estimator
95
# Average weights across estimators for robustness
0
return
sum
# Normalize
Step 2: Implement Exploration-Based Advantage Function
Create advantage function that rewards exploring high-value, low-confidence regions.
classExplorationBasedAdvantageFunction:
"""
Quantifies value of exploring uncertain, high-potential regions.
"""def__init__(self, model, value_model, exploration_bonus_weight=0.5):
self.model = model
self.value_model = value_model
self.exploration_bonus_weight = exploration_bonus_weight
defcompute_advantage(self, trajectories, returns):
"""
Compute advantage with exploration bonus.
Args:
trajectories: List of (state, action, logprob) sequences
returns: Actual returns from trajectories
Returns:
Advantage estimates [num_trajectories]
"""
advantages = []
for trajectory, trajectory_return inzip(trajectories, returns):
# Baseline value estimate
baseline = self.value_model.estimate_trajectory_value(trajectory)
# Standard advantage
standard_advantage = trajectory_return - baseline
# Exploration bonus: higher when trajectory ventures into novel territory
exploration_bonus = self._compute_exploration_bonus(trajectory)
# Combined advantage
total_advantage = (
standard_advantage +
self.exploration_bonus_weight * exploration_bonus
)
advantages.append(total_advantage)
return np.array(advantages)
def_compute_exploration_bonus(self, trajectory):
"""
Compute bonus for exploring uncertain regions.
Args:
trajectory: Sequence of states and actions
Returns:
Exploration bonus value
"""# Measure uncertainty in value estimates
bonus = 0.0for state in trajectory:
# Get value estimate from value model
value_estimate = self.value_model.predict(state)
# Measure uncertainty (use bootstrap ensemble or variance)
uncertainty = self.value_model.estimate_uncertainty(state)
# Higher uncertainty = higher exploration value
bonus += uncertainty
# Normalize by trajectory length
bonus /= len(trajectory)
return bonus
defdetect_capability_frontier(self, problem_pool):
"""
Identify problems at the edge of model capability.
Args:
problem_pool: Set of candidate problems
Returns:
Ranked list of frontier problems
"""
frontier_scores = []
for problem in problem_pool:
# Get model's confidence on this problem
model_output = self.model.generate(problem["prompt"])
confidence = self.model.compute_confidence(model_output)
# Check if problem is solvable but difficult
difficulty = estimate_problem_difficulty(problem)
# Problems with high difficulty but not impossible are frontier
frontier_score = difficulty * (1.0 - confidence)
frontier_scores.append({
"problem": problem,
"frontier_score": frontier_score,
"difficulty": difficulty,
"model_confidence": confidence
})
# Sort by frontier score
frontier_scores.sort(key=lambda x: x["frontier_score"], reverse=True)
return frontier_scores
Step 3: Implement Hybrid Policy Optimization
Combine on-policy verification rewards with off-policy external data.
defhybrid_policy_optimization_step(
model,
on_policy_batch,
external_batch,
importance_weights,
advantage_function,
learning_rate=1e-4,
on_policy_weight=0.7,
external_weight=0.3):
"""
Perform optimization step using hybrid on/off-policy objectives.
Args:
model: Policy model to optimize
on_policy_batch: Batch from model rollouts with verifiable rewards
external_batch: Batch from external training data
importance_weights: Importance weights for external data
advantage_function: Advantage estimator
learning_rate: Optimizer learning rate
on_policy_weight: Weight for on-policy objective
external_weight: Weight for external data objective
Returns:
Updated model and loss metrics
"""# Compute on-policy loss (REINFORCE with baselines)
on_policy_loss = compute_on_policy_loss(
model,
on_policy_batch,
advantage_function
)
# Compute off-policy loss (importance-weighted)
external_loss = compute_importance_weighted_loss(
model,
external_batch,
importance_weights,
target_logprobs=compute_target_logprobs(external_batch)
)
# Hybrid objective
total_loss = (
on_policy_weight * on_policy_loss +
external_weight * external_loss
)
# Gradient step
model.optimizer.zero_grad()
total_loss.backward()
# Gradient clipping to prevent instability
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
model.optimizer.step()
return {
"total_loss": total_loss.item(),
"on_policy_loss": on_policy_loss.item(),
"external_loss": external_loss.item()
}
defcompute_importance_weighted_loss(model, batch, weights, target_logprobs):
"""
Compute importance-weighted loss for off-policy data.
Args:
model: Policy model
batch: Data batch
weights: Importance weights
target_logprobs: Target log probabilities from source model
Returns:
Weighted loss value
"""
model_logprobs = model.get_logprobs(batch)
# Importance-weighted KL divergence
kl_div = model_logprobs - target_logprobs
weighted_loss = (weights * kl_div).mean()
return weighted_loss
Step 4: Training Loop with Capability Boundary Detection
Integrate all components into training loop.
deftrain_with_rl_plus(
model,
verification_function,
problem_pool,
external_training_data,
config
):
"""
Train model with RL-PLUS to prevent capability collapse.
Args:
model: LLM to train
verification_function: Function to verify correct solutions
problem_pool: Set of candidate problems
external_training_data: External training data
config: Training configuration
Returns:
Trained model with expanded capabilities
"""
value_model = train_value_model(problem_pool)
advantage_function = ExplorationBasedAdvantageFunction(
model,
value_model,
exploration_bonus_weight=config.exploration_weight
)
for iteration inrange(config.num_iterations):
# 1. Identify capability frontier
frontier = advantage_function.detect_capability_frontier(problem_pool)
# 2. Rollout on frontier problems (on-policy)
on_policy_rollouts = []
for frontier_item in frontier[:config.frontier_batch_size]:
problem = frontier_item["problem"]
# Generate solution
solution = model.generate(problem["prompt"])
# Verify correctness
is_correct = verification_function(solution, problem["gold"])
# Compute return
return_value = 1.0if is_correct else0.0
on_policy_rollouts.append({
"problem": problem,
"solution": solution,
"return": return_value,
"is_correct": is_correct
})
# 3. Sample from external data with importance weighting
external_sample = sample_external_batch(
external_training_data,
config.external_batch_size
)
# 4. Compute importance weights
importance_weights = multi_importance_sample(
model,
external_sample,
on_policy_rollouts,
num_samples=3
)
# 5. Compute advantages with exploration bonus
on_policy_advantages = advantage_function.compute_advantage(
on_policy_rollouts,
returns=[r["return"] for r in on_policy_rollouts]
)
# 6. Optimization step
loss_metrics = hybrid_policy_optimization_step(
model,
on_policy_rollouts,
external_sample,
importance_weights,
advantage_function,
learning_rate=config.learning_rate
)
if iteration % config.log_freq == 0:
print(f"Iteration {iteration}: Loss={loss_metrics['total_loss']:.4f}")
return model
Practical Guidance
When to Use RL-PLUS
Capability expansion tasks: Models must solve problems beyond baseline ability
Sparse reward environments: Where on-policy exploration is inefficient
Hybrid data availability: Mix of verified on-policy and external off-policy data
Reasoning tasks: Math, coding where exploration of reasoning paths matters
When NOT to Use RL-PLUS
Simple task adaptation: Where verification rewards suffice
Abundant on-policy data: No need for external data exploitation
External weight: 0.2-0.4 (supplement with external data)
Exploration bonus weight: 0.3-0.7 (depends on frontier difficulty)
Importance weight clipping: Log-space clipping at ±2.0
Frontier batch size: 4-8 problems per iteration
Key Insights
The key insight is that standard RLVR narrows capability by optimizing solely on verifiable rewards. By explicitly incentivizing exploration of high-uncertainty regions combined with importance-weighted external data, models can break through their baseline boundaries. The hybrid approach prevents collapse by diversifying the optimization signal.
