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