Improves LLM reasoning by rewarding correct solutions that exhibit rare high-level strategies, preventing exploration collapse and discovering more diverse solution approaches across mathematics, physics, and medical reasoning.
Improves LLM reasoning by rewarding correct solutions that exhibit rare high-level strategies, preventing exploration collapse and discovering more diverse solution approaches across mathematics, physics, and medical reasoning.
Overview
Enhance exploration in LLM reasoning by rewarding correct solutions based on their rarity of approach. Rather than treating all correct solutions equally, provide higher rewards for solutions using novel high-level strategies, preventing models from converging to single solution patterns.
When to Use
For reasoning tasks where multiple solution approaches exist
When you want to maximize the diversity of reasoning strategies learned
For improving pass@k metrics by exploring different solution paths
For creative problem-solving where approach diversity is valuable
When NOT to Use
For tasks with unique optimal solutions
When solution diversity doesn't matter for the application
For single-solution problems
When computational overhead of strategy analysis is unacceptable
Key Technical Components
Solution Strategy Clustering
Group solutions by high-level reasoning approach, ignoring surface-level variation.
# Strategy clustering for exploration diversityclassSolutionStrategyClusterer:
def__init__(self, strategy_classifier_model):
self.classifier = strategy_classifier_model
defextract_strategy(self, solution_text, problem):
"""Identify high-level strategy from solution"""
prompt = f"""
Analyze this {problem['domain']} solution:
{solution_text}
What high-level strategy does it use? (e.g., algebraic, graphical, numerical, recursive)
Ignore surface-level variations in expression.
"""
strategy = self.classifier.classify(prompt)
return strategy
defcluster_solutions(self, solutions, problem):
"""Group solutions by strategy"""
strategy_groups = {}
for solution solutions:
strategy = .extract_strategy(solution[], problem)
strategy strategy_groups:
strategy_groups[strategy] = []
strategy_groups[strategy].append(solution)
strategy_groups
():
strategy_counts = {}
s all_strategies:
strategy_counts[s] = strategy_counts.get(s, ) +
frequency = strategy_counts.get(strategy, ) / (all_strategies)
rarity = / (frequency + )
normalized_rarity = / (rarity + )
normalized_rarity
in
self
"text"
if
not
in
return
def
get_strategy_rarity
self, strategy, all_strategies
"""Compute rarity score for a strategy"""
for
in
0
1
# Rarity = inverse of frequency
1
len
1.0
1e-6
# Normalize to [0, 1]
1.0
1.0
# Inverse sigmoid
return
Uniqueness-Aware Reward Assignment
Assign higher rewards to rare but correct solutions.
# Uniqueness-aware reward computationclassUniquenessReward:
defcompute_reward(self, solution, is_correct, strategy, all_strategies, base_reward=1.0):
"""Compute reward based on correctness and strategy rarity"""ifnot is_correct:
return0.0# Get rarity score for this strategy
rarity_score = SolutionStrategyClusterer().get_strategy_rarity(
strategy,
all_strategies
)
# Reward formula: base * (1 + rarity_bonus)
rarity_bonus = 0.5 * rarity_score # Up to 50% bonus for rare strategies
final_reward = base_reward * (1.0 + rarity_bonus)
return final_reward
defbatch_compute_rewards(self, solutions, problem):
"""Compute rewards for all solutions in a rollout"""# Determine correctness for each
correctness = [self.evaluate_correctness(sol, problem) for sol in solutions]
# Extract strategies
clusterer = SolutionStrategyClusterer(self.get_classifier())
all_strategies = [clusterer.extract_strategy(sol["text"], problem) for sol in solutions]
# Compute rewards
rewards = []
for solution, is_correct, strategy inzip(solutions, correctness, all_strategies):
reward = self.compute_reward(
solution,
is_correct,
strategy,
all_strategies
)
rewards.append(reward)
return rewards
defevaluate_correctness(self, solution, problem):
"""Check if solution is correct"""# Domain-specific evaluationif problem["domain"] == "math":
returnself.verify_math_solution(solution, problem)
elif problem["domain"] == "physics":
returnself.verify_physics_solution(solution, problem)
elif problem["domain"] == "medical":
returnself.verify_medical_diagnosis(solution, problem)
returnFalsedefget_classifier(self):
"""Initialize strategy classifier model"""# Would load actual model in practicepass
Rollout Analysis and Re-weighting
Analyze rollout diversity and re-weight samples.
# Rollout analysis for diversityclassRolloutAnalyzer:
defanalyze_rollout(self, solutions, problem):
"""Assess diversity and correctness of rollout"""
clusterer = SolutionStrategyClusterer(self.get_classifier())
# Cluster solutions by strategy
strategy_groups = clusterer.cluster_solutions(solutions, problem)
# Compute diversity metrics
num_strategies = len(strategy_groups)
total_correct = sum(
1for sol in solutions
if clusterer.extract_strategy(sol, problem) and sol["is_correct"]
)
diversity_score = num_strategies / len(solutions)
correctness_score = total_correct / len(solutions)
return {
"diversity_score": diversity_score,
"correctness_score": correctness_score,
"num_strategies": num_strategies,
"strategy_distribution": {
strategy: len(sols) for strategy, sols in strategy_groups.items()
}
}
defshould_reweight_rollout(self, analysis):
"""Determine if rollout needs re-weighting"""# Re-weight if collapse detectedif analysis["diversity_score"] < 0.5:
returnTrue# Don't re-weight if good diversityreturnFalsedefreweight_samples(self, solutions, analysis):
"""Re-weight solutions to encourage diversity"""
reweighted = []
clusterer = SolutionStrategyClusterer(self.get_classifier())
for solution in solutions:
strategy = clusterer.extract_strategy(solution, None)
# Higher weight for underrepresented strategies
base_weight = 1.0 / analysis["strategy_distribution"].get(strategy, 1)
# Multiply by correctness score
weight = base_weight * (1.0if solution["is_correct"] else0.5)
reweighted.append({
"solution": solution,
"weight": weight
})
# Normalize weights
total_weight = sum(w["weight"] for w in reweighted)
for item in reweighted:
item["weight"] /= total_weight
return reweighted
Training Loop with Exploration Tracking
Integrate uniqueness rewards into RL training.
# RL training with uniqueness rewardsclassUniquenessRL:
def__init__(self, policy_model):
self.policy = policy_model
self.exploration_history = []
deftrain_step(self, problem, num_rollouts=10):
"""Training step with uniqueness-aware rewards"""
all_solutions = []
all_rewards = []
# Generate multiple solutionsfor _ inrange(num_rollouts):
solution = self.policy.generate(problem)
all_solutions.append(solution)
# Compute uniqueness-aware rewards
reward_computer = UniquenessReward()
rewards = reward_computer.batch_compute_rewards(all_solutions, problem)
all_rewards.extend(rewards)
# Analyze rollout diversity
analyzer = RolloutAnalyzer()
analysis = analyzer.analyze_rollout(all_solutions, problem)
self.exploration_history.append(analysis)
# If collapse detected, apply re-weightingif analyzer.should_reweight_rollout(analysis):
solutions = analyzer.reweight_samples(all_solutions, analysis)
rewards = [item["weight"] * reward for item, reward inzip(solutions, all_rewards)]
# Policy gradient update
total_loss = 0.0for solution, reward inzip(all_solutions, all_rewards):
log_prob = self.policy.get_log_prob(solution)
loss = -log_prob * reward
total_loss += loss
self.policy.backward(total_loss / len(all_solutions))
self.policy.optimize()
return {
"loss": (total_loss / len(all_solutions)).item(),
"diversity": analysis["diversity_score"],
"avg_reward": sum(all_rewards) / len(all_rewards)
}
defevaluate_exploration(self, test_problems, num_rollouts=10):
"""Measure exploration quality across test set"""
exploration_metrics = {}
for problem in test_problems:
solutions = [self.policy.generate(problem) for _ inrange(num_rollouts)]
analyzer = RolloutAnalyzer()
analysis = analyzer.analyze_rollout(solutions, problem)
exploration_metrics[problem["id"]] = analysis
# Aggregate metrics
avg_diversity = np.mean([a["diversity_score"] for a in exploration_metrics.values()])
avg_correctness = np.mean([a["correctness_score"] for a in exploration_metrics.values()])
return {
"avg_diversity": avg_diversity,
"avg_correctness": avg_correctness,
"per_problem": exploration_metrics
}
Performance Characteristics
Consistent improvements on pass@k metrics across domains
Maintains or improves pass@1 (single attempt) performance
Increases solution strategy diversity by 2-3x
Works across mathematics, physics, and medical reasoning
Integration Pattern
Generate multiple solutions for each problem (10-20 samples)
Evaluate correctness for each solution
Extract high-level strategy for each correct solution
Cluster solutions by strategy, compute rarity
Assign rewards based on correctness × rarity
Perform policy gradient update with uniqueness-aware rewards
Monitor exploration diversity; if collapse detected, re-weight