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.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
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)