Improves LLM reasoning by decomposing RL objectives into intermediate process rewards assigned to reasoning steps, improving both final accuracy and reasoning capacity without expensive Monte Carlo Tree Search.
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.
Improves LLM reasoning by decomposing RL objectives into intermediate process rewards assigned to reasoning steps, improving both final accuracy and reasoning capacity without expensive Monte Carlo Tree Search.
Overview
Enhance LLM reasoning capabilities by providing reward signals at intermediate steps of the reasoning process rather than only at the end. This approach decomposes entropy-regularized RL objectives into granular process-level rewards that guide step-by-step improvement.
When to Use
For complex reasoning tasks (math, logic, science, coding)
When you want to improve pass@k performance
For multi-step problem-solving where intermediate correctness matters
When computational efficiency is important (vs. MCTS or expensive sampling)
When NOT to Use
For single-step tasks without intermediate reasoning
When intermediate ground truth is unavailable
For real-time applications where reward computation adds latency
For tasks where final-outcome-only feedback is sufficient
Key Technical Components
Process Reward Decomposition
Break down the global RL objective into step-level rewards.
# Process reward decompositionclassProcessRewardLearner:
def__init__(self, reference_model):
self.reference_model = reference_model
self.process_rewards = {}
defdecompose_objective(self, trajectory, final_reward):
"""Decompose end-reward into process rewards"""# Original objective: max R(trajectory) - KL(policy || reference)# Decomposed: sum of process rewards for each step
trajectory_length = len(trajectory)
base_reward = final_reward / trajectory_length
process_rewards = []
for step_idx, step inenumerate(trajectory):
# Assign step-level reward
step_reward = self.compute_step_reward(
step,
step_idx,
trajectory,
final_reward
)
process_rewards.append(step_reward)
return process_rewards
():
step_logprob = step[]
reference_logprob = .reference_model.get_logprob(step[])
kl_penalty = step_logprob - reference_logprob
progress_bonus = .estimate_progress(idx, full_trajectory)
progress_bonus - * kl_penalty
def
compute_step_reward
self, step, idx, full_trajectory, final_reward
"""Compute reward for individual reasoning step"""
# Reward based on:
# 1. Step correctness (if available)
# 2. Progress toward solution
# 3. KL penalty vs reference model
"log_probability"
self
"text"
self
return
0.1
Step-Level Correctness Annotation
When available, use ground truth to label intermediate steps.
# Step correctness annotationdefannotate_step_correctness(trajectory, problem, solution_steps):
"""Mark which reasoning steps are correct"""
annotations = []
for i, step inenumerate(trajectory):
if i < len(solution_steps):
is_correct = matches_solution_step(step, solution_steps[i])
else:
is_correct = None# Unknown
annotations.append({
"step_index": i,
"text": step["text"],
"is_correct": is_correct,
"confidence": compute_confidence(step, solution_steps[i])
})
return annotations
Entropy Regularization Integration
Incorporate KL divergence penalty to maintain exploration.
# Entropy-regularized process rewardsclassEntropyRegularizedPRL:
def__init__(self, beta=0.1, reference_model=None):
self.beta = beta # Entropy regularization coefficientself.reference_model = reference_model
defcompute_regularized_reward(self, step_text, reference_logprob):
"""Add entropy regularization to step reward"""# KL(policy || reference) = E[log(p) - log(q)]
policy_logprob = self.policy_model(step_text)
kl_divergence = policy_logprob - reference_logprob
# Entropy regularization penalizes KL
regularized_reward = -self.beta * kl_divergence
return regularized_reward
defbatch_compute_process_rewards(self, trajectories):
"""Compute process rewards for batch of trajectories"""
all_rewards = []
for trajectory in trajectories:
rewards = []
for step in trajectory:
ref_logprob = self.reference_model(step)
reward = self.compute_regularized_reward(step, ref_logprob)
rewards.append(reward)
all_rewards.append(rewards)
return all_rewards
Advantage Estimation
Compute advantages for policy gradient updates.
# Advantage estimation from process rewardsdefcompute_advantages(process_rewards, discount_factor=0.99):
"""Convert process rewards to advantages for PG update"""
advantages = []
cumulative_return = 0# Reverse traversal for discount computationfor reward inreversed(process_rewards):
cumulative_return = reward + discount_factor * cumulative_return
advantages.insert(0, cumulative_return)
# Normalize for stability
advantages = (advantages - np.mean(advantages)) / (np.std(advantages) + 1e-8)
return advantages
Training Loop Integration
Incorporate process rewards into standard policy gradient training.
# PRL training loopclassPRLTrainer:
def__init__(self, policy_model, reference_model, beta=0.1):
self.policy = policy_model
self.reference = reference_model
self.beta = beta
deftrain_step(self, problem, trajectories, final_rewards):
"""Single training step using process rewards"""
losses = []
for trajectory, final_reward inzip(trajectories, final_rewards):
# 1. Decompose end-reward into process rewards
process_rewards = self.decompose_objective(trajectory, final_reward)
# 2. Add entropy regularization
regularized_rewards = self.add_entropy_penalty(
trajectory,
process_rewards
)
# 3. Compute advantages
advantages = compute_advantages(regularized_rewards)
# 4. Policy gradient lossfor step, advantage inzip(trajectory, advantages):
step_logprob = self.policy(step)
loss = -step_logprob * advantage # PG loss
losses.append(loss)
# Optimization step
total_loss = sum(losses) / len(losses)
self.policy.backward(total_loss)
self.policy.optimize()
return total_loss.item()
defdecompose_objective(self, trajectory, final_reward):
"""Decompose final reward to process level"""return ProcessRewardLearner().decompose_objective(trajectory, final_reward)
defadd_entropy_penalty(self, trajectory, process_rewards):
"""Add entropy regularization to rewards"""
regularized = []
for step, reward inzip(trajectory, process_rewards):
ref_logprob = self.reference(step)
kl_penalty = self.compute_kl(step, ref_logprob)
regularized.append(reward - self.beta * kl_penalty)
return regularized
defcompute_kl(self, step, ref_logprob):
"""Compute KL divergence for step"""
policy_logprob = self.policy(step)
return policy_logprob - ref_logprob
Performance Measurement
Track improvements in both pass@1 and pass@k metrics.
# Performance trackingclassPerformanceTracker:
def__init__(self):
self.pass_at_1 = []
self.pass_at_k = []
self.reasoning_breadth = [] # Coverage of solution approachesdefevaluate(self, model, test_set, k=5):
"""Measure reasoning improvements"""# Pass@1: single attempt success
pass_1 = sum(
1for problem in test_set
if model.solve(problem, attempts=1)
) / len(test_set)
# Pass@k: success within k attempts
pass_k = sum(
1for problem in test_set
ifany(model.solve(problem, attempts=1) for _ inrange(k))
) / len(test_set)
# Reasoning breadth: diversity of approaches explored
breadth = self.measure_approach_diversity(model, test_set)
self.pass_at_1.append(pass_1)
self.pass_at_k.append(pass_k)
self.reasoning_breadth.append(breadth)
return {"pass@1": pass_1, "pass@k": pass_k, "breadth": breadth}
defmeasure_approach_diversity(self, model, test_set):
"""Count distinct reasoning approaches discovered"""
approaches = set()
for problem in test_set:
for attempt inrange(5):
trajectory = model.generate_trajectory(problem)
approach = self.classify_approach(trajectory)
approaches.add(approach)
returnlen(approaches)
