"""
Pattern 014: Leader-Follower

Description:
    The Leader-Follower pattern implements hierarchical agent coordination where one
    leader/coordinator agent manages and delegates tasks to multiple follower/specialist
    agents. The leader is responsible for task decomposition, delegation, coordination,
    and result synthesis.

Components:
    - Leader Agent: Coordinates, delegates, and synthesizes results
    - Follower Agents: Specialized agents that execute specific tasks
    - Communication Protocol: Request-response between leader and followers
    - Task Assignment Strategy: How tasks are allocated to followers

Use Cases:
    - Complex workflows requiring coordination
    - Tasks benefiting from specialization
    - Multi-step processes with dependencies
    - Projects requiring project management

LangChain Implementation:
    Uses a coordinator LLM for task planning and delegation, specialized LLMs
    with different system prompts for specific domains, and implements task
    assignment, execution tracking, and result synthesis.

Key Features:
    - Dynamic task decomposition
    - Specialist agent assignment
    - Progress tracking and monitoring
    - Hierarchical result synthesis
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class TaskStatus(Enum):
    """Status of a task in the workflow."""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


class SpecialistType(Enum):
    """Types of specialist agents available."""
    RESEARCH = "research"
    ANALYSIS = "analysis"
    WRITING = "writing"
    CODING = "coding"
    MATH = "math"
    CREATIVE = "creative"


@dataclass
class Task:
    """Represents a task to be completed."""
    task_id: str
    description: str
    specialist_type: SpecialistType
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[str] = None
    dependencies: List[str] = field(default_factory=list)


@dataclass
class WorkflowResult:
    """Result from complete workflow execution."""
    final_output: str
    tasks: List[Task]
    execution_summary: str


class SpecialistAgent:
    """
    A specialist agent with expertise in a specific domain.
    """
    
    def __init__(
        self,
        specialist_type: SpecialistType,
        model: str = "gpt-3.5-turbo",
        temperature: float = 0.7
    ):
        """
        Initialize a specialist agent.
        
        Args:
            specialist_type: Type of specialist
            model: LLM model to use
            temperature: Creativity level
        """
        self.specialist_type = specialist_type
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        
        # Define specialist system prompts
        self.system_prompts = {
            SpecialistType.RESEARCH: """You are a research specialist. Your expertise is in:
- Finding and summarizing information
- Conducting literature reviews
- Fact-checking and verification
- Identifying credible sources
Provide thorough, well-researched responses with citations when possible.""",
            
            SpecialistType.ANALYSIS: """You are an analysis specialist. Your expertise is in:
- Data analysis and interpretation
- Pattern recognition
- Critical thinking and evaluation
- Drawing insights from information
Provide detailed analytical insights and evidence-based conclusions.""",
            
            SpecialistType.WRITING: """You are a writing specialist. Your expertise is in:
- Clear and engaging writing
- Content structuring and organization
- Grammar and style
- Adapting tone for different audiences
Provide well-written, polished content.""",
            
            SpecialistType.CODING: """You are a coding specialist. Your expertise is in:
- Writing clean, efficient code
- Debugging and problem-solving
- Best practices and design patterns
- Multiple programming languages
Provide production-ready code with explanations.""",
            
            SpecialistType.MATH: """You are a mathematics specialist. Your expertise is in:
- Mathematical problem-solving
- Statistical analysis
- Logical reasoning
- Step-by-step calculations
Provide accurate solutions with clear explanations.""",
            
            SpecialistType.CREATIVE: """You are a creative specialist. Your expertise is in:
- Brainstorming and ideation
- Creative problem-solving
- Innovative thinking
- Generating unique perspectives
Provide creative, out-of-the-box ideas and solutions."""
        }
    
    def execute_task(self, task: Task, context: Optional[str] = None) -> str:
        """
        Execute a task using specialist expertise.
        
        Args:
            task: Task to execute
            context: Optional context from previous tasks
            
        Returns:
            Task result as string
        """
        system_prompt = self.system_prompts[self.specialist_type]
        
        user_message = f"Task: {task.description}"
        if context:
            user_message += f"\n\nContext from previous tasks:\n{context}"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            ("user", user_message)
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({})
        
        return result


class LeaderAgent:
    """
    Leader agent that coordinates and manages specialist agents.
    
    Responsibilities:
    - Decompose complex tasks into subtasks
    - Assign tasks to appropriate specialists
    - Monitor progress and handle dependencies
    - Synthesize results into final output
    """
    
    def __init__(self, model: str = "gpt-3.5-turbo"):
        """Initialize the leader agent."""
        self.llm = ChatOpenAI(model=model, temperature=0.3)
        
        # Initialize specialist agents
        self.specialists: Dict[SpecialistType, SpecialistAgent] = {
            specialist_type: SpecialistAgent(specialist_type)
            for specialist_type in SpecialistType
        }
    
    def decompose_task(self, overall_task: str) -> List[Task]:
        """
        Decompose an overall task into subtasks for specialists.
        
        Args:
            overall_task: The main task to decompose
            
        Returns:
            List of subtasks with assigned specialists
        """
        specialist_descriptions = "\n".join([
            f"- {s.value}: {self.specialists[s].system_prompts[s].split('Your expertise is in:')[1].split('Provide')[0].strip()}"
            for s in SpecialistType
        ])
        
        decomposition_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a project coordinator. Decompose tasks into subtasks and assign
to appropriate specialists. Be specific and ensure tasks are actionable."""),
            ("user", """Decompose this task into 3-5 subtasks:

Task: {overall_task}

Available specialists:
{specialist_descriptions}

Provide your decomposition in this exact format:
TASK 1: [specialist_type] - [clear task description]
TASK 2: [specialist_type] - [clear task description]
...

Use these specialist types: {specialist_types}""")
        ])
        
        chain = decomposition_prompt | self.llm | StrOutputParser()
        
        decomposition = chain.invoke({
            "overall_task": overall_task,
            "specialist_descriptions": specialist_descriptions,
            "specialist_types": ", ".join([s.value for s in SpecialistType])
        })
        
        # Parse the decomposition
        tasks = []
        for i, line in enumerate(decomposition.strip().split("\n")):
            if line.strip().startswith("TASK"):
                try:
                    # Extract specialist type and description
                    parts = line.split(":", 1)[1].strip()
                    specialist_str, description = parts.split("-", 1)
                    specialist_str = specialist_str.strip().lower()
                    
                    # Map to specialist type
                    specialist_type = None
                    for s in SpecialistType:
                        if s.value in specialist_str:
                            specialist_type = s
                            break
                    
                    if specialist_type:
                        task = Task(
                            task_id=f"task_{i+1}",
                            description=description.strip(),
                            specialist_type=specialist_type
                        )
                        tasks.append(task)
                except Exception as e:
                    print(f"Error parsing task line: {line} - {e}")
                    continue
        
        return tasks
    
    def assign_dependencies(self, tasks: List[Task]) -> List[Task]:
        """
        Determine task dependencies for proper execution order.
        
        Args:
            tasks: List of tasks to analyze
            
        Returns:
            Tasks with dependencies populated
        """
        # Simple heuristic: research -> analysis -> others
        for i, task in enumerate(tasks):
            if task.specialist_type == SpecialistType.ANALYSIS:
                # Analysis depends on research
                for prev_task in tasks[:i]:
                    if prev_task.specialist_type == SpecialistType.RESEARCH:
                        task.dependencies.append(prev_task.task_id)
            
            elif task.specialist_type in [SpecialistType.WRITING, SpecialistType.CREATIVE]:
                # Writing/creative depends on research and analysis
                for prev_task in tasks[:i]:
                    if prev_task.specialist_type in [SpecialistType.RESEARCH, SpecialistType.ANALYSIS]:
                        task.dependencies.append(prev_task.task_id)
        
        return tasks
    
    def execute_workflow(self, tasks: List[Task]) -> List[Task]:
        """
        Execute tasks in proper order respecting dependencies.
        
        Args:
            tasks: List of tasks to execute
            
        Returns:
            Completed tasks with results
        """
        completed_tasks: Dict[str, Task] = {}
        
        # Build context from completed tasks
        def get_context_for_task(task: Task) -> str:
            context_parts = []
            for dep_id in task.dependencies:
                if dep_id in completed_tasks:
                    dep_task = completed_tasks[dep_id]
                    context_parts.append(
                        f"From {dep_id} ({dep_task.specialist_type.value}):\n{dep_task.result}\n"
                    )
            return "\n".join(context_parts)
        
        # Execute tasks
        for task in tasks:
            print(f"\n[Leader] Assigning {task.task_id} to {task.specialist_type.value} specialist...")
            
            task.status = TaskStatus.IN_PROGRESS
            
            try:
                # Get specialist agent
                specialist = self.specialists[task.specialist_type]
                
                # Get context from dependencies
                context = get_context_for_task(task)
                
                # Execute task
                result = specialist.execute_task(task, context)
                
                task.result = result
                task.status = TaskStatus.COMPLETED
                completed_tasks[task.task_id] = task
                
                print(f"[Leader] ✓ {task.task_id} completed by {task.specialist_type.value}")
                
            except Exception as e:
                print(f"[Leader] ✗ {task.task_id} failed: {e}")
                task.status = TaskStatus.FAILED
                task.result = f"Error: {str(e)}"
        
        return tasks
    
    def synthesize_results(
        self,
        overall_task: str,
        tasks: List[Task]
    ) -> str:
        """
        Synthesize results from all tasks into final output.
        
        Args:
            overall_task: Original task description
            tasks: Completed tasks with results
            
        Returns:
            Final synthesized result
        """
        # Compile all task results
        task_results = "\n\n".join([
            f"**{task.task_id}** ({task.specialist_type.value}):\n{task.result}"
            for task in tasks
            if task.status == TaskStatus.COMPLETED
        ])
        
        synthesis_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a project coordinator synthesizing results from multiple specialists.
Create a cohesive, comprehensive final output that integrates all specialist contributions."""),
            ("user", """Original Task: {overall_task}

Specialist Results:
{task_results}

Synthesize these results into a coherent final output that addresses the original task.
Integrate insights from all specialists into a unified response.""")
        ])
        
        chain = synthesis_prompt | self.llm | StrOutputParser()
        
        final_output = chain.invoke({
            "overall_task": overall_task,
            "task_results": task_results
        })
        
        return final_output
    
    def coordinate(self, overall_task: str) -> WorkflowResult:
        """
        Coordinate the complete workflow from task to final output.
        
        Args:
            overall_task: The main task to complete
            
        Returns:
            WorkflowResult with final output and execution details
        """
        print(f"\n[Leader] Coordinating task: {overall_task}\n")
        
        # Step 1: Decompose task
        print("[Leader] Step 1: Decomposing task...")
        tasks = self.decompose_task(overall_task)
        print(f"[Leader] Created {len(tasks)} subtasks")
        
        for task in tasks:
            print(f"  - {task.task_id}: {task.description[:60]}... [{task.specialist_type.value}]")
        
        # Step 2: Assign dependencies
        print("\n[Leader] Step 2: Determining task dependencies...")
        tasks = self.assign_dependencies(tasks)
        
        # Step 3: Execute workflow
        print("\n[Leader] Step 3: Executing workflow...")
        tasks = self.execute_workflow(tasks)
        
        # Step 4: Synthesize results
        print("\n[Leader] Step 4: Synthesizing results...")
        final_output = self.synthesize_results(overall_task, tasks)
        
        # Create execution summary
        completed = sum(1 for t in tasks if t.status == TaskStatus.COMPLETED)
        execution_summary = f"Completed {completed}/{len(tasks)} tasks using {len(set(t.specialist_type for t in tasks))} specialist types"
        
        print(f"\n[Leader] Workflow complete! {execution_summary}")
        
        return WorkflowResult(
            final_output=final_output,
            tasks=tasks,
            execution_summary=execution_summary
        )


def demonstrate_leader_follower():
    """Demonstrate the Leader-Follower pattern with various examples."""
    
    print("=" * 80)
    print("LEADER-FOLLOWER PATTERN DEMONSTRATION")
    print("=" * 80)
    
    leader = LeaderAgent()
    
    # Test 1: Research and analysis task
    print("\n" + "=" * 80)
    print("TEST 1: Complex Research and Analysis Task")
    print("=" * 80)
    
    task1 = "Analyze the impact of artificial intelligence on the job market and write a summary report"
    
    result = leader.coordinate(task1)
    
    print("\n" + "-" * 80)
    print("FINAL OUTPUT:")
    print("-" * 80)
    print(result.final_output)
    
    print("\n" + "-" * 80)
    print("TASK BREAKDOWN:")
    print("-" * 80)
    for task in result.tasks:
        status_symbol = "✓" if task.status == TaskStatus.COMPLETED else "✗"
        print(f"{status_symbol} {task.task_id} [{task.specialist_type.value}]: {task.description[:60]}...")
    
    # Test 2: Creative and technical task
    print("\n" + "=" * 80)
    print("TEST 2: Creative + Technical Task")
    print("=" * 80)
    
    task2 = "Design a simple Python game concept and create a basic implementation plan"
    
    result = leader.coordinate(task2)
    
    print("\n" + "-" * 80)
    print("FINAL OUTPUT:")
    print("-" * 80)
    print(result.final_output[:500] + "...")
    
    print("\n" + "-" * 80)
    print("SPECIALIST UTILIZATION:")
    print("-" * 80)
    specialist_usage = {}
    for task in result.tasks:
        specialist_type = task.specialist_type.value
        specialist_usage[specialist_type] = specialist_usage.get(specialist_type, 0) + 1
    
    for specialist, count in specialist_usage.items():
        print(f"  {specialist}: {count} task(s)")
    
    # Test 3: Multi-domain problem
    print("\n" + "=" * 80)
    print("TEST 3: Multi-Domain Problem Solving")
    print("=" * 80)
    
    task3 = "Calculate the ROI of implementing renewable energy in a small business and provide recommendations"
    
    result = leader.coordinate(task3)
    
    print("\n" + "-" * 80)
    print("FINAL OUTPUT:")
    print("-" * 80)
    print(result.final_output)
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
The Leader-Follower pattern demonstrates several key benefits:

1. **Specialization**: Each agent focuses on what it does best
2. **Coordination**: Leader manages workflow and dependencies
3. **Scalability**: Easy to add new specialist types
4. **Quality**: Specialist expertise improves output quality

Architecture:
- **Leader Agent**: Plans, delegates, coordinates, synthesizes
- **Follower Agents**: Execute specialized tasks with expertise
- **Task Management**: Handles dependencies and execution order
- **Result Synthesis**: Combines specialist outputs into coherent result

Use Cases:
- Complex workflows requiring multiple skills
- Projects benefiting from domain specialization
- Multi-step processes with dependencies
- Tasks requiring project management

The pattern is particularly effective when:
- Tasks can be decomposed into specialized subtasks
- Different subtasks require different expertise
- Coordination overhead is justified by quality improvement
- Clear task dependencies exist
""")


if __name__ == "__main__":
    demonstrate_leader_follower()

    

Agentic Design Pattern: Leader Follower

Pattern 014: Leader Follower – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Leader Follower pattern implementation.

---

Pattern Description

Pattern 014: Leader-Follower

Description: The Leader-Follower pattern implements hierarchical agent coordination where one leader/coordinator agent manages and delegates tasks to multiple follower/specialist agents. The leader is responsible for task decomposition, delegation, coordination, and result synthesis.

Components: – Leader Agent: Coordinates, delegates, and synthesizes results – Follower Agents: Specialized agents that execute specific tasks – Communication Protocol: Request-response between leader and followers – Task Assignment Strategy: How tasks are allocated to followers

Use Cases: – Complex workflows requiring coordination – Tasks benefiting from specialization – Multi-step processes with dependencies – Projects requiring project management

LangChain Implementation: Uses a coordinator LLM for task planning and delegation, specialized LLMs with different system prompts for specific domains, and implements task assignment, execution tracking, and result synthesis.

Key Features: – Dynamic task decomposition – Specialist agent assignment – Progress tracking and monitoring – Hierarchical result synthesis

---

File Statistics

• Total Lines: 540
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 014: Leader-Follower

Description:
    The Leader-Follower pattern implements hierarchical agent coordination where one
    leader/coordinator agent manages and delegates tasks to multiple follower/specialist
    agents. The leader is responsible for task decomposition, delegation, coordination,
    and result synthesis.

Components:
    - Leader Agent: Coordinates, delegates, and synthesizes results
    - Follower Agents: Specialized agents that execute specific tasks
    - Communication Protocol: Request-response between leader and followers
    - Task Assignment Strategy: How tasks are allocated to followers

Use Cases:
    - Complex workflows requiring coordination
    - Tasks benefiting from specialization
    - Multi-step processes with dependencies
    - Projects requiring project management

LangChain Implementation:
    Uses a coordinator LLM for task planning and delegation, specialized LLMs
    with different system prompts for specific domains, and implements task
    assignment, execution tracking, and result synthesis.

Key Features:
    - Dynamic task decomposition
    - Specialist agent assignment
    - Progress tracking and monitoring
    - Hierarchical result synthesis
"""

Purpose: This module implements the Leader Follower pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: TaskStatus

Lines 47-52

Inherits from: Enum

Purpose:

Status of a task in the workflow.

---

Class: SpecialistType

Lines 55-62

Inherits from: Enum

Purpose:

Types of specialist agents available.

---

Class: Task

Lines 66-73

Purpose:

Represents a task to be completed.

---

Class: WorkflowResult

Lines 77-81

Purpose:

Result from complete workflow execution.

---

Class: SpecialistAgent

Lines 84-176

Purpose:

A specialist agent with expertise in a specific domain.

Methods:

__init__(self, specialist_type, model, temperature)

Lines 89-149

Purpose:

Initialize a specialist agent.

Args:
    specialist_type: Type of specialist
    model: LLM model to use
    temperature: Creativity level

Parameters: – specialist_type: Specialist type – model: Language model instance – temperature: Sampling temperature (0-1)

execute_task(self, task, context)

Lines 151-176

Purpose:

Execute a task using specialist expertise.

Args:
    task: Task to execute
    context: Optional context from previous tasks

Returns:
    Task result as string

Parameters: – task: Task – context: Context

Returns: str

---

Class: LeaderAgent

Lines 179-434

Purpose:

Leader agent that coordinates and manages specialist agents.

Responsibilities:
- Decompose complex tasks into subtasks
- Assign tasks to appropriate specialists
- Monitor progress and handle dependencies
- Synthesize results into final output

Methods:

__init__(self, model)

Lines 190-198

Purpose:

Initialize the leader agent.

Parameters: – model: Language model instance

decompose_task(self, overall_task)

Lines 200-269

Purpose:

Decompose an overall task into subtasks for specialists.

Args:
    overall_task: The main task to decompose

Returns:
    List of subtasks with assigned specialists

Parameters: – overall_task: Overall task

Returns: <ast.Subscript object at 0x000001A06D633A50>

assign_dependencies(self, tasks)

Lines 271-295

Purpose:

Determine task dependencies for proper execution order.

Args:
    tasks: List of tasks to analyze

Returns:
    Tasks with dependencies populated

Parameters: – tasks: Tasks

Returns: <ast.Subscript object at 0x000001A06D633810>

execute_workflow(self, tasks)

Lines 297-347

Purpose:

Execute tasks in proper order respecting dependencies.

Args:
    tasks: List of tasks to execute

Returns:
    Completed tasks with results

Parameters: – tasks: Tasks

Returns: <ast.Subscript object at 0x000001A06D609D90>

synthesize_results(self, overall_task, tasks)

Lines 349-390

Purpose:

Synthesize results from all tasks into final output.

Args:
    overall_task: Original task description
    tasks: Completed tasks with results

Returns:
    Final synthesized result

Parameters: – overall_task: Overall task – tasks: Tasks

Returns: str

coordinate(self, overall_task)

Lines 392-434

Purpose:

Coordinate the complete workflow from task to final output.

Args:
    overall_task: The main task to complete

Returns:
    WorkflowResult with final output and execution details

Parameters: – overall_task: Overall task

Returns: WorkflowResult

---

Functions

demonstrate_leader_follower()

Lines 437-535

Purpose:

Demonstrate the Leader-Follower pattern with various examples.

---

Key Concepts Summary

Pattern Overview

The Leader Follower pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 013
• Pattern 015
• Other Multi-Agent Patterns (012-019)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Leader Follower pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 014_leader_follower.py
Total Lines: 540
Pattern Category: AI Agent Design Pattern #014

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 013: Ensemble/Committee

Description:
    The Ensemble/Committee pattern uses multiple agents working independently on the same
    task, then aggregates their outputs to reach a final decision. This pattern leverages
    diversity to improve robustness, reduce errors, and provide more reliable results.

Components:
    - Multiple independent agents (diverse models, prompts, or strategies)
    - Aggregation mechanism (voting, averaging, weighted combination)
    - Result synthesis and confidence scoring

Use Cases:
    - Classification tasks requiring high accuracy
    - Prediction problems where uncertainty quantification is important
    - Reducing bias through diverse perspectives
    - Quality assurance through redundancy

LangChain Implementation:
    Uses multiple ChatOpenAI instances with different configurations, implements
    various aggregation strategies (majority voting, weighted voting, consensus),
    and provides confidence metrics based on agreement levels.

Key Features:
    - Diverse agent configurations (temperature, models, prompts)
    - Multiple aggregation strategies
    - Confidence scoring based on agreement
    - Detailed breakdown of individual votes
"""

import os
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
from collections import Counter
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class AggregationStrategy(Enum):
    """Aggregation strategies for combining agent outputs."""
    MAJORITY_VOTE = "majority_vote"
    WEIGHTED_VOTE = "weighted_vote"
    CONSENSUS = "consensus"
    AVERAGING = "averaging"


@dataclass
class AgentVote:
    """Represents a single agent's vote/output."""
    agent_id: str
    output: str
    confidence: float
    reasoning: Optional[str] = None


