"""
Pattern 78: Reflection Pattern
Agents reflect on their outputs, identify improvements, and refine responses.
Implements self-critique and iterative improvement through reflection.
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
@dataclass
class ReflectionCycle:
"""Single reflection cycle."""
cycle_id: int
original_output: str
reflection: str
improvements: List[str]
refined_output: str
quality_score: float # 0 to 1
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class ReflectionMetrics:
"""Metrics for reflection pattern."""
total_cycles: int
avg_quality_improvement: float
final_quality_score: float
total_improvements: int
class ReflectiveAgent:
"""Agent with reflection capabilities."""
def __init__(self, name: str, model_name: str = "gpt-4"):
self.name = name
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.reflection_history: List[ReflectionCycle] = []
def generate(self, task: str) -> str:
"""Generate initial response."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}. Provide your best answer to the task."),
("human", "{task}")
])
chain = prompt | self.llm
response = chain.invoke({"task": task})
return str(response.content)
def reflect(self, task: str, output: str) -> tuple[str, List[str]]:
"""Reflect on output and identify improvements."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name} reflecting on your previous response. "
f"Be critical and identify specific improvements."),
("human", f"Task: {task}\n\n"
f"Your previous response:\n{output}\n\n"
f"Reflect on your response:\n"
f"1. What did you do well?\n"
f"2. What could be improved?\n"
f"3. What specific changes would make it better?\n\n"
f"Provide your reflection and list 2-3 specific improvements.")
])
chain = prompt | self.llm
response = chain.invoke({})
reflection_text = str(response.content)
# Extract improvements
improvements = []
lines = reflection_text.split('\n')
for line in lines:
line = line.strip()
if line and (line[0].isdigit() or line.startswith('-') or line.startswith('ā¢')):
improvements.append(line.lstrip('0123456789.-ā¢) ').strip())
return reflection_text, improvements[:5] # Max 5 improvements
def refine(self, task: str, previous_output: str,
improvements: List[str]) -> str:
"""Refine output based on improvements."""
improvements_text = "\n".join([f"- {imp}" for imp in improvements])
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}. Improve your previous response based on reflections."),
("human", f"Task: {task}\n\n"
f"Previous response:\n{previous_output}\n\n"
f"Identified improvements:\n{improvements_text}\n\n"
f"Provide an improved response that addresses these points.")
])
chain = prompt | self.llm
response = chain.invoke({})
return str(response.content)
def evaluate_quality(self, task: str, output: str) -> float:
"""Evaluate quality of output (0 to 1)."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are an objective quality evaluator."),
("human", f"Task: {task}\n\n"
f"Response: {output}\n\n"
f"Rate the quality of this response on a scale of 0 to 1, where:\n"
f"0 = Very poor, 0.5 = Acceptable, 1 = Excellent\n\n"
f"Consider accuracy, completeness, clarity, and relevance.\n"
f"Respond with just a number between 0 and 1.")
])
chain = prompt | self.llm
response = chain.invoke({})
try:
# Extract number from response
import re
numbers = re.findall(r'0\.\d+|1\.0|1', str(response.content))
if numbers:
score = float(numbers[0])
return min(1.0, max(0.0, score))
except:
pass
return 0.5 # Default score
class ReflectionPattern:
"""
Reflection Pattern - Self-critique and iterative improvement.
Agent generates output, reflects on it, identifies improvements,
and refines the response through multiple reflection cycles.
"""
def __init__(self, max_cycles: int = 3, quality_threshold: float = 0.85):
self.max_cycles = max_cycles
self.quality_threshold = quality_threshold
self.agent: Optional[ReflectiveAgent] = None
def set_agent(self, agent: ReflectiveAgent):
"""Set the reflective agent."""
self.agent = agent
def execute(self, task: str) -> tuple[str, List[ReflectionCycle]]:
"""Execute reflection pattern on task."""
if not self.agent:
raise ValueError("No agent set for reflection")
print(f"\nš Starting reflection pattern for: {task}\n")
# Initial generation
output = self.agent.generate(task)
quality = self.agent.evaluate_quality(task, output)
print(f"š Initial output (quality: {quality:.2f}):")
print(f" {output[:150]}...")
reflection_cycles = []
# Reflection cycles
for cycle_num in range(self.max_cycles):
print(f"\nš Reflection Cycle {cycle_num + 1}:")
# Reflect
reflection, improvements = self.agent.reflect(task, output)
print(f" Found {len(improvements)} improvements:")
for imp in improvements[:3]:
print(f" - {imp[:80]}...")
# Refine
refined_output = self.agent.refine(task, output, improvements)
new_quality = self.agent.evaluate_quality(task, refined_output)
print(f" New quality: {new_quality:.2f} (was {quality:.2f})")
# Record cycle
cycle = ReflectionCycle(
cycle_id=cycle_num,
original_output=output,
reflection=reflection,
improvements=improvements,
refined_output=refined_output,
quality_score=new_quality
)
reflection_cycles.append(cycle)
self.agent.reflection_history.append(cycle)
# Update for next cycle
output = refined_output
quality = new_quality
# Check if quality threshold met
if quality >= self.quality_threshold:
print(f"\nā
Quality threshold reached: {quality:.2f} >= {self.quality_threshold}")
break
print(f"\nā Reflection complete after {len(reflection_cycles)} cycles")
print(f" Final output (quality: {quality:.2f}):")
print(f" {output[:150]}...")
return output, reflection_cycles
def get_metrics(self, cycles: List[ReflectionCycle]) -> ReflectionMetrics:
"""Calculate metrics from reflection cycles."""
if not cycles:
return ReflectionMetrics(0, 0.0, 0.0, 0)
quality_improvements = []
for i in range(len(cycles) - 1):
improvement = cycles[i + 1].quality_score - cycles[i].quality_score
quality_improvements.append(improvement)
avg_improvement = sum(quality_improvements) / len(quality_improvements) if quality_improvements else 0
total_improvements = sum(len(cycle.improvements) for cycle in cycles)
final_quality = cycles[-1].quality_score
return ReflectionMetrics(
total_cycles=len(cycles),
avg_quality_improvement=avg_improvement,
final_quality_score=final_quality,
total_improvements=total_improvements
)
# Example usage
if __name__ == "__main__":
print("=== Reflection Pattern Demo ===\n")
# Create reflective agent
print("1. Creating reflective agent...")
agent = ReflectiveAgent("Thinker")
# Create reflection pattern
pattern = ReflectionPattern(max_cycles=3, quality_threshold=0.85)
pattern.set_agent(agent)
# Test tasks
tasks = [
"Explain quantum computing to a 10-year-old",
"Write a strategy for reducing carbon emissions in cities",
"Design an algorithm to detect fake news"
]
all_metrics = []
for i, task in enumerate(tasks, 1):
print(f"\n{'='*60}")
print(f"Task {i}/{len(tasks)}")
print(f"{'='*60}")
# Execute reflection
final_output, cycles = pattern.execute(task)
# Get metrics
metrics = pattern.get_metrics(cycles)
all_metrics.append(metrics)
print(f"\nš Metrics for this task:")
print(f" Reflection cycles: {metrics.total_cycles}")
print(f" Avg quality improvement: {metrics.avg_quality_improvement:+.3f}")
print(f" Final quality: {metrics.final_quality_score:.2f}")
print(f" Total improvements identified: {metrics.total_improvements}")
# Overall statistics
print(f"\n\n{'='*60}")
print("š Overall Statistics")
print(f"{'='*60}")
avg_cycles = sum(m.total_cycles for m in all_metrics) / len(all_metrics)
avg_final_quality = sum(m.final_quality_score for m in all_metrics) / len(all_metrics)
total_improvements = sum(m.total_improvements for m in all_metrics)
print(f" Tasks processed: {len(tasks)}")
print(f" Avg reflection cycles: {avg_cycles:.1f}")
print(f" Avg final quality: {avg_final_quality:.2f}")
print(f" Total improvements: {total_improvements}")
print("\nā Reflection pattern enabled self-improvement!")
Agentic Architecture Pattern: Reflection
Pattern 78: 78 Reflection – Line-by-Line Explanation
Overview
Pattern 78: Reflection Pattern
Agents reflect on their outputs, identify improvements, and refine responses. Implements self-critique and iterative improvement through reflection.
Code Structure Analysis
File Information
- Pattern Number: 78
- Pattern Name: 78 Reflection
- File Path:
patterns\execution_control\78_reflection.py - Total Lines: 286
Detailed Code Explanation
Module Docstring
"""
Pattern 78: Reflection Pattern
Agents reflect on their outputs, identify improvements, and refine responses.
Implements self-critique and iterative improvement through reflection.
"""
Explanation: This module docstring describes the purpose and key features of the 78 Reflection pattern.
Imports Section
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime: Date and time utilities
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
Classes
ReflectionCycle (Line 16)
Description: Single reflection cycle.
ReflectionMetrics (Line 28)
Description: Metrics for reflection pattern.
ReflectiveAgent (Line 36)
Description: Agent with reflection capabilities.
Methods:
__init__ (Line 39)
def __init__(self, name, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– model_name: LLM model to use (e.g., “gpt-4”)
generate (Line 44)
def generate(self, task):
"""Generate initial response."""
Purpose: Generate initial response.
Parameters:
– self: Instance reference
– task: Parameter for task
reflect (Line 56)
def reflect(self, task, output):
"""Reflect on output and identify improvements."""
Purpose: Reflect on output and identify improvements.
Parameters:
– self: Instance reference
– task: Parameter for task
– output: Parameter for output
refine (Line 85)
def refine(self, task, previous_output, improvements):
"""Refine output based on improvements."""
Purpose: Refine output based on improvements.
Parameters:
– self: Instance reference
– task: Parameter for task
– previous_output: Parameter for previous_output
– improvements: Parameter for improvements
evaluate_quality (Line 103)
def evaluate_quality(self, task, output):
"""Evaluate quality of output (0 to 1)."""
Purpose: Evaluate quality of output (0 to 1).
Parameters:
– self: Instance reference
– task: Parameter for task
– output: Parameter for output
ReflectionPattern (Line 131)
Description: Reflection Pattern – Self-critique and iterative improvement.
Agent generates output, reflects on it, identifies improvements, and refines the response through multiple reflection cycles.
Methods:
__init__ (Line 139)
def __init__(self, max_cycles, quality_threshold):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– max_cycles: Parameter for max_cycles
– quality_threshold: Parameter for quality_threshold
set_agent (Line 144)
def set_agent(self, agent):
"""Set the reflective agent."""
Purpose: Set the reflective agent.
Parameters:
– self: Instance reference
– agent: Parameter for agent
execute (Line 148)
def execute(self, task):
"""Execute reflection pattern on task."""
Purpose: Execute reflection pattern on task.
Parameters:
– self: Instance reference
– task: Parameter for task
get_metrics (Line 209)
def get_metrics(self, cycles):
"""Calculate metrics from reflection cycles."""
Purpose: Calculate metrics from reflection cycles.
Parameters:
– self: Instance reference
– cycles: Parameter for cycles
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = ReflectionCycle(name="example", model_name="gpt-4")
# Configure and execute
result = pattern.execute(data={"key": "value"})
# Check results
print(result)
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities
Leave a Reply