aishastr.live

Agentic Architecture Pattern: Self Refinement

Written by

admin

in


    """
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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts