"""
Pattern 79: Self-Refinement Pattern
Agents iteratively refine outputs using self-feedback without external critique.
Automatically improves responses through multiple refinement iterations.
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class RefinementCriteria(Enum):
"""Criteria for refinement evaluation."""
CLARITY = "clarity"
ACCURACY = "accuracy"
COMPLETENESS = "completeness"
COHERENCE = "coherence"
CONCISENESS = "conciseness"
@dataclass
class RefinementStep:
"""Single refinement iteration."""
step_number: int
original_text: str
refined_text: str
improvements_made: List[str]
criteria_scores: Dict[RefinementCriteria, float]
timestamp: datetime = field(default_factory=datetime.now)
class SelfRefiningAgent:
"""Agent with self-refinement capabilities."""
def __init__(self, name: str, model_name: str = "gpt-4"):
self.name = name
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.refinement_history: List[RefinementStep] = []
def generate_initial(self, task: str) -> str:
"""Generate initial response."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}. Provide your response to the task."),
("human", "{task}")
])
chain = prompt | self.llm
response = chain.invoke({"task": task})
return str(response.content)
def self_critique(self, text: str, criteria: List[RefinementCriteria]) -> tuple[List[str], Dict[RefinementCriteria, float]]:
"""Critique own output against criteria."""
criteria_text = "\n".join([f"- {c.value.capitalize()}" for c in criteria])
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name} critiquing your own work. Be honest and specific."),
("human", f"Your text:\n{text}\n\n"
f"Evaluate your text on these criteria:\n{criteria_text}\n\n"
f"For each criterion:\n"
f"1. Rate it from 0.0 to 1.0\n"
f"2. Identify specific improvements needed\n\n"
f"Format: criterion_name: score | improvement")
])
chain = prompt | self.llm
response = chain.invoke({})
# Parse response
improvements = []
scores = {}
for line in str(response.content).split('\n'):
line = line.strip()
if '|' in line:
parts = line.split('|')
if len(parts) >= 2:
improvement = parts[1].strip()
if improvement:
improvements.append(improvement)
# Try to extract score
score_part = parts[0].strip()
for criterion in criteria:
if criterion.value in score_part.lower():
import re
nums = re.findall(r'0\.\d+|1\.0|1', score_part)
if nums:
try:
scores[criterion] = float(nums[0])
except:
scores[criterion] = 0.5
# Fill in missing scores
for criterion in criteria:
if criterion not in scores:
scores[criterion] = 0.5
return improvements[:5], scores
def refine_text(self, original: str, improvements: List[str]) -> str:
"""Refine text based on identified improvements."""
improvements_text = "\n".join([f"{i+1}. {imp}" for i, imp in enumerate(improvements)])
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}. Improve your text based on self-critique."),
("human", f"Original text:\n{original}\n\n"
f"Improvements to make:\n{improvements_text}\n\n"
f"Provide the refined version that addresses these improvements.")
])
chain = prompt | self.llm
response = chain.invoke({})
return str(response.content)
class SelfRefinementPattern:
"""
Self-Refinement Pattern - Iterative self-improvement without external feedback.
Agent generates output, self-critiques against criteria, and refines
through multiple iterations until quality threshold is met.
"""
def __init__(self, max_iterations: int = 3,
criteria: Optional[List[RefinementCriteria]] = None):
self.max_iterations = max_iterations
self.criteria = criteria or [
RefinementCriteria.CLARITY,
RefinementCriteria.ACCURACY,
RefinementCriteria.COMPLETENESS
]
self.agent: Optional[SelfRefiningAgent] = None
def set_agent(self, agent: SelfRefiningAgent):
"""Set the self-refining agent."""
self.agent = agent
def refine(self, task: str, quality_threshold: float = 0.8) -> tuple[str, List[RefinementStep]]:
"""Execute self-refinement process."""
if not self.agent:
raise ValueError("No agent set")
print(f"\nš Starting self-refinement: {task}\n")
# Initial generation
text = self.agent.generate_initial(task)
print(f"š Initial output:\n {text[:200]}...\n")
steps = []
# Refinement iterations
for iteration in range(self.max_iterations):
print(f"š§ Refinement iteration {iteration + 1}:")
# Self-critique
improvements, scores = self.agent.self_critique(text, self.criteria)
avg_score = sum(scores.values()) / len(scores) if scores else 0
print(f" Criteria scores:")
for criterion, score in scores.items():
print(f" - {criterion.value}: {score:.2f}")
print(f" Average: {avg_score:.2f}")
# Check if quality threshold met
if avg_score >= quality_threshold:
print(f" ā
Quality threshold reached: {avg_score:.2f} >= {quality_threshold}")
break
# Refine
print(f" Applying {len(improvements)} improvements...")
refined = self.agent.refine_text(text, improvements)
# Record step
step = RefinementStep(
step_number=iteration,
original_text=text,
refined_text=refined,
improvements_made=improvements,
criteria_scores=scores
)
steps.append(step)
self.agent.refinement_history.append(step)
text = refined
print(f"\nā
Final refined output:\n {text[:200]}...\n")
return text, steps
def get_improvement_metrics(self, steps: List[RefinementStep]) -> Dict[str, Any]:
"""Calculate improvement metrics."""
if not steps:
return {"iterations": 0}
initial_scores = steps[0].criteria_scores if steps else {}
final_scores = steps[-1].criteria_scores if steps else {}
improvements = {}
for criterion in self.criteria:
initial = initial_scores.get(criterion, 0)
final = final_scores.get(criterion, 0)
improvements[criterion.value] = final - initial
total_improvements = sum(len(step.improvements_made) for step in steps)
return {
"iterations": len(steps),
"total_improvements_applied": total_improvements,
"score_improvements": improvements,
"initial_avg_score": sum(initial_scores.values()) / len(initial_scores) if initial_scores else 0,
"final_avg_score": sum(final_scores.values()) / len(final_scores) if final_scores else 0
}
# Example usage
if __name__ == "__main__":
print("=== Self-Refinement Pattern Demo ===\n")
# Create self-refining agent
print("1. Creating self-refining agent...")
agent = SelfRefiningAgent("Refiner")
# Create pattern with criteria
pattern = SelfRefinementPattern(
max_iterations=3,
criteria=[
RefinementCriteria.CLARITY,
RefinementCriteria.ACCURACY,
RefinementCriteria.COMPLETENESS,
RefinementCriteria.COHERENCE
]
)
pattern.set_agent(agent)
# Test tasks
tasks = [
"Explain blockchain technology",
"Write a product description for a smart watch",
"Describe the scientific method"
]
all_metrics = []
for i, task in enumerate(tasks, 1):
print(f"\n{'='*70}")
print(f"Task {i}/{len(tasks)}: {task}")
print(f"{'='*70}")
# Execute self-refinement
final_text, steps = pattern.refine(task, quality_threshold=0.75)
# Get metrics
metrics = pattern.get_improvement_metrics(steps)
all_metrics.append(metrics)
print(f"š Refinement Metrics:")
print(f" Iterations: {metrics['iterations']}")
print(f" Total improvements: {metrics['total_improvements_applied']}")
print(f" Initial avg score: {metrics['initial_avg_score']:.2f}")
print(f" Final avg score: {metrics['final_avg_score']:.2f}")
print(f" Overall improvement: {metrics['final_avg_score'] - metrics['initial_avg_score']:+.2f}")
print(f"\n Score improvements by criterion:")
for criterion, improvement in metrics['score_improvements'].items():
print(f" - {criterion}: {improvement:+.2f}")
# Overall statistics
print(f"\n\n{'='*70}")
print("š Overall Statistics")
print(f"{'='*70}")
avg_iterations = sum(m['iterations'] for m in all_metrics) / len(all_metrics)
avg_improvements = sum(m['total_improvements_applied'] for m in all_metrics) / len(all_metrics)
avg_score_gain = sum(m['final_avg_score'] - m['initial_avg_score'] for m in all_metrics) / len(all_metrics)
print(f" Tasks processed: {len(tasks)}")
print(f" Avg iterations: {avg_iterations:.1f}")
print(f" Avg improvements per task: {avg_improvements:.1f}")
print(f" Avg score improvement: {avg_score_gain:+.2f}")
# Show refinement progression for last task
if agent.refinement_history:
print(f"\n Refinement progression (last task):")
for step in agent.refinement_history[-3:]:
avg = sum(step.criteria_scores.values()) / len(step.criteria_scores)
print(f" Step {step.step_number + 1}: {avg:.2f} -> {len(step.improvements_made)} improvements")
print("\nā Self-refinement pattern enabled automatic quality improvement!")
Agentic Architecture Pattern: Self Refinement
Pattern 79: 79 Self Refinement – Line-by-Line Explanation
Overview
Pattern 79: Self-Refinement Pattern
Agents iteratively refine outputs using self-feedback without external critique. Automatically improves responses through multiple refinement iterations.
Code Structure Analysis
File Information
- Pattern Number: 79
- Pattern Name: 79 Self Refinement
- File Path:
patterns\execution_control\79_self_refinement.py - Total Lines: 294
Detailed Code Explanation
Module Docstring
"""
Pattern 79: Self-Refinement Pattern
Agents iteratively refine outputs using self-feedback without external critique.
Automatically improves responses through multiple refinement iterations.
"""
Explanation: This module docstring describes the purpose and key features of the 79 Self Refinement pattern.
Imports Section
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
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 enum import Enum: Enumeration support for fixed sets of values
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
Enumerations
RefinementCriteria (Line 16)
class RefinementCriteria(Enum):
CLARITY
ACCURACY
COMPLETENESS
COHERENCE
CONCISENESS
Purpose: Criteria for refinement evaluation.
Values:
– CLARITY: Represents clarity
– ACCURACY: Represents accuracy
– COMPLETENESS: Represents completeness
– COHERENCE: Represents coherence
– CONCISENESS: Represents conciseness
Classes
RefinementCriteria (Line 16)
Description: Criteria for refinement evaluation.
RefinementStep (Line 26)
Description: Single refinement iteration.
SelfRefiningAgent (Line 36)
Description: Agent with self-refinement 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_initial (Line 44)
def generate_initial(self, task):
"""Generate initial response."""
Purpose: Generate initial response.
Parameters:
– self: Instance reference
– task: Parameter for task
self_critique (Line 56)
def self_critique(self, text, criteria):
"""Critique own output against criteria."""
Purpose: Critique own output against criteria.
Parameters:
– self: Instance reference
– text: Parameter for text
– criteria: Parameter for criteria
refine_text (Line 105)
def refine_text(self, original, improvements):
"""Refine text based on identified improvements."""
Purpose: Refine text based on identified improvements.
Parameters:
– self: Instance reference
– original: Parameter for original
– improvements: Parameter for improvements
SelfRefinementPattern (Line 122)
Description: Self-Refinement Pattern – Iterative self-improvement without external feedback.
Agent generates output, self-critiques against criteria, and refines through multiple iterations until quality threshold is met.
Methods:
__init__ (Line 130)
def __init__(self, max_iterations, criteria):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– max_iterations: Parameter for max_iterations
– criteria: Parameter for criteria
set_agent (Line 140)
def set_agent(self, agent):
"""Set the self-refining agent."""
Purpose: Set the self-refining agent.
Parameters:
– self: Instance reference
– agent: Parameter for agent
refine (Line 144)
def refine(self, task, quality_threshold):
"""Execute self-refinement process."""
Purpose: Execute self-refinement process.
Parameters:
– self: Instance reference
– task: Parameter for task
– quality_threshold: Parameter for quality_threshold
get_improvement_metrics (Line 195)
def get_improvement_metrics(self, steps):
"""Calculate improvement metrics."""
Purpose: Calculate improvement metrics.
Parameters:
– self: Instance reference
– steps: Parameter for steps
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 = RefinementCriteria(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