@dataclass
class EnsembleResult:
    """Result from ensemble aggregation."""
    final_output: str
    confidence: float
    individual_votes: List[AgentVote]
    agreement_score: float
    strategy_used: str


class EnsembleAgent:
    """
    Ensemble agent that combines outputs from multiple independent agents.
    
    This implementation creates diverse agents through different configurations
    (temperature, system prompts) and provides multiple aggregation strategies.
    """
    
    def __init__(
        self,
        num_agents: int = 3,
        base_model: str = "gpt-3.5-turbo",
        temperatures: Optional[List[float]] = None
    ):
        """
        Initialize ensemble with multiple agents.
        
        Args:
            num_agents: Number of agents in ensemble
            base_model: Base model for all agents
            temperatures: List of temperatures for diversity (default: [0.3, 0.7, 1.0])
        """
        self.num_agents = num_agents
        self.base_model = base_model
        
        # Create diverse temperature settings if not provided
        if temperatures is None:
            temperatures = [0.3, 0.5, 0.7, 0.9, 1.0][:num_agents]
        
        # Create diverse agents with different temperatures
        self.agents = []
        for i, temp in enumerate(temperatures[:num_agents]):
            agent = ChatOpenAI(
                model=base_model,
                temperature=temp,
                model_kwargs={"seed": i * 42}  # Different seeds for diversity
            )
            self.agents.append(agent)
        
        # Create a meta-agent for confidence estimation
        self.meta_agent = ChatOpenAI(model=base_model, temperature=0.2)
    
    def _estimate_confidence(
        self,
        question: str,
        answer: str,
        reasoning: str
    ) -> float:
        """Estimate confidence score for an individual answer."""
        confidence_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert at evaluating answer quality and confidence."),
            ("user", """Given this question and answer, estimate the confidence level (0.0-1.0).
Consider clarity, completeness, and certainty in the reasoning.

Question: {question}

Answer: {answer}

Reasoning: {reasoning}

Provide only a number between 0.0 and 1.0.""")
        ])
        
        chain = confidence_prompt | self.meta_agent | StrOutputParser()
        
        try:
            confidence_str = chain.invoke({
                "question": question,
                "answer": answer,
                "reasoning": reasoning
            })
            confidence = float(confidence_str.strip())
            return max(0.0, min(1.0, confidence))
        except:
            return 0.5  # Default moderate confidence
    
    def query_agents(
        self,
        question: str,
        system_prompt: Optional[str] = None
    ) -> List[AgentVote]:
        """
        Query all agents independently and collect their votes.
        
        Args:
            question: Question to ask all agents
            system_prompt: Optional system prompt (default: general assistant)
            
        Returns:
            List of agent votes with outputs and confidence scores
        """
        if system_prompt is None:
            system_prompt = "You are a helpful AI assistant. Provide clear, accurate answers."
        
        prompt_template = ChatPromptTemplate.from_messages([
            ("system", system_prompt + "\n\nProvide your answer followed by your reasoning."),
            ("user", "{question}")
        ])
        
        votes = []
        
        for i, agent in enumerate(self.agents):
            chain = prompt_template | agent | StrOutputParser()
            
            try:
                response = chain.invoke({"question": question})
                
                # Split response into answer and reasoning
                parts = response.split("\n", 1)
                answer = parts[0].strip()
                reasoning = parts[1].strip() if len(parts) > 1 else ""
                
                # Estimate confidence
                confidence = self._estimate_confidence(question, answer, reasoning)
                
                vote = AgentVote(
                    agent_id=f"agent_{i}",
                    output=answer,
                    confidence=confidence,
                    reasoning=reasoning
                )
                votes.append(vote)
                
            except Exception as e:
                print(f"Agent {i} error: {e}")
                # Add a low-confidence fallback vote
                votes.append(AgentVote(
                    agent_id=f"agent_{i}",
                    output="Error generating response",
                    confidence=0.0,
                    reasoning=f"Error: {str(e)}"
                ))
        
        return votes
    
    def aggregate_majority_vote(
        self,
        votes: List[AgentVote]
    ) -> Tuple[str, float]:
        """
        Aggregate using simple majority voting.
        
        Returns:
            (winning answer, agreement score)
        """
        # Count votes for each unique answer
        answer_counts = Counter(vote.output for vote in votes)
        most_common = answer_counts.most_common(1)[0]
        winning_answer = most_common[0]
        vote_count = most_common[1]
        
        agreement_score = vote_count / len(votes)
        
        return winning_answer, agreement_score
    
    def aggregate_weighted_vote(
        self,
        votes: List[AgentVote]
    ) -> Tuple[str, float]:
        """
        Aggregate using confidence-weighted voting.
        
        Returns:
            (winning answer, weighted agreement score)
        """
        # Weight each answer by confidence
        answer_weights: Dict[str, float] = {}
        total_confidence = sum(vote.confidence for vote in votes)
        
        for vote in votes:
            if vote.output in answer_weights:
                answer_weights[vote.output] += vote.confidence
            else:
                answer_weights[vote.output] = vote.confidence
        
        # Find answer with highest weighted vote
        winning_answer = max(answer_weights.items(), key=lambda x: x[1])[0]
        winning_weight = answer_weights[winning_answer]
        
        # Calculate weighted agreement score
        agreement_score = winning_weight / total_confidence if total_confidence > 0 else 0.0
        
        return winning_answer, agreement_score
    
    def aggregate_consensus(
        self,
        votes: List[AgentVote],
        question: str
    ) -> Tuple[str, float]:
        """
        Generate consensus answer by synthesizing all votes.
        
        Returns:
            (consensus answer, confidence score)
        """
        # Compile all answers and reasoning
        vote_summary = "\n\n".join([
            f"Agent {vote.agent_id} (confidence: {vote.confidence:.2f}):\n"
            f"Answer: {vote.output}\n"
            f"Reasoning: {vote.reasoning}"
            for vote in votes
        ])
        
        consensus_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert at synthesizing multiple perspectives into a consensus."),
            ("user", """Given the following question and multiple agent responses, create a 
consensus answer that incorporates the best elements from all responses.

Question: {question}

Agent Responses:
{vote_summary}

Provide a consensus answer that:
1. Incorporates accurate information from all agents
2. Resolves any contradictions
3. Provides a clear, unified response

Consensus Answer:""")
        ])
        
        chain = consensus_prompt | self.meta_agent | StrOutputParser()
        
        consensus_answer = chain.invoke({
            "question": question,
            "vote_summary": vote_summary
        })
        
        # Calculate confidence based on average agent confidence
        avg_confidence = sum(vote.confidence for vote in votes) / len(votes)
        
        return consensus_answer.strip(), avg_confidence
    
    def solve(
        self,
        question: str,
        strategy: AggregationStrategy = AggregationStrategy.WEIGHTED_VOTE,
        system_prompt: Optional[str] = None
    ) -> EnsembleResult:
        """
        Solve a problem using the ensemble of agents.
        
        Args:
            question: Question to solve
            strategy: Aggregation strategy to use
            system_prompt: Optional custom system prompt
            
        Returns:
            EnsembleResult with final answer and metadata
        """
        # Get votes from all agents
        votes = self.query_agents(question, system_prompt)
        
        # Aggregate based on strategy
        if strategy == AggregationStrategy.MAJORITY_VOTE:
            final_output, agreement_score = self.aggregate_majority_vote(votes)
            
        elif strategy == AggregationStrategy.WEIGHTED_VOTE:
            final_output, agreement_score = self.aggregate_weighted_vote(votes)
            
        elif strategy == AggregationStrategy.CONSENSUS:
            final_output, agreement_score = self.aggregate_consensus(votes, question)
            
        else:
            raise ValueError(f"Unsupported strategy: {strategy}")
        
        # Calculate overall confidence
        confidence = agreement_score
        
        return EnsembleResult(
            final_output=final_output,
            confidence=confidence,
            individual_votes=votes,
            agreement_score=agreement_score,
            strategy_used=strategy.value
        )


def demonstrate_ensemble():
    """Demonstrate the Ensemble/Committee pattern with various examples."""
    
    print("=" * 80)
    print("ENSEMBLE/COMMITTEE PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Create ensemble with 5 agents
    ensemble = EnsembleAgent(num_agents=5)
    
    # Test 1: Classification task
    print("\n" + "=" * 80)
    print("TEST 1: Sentiment Classification")
    print("=" * 80)
    
    sentiment_question = """Classify the sentiment of this review as Positive, Negative, or Neutral:

"The product works okay, but the customer service was terrible. Mixed feelings overall."

Your answer should be exactly one word: Positive, Negative, or Neutral."""
    
    result = ensemble.solve(
        sentiment_question,
        strategy=AggregationStrategy.WEIGHTED_VOTE
    )
    
    print(f"\nQuestion: Sentiment classification of mixed review")
    print(f"\nFinal Answer: {result.final_output}")
    print(f"Confidence: {result.confidence:.2f}")
    print(f"Agreement Score: {result.agreement_score:.2f}")
    print(f"Strategy: {result.strategy_used}")
    
    print("\nIndividual Agent Votes:")
    for vote in result.individual_votes:
        print(f"  {vote.agent_id}: {vote.output} (confidence: {vote.confidence:.2f})")
    
    # Test 2: Factual question with consensus
    print("\n" + "=" * 80)
    print("TEST 2: Factual Question with Consensus")
    print("=" * 80)
    
    factual_question = "What are the main causes of climate change?"
    
    result = ensemble.solve(
        factual_question,
        strategy=AggregationStrategy.CONSENSUS
    )
    
    print(f"\nQuestion: {factual_question}")
    print(f"\nConsensus Answer:\n{result.final_output}")
    print(f"\nConfidence: {result.confidence:.2f}")
    print(f"Agreement Score: {result.agreement_score:.2f}")
    
    # Test 3: Multiple choice with majority vote
    print("\n" + "=" * 80)
    print("TEST 3: Multiple Choice with Majority Vote")
    print("=" * 80)
    
    mc_question = """Which of the following is the capital of Australia?
A) Sydney
B) Melbourne
C) Canberra
D) Brisbane

Answer with only the letter (A, B, C, or D)."""
    
    result = ensemble.solve(
        mc_question,
        strategy=AggregationStrategy.MAJORITY_VOTE
    )
    
    print(f"\nQuestion: Capital of Australia (multiple choice)")
    print(f"\nFinal Answer: {result.final_output}")
    print(f"Agreement Score: {result.agreement_score:.2f}")
    
    print("\nIndividual Agent Votes:")
    for vote in result.individual_votes:
        print(f"  {vote.agent_id}: {vote.output}")
    
    # Test 4: Comparison of strategies
    print("\n" + "=" * 80)
    print("TEST 4: Strategy Comparison")
    print("=" * 80)
    
    comparison_question = "Is artificial intelligence dangerous? Answer Yes or No, then explain."
    
    print(f"\nQuestion: {comparison_question}\n")
    
    for strategy in [AggregationStrategy.MAJORITY_VOTE, 
                     AggregationStrategy.WEIGHTED_VOTE,
                     AggregationStrategy.CONSENSUS]:
        
        result = ensemble.solve(comparison_question, strategy=strategy)
        
        print(f"\nStrategy: {strategy.value}")
        print(f"Answer: {result.final_output[:150]}...")
        print(f"Confidence: {result.confidence:.2f}")
        print(f"Agreement: {result.agreement_score:.2f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
The Ensemble/Committee pattern demonstrates several key benefits:

1. **Robustness**: Multiple agents reduce the impact of individual errors
2. **Diversity**: Different temperatures create varied perspectives
3. **Confidence Scoring**: Agreement levels indicate reliability
4. **Flexibility**: Multiple aggregation strategies for different use cases

Aggregation Strategies:
- **Majority Vote**: Simple, works well for discrete choices
- **Weighted Vote**: Considers confidence, better for uncertain domains
- **Consensus**: Synthesizes responses, best for complex questions

Use Cases:
- High-stakes classification tasks
- Fact-checking and validation
- Reducing model bias
- Quality assurance through redundancy

The pattern is particularly effective when:
- Individual agent accuracy is moderate but independent
- Diversity in approaches improves results
- Confidence/uncertainty quantification is important
- Explainability of decisions is needed
""")


if __name__ == "__main__":
    demonstrate_ensemble()

    

Agentic Design Pattern: Ensemble Committee

Pattern 013: Ensemble Committee – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Ensemble Committee pattern implementation.

---

Pattern Description

Pattern 013: Ensemble/Committee

Description: The Ensemble/Committee pattern uses multiple agents working independently on the same task, then aggregates their outputs to reach a final decision. This pattern leverages diversity to improve robustness, reduce errors, and provide more reliable results.

Components: – Multiple independent agents (diverse models, prompts, or strategies) – Aggregation mechanism (voting, averaging, weighted combination) – Result synthesis and confidence scoring

Use Cases: – Classification tasks requiring high accuracy – Prediction problems where uncertainty quantification is important – Reducing bias through diverse perspectives – Quality assurance through redundancy

LangChain Implementation: Uses multiple ChatOpenAI instances with different configurations, implements various aggregation strategies (majority voting, weighted voting, consensus), and provides confidence metrics based on agreement levels.

Key Features: – Diverse agent configurations (temperature, models, prompts) – Multiple aggregation strategies – Confidence scoring based on agreement – Detailed breakdown of individual votes

---

File Statistics

• Total Lines: 483
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 013: Ensemble/Committee

Description:
    The Ensemble/Committee pattern uses multiple agents working independently on the same
    task, then aggregates their outputs to reach a final decision. This pattern leverages
    diversity to improve robustness, reduce errors, and provide more reliable results.

Components:
    - Multiple independent agents (diverse models, prompts, or strategies)
    - Aggregation mechanism (voting, averaging, weighted combination)
    - Result synthesis and confidence scoring

Use Cases:
    - Classification tasks requiring high accuracy
    - Prediction problems where uncertainty quantification is important
    - Reducing bias through diverse perspectives
    - Quality assurance through redundancy

LangChain Implementation:
    Uses multiple ChatOpenAI instances with different configurations, implements
    various aggregation strategies (majority voting, weighted voting, consensus),
    and provides confidence metrics based on agreement levels.

Key Features:
    - Diverse agent configurations (temperature, models, prompts)
    - Multiple aggregation strategies
    - Confidence scoring based on agreement
    - Detailed breakdown of individual votes
"""

Purpose: This module implements the Ensemble Committee pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.Counter
import dataclasses.dataclass
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – collections.Counter: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: AggregationStrategy

Lines 46-51

Inherits from: Enum

Purpose:

Aggregation strategies for combining agent outputs.

---

Class: AgentVote

Lines 55-60

Purpose:

Represents a single agent's vote/output.

---

Class: EnsembleResult

Lines 64-70

Purpose:

Result from ensemble aggregation.

---

Class: EnsembleAgent

Lines 73-348

Purpose:

Ensemble agent that combines outputs from multiple independent agents.

This implementation creates diverse agents through different configurations
(temperature, system prompts) and provides multiple aggregation strategies.

Methods:

__init__(self, num_agents, base_model, temperatures)

Lines 81-113

Purpose:

Initialize ensemble with multiple agents.

Args:
    num_agents: Number of agents in ensemble
    base_model: Base model for all agents
    temperatures: List of temperatures for diversity (default: [0.3, 0.7, 1.0])

Parameters: – num_agents: Num agents – base_model: Language model instance – temperatures: Sampling temperature (0-1)

_estimate_confidence(self, question, answer, reasoning)

Lines 115-147

Purpose:

Estimate confidence score for an individual answer.

Parameters: – question: Question to answer – answer: Answer – reasoning: Reasoning

Returns: float

query_agents(self, question, system_prompt)

Lines 149-206

Purpose:

Query all agents independently and collect their votes.

Args:
    question: Question to ask all agents
    system_prompt: Optional system prompt (default: general assistant)

Returns:
    List of agent votes with outputs and confidence scores

Parameters: – question: Question to answer – system_prompt: Prompt template or text

Returns: <ast.Subscript object at 0x000001A06D17AC50>

aggregate_majority_vote(self, votes)

Lines 208-226

Purpose:

Aggregate using simple majority voting.

Returns:
    (winning answer, agreement score)

Parameters: – votes: Votes

Returns: <ast.Subscript object at 0x000001A06D169190>

aggregate_weighted_vote(self, votes)

Lines 228-255

Purpose:

Aggregate using confidence-weighted voting.

Returns:
    (winning answer, weighted agreement score)

Parameters: – votes: Votes

Returns: <ast.Subscript object at 0x000001A06D6F7A90>

aggregate_consensus(self, votes, question)

Lines 257-304

Purpose:

Generate consensus answer by synthesizing all votes.

Returns:
    (consensus answer, confidence score)

Parameters: – votes: Votes – question: Question to answer

Returns: <ast.Subscript object at 0x000001A06D7038D0>

solve(self, question, strategy, system_prompt)

Lines 306-348

Purpose:

Solve a problem using the ensemble of agents.

Args:
    question: Question to solve
    strategy: Aggregation strategy to use
    system_prompt: Optional custom system prompt

Returns:
    EnsembleResult with final answer and metadata

Parameters: – question: Question to answer – strategy: Strategy – system_prompt: Prompt template or text

Returns: EnsembleResult

---

Functions

demonstrate_ensemble()

Lines 351-478

Purpose:

Demonstrate the Ensemble/Committee pattern with various examples.

---

Key Concepts Summary

Pattern Overview

The Ensemble Committee pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 012
• Pattern 014
• Other Multi-Agent Patterns (012-019)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Ensemble Committee pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 013_ensemble_committee.py
Total Lines: 483
Pattern Category: AI Agent Design Pattern #013

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 012: Debate/Discussion (Multi-Agent)

Description:
    Multiple agents engage in debate or discussion to reach better conclusions.
    Each agent represents different perspectives, argues for their position,
    and the system synthesizes the best answer from the debate. This reduces
    bias and improves reasoning quality through adversarial collaboration.

Key Concepts:
    - Multiple Perspectives: Each agent takes different viewpoint
    - Argument Generation: Agents present reasoned arguments
    - Counter-Arguments: Agents respond to each other's points
    - Synthesis: Final answer combines best insights from debate
    - Moderation: Optional moderator guides discussion

Benefits:
    - Reduced individual bias
    - More thorough exploration of problem space
    - Error correction through peer review
    - Higher quality conclusions

Use Cases:
    - Complex decision-making
    - Ethical dilemmas
    - Strategy formulation
    - Content review and validation

LangChain Implementation:
    Multiple LLM instances with different personas engaging in structured debate.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


class AgentRole(Enum):
    """Roles agents can take in debate."""
    PROPONENT = "proponent"
    OPPONENT = "opponent"
    SKEPTIC = "skeptic"
    PRAGMATIST = "pragmatist"
    MODERATOR = "moderator"


@dataclass
class DebateMessage:
    """Represents a message in the debate."""
    round: int
    agent_id: str
    role: AgentRole
    content: str
    responding_to: Optional[str] = None


@dataclass
class DebateTranscript:
    """Complete transcript of a debate."""
    question: str
    messages: List[DebateMessage] = field(default_factory=list)
    final_synthesis: str = ""
    
    def add_message(self, message: DebateMessage):
        """Add a message to the transcript."""
        self.messages.append(message)
    
    def get_round_messages(self, round_num: int) -> List[DebateMessage]:
        """Get all messages from a specific round."""
        return [msg for msg in self.messages if msg.round == round_num]


class DebateAgent:
    """Individual agent participating in debate."""
    
    def __init__(self, agent_id: str, role: AgentRole, 
                 model_name: str = "gpt-3.5-turbo"):
        """
        Initialize a debate agent.
        
        Args:
            agent_id: Unique identifier for the agent
            role: Role the agent plays in debate
            model_name: Name of the OpenAI model
        """
        self.agent_id = agent_id
        self.role = role
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,  # Some creativity for diverse perspectives
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        # Role-specific system prompts
        self.system_prompts = {
            AgentRole.PROPONENT: "You are advocating FOR the position. Present strong arguments supporting it.",
            AgentRole.OPPONENT: "You are arguing AGAINST the position. Present strong counterarguments.",
            AgentRole.SKEPTIC: "You are a critical skeptic. Question assumptions and identify weaknesses.",
            AgentRole.PRAGMATIST: "You focus on practical implications and real-world feasibility."
        }
    
    def generate_opening_statement(self, question: str) -> str:
        """Generate opening statement for the debate."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""{self.system_prompts.get(self.role, 'You are a thoughtful debater.')}
            
This is your opening statement. Present your main position clearly and concisely."""),
            ("human", "Question: {question}\n\nYour opening statement:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        return chain.invoke({"question": question})
    
    def respond_to_arguments(self, question: str, previous_messages: List[DebateMessage]) -> str:
        """Generate response to previous arguments."""
        # Build context from previous messages
        context_parts = []
        for msg in previous_messages:
            context_parts.append(f"{msg.agent_id} ({msg.role.value}): {msg.content}")
        
        context = "\n\n".join(context_parts)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""{self.system_prompts.get(self.role, 'You are a thoughtful debater.')}

Respond to the previous arguments. Address key points, present counter-arguments where appropriate,
and strengthen your position."""),
            ("human", """Question: {question}

Previous Arguments:
{context}

Your response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "question": question,
            "context": context
        })


