Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
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.
