Improve RL training efficiency by embedding explicit reflection and revision loops. Models generate initial responses, receive feedback, produce self-reflections describing improvements, revise their attempts, and distill successful corrections into the base policy. Achieves up to 81% improvement on complex tasks through structured behavioral change.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
Improve RL training efficiency by embedding explicit reflection and revision loops. Models generate initial responses, receive feedback, produce self-reflections describing improvements, revise their attempts, and distill successful corrections into the base policy. Achieves up to 81% improvement on complex tasks through structured behavioral change.
Experiential Reinforcement Learning
Problem Context
Standard RL trains models to implicitly discover how failures should translate into behavioral change through undirected trial-and-error. This is inefficient and wastes signal from corrective feedback. Experiential RL makes the learning process explicit: models generate reflections on failures, attempt revisions based on feedback, and internalize successful corrections. This transforms reactive behavior optimization into structured experience learning.
Core Concept
ERL operates in three phases: (1) initial attempt on a task, (2) reflection and revision (model generates explanation of improvements and attempts again), (3) internalization via distillation. A cross-episode reflection memory stores successful corrective patterns discovered during training, enabling reuse across tasks.
Architecture Overview
Initial attempt: Generate first response to task
Reflection generation: Produce reasoning about improvements given feedback
Revision step: Attempt task again using reflection insights
Success detection: Identify when revision succeeds
Distillation: Consolidate successful revisions into base policy
Reflection memory: Store and retrieve successful correction patterns
"""Retrieve similar successful reflections for in-context learning."""
if
not
in
self
return
self
if
is
None
# Simple: return most recent
return
# With embeddings: return most similar
# (Simplified; would use vector similarity)
return
Step 2: Generate reflections
classReflectionGenerator:
"""Generate reflections on failures and improvements."""def__init__(self, model):
self.model = model
defgenerate_reflection(
self,
task: str,
initial_response: str,
feedback: str,
reflection_memory: ReflectionMemory = None,
task_type: str = None) -> str:
"""
Generate reflection on failure and proposed improvements.
Args:
task: Task description
initial_response: Initial attempt
feedback: Environmental feedback on failure
reflection_memory: Optional memory of past corrections
task_type: Category of task for memory retrieval
Returns:
reflection: Reasoning about improvements
"""# Retrieve similar successful reflections for in-context examples
in_context_examples = ""if reflection_memory and task_type:
similar = reflection_memory.retrieve_similar_reflections(task_type, top_k=2)
if similar:
in_context_examples = "Similar successful corrections:\n"for ex in similar[:2]:
in_context_examples += (
f"Initial: {ex['initial'][:100]}...\n"f"Reflection: {ex['reflection']}\n"f"Revised: {ex['revised'][:100]}...\n\n"
)
# Construct reflection prompt
prompt = f"""
Task: {task}
Initial attempt: {initial_response}
Feedback on failure: {feedback}{in_context_examples}
Analyze the failure and generate a reflection on what went wrong and how to improve. Be specific about the changes needed.
Reflection:"""
reflection = self.model.generate(prompt, max_tokens=200, temperature=0.7)
return reflection.strip()
Step 3: Generate revision based on reflection
classRevisionGenerator:
"""Generate revised responses using reflection insights."""def__init__(self, model):
self.model = model
defgenerate_revision(
self,
task: str,
initial_response: str,
reflection: str,
max_tokens: int = 500) -> str:
"""
Generate revised attempt using reflection.
Args:
task: Original task
initial_response: First attempt that failed
reflection: Generated reflection on improvements
max_tokens: Maximum response length
Returns:
revised_response: Improved attempt
"""
prompt = f"""
Task: {task}
Previous attempt that failed: {initial_response}
Analysis of improvements needed: {reflection}
Based on the analysis, provide a revised solution. Apply all improvements identified in the analysis.
Revised solution:"""
revised = self.model.generate(prompt, max_tokens=max_tokens, temperature=0.6)
return revised.strip()
Step 4: Experiential RL training loop
classExperientialRL:
"""Full ERL training with reflection, revision, and distillation."""def__init__(
self,
model,
optimizer,
verifier,
reflection_memory: ReflectionMemory = None):
self.model = model
self.optimizer = optimizer
self.verifier = verifier
self.reflection_generator = ReflectionGenerator(model)
self.revision_generator = RevisionGenerator(model)
self.reflection_memory = reflection_memory or ReflectionMemory()
defexperiential_episode(
self,
task: str,
task_type: str = None,
max_revisions: int = 2) -> Dict:
"""
Execute single ERL episode with reflection and revision.
Returns:
episode_data: {initial, reflection, revision, success, log_probs}
"""
episode_data = {
'task': task,
'attempts': [],
'success': False,
'final_reflection': None
}
# Initial attempt
initial_response, log_probs_initial = self.model.generate_with_logprobs(
task, max_tokens=500
)
episode_data['attempts'].append({
'response': initial_response,
'log_probs': log_probs_initial,
'is_revision': False
})
initial_success = self.verifier(initial_response, task)
if initial_success:
episode_data['success'] = Truereturn episode_data
# Get feedback
feedback = self._generate_feedback(initial_response, task)
# Reflection and revision loopfor revision_idx inrange(max_revisions):
# Generate reflection
reflection = self.reflection_generator.generate_reflection(
task, initial_response, feedback,
self.reflection_memory, task_type
)
episode_data['final_reflection'] = reflection
# Generate revision
revised_response, log_probs_revised = self.model.generate_with_logprobs(
f"{task}\n\nAnalysis of improvements: {reflection}\n\nRevised solution:",
max_tokens=500
)
episode_data['attempts'].append({
'response': revised_response,
'log_probs': log_probs_revised,
'is_revision': True,
'reflection': reflection
})
# Check success
revision_success = self.verifier(revised_response, task)
if revision_success:
episode_data['success'] = True# Store successful correctionif task_type:
self.reflection_memory.store_successful_correction(
task_type,
initial_response,
reflection,
revised_response,
task
)
return episode_data
# Update for next iteration
initial_response = revised_response
feedback = self._generate_feedback(revised_response, task)
return episode_data
def_generate_feedback(self, response: str, task: str) -> str:
"""Generate feedback on response failure."""# Simplified: use rule-based or LLM feedbackreturnf"Your response to '{task[:50]}...' was incorrect."defdistill_successful_revision(
self,
episode_data: Dict) -> float:
"""
Distill successful revision into base policy via supervised loss.
Args:
episode_data: Episode with successful revision
Returns:
loss: Distillation loss
"""ifnot episode_data['success']:
return0.0# Find successful revision attempt
successful_attempt = Nonefor attempt in episode_data['attempts']:
if attempt['is_revision']:
successful_attempt = attempt
breakif successful_attempt isNone:
return0.0# Supervised fine-tuning on successful response
task = episode_data['task']
successful_response = successful_attempt['response']
# Forward pass: model predicts successful response
prompt_logits = self.model.forward(task)
response_logits = self.model.forward(
f"{task}\n\nAnswer: {successful_response}"
)
# Cross-entropy loss on successful tokens
loss = self._compute_ce_loss(response_logits, successful_response)
# Backward and optimizeself.optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
self.optimizer.step()
return loss.item()
def_compute_ce_loss(self, logits: torch.Tensor, target_text: str) -> torch.Tensor:
"""Compute cross-entropy loss (simplified)."""# Simplified: would tokenize and compute actual CEreturn torch.tensor(0.0, requires_grad=True)