class DebateSystem:
    """System for managing multi-agent debate."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", num_rounds: int = 2):
        """
        Initialize the debate system.
        
        Args:
            model_name: Name of the OpenAI model
            num_rounds: Number of debate rounds
        """
        self.model_name = model_name
        self.num_rounds = num_rounds
        self.moderator_llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def conduct_debate(self, question: str, 
                      agent_roles: Optional[List[AgentRole]] = None,
                      verbose: bool = True) -> DebateTranscript:
        """
        Conduct a structured debate on the question.
        
        Args:
            question: Question to debate
            agent_roles: Roles for agents (default: proponent and opponent)
            verbose: Whether to print debate progress
            
        Returns:
            Complete debate transcript with synthesis
        """
        if agent_roles is None:
            agent_roles = [AgentRole.PROPONENT, AgentRole.OPPONENT]
        
        transcript = DebateTranscript(question=question)
        
        if verbose:
            print(f"\nQuestion: {question}\n")
            print(f"{'='*60}")
            print(f"MULTI-AGENT DEBATE")
            print(f"{'='*60}")
            print(f"Participants: {len(agent_roles)}")
            print(f"Rounds: {self.num_rounds}\n")
        
        # Create agents
        agents = []
        for i, role in enumerate(agent_roles):
            agent = DebateAgent(
                agent_id=f"Agent_{i+1}",
                role=role,
                model_name=self.model_name
            )
            agents.append(agent)
            if verbose:
                print(f"  {agent.agent_id}: {role.value}")
        
        print()
        
        # Round 1: Opening statements
        if verbose:
            print(f"\n{'='*60}")
            print(f"ROUND 1: Opening Statements")
            print('='*60)
        
        for agent in agents:
            statement = agent.generate_opening_statement(question)
            
            message = DebateMessage(
                round=1,
                agent_id=agent.agent_id,
                role=agent.role,
                content=statement
            )
            transcript.add_message(message)
            
            if verbose:
                print(f"\n{agent.agent_id} ({agent.role.value}):")
                print(f"{statement}")
        
        # Subsequent rounds: Responses and rebuttals
        for round_num in range(2, self.num_rounds + 1):
            if verbose:
                print(f"\n{'='*60}")
                print(f"ROUND {round_num}: Discussion")
                print('='*60)
            
            # Get previous round messages for context
            prev_messages = transcript.get_round_messages(round_num - 1)
            
            for agent in agents:
                # Agent responds to previous arguments
                response = agent.respond_to_arguments(question, prev_messages)
                
                message = DebateMessage(
                    round=round_num,
                    agent_id=agent.agent_id,
                    role=agent.role,
                    content=response
                )
                transcript.add_message(message)
                
                if verbose:
                    print(f"\n{agent.agent_id} ({agent.role.value}):")
                    print(f"{response}")
        
        # Synthesis: Combine insights from debate
        if verbose:
            print(f"\n{'='*60}")
            print("SYNTHESIS")
            print('='*60)
        
        synthesis = self._synthesize_debate(question, transcript)
        transcript.final_synthesis = synthesis
        
        if verbose:
            print(f"\n{synthesis}")
        
        return transcript
    
    def _synthesize_debate(self, question: str, transcript: DebateTranscript) -> str:
        """
        Synthesize the debate into a final answer.
        
        Args:
            question: Original question
            transcript: Complete debate transcript
            
        Returns:
            Synthesized answer
        """
        # Build debate summary
        debate_text = []
        for msg in transcript.messages:
            debate_text.append(f"[Round {msg.round}] {msg.agent_id} ({msg.role.value}):\n{msg.content}")
        
        debate_summary = "\n\n".join(debate_text)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a moderator synthesizing a debate. 
            
Review all arguments presented and create a balanced, thoughtful conclusion that:
1. Acknowledges strong points from all sides
2. Identifies areas of consensus
3. Notes remaining disagreements
4. Provides a nuanced final answer

Be objective and fair to all perspectives."""),
            ("human", """Question: {question}

Debate Transcript:
{debate}

Synthesis:""")
        ])
        
        chain = prompt | self.moderator_llm | StrOutputParser()
        synthesis = chain.invoke({
            "question": question,
            "debate": debate_summary
        })
        
        return synthesis


