aishastr.live

Agentic Architecture Pattern: Iterative Refinement

Written by

admin

in


    """
Pattern 80: Iterative Refinement Pattern

Progressively improves output through multiple refinement iterations with
external feedback. Each iteration addresses specific issues identified in feedback.
"""

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 FeedbackType(Enum):
    """Types of feedback."""
    POSITIVE = "positive"
    NEGATIVE = "negative"
    SUGGESTION = "suggestion"
    CORRECTION = "correction"


@dataclass
class Feedback:
    """External feedback on output."""
    iteration: int
    feedback_type: FeedbackType
    content: str
    severity: float = 0.5  # 0.0 (minor) to 1.0 (critical)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class RefinementIteration:
    """Single refinement iteration."""
    iteration_number: int
    input_query: str
    output: str
    feedback: List[Feedback]
    improvements: List[str]
    timestamp: datetime = field(default_factory=datetime.now)


class IterativeRefinementAgent:
    """Agent that refines output based on feedback."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.iterations: List[RefinementIteration] = []
    
    def generate_initial(self, query: str) -> str:
        """Generate initial output."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. Provide a comprehensive response."),
            ("human", "{query}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"query": query})
        return response.content
    
    def refine_with_feedback(self, query: str, previous_output: str, 
                            feedback_list: List[Feedback]) -> tuple[str, List[str]]:
        """Refine output based on feedback."""
        feedback_text = "\n".join([
            f"- [{fb.feedback_type.value}] {fb.content} (severity: {fb.severity})"
            for fb in feedback_list
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. Improve your previous response based on feedback."),
            ("human", """Original Query: {query}

Previous Response: {previous_output}

Feedback Received:
{feedback}

Provide an improved response that addresses ALL feedback points. 
Then list specific improvements made.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "query": query,
            "previous_output": previous_output,
            "feedback": feedback_text
        })
        
        # Parse improvements (assume agent lists them)
        output = response.content
        improvements = self._extract_improvements(output)
        
        return output, improvements
    
    def _extract_improvements(self, output: str) -> List[str]:
        """Extract improvement list from output."""
        lines = output.split("\n")
        improvements = []
        in_improvements = False
        
        for line in lines:
            if "improvement" in line.lower() or "changes made" in line.lower():
                in_improvements = True
                continue
            if in_improvements and line.strip().startswith(("-", "*", "ā€¢")):
                improvements.append(line.strip()[1:].strip())
        
        return improvements if improvements else ["General improvements applied"]


class IterativeRefinementPattern:
    """
    Iterative Refinement Pattern - Progressive improvement with feedback.
    
    Generates initial output, receives external feedback, refines iteratively
    until quality criteria met or max iterations reached.
    """
    
    def __init__(self, max_iterations: int = 5):
        self.max_iterations = max_iterations
        self.agents: Dict[str, IterativeRefinementAgent] = {}
    
    def register_agent(self, agent: IterativeRefinementAgent):
        """Register refinement agent."""
        self.agents[agent.name] = agent
    
    def run(self, query: str, feedback_provider: Callable[[str, int], List[Feedback]],
            stopping_criterion: Callable[[List[RefinementIteration]], bool] = None) -> Dict[str, Any]:
        """
        Execute iterative refinement.
        
        Args:
            query: Input query/task
            feedback_provider: Function that generates feedback for output
            stopping_criterion: Optional function to determine when to stop
        """
        print(f"=== Iterative Refinement Pattern ===\n")
        print(f"Query: {query}\n")
        
        for agent in self.agents.values():
            print(f"\n{agent.name} - Starting refinement process...")
            
            # Initial generation
            print(f"\n  Iteration 1: Initial generation")
            output = agent.generate_initial(query)
            print(f"  Output: {output[:150]}...")
            
            # Get feedback
            feedback_list = feedback_provider(output, 1)
            print(f"  Feedback: {len(feedback_list)} points received")
            
            iteration = RefinementIteration(
                iteration_number=1,
                input_query=query,
                output=output,
                feedback=feedback_list,
                improvements=[]
            )
            agent.iterations.append(iteration)
            
            # Refinement loop
            for i in range(2, self.max_iterations + 1):
                if stopping_criterion and stopping_criterion(agent.iterations):
                    print(f"\n  Stopping criterion met at iteration {i-1}")
                    break
                
                if not feedback_list or all(fb.feedback_type == FeedbackType.POSITIVE 
                                           for fb in feedback_list):
                    print(f"\n  All feedback positive, stopping at iteration {i-1}")
                    break
                
                print(f"\n  Iteration {i}: Refining based on feedback")
                output, improvements = agent.refine_with_feedback(
                    query, output, feedback_list
                )
                print(f"  Output: {output[:150]}...")
                print(f"  Improvements: {len(improvements)} changes made")
                
                # Get new feedback
                feedback_list = feedback_provider(output, i)
                print(f"  Feedback: {len(feedback_list)} points received")
                
                iteration = RefinementIteration(
                    iteration_number=i,
                    input_query=query,
                    output=output,
                    feedback=feedback_list,
                    improvements=improvements
                )
                agent.iterations.append(iteration)
        
        return {
            "pattern": "IterativeRefinement",
            "agents": len(self.agents),
            "iterations_per_agent": {name: len(agent.iterations) 
                                    for name, agent in self.agents.items()},
            "final_outputs": {name: agent.iterations[-1].output 
                            for name, agent in self.agents.items()}
        }


# Example usage
if __name__ == "__main__":
    from typing import Callable
    
    print("=== Iterative Refinement Pattern Demo ===\n")
    
    # Create pattern
    pattern = IterativeRefinementPattern(max_iterations=4)
    
    # Create agent
    agent = IterativeRefinementAgent("ContentWriter")
    pattern.register_agent(agent)
    
    # Simulated feedback provider
    def provide_feedback(output: str, iteration: int) -> List[Feedback]:
        """Simulate external reviewer feedback."""
        if iteration == 1:
            return [
                Feedback(iteration, FeedbackType.NEGATIVE, 
                        "Add more specific examples", severity=0.8),
                Feedback(iteration, FeedbackType.SUGGESTION,
                        "Structure with clear headings", severity=0.6),
                Feedback(iteration, FeedbackType.CORRECTION,
                        "Fix technical accuracy in section 2", severity=0.9)
            ]
        elif iteration == 2:
            return [
                Feedback(iteration, FeedbackType.POSITIVE,
                        "Examples are now clear", severity=0.0),
                Feedback(iteration, FeedbackType.SUGGESTION,
                        "Add a conclusion section", severity=0.5)
            ]
        elif iteration == 3:
            return [
                Feedback(iteration, FeedbackType.POSITIVE,
                        "Excellent structure and content", severity=0.0)
            ]
        return []
    
    # Run refinement
    query = "Write a technical article about microservices architecture"
    summary = pattern.run(query, provide_feedback)
    
    print(f"\n\n=== Summary ===")
    print(f"Total iterations: {summary['iterations_per_agent']['ContentWriter']}")
    print(f"Final output length: {len(summary['final_outputs']['ContentWriter'])} chars")
    
    print(f"\n=== Refinement History ===")
    for iteration in agent.iterations:
        print(f"\nIteration {iteration.iteration_number}:")
        print(f"  Feedback points: {len(iteration.feedback)}")
        print(f"  Improvements made: {len(iteration.improvements)}")
        if iteration.improvements:
            print(f"  Changes: {iteration.improvements[0][:60]}...")
    
    print("\nāœ“ Iterative refinement with external feedback complete!")

Agentic Architecture Pattern: Iterative Refinement

Pattern 80: 80 Iterative Refinement – Line-by-Line Explanation

Overview

Pattern 80: Iterative Refinement Pattern

Progressively improves output through multiple refinement iterations with external feedback. Each iteration addresses specific issues identified in feedback.

Code Structure Analysis

File Information

  • Pattern Number: 80
  • Pattern Name: 80 Iterative Refinement
  • File Path: patterns\execution_control\80_iterative_refinement.py
  • Total Lines: 261

Detailed Code Explanation

Module Docstring

"""
Pattern 80: Iterative Refinement Pattern

Progressively improves output through multiple refinement iterations with
external feedback. Each iteration addresses specific issues identified in feedback.
"""

Explanation: This module docstring describes the purpose and key features of the 80 Iterative 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
    from typing import Callable  # 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 – from typing import Callable: Type hints for better code documentation

Enumerations

FeedbackType (Line 16)

class FeedbackType(Enum):
    POSITIVE
    NEGATIVE
    SUGGESTION
    CORRECTION

Purpose: Types of feedback.

Values: – POSITIVE: Represents positive – NEGATIVE: Represents negative – SUGGESTION: Represents suggestion – CORRECTION: Represents correction

Classes

FeedbackType (Line 16)

Description: Types of feedback.

Feedback (Line 25)

Description: External feedback on output.

RefinementIteration (Line 35)

Description: Single refinement iteration.

IterativeRefinementAgent (Line 45)

Description: Agent that refines output based on feedback.

Methods:

__init__ (Line 48)
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 53)
def generate_initial(self, query):
    """Generate initial output."""

Purpose: Generate initial output.

Parameters: – self: Instance reference – query: Parameter for query

refine_with_feedback (Line 64)
def refine_with_feedback(self, query, previous_output, feedback_list):
    """Refine output based on feedback."""

Purpose: Refine output based on feedback.

Parameters: – self: Instance reference – query: Parameter for query – previous_output: Parameter for previous_output – feedback_list: Parameter for feedback_list

_extract_improvements (Line 98)
def _extract_improvements(self, output):
    """Extract improvement list from output."""

Purpose: Extract improvement list from output.

Parameters: – self: Instance reference – output: Parameter for output

IterativeRefinementPattern (Line 114)

Description: Iterative Refinement Pattern – Progressive improvement with feedback.

Generates initial output, receives external feedback, refines iteratively until quality criteria met or max iterations reached.

Methods:

__init__ (Line 122)
def __init__(self, max_iterations):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – max_iterations: Parameter for max_iterations

register_agent (Line 126)
def register_agent(self, agent):
    """Register refinement agent."""

Purpose: Register refinement agent.

Parameters: – self: Instance reference – agent: Parameter for agent

run (Line 130)
def run(self, query, feedback_provider, stopping_criterion):
    """Execute iterative refinement.

Args:
    query: Input query/task
    feedback_provider: Function that generates feedback for output
    stopping_criterion: Optional function to determine when to stop"""

Purpose: Execute iterative refinement.

Args: query: Input query/task feedback_provider: Function that generates feedback for output stopping_criterion: Optional function to determine when to stop

Parameters: – self: Instance reference – query: Parameter for query – feedback_provider: Parameter for feedback_provider – stopping_criterion: Parameter for stopping_criterion

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 = FeedbackType(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