def demonstrate_debate_pattern():
    """Demonstrates the Debate/Discussion pattern."""
    
    print("=" * 80)
    print("PATTERN 012: Debate/Discussion (Multi-Agent)")
    print("=" * 80)
    print()
    print("Multi-Agent Debate improves reasoning through:")
    print("1. Multiple Perspectives: Different viewpoints represented")
    print("2. Adversarial Collaboration: Agents challenge each other")
    print("3. Iterative Refinement: Arguments evolve through rounds")
    print("4. Synthesis: Best insights combined into final answer")
    print()
    
    # Create debate system
    system = DebateSystem(num_rounds=2)
    
    # Test questions
    questions = [
        "Should artificial intelligence development be regulated by governments?",
        "Is remote work more productive than office work for software development?"
    ]
    
    for idx, question in enumerate(questions, 1):
        print(f"\n{'='*80}")
        print(f"Debate {idx}")
        print('='*80)
        
        try:
            # Conduct debate with different roles
            roles = [AgentRole.PROPONENT, AgentRole.OPPONENT, AgentRole.SKEPTIC]
            
            transcript = system.conduct_debate(
                question=question,
                agent_roles=roles,
                verbose=True
            )
            
            print(f"\n\n{'='*80}")
            print("DEBATE SUMMARY")
            print('='*80)
            print(f"\nQuestion: {transcript.question}")
            print(f"Total Messages: {len(transcript.messages)}")
            print(f"Rounds: {system.num_rounds}")
            print(f"\nFinal Synthesis:\n{transcript.final_synthesis}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("DEBATE/DISCUSSION PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Multiple Agent Roles: Proponent, Opponent, Skeptic")
    print("2. Structured Rounds: Opening statements followed by discussion")
    print("3. Contextual Responses: Agents respond to previous arguments")
    print("4. Moderator Synthesis: Balanced conclusion from all perspectives")
    print("5. Transcript Management: Complete debate history maintained")
    print()
    print("Advantages:")
    print("- Reduces individual bias")
    print("- More thorough problem exploration")
    print("- Error correction through peer review")
    print("- Higher quality conclusions")
    print()
    print("When to use Debate Pattern:")
    print("- Complex decision-making")
    print("- Ethical or controversial questions")
    print("- Strategy formulation")
    print("- Content validation and review")
    print()
    print("LangChain Components Used:")
    print("- Multiple ChatOpenAI instances: Different agents")
    print("- Role-based system prompts: Agent personas")
    print("- Structured debate management: Rounds and synthesis")
    print("- Context building: Previous arguments inform responses")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_debate_pattern()

    

Agentic Design Pattern: Debate Discussion

Pattern 012: Debate Discussion – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Debate Discussion pattern implementation.

---

Pattern Description

Pattern 012: Debate/Discussion (Multi-Agent)

Description: Multiple agents engage in debate or discussion to reach better conclusions. Each agent represents different perspectives, argues for their position, and the system synthesizes the best answer from the debate. This reduces bias and improves reasoning quality through adversarial collaboration.

Key Concepts: – Multiple Perspectives: Each agent takes different viewpoint – Argument Generation: Agents present reasoned arguments – Counter-Arguments: Agents respond to each other’s points – Synthesis: Final answer combines best insights from debate – Moderation: Optional moderator guides discussion

Benefits: – Reduced individual bias – More thorough exploration of problem space – Error correction through peer review – Higher quality conclusions

Use Cases: – Complex decision-making – Ethical dilemmas – Strategy formulation – Content review and validation

LangChain Implementation: Multiple LLM instances with different personas engaging in structured debate.

---

File Statistics

• Total Lines: 409
• Classes: 5
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 012: Debate/Discussion (Multi-Agent)

Description:
    Multiple agents engage in debate or discussion to reach better conclusions.
    Each agent represents different perspectives, argues for their position,
    and the system synthesizes the best answer from the debate. This reduces
    bias and improves reasoning quality through adversarial collaboration.

Key Concepts:
    - Multiple Perspectives: Each agent takes different viewpoint
    - Argument Generation: Agents present reasoned arguments
    - Counter-Arguments: Agents respond to each other's points
    - Synthesis: Final answer combines best insights from debate
    - Moderation: Optional moderator guides discussion

Benefits:
    - Reduced individual bias
    - More thorough exploration of problem space
    - Error correction through peer review
    - Higher quality conclusions

Use Cases:
    - Complex decision-making
    - Ethical dilemmas
    - Strategy formulation
    - Content review and validation

LangChain Implementation:
    Multiple LLM instances with different personas engaging in structured debate.
"""

Purpose: This module implements the Debate Discussion pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: AgentRole

Lines 47-53

Inherits from: Enum

Purpose:

Roles agents can take in debate.

---

Class: DebateMessage

Lines 57-63

Purpose:

Represents a message in the debate.

---

Class: DebateTranscript

Lines 67-79

Purpose:

Complete transcript of a debate.

Methods:

add_message(self, message)

Lines 73-75

Purpose:

Add a message to the transcript.

Parameters: – message: Message

get_round_messages(self, round_num)

Lines 77-79

Purpose:

Get all messages from a specific round.

Parameters: – round_num: Round num

Returns: <ast.Subscript object at 0x000001A06D60F4D0>

---

Class: DebateAgent

Lines 82-149

Purpose:

Individual agent participating in debate.

Methods:

__init__(self, agent_id, role, model_name)

Lines 85-109

Purpose:

Initialize a debate agent.

Args:
    agent_id: Unique identifier for the agent
    role: Role the agent plays in debate
    model_name: Name of the OpenAI model

Parameters: – agent_id: Agent id – role: Role – model_name: Language model instance

generate_opening_statement(self, question)

Lines 111-121

Purpose:

Generate opening statement for the debate.

Parameters: – question: Question to answer

Returns: str

respond_to_arguments(self, question, previous_messages)

Lines 123-149

Purpose:

Generate response to previous arguments.

Parameters: – question: Question to answer – previous_messages: Previous messages

Returns: str

---

Class: DebateSystem

Lines 152-315

Purpose:

System for managing multi-agent debate.

Methods:

__init__(self, model_name, num_rounds)

Lines 155-169

Purpose:

Initialize the debate system.

Args:
    model_name: Name of the OpenAI model
    num_rounds: Number of debate rounds

Parameters: – model_name: Language model instance – num_rounds: Num rounds

conduct_debate(self, question, agent_roles, verbose)

Lines 171-271

Purpose:

Conduct a structured debate on the question.

Args:
    question: Question to debate
    agent_roles: Roles for agents (default: proponent and opponent)
    verbose: Whether to print debate progress

Returns:
    Complete debate transcript with synthesis

Parameters: – question: Question to answer – agent_roles: Agent roles – verbose: Enable detailed logging

Returns: DebateTranscript

_synthesize_debate(self, question, transcript)

Lines 273-315

Purpose:

Synthesize the debate into a final answer.

Args:
    question: Original question
    transcript: Complete debate transcript

Returns:
    Synthesized answer

Parameters: – question: Question to answer – transcript: Transcript

Returns: str

---

Functions

demonstrate_debate_pattern()

Lines 318-398

Purpose:

Demonstrates the Debate/Discussion pattern.

---

Key Concepts Summary

Pattern Overview

The Debate Discussion pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 011
• Pattern 013
• Other Multi-Agent Patterns (012-019)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Debate Discussion pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 012_debate_discussion.py
Total Lines: 409
Pattern Category: AI Agent Design Pattern #012

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 011: Metacognitive Monitoring

Description:
    Agent monitors its own thinking process, confidence levels, and decision-making.
    Provides self-awareness about reasoning quality, uncertainty, and when to seek
    help or alternative strategies. Essential for safe and reliable AI systems.

Key Concepts:
    - Confidence Estimation: Quantify certainty in outputs
    - Uncertainty Quantification: Identify areas of low confidence
    - Self-Monitoring: Track reasoning quality during execution
    - Strategy Selection: Choose appropriate approaches based on confidence
    - Error Detection: Identify potential mistakes proactively

Use Cases:
    - Safety-critical applications
    - Decision validation and verification
    - Knowing when to escalate to humans
    - Quality control and error detection

LangChain Implementation:
    Confidence scoring chains with self-monitoring and adaptive strategies.
"""

import os
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


class ConfidenceLevel(Enum):
    """Confidence levels for reasoning."""
    VERY_LOW = "very_low"
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    VERY_HIGH = "very_high"


@dataclass
class MetacognitiveState:
    """Represents the agent's metacognitive state."""
    task: str
    current_step: str
    confidence: float = 0.5  # 0.0 to 1.0
    uncertainty_sources: List[str] = field(default_factory=list)
    alternative_strategies: List[str] = field(default_factory=list)
    should_seek_help: bool = False
    reasoning_quality: float = 0.5
    
    def get_confidence_level(self) -> ConfidenceLevel:
        """Convert numeric confidence to level."""
        if self.confidence >= 0.9:
            return ConfidenceLevel.VERY_HIGH
        elif self.confidence >= 0.7:
            return ConfidenceLevel.HIGH
        elif self.confidence >= 0.5:
            return ConfidenceLevel.MEDIUM
        elif self.confidence >= 0.3:
            return ConfidenceLevel.LOW
        else:
            return ConfidenceLevel.VERY_LOW


class MetacognitiveAgent:
    """Agent with metacognitive monitoring capabilities."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo",
                 confidence_threshold: float = 0.6,
                 help_threshold: float = 0.4):
        """
        Initialize the Metacognitive Agent.
        
        Args:
            model_name: Name of the OpenAI model
            confidence_threshold: Minimum acceptable confidence
            help_threshold: Threshold below which to seek help
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.confidence_threshold = confidence_threshold
        self.help_threshold = help_threshold
        self.metacognitive_log: List[MetacognitiveState] = []
    
    def estimate_confidence(self, task: str, reasoning: str, answer: str) -> Tuple[float, List[str]]:
        """
        Estimate confidence in the answer and identify uncertainty sources.
        
        Args:
            task: The task being solved
            reasoning: The reasoning process
            answer: The proposed answer
            
        Returns:
            Tuple of (confidence_score, uncertainty_sources)
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate your confidence in this answer. Consider:
1. Completeness of reasoning
2. Availability of relevant information
3. Complexity of the task
4. Potential ambiguities or uncertainties

Provide:
CONFIDENCE: [0.0-1.0]
UNCERTAINTY_SOURCES:
- [source 1]
- [source 2]
...

Be honest about limitations and uncertainties."""),
            ("human", """Task: {task}

Reasoning:
{reasoning}

Answer: {answer}

Confidence Evaluation:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "task": task,
            "reasoning": reasoning,
            "answer": answer
        })
        
        # Parse confidence and uncertainty sources
        confidence = 0.5
        uncertainty_sources = []
        
        for line in response.split('\n'):
            line = line.strip()
            if line.upper().startswith('CONFIDENCE:'):
                try:
                    conf_str = line.split(':', 1)[1].strip()
                    confidence = float(conf_str)
                except:
                    pass
            elif line.startswith('-') and 'UNCERTAINTY_SOURCES' in response.upper():
                source = line.lstrip('- ').strip()
                if source:
                    uncertainty_sources.append(source)
        
        return confidence, uncertainty_sources
    
    def identify_alternative_strategies(self, task: str, current_approach: str,
                                       issues: List[str]) -> List[str]:
        """
        Identify alternative strategies when current approach has low confidence.
        
        Args:
            task: The task being solved
            current_approach: Current approach being used
            issues: Identified issues with current approach
            
        Returns:
            List of alternative strategies
        """
        issues_text = "\n".join([f"- {issue}" for issue in issues])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Suggest alternative strategies for solving this task given the identified issues.
Provide 2-3 concrete alternative approaches.

Format:
1. [Strategy 1]
2. [Strategy 2]
3. [Strategy 3]"""),
            ("human", """Task: {task}

Current Approach: {current_approach}

Issues:
{issues}

Alternative Strategies:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "task": task,
            "current_approach": current_approach,
            "issues": issues_text
        })
        
        # Parse strategies
        strategies = []
        for line in response.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                strategy = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                if strategy:
                    strategies.append(strategy)
        
        return strategies[:3]
    
    def evaluate_reasoning_quality(self, reasoning: str) -> float:
        """
        Evaluate the quality of the reasoning process.
        
        Args:
            reasoning: The reasoning to evaluate
            
        Returns:
            Quality score 0.0-1.0
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate the quality of this reasoning. Consider:
- Logical consistency
- Completeness
- Clarity
- Rigor

Provide a quality score from 0.0 (poor) to 1.0 (excellent).
Respond with just the number."""),
            ("human", """Reasoning:
{reasoning}

Quality Score (0.0-1.0):""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            response = chain.invoke({"reasoning": reasoning})
            score = float(response.strip())
            return max(0.0, min(1.0, score))
        except:
            return 0.5
    
    def solve_with_monitoring(self, task: str, verbose: bool = True) -> Dict[str, Any]:
        """
        Solve a task with metacognitive monitoring.
        
        Args:
            task: The task to solve
            verbose: Whether to print monitoring information
            
        Returns:
            Dictionary with solution and metacognitive information
        """
        if verbose:
            print(f"\nTask: {task}\n")
            print("="*60)
            print("METACOGNITIVE MONITORING ACTIVE")
            print("="*60)
        
        # Step 1: Initial reasoning
        if verbose:
            print("\nStep 1: Initial Reasoning")
        
        reasoning_prompt = ChatPromptTemplate.from_messages([
            ("system", "Solve this task with clear step-by-step reasoning."),
            ("human", "{task}")
        ])
        
        chain = reasoning_prompt | self.llm | StrOutputParser()
        reasoning = chain.invoke({"task": task})
        
        # Extract answer
        answer_lines = reasoning.strip().split('\n')
        answer = answer_lines[-1] if answer_lines else reasoning[:100]
        
        if verbose:
            print(f"Reasoning: {reasoning[:200]}...")
            print(f"Answer: {answer}")
        
        # Step 2: Estimate confidence
        if verbose:
            print("\nStep 2: Confidence Estimation")
        
        confidence, uncertainty_sources = self.estimate_confidence(task, reasoning, answer)
        
        # Step 3: Evaluate reasoning quality
        reasoning_quality = self.evaluate_reasoning_quality(reasoning)
        
        if verbose:
            print(f"Confidence: {confidence:.2f} ({ConfidenceLevel(confidence >= 0.9 and 'very_high' or confidence >= 0.7 and 'high' or confidence >= 0.5 and 'medium' or confidence >= 0.3 and 'low' or 'very_low').value})")
            print(f"Reasoning Quality: {reasoning_quality:.2f}")
            if uncertainty_sources:
                print(f"Uncertainty Sources: {len(uncertainty_sources)}")
                for source in uncertainty_sources:
                    print(f"  - {source}")
        
        # Step 4: Determine if help needed
        should_seek_help = confidence < self.help_threshold
        
        # Step 5: Identify alternative strategies if confidence low
        alternative_strategies = []
        if confidence < self.confidence_threshold:
            if verbose:
                print(f"\n⚠️ Low confidence detected (threshold: {self.confidence_threshold})")
                print("Step 3: Identifying Alternative Strategies")
            
            alternative_strategies = self.identify_alternative_strategies(
                task, "Initial direct reasoning", uncertainty_sources
            )
            
            if verbose:
                print("Alternative Strategies:")
                for i, strategy in enumerate(alternative_strategies, 1):
                    print(f"  {i}. {strategy}")
        
        # Step 6: Decide on action
        if should_seek_help:
            action = "ESCALATE_TO_HUMAN"
            recommendation = "Task complexity or uncertainty requires human oversight"
        elif confidence < self.confidence_threshold:
            action = "TRY_ALTERNATIVE_STRATEGY"
            recommendation = f"Recommend trying: {alternative_strategies[0] if alternative_strategies else 'alternative approach'}"
        else:
            action = "PROCEED_WITH_ANSWER"
            recommendation = "Confidence sufficient to proceed"
        
        if verbose:
            print(f"\nStep 4: Decision")
            print(f"Action: {action}")
            print(f"Recommendation: {recommendation}")
        
        # Create metacognitive state
        state = MetacognitiveState(
            task=task,
            current_step="Solution",
            confidence=confidence,
            uncertainty_sources=uncertainty_sources,
            alternative_strategies=alternative_strategies,
            should_seek_help=should_seek_help,
            reasoning_quality=reasoning_quality
        )
        self.metacognitive_log.append(state)
        
        return {
            "task": task,
            "reasoning": reasoning,
            "answer": answer,
            "confidence": confidence,
            "confidence_level": state.get_confidence_level().value,
            "reasoning_quality": reasoning_quality,
            "uncertainty_sources": uncertainty_sources,
            "alternative_strategies": alternative_strategies,
            "should_seek_help": should_seek_help,
            "recommended_action": action,
            "recommendation": recommendation
        }


def demonstrate_metacognitive_monitoring():
    """Demonstrates the Metacognitive Monitoring pattern."""
    
    print("=" * 80)
    print("PATTERN 011: Metacognitive Monitoring")
    print("=" * 80)
    print()
    print("Metacognitive Monitoring enables self-awareness through:")
    print("1. Confidence Estimation: Quantify certainty in outputs")
    print("2. Uncertainty Detection: Identify sources of uncertainty")
    print("3. Quality Assessment: Evaluate reasoning quality")
    print("4. Strategy Selection: Choose alternative approaches when needed")
    print("5. Help-Seeking: Know when to escalate to humans")
    print()
    
    # Create agent
    agent = MetacognitiveAgent(
        confidence_threshold=0.6,
        help_threshold=0.4
    )
    
    # Test tasks with varying complexity
    tasks = [
        "What is 15 + 27?",  # High confidence expected
        "Explain the implications of quantum entanglement for future communication technologies",  # Medium confidence
        "Should Company X acquire Company Y given uncertain market conditions?"  # Low confidence
    ]
    
    for idx, task in enumerate(tasks, 1):
        print(f"\n{'='*80}")
        print(f"Example {idx}")
        print('='*80)
        
        try:
            result = agent.solve_with_monitoring(task, verbose=True)
            
            print(f"\n\n{'='*80}")
            print("METACOGNITIVE SUMMARY")
            print('='*80)
            print(f"\nTask: {result['task']}")
            print(f"Answer: {result['answer']}")
            print(f"\nConfidence: {result['confidence']:.2f} ({result['confidence_level']})")
            print(f"Reasoning Quality: {result['reasoning_quality']:.2f}")
            print(f"Should Seek Help: {'Yes' if result['should_seek_help'] else 'No'}")
            print(f"\nRecommended Action: {result['recommended_action']}")
            print(f"Recommendation: {result['recommendation']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("METACOGNITIVE MONITORING PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Confidence Estimation: Quantified certainty levels")
    print("2. Uncertainty Sources: Identified specific areas of uncertainty")
    print("3. Quality Assessment: Evaluated reasoning quality")
    print("4. Alternative Strategies: Suggested alternatives for low confidence")
    print("5. Decision Making: Determined appropriate actions based on confidence")
    print()
    print("Advantages:")
    print("- Self-awareness of limitations")
    print("- Proactive error detection")
    print("- Intelligent help-seeking")
    print("- Adaptive strategy selection")
    print()
    print("When to use Metacognitive Monitoring:")
    print("- Safety-critical applications")
    print("- High-stakes decision-making")
    print("- Uncertain or ambiguous tasks")
    print("- When transparency is required")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Confidence estimation and evaluation")
    print("- StrOutputParser: Parse monitoring outputs")
    print("- Threshold-based decision making")
    print("- Structured metacognitive state tracking")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_metacognitive_monitoring()

    

Agentic Design Pattern: Metacognitive Monitoring

Pattern 011: Metacognitive Monitoring – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Metacognitive Monitoring pattern implementation.

---

Pattern Description

Pattern 011: Metacognitive Monitoring

Description: Agent monitors its own thinking process, confidence levels, and decision-making. Provides self-awareness about reasoning quality, uncertainty, and when to seek help or alternative strategies. Essential for safe and reliable AI systems.

Key Concepts: – Confidence Estimation: Quantify certainty in outputs – Uncertainty Quantification: Identify areas of low confidence – Self-Monitoring: Track reasoning quality during execution – Strategy Selection: Choose appropriate approaches based on confidence – Error Detection: Identify potential mistakes proactively

Use Cases: – Safety-critical applications – Decision validation and verification – Knowing when to escalate to humans – Quality control and error detection

LangChain Implementation: Confidence scoring chains with self-monitoring and adaptive strategies.

---

File Statistics

• Total Lines: 454
• Classes: 3
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 011: Metacognitive Monitoring

Description:
    Agent monitors its own thinking process, confidence levels, and decision-making.
    Provides self-awareness about reasoning quality, uncertainty, and when to seek
    help or alternative strategies. Essential for safe and reliable AI systems.

Key Concepts:
    - Confidence Estimation: Quantify certainty in outputs
    - Uncertainty Quantification: Identify areas of low confidence
    - Self-Monitoring: Track reasoning quality during execution
    - Strategy Selection: Choose appropriate approaches based on confidence
    - Error Detection: Identify potential mistakes proactively

Use Cases:
    - Safety-critical applications
    - Decision validation and verification
    - Knowing when to escalate to humans
    - Quality control and error detection

LangChain Implementation:
    Confidence scoring chains with self-monitoring and adaptive strategies.
"""

Purpose: This module implements the Metacognitive Monitoring pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ConfidenceLevel

Lines 40-46

Inherits from: Enum

Purpose:

Confidence levels for reasoning.

---

Class: MetacognitiveState

Lines 50-71

Purpose:

Represents the agent's metacognitive state.

Methods:

get_confidence_level(self)

Lines 60-71

Purpose:

Convert numeric confidence to level.

Parameters:

Returns: ConfidenceLevel

---

Class: MetacognitiveAgent

Lines 74-359

Purpose:

Agent with metacognitive monitoring capabilities.

Methods:

__init__(self, model_name, confidence_threshold, help_threshold)

Lines 77-96

Purpose:

Initialize the Metacognitive Agent.

Args:
    model_name: Name of the OpenAI model
    confidence_threshold: Minimum acceptable confidence
    help_threshold: Threshold below which to seek help

Parameters: – model_name: Language model instance – confidence_threshold: Confidence threshold – help_threshold: Help threshold

estimate_confidence(self, task, reasoning, answer)

Lines 98-159

Purpose:

Estimate confidence in the answer and identify uncertainty sources.

Args:
    task: The task being solved
    reasoning: The reasoning process
    answer: The proposed answer

Returns:
    Tuple of (confidence_score, uncertainty_sources)

Parameters: – task: Task – reasoning: Reasoning – answer: Answer

Returns: <ast.Subscript object at 0x000001A06D1592D0>

identify_alternative_strategies(self, task, current_approach, issues)

Lines 161-210

Purpose:

Identify alternative strategies when current approach has low confidence.

Args:
    task: The task being solved
    current_approach: Current approach being used
    issues: Identified issues with current approach

Returns:
    List of alternative strategies

Parameters: – task: Task – current_approach: Current approach – issues: Issues

Returns: <ast.Subscript object at 0x000001A06D700F50>

evaluate_reasoning_quality(self, reasoning)

Lines 212-244

Purpose:

Evaluate the quality of the reasoning process.

Args:
    reasoning: The reasoning to evaluate

Returns:
    Quality score 0.0-1.0

Parameters: – reasoning: Reasoning

Returns: float

solve_with_monitoring(self, task, verbose)

Lines 246-359

Purpose:

Solve a task with metacognitive monitoring.

Args:
    task: The task to solve
    verbose: Whether to print monitoring information

Returns:
    Dictionary with solution and metacognitive information

Parameters: – task: Task – verbose: Enable detailed logging

Returns: <ast.Subscript object at 0x000001A06D156C50>

---

Functions

demonstrate_metacognitive_monitoring()

Lines 362-443

Purpose:

Demonstrates the Metacognitive Monitoring pattern.

---

Key Concepts Summary

Pattern Overview

The Metacognitive Monitoring pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 010
• Pattern 012
• Other Reasoning Patterns (001-011)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Metacognitive Monitoring pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 011_metacognitive_monitoring.py
Total Lines: 454
Pattern Category: AI Agent Design Pattern #011

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 010: Constitutional AI

Description:
    Ensures agent behavior aligns with explicit principles or "constitution."
    The agent generates responses, evaluates them against constitutional principles,
    and revises if violations are found. This creates safer, more aligned AI systems.

Key Concepts:
    - Constitution: Set of principles/rules the agent must follow
    - Self-Critique: Agent evaluates its own outputs against principles
    - Revision: Generate improved response that adheres to constitution
    - Iterative Alignment: Multiple rounds of critique and revision
    - Transparency: Explicit principles make behavior interpretable

Constitutional Principles Examples:
    - Harmlessness: Avoid harmful, dangerous, or offensive content
    - Helpfulness: Provide useful, relevant information
    - Honesty: Be truthful, acknowledge uncertainty
    - Privacy: Respect privacy and confidentiality
    - Fairness: Avoid bias and discrimination

Use Cases:
    - Production AI systems requiring safety
    - Content moderation and filtering
    - Ethical decision-making systems
    - Regulated domains (healthcare, finance, legal)

LangChain Implementation:
    Constitutional chain with critique and revision loops based on principles.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


@dataclass
class Principle:
    """Represents a constitutional principle."""
    name: str
    description: str
    critique_prompt: str
    weight: float = 1.0  # Importance weight


@dataclass
class ViolationReport:
    """Report of constitutional violations."""
    principle: str
    violated: bool
    explanation: str
    severity: float = 0.0  # 0.0 (no violation) to 1.0 (severe)


@dataclass
class ConstitutionalReview:
    """Complete review against constitution."""
    original_output: str
    violations: List[ViolationReport] = field(default_factory=list)
    total_severity: float = 0.0
    passes: bool = True
    
    def calculate_severity(self):
        """Calculate total severity of violations."""
        self.total_severity = sum(v.severity for v in self.violations if v.violated)
        self.passes = self.total_severity == 0.0


class Constitution:
    """Set of principles for constitutional AI."""
    
    def __init__(self, name: str = "Default Constitution"):
        """Initialize constitution with default or custom principles."""
        self.name = name
        self.principles: List[Principle] = []
    
    def add_principle(self, principle: Principle):
        """Add a principle to the constitution."""
        self.principles.append(principle)
    
    def get_principles_text(self) -> str:
        """Get formatted text of all principles."""
        lines = [f"Constitution: {self.name}\n"]
        for i, p in enumerate(self.principles, 1):
            lines.append(f"{i}. {p.name}: {p.description}")
        return "\n".join(lines)
    
    @staticmethod
    def get_default() -> 'Constitution':
        """Get default constitution with common principles."""
        constitution = Constitution("Default AI Constitution")
        
        # Harmlessness
        constitution.add_principle(Principle(
            name="Harmlessness",
            description="Responses must not include harmful, dangerous, illegal, or offensive content",
            critique_prompt="""Does the response contain any harmful, dangerous, illegal, or offensive content?
Consider: violence, illegal activities, self-harm, harassment, discrimination.
Answer YES if violations exist, NO if the response is harmless."""
        ))
        
        # Helpfulness
        constitution.add_principle(Principle(
            name="Helpfulness",
            description="Responses should be helpful, relevant, and answer the user's question",
            critique_prompt="""Is the response helpful and relevant to the user's query?
Does it provide useful information or assistance?
Answer YES if unhelpful, NO if the response is helpful."""
        ))
        
        # Honesty
        constitution.add_principle(Principle(
            name="Honesty",
            description="Responses should be truthful and acknowledge uncertainty when appropriate",
            critique_prompt="""Does the response contain false information or make claims without acknowledging uncertainty?
Answer YES if dishonest or overconfident, NO if honest and appropriately uncertain."""
        ))
        
        # Privacy
        constitution.add_principle(Principle(
            name="Privacy",
            description="Responses must respect privacy and not request or share personal information inappropriately",
            critique_prompt="""Does the response inappropriately request, share, or handle personal information?
Answer YES if privacy is violated, NO if privacy is respected."""
        ))
        
        # Fairness
        constitution.add_principle(Principle(
            name="Fairness",
            description="Responses should be fair and avoid bias or discrimination",
            critique_prompt="""Does the response show bias or discrimination based on race, gender, religion, or other protected characteristics?
Answer YES if biased or discriminatory, NO if fair and balanced."""
        ))
        
        return constitution


class ConstitutionalAI:
    """Agent that follows constitutional principles."""
    
    def __init__(self, constitution: Constitution,
                 model_name: str = "gpt-3.5-turbo",
                 max_revisions: int = 3,
                 severity_threshold: float = 0.3):
        """
        Initialize Constitutional AI agent.
        
        Args:
            constitution: The constitution to follow
            model_name: Name of the OpenAI model
            max_revisions: Maximum number of revision attempts
            severity_threshold: Severity threshold for triggering revision
        """
        self.constitution = constitution
        self.max_revisions = max_revisions
        self.severity_threshold = severity_threshold
        
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.critic_llm = ChatOpenAI(
            model=model_name,
            temperature=0.2,  # Lower temperature for consistent critique
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def generate_initial(self, query: str, context: str = "") -> str:
        """
        Generate initial response to query.
        
        Args:
            query: User query
            context: Additional context
            
        Returns:
            Initial response
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant. Respond to the user's query."),
            ("human", """{context}

Query: {query}

Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "query": query,
            "context": context if context else ""
        })
        
        return response.strip()
    
    def critique_against_principle(self, output: str, query: str,
                                   principle: Principle) -> ViolationReport:
        """
        Critique output against a single principle.
        
        Args:
            output: Output to critique
            query: Original query
            principle: Principle to check against
            
        Returns:
            ViolationReport indicating if principle is violated
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are evaluating whether a response violates the principle: "{principle.name}"

Principle: {principle.description}

{principle.critique_prompt}

If YES (violation), also rate severity from 0.1 (minor) to 1.0 (severe).

Format your response as:
VIOLATION: YES/NO
SEVERITY: [0.0-1.0]
EXPLANATION: [Brief explanation]"""),
            ("human", """Query: {query}

Response to evaluate:
{output}

Evaluation:""")
        ])
        
        chain = prompt | self.critic_llm | StrOutputParser()
        critique_text = chain.invoke({
            "query": query,
            "output": output
        })
        
        # Parse critique
        violated = False
        severity = 0.0
        explanation = ""
        
        for line in critique_text.split('\n'):
            line = line.strip()
            if line.upper().startswith('VIOLATION:'):
                violated = 'YES' in line.upper()
            elif line.upper().startswith('SEVERITY:'):
                try:
                    severity_str = line.split(':', 1)[1].strip()
                    severity = float(severity_str)
                except:
                    severity = 0.5 if violated else 0.0
            elif line.upper().startswith('EXPLANATION:'):
                explanation = line.split(':', 1)[1].strip()
        
        return ViolationReport(
            principle=principle.name,
            violated=violated,
            explanation=explanation,
            severity=severity if violated else 0.0
        )
    
    def review(self, output: str, query: str) -> ConstitutionalReview:
        """
        Review output against all constitutional principles.
        
        Args:
            output: Output to review
            query: Original query
            
        Returns:
            ConstitutionalReview with all violations
        """
        review = ConstitutionalReview(original_output=output)
        
        for principle in self.constitution.principles:
            violation = self.critique_against_principle(output, query, principle)
            review.violations.append(violation)
        
        review.calculate_severity()
        
        return review
    
    def revise(self, query: str, output: str, review: ConstitutionalReview) -> str:
        """
        Revise output to address constitutional violations.
        
        Args:
            query: Original query
            output: Current output with violations
            review: Review identifying violations
            
        Returns:
            Revised output addressing violations
        """
        # Build violation summary
        violations_text = []
        for v in review.violations:
            if v.violated:
                violations_text.append(f"- {v.principle}: {v.explanation} (severity: {v.severity:.2f})")
        
        violations_summary = "\n".join(violations_text) if violations_text else "No violations"
        
        # Get principles text
        principles_text = self.constitution.get_principles_text()
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are revising a response to address constitutional violations.

{principles_text}

Revise the response to:
1. Address all identified violations
2. Maintain helpfulness and relevance
3. Fully comply with all principles

Generate a new, improved response that adheres to the constitution."""),
            ("human", """Original Query: {query}

Current Response:
{output}

Violations Found:
{violations}

Revised Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        revised = chain.invoke({
            "query": query,
            "output": output,
            "violations": violations_summary
        })
        
        return revised.strip()
    
    def generate_with_constitution(self, query: str, context: str = "",
                                   verbose: bool = True) -> Dict[str, Any]:
        """
        Generate response following constitutional principles.
        
        Args:
            query: User query
            context: Additional context
            verbose: Whether to print progress
            
        Returns:
            Dictionary with final response and review history
        """
        if verbose:
            print(f"\nQuery: {query}\n")
            print(f"Constitution: {self.constitution.name}")
            print(f"Principles: {len(self.constitution.principles)}\n")
        
        # Generate initial response
        if verbose:
            print("="*60)
            print("INITIAL GENERATION")
            print("="*60)
        
        current_output = self.generate_initial(query, context)
        
        if verbose:
            print(f"\nGenerated:\n{current_output[:200]}..." if len(current_output) > 200 else f"\nGenerated:\n{current_output}")
        
        reviews = []
        revision_count = 0
        
        # Review and revise loop
        for iteration in range(self.max_revisions + 1):
            if verbose:
                print(f"\n{'='*60}")
                print(f"CONSTITUTIONAL REVIEW {iteration + 1}")
                print("="*60)
            
            # Review against constitution
            review = self.review(current_output, query)
            reviews.append(review)
            
            if verbose:
                print(f"\nViolations Found: {sum(1 for v in review.violations if v.violated)}/{len(review.violations)}")
                print(f"Total Severity: {review.total_severity:.2f}")
                
                for v in review.violations:
                    if v.violated:
                        print(f"  ⚠ {v.principle}: {v.explanation} (severity: {v.severity:.2f})")
                    else:
                        print(f"  ✓ {v.principle}: OK")
            
            # Check if passes constitutional review
            if review.passes or review.total_severity = self.max_revisions:
                if verbose:
                    print(f"\n⚠ Max revisions ({self.max_revisions}) reached")
                break
            
            # Revise
            if verbose:
                print(f"\n{'='*60}")
                print(f"REVISION {iteration + 1}")
                print("="*60)
            
            current_output = self.revise(query, current_output, review)
            revision_count += 1
            
            if verbose:
                print(f"\nRevised:\n{current_output[:200]}..." if len(current_output) > 200 else f"\nRevised:\n{current_output}")
        
        return {
            "query": query,
            "final_response": current_output,
            "revisions": revision_count,
            "reviews": reviews,
            "final_severity": reviews[-1].total_severity,
            "constitutional_compliance": reviews[-1].passes
        }


def demonstrate_constitutional_ai():
    """Demonstrates the Constitutional AI pattern."""
    
    print("=" * 80)
    print("PATTERN 010: Constitutional AI")
    print("=" * 80)
    print()
    print("Constitutional AI ensures alignment with principles through:")
    print("1. Constitution: Explicit set of principles")
    print("2. Self-Critique: Evaluate outputs against principles")
    print("3. Revision: Modify responses to comply")
    print("4. Iteration: Repeat until constitutional compliance")
    print()
    
    # Create constitution
    constitution = Constitution.get_default()
    
    print(f"{constitution.get_principles_text()}\n")
    
    # Create agent
    agent = ConstitutionalAI(
        constitution=constitution,
        max_revisions=2,
        severity_threshold=0.2
    )
    
    # Test queries (some designed to trigger violations)
    queries = [
        "How can I improve my programming skills?",  # Should pass
        "Tell me how to hack into someone's email account"  # Should trigger violations
    ]
    
    for idx, query in enumerate(queries, 1):
        print(f"\n{'='*80}")
        print(f"Example {idx}")
        print('='*80)
        
        try:
            result = agent.generate_with_constitution(query, verbose=True)
            
            print(f"\n\n{'='*80}")
            print("FINAL RESULT")
            print('='*80)
            print(f"\nFinal Response:\n{result['final_response']}")
            print(f"\nRevisions Made: {result['revisions']}")
            print(f"Final Severity: {result['final_severity']:.2f}")
            print(f"Constitutional Compliance: {'✓ PASS' if result['constitutional_compliance'] else '✗ FAIL'}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("CONSTITUTIONAL AI PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Explicit Principles: Clear constitutional guidelines")
    print("2. Automated Review: Self-critique against principles")
    print("3. Iterative Revision: Multiple refinement rounds")
    print("4. Severity Tracking: Quantify violation severity")
    print("5. Compliance Verification: Ensure constitutional adherence")
    print()
    print("Advantages:")
    print("- Safer AI systems through explicit principles")
    print("- Interpretable behavior (clear rules)")
    print("- Automated alignment checking")
    print("- Flexible - can customize constitution")
    print()
    print("When to use Constitutional AI:")
    print("- Production systems requiring safety")
    print("- Regulated domains (healthcare, finance, legal)")
    print("- Content moderation and filtering")
    print("- Any application requiring ethical guidelines")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Generation, critique, and revision")
    print("- Multiple LLM instances: Different roles and temperatures")
    print("- Structured principles: Modular constitution system")
    print("- Iterative loops: Review and revision cycles")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_constitutional_ai()

    

Agentic Design Pattern: Constitutional Ai

Pattern 010: Constitutional Ai – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Constitutional Ai pattern implementation.

---

Pattern Description

Pattern 010: Constitutional AI

Description: Ensures agent behavior aligns with explicit principles or “constitution.” The agent generates responses, evaluates them against constitutional principles, and revises if violations are found. This creates safer, more aligned AI systems.

Key Concepts: – Constitution: Set of principles/rules the agent must follow – Self-Critique: Agent evaluates its own outputs against principles – Revision: Generate improved response that adheres to constitution – Iterative Alignment: Multiple rounds of critique and revision – Transparency: Explicit principles make behavior interpretable

Constitutional Principles Examples: – Harmlessness: Avoid harmful, dangerous, or offensive content – Helpfulness: Provide useful, relevant information – Honesty: Be truthful, acknowledge uncertainty – Privacy: Respect privacy and confidentiality – Fairness: Avoid bias and discrimination

Use Cases: – Production AI systems requiring safety – Content moderation and filtering – Ethical decision-making systems – Regulated domains (healthcare, finance, legal)

LangChain Implementation: Constitutional chain with critique and revision loops based on principles.

---

File Statistics

• Total Lines: 528
• Classes: 5
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 010: Constitutional AI

Description:
    Ensures agent behavior aligns with explicit principles or "constitution."
    The agent generates responses, evaluates them against constitutional principles,
    and revises if violations are found. This creates safer, more aligned AI systems.

Key Concepts:
    - Constitution: Set of principles/rules the agent must follow
    - Self-Critique: Agent evaluates its own outputs against principles
    - Revision: Generate improved response that adheres to constitution
    - Iterative Alignment: Multiple rounds of critique and revision
    - Transparency: Explicit principles make behavior interpretable

Constitutional Principles Examples:
    - Harmlessness: Avoid harmful, dangerous, or offensive content
    - Helpfulness: Provide useful, relevant information
    - Honesty: Be truthful, acknowledge uncertainty
    - Privacy: Respect privacy and confidentiality
    - Fairness: Avoid bias and discrimination

Use Cases:
    - Production AI systems requiring safety
    - Content moderation and filtering
    - Ethical decision-making systems
    - Regulated domains (healthcare, finance, legal)

LangChain Implementation:
    Constitutional chain with critique and revision loops based on principles.
"""

Purpose: This module implements the Constitutional Ai pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: Principle

Lines 48-53

Purpose:

Represents a constitutional principle.

---

Class: ViolationReport

Lines 57-62

Purpose:

Report of constitutional violations.

---

Class: ConstitutionalReview

Lines 66-76

Purpose:

Complete review against constitution.

Methods:

calculate_severity(self)

Lines 73-76

Purpose:

Calculate total severity of violations.

Parameters:

---

Class: Constitution

Lines 79-145

Purpose:

Set of principles for constitutional AI.

Methods:

__init__(self, name)

Lines 82-85

Purpose:

Initialize constitution with default or custom principles.

Parameters: – name: Name

add_principle(self, principle)

Lines 87-89

Purpose:

Add a principle to the constitution.

Parameters: – principle: Principle

get_principles_text(self)

Lines 91-96

Purpose:

Get formatted text of all principles.

Parameters:

Returns: str

get_default()

Lines 99-145

Decorators: staticmethod

Purpose:

Get default constitution with common principles.

Returns: Constitution

---

Class: ConstitutionalAI

Lines 148-432

Purpose:

Agent that follows constitutional principles.

Methods:

__init__(self, constitution, model_name, max_revisions, severity_threshold)

Lines 151-178

Purpose:

Initialize Constitutional AI agent.

Args:
    constitution: The constitution to follow
    model_name: Name of the OpenAI model
    max_revisions: Maximum number of revision attempts
    severity_threshold: Severity threshold for triggering revision

Parameters: – constitution: Constitution – model_name: Language model instance – max_revisions: Max revisions – severity_threshold: Severity threshold

generate_initial(self, query, context)

Lines 180-206

Purpose:

Generate initial response to query.

Args:
    query: User query
    context: Additional context

Returns:
    Initial response

Parameters: – query: Input query or question – context: Context

Returns: str

critique_against_principle(self, output, query, principle)

Lines 208-271

Purpose:

Critique output against a single principle.

Args:
    output: Output to critique
    query: Original query
    principle: Principle to check against

Returns:
    ViolationReport indicating if principle is violated

Parameters: – output: Output – query: Input query or question – principle: Principle

Returns: ViolationReport

review(self, output, query)

Lines 273-292

Purpose:

Review output against all constitutional principles.

Args:
    output: Output to review
    query: Original query

Returns:
    ConstitutionalReview with all violations

Parameters: – output: Output – query: Input query or question

Returns: ConstitutionalReview

revise(self, query, output, review)

Lines 294-346

Purpose:

Revise output to address constitutional violations.

Args:
    query: Original query
    output: Current output with violations
    review: Review identifying violations

Returns:
    Revised output addressing violations

Parameters: – query: Input query or question – output: Output – review: Review

Returns: str

generate_with_constitution(self, query, context, verbose)

Lines 348-432

Purpose:

Generate response following constitutional principles.

Args:
    query: User query
    context: Additional context
    verbose: Whether to print progress

Returns:
    Dictionary with final response and review history

Parameters: – query: Input query or question – context: Context – verbose: Enable detailed logging

Returns: <ast.Subscript object at 0x000001A06D607710>

---

Functions

demonstrate_constitutional_ai()

Lines 435-517

Purpose:

Demonstrates the Constitutional AI pattern.

---

Key Concepts Summary

Pattern Overview

The Constitutional Ai pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 009
• Pattern 011
• Other Reasoning Patterns (001-011)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Constitutional Ai pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 010_constitutional_ai.py
Total Lines: 528
Pattern Category: AI Agent Design Pattern #010

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 009: Self-Refine

Description:
    Iteratively improves outputs through self-critique and refinement.
    The agent generates an initial output, evaluates it, identifies weaknesses,
    and produces an improved version. This cycle repeats until quality threshold
    is met or max iterations reached.

Key Concepts:
    - Initial Generation: Create first draft output
    - Self-Critique: Analyze weaknesses and areas for improvement
    - Refinement: Generate improved version based on critique
    - Iteration: Repeat until satisfactory or max iterations
    - Quality Tracking: Monitor improvement across iterations

Benefits:
    - Higher quality outputs through iteration
    - Self-improvement without external feedback
    - Transparency in refinement process
    - Handles complex generation tasks

Use Cases:
    - Content creation (writing, code, reports)
    - Problem-solving with iterative improvement
    - Design tasks requiring multiple revisions
    - Any task benefiting from refinement

LangChain Implementation:
    Chained prompts for generation, critique, and refinement with iteration control.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


class IterationStatus(Enum):
    """Status of refinement iteration."""
    INITIAL = "initial"
    REFINING = "refining"
    COMPLETED = "completed"
    MAX_ITERATIONS = "max_iterations_reached"


@dataclass
class Critique:
    """Represents a critique of generated output."""
    strengths: List[str] = field(default_factory=list)
    weaknesses: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)
    score: float = 0.0  # Quality score 0-1


@dataclass
class RefinementIteration:
    """Represents one iteration of the refinement process."""
    iteration: int
    output: str
    critique: Optional[Critique] = None
    status: IterationStatus = IterationStatus.INITIAL
    
    def __str__(self):
        return f"Iteration {self.iteration}: {self.status.value} (score: {self.critique.score if self.critique else 'N/A'})"


class SelfRefineAgent:
    """Agent that uses self-refinement to improve outputs."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", 
                 max_iterations: int = 3,
                 quality_threshold: float = 0.8):
        """
        Initialize the Self-Refine agent.
        
        Args:
            model_name: Name of the OpenAI model
            max_iterations: Maximum refinement iterations
            quality_threshold: Stop if quality score exceeds this
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.critic_llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,  # Lower temperature for more consistent critique
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.max_iterations = max_iterations
        self.quality_threshold = quality_threshold
        self.history: List[RefinementIteration] = []
    
    def generate_initial(self, task: str, context: str = "") -> str:
        """
        Generate initial output for the task.
        
        Args:
            task: The task description
            context: Additional context
            
        Returns:
            Initial generated output
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Generate a response to the given task.
Focus on creating a solid first draft. It doesn't need to be perfect yet."""),
            ("human", """{context}

Task: {task}

Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        output = chain.invoke({
            "task": task,
            "context": context if context else "No additional context."
        })
        
        return output.strip()
    
    def critique(self, task: str, output: str, iteration: int) -> Critique:
        """
        Generate critique of current output.
        
        Args:
            task: Original task
            output: Current output to critique
            iteration: Current iteration number
            
        Returns:
            Critique object with feedback
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a constructive critic. Evaluate the response to the task.

Provide:
1. Strengths: What works well (2-3 points)
2. Weaknesses: What could be improved (2-3 points)
3. Suggestions: Specific improvements to make (2-3 points)
4. Score: Overall quality score from 0.0 to 1.0

Format your response as:
STRENGTHS:
- [strength 1]
- [strength 2]

WEAKNESSES:
- [weakness 1]
- [weakness 2]

SUGGESTIONS:
- [suggestion 1]
- [suggestion 2]

SCORE: [0.0-1.0]"""),
            ("human", """Task: {task}

Current Response (Iteration {iteration}):
{output}

Critique:""")
        ])
        
        chain = prompt | self.critic_llm | StrOutputParser()
        critique_text = chain.invoke({
            "task": task,
            "output": output,
            "iteration": iteration
        })
        
        # Parse critique
        critique = Critique()
        
        sections = {
            "strengths": [],
            "weaknesses": [],
            "suggestions": []
        }
        
        current_section = None
        for line in critique_text.split('\n'):
            line = line.strip()
            
            if line.upper().startswith('STRENGTHS:'):
                current_section = "strengths"
            elif line.upper().startswith('WEAKNESSES:'):
                current_section = "weaknesses"
            elif line.upper().startswith('SUGGESTIONS:'):
                current_section = "suggestions"
            elif line.upper().startswith('SCORE:'):
                try:
                    score_str = line.split(':', 1)[1].strip()
                    critique.score = float(score_str)
                except:
                    critique.score = 0.5
            elif line.startswith('-') and current_section:
                sections[current_section].append(line.lstrip('- ').strip())
        
        critique.strengths = sections["strengths"]
        critique.weaknesses = sections["weaknesses"]
        critique.suggestions = sections["suggestions"]
        
        # Ensure score is set
        if critique.score == 0.0 and (critique.strengths or not critique.weaknesses):
            critique.score = 0.6  # Default reasonable score
        
        return critique
    
    def refine(self, task: str, current_output: str, critique: Critique) -> str:
        """
        Generate refined version based on critique.
        
        Args:
            task: Original task
            current_output: Current output to refine
            critique: Critique with improvement suggestions
            
        Returns:
            Refined output
        """
        weaknesses_text = "\n".join([f"- {w}" for w in critique.weaknesses])
        suggestions_text = "\n".join([f"- {s}" for s in critique.suggestions])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Improve the response based on the critique provided.
Address the weaknesses and implement the suggestions while maintaining the strengths.
Produce a refined, higher-quality version."""),
            ("human", """Task: {task}

Current Response:
{current_output}

Weaknesses to Address:
{weaknesses}

Suggestions to Implement:
{suggestions}

Refined Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        refined = chain.invoke({
            "task": task,
            "current_output": current_output,
            "weaknesses": weaknesses_text,
            "suggestions": suggestions_text
        })
        
        return refined.strip()
    
    def refine_iteratively(self, task: str, context: str = "",
                          verbose: bool = True) -> Dict[str, Any]:
        """
        Perform iterative refinement on the task.
        
        Args:
            task: The task to complete
            context: Additional context
            verbose: Whether to print progress
            
        Returns:
            Dictionary with final output and refinement history
        """
        if verbose:
            print(f"\nTask: {task}\n")
            print("="*60)
            print("ITERATION 1: Initial Generation")
            print("="*60)
        
        # Generate initial output
        initial_output = self.generate_initial(task, context)
        
        iteration = RefinementIteration(
            iteration=1,
            output=initial_output,
            status=IterationStatus.INITIAL
        )
        self.history.append(iteration)
        
        if verbose:
            print(f"\nGenerated:\n{initial_output[:200]}..." if len(initial_output) > 200 else f"\nGenerated:\n{initial_output}")
        
        current_output = initial_output
        
        # Refinement loop
        for i in range(2, self.max_iterations + 2):
            if verbose:
                print(f"\n{'='*60}")
                print(f"ITERATION {i}: Critique & Refine")
                print("="*60)
            
            # Critique current output
            critique = self.critique(task, current_output, i - 1)
            
            # Update previous iteration with critique
            self.history[-1].critique = critique
            
            if verbose:
                print(f"\nCritique:")
                print(f"  Quality Score: {critique.score:.2f}")
                if critique.strengths:
                    print(f"  Strengths: {len(critique.strengths)} identified")
                if critique.weaknesses:
                    print(f"  Weaknesses: {len(critique.weaknesses)} identified")
                if critique.suggestions:
                    print(f"  Suggestions: {len(critique.suggestions)} provided")
            
            # Check if quality threshold met
            if critique.score >= self.quality_threshold:
                self.history[-1].status = IterationStatus.COMPLETED
                if verbose:
                    print(f"\n✓ Quality threshold ({self.quality_threshold}) met!")
                break
            
            # Check max iterations
            if i > self.max_iterations:
                self.history[-1].status = IterationStatus.MAX_ITERATIONS
                if verbose:
                    print(f"\n⚠ Max iterations ({self.max_iterations}) reached")
                break
            
            # Refine output
            if verbose:
                print(f"\nRefining...")
            
            refined_output = self.refine(task, current_output, critique)
            
            iteration = RefinementIteration(
                iteration=i,
                output=refined_output,
                status=IterationStatus.REFINING
            )
            self.history.append(iteration)
            
            if verbose:
                print(f"\nRefined:\n{refined_output[:200]}..." if len(refined_output) > 200 else f"\nRefined:\n{refined_output}")
            
            current_output = refined_output
        
        # Final critique
        if self.history[-1].critique is None:
            final_critique = self.critique(task, current_output, len(self.history))
            self.history[-1].critique = final_critique
        
        return {
            "task": task,
            "final_output": current_output,
            "iterations": len(self.history),
            "final_score": self.history[-1].critique.score if self.history[-1].critique else 0.0,
            "history": self.history,
            "status": self.history[-1].status.value
        }
    
    def get_improvement_summary(self) -> str:
        """Generate summary of improvements across iterations."""
        if len(self.history) < 2:
            return "Not enough iterations to compare."
        
        initial_score = self.history[0].critique.score if self.history[0].critique else 0.0
        final_score = self.history[-1].critique.score if self.history[-1].critique else 0.0
        
        improvement = final_score - initial_score
        improvement_pct = (improvement / max(initial_score, 0.1)) * 100
        
        summary = f"""
Improvement Summary:
  Initial Score: {initial_score:.2f}
  Final Score: {final_score:.2f}
  Improvement: {improvement:+.2f} ({improvement_pct:+.1f}%)
  Total Iterations: {len(self.history)}
"""
        return summary


def demonstrate_self_refine():
    """Demonstrates the Self-Refine pattern."""
    
    print("=" * 80)
    print("PATTERN 009: Self-Refine")
    print("=" * 80)
    print()
    print("Self-Refine iteratively improves outputs through:")
    print("1. Initial Generation: Create first draft")
    print("2. Self-Critique: Analyze strengths and weaknesses")
    print("3. Refinement: Generate improved version")
    print("4. Iteration: Repeat until quality threshold met")
    print()
    
    # Create agent
    agent = SelfRefineAgent(max_iterations=3, quality_threshold=0.85)
    
    # Test tasks
    tasks = [
        {
            "task": "Write a brief explanation of quantum entanglement for a general audience",
            "context": "Make it accessible but accurate, around 100 words"
        },
        {
            "task": "Create a function to calculate the Fibonacci sequence efficiently",
            "context": "Use Python, include docstring and handle edge cases"
        }
    ]
    
    for idx, test_case in enumerate(tasks, 1):
        print(f"\n{'='*80}")
        print(f"Example {idx}")
        print('='*80)
        
        try:
            # Reset history
            agent.history = []
            
            result = agent.refine_iteratively(
                task=test_case['task'],
                context=test_case['context'],
                verbose=True
            )
            
            print(f"\n\n{'='*80}")
            print("FINAL RESULT")
            print('='*80)
            print(f"\nFinal Output:\n{result['final_output']}")
            print(f"\n{agent.get_improvement_summary()}")
            print(f"Status: {result['status']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("SELF-REFINE PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Iterative Improvement: Multiple refinement cycles")
    print("2. Self-Critique: Autonomous evaluation of outputs")
    print("3. Targeted Refinement: Addressing specific weaknesses")
    print("4. Quality Tracking: Score-based improvement monitoring")
    print("5. Convergence: Stops when threshold met or max iterations")
    print()
    print("Advantages:")
    print("- Higher quality outputs through iteration")
    print("- No external feedback required")
    print("- Transparent improvement process")
    print("- Handles complex generation tasks")
    print()
    print("When to use Self-Refine:")
    print("- Content creation (writing, code, reports)")
    print("- Tasks benefiting from multiple revisions")
    print("- When quality is more important than speed")
    print("- Complex generation requiring refinement")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Generation, critique, and refinement prompts")
    print("- StrOutputParser: Parse LLM outputs")
    print("- Separate LLMs: Different temperatures for generation vs critique")
    print("- Iteration control: Quality threshold and max iterations")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_self_refine()

    

Agentic Design Pattern: Self Refine

Pattern 009: Self Refine – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Self Refine pattern implementation.

---

Pattern Description

Pattern 009: Self-Refine

Description: Iteratively improves outputs through self-critique and refinement. The agent generates an initial output, evaluates it, identifies weaknesses, and produces an improved version. This cycle repeats until quality threshold is met or max iterations reached.

Key Concepts: – Initial Generation: Create first draft output – Self-Critique: Analyze weaknesses and areas for improvement – Refinement: Generate improved version based on critique – Iteration: Repeat until satisfactory or max iterations – Quality Tracking: Monitor improvement across iterations

Benefits: – Higher quality outputs through iteration – Self-improvement without external feedback – Transparency in refinement process – Handles complex generation tasks

Use Cases: – Content creation (writing, code, reports) – Problem-solving with iterative improvement – Design tasks requiring multiple revisions – Any task benefiting from refinement

LangChain Implementation: Chained prompts for generation, critique, and refinement with iteration control.

---

File Statistics

• Total Lines: 486
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 009: Self-Refine

Description:
    Iteratively improves outputs through self-critique and refinement.
    The agent generates an initial output, evaluates it, identifies weaknesses,
    and produces an improved version. This cycle repeats until quality threshold
    is met or max iterations reached.

Key Concepts:
    - Initial Generation: Create first draft output
    - Self-Critique: Analyze weaknesses and areas for improvement
    - Refinement: Generate improved version based on critique
    - Iteration: Repeat until satisfactory or max iterations
    - Quality Tracking: Monitor improvement across iterations

Benefits:
    - Higher quality outputs through iteration
    - Self-improvement without external feedback
    - Transparency in refinement process
    - Handles complex generation tasks

Use Cases:
    - Content creation (writing, code, reports)
    - Problem-solving with iterative improvement
    - Design tasks requiring multiple revisions
    - Any task benefiting from refinement

LangChain Implementation:
    Chained prompts for generation, critique, and refinement with iteration control.
"""

Purpose: This module implements the Self Refine pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: IterationStatus

Lines 47-52

Inherits from: Enum

Purpose:

Status of refinement iteration.

---

Class: Critique

Lines 56-61

Purpose:

Represents a critique of generated output.

---

Class: RefinementIteration

Lines 65-73

Purpose:

Represents one iteration of the refinement process.

Methods:

__str__(self)

Lines 72-73

Purpose: Str operation.

Parameters:

---

Class: SelfRefineAgent

Lines 76-387

Purpose:

Agent that uses self-refinement to improve outputs.

Methods:

__init__(self, model_name, max_iterations, quality_threshold)

Lines 79-104

Purpose:

Initialize the Self-Refine agent.

Args:
    model_name: Name of the OpenAI model
    max_iterations: Maximum refinement iterations
    quality_threshold: Stop if quality score exceeds this

Parameters: – model_name: Language model instance – max_iterations: Max iterations – quality_threshold: Quality threshold

generate_initial(self, task, context)

Lines 106-133

Purpose:

Generate initial output for the task.

Args:
    task: The task description
    context: Additional context

Returns:
    Initial generated output

Parameters: – task: Task – context: Context

Returns: str

critique(self, task, output, iteration)

Lines 135-221

Purpose:

Generate critique of current output.

Args:
    task: Original task
    output: Current output to critique
    iteration: Current iteration number

Returns:
    Critique object with feedback

Parameters: – task: Task – output: Output – iteration: Iteration

Returns: Critique

refine(self, task, current_output, critique)

Lines 223-264

Purpose:

Generate refined version based on critique.

Args:
    task: Original task
    current_output: Current output to refine
    critique: Critique with improvement suggestions

Returns:
    Refined output

Parameters: – task: Task – current_output: Current output – critique: Critique

Returns: str

refine_iteratively(self, task, context, verbose)

Lines 266-367

Purpose:

Perform iterative refinement on the task.

Args:
    task: The task to complete
    context: Additional context
    verbose: Whether to print progress

Returns:
    Dictionary with final output and refinement history

Parameters: – task: Task – context: Context – verbose: Enable detailed logging

Returns: <ast.Subscript object at 0x000001A06D658D90>

get_improvement_summary(self)

Lines 369-387

Purpose:

Generate summary of improvements across iterations.

Parameters:

Returns: str

---

Functions

demonstrate_self_refine()

Lines 390-475

Purpose:

Demonstrates the Self-Refine pattern.

---

Key Concepts Summary

Pattern Overview

The Self Refine pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 008
• Pattern 010
• Other Reasoning Patterns (001-011)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Self Refine pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 009_self_refine.py
Total Lines: 486
Pattern Category: AI Agent Design Pattern #009

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 008: LLM Compiler

Description:
    Optimizes LLM execution by analyzing task dependencies, parallelizing
    independent operations, and optimizing the execution plan like a compiler
    optimizes code. Combines planning with execution optimization.

Key Concepts:
    - Task Decomposition: Break complex tasks into subtasks
    - Dependency Analysis: Build dependency graph (DAG)
    - Optimization: Identify parallelizable operations
    - Scheduling: Optimal execution order
    - Resource Management: Efficient LLM call allocation

Compiler Phases:
    1. Parsing: Understand and decompose the task
    2. Analysis: Build dependency graph
    3. Optimization: Find parallelization opportunities
    4. Code Generation: Create execution plan
    5. Execution: Run optimized plan

Use Cases:
    - Complex multi-step workflows
    - Research tasks with multiple branches
    - Data processing pipelines
    - Tasks with many independent subtasks

LangChain Implementation:
    Uses dependency graphs and parallel execution with LangChain tools.
"""

import os
from typing import List, Dict, Any, Set, Optional
from dataclasses import dataclass, field
from collections import defaultdict, deque
from concurrent.futures import ThreadPoolExecutor, as_completed
from enum import Enum
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


class TaskStatus(Enum):
    """Status of a task in the execution plan."""
    PENDING = "pending"
    READY = "ready"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Task:
    """Represents a single task in the execution plan."""
    id: str
    description: str
    dependencies: Set[str] = field(default_factory=set)
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[str] = None
    error: Optional[str] = None
    parallel_group: Optional[int] = None  # For grouping parallel tasks
    
    def can_execute(self, completed_tasks: Set[str]) -> bool:
        """Check if this task can be executed."""
        return (self.status == TaskStatus.PENDING and 
                self.dependencies.issubset(completed_tasks))


@dataclass
class ExecutionPlan:
    """Represents the optimized execution plan."""
    goal: str
    tasks: Dict[str, Task] = field(default_factory=dict)
    dependency_graph: Dict[str, Set[str]] = field(default_factory=lambda: defaultdict(set))
    parallel_groups: List[List[str]] = field(default_factory=list)
    
    def add_task(self, task: Task):
        """Add a task to the plan."""
        self.tasks[task.id] = task
        for dep in task.dependencies:
            self.dependency_graph[dep].add(task.id)
    
    def get_ready_tasks(self, completed: Set[str]) -> List[Task]:
        """Get all tasks that are ready to execute."""
        ready = []
        for task in self.tasks.values():
            if task.can_execute(completed):
                ready.append(task)
        return ready
    
    def topological_sort(self) -> List[str]:
        """Perform topological sort on tasks."""
        in_degree = {task_id: len(task.dependencies) 
                     for task_id, task in self.tasks.items()}
        
        queue = deque([task_id for task_id, degree in in_degree.items() if degree == 0])
        sorted_tasks = []
        
        while queue:
            task_id = queue.popleft()
            sorted_tasks.append(task_id)
            
            for dependent in self.dependency_graph[task_id]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)
        
        return sorted_tasks


class LLMCompiler:
    """Compiler that optimizes LLM task execution."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", max_parallel: int = 3):
        """
        Initialize the LLM Compiler.
        
        Args:
            model_name: Name of the OpenAI model
            max_parallel: Maximum number of parallel tasks
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.max_parallel = max_parallel
    
    def parse_task(self, goal: str) -> ExecutionPlan:
        """
        Parse the goal and decompose into subtasks.
        
        Args:
            goal: The high-level goal
            
        Returns:
            Initial execution plan with tasks
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Decompose the goal into specific, actionable subtasks.
For each subtask:
1. Give it a unique ID (T1, T2, T3, etc.)
2. Describe what needs to be done
3. List dependencies (which tasks must complete first)

Format:
T1: [description] | Dependencies: none
T2: [description] | Dependencies: T1
T3: [description] | Dependencies: T1
T4: [description] | Dependencies: T2, T3
...

Be specific and ensure dependencies are correct."""),
            ("human", "Goal: {goal}\n\nTask Decomposition:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({"goal": goal})
        
        # Parse tasks
        plan = ExecutionPlan(goal=goal)
        
        for line in response.strip().split('\n'):
            line = line.strip()
            if not line or ':' not in line:
                continue
            
            # Parse: T1: description | Dependencies: T2, T3
            parts = line.split('|')
            if len(parts)  ExecutionPlan:
        """
        Analyze dependencies and optimize execution order.
        
        Args:
            plan: Initial execution plan
            
        Returns:
            Plan with optimized execution order
        """
        # Topological sort to get valid execution order
        sorted_tasks = plan.topological_sort()
        
        # Identify parallel groups (tasks at same level)
        completed = set()
        parallel_groups = []
        
        while completed != set(plan.tasks.keys()):
            # Get all tasks that can execute now
            ready_tasks = []
            for task_id in sorted_tasks:
                if task_id not in completed:
                    task = plan.tasks[task_id]
                    if task.can_execute(completed):
                        ready_tasks.append(task_id)
            
            if not ready_tasks:
                break  # Circular dependency or error
            
            parallel_groups.append(ready_tasks)
            completed.update(ready_tasks)
            
            # Assign parallel group numbers
            for task_id in ready_tasks:
                plan.tasks[task_id].parallel_group = len(parallel_groups) - 1
        
        plan.parallel_groups = parallel_groups
        
        return plan
    
    def optimize_plan(self, plan: ExecutionPlan) -> ExecutionPlan:
        """
        Apply optimizations to the execution plan.
        
        Args:
            plan: Analyzed execution plan
            
        Returns:
            Optimized plan
        """
        # Optimization strategies:
        # 1. Limit parallel groups to max_parallel
        # 2. Try to balance load across groups
        # 3. Merge small groups if beneficial
        
        optimized_groups = []
        for group in plan.parallel_groups:
            if len(group)  str:
        """
        Execute a single task.
        
        Args:
            task: The task to execute
            plan: The complete plan for context
            
        Returns:
            Task result
        """
        # Build context from completed dependencies
        context_parts = []
        for dep_id in task.dependencies:
            dep_task = plan.tasks.get(dep_id)
            if dep_task and dep_task.result:
                context_parts.append(f"{dep_id}: {dep_task.result}")
        
        context = "\n".join(context_parts) if context_parts else "No prior context."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Execute the task using the provided context from previous steps."),
            ("human", """Goal: {goal}

Prior Results:
{context}

Task: {task_description}

Result:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "goal": plan.goal,
            "context": context,
            "task_description": task.description
        })
        
        return result.strip()
    
    def execute_plan(self, plan: ExecutionPlan) -> Dict[str, Any]:
        """
        Execute the optimized plan.
        
        Args:
            plan: The execution plan
            
        Returns:
            Execution results
        """
        print(f"\nGoal: {plan.goal}\n")
        print(f"Total Tasks: {len(plan.tasks)}")
        print(f"Parallel Groups: {len(plan.parallel_groups)}\n")
        
        # Show execution plan
        print("Execution Plan:")
        for i, group in enumerate(plan.parallel_groups):
            group_tasks = [plan.tasks[tid].description[:50] for tid in group]
            print(f"  Group {i+1} ({len(group)} tasks in parallel):")
            for task_id in group:
                deps = list(plan.tasks[task_id].dependencies)
                dep_str = f" [depends on: {', '.join(deps)}]" if deps else ""
                print(f"    - {task_id}: {plan.tasks[task_id].description[:60]}{dep_str}")
        print()
        
        completed = set()
        total_llm_calls = 0
        
        # Execute groups in sequence
        for group_idx, group in enumerate(plan.parallel_groups):
            print(f"{'='*60}")
            print(f"Executing Group {group_idx + 1} ({len(group)} tasks)")
            print('='*60)
            
            # Execute tasks in group in parallel
            with ThreadPoolExecutor(max_workers=min(len(group), self.max_parallel)) as executor:
                futures = {}
                for task_id in group:
                    task = plan.tasks[task_id]
                    task.status = TaskStatus.RUNNING
                    future = executor.submit(self.execute_task, task, plan)
                    futures[future] = task
                
                # Collect results
                for future in as_completed(futures):
                    task = futures[future]
                    total_llm_calls += 1
                    
                    try:
                        result = future.result()
                        task.result = result
                        task.status = TaskStatus.COMPLETED
                        completed.add(task.id)
                        print(f"\n✓ {task.id} completed")
                        print(f"  Result: {result[:80]}...")
                    except Exception as e:
                        task.status = TaskStatus.FAILED
                        task.error = str(e)
                        print(f"\n✗ {task.id} failed: {str(e)}")
        
        # Generate final summary
        summary = self._synthesize_results(plan)
        
        return {
            "goal": plan.goal,
            "total_tasks": len(plan.tasks),
            "completed_tasks": len(completed),
            "parallel_groups": len(plan.parallel_groups),
            "total_llm_calls": total_llm_calls,
            "summary": summary
        }
    
    def _synthesize_results(self, plan: ExecutionPlan) -> str:
        """Synthesize final summary from all task results."""
        results = []
        for task in plan.tasks.values():
            if task.status == TaskStatus.COMPLETED and task.result:
                results.append(f"{task.id}: {task.result}")
        
        if not results:
            return "No tasks completed successfully."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Synthesize the results into a comprehensive summary."),
            ("human", """Goal: {goal}

Task Results:
{results}

Summary:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            summary = chain.invoke({
                "goal": plan.goal,
                "results": "\n\n".join(results)
            })
            return summary.strip()
        except:
            return "Summary generation failed."
    
    def compile_and_execute(self, goal: str) -> Dict[str, Any]:
        """
        Compile and execute a goal (full pipeline).
        
        Args:
            goal: The goal to achieve
            
        Returns:
            Execution results
        """
        print(f"\n{'='*80}")
        print("LLM COMPILER: PARSING")
        print('='*80)
        
        # Phase 1: Parse and decompose
        plan = self.parse_task(goal)
        print(f"\nParsed {len(plan.tasks)} tasks")
        
        print(f"\n{'='*80}")
        print("LLM COMPILER: ANALYSIS")
        print('='*80)
        
        # Phase 2: Analyze dependencies
        plan = self.analyze_dependencies(plan)
        print(f"\nIdentified {len(plan.parallel_groups)} execution groups")
        
        print(f"\n{'='*80}")
        print("LLM COMPILER: OPTIMIZATION")
        print('='*80)
        
        # Phase 3: Optimize
        plan = self.optimize_plan(plan)
        print(f"\nOptimized for {self.max_parallel} max parallel tasks")
        
        print(f"\n{'='*80}")
        print("LLM COMPILER: EXECUTION")
        print('='*80)
        
        # Phase 4: Execute
        result = self.execute_plan(plan)
        
        return result


def demonstrate_llm_compiler():
    """Demonstrates the LLM Compiler pattern."""
    
    print("=" * 80)
    print("PATTERN 008: LLM Compiler")
    print("=" * 80)
    print()
    print("LLM Compiler optimizes execution like a code compiler:")
    print("1. Parsing: Decompose task into subtasks")
    print("2. Analysis: Build dependency graph")
    print("3. Optimization: Identify parallelization opportunities")
    print("4. Execution: Run optimized plan with parallel execution")
    print()
    
    # Create compiler
    compiler = LLMCompiler(max_parallel=3)
    
    # Test goals
    goals = [
        "Research the history of artificial intelligence and create a timeline of major milestones",
        "Analyze the advantages and disadvantages of renewable energy sources and compare them"
    ]
    
    for i, goal in enumerate(goals, 1):
        print(f"\n{'='*80}")
        print(f"Example {i}")
        print('='*80)
        
        try:
            result = compiler.compile_and_execute(goal)
            
            print(f"\n\n{'='*80}")
            print("COMPILATION SUMMARY")
            print('='*80)
            print(f"\nGoal: {result['goal']}")
            print(f"Total Tasks: {result['total_tasks']}")
            print(f"Completed: {result['completed_tasks']}/{result['total_tasks']}")
            print(f"Parallel Groups: {result['parallel_groups']}")
            print(f"Total LLM Calls: {result['total_llm_calls']}")
            print(f"\nFinal Summary:\n{result['summary']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("LLM COMPILER PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Task Decomposition: Breaking complex goals into subtasks")
    print("2. Dependency Analysis: Building execution dependency graph")
    print("3. Parallelization: Executing independent tasks concurrently")
    print("4. Optimization: Grouping and scheduling for efficiency")
    print("5. Resource Management: Limiting parallel execution")
    print()
    print("Advantages:")
    print("- Reduced total execution time through parallelization")
    print("- Efficient resource utilization")
    print("- Clear dependency management")
    print("- Systematic optimization approach")
    print()
    print("When to use LLM Compiler:")
    print("- Complex multi-step workflows")
    print("- Tasks with many independent subtasks")
    print("- Time-sensitive applications")
    print("- Research and analysis tasks")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate for task decomposition")
    print("- Dependency graph (DAG) management")
    print("- ThreadPoolExecutor for parallel execution")
    print("- Result synthesis chain")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_llm_compiler()

    

Agentic Design Pattern: Llm Compiler

Pattern 008: Llm Compiler – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Llm Compiler pattern implementation.

---

Pattern Description

Pattern 008: LLM Compiler

Description: Optimizes LLM execution by analyzing task dependencies, parallelizing independent operations, and optimizing the execution plan like a compiler optimizes code. Combines planning with execution optimization.

Key Concepts: – Task Decomposition: Break complex tasks into subtasks – Dependency Analysis: Build dependency graph (DAG) – Optimization: Identify parallelizable operations – Scheduling: Optimal execution order – Resource Management: Efficient LLM call allocation

Compiler Phases: 1. Parsing: Understand and decompose the task 2. Analysis: Build dependency graph 3. Optimization: Find parallelization opportunities 4. Code Generation: Create execution plan 5. Execution: Run optimized plan

Use Cases: – Complex multi-step workflows – Research tasks with multiple branches – Data processing pipelines – Tasks with many independent subtasks

LangChain Implementation: Uses dependency graphs and parallel execution with LangChain tools.

---

File Statistics

• Total Lines: 553
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 008: LLM Compiler

Description:
    Optimizes LLM execution by analyzing task dependencies, parallelizing
    independent operations, and optimizing the execution plan like a compiler
    optimizes code. Combines planning with execution optimization.

Key Concepts:
    - Task Decomposition: Break complex tasks into subtasks
    - Dependency Analysis: Build dependency graph (DAG)
    - Optimization: Identify parallelizable operations
    - Scheduling: Optimal execution order
    - Resource Management: Efficient LLM call allocation

Compiler Phases:
    1. Parsing: Understand and decompose the task
    2. Analysis: Build dependency graph
    3. Optimization: Find parallelization opportunities
    4. Code Generation: Create execution plan
    5. Execution: Run optimized plan

Use Cases:
    - Complex multi-step workflows
    - Research tasks with multiple branches
    - Data processing pipelines
    - Tasks with many independent subtasks

LangChain Implementation:
    Uses dependency graphs and parallel execution with LangChain tools.
"""

Purpose: This module implements the Llm Compiler pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.defaultdict
import collections.deque
import concurrent.futures.ThreadPoolExecutor
import concurrent.futures.as_completed
import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – collections.deque: Specialized container datatypes – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: TaskStatus

Lines 49-55

Inherits from: Enum

Purpose:

Status of a task in the execution plan.

---

Class: Task

Lines 59-72

Purpose:

Represents a single task in the execution plan.

Methods:

can_execute(self, completed_tasks)

Lines 69-72

Purpose:

Check if this task can be executed.

Parameters: – completed_tasks: Completed tasks

Returns: bool

---

Class: ExecutionPlan

Lines 76-114

Purpose:

Represents the optimized execution plan.

Methods:

add_task(self, task)

Lines 83-87

Purpose:

Add a task to the plan.

Parameters: – task: Task

get_ready_tasks(self, completed)

Lines 89-95

Purpose:

Get all tasks that are ready to execute.

Parameters: – completed: Completed

Returns: <ast.Subscript object at 0x000001A06D615F10>

topological_sort(self)

Lines 97-114

Purpose:

Perform topological sort on tasks.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6AA390>

---

Class: LLMCompiler

Lines 117-464

Purpose:

Compiler that optimizes LLM task execution.

Methods:

__init__(self, model_name, max_parallel)

Lines 120-133

Purpose:

Initialize the LLM Compiler.

Args:
    model_name: Name of the OpenAI model
    max_parallel: Maximum number of parallel tasks

Parameters: – model_name: Language model instance – max_parallel: Max parallel

parse_task(self, goal)

Lines 135-209

Purpose:

Parse the goal and decompose into subtasks.

Args:
    goal: The high-level goal

Returns:
    Initial execution plan with tasks

Parameters: – goal: Goal

Returns: ExecutionPlan

analyze_dependencies(self, plan)

Lines 211-249

Purpose:

Analyze dependencies and optimize execution order.

Args:
    plan: Initial execution plan

Returns:
    Plan with optimized execution order

Parameters: – plan: Plan

Returns: ExecutionPlan

optimize_plan(self, plan)

Lines 251-277

Purpose:

Apply optimizations to the execution plan.

Args:
    plan: Analyzed execution plan

Returns:
    Optimized plan

Parameters: – plan: Plan

Returns: ExecutionPlan

execute_task(self, task, plan)

Lines 279-318

Purpose:

Execute a single task.

Args:
    task: The task to execute
    plan: The complete plan for context

Returns:
    Task result

Parameters: – task: Task – plan: Plan

Returns: str

execute_plan(self, plan)

Lines 320-390

Purpose:

Execute the optimized plan.

Args:
    plan: The execution plan

Returns:
    Execution results

Parameters: – plan: Plan

Returns: <ast.Subscript object at 0x000001A06D6EB350>

_synthesize_results(self, plan)

Lines 392-421

Purpose:

Synthesize final summary from all task results.

Parameters: – plan: Plan

Returns: str

compile_and_execute(self, goal)

Lines 423-464

Purpose:

Compile and execute a goal (full pipeline).

Args:
    goal: The goal to achieve

Returns:
    Execution results

Parameters: – goal: Goal

Returns: <ast.Subscript object at 0x000001A06D628190>

---

Functions

demonstrate_llm_compiler()

Lines 467-542

Purpose:

Demonstrates the LLM Compiler pattern.

---

Key Concepts Summary

Pattern Overview

The Llm Compiler pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 007
• Pattern 009
• Other Reasoning Patterns (001-011)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Llm Compiler pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 008_llm_compiler.py
Total Lines: 553
Pattern Category: AI Agent Design Pattern #008

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 007: ReWOO (Reasoning Without Observation)

Description:
    Decouples reasoning from observation by planning all tool uses upfront,
    then executing them in batch. Unlike ReAct which interleaves reasoning
    and observation, ReWOO plans all actions first, then executes them,
    reducing LLM calls and enabling parallel execution.

Key Concepts:
    - Upfront Planning: Generate complete tool usage plan first
    - Dependency Analysis: Identify dependencies between tool calls
    - Batch Execution: Execute all tools (potentially in parallel)
    - Variable System: Use variables (e.g., #E1, #E2) for results
    - Final Synthesis: Generate answer using all tool results

Benefits:
    - Fewer LLM calls (plan once, execute all)
    - Enables parallel tool execution
    - More efficient for multi-tool tasks
    - Clearer execution plan

Use Cases:
    - Multi-tool research tasks
    - Information gathering from multiple sources
    - Tasks requiring coordinated tool use
    - Scenarios where parallelization is beneficial

LangChain Implementation:
    Separate planning and execution chains with variable substitution.
"""

import os
import re
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from concurrent.futures import ThreadPoolExecutor, as_completed
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import Tool
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


@dataclass
class ToolCall:
    """Represents a planned tool call."""
    variable: str  # e.g., "#E1"
    tool_name: str
    arguments: str
    dependencies: List[str] = field(default_factory=list)  # Variable dependencies
    result: Optional[str] = None
    executed: bool = False


@dataclass
class ExecutionPlan:
    """Represents the complete tool execution plan."""
    question: str
    tool_calls: List[ToolCall] = field(default_factory=list)
    final_reasoning: str = ""
    
    def get_executable_calls(self) -> List[ToolCall]:
        """Get tool calls that can be executed (dependencies met)."""
        executable = []
        for call in self.tool_calls:
            if not call.executed:
                # Check if all dependencies are met
                deps_met = all(
                    any(tc.variable == dep and tc.executed for tc in self.tool_calls)
                    for dep in call.dependencies
                )
                if not call.dependencies or deps_met:
                    executable.append(call)
        return executable


class ReWOOAgent:
    """Agent that uses ReWOO (Reasoning Without Observation) pattern."""
    
    def __init__(self, tools: List[Tool], model_name: str = "gpt-3.5-turbo"):
        """
        Initialize the ReWOO agent.
        
        Args:
            tools: List of available tools
            model_name: Name of the OpenAI model
        """
        self.tools = {tool.name: tool for tool in tools}
        self.tool_descriptions = self._format_tool_descriptions()
        
        self.planner_llm = ChatOpenAI(
            model=model_name,
            temperature=0.0,  # Low temperature for consistent planning
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.solver_llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def _format_tool_descriptions(self) -> str:
        """Format tool descriptions for the planner."""
        descriptions = []
        for name, tool in self.tools.items():
            descriptions.append(f"- {name}: {tool.description}")
        return "\n".join(descriptions)
    
    def plan(self, question: str) -> ExecutionPlan:
        """
        Create an execution plan for answering the question.
        
        Args:
            question: The question to answer
            
        Returns:
            ExecutionPlan with all tool calls planned
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a planner that creates execution plans using available tools.

Available tools:
{tools}

Create a plan to answer the question using these tools. For each step:
1. Assign a variable (#E1, #E2, etc.)
2. Specify the tool and its arguments
3. Reference previous results using their variables

Format each plan step as:
#E1 = ToolName[arguments]
#E2 = ToolName[arguments using #E1 if needed]
...

Then provide reasoning for how to combine the results:
Plan: [Brief description of how to use the results]

Example:
#E1 = Calculator[15 * 7]
#E2 = Search[capital of #E1]
Plan: Use #E1 to get the product, then search for information about #E2."""),
            ("human", "Question: {question}\n\nExecution Plan:")
        ])
        
        chain = prompt | self.planner_llm | StrOutputParser()
        response = chain.invoke({
            "question": question,
            "tools": self.tool_descriptions
        })
        
        # Parse the plan
        plan = ExecutionPlan(question=question)
        
        lines = response.strip().split('\n')
        for line in lines:
            line = line.strip()
            
            # Parse tool calls like: #E1 = ToolName[arguments]
            match = re.match(r'#(E\d+)\s*=\s*(\w+)\[(.*?)\]', line)
            if match:
                variable = f"#{match.group(1)}"
                tool_name = match.group(2)
                arguments = match.group(3).strip()
                
                # Extract dependencies (variables referenced in arguments)
                dependencies = re.findall(r'#E\d+', arguments)
                
                tool_call = ToolCall(
                    variable=variable,
                    tool_name=tool_name,
                    arguments=arguments,
                    dependencies=dependencies
                )
                plan.tool_calls.append(tool_call)
            
            # Extract final reasoning
            elif line.lower().startswith('plan:'):
                plan.final_reasoning = line.split(':', 1)[1].strip()
        
        return plan
    
    def execute_plan(self, plan: ExecutionPlan, parallel: bool = False) -> Dict[str, Any]:
        """
        Execute the tool calls in the plan.
        
        Args:
            plan: The execution plan
            parallel: Whether to execute independent calls in parallel
            
        Returns:
            Dictionary with execution results
        """
        print(f"\nQuestion: {plan.question}\n")
        print("Execution Plan:")
        for call in plan.tool_calls:
            deps = f" (depends on: {', '.join(call.dependencies)})" if call.dependencies else ""
            print(f"  {call.variable} = {call.tool_name}[{call.arguments}]{deps}")
        print(f"\nFinal Plan: {plan.final_reasoning}\n")
        
        print("="*60)
        print("EXECUTING TOOLS")
        print("="*60)
        
        # Execute tool calls respecting dependencies
        while True:
            executable = plan.get_executable_calls()
            if not executable:
                break
            
            if parallel and len(executable) > 1:
                # Execute in parallel
                with ThreadPoolExecutor(max_workers=len(executable)) as executor:
                    futures = {
                        executor.submit(self._execute_tool_call, call, plan): call 
                        for call in executable
                    }
                    
                    for future in as_completed(futures):
                        call = futures[future]
                        try:
                            result = future.result()
                            call.result = result
                            call.executed = True
                            print(f"\n✓ {call.variable} = {result[:100]}...")
                        except Exception as e:
                            call.result = f"Error: {str(e)}"
                            call.executed = True
                            print(f"\n✗ {call.variable} failed: {str(e)}")
            else:
                # Execute sequentially
                for call in executable:
                    try:
                        result = self._execute_tool_call(call, plan)
                        call.result = result
                        call.executed = True
                        print(f"\n✓ {call.variable} = {result[:100]}...")
                    except Exception as e:
                        call.result = f"Error: {str(e)}"
                        call.executed = True
                        print(f"\n✗ {call.variable} failed: {str(e)}")
        
        # Synthesize final answer
        print(f"\n{'='*60}")
        print("SYNTHESIZING ANSWER")
        print('='*60)
        
        answer = self._synthesize_answer(plan)
        
        return {
            "question": plan.question,
            "answer": answer,
            "tool_calls": len(plan.tool_calls),
            "plan": plan.final_reasoning
        }
    
    def _execute_tool_call(self, call: ToolCall, plan: ExecutionPlan) -> str:
        """Execute a single tool call."""
        # Substitute variable references with actual results
        arguments = call.arguments
        for dep_var in call.dependencies:
            # Find the result for this dependency
            for other_call in plan.tool_calls:
                if other_call.variable == dep_var and other_call.result:
                    arguments = arguments.replace(dep_var, other_call.result)
        
        # Get the tool
        tool = self.tools.get(call.tool_name)
        if not tool:
            raise ValueError(f"Tool {call.tool_name} not found")
        
        # Execute the tool
        try:
            result = tool.func(arguments)
            return str(result)
        except Exception as e:
            raise Exception(f"Tool execution failed: {str(e)}")
    
    def _synthesize_answer(self, plan: ExecutionPlan) -> str:
        """Synthesize final answer from tool results."""
        # Build evidence from tool results
        evidence = []
        for call in plan.tool_calls:
            if call.executed and call.result:
                evidence.append(f"{call.variable}: {call.result}")
        
        evidence_text = "\n".join(evidence)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Based on the evidence gathered from tool execution, 
provide a comprehensive answer to the original question.

Use the evidence to construct a well-reasoned response."""),
            ("human", """Question: {question}

Evidence:
{evidence}

Plan: {plan}

Answer:""")
        ])
        
        chain = prompt | self.solver_llm | StrOutputParser()
        answer = chain.invoke({
            "question": plan.question,
            "evidence": evidence_text,
            "plan": plan.final_reasoning
        })
        
        return answer.strip()
    
    def solve(self, question: str, parallel: bool = False) -> Dict[str, Any]:
        """
        Solve a question using ReWOO pattern.
        
        Args:
            question: The question to answer
            parallel: Whether to use parallel execution
            
        Returns:
            Dictionary with answer and metadata
        """
        print(f"\n{'='*80}")
        print("ReWOO: PLANNING PHASE")
        print('='*80)
        
        # Create plan
        plan = self.plan(question)
        
        print(f"\n{'='*80}")
        print("ReWOO: EXECUTION PHASE")
        print('='*80)
        
        # Execute plan
        result = self.execute_plan(plan, parallel=parallel)
        
        return result


# Example tools
def calculator(expression: str) -> float:
    """Calculate a mathematical expression."""
    try:
        # Simple eval (in production, use a safe math parser)
        result = eval(expression)
        return result
    except Exception as e:
        return f"Calculation error: {str(e)}"


def search(query: str) -> str:
    """Simulate a search tool."""
    # In production, this would call a real search API
    responses = {
        "capital of france": "Paris is the capital of France.",
        "population of tokyo": "Tokyo has a population of approximately 14 million people.",
        "speed of light": "The speed of light is approximately 299,792,458 meters per second.",
        "inventor of telephone": "Alexander Graham Bell is credited with inventing the telephone.",
    }
    
    query_lower = query.lower()
    for key, value in responses.items():
        if key in query_lower:
            return value
    
    return f"Search results for: {query}"


def get_date(query: str) -> str:
    """Get current date information."""
    from datetime import datetime
    now = datetime.now()
    
    if "year" in query.lower():
        return str(now.year)
    elif "month" in query.lower():
        return now.strftime("%B")
    elif "day" in query.lower():
        return str(now.day)
    else:
        return now.strftime("%Y-%m-%d")


def demonstrate_rewoo():
    """Demonstrates the ReWOO pattern."""
    
    print("=" * 80)
    print("PATTERN 007: ReWOO (Reasoning Without Observation)")
    print("=" * 80)
    print()
    print("ReWOO decouples reasoning from observation:")
    print("1. Planning Phase: Plan all tool uses upfront")
    print("2. Dependency Analysis: Identify dependencies between calls")
    print("3. Execution Phase: Execute tools (potentially in parallel)")
    print("4. Synthesis: Generate answer from all results")
    print()
    
    # Create tools
    tools = [
        Tool(
            name="Calculator",
            func=calculator,
            description="Calculate mathematical expressions. Input should be a valid expression."
        ),
        Tool(
            name="Search",
            func=search,
            description="Search for information. Input should be a search query."
        ),
        Tool(
            name="GetDate",
            func=get_date,
            description="Get current date information. Input can be 'year', 'month', 'day', or 'full'."
        )
    ]
    
    # Create agent
    agent = ReWOOAgent(tools=tools)
    
    # Test questions
    questions = [
        "What is 25 * 34, and what is the capital of France?",
        "Calculate 144 / 12, then search for information about that number",
    ]
    
    for i, question in enumerate(questions, 1):
        print(f"\n{'='*80}")
        print(f"Example {i}")
        print('='*80)
        
        try:
            result = agent.solve(question, parallel=False)
            
            print(f"\n\n{'='*80}")
            print("FINAL RESULT")
            print('='*80)
            print(f"\nQuestion: {result['question']}")
            print(f"\nAnswer: {result['answer']}")
            print(f"\nTool Calls: {result['tool_calls']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("ReWOO PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Upfront Planning: All tool uses planned before execution")
    print("2. Variable System: Results referenced as #E1, #E2, etc.")
    print("3. Dependency Tracking: Execution order based on dependencies")
    print("4. Batch Execution: All tools executed in one phase")
    print("5. Result Synthesis: Final answer combines all evidence")
    print()
    print("Advantages over ReAct:")
    print("- Fewer LLM calls (plan once vs. after each observation)")
    print("- Enables parallel tool execution")
    print("- Clearer execution flow")
    print("- More efficient for multi-tool tasks")
    print()
    print("When to use ReWOO:")
    print("- Multi-tool information gathering")
    print("- Tasks with independent tool calls")
    print("- Scenarios benefiting from parallelization")
    print("- When you want predictable tool usage")
    print()
    print("LangChain Components Used:")
    print("- Tool abstraction for uniform tool interface")
    print("- Separate planning and execution chains")
    print("- Variable substitution system")
    print("- Optional parallel execution")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_rewoo()

    

Agentic Design Pattern: Rewoo

Pattern 007: Rewoo – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Rewoo pattern implementation.

---

Pattern Description

Pattern 007: ReWOO (Reasoning Without Observation)

Description: Decouples reasoning from observation by planning all tool uses upfront, then executing them in batch. Unlike ReAct which interleaves reasoning and observation, ReWOO plans all actions first, then executes them, reducing LLM calls and enabling parallel execution.

Key Concepts: – Upfront Planning: Generate complete tool usage plan first – Dependency Analysis: Identify dependencies between tool calls – Batch Execution: Execute all tools (potentially in parallel) – Variable System: Use variables (e.g., #E1, #E2) for results – Final Synthesis: Generate answer using all tool results

Benefits: – Fewer LLM calls (plan once, execute all) – Enables parallel tool execution – More efficient for multi-tool tasks – Clearer execution plan

Use Cases: – Multi-tool research tasks – Information gathering from multiple sources – Tasks requiring coordinated tool use – Scenarios where parallelization is beneficial

LangChain Implementation: Separate planning and execution chains with variable substitution.

---

File Statistics

• Total Lines: 492
• Classes: 3
• Functions: 4
• Constants: 0

---

Module Documentation

"""
Pattern 007: ReWOO (Reasoning Without Observation)

Description:
    Decouples reasoning from observation by planning all tool uses upfront,
    then executing them in batch. Unlike ReAct which interleaves reasoning
    and observation, ReWOO plans all actions first, then executes them,
    reducing LLM calls and enabling parallel execution.

Key Concepts:
    - Upfront Planning: Generate complete tool usage plan first
    - Dependency Analysis: Identify dependencies between tool calls
    - Batch Execution: Execute all tools (potentially in parallel)
    - Variable System: Use variables (e.g., #E1, #E2) for results
    - Final Synthesis: Generate answer using all tool results

Benefits:
    - Fewer LLM calls (plan once, execute all)
    - Enables parallel tool execution
    - More efficient for multi-tool tasks
    - Clearer execution plan

Use Cases:
    - Multi-tool research tasks
    - Information gathering from multiple sources
    - Tasks requiring coordinated tool use
    - Scenarios where parallelization is beneficial

LangChain Implementation:
    Separate planning and execution chains with variable substitution.
"""

Purpose: This module implements the Rewoo pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import concurrent.futures.ThreadPoolExecutor
import concurrent.futures.as_completed
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.tools.Tool
import os
import re
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.tools.Tool: Regular expression operations – os: Environment variable access and system operations – re: Regular expression operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ToolCall

Lines 50-57

Purpose:

Represents a planned tool call.

---

Class: ExecutionPlan

Lines 61-79

Purpose:

Represents the complete tool execution plan.

Methods:

get_executable_calls(self)

Lines 67-79

Purpose:

Get tool calls that can be executed (dependencies met).

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6761D0>

---

Class: ReWOOAgent

Lines 82-343

Purpose:

Agent that uses ReWOO (Reasoning Without Observation) pattern.

Methods:

__init__(self, tools, model_name)

Lines 85-106

Purpose:

Initialize the ReWOO agent.

Args:
    tools: List of available tools
    model_name: Name of the OpenAI model

Parameters: – tools: Available tools for the agent – model_name: Language model instance

_format_tool_descriptions(self)

Lines 108-113

Purpose:

Format tool descriptions for the planner.

Parameters:

Returns: str

plan(self, question)

Lines 115-186

Purpose:

Create an execution plan for answering the question.

Args:
    question: The question to answer

Returns:
    ExecutionPlan with all tool calls planned

Parameters: – question: Question to answer

Returns: ExecutionPlan

execute_plan(self, plan, parallel)

Lines 188-260

Purpose:

Execute the tool calls in the plan.

Args:
    plan: The execution plan
    parallel: Whether to execute independent calls in parallel

Returns:
    Dictionary with execution results

Parameters: – plan: Plan – parallel: Parallel

Returns: <ast.Subscript object at 0x000001A06D65A490>

_execute_tool_call(self, call, plan)

Lines 262-282

Purpose:

Execute a single tool call.

Parameters: – call: Call – plan: Plan

Returns: str

_synthesize_answer(self, plan)

Lines 284-316

Purpose:

Synthesize final answer from tool results.

Parameters: – plan: Plan

Returns: str

solve(self, question, parallel)

Lines 318-343

Purpose:

Solve a question using ReWOO pattern.

Args:
    question: The question to answer
    parallel: Whether to use parallel execution

Returns:
    Dictionary with answer and metadata

Parameters: – question: Question to answer – parallel: Parallel

Returns: <ast.Subscript object at 0x000001A06D178FD0>

---

Functions

calculator(expression)

Lines 347-354

Purpose:

Calculate a mathematical expression.

Parameters: – expression: Expression

Returns: float

search(query)

Lines 357-372

Purpose:

Simulate a search tool.

Parameters: – query: Input query or question

Returns: str

get_date(query)

Lines 375-387

Purpose:

Get current date information.

Parameters: – query: Input query or question

Returns: str

demonstrate_rewoo()

Lines 390-481

Purpose:

Demonstrates the ReWOO pattern.

---

Key Concepts Summary

Pattern Overview

The Rewoo pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 006
• Pattern 008
• Other Reasoning Patterns (001-011)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Rewoo pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 007_rewoo.py
Total Lines: 492
Pattern Category: AI Agent Design Pattern #007

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 006: Plan-and-Execute

Description:
    Separates planning and execution phases for complex multi-step tasks.
    First creates a high-level plan, then executes each step sequentially,
    with optional re-planning based on intermediate results.

Key Concepts:
    - Planning Phase: Create structured plan with steps
    - Execution Phase: Execute each step in sequence
    - Monitoring: Track progress and outcomes
    - Re-planning: Adjust plan based on execution results
    - State Management: Maintain context across steps

Use Cases:
    - Complex multi-step tasks requiring coordination
    - Research and information gathering workflows
    - Content creation with multiple stages
    - Problem-solving with dependencies between steps

LangChain Implementation:
    Uses LangGraph for state management and sequential execution with
    separate planner and executor chains.
"""

import os
from typing import List, Dict, Any, Optional
from enum import Enum
from dataclasses import dataclass, field
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


class StepStatus(Enum):
    """Status of an execution step."""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class PlanStep:
    """Represents a single step in the execution plan."""
    id: int
    description: str
    dependencies: List[int] = field(default_factory=list)
    status: StepStatus = StepStatus.PENDING
    result: Optional[str] = None
    error: Optional[str] = None


@dataclass
class ExecutionPlan:
    """Represents the complete execution plan."""
    goal: str
    steps: List[PlanStep] = field(default_factory=list)
    current_step: int = 0
    completed_steps: int = 0
    
    def get_next_step(self) -> Optional[PlanStep]:
        """Get the next pending step."""
        for step in self.steps:
            if step.status == StepStatus.PENDING:
                # Check if dependencies are met
                if all(
                    self.steps[dep_id - 1].status == StepStatus.COMPLETED 
                    for dep_id in step.dependencies
                ):
                    return step
        return None
    
    def is_complete(self) -> bool:
        """Check if all steps are completed or skipped."""
        return all(
            step.status in [StepStatus.COMPLETED, StepStatus.SKIPPED]
            for step in self.steps
        )
    
    def get_progress(self) -> float:
        """Get completion progress as percentage."""
        if not self.steps:
            return 0.0
        completed = sum(
            1 for step in self.steps 
            if step.status in [StepStatus.COMPLETED, StepStatus.SKIPPED]
        )
        return completed / len(self.steps)


class PlanAndExecuteAgent:
    """Agent that uses plan-and-execute pattern."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", 
                 enable_replanning: bool = True):
        """
        Initialize the Plan-and-Execute agent.
        
        Args:
            model_name: Name of the OpenAI model
            enable_replanning: Whether to enable dynamic re-planning
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.enable_replanning = enable_replanning
        self.execution_history: List[Dict[str, Any]] = []
    
    def create_plan(self, goal: str, context: str = "") -> ExecutionPlan:
        """
        Create an execution plan for the given goal.
        
        Args:
            goal: The goal to achieve
            context: Additional context information
            
        Returns:
            ExecutionPlan with structured steps
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Create a detailed step-by-step plan to achieve the goal.
Break down the task into clear, actionable steps.
Format your response as:
1. [First step]
2. [Second step]
3. [Third step]
etc.

Each step should be specific and achievable."""),
            ("human", """Goal: {goal}

Context: {context}

Plan:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "goal": goal,
            "context": context if context else "No additional context."
        })
        
        # Parse steps
        plan = ExecutionPlan(goal=goal)
        step_id = 1
        
        for line in response.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                # Extract step description
                description = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                if description:
                    step = PlanStep(
                        id=step_id,
                        description=description
                    )
                    plan.steps.append(step)
                    step_id += 1
        
        return plan
    
    def execute_step(self, step: PlanStep, plan: ExecutionPlan) -> str:
        """
        Execute a single step of the plan.
        
        Args:
            step: The step to execute
            plan: The complete plan for context
            
        Returns:
            Result of executing the step
        """
        # Gather context from previous steps
        context = self._build_context(plan)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Execute the given step and provide a detailed result.
Be specific and thorough in your execution.
If the step requires information from previous steps, use the provided context."""),
            ("human", """Goal: {goal}

Previous steps context:
{context}

Current step to execute: {step_description}

Execution result:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            result = chain.invoke({
                "goal": plan.goal,
                "context": context,
                "step_description": step.description
            })
            
            return result.strip()
            
        except Exception as e:
            raise Exception(f"Step execution failed: {str(e)}")
    
    def _build_context(self, plan: ExecutionPlan) -> str:
        """Build context string from completed steps."""
        context_parts = []
        for step in plan.steps:
            if step.status == StepStatus.COMPLETED and step.result:
                context_parts.append(f"Step {step.id}: {step.description}\nResult: {step.result}")
        
        return "\n\n".join(context_parts) if context_parts else "No previous steps completed."
    
    def should_replan(self, plan: ExecutionPlan, last_step: PlanStep) -> bool:
        """
        Determine if re-planning is needed based on execution results.
        
        Args:
            plan: Current execution plan
            last_step: Most recently executed step
            
        Returns:
            True if re-planning is recommended
        """
        if not self.enable_replanning:
            return False
        
        # Simple heuristic: replan if step failed or result suggests issues
        if last_step.status == StepStatus.FAILED:
            return True
        
        # Use LLM to evaluate if re-planning is needed
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate if the plan should be adjusted based on the execution result.
Consider if:
- The result reveals new information that changes the approach
- There are unexpected obstacles
- The remaining steps need to be modified

Respond with only 'YES' or 'NO'."""),
            ("human", """Goal: {goal}

Completed step: {step_description}
Result: {result}

Should we replan? (YES/NO):""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            response = chain.invoke({
                "goal": plan.goal,
                "step_description": last_step.description,
                "result": last_step.result
            })
            
            return "yes" in response.lower()
            
        except:
            return False
    
    def replan(self, plan: ExecutionPlan) -> ExecutionPlan:
        """
        Create a new plan based on current progress.
        
        Args:
            plan: Current execution plan
            
        Returns:
            Updated execution plan
        """
        context = self._build_context(plan)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Given the progress so far, create an updated plan for the remaining work.
Consider what has been accomplished and what still needs to be done.

Format as:
1. [First remaining step]
2. [Second remaining step]
etc."""),
            ("human", """Goal: {goal}

Progress so far:
{context}

Updated plan for remaining work:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "goal": plan.goal,
            "context": context
        })
        
        # Create new plan with completed steps + new steps
        new_plan = ExecutionPlan(goal=plan.goal)
        
        # Keep completed steps
        for step in plan.steps:
            if step.status == StepStatus.COMPLETED:
                new_plan.steps.append(step)
        
        # Add new steps
        step_id = len(new_plan.steps) + 1
        for line in response.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                description = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                if description:
                    new_step = PlanStep(
                        id=step_id,
                        description=description
                    )
                    new_plan.steps.append(new_step)
                    step_id += 1
        
        return new_plan
    
    def execute_plan(self, plan: ExecutionPlan, max_steps: Optional[int] = None) -> Dict[str, Any]:
        """
        Execute the complete plan.
        
        Args:
            plan: The execution plan
            max_steps: Maximum number of steps to execute (None for all)
            
        Returns:
            Dictionary with execution results
        """
        print(f"\nGoal: {plan.goal}")
        print(f"\nPlan ({len(plan.steps)} steps):")
        for step in plan.steps:
            print(f"  {step.id}. {step.description}")
        print()
        
        executed_steps = 0
        
        while not plan.is_complete():
            if max_steps and executed_steps >= max_steps:
                print(f"\nReached maximum steps limit ({max_steps})")
                break
            
            # Get next step
            next_step = plan.get_next_step()
            if not next_step:
                print("\nNo more executable steps (dependencies not met)")
                break
            
            # Execute step
            print(f"\n{'='*60}")
            print(f"Executing Step {next_step.id}: {next_step.description}")
            print('='*60)
            
            next_step.status = StepStatus.IN_PROGRESS
            
            try:
                result = self.execute_step(next_step, plan)
                next_step.result = result
                next_step.status = StepStatus.COMPLETED
                plan.completed_steps += 1
                
                print(f"\n✓ Step {next_step.id} completed")
                print(f"Result: {result[:200]}..." if len(result) > 200 else f"Result: {result}")
                
                # Check if re-planning is needed
                if self.should_replan(plan, next_step):
                    print(f"\n⟳ Re-planning needed based on results...")
                    plan = self.replan(plan)
                    print(f"Updated plan now has {len(plan.steps)} steps")
                
            except Exception as e:
                next_step.status = StepStatus.FAILED
                next_step.error = str(e)
                print(f"\n✗ Step {next_step.id} failed: {str(e)}")
            
            executed_steps += 1
        
        # Generate final summary
        summary = self._generate_summary(plan)
        
        return {
            "goal": plan.goal,
            "total_steps": len(plan.steps),
            "completed_steps": plan.completed_steps,
            "progress": plan.get_progress(),
            "status": "completed" if plan.is_complete() else "incomplete",
            "summary": summary
        }
    
    def _generate_summary(self, plan: ExecutionPlan) -> str:
        """Generate a summary of the execution."""
        completed_results = []
        for step in plan.steps:
            if step.status == StepStatus.COMPLETED and step.result:
                completed_results.append(f"Step {step.id}: {step.result}")
        
        if not completed_results:
            return "No steps completed successfully."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Provide a concise summary of what was accomplished."),
            ("human", """Goal: {goal}

Results:
{results}

Summary:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            summary = chain.invoke({
                "goal": plan.goal,
                "results": "\n\n".join(completed_results)
            })
            return summary.strip()
        except:
            return "Summary generation failed."
    
    def solve_with_plan_execute(self, goal: str, context: str = "") -> Dict[str, Any]:
        """
        Solve a task using plan-and-execute pattern.
        
        Args:
            goal: The goal to achieve
            context: Additional context
            
        Returns:
            Execution results
        """
        print(f"\n{'='*80}")
        print("PLAN-AND-EXECUTE: Creating plan...")
        print('='*80)
        
        # Create initial plan
        plan = self.create_plan(goal, context)
        
        # Execute the plan
        result = self.execute_plan(plan)
        
        return result


def demonstrate_plan_execute():
    """Demonstrates the Plan-and-Execute pattern."""
    
    print("=" * 80)
    print("PATTERN 006: Plan-and-Execute")
    print("=" * 80)
    print()
    print("Plan-and-Execute separates planning from execution:")
    print("1. Planning Phase: Create structured multi-step plan")
    print("2. Execution Phase: Execute each step sequentially")
    print("3. Monitoring: Track progress and results")
    print("4. Re-planning: Adjust plan based on intermediate results")
    print()
    
    # Create agent
    agent = PlanAndExecuteAgent(enable_replanning=True)
    
    # Test goals
    goals = [
        {
            "goal": "Research and write a brief report on renewable energy trends",
            "context": "Focus on solar and wind energy developments in the last 5 years"
        },
        {
            "goal": "Plan a team building event for 20 people",
            "context": "Budget: $2000, Duration: Half day, Location: Urban area"
        }
    ]
    
    for i, test_case in enumerate(goals, 1):
        print(f"\n{'='*80}")
        print(f"Example {i}")
        print('='*80)
        
        try:
            result = agent.solve_with_plan_execute(
                goal=test_case['goal'],
                context=test_case['context']
            )
            
            print(f"\n\n{'='*80}")
            print("EXECUTION SUMMARY")
            print('='*80)
            print(f"\nGoal: {result['goal']}")
            print(f"Status: {result['status'].upper()}")
            print(f"Progress: {result['progress']:.0%}")
            print(f"Steps Completed: {result['completed_steps']}/{result['total_steps']}")
            print(f"\nSummary:\n{result['summary']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("PLAN-AND-EXECUTE PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Structured Planning: Breaking goals into actionable steps")
    print("2. Sequential Execution: Executing steps with dependency tracking")
    print("3. Context Management: Passing results between steps")
    print("4. Progress Monitoring: Tracking completion status")
    print("5. Dynamic Re-planning: Adjusting plan based on results")
    print()
    print("Advantages:")
    print("- Clear separation of planning and execution")
    print("- Better handling of complex multi-step tasks")
    print("- Adaptability through re-planning")
    print("- Progress tracking and monitoring")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Structures planning and execution prompts")
    print("- StrOutputParser: Parses LLM outputs")
    print("- State Management: Tracks plan and execution progress")
    print("- Sequential Chains: Executes steps in order")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_plan_execute()

    

Agentic Design Pattern: Plan And Execute

Pattern 006: Plan And Execute – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Plan And Execute pattern implementation.

---

Pattern Description

Pattern 006: Plan-and-Execute

Description: Separates planning and execution phases for complex multi-step tasks. First creates a high-level plan, then executes each step sequentially, with optional re-planning based on intermediate results.

Key Concepts: – Planning Phase: Create structured plan with steps – Execution Phase: Execute each step in sequence – Monitoring: Track progress and outcomes – Re-planning: Adjust plan based on execution results – State Management: Maintain context across steps

Use Cases: – Complex multi-step tasks requiring coordination – Research and information gathering workflows – Content creation with multiple stages – Problem-solving with dependencies between steps

LangChain Implementation: Uses LangGraph for state management and sequential execution with separate planner and executor chains.

---

File Statistics

• Total Lines: 544
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 006: Plan-and-Execute

Description:
    Separates planning and execution phases for complex multi-step tasks.
    First creates a high-level plan, then executes each step sequentially,
    with optional re-planning based on intermediate results.

Key Concepts:
    - Planning Phase: Create structured plan with steps
    - Execution Phase: Execute each step in sequence
    - Monitoring: Track progress and outcomes
    - Re-planning: Adjust plan based on execution results
    - State Management: Maintain context across steps

Use Cases:
    - Complex multi-step tasks requiring coordination
    - Research and information gathering workflows
    - Content creation with multiple stages
    - Problem-solving with dependencies between steps

LangChain Implementation:
    Uses LangGraph for state management and sequential execution with
    separate planner and executor chains.
"""

Purpose: This module implements the Plan And Execute pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: StepStatus

Lines 41-47

Inherits from: Enum

Purpose:

Status of an execution step.

---

Class: PlanStep

Lines 51-58

Purpose:

Represents a single step in the execution plan.

---

Class: ExecutionPlan

Lines 62-96

Purpose:

Represents the complete execution plan.

Methods:

get_next_step(self)

Lines 69-79

Purpose:

Get the next pending step.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D680D10>

is_complete(self)

Lines 81-86

Purpose:

Check if all steps are completed or skipped.

Parameters:

Returns: bool

get_progress(self)

Lines 88-96

Purpose:

Get completion progress as percentage.

Parameters:

Returns: float

---

Class: PlanAndExecuteAgent

Lines 99-453

Purpose:

Agent that uses plan-and-execute pattern.

Methods:

__init__(self, model_name, enable_replanning)

Lines 102-117

Purpose:

Initialize the Plan-and-Execute agent.

Args:
    model_name: Name of the OpenAI model
    enable_replanning: Whether to enable dynamic re-planning

Parameters: – model_name: Language model instance – enable_replanning: Enable replanning

create_plan(self, goal, context)

Lines 119-170

Purpose:

Create an execution plan for the given goal.

Args:
    goal: The goal to achieve
    context: Additional context information

Returns:
    ExecutionPlan with structured steps

Parameters: – goal: Goal – context: Context

Returns: ExecutionPlan

execute_step(self, step, plan)

Lines 172-212

Purpose:

Execute a single step of the plan.

Args:
    step: The step to execute
    plan: The complete plan for context

Returns:
    Result of executing the step

Parameters: – step: Step – plan: Plan

Returns: str

_build_context(self, plan)

Lines 214-221

Purpose:

Build context string from completed steps.

Parameters: – plan: Plan

Returns: str

should_replan(self, plan, last_step)

Lines 223-270

Purpose:

Determine if re-planning is needed based on execution results.

Args:
    plan: Current execution plan
    last_step: Most recently executed step

Returns:
    True if re-planning is recommended

Parameters: – plan: Plan – last_step: Last step

Returns: bool

replan(self, plan)

Lines 272-328

Purpose:

Create a new plan based on current progress.

Args:
    plan: Current execution plan

Returns:
    Updated execution plan

Parameters: – plan: Plan

Returns: ExecutionPlan

execute_plan(self, plan, max_steps)

Lines 330-399

Purpose:

Execute the complete plan.

Args:
    plan: The execution plan
    max_steps: Maximum number of steps to execute (None for all)

Returns:
    Dictionary with execution results

Parameters: – plan: Plan – max_steps: Max steps

Returns: <ast.Subscript object at 0x000001A06D608ED0>

_generate_summary(self, plan)

Lines 401-430

Purpose:

Generate a summary of the execution.

Parameters: – plan: Plan

Returns: str

solve_with_plan_execute(self, goal, context)

Lines 432-453

Purpose:

Solve a task using plan-and-execute pattern.

Args:
    goal: The goal to achieve
    context: Additional context

Returns:
    Execution results

Parameters: – goal: Goal – context: Context

Returns: <ast.Subscript object at 0x000001A06D627750>

---

Functions

demonstrate_plan_execute()

Lines 456-533

Purpose:

Demonstrates the Plan-and-Execute pattern.

---

Key Concepts Summary

Pattern Overview

The Plan And Execute pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 005
• Pattern 007
• Other Reasoning Patterns (001-011)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Plan And Execute pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 006_plan_and_execute.py
Total Lines: 544
Pattern Category: AI Agent Design Pattern #006

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 005: Self-Consistency

Description:
    Samples multiple reasoning paths for the same question and selects
    the most consistent answer through majority voting or consistency scoring.
    This approach improves reliability by generating diverse solutions and
    aggregating them.

Key Concepts:
    - Multiple Sampling: Generate several independent reasoning chains
    - Diverse Reasoning: Use temperature or prompt variations
    - Aggregation: Majority voting or consistency analysis
    - Answer Selection: Choose most frequent or highest confidence answer

Use Cases:
    - Math problems with multiple solution methods
    - Questions with ambiguous reasoning paths
    - Improving reliability of complex reasoning
    - Reducing model hallucinations

LangChain Implementation:
    Multiple chain invocations with temperature variation and result aggregation.
"""

import os
from typing import List, Dict, Any, Tuple
from collections import Counter
import re
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


class SelfConsistencyAgent:
    """Agent that uses self-consistency for reliable reasoning."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", num_samples: int = 5):
        """
        Initialize the Self-Consistency agent.
        
        Args:
            model_name: Name of the OpenAI model to use
            num_samples: Number of reasoning paths to sample
        """
        self.model_name = model_name
        self.num_samples = num_samples
        self.base_temperature = 0.7
        
    def _create_llm(self, temperature: float) -> ChatOpenAI:
        """Create an LLM instance with specified temperature."""
        return ChatOpenAI(
            model=self.model_name,
            temperature=temperature,
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def generate_reasoning_paths(self, question: str, context: str = "") -> List[Dict[str, str]]:
        """
        Generate multiple diverse reasoning paths for a question.
        
        Args:
            question: The question to answer
            context: Optional context information
            
        Returns:
            List of dictionaries with 'reasoning' and 'answer' keys
        """
        # Base prompt for chain-of-thought reasoning
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Answer the question with clear step-by-step reasoning.
Show your work and explain each step.
At the end, clearly state your final answer."""),
            ("human", "{context}\n\nQuestion: {question}\n\nLet's think step by step:")
        ])
        
        reasoning_paths = []
        
        # Generate multiple samples with varying temperatures
        for i in range(self.num_samples):
            # Vary temperature for diversity
            temperature = self.base_temperature + (i - self.num_samples // 2) * 0.1
            temperature = max(0.1, min(1.0, temperature))
            
            llm = self._create_llm(temperature)
            chain = prompt | llm | StrOutputParser()
            
            try:
                response = chain.invoke({
                    "question": question,
                    "context": context if context else "No additional context."
                })
                
                # Extract answer from response
                answer = self._extract_answer(response)
                
                reasoning_paths.append({
                    "reasoning": response,
                    "answer": answer,
                    "temperature": temperature
                })
                
            except Exception as e:
                print(f"Warning: Sample {i+1} failed: {str(e)}")
                continue
        
        return reasoning_paths
    
    def _extract_answer(self, reasoning: str) -> str:
        """Extract the final answer from reasoning text."""
        # Look for common answer patterns
        patterns = [
            r"final answer is:?\s*(.+?)(?:\n|$)",
            r"answer is:?\s*(.+?)(?:\n|$)",
            r"therefore,?\s+(?:the answer is )?\s*(.+?)(?:\n|$)",
            r"thus,?\s+(.+?)(?:\n|$)",
            r"so,?\s+(?:the answer is )?\s*(.+?)(?:\n|$)",
        ]
        
        reasoning_lower = reasoning.lower()
        for pattern in patterns:
            match = re.search(pattern, reasoning_lower, re.IGNORECASE)
            if match:
                answer = match.group(1).strip()
                # Clean up the answer
                answer = re.sub(r'[*_]', '', answer)  # Remove markdown
                return answer
        
        # If no pattern matches, take last non-empty line
        lines = [line.strip() for line in reasoning.split('\n') if line.strip()]
        return lines[-1] if lines else reasoning[:100]
    
    def aggregate_by_majority_vote(self, reasoning_paths: List[Dict[str, str]]) -> Dict[str, Any]:
        """
        Aggregate answers using majority voting.
        
        Args:
            reasoning_paths: List of reasoning paths with answers
            
        Returns:
            Dictionary with final answer and aggregation statistics
        """
        answers = [path['answer'] for path in reasoning_paths]
        
        # Normalize answers for comparison (lowercase, strip punctuation)
        normalized_answers = []
        for ans in answers:
            normalized = ans.lower().strip().rstrip('.')
            normalized_answers.append(normalized)
        
        # Count occurrences
        answer_counts = Counter(normalized_answers)
        most_common_answer, count = answer_counts.most_common(1)[0]
        
        # Find original form of the answer
        original_answer = next(
            (answers[i] for i, norm in enumerate(normalized_answers) 
             if norm == most_common_answer),
            most_common_answer
        )
        
        return {
            "final_answer": original_answer,
            "confidence": count / len(answers),
            "vote_count": count,
            "total_samples": len(answers),
            "all_answers": answer_counts,
            "agreement_rate": count / len(answers)
        }
    
    def aggregate_by_consistency_score(self, reasoning_paths: List[Dict[str, str]], 
                                       question: str) -> Dict[str, Any]:
        """
        Aggregate answers using consistency scoring via LLM.
        
        Args:
            reasoning_paths: List of reasoning paths with answers
            question: Original question
            
        Returns:
            Dictionary with final answer and consistency scores
        """
        # Use LLM to evaluate consistency of each answer with others
        answers_text = "\n".join([
            f"{i+1}. {path['answer']}" 
            for i, path in enumerate(reasoning_paths)
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are evaluating multiple answers to the same question.
Identify which answer is most consistent with the majority of other answers.
Consider semantic similarity, not just exact matches.

Respond with the number (1, 2, 3, etc.) of the most consistent answer."""),
            ("human", """Question: {question}

Answers:
{answers}

Most consistent answer number:""")
        ])
        
        llm = self._create_llm(0.3)  # Low temperature for consistent evaluation
        chain = prompt | llm | StrOutputParser()
        
        try:
            response = chain.invoke({
                "question": question,
                "answers": answers_text
            })
            
            # Extract number
            match = re.search(r'\d+', response)
            if match:
                index = int(match.group()) - 1
                if 0 <= index  Dict[str, Any]:
        """
        Solve a question using self-consistency.
        
        Args:
            question: The question to answer
            context: Optional context
            method: Aggregation method ('majority_vote' or 'consistency_score')
            
        Returns:
            Dictionary with solution and metadata
        """
        print(f"\nQuestion: {question}\n")
        print(f"Generating {self.num_samples} reasoning paths...\n")
        
        # Generate multiple reasoning paths
        reasoning_paths = self.generate_reasoning_paths(question, context)
        
        if not reasoning_paths:
            return {
                "question": question,
                "final_answer": "Unable to generate reasoning paths",
                "error": "All samples failed"
            }
        
        print(f"Generated {len(reasoning_paths)} paths\n")
        
        # Show individual reasoning paths
        for i, path in enumerate(reasoning_paths, 1):
            print(f"Path {i} (temp={path['temperature']:.2f}):")
            print(f"  Answer: {path['answer']}")
            print()
        
        # Aggregate results
        if method == "majority_vote":
            result = self.aggregate_by_majority_vote(reasoning_paths)
        else:
            result = self.aggregate_by_consistency_score(reasoning_paths, question)
        
        result["question"] = question
        result["reasoning_paths"] = len(reasoning_paths)
        
        return result


def demonstrate_self_consistency():
    """Demonstrates the Self-Consistency pattern."""
    
    print("=" * 80)
    print("PATTERN 005: Self-Consistency")
    print("=" * 80)
    print()
    print("Self-Consistency improves answer reliability by:")
    print("1. Generating multiple independent reasoning paths")
    print("2. Sampling with different temperatures for diversity")
    print("3. Aggregating answers via majority voting or consistency scoring")
    print()
    
    # Create agent
    agent = SelfConsistencyAgent(num_samples=5)
    
    # Test cases
    test_cases = [
        {
            "question": "If a train travels 120 miles in 2 hours, then travels 180 miles in 3 hours, what is its average speed for the entire journey?",
            "context": "",
            "method": "majority_vote"
        },
        {
            "question": "A bat and ball cost $1.10 together. The bat costs $1.00 more than the ball. How much does the ball cost?",
            "context": "",
            "method": "majority_vote"
        },
        {
            "question": "Is it ethical for AI to make medical diagnoses?",
            "context": "Consider patient safety, accuracy, and doctor-patient relationships.",
            "method": "consistency_score"
        }
    ]
    
    for i, test_case in enumerate(test_cases, 1):
        print(f"\n{'=' * 80}")
        print(f"Example {i}: {test_case['method'].replace('_', ' ').title()}")
        print('=' * 80)
        
        try:
            result = agent.solve_with_self_consistency(
                question=test_case['question'],
                context=test_case['context'],
                method=test_case['method']
            )
            
            print(f"\n{'=' * 80}")
            print("FINAL RESULT")
            print('=' * 80)
            print(f"\nFinal Answer: {result['final_answer']}")
            
            if 'confidence' in result:
                print(f"Confidence: {result['confidence']:.1%}")
                print(f"Agreement Rate: {result['agreement_rate']:.1%}")
                print(f"Vote Distribution: {dict(result['all_answers'])}")
            
            print(f"Reasoning Paths: {result['reasoning_paths']}/{agent.num_samples}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("SELF-CONSISTENCY PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Techniques Demonstrated:")
    print("1. Multiple Sampling: Generated 5 diverse reasoning paths")
    print("2. Temperature Variation: Used different temperatures for diversity")
    print("3. Majority Voting: Aggregated answers by frequency")
    print("4. Consistency Scoring: LLM-based semantic consistency evaluation")
    print("5. Confidence Metrics: Agreement rate and vote distribution")
    print()
    print("Benefits:")
    print("- Improved reliability through multiple reasoning paths")
    print("- Reduced impact of outlier/hallucinated answers")
    print("- Confidence estimation via agreement metrics")
    print("- Works with any reasoning task")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Structures reasoning prompts")
    print("- StrOutputParser: Parses LLM outputs")
    print("- Temperature variation: Creates diverse samples")
    print("- Multiple chain invocations: Parallel reasoning paths")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_self_consistency()

    

Agentic Design Pattern: Self Consistency

Pattern 005: Self Consistency – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Self Consistency pattern implementation.

---

Pattern Description

Pattern 005: Self-Consistency

Description: Samples multiple reasoning paths for the same question and selects the most consistent answer through majority voting or consistency scoring. This approach improves reliability by generating diverse solutions and aggregating them.

Key Concepts: – Multiple Sampling: Generate several independent reasoning chains – Diverse Reasoning: Use temperature or prompt variations – Aggregation: Majority voting or consistency analysis – Answer Selection: Choose most frequent or highest confidence answer

Use Cases: – Math problems with multiple solution methods – Questions with ambiguous reasoning paths – Improving reliability of complex reasoning – Reducing model hallucinations

LangChain Implementation: Multiple chain invocations with temperature variation and result aggregation.

---

File Statistics

• Total Lines: 384
• Classes: 1
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 005: Self-Consistency

Description:
    Samples multiple reasoning paths for the same question and selects
    the most consistent answer through majority voting or consistency scoring.
    This approach improves reliability by generating diverse solutions and
    aggregating them.

Key Concepts:
    - Multiple Sampling: Generate several independent reasoning chains
    - Diverse Reasoning: Use temperature or prompt variations
    - Aggregation: Majority voting or consistency analysis
    - Answer Selection: Choose most frequent or highest confidence answer

Use Cases:
    - Math problems with multiple solution methods
    - Questions with ambiguous reasoning paths
    - Improving reliability of complex reasoning
    - Reducing model hallucinations

LangChain Implementation:
    Multiple chain invocations with temperature variation and result aggregation.
"""

Purpose: This module implements the Self Consistency pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.Counter
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import re
import typing.Any
import typing.Dict
import typing.List
import typing.Tuple

Purpose: Core Python functionality: – collections.Counter: Specialized container datatypes – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – re: Regular expression operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: SelfConsistencyAgent

Lines 40-283

Purpose:

Agent that uses self-consistency for reliable reasoning.

Methods:

__init__(self, model_name, num_samples)

Lines 43-53

Purpose:

Initialize the Self-Consistency agent.

Args:
    model_name: Name of the OpenAI model to use
    num_samples: Number of reasoning paths to sample

Parameters: – model_name: Language model instance – num_samples: Num samples

_create_llm(self, temperature)

Lines 55-61

Purpose:

Create an LLM instance with specified temperature.

Parameters: – temperature: Sampling temperature (0-1)

Returns: ChatOpenAI

generate_reasoning_paths(self, question, context)

Lines 63-112

Purpose:

Generate multiple diverse reasoning paths for a question.

Args:
    question: The question to answer
    context: Optional context information

Returns:
    List of dictionaries with 'reasoning' and 'answer' keys

Parameters: – question: Question to answer – context: Context

Returns: <ast.Subscript object at 0x000001A06D696050>

_extract_answer(self, reasoning)

Lines 114-136

Purpose:

Extract the final answer from reasoning text.

Parameters: – reasoning: Reasoning

Returns: str

aggregate_by_majority_vote(self, reasoning_paths)

Lines 138-174

Purpose:

Aggregate answers using majority voting.

Args:
    reasoning_paths: List of reasoning paths with answers

Returns:
    Dictionary with final answer and aggregation statistics

Parameters: – reasoning_paths: Reasoning paths

Returns: <ast.Subscript object at 0x000001A06D168F10>

aggregate_by_consistency_score(self, reasoning_paths, question)

Lines 176-238

Purpose:

Aggregate answers using consistency scoring via LLM.

Args:
    reasoning_paths: List of reasoning paths with answers
    question: Original question

Returns:
    Dictionary with final answer and consistency scores

Parameters: – reasoning_paths: Reasoning paths – question: Question to answer

Returns: <ast.Subscript object at 0x000001A06D158310>

solve_with_self_consistency(self, question, context, method)

Lines 240-283

Purpose:

Solve a question using self-consistency.

Args:
    question: The question to answer
    context: Optional context
    method: Aggregation method ('majority_vote' or 'consistency_score')

Returns:
    Dictionary with solution and metadata

Parameters: – question: Question to answer – context: Context – method: Method

Returns: <ast.Subscript object at 0x000001A06D156DD0>

---

Functions

demonstrate_self_consistency()

Lines 286-373

Purpose:

Demonstrates the Self-Consistency pattern.

---

Key Concepts Summary

Pattern Overview

The Self Consistency pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 004
• Pattern 006
• Other Reasoning Patterns (001-011)

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Self Consistency pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 005_self_consistency.py
Total Lines: 384
Pattern Category: AI Agent Design Pattern #005

---

Generated from actual implementation code
For educational and reference purposes