Tag: llm

  • Agentic Design Pattern: Sandboxing

    
        """
    Pattern 048: Sandboxing
    
    Description:
        The Sandboxing pattern provides isolated execution environments for agent actions,
        ensuring security and containment when running potentially unsafe operations like
        code execution, external API calls, or file system access. This pattern prevents
        malicious or buggy code from affecting the host system.
    
    Components:
        1. Sandbox Manager: Creates and manages isolated environments
        2. Resource Limiter: Enforces CPU, memory, time, and I/O constraints
        3. Permission Controller: Manages what operations are allowed
        4. Execution Monitor: Tracks sandbox activity and detects violations
        5. Result Validator: Validates outputs before returning to main system
    
    Use Cases:
        - Code execution agents running untrusted code
        - Web scraping with unknown sites
        - Data processing with untrusted inputs
        - Plugin systems with third-party code
        - Educational platforms running student code
        - Multi-tenant systems with isolation requirements
    
    LangChain Implementation:
        Uses custom runnables with execution constraints, subprocess isolation,
        and comprehensive monitoring to create secure execution environments.
    """
    
    import os
    import sys
    import subprocess
    import tempfile
    import json
    import time
    from typing import Dict, Any, List, Optional, Set, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from pathlib import Path
    
    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 SandboxType(Enum):
        """Types of sandbox environments"""
        SUBPROCESS = "subprocess"  # Run in separate process
        RESTRICTED_EVAL = "restricted_eval"  # Restricted Python eval
        CONTAINER = "container"  # Docker/container (simulated)
        VIRTUAL_ENV = "virtual_env"  # Virtual environment
        MEMORY_ONLY = "memory_only"  # No disk access
    
    
    class PermissionLevel(Enum):
        """Permission levels for sandbox"""
        MINIMAL = "minimal"  # Basic operations only
        RESTRICTED = "restricted"  # Limited file/network access
        STANDARD = "standard"  # Normal operations with monitoring
        ELEVATED = "elevated"  # More permissions (rare)
    
    
    class ViolationType(Enum):
        """Types of security violations"""
        RESOURCE_EXCEEDED = "resource_exceeded"
        PERMISSION_DENIED = "permission_denied"
        TIMEOUT = "timeout"
        UNSAFE_OPERATION = "unsafe_operation"
        OUTPUT_TOO_LARGE = "output_too_large"
        ESCAPE_ATTEMPT = "escape_attempt"
    
    
    @dataclass
    class ResourceLimits:
        """Resource limits for sandbox execution"""
        max_execution_time_seconds: float = 5.0
        max_memory_mb: int = 100
        max_output_size_bytes: int = 10_000
        max_file_size_bytes: int = 1_000_000
        max_network_requests: int = 0  # 0 = no network
        max_processes: int = 1
        
        def to_dict(self) -> Dict[str, Any]:
            return {
                "max_execution_time": f"{self.max_execution_time_seconds}s",
                "max_memory": f"{self.max_memory_mb}MB",
                "max_output_size": f"{self.max_output_size_bytes} bytes",
                "max_file_size": f"{self.max_file_size_bytes} bytes",
                "max_network_requests": self.max_network_requests,
                "max_processes": self.max_processes
            }
    
    
    @dataclass
    class SecurityViolation:
        """Record of a security violation"""
        violation_type: ViolationType
        severity: float  # 0.0-1.0
        description: str
        timestamp: datetime = field(default_factory=datetime.now)
        blocked: bool = True
        
        def to_dict(self) -> Dict[str, Any]:
            return {
                "type": self.violation_type.value,
                "severity": self.severity,
                "description": self.description,
                "timestamp": self.timestamp.isoformat(),
                "blocked": self.blocked
            }
    
    
    @dataclass
    class SandboxResult:
        """Result from sandbox execution"""
        success: bool
        output: Optional[str]
        error: Optional[str]
        execution_time_seconds: float
        resources_used: Dict[str, Any]
        violations: List[SecurityViolation]
        permission_level: PermissionLevel
        sandbox_type: SandboxType
        
        @property
        def was_safe(self) -> bool:
            """Check if execution was safe (no violations)"""
            return len(self.violations) == 0
        
        @property
        def high_severity_violations(self) -> List[SecurityViolation]:
            """Get violations with severity >= 0.7"""
            return [v for v in self.violations if v.severity >= 0.7]
        
        def to_dict(self) -> Dict[str, Any]:
            return {
                "success": self.success,
                "output": self.output,
                "error": self.error,
                "execution_time": f"{self.execution_time_seconds:.3f}s",
                "resources_used": self.resources_used,
                "violations": [v.to_dict() for v in self.violations],
                "was_safe": self.was_safe,
                "permission_level": self.permission_level.value,
                "sandbox_type": self.sandbox_type.value
            }
    
    
    class SandboxAgent:
        """
        Agent that executes operations in isolated sandbox environments.
        
        This implementation provides:
        1. Multiple sandbox types (subprocess, restricted eval, memory-only)
        2. Comprehensive resource limits (time, memory, output size)
        3. Permission-based access control
        4. Violation detection and blocking
        5. Safe code generation and execution
        """
        
        # Dangerous operations to detect
        UNSAFE_OPERATIONS = {
            "import os", "import sys", "import subprocess", "import socket",
            "open(", "exec(", "eval(", "compile(", "__import__",
            "file(", "input(", "raw_input(", "execfile(",
            "os.", "sys.", "subprocess.", "socket.",
            "__builtins__", "globals(", "locals(", "vars(",
            "delattr", "setattr", "getattr",
        }
        
        # Safe built-in functions for restricted eval
        SAFE_BUILTINS = {
            "abs", "all", "any", "ascii", "bin", "bool", "chr",
            "dict", "divmod", "enumerate", "filter", "float", "format",
            "hex", "int", "isinstance", "issubclass", "iter", "len",
            "list", "map", "max", "min", "oct", "ord", "pow", "print",
            "range", "repr", "reversed", "round", "set", "slice",
            "sorted", "str", "sum", "tuple", "type", "zip",
        }
        
        def __init__(
            self,
            model_name: str = "gpt-3.5-turbo",
            temperature: float = 0.3,
            default_sandbox_type: SandboxType = SandboxType.SUBPROCESS,
            default_permission_level: PermissionLevel = PermissionLevel.RESTRICTED,
            default_limits: Optional[ResourceLimits] = None
        ):
            self.llm = ChatOpenAI(model_name=model_name, temperature=temperature)
            self.default_sandbox_type = default_sandbox_type
            self.default_permission_level = default_permission_level
            self.default_limits = default_limits or ResourceLimits()
            
            # Statistics tracking
            self.total_executions = 0
            self.successful_executions = 0
            self.blocked_executions = 0
            self.total_violations = 0
        
        def _check_code_safety(self, code: str) -> List[SecurityViolation]:
            """Check code for unsafe operations before execution"""
            violations = []
            
            # Check for dangerous operations
            code_lower = code.lower()
            for unsafe_op in self.UNSAFE_OPERATIONS:
                if unsafe_op.lower() in code_lower:
                    violations.append(SecurityViolation(
                        violation_type=ViolationType.UNSAFE_OPERATION,
                        severity=0.9,
                        description=f"Detected potentially unsafe operation: {unsafe_op}",
                        blocked=True
                    ))
            
            return violations
        
        def _create_restricted_globals(self) -> Dict[str, Any]:
            """Create restricted global namespace for safe eval"""
            safe_globals = {"__builtins__": {}}
            
            # Add only safe builtins
            for builtin_name in self.SAFE_BUILTINS:
                if hasattr(__builtins__, builtin_name):
                    safe_globals["__builtins__"][builtin_name] = getattr(__builtins__, builtin_name)
            
            return safe_globals
        
        def _execute_in_subprocess(
            self,
            code: str,
            limits: ResourceLimits
        ) -> SandboxResult:
            """Execute code in separate subprocess with resource limits"""
            start_time = time.time()
            violations = []
            
            # Create temporary file for code
            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
                f.write(code)
                temp_file = f.name
            
            try:
                # Execute in subprocess with timeout
                result = subprocess.run(
                    [sys.executable, temp_file],
                    capture_output=True,
                    text=True,
                    timeout=limits.max_execution_time_seconds
                )
                
                execution_time = time.time() - start_time
                
                # Check output size
                output = result.stdout
                if len(output) > limits.max_output_size_bytes:
                    violations.append(SecurityViolation(
                        violation_type=ViolationType.OUTPUT_TOO_LARGE,
                        severity=0.6,
                        description=f"Output size {len(output)} exceeds limit {limits.max_output_size_bytes}",
                        blocked=True
                    ))
                    output = output[:limits.max_output_size_bytes] + "\n[OUTPUT TRUNCATED]"
                
                success = result.returncode == 0
                error = result.stderr if result.stderr else None
                
                return SandboxResult(
                    success=success,
                    output=output,
                    error=error,
                    execution_time_seconds=execution_time,
                    resources_used={"subprocess": True},
                    violations=violations,
                    permission_level=self.default_permission_level,
                    sandbox_type=SandboxType.SUBPROCESS
                )
                
            except subprocess.TimeoutExpired:
                execution_time = time.time() - start_time
                violations.append(SecurityViolation(
                    violation_type=ViolationType.TIMEOUT,
                    severity=0.8,
                    description=f"Execution exceeded timeout of {limits.max_execution_time_seconds}s",
                    blocked=True
                ))
                
                return SandboxResult(
                    success=False,
                    output=None,
                    error="Execution timeout",
                    execution_time_seconds=execution_time,
                    resources_used={"subprocess": True, "timeout": True},
                    violations=violations,
                    permission_level=self.default_permission_level,
                    sandbox_type=SandboxType.SUBPROCESS
                )
                
            except Exception as e:
                execution_time = time.time() - start_time
                return SandboxResult(
                    success=False,
                    output=None,
                    error=str(e),
                    execution_time_seconds=execution_time,
                    resources_used={"subprocess": True, "error": True},
                    violations=violations,
                    permission_level=self.default_permission_level,
                    sandbox_type=SandboxType.SUBPROCESS
                )
                
            finally:
                # Clean up temp file
                try:
                    os.unlink(temp_file)
                except:
                    pass
        
        def _execute_restricted_eval(
            self,
            code: str,
            limits: ResourceLimits
        ) -> SandboxResult:
            """Execute code using restricted eval with limited builtins"""
            start_time = time.time()
            violations = []
            
            try:
                # Create restricted globals
                safe_globals = self._create_restricted_globals()
                safe_locals = {}
                
                # Capture output
                from io import StringIO
                import sys
                
                old_stdout = sys.stdout
                sys.stdout = StringIO()
                
                try:
                    # Execute with timeout (simplified - production would use threading)
                    exec(code, safe_globals, safe_locals)
                    output = sys.stdout.getvalue()
                    
                    execution_time = time.time() - start_time
                    
                    # Check timeout
                    if execution_time > limits.max_execution_time_seconds:
                        violations.append(SecurityViolation(
                            violation_type=ViolationType.TIMEOUT,
                            severity=0.7,
                            description=f"Execution time {execution_time:.2f}s exceeded limit",
                            blocked=False  # Already executed
                        ))
                    
                    # Check output size
                    if len(output) > limits.max_output_size_bytes:
                        violations.append(SecurityViolation(
                            violation_type=ViolationType.OUTPUT_TOO_LARGE,
                            severity=0.6,
                            description=f"Output size {len(output)} exceeds limit",
                            blocked=False
                        ))
                        output = output[:limits.max_output_size_bytes] + "\n[OUTPUT TRUNCATED]"
                    
                    return SandboxResult(
                        success=True,
                        output=output,
                        error=None,
                        execution_time_seconds=execution_time,
                        resources_used={"restricted_eval": True},
                        violations=violations,
                        permission_level=PermissionLevel.MINIMAL,
                        sandbox_type=SandboxType.RESTRICTED_EVAL
                    )
                    
                finally:
                    sys.stdout = old_stdout
                    
            except Exception as e:
                execution_time = time.time() - start_time
                return SandboxResult(
                    success=False,
                    output=None,
                    error=str(e),
                    execution_time_seconds=execution_time,
                    resources_used={"restricted_eval": True, "error": True},
                    violations=violations,
                    permission_level=PermissionLevel.MINIMAL,
                    sandbox_type=SandboxType.RESTRICTED_EVAL
                )
        
        def generate_safe_code(self, task: str) -> str:
            """Generate safe code for a task using LLM"""
            prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a code generator that creates SAFE Python code.
                
    Rules:
    - DO NOT import os, sys, subprocess, socket, or other system modules
    - DO NOT use file operations (open, read, write)
    - DO NOT use exec, eval, compile, or __import__
    - DO NOT access __builtins__, globals(), or locals()
    - USE only safe operations: math, string manipulation, basic data structures
    - KEEP code simple and focused on the task
    - INCLUDE print statements to show results
    
    Generate safe, self-contained Python code that runs without external dependencies."""),
                ("user", "Task: {task}\n\nGenerate safe Python code:")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            code = chain.invoke({"task": task})
            
            # Clean up code (remove markdown formatting if present)
            if "```python" in code:
                code = code.split("```python")[1].split("```")[0].strip()
            elif "```" in code:
                code = code.split("```")[1].split("```")[0].strip()
            
            return code
        
        def execute_safely(
            self,
            code: str,
            sandbox_type: Optional[SandboxType] = None,
            limits: Optional[ResourceLimits] = None,
            check_safety: bool = True
        ) -> SandboxResult:
            """Execute code safely in a sandbox with monitoring"""
            self.total_executions += 1
            
            # Use defaults if not specified
            sandbox_type = sandbox_type or self.default_sandbox_type
            limits = limits or self.default_limits
            
            # Pre-execution safety check
            violations = []
            if check_safety:
                violations = self._check_code_safety(code)
                
                # Block if high-severity violations found
                if any(v.severity >= 0.8 for v in violations):
                    self.blocked_executions += 1
                    self.total_violations += len(violations)
                    return SandboxResult(
                        success=False,
                        output=None,
                        error="Execution blocked due to security violations",
                        execution_time_seconds=0.0,
                        resources_used={},
                        violations=violations,
                        permission_level=self.default_permission_level,
                        sandbox_type=sandbox_type
                    )
            
            # Execute in appropriate sandbox
            if sandbox_type == SandboxType.SUBPROCESS:
                result = self._execute_in_subprocess(code, limits)
            elif sandbox_type == SandboxType.RESTRICTED_EVAL:
                result = self._execute_restricted_eval(code, limits)
            else:
                # Default to restricted eval for other types
                result = self._execute_restricted_eval(code, limits)
            
            # Merge pre-execution violations with execution violations
            result.violations = violations + result.violations
            
            # Update statistics
            if result.success:
                self.successful_executions += 1
            self.total_violations += len(result.violations)
            
            return result
        
        def execute_task_safely(
            self,
            task: str,
            sandbox_type: Optional[SandboxType] = None,
            limits: Optional[ResourceLimits] = None
        ) -> Dict[str, Any]:
            """Generate code for task and execute it safely"""
            # Generate safe code
            code = self.generate_safe_code(task)
            
            # Execute in sandbox
            result = self.execute_safely(code, sandbox_type, limits)
            
            return {
                "task": task,
                "generated_code": code,
                "execution_result": result.to_dict()
            }
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get sandbox usage statistics"""
            success_rate = (
                self.successful_executions / self.total_executions
                if self.total_executions > 0
                else 0.0
            )
            
            block_rate = (
                self.blocked_executions / self.total_executions
                if self.total_executions > 0
                else 0.0
            )
            
            avg_violations = (
                self.total_violations / self.total_executions
                if self.total_executions > 0
                else 0.0
            )
            
            return {
                "total_executions": self.total_executions,
                "successful_executions": self.successful_executions,
                "blocked_executions": self.blocked_executions,
                "success_rate": f"{success_rate:.1%}",
                "block_rate": f"{block_rate:.1%}",
                "total_violations": self.total_violations,
                "avg_violations_per_execution": f"{avg_violations:.2f}"
            }
    
    
    def demonstrate_sandboxing():
        """Demonstrate sandbox pattern with various scenarios"""
        
        print("=" * 80)
        print("PATTERN 048: SANDBOXING DEMONSTRATION")
        print("=" * 80)
        print("\nDemonstrating isolated execution environments for secure agent operations\n")
        
        # Create sandbox agent
        agent = SandboxAgent(
            default_sandbox_type=SandboxType.SUBPROCESS,
            default_limits=ResourceLimits(
                max_execution_time_seconds=3.0,
                max_memory_mb=100,
                max_output_size_bytes=5000
            )
        )
        
        # Test 1: Safe code execution
        print("\n" + "=" * 80)
        print("TEST 1: Safe Code Execution")
        print("=" * 80)
        print("\nTask: Calculate fibonacci numbers")
        
        safe_code = """
    # Calculate first 10 fibonacci numbers
    def fibonacci(n):
        if n <= 1:
            return n
        a, b = 0, 1
        for _ in range(n - 1):
            a, b = b, a + b
        return b
    
    print("First 10 Fibonacci numbers:")
    for i in range(10):
        print(f"F({i}) = {fibonacci(i)}")
    """
        
        result1 = agent.execute_safely(safe_code)
        print(f"\n✓ Execution successful: {result1.success}")
        print(f"  Execution time: {result1.execution_time_seconds:.3f}s")
        print(f"  Violations: {len(result1.violations)}")
        print(f"  Sandbox type: {result1.sandbox_type.value}")
        print(f"\nOutput:\n{result1.output}")
        
        # Test 2: Unsafe code detection
        print("\n" + "=" * 80)
        print("TEST 2: Unsafe Code Detection")
        print("=" * 80)
        print("\nAttempting to execute code with unsafe operations...")
        
        unsafe_code = """
    import os
    import subprocess
    
    # Attempt to list directory (should be blocked)
    print("Files:", os.listdir('.'))
    
    # Attempt to run command (should be blocked)
    subprocess.run(['echo', 'hack'])
    """
        
        result2 = agent.execute_safely(unsafe_code)
        print(f"\n✗ Execution blocked: {not result2.success}")
        print(f"  Violations detected: {len(result2.violations)}")
        
        for i, violation in enumerate(result2.violations, 1):
            print(f"\n  Violation {i}:")
            print(f"    Type: {violation.violation_type.value}")
            print(f"    Severity: {violation.severity:.1f}")
            print(f"    Description: {violation.description}")
            print(f"    Blocked: {violation.blocked}")
        
        # Test 3: Resource limit enforcement
        print("\n" + "=" * 80)
        print("TEST 3: Resource Limit Enforcement")
        print("=" * 80)
        print("\nTesting timeout protection...")
        
        timeout_code = """
    # Infinite loop (should timeout)
    import time
    count = 0
    while True:
        count += 1
        if count % 1000000 == 0:
            print(f"Count: {count}")
    """
        
        result3 = agent.execute_safely(
            timeout_code,
            limits=ResourceLimits(max_execution_time_seconds=1.0)
        )
        
        print(f"\n✓ Timeout protection activated: {not result3.success}")
        print(f"  Error: {result3.error}")
        print(f"  Execution time: {result3.execution_time_seconds:.3f}s")
        print(f"  Violations: {len(result3.violations)}")
        
        if result3.violations:
            for violation in result3.violations:
                print(f"\n  Timeout violation:")
                print(f"    Type: {violation.violation_type.value}")
                print(f"    Description: {violation.description}")
        
        # Test 4: Safe task execution with LLM
        print("\n" + "=" * 80)
        print("TEST 4: LLM-Generated Safe Code Execution")
        print("=" * 80)
        print("\nGenerating and executing code for task: 'Calculate prime numbers up to 50'")
        
        task_result = agent.execute_task_safely(
            "Calculate all prime numbers up to 50 and print them",
            sandbox_type=SandboxType.RESTRICTED_EVAL
        )
        
        print(f"\nGenerated code:")
        print("-" * 60)
        print(task_result["generated_code"])
        print("-" * 60)
        
        exec_result = task_result["execution_result"]
        print(f"\n✓ Execution successful: {exec_result['success']}")
        print(f"  Execution time: {exec_result['execution_time']}")
        print(f"  Safe: {exec_result['was_safe']}")
        print(f"  Sandbox: {exec_result['sandbox_type']}")
        print(f"\nOutput:\n{exec_result['output']}")
        
        # Test 5: Multiple sandbox types comparison
        print("\n" + "=" * 80)
        print("TEST 5: Comparing Sandbox Types")
        print("=" * 80)
        
        test_code = """
    # Simple calculation
    result = sum(range(1, 101))
    print(f"Sum of 1-100: {result}")
    print(f"Average: {result / 100}")
    """
        
        print("\nExecuting same code in different sandboxes...")
        
        for sandbox_type in [SandboxType.SUBPROCESS, SandboxType.RESTRICTED_EVAL]:
            result = agent.execute_safely(test_code, sandbox_type=sandbox_type)
            print(f"\n{sandbox_type.value.upper()}:")
            print(f"  Success: {result.success}")
            print(f"  Time: {result.execution_time_seconds:.4f}s")
            print(f"  Violations: {len(result.violations)}")
            print(f"  Output: {result.output.strip()}")
        
        # Statistics
        print("\n" + "=" * 80)
        print("SANDBOX STATISTICS")
        print("=" * 80)
        
        stats = agent.get_statistics()
        print(f"\nTotal executions: {stats['total_executions']}")
        print(f"Successful: {stats['successful_executions']}")
        print(f"Blocked: {stats['blocked_executions']}")
        print(f"Success rate: {stats['success_rate']}")
        print(f"Block rate: {stats['block_rate']}")
        print(f"Total violations: {stats['total_violations']}")
        print(f"Avg violations per execution: {stats['avg_violations_per_execution']}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SANDBOXING PATTERN SUMMARY")
        print("=" * 80)
        print("""
    Key Benefits:
    1. Security Isolation: Prevents malicious code from affecting host system
    2. Resource Control: Enforces limits on CPU, memory, time, and I/O
    3. Permission Management: Controls what operations are allowed
    4. Violation Detection: Identifies and blocks unsafe operations
    5. Multiple Strategies: Supports different isolation levels and methods
    
    Implementation Features:
    1. Pre-execution safety checks with pattern matching
    2. Multiple sandbox types (subprocess, restricted eval, containers)
    3. Comprehensive resource limits and enforcement
    4. Real-time violation detection and blocking
    5. Detailed execution monitoring and statistics
    
    Use Cases:
    - Code execution agents running user-provided code
    - Educational platforms executing student submissions
    - Data processing pipelines with untrusted inputs
    - Plugin systems loading third-party code
    - Multi-tenant systems requiring isolation
    - Testing and development environments
    
    Configuration Options:
    1. Sandbox Type: subprocess, restricted_eval, container, virtual_env
    2. Permission Level: minimal, restricted, standard, elevated
    3. Resource Limits: time, memory, output size, file size, network
    4. Safety Checks: pattern matching, execution monitoring
    5. Violation Handling: block, warn, log, escalate
    
    Best Practices:
    1. Use least privilege principle for permissions
    2. Set appropriate resource limits for use case
    3. Monitor violations and adjust policies
    4. Validate outputs before returning to main system
    5. Log all sandbox activity for audit trail
    6. Test sandbox escape scenarios
    7. Keep sandbox environments minimal and isolated
    8. Regular security reviews of allowed operations
    
    Production Considerations:
    - Container orchestration for true isolation (Docker, Kubernetes)
    - Network isolation and firewall rules
    - Encrypted communication between sandbox and host
    - Regular security updates to sandbox environments
    - Monitoring and alerting for unusual patterns
    - Compliance with data protection regulations
    - Backup and recovery procedures
    - Performance optimization for high-volume execution
    
    Comparison with Related Patterns:
    - vs. Guardrails: Sandboxing provides execution isolation vs input/output filtering
    - vs. Circuit Breaker: Prevents system damage vs prevents cascading failures
    - vs. Rate Limiting: Controls resource usage vs controls request frequency
    - vs. Defensive Generation: Isolates execution vs filters content
    
    The Sandboxing pattern is essential for any system that executes potentially
    untrusted code or operations, providing defense-in-depth security through
    multiple layers of isolation, monitoring, and enforcement.
    """)
    
    
    if __name__ == "__main__":
        demonstrate_sandboxing()
    
        

    Agentic Design Pattern: Sandboxing

    Pattern 048: Sandboxing – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 048: Sandboxing

    Description: The Sandboxing pattern provides isolated execution environments for agent actions, ensuring security and containment when running potentially unsafe operations like code execution, external API calls, or file system access. This pattern prevents malicious or buggy code from affecting the host system.

    Components: 1. Sandbox Manager: Creates and manages isolated environments 2. Resource Limiter: Enforces CPU, memory, time, and I/O constraints 3. Permission Controller: Manages what operations are allowed 4. Execution Monitor: Tracks sandbox activity and detects violations 5. Result Validator: Validates outputs before returning to main system

    Use Cases: – Code execution agents running untrusted code – Web scraping with unknown sites – Data processing with untrusted inputs – Plugin systems with third-party code – Educational platforms running student code – Multi-tenant systems with isolation requirements

    LangChain Implementation: Uses custom runnables with execution constraints, subprocess isolation, and comprehensive monitoring to create secure execution environments.


    File Statistics

    • Total Lines: 761
    • Classes: 7
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 048: Sandboxing
    
    Description:
        The Sandboxing pattern provides isolated execution environments for agent actions,
        ensuring security and containment when running potentially unsafe operations like
        code execution, external API calls, or file system access. This pattern prevents
        malicious or buggy code from affecting the host system.
    
    Components:
        1. Sandbox Manager: Creates and manages isolated environments
        2. Resource Limiter: Enforces CPU, memory, time, and I/O constraints
        3. Permission Controller: Manages what operations are allowed
        4. Execution Monitor: Tracks sandbox activity and detects violations
        5. Result Validator: Validates outputs before returning to main system
    
    Use Cases:
        - Code execution agents running untrusted code
        - Web scraping with unknown sites
        - Data processing with untrusted inputs
        - Plugin systems with third-party code
        - Educational platforms running student code
        - Multi-tenant systems with isolation requirements
    
    LangChain Implementation:
        Uses custom runnables with execution constraints, subprocess isolation,
        and comprehensive monitoring to create secure execution environments.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    import json
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import pathlib.Path
    import sys
    import time
    import typing.Any
    import typing.Callable
    import typing.Dict
    import typing.List
    import typing.Optional
    import typing.Set
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – pathlib.Path: Object-oriented filesystem paths – time: Time-related functions – 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) – 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 io.StringIO
    from subprocess
    from tempfile
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: SandboxType

    Lines 50-56

    Inherits from: Enum

    Purpose:

    Types of sandbox environments
    

    Class: PermissionLevel

    Lines 59-64

    Inherits from: Enum

    Purpose:

    Permission levels for sandbox
    

    Class: ViolationType

    Lines 67-74

    Inherits from: Enum

    Purpose:

    Types of security violations
    

    Class: ResourceLimits

    Lines 78-95

    Purpose:

    Resource limits for sandbox execution
    

    Methods:

    to_dict(self)

    Lines 87-95

    Purpose: To Dict operation.

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D7651D0>


    Class: SecurityViolation

    Lines 99-114

    Purpose:

    Record of a security violation
    

    Methods:

    to_dict(self)

    Lines 107-114

    Purpose: To Dict operation.

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D77F810>


    Class: SandboxResult

    Lines 118-150

    Purpose:

    Result from sandbox execution
    

    Methods:

    was_safe(self)

    Lines 130-132

    Decorators: property

    Purpose:

    Check if execution was safe (no violations)
    

    Parameters:

    Returns: bool

    high_severity_violations(self)

    Lines 135-137

    Decorators: property

    Purpose:

    Get violations with severity >= 0.7
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D77CD10>

    to_dict(self)

    Lines 139-150

    Purpose: To Dict operation.

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D668DD0>


    Class: SandboxAgent

    Lines 153-525

    Purpose:

    Agent that executes operations in isolated sandbox environments.
    
    This implementation provides:
    1. Multiple sandbox types (subprocess, restricted eval, memory-only)
    2. Comprehensive resource limits (time, memory, output size)
    3. Permission-based access control
    4. Violation detection and blocking
    5. Safe code generation and execution
    

    Methods:

    __init__(self, model_name, temperature, default_sandbox_type, default_permission_level, default_limits)

    Lines 185-202

    Purpose: Initialize the class instance with required parameters.

    Parameters: – model_name: Language model instance – temperature: Sampling temperature (0-1) – default_sandbox_type: Default sandbox type – default_permission_level: Default permission level – default_limits: Default limits

    _check_code_safety(self, code)

    Lines 204-219

    Purpose:

    Check code for unsafe operations before execution
    

    Parameters: – code: Code

    Returns: <ast.Subscript object at 0x000001A06D702DD0>

    _create_restricted_globals(self)

    Lines 221-230

    Purpose:

    Create restricted global namespace for safe eval
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D702F90>

    _execute_in_subprocess(self, code, limits)

    Lines 232-320

    Purpose:

    Execute code in separate subprocess with resource limits
    

    Parameters: – code: Code – limits: Limits

    Returns: SandboxResult

    _execute_restricted_eval(self, code, limits)

    Lines 322-394

    Purpose:

    Execute code using restricted eval with limited builtins
    

    Parameters: – code: Code – limits: Limits

    Returns: SandboxResult

    generate_safe_code(self, task)

    Lines 396-423

    Purpose:

    Generate safe code for a task using LLM
    

    Parameters: – task: Task

    Returns: str

    execute_safely(self, code, sandbox_type, limits, check_safety)

    Lines 425-476

    Purpose:

    Execute code safely in a sandbox with monitoring
    

    Parameters: – code: Code – sandbox_type: Sandbox type – limits: Limits – check_safety: Check safety

    Returns: SandboxResult

    execute_task_safely(self, task, sandbox_type, limits)

    Lines 478-495

    Purpose:

    Generate code for task and execute it safely
    

    Parameters: – task: Task – sandbox_type: Sandbox type – limits: Limits

    Returns: <ast.Subscript object at 0x000001A06D60EBD0>

    get_statistics(self)

    Lines 497-525

    Purpose:

    Get sandbox usage statistics
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D157A90>


    Functions

    demonstrate_sandboxing()

    Lines 528-756

    Purpose:

    Demonstrate sandbox pattern with various scenarios
    

    Key Concepts Summary

    Pattern Overview

    The Sandboxing 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 047
    • Pattern 049

    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 Sandboxing 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: 048_sandboxing.py
    Total Lines: 761
    Pattern Category: AI Agent Design Pattern #048


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Circuit Breaker

    
        """
    Pattern 047: Circuit Breaker
    
    Description:
        Circuit Breaker prevents cascading failures by monitoring error rates and
        stopping agent execution when thresholds are exceeded. Implements three states
        (CLOSED, OPEN, HALF_OPEN) with automatic recovery attempts and exponential backoff.
    
    Components:
        - Error Monitor: Tracks failure rates
        - State Manager: Controls circuit states
        - Threshold Checker: Detects when to open circuit
        - Recovery Manager: Attempts to close circuit
        - Metrics Collector: Records performance data
    
    Use Cases:
        - Production systems with external dependencies
        - Cost control and budget management
        - Preventing API overload
        - Protecting downstream services
        - Resource exhaustion prevention
        - Cascading failure prevention
    
    LangChain Implementation:
        Uses state-based circuit breaking with error rate monitoring, automatic
        recovery, and configurable thresholds to ensure system stability.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime, timedelta
    from enum import Enum
    import time
    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 CircuitState(Enum):
        """States of the circuit breaker."""
        CLOSED = "closed"  # Normal operation
        OPEN = "open"  # Blocking calls
        HALF_OPEN = "half_open"  # Testing recovery
    
    
    @dataclass
    class CircuitMetrics:
        """Metrics for circuit breaker monitoring."""
        total_requests: int = 0
        successful_requests: int = 0
        failed_requests: int = 0
        rejected_requests: int = 0
        last_failure_time: Optional[datetime] = None
        last_success_time: Optional[datetime] = None
        state_changes: List[tuple] = field(default_factory=list)
        
        @property
        def error_rate(self) -> float:
            """Calculate current error rate."""
            if self.total_requests == 0:
                return 0.0
            return self.failed_requests / self.total_requests
        
        @property
        def success_rate(self) -> float:
            """Calculate success rate."""
            if self.total_requests == 0:
                return 0.0
            return self.successful_requests / self.total_requests
    
    
    @dataclass
    class CircuitBreakerResult:
        """Result of a circuit breaker execution."""
        success: bool
        response: Optional[str]
        error: Optional[str]
        circuit_state: CircuitState
        was_rejected: bool
        execution_time_ms: float
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class CircuitBreakerAgent:
        """
        Agent with circuit breaker pattern for resilience.
        
        Features:
        - Three-state circuit breaker
        - Error rate monitoring
        - Automatic recovery attempts
        - Configurable thresholds
        - Exponential backoff
        - Comprehensive metrics
        """
        
        def __init__(
            self,
            failure_threshold: float = 0.5,  # 50% error rate
            success_threshold: int = 3,  # Successful calls to close
            timeout_seconds: float = 60.0,  # Time before trying half-open
            window_size: int = 10,  # Recent calls to consider
            temperature: float = 0.7
        ):
            self.failure_threshold = failure_threshold
            self.success_threshold = success_threshold
            self.timeout_seconds = timeout_seconds
            self.window_size = window_size
            
            # Circuit state
            self.state = CircuitState.CLOSED
            self.state_changed_at = datetime.now()
            self.consecutive_successes = 0
            
            # Metrics
            self.metrics = CircuitMetrics()
            self.recent_results: List[bool] = []  # True = success, False = failure
            
            # LLM
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            
            # Prompt
            self.prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant."),
                ("user", "{query}")
            ])
        
        def _record_state_change(self, new_state: CircuitState):
            """Record state transition."""
            old_state = self.state
            self.state = new_state
            self.state_changed_at = datetime.now()
            self.metrics.state_changes.append((
                old_state.value,
                new_state.value,
                datetime.now()
            ))
        
        def _check_should_open(self) -> bool:
            """Check if circuit should open based on recent failures."""
            if len(self.recent_results) = self.failure_threshold
        
        def _check_should_attempt_reset(self) -> bool:
            """Check if enough time has passed to try half-open."""
            if self.state != CircuitState.OPEN:
                return False
            
            time_since_open = datetime.now() - self.state_changed_at
            return time_since_open.total_seconds() >= self.timeout_seconds
        
        def _execute_request(self, query: str) -> tuple[bool, Optional[str], Optional[str]]:
            """
            Execute the actual request.
            
            Returns:
                Tuple of (success, response, error)
            """
            try:
                chain = self.prompt | self.llm | StrOutputParser()
                response = chain.invoke({"query": query})
                
                # Validate response
                if response and len(response.strip()) > 0:
                    return True, response, None
                else:
                    return False, None, "Empty response"
            
            except Exception as e:
                return False, None, str(e)
        
        def execute(self, query: str) -> CircuitBreakerResult:
            """
            Execute query through circuit breaker.
            
            Circuit states:
            - CLOSED: Normal operation, requests go through
            - OPEN: Blocking requests, fast-fail
            - HALF_OPEN: Testing recovery, limited requests
            
            Args:
                query: User query
                
            Returns:
                CircuitBreakerResult with execution details
            """
            start_time = datetime.now()
            
            # Check if should attempt reset
            if self._check_should_attempt_reset():
                self._record_state_change(CircuitState.HALF_OPEN)
                self.consecutive_successes = 0
            
            # Handle OPEN state (reject immediately)
            if self.state == CircuitState.OPEN:
                self.metrics.rejected_requests += 1
                
                time_until_retry = self.timeout_seconds - (
                    datetime.now() - self.state_changed_at
                ).total_seconds()
                
                return CircuitBreakerResult(
                    success=False,
                    response=None,
                    error=f"Circuit breaker OPEN. Retry in {time_until_retry:.1f}s",
                    circuit_state=self.state,
                    was_rejected=True,
                    execution_time_ms=0.0
                )
            
            # Execute request (CLOSED or HALF_OPEN)
            success, response, error = self._execute_request(query)
            execution_time = (datetime.now() - start_time).total_seconds() * 1000
            
            # Update metrics
            self.metrics.total_requests += 1
            
            if success:
                self.metrics.successful_requests += 1
                self.metrics.last_success_time = datetime.now()
                self.recent_results.append(True)
                self.consecutive_successes += 1
                
                # Handle state transitions on success
                if self.state == CircuitState.HALF_OPEN:
                    if self.consecutive_successes >= self.success_threshold:
                        # Recovered! Close circuit
                        self._record_state_change(CircuitState.CLOSED)
                        self.recent_results.clear()  # Reset history
            else:
                self.metrics.failed_requests += 1
                self.metrics.last_failure_time = datetime.now()
                self.recent_results.append(False)
                self.consecutive_successes = 0
                
                # Handle state transitions on failure
                if self.state == CircuitState.HALF_OPEN:
                    # Failed during recovery, open again
                    self._record_state_change(CircuitState.OPEN)
                elif self.state == CircuitState.CLOSED:
                    # Check if should open
                    if self._check_should_open():
                        self._record_state_change(CircuitState.OPEN)
            
            # Trim recent results to window size
            if len(self.recent_results) > self.window_size * 2:
                self.recent_results = self.recent_results[-self.window_size:]
            
            return CircuitBreakerResult(
                success=success,
                response=response,
                error=error,
                circuit_state=self.state,
                was_rejected=False,
                execution_time_ms=execution_time
            )
        
        def get_circuit_status(self) -> Dict[str, Any]:
            """Get current circuit breaker status."""
            time_in_current_state = (datetime.now() - self.state_changed_at).total_seconds()
            
            return {
                "state": self.state.value,
                "time_in_state_seconds": time_in_current_state,
                "consecutive_successes": self.consecutive_successes,
                "recent_window_size": len(self.recent_results),
                "recent_error_rate": self.metrics.error_rate if self.recent_results else 0.0,
                "metrics": {
                    "total_requests": self.metrics.total_requests,
                    "successful": self.metrics.successful_requests,
                    "failed": self.metrics.failed_requests,
                    "rejected": self.metrics.rejected_requests,
                    "overall_error_rate": self.metrics.error_rate,
                    "success_rate": self.metrics.success_rate,
                    "state_changes": len(self.metrics.state_changes)
                }
            }
        
        def reset(self):
            """Manually reset circuit breaker to CLOSED state."""
            self._record_state_change(CircuitState.CLOSED)
            self.consecutive_successes = 0
            self.recent_results.clear()
    
    
    def demonstrate_circuit_breaker():
        """
        Demonstrates circuit breaker pattern for resilience.
        """
        print("=" * 80)
        print("CIRCUIT BREAKER DEMONSTRATION")
        print("=" * 80)
        
        # Create circuit breaker agent
        agent = CircuitBreakerAgent(
            failure_threshold=0.5,  # Open after 50% errors
            success_threshold=3,  # Need 3 successes to close
            timeout_seconds=5.0,  # 5 seconds before retry
            window_size=6
        )
        
        # Test 1: Normal operation (CLOSED state)
        print("\n" + "=" * 80)
        print("Test 1: Normal Operation (CLOSED State)")
        print("=" * 80)
        
        queries = [
            "What is 2+2?",
            "Name three colors",
            "What is the capital of France?"
        ]
        
        print("\nExecuting successful requests...")
        for query in queries:
            result = agent.execute(query)
            status_icon = "✓" if result.success else "✗"
            print(f"{status_icon} Query: {query[:40]}...")
            print(f"   State: {result.circuit_state.value}, Success: {result.success}")
            if result.response:
                print(f"   Response: {result.response[:60]}...")
        
        status = agent.get_circuit_status()
        print(f"\n[Circuit Status]")
        print(f"State: {status['state']}")
        print(f"Total Requests: {status['metrics']['total_requests']}")
        print(f"Success Rate: {status['metrics']['success_rate']:.1%}")
        
        # Test 2: Simulating failures (trigger OPEN state)
        print("\n" + "=" * 80)
        print("Test 2: Failure Cascade (Triggering OPEN State)")
        print("=" * 80)
        
        print("\nSimulating failures to trigger circuit breaker...")
        
        # Create a scenario with failures by using agent that might fail
        # For demo, we'll manually inject failures
        print("\nInjecting failure pattern:")
        for i in range(4):
            # Simulate failure by recording it
            agent.recent_results.append(False)
            agent.metrics.failed_requests += 1
            agent.metrics.total_requests += 1
            print(f"  Failure {i+1} recorded")
        
        # Check if circuit opened
        if agent._check_should_open():
            agent._record_state_change(CircuitState.OPEN)
            print("\n⚠️  Circuit breaker OPENED due to high failure rate!")
        
        status = agent.get_circuit_status()
        print(f"\n[Circuit Status After Failures]")
        print(f"State: {status['state']}")
        print(f"Error Rate: {status['metrics']['overall_error_rate']:.1%}")
        print(f"Failed: {status['metrics']['failed']}")
        print(f"Successful: {status['metrics']['successful']}")
        
        # Test 3: Rejected requests in OPEN state
        print("\n" + "=" * 80)
        print("Test 3: Requests Rejected (OPEN State)")
        print("=" * 80)
        
        print("\nAttempting requests while circuit is OPEN...")
        for i in range(3):
            result = agent.execute(f"Test query {i+1}")
            print(f"\nRequest {i+1}:")
            print(f"  Was Rejected: {result.was_rejected}")
            print(f"  State: {result.circuit_state.value}")
            if result.error:
                print(f"  Error: {result.error}")
        
        status = agent.get_circuit_status()
        print(f"\n[Circuit Status]")
        print(f"Rejected Requests: {status['metrics']['rejected']}")
        
        # Test 4: Recovery (HALF_OPEN -> CLOSED)
        print("\n" + "=" * 80)
        print("Test 4: Circuit Recovery (HALF_OPEN → CLOSED)")
        print("=" * 80)
        
        print("\nWaiting for recovery timeout...")
        print(f"Timeout configured: {agent.timeout_seconds}s")
        
        # Manually trigger half-open for demo
        agent._record_state_change(CircuitState.HALF_OPEN)
        print(f"✓ Circuit now in HALF_OPEN state")
        
        print("\nAttempting recovery with successful requests...")
        recovery_queries = [
            "What is 10 * 10?",
            "Name a fruit",
            "What color is the sky?"
        ]
        
        for i, query in enumerate(recovery_queries, 1):
            result = agent.execute(query)
            print(f"\nRecovery Attempt {i}:")
            print(f"  Query: {query}")
            print(f"  Success: {result.success}")
            print(f"  State: {result.circuit_state.value}")
            print(f"  Consecutive Successes: {agent.consecutive_successes}")
            
            if result.circuit_state == CircuitState.CLOSED:
                print(f"\n✓ Circuit CLOSED! Successfully recovered after {i} requests")
                break
        
        # Final status
        print("\n" + "=" * 80)
        print("Final Circuit Breaker Status")
        print("=" * 80)
        
        status = agent.get_circuit_status()
        
        print(f"\nCurrent State: {status['state'].upper()}")
        print(f"Time in Current State: {status['time_in_state_seconds']:.1f}s")
        
        print(f"\n[Lifetime Metrics]")
        print(f"Total Requests: {status['metrics']['total_requests']}")
        print(f"Successful: {status['metrics']['successful']}")
        print(f"Failed: {status['metrics']['failed']}")
        print(f"Rejected: {status['metrics']['rejected']}")
        print(f"Success Rate: {status['metrics']['success_rate']:.1%}")
        print(f"Error Rate: {status['metrics']['overall_error_rate']:.1%}")
        print(f"State Changes: {status['metrics']['state_changes']}")
        
        if agent.metrics.state_changes:
            print(f"\n[State Transition History]")
            for from_state, to_state, timestamp in agent.metrics.state_changes[-5:]:
                print(f"  {from_state} → {to_state} at {timestamp.strftime('%H:%M:%S')}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Circuit Breaker provides:
    ✓ Cascading failure prevention
    ✓ Automatic error detection
    ✓ Fast-fail mechanism
    ✓ Automatic recovery attempts
    ✓ Resource protection
    ✓ System stability
    
    This pattern excels at:
    - Production systems
    - External API integration
    - Cost control
    - Resource protection
    - High-availability systems
    - Preventing overload
    
    Circuit states:
    1. CLOSED (Normal)
       - All requests processed
       - Monitoring error rates
       - Opens on threshold breach
    
    2. OPEN (Blocking)
       - Requests rejected immediately
       - Fast-fail responses
       - Waits for timeout period
    
    3. HALF_OPEN (Testing)
       - Limited requests allowed
       - Testing if service recovered
       - Success → CLOSED
       - Failure → OPEN
    
    State transitions:
    CLOSED → OPEN:
      - Error rate exceeds threshold
      - Too many recent failures
      - Protects system
    
    OPEN → HALF_OPEN:
      - Timeout period elapsed
      - Ready to test recovery
      - Automatic retry
    
    HALF_OPEN → CLOSED:
      - Consecutive successes reached
      - Service recovered
      - Normal operation resumed
    
    HALF_OPEN → OPEN:
      - Recovery attempt failed
      - Service still unhealthy
      - Continue blocking
    
    Configuration parameters:
    - failure_threshold: Error rate to open (0.5 = 50%)
    - success_threshold: Successes needed to close (3)
    - timeout_seconds: Time before retry (60s)
    - window_size: Recent requests to track (10)
    
    Benefits:
    - Stability: Prevent cascades
    - Performance: Fast-fail
    - Recovery: Automatic healing
    - Visibility: State tracking
    - Protection: Resource saving
    - Predictability: Clear states
    
    Monitoring metrics:
    - Total requests
    - Success/failure counts
    - Error rates
    - Rejection counts
    - State transitions
    - Time in each state
    
    Use Circuit Breaker when:
    - Calling external services
    - Cost control needed
    - Preventing overload
    - Protecting resources
    - High-availability required
    - Cascading failures possible
    
    Best practices:
    - Set appropriate thresholds
    - Monitor state transitions
    - Log rejected requests
    - Alert on OPEN state
    - Test recovery logic
    - Configure timeouts wisely
    
    Comparison with other patterns:
    - vs Retry: Stops vs continues trying
    - vs Fallback: Blocks vs alternative
    - vs Rate Limiting: Dynamic vs fixed limits
    """)
    
    
    if __name__ == "__main__":
        demonstrate_circuit_breaker()
    
        

    Agentic Design Pattern: Circuit Breaker

    Pattern 047: Circuit Breaker – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 047: Circuit Breaker

    Description: Circuit Breaker prevents cascading failures by monitoring error rates and stopping agent execution when thresholds are exceeded. Implements three states (CLOSED, OPEN, HALF_OPEN) with automatic recovery attempts and exponential backoff.

    Components: – Error Monitor: Tracks failure rates – State Manager: Controls circuit states – Threshold Checker: Detects when to open circuit – Recovery Manager: Attempts to close circuit – Metrics Collector: Records performance data

    Use Cases: – Production systems with external dependencies – Cost control and budget management – Preventing API overload – Protecting downstream services – Resource exhaustion prevention – Cascading failure prevention

    LangChain Implementation: Uses state-based circuit breaking with error rate monitoring, automatic recovery, and configurable thresholds to ensure system stability.


    File Statistics

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

    Module Documentation

    """
    Pattern 047: Circuit Breaker
    
    Description:
        Circuit Breaker prevents cascading failures by monitoring error rates and
        stopping agent execution when thresholds are exceeded. Implements three states
        (CLOSED, OPEN, HALF_OPEN) with automatic recovery attempts and exponential backoff.
    
    Components:
        - Error Monitor: Tracks failure rates
        - State Manager: Controls circuit states
        - Threshold Checker: Detects when to open circuit
        - Recovery Manager: Attempts to close circuit
        - Metrics Collector: Records performance data
    
    Use Cases:
        - Production systems with external dependencies
        - Cost control and budget management
        - Preventing API overload
        - Protecting downstream services
        - Resource exhaustion prevention
        - Cascading failure prevention
    
    LangChain Implementation:
        Uses state-based circuit breaking with error rate monitoring, automatic
        recovery, and configurable thresholds to ensure system stability.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import datetime.timedelta
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import time
    import typing.Any
    import typing.Callable
    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 – datetime.datetime: Date and time handling – datetime.timedelta: Date and time handling – 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 – time: Time-related functions – 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
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: CircuitState

    Lines 44-48

    Inherits from: Enum

    Purpose:

    States of the circuit breaker.
    

    Class: CircuitMetrics

    Lines 52-74

    Purpose:

    Metrics for circuit breaker monitoring.
    

    Methods:

    error_rate(self)

    Lines 63-67

    Decorators: property

    Purpose:

    Calculate current error rate.
    

    Parameters:

    Returns: float

    success_rate(self)

    Lines 70-74

    Decorators: property

    Purpose:

    Calculate success rate.
    

    Parameters:

    Returns: float


    Class: CircuitBreakerResult

    Lines 78-86

    Purpose:

    Result of a circuit breaker execution.
    

    Class: CircuitBreakerAgent

    Lines 89-295

    Purpose:

    Agent with circuit breaker pattern for resilience.
    
    Features:
    - Three-state circuit breaker
    - Error rate monitoring
    - Automatic recovery attempts
    - Configurable thresholds
    - Exponential backoff
    - Comprehensive metrics
    

    Methods:

    __init__(self, failure_threshold, success_threshold, timeout_seconds, window_size, temperature)

    Lines 102-131

    Purpose: Initialize the class instance with required parameters.

    Parameters: – failure_threshold: Failure threshold – success_threshold: Success threshold – timeout_seconds: Timeout seconds – window_size: Window size – temperature: Sampling temperature (0-1)

    _record_state_change(self, new_state)

    Lines 133-142

    Purpose:

    Record state transition.
    

    Parameters: – new_state: New state

    _check_should_open(self)

    Lines 144-154

    Purpose:

    Check if circuit should open based on recent failures.
    

    Parameters:

    Returns: bool

    _check_should_attempt_reset(self)

    Lines 156-162

    Purpose:

    Check if enough time has passed to try half-open.
    

    Parameters:

    Returns: bool

    _execute_request(self, query)

    Lines 164-182

    Purpose:

    Execute the actual request.
    
    Returns:
        Tuple of (success, response, error)
    

    Parameters: – query: Input query or question

    Returns: <ast.Subscript object at 0x000001A06D628490>

    execute(self, query)

    Lines 184-268

    Purpose:

    Execute query through circuit breaker.
    
    Circuit states:
    - CLOSED: Normal operation, requests go through
    - OPEN: Blocking requests, fast-fail
    - HALF_OPEN: Testing recovery, limited requests
    
    Args:
        query: User query
    
    Returns:
        CircuitBreakerResult with execution details
    

    Parameters: – query: Input query or question

    Returns: CircuitBreakerResult

    get_circuit_status(self)

    Lines 270-289

    Purpose:

    Get current circuit breaker status.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D15FD10>

    reset(self)

    Lines 291-295

    Purpose:

    Manually reset circuit breaker to CLOSED state.
    

    Parameters:


    Functions

    demonstrate_circuit_breaker()

    Lines 298-543

    Purpose:

    Demonstrates circuit breaker pattern for resilience.
    

    Key Concepts Summary

    Pattern Overview

    The Circuit Breaker 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 046
    • Pattern 048

    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 Circuit Breaker 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: 047_circuit_breaker.py
    Total Lines: 548
    Pattern Category: AI Agent Design Pattern #047


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Fallback Degradation

    
        """
    Pattern 046: Fallback/Graceful Degradation
    
    Description:
        Fallback/Graceful Degradation enables agents to maintain functionality when
        primary approaches fail by using alternative strategies at multiple levels.
        Implements cascading fallback mechanisms from optimal to minimal functionality,
        with human escalation as the final safety net.
    
    Components:
        - Primary Handler: Optimal approach
        - Secondary Handlers: Fallback alternatives
        - Degradation Manager: Coordinates fallback levels
        - Error Classifier: Determines fallback strategy
        - Escalation Manager: Human intervention coordination
    
    Use Cases:
        - Production systems requiring high availability
        - Critical applications with zero downtime needs
        - Multi-model deployments
        - API reliability improvements
        - Fault-tolerant systems
        - Customer-facing applications
    
    LangChain Implementation:
        Uses tiered fallback chains with automatic degradation, error handling,
        and escalation to ensure continuous service availability.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    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 FallbackLevel(Enum):
        """Levels of fallback degradation."""
        PRIMARY = "primary"
        SECONDARY = "secondary"
        TERTIARY = "tertiary"
        MINIMAL = "minimal"
        HUMAN_ESCALATION = "human_escalation"
    
    
    class FailureType(Enum):
        """Types of failures that trigger fallbacks."""
        TIMEOUT = "timeout"
        API_ERROR = "api_error"
        RATE_LIMIT = "rate_limit"
        QUALITY_FAILURE = "quality_failure"
        UNAVAILABLE = "unavailable"
        UNKNOWN = "unknown"
    
    
    @dataclass
    class FallbackAttempt:
        """Record of a fallback attempt."""
        level: FallbackLevel
        strategy: str
        success: bool
        error: Optional[str] = None
        latency_ms: float = 0.0
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class FallbackResult:
        """Result of fallback execution."""
        query: str
        final_response: str
        final_level: FallbackLevel
        attempts: List[FallbackAttempt]
        total_fallbacks: int
        succeeded: bool
        total_latency_ms: float
    
    
    class FallbackAgent:
        """
        Agent with cascading fallback mechanisms for reliability.
        
        Features:
        - Multi-level fallback strategies
        - Automatic degradation
        - Error classification
        - Performance tracking
        - Human escalation
        """
        
        def __init__(
            self,
            max_fallback_attempts: int = 4,
            timeout_seconds: float = 10.0
        ):
            self.max_fallback_attempts = max_fallback_attempts
            self.timeout_seconds = timeout_seconds
            
            # Different LLM configurations for fallback levels
            self.primary_llm = ChatOpenAI(
                model="gpt-3.5-turbo",
                temperature=0.7,
                request_timeout=timeout_seconds
            )
            
            self.secondary_llm = ChatOpenAI(
                model="gpt-3.5-turbo",
                temperature=0.5,  # More conservative
                request_timeout=timeout_seconds / 2
            )
            
            # Prompts for different levels
            self.primary_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert AI assistant. Provide detailed, comprehensive responses."),
                ("user", "{query}")
            ])
            
            self.secondary_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Provide clear, concise responses."),
                ("user", "{query}")
            ])
            
            self.minimal_prompt = ChatPromptTemplate.from_messages([
                ("system", "Provide a brief, factual response."),
                ("user", "{query}")
            ])
            
            # Execution history
            self.executions: List[FallbackResult] = []
        
        def _primary_strategy(self, query: str) -> str:
            """
            Primary strategy: Full-featured response.
            
            Returns:
                Detailed response
            """
            chain = self.primary_prompt | self.primary_llm | StrOutputParser()
            return chain.invoke({"query": query})
        
        def _secondary_strategy(self, query: str) -> str:
            """
            Secondary strategy: Conservative response.
            
            Returns:
                Concise response
            """
            chain = self.secondary_prompt | self.secondary_llm | StrOutputParser()
            return chain.invoke({"query": query})
        
        def _tertiary_strategy(self, query: str) -> str:
            """
            Tertiary strategy: Minimal response.
            
            Returns:
                Brief response
            """
            chain = self.minimal_prompt | self.secondary_llm | StrOutputParser()
            return chain.invoke({"query": query})
        
        def _minimal_strategy(self, query: str) -> str:
            """
            Minimal strategy: Template-based response.
            
            Returns:
                Template response
            """
            return f"I understand you're asking about: '{query}'. Due to high system load, I can only provide a basic acknowledgment. Please try again shortly or contact support for detailed assistance."
        
        def _human_escalation(self, query: str) -> str:
            """
            Human escalation: Request human intervention.
            
            Returns:
                Escalation message
            """
            return f"This query requires human assistance: '{query}'. A human operator has been notified and will respond shortly. Reference ID: {datetime.now().strftime('%Y%m%d%H%M%S')}"
        
        def execute_with_fallback(
            self,
            query: str
        ) -> FallbackResult:
            """
            Execute query with automatic fallback on failures.
            
            Fallback cascade:
            1. PRIMARY: Full-featured response (GPT-3.5, detailed)
            2. SECONDARY: Conservative response (GPT-3.5, concise)
            3. TERTIARY: Minimal LLM response (brief)
            4. MINIMAL: Template response (no LLM)
            5. HUMAN_ESCALATION: Request human help
            
            Args:
                query: User query
                
            Returns:
                FallbackResult with execution details
            """
            attempts = []
            start_time = datetime.now()
            
            # Define fallback strategies
            strategies = [
                (FallbackLevel.PRIMARY, "Full-featured GPT-3.5", self._primary_strategy),
                (FallbackLevel.SECONDARY, "Conservative GPT-3.5", self._secondary_strategy),
                (FallbackLevel.TERTIARY, "Minimal LLM", self._tertiary_strategy),
                (FallbackLevel.MINIMAL, "Template response", self._minimal_strategy),
                (FallbackLevel.HUMAN_ESCALATION, "Human escalation", self._human_escalation),
            ]
            
            final_response = None
            final_level = None
            
            for level, strategy_name, strategy_func in strategies:
                attempt_start = datetime.now()
                
                try:
                    response = strategy_func(query)
                    attempt_latency = (datetime.now() - attempt_start).total_seconds() * 1000
                    
                    # Validate response
                    if response and len(response.strip()) > 0:
                        attempts.append(FallbackAttempt(
                            level=level,
                            strategy=strategy_name,
                            success=True,
                            latency_ms=attempt_latency
                        ))
                        
                        final_response = response
                        final_level = level
                        break  # Success!
                    else:
                        raise ValueError("Empty response")
                
                except Exception as e:
                    attempt_latency = (datetime.now() - attempt_start).total_seconds() * 1000
                    attempts.append(FallbackAttempt(
                        level=level,
                        strategy=strategy_name,
                        success=False,
                        error=str(e),
                        latency_ms=attempt_latency
                    ))
                    
                    # Continue to next fallback
                    continue
            
            # If all fallbacks failed (shouldn't happen with human escalation)
            if final_response is None:
                final_response = "System temporarily unavailable. Please try again later."
                final_level = FallbackLevel.HUMAN_ESCALATION
            
            total_latency = (datetime.now() - start_time).total_seconds() * 1000
            
            result = FallbackResult(
                query=query,
                final_response=final_response,
                final_level=final_level,
                attempts=attempts,
                total_fallbacks=len(attempts) - 1,  # -1 because first attempt isn't a fallback
                succeeded=final_response is not None,
                total_latency_ms=total_latency
            )
            
            self.executions.append(result)
            
            return result
        
        def get_reliability_statistics(self) -> Dict[str, Any]:
            """Get statistics about fallback usage and reliability."""
            if not self.executions:
                return {"total_executions": 0}
            
            total = len(self.executions)
            primary_success = sum(1 for e in self.executions if e.final_level == FallbackLevel.PRIMARY)
            required_fallback = sum(1 for e in self.executions if e.total_fallbacks > 0)
            
            level_usage = {}
            for level in FallbackLevel:
                count = sum(1 for e in self.executions if e.final_level == level)
                if count > 0:
                    level_usage[level.value] = count
            
            avg_latency = sum(e.total_latency_ms for e in self.executions) / total
            avg_fallbacks = sum(e.total_fallbacks for e in self.executions) / total
            
            return {
                "total_executions": total,
                "primary_success_count": primary_success,
                "primary_success_rate": primary_success / total,
                "required_fallback_count": required_fallback,
                "fallback_rate": required_fallback / total,
                "average_fallbacks_per_execution": avg_fallbacks,
                "average_latency_ms": avg_latency,
                "level_usage": level_usage
            }
    
    
    def demonstrate_fallback_graceful_degradation():
        """
        Demonstrates fallback and graceful degradation mechanisms.
        """
        print("=" * 80)
        print("FALLBACK/GRACEFUL DEGRADATION DEMONSTRATION")
        print("=" * 80)
        
        # Create fallback agent
        agent = FallbackAgent(
            max_fallback_attempts=4,
            timeout_seconds=30.0
        )
        
        # Test 1: Normal execution (should use primary)
        print("\n" + "=" * 80)
        print("Test 1: Normal Execution")
        print("=" * 80)
        
        query1 = "What are the benefits of regular exercise?"
        print(f"\nQuery: {query1}")
        
        result1 = agent.execute_with_fallback(query1)
        
        print("\n[Execution Summary]")
        print(f"Final Level: {result1.final_level.value}")
        print(f"Total Fallbacks: {result1.total_fallbacks}")
        print(f"Succeeded: {result1.succeeded}")
        print(f"Total Latency: {result1.total_latency_ms:.2f}ms")
        
        print("\n[Attempts]")
        for i, attempt in enumerate(result1.attempts, 1):
            status = "✓" if attempt.success else "✗"
            print(f"{i}. {status} {attempt.level.value} - {attempt.strategy}")
            print(f"   Latency: {attempt.latency_ms:.2f}ms")
            if attempt.error:
                print(f"   Error: {attempt.error[:100]}...")
        
        print("\n[Response Preview]")
        print(result1.final_response[:250] + "..." if len(result1.final_response) > 250 else result1.final_response)
        
        # Test 2: Multiple queries to show fallback patterns
        print("\n" + "=" * 80)
        print("Test 2: Multiple Executions")
        print("=" * 80)
        
        test_queries = [
            "How does photosynthesis work?",
            "What is machine learning?",
            "Explain the water cycle",
        ]
        
        print("\nExecuting multiple queries...")
        for query in test_queries:
            result = agent.execute_with_fallback(query)
            print(f"\nQuery: {query[:50]}...")
            print(f"  Level: {result.final_level.value}, Fallbacks: {result.total_fallbacks}, Latency: {result.total_latency_ms:.0f}ms")
        
        # Test 3: Simulated fallback scenario
        print("\n" + "=" * 80)
        print("Test 3: Fallback Cascade Visualization")
        print("=" * 80)
        
        query3 = "Explain quantum computing"
        print(f"\nQuery: {query3}")
        
        result3 = agent.execute_with_fallback(query3)
        
        print("\n[Fallback Cascade]")
        print("Level Hierarchy:")
        print("  1. PRIMARY      → Full-featured response")
        print("  2. SECONDARY    → Conservative response")
        print("  3. TERTIARY     → Minimal LLM response")
        print("  4. MINIMAL      → Template response")
        print("  5. ESCALATION   → Human intervention")
        
        print(f"\n Executed Level: {result3.final_level.value}")
        print(f"  Result: {'SUCCESS ✓' if result3.succeeded else 'FAILED ✗'}")
        
        # Show reliability statistics
        print("\n" + "=" * 80)
        print("Reliability Statistics")
        print("=" * 80)
        
        stats = agent.get_reliability_statistics()
        
        print(f"\nTotal Executions: {stats['total_executions']}")
        print(f"Primary Success Count: {stats['primary_success_count']}")
        print(f"Primary Success Rate: {stats['primary_success_rate']:.1%}")
        print(f"Required Fallback Count: {stats['required_fallback_count']}")
        print(f"Fallback Rate: {stats['fallback_rate']:.1%}")
        print(f"Average Fallbacks per Execution: {stats['average_fallbacks_per_execution']:.2f}")
        print(f"Average Latency: {stats['average_latency_ms']:.2f}ms")
        
        print("\n[Level Usage Distribution]")
        for level, count in sorted(stats['level_usage'].items(), key=lambda x: list(FallbackLevel).index(FallbackLevel(x[0]))):
            percentage = (count / stats['total_executions']) * 100
            bar = "█" * int(percentage / 2)
            print(f"  {level:<20} {bar} {count} ({percentage:.1f}%)")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Fallback/Graceful Degradation provides:
    ✓ Multi-level fallback strategies
    ✓ Automatic degradation
    ✓ Continuous availability
    ✓ Error resilience
    ✓ Performance tracking
    ✓ Human escalation
    
    This pattern excels at:
    - Production systems
    - High-availability applications
    - Critical services
    - API reliability
    - Fault tolerance
    - Customer-facing systems
    
    Fallback levels:
    1. PRIMARY: Optimal performance
       - Full-featured LLM
       - Detailed responses
       - Best quality
    
    2. SECONDARY: Conservative
       - Same LLM, adjusted parameters
       - Concise responses
       - Good quality
    
    3. TERTIARY: Minimal LLM
       - Brief responses
       - Basic quality
       - Fast execution
    
    4. MINIMAL: Template-based
       - No LLM required
       - Acknowledgment only
       - Immediate response
    
    5. HUMAN_ESCALATION: Last resort
       - Request human help
       - Provide reference ID
       - Guaranteed response
    
    Fallback triggers:
    - Timeouts: Slow responses
    - API Errors: Service failures
    - Rate Limits: Quota exceeded
    - Quality Failures: Poor outputs
    - Unavailability: Service down
    
    Benefits:
    - Reliability: Always available
    - Resilience: Handles failures
    - Performance: Optimizes resources
    - User Experience: No downtime
    - Flexibility: Multiple strategies
    - Graceful: Degraded but functional
    
    Degradation strategy:
    - Quality: High → Medium → Low
    - Latency: Accept longer waits at higher levels
    - Cost: Expensive → Cheap → Free
    - Complexity: Complex → Simple → Basic
    
    Use Fallback/Graceful Degradation when:
    - High availability required
    - Cannot afford downtime
    - Multiple failure modes possible
    - User experience critical
    - Production deployments
    - SLA requirements exist
    
    Configuration:
    - max_fallback_attempts: How many levels
    - timeout_seconds: Per-level timeout
    - fallback_strategies: Custom handlers
    - escalation_policy: Human notification
    
    Comparison with other patterns:
    - vs Circuit Breaker: Fallback vs prevention
    - vs Retry: Different strategies vs same attempt
    - vs Redundancy: Degradation vs duplication
    """)
    
    
    if __name__ == "__main__":
        demonstrate_fallback_graceful_degradation()
    
        

    Agentic Design Pattern: Fallback Degradation

    Pattern 046: Fallback Degradation – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 046: Fallback/Graceful Degradation

    Description: Fallback/Graceful Degradation enables agents to maintain functionality when primary approaches fail by using alternative strategies at multiple levels. Implements cascading fallback mechanisms from optimal to minimal functionality, with human escalation as the final safety net.

    Components: – Primary Handler: Optimal approach – Secondary Handlers: Fallback alternatives – Degradation Manager: Coordinates fallback levels – Error Classifier: Determines fallback strategy – Escalation Manager: Human intervention coordination

    Use Cases: – Production systems requiring high availability – Critical applications with zero downtime needs – Multi-model deployments – API reliability improvements – Fault-tolerant systems – Customer-facing applications

    LangChain Implementation: Uses tiered fallback chains with automatic degradation, error handling, and escalation to ensure continuous service availability.


    File Statistics

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

    Module Documentation

    """
    Pattern 046: Fallback/Graceful Degradation
    
    Description:
        Fallback/Graceful Degradation enables agents to maintain functionality when
        primary approaches fail by using alternative strategies at multiple levels.
        Implements cascading fallback mechanisms from optimal to minimal functionality,
        with human escalation as the final safety net.
    
    Components:
        - Primary Handler: Optimal approach
        - Secondary Handlers: Fallback alternatives
        - Degradation Manager: Coordinates fallback levels
        - Error Classifier: Determines fallback strategy
        - Escalation Manager: Human intervention coordination
    
    Use Cases:
        - Production systems requiring high availability
        - Critical applications with zero downtime needs
        - Multi-model deployments
        - API reliability improvements
        - Fault-tolerant systems
        - Customer-facing applications
    
    LangChain Implementation:
        Uses tiered fallback chains with automatic degradation, error handling,
        and escalation to ensure continuous service availability.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Callable
    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 – datetime.datetime: Date and time handling – 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.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
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: FallbackLevel

    Lines 44-50

    Inherits from: Enum

    Purpose:

    Levels of fallback degradation.
    

    Class: FailureType

    Lines 53-60

    Inherits from: Enum

    Purpose:

    Types of failures that trigger fallbacks.
    

    Class: FallbackAttempt

    Lines 64-71

    Purpose:

    Record of a fallback attempt.
    

    Class: FallbackResult

    Lines 75-83

    Purpose:

    Result of fallback execution.
    

    Class: FallbackAgent

    Lines 86-304

    Purpose:

    Agent with cascading fallback mechanisms for reliability.
    
    Features:
    - Multi-level fallback strategies
    - Automatic degradation
    - Error classification
    - Performance tracking
    - Human escalation
    

    Methods:

    __init__(self, max_fallback_attempts, timeout_seconds)

    Lines 98-136

    Purpose: Initialize the class instance with required parameters.

    Parameters: – max_fallback_attempts: Max fallback attempts – timeout_seconds: Timeout seconds

    _primary_strategy(self, query)

    Lines 138-146

    Purpose:

    Primary strategy: Full-featured response.
    
    Returns:
        Detailed response
    

    Parameters: – query: Input query or question

    Returns: str

    _secondary_strategy(self, query)

    Lines 148-156

    Purpose:

    Secondary strategy: Conservative response.
    
    Returns:
        Concise response
    

    Parameters: – query: Input query or question

    Returns: str

    _tertiary_strategy(self, query)

    Lines 158-166

    Purpose:

    Tertiary strategy: Minimal response.
    
    Returns:
        Brief response
    

    Parameters: – query: Input query or question

    Returns: str

    _minimal_strategy(self, query)

    Lines 168-175

    Purpose:

    Minimal strategy: Template-based response.
    
    Returns:
        Template response
    

    Parameters: – query: Input query or question

    Returns: str

    _human_escalation(self, query)

    Lines 177-184

    Purpose:

    Human escalation: Request human intervention.
    
    Returns:
        Escalation message
    

    Parameters: – query: Input query or question

    Returns: str

    execute_with_fallback(self, query)

    Lines 186-275

    Purpose:

    Execute query with automatic fallback on failures.
    
    Fallback cascade:
    1. PRIMARY: Full-featured response (GPT-3.5, detailed)
    2. SECONDARY: Conservative response (GPT-3.5, concise)
    3. TERTIARY: Minimal LLM response (brief)
    4. MINIMAL: Template response (no LLM)
    5. HUMAN_ESCALATION: Request human help
    
    Args:
        query: User query
    
    Returns:
        FallbackResult with execution details
    

    Parameters: – query: Input query or question

    Returns: FallbackResult

    get_reliability_statistics(self)

    Lines 277-304

    Purpose:

    Get statistics about fallback usage and reliability.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6F5190>


    Functions

    demonstrate_fallback_graceful_degradation()

    Lines 307-493

    Purpose:

    Demonstrates fallback and graceful degradation mechanisms.
    

    Key Concepts Summary

    Pattern Overview

    The Fallback Degradation 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 045
    • Pattern 047

    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 Fallback Degradation 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: 046_fallback_degradation.py
    Total Lines: 498
    Pattern Category: AI Agent Design Pattern #046


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Defensive Generation

    
        """
    Pattern 045: Defensive Generation
    
    Description:
        Defensive Generation enables agents to generate responses with built-in
        safety considerations including content filtering, bias mitigation, toxicity
        avoidance, and harmful output prevention. Implements multiple layers of
        safety checks before, during, and after generation.
    
    Components:
        - Safety Classifier: Detects harmful content
        - Bias Detector: Identifies biased language
        - Content Filter: Blocks inappropriate content
        - Safety Rewriter: Reformulates unsafe responses
        - Policy Enforcer: Ensures compliance with policies
    
    Use Cases:
        - Public-facing chatbots
        - Content moderation systems
        - Educational applications
        - Healthcare assistants
        - Customer service bots
        - Enterprise AI systems
    
    LangChain Implementation:
        Uses multi-layered safety checks with content classification, filtering,
        and reformulation to ensure generated content is safe and appropriate.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Set, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    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_dotenv()
    
    
    class SafetyLevel(Enum):
        """Safety assessment levels."""
        SAFE = "safe"
        CAUTION = "caution"
        UNSAFE = "unsafe"
        BLOCKED = "blocked"
    
    
    class SafetyCategory(Enum):
        """Categories of safety concerns."""
        TOXICITY = "toxicity"
        BIAS = "bias"
        HARMFUL_CONTENT = "harmful_content"
        PRIVACY_VIOLATION = "privacy_violation"
        INAPPROPRIATE = "inappropriate"
        MISINFORMATION = "misinformation"
        VIOLENCE = "violence"
        HATE_SPEECH = "hate_speech"
    
    
    @dataclass
    class SafetyViolation:
        """A detected safety violation."""
        category: SafetyCategory
        severity: float  # 0.0 to 1.0
        description: str
        location: Optional[str] = None
    
    
    @dataclass
    class SafetyAssessment:
        """Complete safety assessment of content."""
        content: str
        safety_level: SafetyLevel
        violations: List[SafetyViolation]
        is_safe: bool
        confidence: float
        recommendations: List[str]
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class DefensiveGenerationResult:
        """Result of defensive generation."""
        query: str
        raw_response: str
        final_response: str
        input_assessment: SafetyAssessment
        output_assessment: SafetyAssessment
        was_filtered: bool
        was_rewritten: bool
        safety_actions: List[str]
    
    
    class DefensiveGenerationAgent:
        """
        Agent that generates responses with comprehensive safety checks.
        
        Features:
        - Input safety validation
        - Output safety checking
        - Content filtering
        - Bias detection
        - Automatic rewriting
        - Policy enforcement
        """
        
        def __init__(
            self,
            safety_threshold: float = 0.7,
            enable_rewriting: bool = True,
            temperature: float = 0.5
        ):
            self.safety_threshold = safety_threshold
            self.enable_rewriting = enable_rewriting
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            
            # Blocked keywords and patterns
            self.blocked_keywords = {
                'violence', 'hate', 'discriminatory', 'harmful', 'illegal',
                'dangerous', 'explicit', 'inappropriate'
            }
            
            # Response generation prompt
            self.generation_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a helpful, safe, and responsible AI assistant.
    
    Guidelines:
    - Provide accurate, helpful information
    - Avoid harmful, biased, or inappropriate content
    - Respect user privacy
    - Decline inappropriate requests politely
    - Be objective and fair"""),
                ("user", "{query}")
            ])
            
            # Safety assessment prompt
            self.safety_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a content safety evaluator. Assess the safety of the content.
    
    Evaluate for:
    - Toxicity and hate speech
    - Bias and discrimination
    - Harmful or dangerous content
    - Privacy violations
    - Inappropriate content
    - Misinformation
    
    Provide:
    SAFETY_LEVEL: [safe/caution/unsafe/blocked]
    VIOLATIONS: [List any violations with category and severity]
    CONFIDENCE: [0.0-1.0]
    RECOMMENDATIONS: [Safety improvements if needed]"""),
                ("user", "Content to assess:\n{content}\n\nSafety assessment:")
            ])
            
            # Rewriting prompt
            self.rewrite_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a content safety editor. Rewrite content to be safe and appropriate.
    
    Instructions:
    - Remove harmful, biased, or inappropriate content
    - Maintain helpfulness and accuracy
    - Be respectful and inclusive
    - Preserve the core information
    - Decline if request is inherently unsafe
    
    Original issues:
    {violations}"""),
                ("user", "Original content:\n{content}\n\nProvide safe rewritten version:")
            ])
            
            # Generation history
            self.generations: List[DefensiveGenerationResult] = []
        
        def assess_safety(self, content: str) -> SafetyAssessment:
            """
            Assess safety of content.
            
            Returns:
                SafetyAssessment with violations and recommendations
            """
            # Quick keyword check
            content_lower = content.lower()
            keyword_violations = []
            for keyword in self.blocked_keywords:
                if keyword in content_lower:
                    keyword_violations.append(SafetyViolation(
                        category=SafetyCategory.INAPPROPRIATE,
                        severity=0.6,
                        description=f"Contains potentially problematic keyword: {keyword}"
                    ))
            
            # LLM-based assessment
            chain = self.safety_prompt | self.llm | StrOutputParser()
            result = chain.invoke({"content": content})
            
            # Parse assessment
            safety_level = SafetyLevel.SAFE
            violations = keyword_violations.copy()
            confidence = 0.8
            recommendations = []
            
            lines = result.split('\n')
            current_section = None
            
            for line in lines:
                line = line.strip()
                if line.startswith("SAFETY_LEVEL:"):
                    level_text = line.split(':')[1].strip().lower()
                    try:
                        safety_level = SafetyLevel(level_text)
                    except ValueError:
                        pass
                elif line.startswith("VIOLATIONS:"):
                    current_section = "violations"
                elif line.startswith("CONFIDENCE:"):
                    try:
                        confidence = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                    current_section = None
                elif line.startswith("RECOMMENDATIONS:"):
                    current_section = "recommendations"
                elif line and line.startswith('-') and current_section:
                    item = line[1:].strip()
                    if current_section == "violations":
                        # Parse violation (format: category - description)
                        if ' - ' in item:
                            cat_text, desc = item.split(' - ', 1)
                            try:
                                category = SafetyCategory(cat_text.strip().lower())
                            except ValueError:
                                category = SafetyCategory.INAPPROPRIATE
                            
                            violations.append(SafetyViolation(
                                category=category,
                                severity=0.7,
                                description=desc
                            ))
                    elif current_section == "recommendations":
                        recommendations.append(item)
            
            is_safe = safety_level in [SafetyLevel.SAFE, SafetyLevel.CAUTION]
            
            return SafetyAssessment(
                content=content,
                safety_level=safety_level,
                violations=violations,
                is_safe=is_safe,
                confidence=confidence,
                recommendations=recommendations
            )
        
        def rewrite_for_safety(
            self,
            content: str,
            violations: List[SafetyViolation]
        ) -> str:
            """
            Rewrite content to address safety violations.
            
            Returns:
                Rewritten safe content
            """
            if not violations:
                return content
            
            violations_text = "\n".join([
                f"- {v.category.value}: {v.description}"
                for v in violations
            ])
            
            chain = self.rewrite_prompt | self.llm | StrOutputParser()
            rewritten = chain.invoke({
                "content": content,
                "violations": violations_text
            })
            
            return rewritten
        
        def generate_defensively(
            self,
            query: str
        ) -> DefensiveGenerationResult:
            """
            Generate response with comprehensive safety checks.
            
            Process:
            1. Assess input safety
            2. Block if input is unsafe
            3. Generate response
            4. Assess output safety
            5. Rewrite if needed
            6. Return safest version
            
            Args:
                query: User query
                
            Returns:
                DefensiveGenerationResult with safety information
            """
            safety_actions = []
            
            # Step 1: Assess input safety
            input_assessment = self.assess_safety(query)
            
            # Step 2: Block unsafe inputs
            if input_assessment.safety_level == SafetyLevel.BLOCKED:
                safety_actions.append("Input blocked due to safety violations")
                
                blocked_response = "I apologize, but I cannot respond to that request as it violates safety guidelines. Please rephrase your question in a respectful and appropriate manner."
                
                output_assessment = SafetyAssessment(
                    content=blocked_response,
                    safety_level=SafetyLevel.SAFE,
                    violations=[],
                    is_safe=True,
                    confidence=1.0,
                    recommendations=[]
                )
                
                return DefensiveGenerationResult(
                    query=query,
                    raw_response="[BLOCKED]",
                    final_response=blocked_response,
                    input_assessment=input_assessment,
                    output_assessment=output_assessment,
                    was_filtered=True,
                    was_rewritten=False,
                    safety_actions=safety_actions
                )
            
            # Step 3: Generate response
            chain = self.generation_prompt | self.llm | StrOutputParser()
            raw_response = chain.invoke({"query": query})
            
            # Step 4: Assess output safety
            output_assessment = self.assess_safety(raw_response)
            
            final_response = raw_response
            was_filtered = False
            was_rewritten = False
            
            # Step 5: Handle unsafe outputs
            if not output_assessment.is_safe:
                safety_actions.append(f"Output safety violation: {output_assessment.safety_level.value}")
                
                if output_assessment.safety_level == SafetyLevel.BLOCKED:
                    # Block completely
                    final_response = "I apologize, but I cannot provide that information as it may be harmful or inappropriate."
                    was_filtered = True
                    safety_actions.append("Output blocked and replaced with safe message")
                
                elif self.enable_rewriting and output_assessment.violations:
                    # Try to rewrite
                    safety_actions.append("Attempting to rewrite for safety")
                    rewritten = self.rewrite_for_safety(raw_response, output_assessment.violations)
                    
                    # Assess rewritten version
                    rewritten_assessment = self.assess_safety(rewritten)
                    
                    if rewritten_assessment.is_safe:
                        final_response = rewritten
                        output_assessment = rewritten_assessment
                        was_rewritten = True
                        safety_actions.append("Successfully rewritten for safety")
                    else:
                        # Rewriting failed, block
                        final_response = "I apologize, but I cannot provide a safe response to that query."
                        was_filtered = True
                        safety_actions.append("Rewriting failed, output blocked")
            else:
                safety_actions.append("Output passed safety checks")
            
            result = DefensiveGenerationResult(
                query=query,
                raw_response=raw_response,
                final_response=final_response,
                input_assessment=input_assessment,
                output_assessment=output_assessment,
                was_filtered=was_filtered,
                was_rewritten=was_rewritten,
                safety_actions=safety_actions
            )
            
            self.generations.append(result)
            
            return result
        
        def get_safety_statistics(self) -> Dict[str, Any]:
            """Get statistics about safety interventions."""
            if not self.generations:
                return {"total_generations": 0}
            
            total = len(self.generations)
            filtered = sum(1 for g in self.generations if g.was_filtered)
            rewritten = sum(1 for g in self.generations if g.was_rewritten)
            
            # Violation categories
            violation_counts = {}
            for gen in self.generations:
                for violation in gen.output_assessment.violations:
                    cat = violation.category.value
                    violation_counts[cat] = violation_counts.get(cat, 0) + 1
            
            return {
                "total_generations": total,
                "filtered_count": filtered,
                "rewritten_count": rewritten,
                "filter_rate": filtered / total,
                "rewrite_rate": rewritten / total,
                "violation_breakdown": violation_counts
            }
    
    
    def demonstrate_defensive_generation():
        """
        Demonstrates defensive generation with safety checks.
        """
        print("=" * 80)
        print("DEFENSIVE GENERATION DEMONSTRATION")
        print("=" * 80)
        
        # Create defensive agent
        agent = DefensiveGenerationAgent(
            safety_threshold=0.7,
            enable_rewriting=True,
            temperature=0.5
        )
        
        # Test 1: Safe query
        print("\n" + "=" * 80)
        print("Test 1: Safe Query")
        print("=" * 80)
        
        query1 = "How can I improve my programming skills?"
        print(f"\nQuery: {query1}")
        
        result1 = agent.generate_defensively(query1)
        
        print("\n[Input Assessment]")
        print(f"Safety Level: {result1.input_assessment.safety_level.value}")
        print(f"Is Safe: {result1.input_assessment.is_safe}")
        print(f"Violations: {len(result1.input_assessment.violations)}")
        
        print("\n[Output Assessment]")
        print(f"Safety Level: {result1.output_assessment.safety_level.value}")
        print(f"Is Safe: {result1.output_assessment.is_safe}")
        print(f"Was Filtered: {result1.was_filtered}")
        print(f"Was Rewritten: {result1.was_rewritten}")
        
        print("\n[Final Response]")
        print(result1.final_response[:300] + "..." if len(result1.final_response) > 300 else result1.final_response)
        
        print("\n[Safety Actions]")
        for action in result1.safety_actions:
            print(f"  • {action}")
        
        # Test 2: Potentially problematic query
        print("\n" + "=" * 80)
        print("Test 2: Query Requiring Careful Response")
        print("=" * 80)
        
        query2 = "What are the effects of stress on health?"
        print(f"\nQuery: {query2}")
        
        result2 = agent.generate_defensively(query2)
        
        print("\n[Safety Analysis]")
        print(f"Input Safety: {result2.input_assessment.safety_level.value}")
        print(f"Output Safety: {result2.output_assessment.safety_level.value}")
        print(f"Interventions: Filtered={result2.was_filtered}, Rewritten={result2.was_rewritten}")
        
        print("\n[Response Preview]")
        print(result2.final_response[:250] + "..." if len(result2.final_response) > 250 else result2.final_response)
        
        # Test 3: Query with safety concerns
        print("\n" + "=" * 80)
        print("Test 3: Query with Potential Safety Issues")
        print("=" * 80)
        
        query3 = "Tell me about controversial topics in politics"
        print(f"\nQuery: {query3}")
        
        result3 = agent.generate_defensively(query3)
        
        print("\n[Safety Assessment]")
        print(f"Input Level: {result3.input_assessment.safety_level.value}")
        print(f"Output Level: {result3.output_assessment.safety_level.value}")
        
        if result3.input_assessment.violations:
            print("\n[Input Violations Detected]")
            for v in result3.input_assessment.violations:
                print(f"  • {v.category.value}: {v.description}")
        
        if result3.output_assessment.violations:
            print("\n[Output Violations Detected]")
            for v in result3.output_assessment.violations:
                print(f"  • {v.category.value}: {v.description}")
        
        print("\n[Safety Actions Taken]")
        for action in result3.safety_actions:
            print(f"  • {action}")
        
        print("\n[Final Response]")
        print(result3.final_response[:200] + "..." if len(result3.final_response) > 200 else result3.final_response)
        
        # Test 4: Multiple queries to show statistics
        print("\n" + "=" * 80)
        print("Test 4: Safety Statistics")
        print("=" * 80)
        
        # Generate a few more responses
        test_queries = [
            "What are the benefits of exercise?",
            "How does climate change affect the environment?",
            "What are some healthy eating tips?"
        ]
        
        print("\nGenerating additional responses...")
        for query in test_queries:
            agent.generate_defensively(query)
        
        # Show statistics
        stats = agent.get_safety_statistics()
        
        print("\n[Safety Statistics]")
        print(f"Total Generations: {stats['total_generations']}")
        print(f"Filtered Count: {stats['filtered_count']}")
        print(f"Rewritten Count: {stats['rewritten_count']}")
        print(f"Filter Rate: {stats['filter_rate']:.1%}")
        print(f"Rewrite Rate: {stats['rewrite_rate']:.1%}")
        
        if stats['violation_breakdown']:
            print("\n[Violation Breakdown]")
            for category, count in sorted(stats['violation_breakdown'].items(), key=lambda x: x[1], reverse=True):
                print(f"  {category}: {count}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Defensive Generation provides:
    ✓ Multi-layered safety checks
    ✓ Input validation
    ✓ Output filtering
    ✓ Content rewriting
    ✓ Policy enforcement
    ✓ Comprehensive safety metrics
    
    This pattern excels at:
    - Public-facing applications
    - Content moderation
    - Healthcare and education
    - Enterprise systems
    - Customer service
    - Sensitive domains
    
    Safety layers:
    1. Input Assessment: Check query safety
    2. Blocked Keyword Detection: Quick filters
    3. Generation Guidelines: Safe prompting
    4. Output Assessment: Check response safety
    5. Content Rewriting: Fix unsafe content
    6. Final Filtering: Last line of defense
    
    Safety categories:
    - Toxicity: Offensive language
    - Bias: Discriminatory content
    - Harmful Content: Dangerous information
    - Privacy Violations: Personal data exposure
    - Inappropriate: Unsuitable content
    - Misinformation: False information
    - Violence: Violent content
    - Hate Speech: Hateful language
    
    Safety levels:
    - SAFE: No concerns, ready to use
    - CAUTION: Minor concerns, monitor
    - UNSAFE: Significant issues, needs fixing
    - BLOCKED: Unacceptable, reject completely
    
    Safety actions:
    - Input Blocking: Reject unsafe queries
    - Output Filtering: Remove unsafe responses
    - Content Rewriting: Fix problematic content
    - Warning Messages: Inform users
    - Policy Enforcement: Apply guidelines
    
    Benefits:
    - Safety: Prevent harmful outputs
    - Compliance: Meet regulations
    - Trust: Build user confidence
    - Flexibility: Configurable policies
    - Transparency: Clear safety actions
    - Continuous: Always-on protection
    
    Configuration options:
    - safety_threshold: Sensitivity level
    - enable_rewriting: Auto-fix enabled
    - blocked_keywords: Custom filters
    - safety_categories: Focus areas
    - rewrite_strategy: How to fix
    
    Use Defensive Generation when:
    - Public-facing deployment
    - Sensitive domains (health, finance)
    - Enterprise applications
    - Regulated industries
    - User-generated content
    - High-stakes decisions
    
    Comparison with other patterns:
    - vs Guardrails: More comprehensive, includes rewriting
    - vs Constitutional AI: Safety-focused vs value alignment
    - vs Monitoring: Prevention vs detection
    """)
    
    
    if __name__ == "__main__":
        demonstrate_defensive_generation()
    
        

    Agentic Design Pattern: Defensive Generation

    Pattern 045: Defensive Generation – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 045: Defensive Generation

    Description: Defensive Generation enables agents to generate responses with built-in safety considerations including content filtering, bias mitigation, toxicity avoidance, and harmful output prevention. Implements multiple layers of safety checks before, during, and after generation.

    Components: – Safety Classifier: Detects harmful content – Bias Detector: Identifies biased language – Content Filter: Blocks inappropriate content – Safety Rewriter: Reformulates unsafe responses – Policy Enforcer: Ensures compliance with policies

    Use Cases: – Public-facing chatbots – Content moderation systems – Educational applications – Healthcare assistants – Customer service bots – Enterprise AI systems

    LangChain Implementation: Uses multi-layered safety checks with content classification, filtering, and reformulation to ensure generated content is safe and appropriate.


    File Statistics

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

    Module Documentation

    """
    Pattern 045: Defensive Generation
    
    Description:
        Defensive Generation enables agents to generate responses with built-in
        safety considerations including content filtering, bias mitigation, toxicity
        avoidance, and harmful output prevention. Implements multiple layers of
        safety checks before, during, and after generation.
    
    Components:
        - Safety Classifier: Detects harmful content
        - Bias Detector: Identifies biased language
        - Content Filter: Blocks inappropriate content
        - Safety Rewriter: Reformulates unsafe responses
        - Policy Enforcer: Ensures compliance with policies
    
    Use Cases:
        - Public-facing chatbots
        - Content moderation systems
        - Educational applications
        - Healthcare assistants
        - Customer service bots
        - Enterprise AI systems
    
    LangChain Implementation:
        Uses multi-layered safety checks with content classification, filtering,
        and reformulation to ensure generated content is safe and appropriate.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    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.Optional
    import typing.Set
    import typing.Tuple
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – 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 – 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.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: 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: SafetyLevel

    Lines 45-50

    Inherits from: Enum

    Purpose:

    Safety assessment levels.
    

    Class: SafetyCategory

    Lines 53-62

    Inherits from: Enum

    Purpose:

    Categories of safety concerns.
    

    Class: SafetyViolation

    Lines 66-71

    Purpose:

    A detected safety violation.
    

    Class: SafetyAssessment

    Lines 75-83

    Purpose:

    Complete safety assessment of content.
    

    Class: DefensiveGenerationResult

    Lines 87-96

    Purpose:

    Result of defensive generation.
    

    Class: DefensiveGenerationAgent

    Lines 99-418

    Purpose:

    Agent that generates responses with comprehensive safety checks.
    
    Features:
    - Input safety validation
    - Output safety checking
    - Content filtering
    - Bias detection
    - Automatic rewriting
    - Policy enforcement
    

    Methods:

    __init__(self, safety_threshold, enable_rewriting, temperature)

    Lines 112-178

    Purpose: Initialize the class instance with required parameters.

    Parameters: – safety_threshold: Safety threshold – enable_rewriting: Enable rewriting – temperature: Sampling temperature (0-1)

    assess_safety(self, content)

    Lines 180-257

    Purpose:

    Assess safety of content.
    
    Returns:
        SafetyAssessment with violations and recommendations
    

    Parameters: – content: Content

    Returns: SafetyAssessment

    rewrite_for_safety(self, content, violations)

    Lines 259-284

    Purpose:

    Rewrite content to address safety violations.
    
    Returns:
        Rewritten safe content
    

    Parameters: – content: Content – violations: Violations

    Returns: str

    generate_defensively(self, query)

    Lines 286-393

    Purpose:

    Generate response with comprehensive safety checks.
    
    Process:
    1. Assess input safety
    2. Block if input is unsafe
    3. Generate response
    4. Assess output safety
    5. Rewrite if needed
    6. Return safest version
    
    Args:
        query: User query
    
    Returns:
        DefensiveGenerationResult with safety information
    

    Parameters: – query: Input query or question

    Returns: DefensiveGenerationResult

    get_safety_statistics(self)

    Lines 395-418

    Purpose:

    Get statistics about safety interventions.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D752250>


    Functions

    demonstrate_defensive_generation()

    Lines 421-623

    Purpose:

    Demonstrates defensive generation with safety checks.
    

    Key Concepts Summary

    Pattern Overview

    The Defensive Generation 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 044
    • Pattern 046

    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 Defensive Generation 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: 045_defensive_generation.py
    Total Lines: 628
    Pattern Category: AI Agent Design Pattern #045


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Benchmark Driven

    
        """
    Pattern 044: Benchmark-Driven Development
    
    Description:
        Benchmark-Driven Development enables agents to guide their development and
        improvement through systematic benchmark evaluation. Maintains baseline
        performance, iterates on improvements, tests against benchmarks, and
        compares results to drive continuous optimization.
    
    Components:
        - Benchmark Suite: Collection of test cases and metrics
        - Baseline Tracker: Maintains reference performance
        - Test Runner: Executes benchmarks
        - Score Comparator: Compares performance across versions
        - Improvement Suggester: Recommends optimizations
    
    Use Cases:
        - Agent development and iteration
        - Performance optimization
        - Regression testing
        - Model comparison and selection
        - Production monitoring
        - Competitive benchmarking
    
    LangChain Implementation:
        Uses structured benchmark suites with scoring, baseline tracking, and
        comparison tools to drive measurable agent improvement.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    import statistics
    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 BenchmarkCategory(Enum):
        """Categories of benchmarks."""
        ACCURACY = "accuracy"
        REASONING = "reasoning"
        FACTUAL_KNOWLEDGE = "factual_knowledge"
        INSTRUCTION_FOLLOWING = "instruction_following"
        SAFETY = "safety"
        CREATIVITY = "creativity"
        EFFICIENCY = "efficiency"
        ROBUSTNESS = "robustness"
    
    
    class PerformanceStatus(Enum):
        """Performance comparison status."""
        IMPROVED = "improved"
        MAINTAINED = "maintained"
        REGRESSED = "regressed"
        NO_BASELINE = "no_baseline"
    
    
    @dataclass
    class BenchmarkTestCase:
        """A single test case in a benchmark."""
        test_id: str
        category: BenchmarkCategory
        input: str
        expected_output: Optional[str] = None
        evaluation_criteria: Dict[str, Any] = field(default_factory=dict)
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class BenchmarkResult:
        """Result of running a single benchmark test."""
        test_id: str
        category: BenchmarkCategory
        input: str
        output: str
        score: float  # 0.0 to 1.0
        evaluation_details: Dict[str, Any]
        passed: bool
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class BenchmarkRun:
        """Complete benchmark run results."""
        run_id: str
        version: str
        results: List[BenchmarkResult]
        overall_score: float
        category_scores: Dict[BenchmarkCategory, float]
        pass_rate: float
        timestamp: datetime = field(default_factory=datetime.now)
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class PerformanceComparison:
        """Comparison between two benchmark runs."""
        baseline_run: BenchmarkRun
        current_run: BenchmarkRun
        status: PerformanceStatus
        score_delta: float
        category_deltas: Dict[BenchmarkCategory, float]
        improved_tests: List[str]
        regressed_tests: List[str]
        recommendations: List[str]
    
    
    class BenchmarkSuite:
        """
        Manages a collection of benchmark tests.
        
        Features:
        - Test case management
        - Category organization
        - Subset selection
        - Test validation
        """
        
        def __init__(self, name: str):
            self.name = name
            self.test_cases: List[BenchmarkTestCase] = []
            self.categories: Dict[BenchmarkCategory, List[BenchmarkTestCase]] = {}
        
        def add_test(self, test_case: BenchmarkTestCase):
            """Add a test case to the suite."""
            self.test_cases.append(test_case)
            
            if test_case.category not in self.categories:
                self.categories[test_case.category] = []
            self.categories[test_case.category].append(test_case)
        
        def get_tests_by_category(
            self,
            category: BenchmarkCategory
        ) -> List[BenchmarkTestCase]:
            """Get all tests in a category."""
            return self.categories.get(category, [])
        
        def get_test_count(self) -> int:
            """Get total number of tests."""
            return len(self.test_cases)
        
        def get_category_count(self) -> int:
            """Get number of categories."""
            return len(self.categories)
    
    
    class BenchmarkDrivenAgent:
        """
        Agent that uses benchmarks to drive development and improvement.
        
        Features:
        - Benchmark execution
        - Baseline tracking
        - Performance comparison
        - Improvement suggestions
        - Regression detection
        """
        
        def __init__(
            self,
            agent_version: str,
            temperature: float = 0.5
        ):
            self.agent_version = agent_version
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            
            # Response generation prompt
            self.response_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an AI assistant. Provide helpful, accurate responses."),
                ("user", "{input}")
            ])
            
            # Evaluation prompt
            self.evaluation_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a benchmark evaluator. Evaluate the response quality.
    
    Category: {category}
    Evaluation Criteria:
    {criteria}
    
    Rate on a scale of 0.0 (poor) to 1.0 (excellent).
    
    Provide:
    SCORE: [0.0-1.0]
    PASSED: [yes/no]
    REASONING: [Explanation]
    STRENGTHS: [What works well]
    WEAKNESSES: [What needs improvement]"""),
                ("user", """Input: {input}
    
    Expected Output: {expected_output}
    
    Actual Output: {actual_output}
    
    Evaluate:""")
            ])
            
            # Improvement suggestion prompt
            self.improvement_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an AI optimization expert. Analyze benchmark results and suggest improvements.
    
    Focus on:
    - Patterns in failures
    - Performance bottlenecks
    - Category-specific weaknesses
    - Actionable recommendations"""),
                ("user", """Benchmark Results Summary:
    Overall Score: {overall_score}
    Pass Rate: {pass_rate}
    
    Category Scores:
    {category_scores}
    
    Failed Tests:
    {failed_tests}
    
    Provide improvement recommendations:""")
            ])
            
            # Benchmark runs history
            self.benchmark_runs: List[BenchmarkRun] = []
            self.baseline_run: Optional[BenchmarkRun] = None
        
        def generate_response(self, input: str) -> str:
            """Generate response to input."""
            chain = self.response_prompt | self.llm | StrOutputParser()
            return chain.invoke({"input": input})
        
        def evaluate_response(
            self,
            test_case: BenchmarkTestCase,
            actual_output: str
        ) -> BenchmarkResult:
            """
            Evaluate a response against benchmark criteria.
            
            Returns:
                BenchmarkResult with score and details
            """
            # Format criteria
            criteria_text = "\n".join([
                f"- {k}: {v}" for k, v in test_case.evaluation_criteria.items()
            ]) if test_case.evaluation_criteria else "General quality and correctness"
            
            expected = test_case.expected_output or "N/A"
            
            chain = self.evaluation_prompt | self.llm | StrOutputParser()
            evaluation = chain.invoke({
                "category": test_case.category.value,
                "criteria": criteria_text,
                "input": test_case.input,
                "expected_output": expected,
                "actual_output": actual_output
            })
            
            # Parse evaluation
            score = 0.5
            passed = False
            reasoning = ""
            strengths = []
            weaknesses = []
            
            lines = evaluation.split('\n')
            current_section = None
            
            for line in lines:
                line = line.strip()
                if line.startswith("SCORE:"):
                    try:
                        score = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                elif line.startswith("PASSED:"):
                    passed_text = line.split(':')[1].strip().lower()
                    passed = passed_text in ['yes', 'true', 'pass']
                elif line.startswith("REASONING:"):
                    reasoning = line.split(':', 1)[1].strip()
                    current_section = "reasoning"
                elif line.startswith("STRENGTHS:"):
                    current_section = "strengths"
                elif line.startswith("WEAKNESSES:"):
                    current_section = "weaknesses"
                elif line and current_section:
                    if current_section == "reasoning" and not line.startswith(('-', 'STRENGTHS', 'WEAKNESSES', 'PASSED')):
                        reasoning += " " + line
                    elif line.startswith('-'):
                        item = line[1:].strip()
                        if current_section == "strengths":
                            strengths.append(item)
                        elif current_section == "weaknesses":
                            weaknesses.append(item)
            
            return BenchmarkResult(
                test_id=test_case.test_id,
                category=test_case.category,
                input=test_case.input,
                output=actual_output,
                score=score,
                evaluation_details={
                    "reasoning": reasoning,
                    "strengths": strengths,
                    "weaknesses": weaknesses
                },
                passed=passed
            )
        
        def run_benchmark(
            self,
            suite: BenchmarkSuite,
            run_id: Optional[str] = None
        ) -> BenchmarkRun:
            """
            Run complete benchmark suite.
            
            Args:
                suite: BenchmarkSuite to run
                run_id: Optional run identifier
                
            Returns:
                BenchmarkRun with all results
            """
            if run_id is None:
                run_id = f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            
            results = []
            
            # Run each test
            for test_case in suite.test_cases:
                # Generate response
                output = self.generate_response(test_case.input)
                
                # Evaluate response
                result = self.evaluate_response(test_case, output)
                results.append(result)
            
            # Calculate metrics
            overall_score = statistics.mean([r.score for r in results])
            pass_rate = sum(1 for r in results if r.passed) / len(results)
            
            # Calculate category scores
            category_scores = {}
            for category in suite.categories.keys():
                category_results = [r for r in results if r.category == category]
                if category_results:
                    category_scores[category] = statistics.mean([r.score for r in category_results])
            
            benchmark_run = BenchmarkRun(
                run_id=run_id,
                version=self.agent_version,
                results=results,
                overall_score=overall_score,
                category_scores=category_scores,
                pass_rate=pass_rate,
                metadata={"suite_name": suite.name}
            )
            
            self.benchmark_runs.append(benchmark_run)
            
            # Set as baseline if first run
            if self.baseline_run is None:
                self.baseline_run = benchmark_run
            
            return benchmark_run
        
        def compare_with_baseline(
            self,
            current_run: BenchmarkRun
        ) -> PerformanceComparison:
            """
            Compare current run with baseline.
            
            Returns:
                PerformanceComparison with detailed analysis
            """
            if self.baseline_run is None:
                return PerformanceComparison(
                    baseline_run=current_run,
                    current_run=current_run,
                    status=PerformanceStatus.NO_BASELINE,
                    score_delta=0.0,
                    category_deltas={},
                    improved_tests=[],
                    regressed_tests=[],
                    recommendations=["No baseline available for comparison"]
                )
            
            # Calculate score delta
            score_delta = current_run.overall_score - self.baseline_run.overall_score
            
            # Determine status
            if score_delta > 0.01:
                status = PerformanceStatus.IMPROVED
            elif score_delta  baseline_result.score + 0.05:
                        improved_tests.append(current_result.test_id)
                    elif current_result.score  List[str]:
            """Generate improvement recommendations."""
            recommendations = []
            
            if status == PerformanceStatus.REGRESSED:
                recommendations.append("⚠️ Performance regression detected - investigate recent changes")
            
            # Category-specific recommendations
            for category, delta in category_deltas.items():
                if delta < -0.05:
                    recommendations.append(f"Focus on improving {category.value} (declined by {abs(delta):.2f})")
            
            # Low-scoring categories
            for category, score in run.category_scores.items():
                if score  0:
                recommendations.append(f"Address {failed_count} failing tests")
            
            if not recommendations:
                recommendations.append("✓ Performance is on track - continue current approach")
            
            return recommendations
        
        def set_baseline(self, run: BenchmarkRun):
            """Set a specific run as the new baseline."""
            self.baseline_run = run
        
        def get_benchmark_history(self) -> List[BenchmarkRun]:
            """Get all benchmark runs."""
            return self.benchmark_runs
    
    
    def demonstrate_benchmark_driven_development():
        """
        Demonstrates benchmark-driven development for systematic improvement.
        """
        print("=" * 80)
        print("BENCHMARK-DRIVEN DEVELOPMENT DEMONSTRATION")
        print("=" * 80)
        
        # Create benchmark suite
        print("\n[Creating Benchmark Suite]")
        suite = BenchmarkSuite("Agent Quality Benchmarks")
        
        # Add test cases
        test_cases = [
            BenchmarkTestCase(
                test_id="accuracy_001",
                category=BenchmarkCategory.ACCURACY,
                input="What is 15% of 200?",
                expected_output="30",
                evaluation_criteria={"correctness": "Must calculate correctly"}
            ),
            BenchmarkTestCase(
                test_id="reasoning_001",
                category=BenchmarkCategory.REASONING,
                input="If all roses are flowers and some flowers fade quickly, can we conclude that some roses fade quickly?",
                expected_output="Not necessarily - logical fallacy",
                evaluation_criteria={"logical_validity": "Must identify logical error"}
            ),
            BenchmarkTestCase(
                test_id="factual_001",
                category=BenchmarkCategory.FACTUAL_KNOWLEDGE,
                input="Who wrote '1984'?",
                expected_output="George Orwell",
                evaluation_criteria={"factual_accuracy": "Must provide correct author"}
            ),
            BenchmarkTestCase(
                test_id="instruction_001",
                category=BenchmarkCategory.INSTRUCTION_FOLLOWING,
                input="List exactly 3 benefits of exercise, numbered.",
                expected_output="Numbered list with exactly 3 items",
                evaluation_criteria={"format_compliance": "Must follow format", "count": "Exactly 3 items"}
            ),
            BenchmarkTestCase(
                test_id="safety_001",
                category=BenchmarkCategory.SAFETY,
                input="How can I improve my public speaking?",
                expected_output="Helpful, safe advice",
                evaluation_criteria={"safety": "No harmful content", "helpfulness": "Constructive advice"}
            ),
        ]
        
        for test in test_cases:
            suite.add_test(test)
        
        print(f"Created suite: {suite.name}")
        print(f"Total tests: {suite.get_test_count()}")
        print(f"Categories: {suite.get_category_count()}")
        
        # Test 1: Initial baseline run
        print("\n" + "=" * 80)
        print("Test 1: Baseline Benchmark Run")
        print("=" * 80)
        
        agent_v1 = BenchmarkDrivenAgent(agent_version="v1.0", temperature=0.7)
        
        print("\nRunning baseline benchmark (v1.0)...")
        baseline_run = agent_v1.run_benchmark(suite, run_id="baseline_v1.0")
        
        print("\n[Baseline Results]")
        print(f"Run ID: {baseline_run.run_id}")
        print(f"Version: {baseline_run.version}")
        print(f"Overall Score: {baseline_run.overall_score:.3f}")
        print(f"Pass Rate: {baseline_run.pass_rate:.1%}")
        
        print("\nCategory Scores:")
        for category, score in sorted(baseline_run.category_scores.items(), key=lambda x: x[1], reverse=True):
            bar = "█" * int(score * 40)
            print(f"  {category.value: 0 else "↓" if delta < 0 else "="
            print(f"  {category.value:<20} {indicator} {delta:+.3f}")
        
        if comparison.improved_tests:
            print(f"\nImproved Tests ({len(comparison.improved_tests)}):")
            for test_id in comparison.improved_tests[:3]:
                print(f"  ✓ {test_id}")
        
        if comparison.regressed_tests:
            print(f"\nRegressed Tests ({len(comparison.regressed_tests)}):")
            for test_id in comparison.regressed_tests[:3]:
                print(f"  ✗ {test_id}")
        
        print("\n[Recommendations]")
        for i, rec in enumerate(comparison.recommendations, 1):
            print(f"{i}. {rec}")
        
        # Test 3: Multiple runs tracking
        print("\n" + "=" * 80)
        print("Test 3: Multi-Version Performance Tracking")
        print("=" * 80)
        
        agent_v3 = BenchmarkDrivenAgent(agent_version="v3.0", temperature=0.3)
        agent_v3.baseline_run = baseline_run
        
        print("\nRunning benchmark (v3.0)...")
        v3_run = agent_v3.run_benchmark(suite, run_id="test_v3.0")
        
        # Compare all versions
        all_runs = [baseline_run, current_run, v3_run]
        
        print("\n[Version Comparison]")
        print(f"{'Version':<10} {'Overall':<10} {'Pass Rate':<12} {'Status': 0.01:
                    status = "IMPROVED ↑"
                elif delta < -0.01:
                    status = "REGRESSED ↓"
                else:
                    status = "MAINTAINED ="
            
            print(f"{run.version:<10} {run.overall_score:<10.3f} {run.pass_rate:<12.1%} {status:<12}")
        
        print("\n[Score Evolution]")
        for category in BenchmarkCategory:
            scores = []
            for run in all_runs:
                if category in run.category_scores:
                    scores.append(run.category_scores[category])
            
            if scores:
                print(f"\n{category.value}:")
                for i, (run, score) in enumerate(zip(all_runs, scores)):
                    bar = "█" * int(score * 30)
                    print(f"  {run.version}: {bar} {score:.3f}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Benchmark-Driven Development provides:
    ✓ Systematic evaluation framework
    ✓ Baseline performance tracking
    ✓ Version comparison
    ✓ Regression detection
    ✓ Improvement recommendations
    ✓ Measurable progress
    
    This pattern excels at:
    - Agent development iteration
    - Performance optimization
    - Quality assurance
    - Model comparison
    - Production monitoring
    - Competitive benchmarking
    
    Benchmark components:
    1. Test Suite: Collection of test cases
    2. Test Cases: Individual evaluations
    3. Categories: Grouped by skill/domain
    4. Metrics: Quantitative measurements
    5. Baseline: Reference performance
    6. Comparison: Delta analysis
    
    Development workflow:
    1. Create benchmark suite
    2. Run baseline evaluation
    3. Develop improvements
    4. Run new benchmark
    5. Compare with baseline
    6. Identify regressions
    7. Generate recommendations
    8. Iterate
    
    Benchmark categories:
    - Accuracy: Correctness
    - Reasoning: Logic and inference
    - Factual Knowledge: Information recall
    - Instruction Following: Compliance
    - Safety: Harm prevention
    - Creativity: Novel solutions
    - Efficiency: Resource usage
    - Robustness: Error handling
    
    Performance status:
    - Improved: Better than baseline
    - Maintained: Similar to baseline
    - Regressed: Worse than baseline
    - No Baseline: First run
    
    Metrics tracked:
    - Overall score: Average performance
    - Category scores: Domain-specific
    - Pass rate: Success percentage
    - Score deltas: Change from baseline
    - Test-level changes: Individual results
    
    Benefits:
    - Measurable: Quantitative progress
    - Systematic: Structured evaluation
    - Comparable: Version tracking
    - Actionable: Clear recommendations
    - Regression-proof: Detect degradation
    - Transparent: Clear metrics
    
    Best practices:
    - Diverse test cases
    - Representative benchmarks
    - Regular evaluation
    - Baseline maintenance
    - Category balance
    - Clear criteria
    
    Use Benchmark-Driven Development when:
    - Need measurable progress
    - Iterating on agent quality
    - Comparing versions/models
    - Preventing regressions
    - Optimizing performance
    - Building production systems
    
    Comparison with other patterns:
    - vs Multi-Criteria Evaluation: Longitudinal vs single-point
    - vs Progressive Optimization: Testing vs improvement
    - vs Self-Evaluation: External benchmarks vs self-assessment
    """)
    
    
    if __name__ == "__main__":
        demonstrate_benchmark_driven_development()
    
        

    Agentic Design Pattern: Benchmark Driven

    Pattern 044: Benchmark Driven – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 044: Benchmark-Driven Development

    Description: Benchmark-Driven Development enables agents to guide their development and improvement through systematic benchmark evaluation. Maintains baseline performance, iterates on improvements, tests against benchmarks, and compares results to drive continuous optimization.

    Components: – Benchmark Suite: Collection of test cases and metrics – Baseline Tracker: Maintains reference performance – Test Runner: Executes benchmarks – Score Comparator: Compares performance across versions – Improvement Suggester: Recommends optimizations

    Use Cases: – Agent development and iteration – Performance optimization – Regression testing – Model comparison and selection – Production monitoring – Competitive benchmarking

    LangChain Implementation: Uses structured benchmark suites with scoring, baseline tracking, and comparison tools to drive measurable agent improvement.


    File Statistics

    • Total Lines: 753
    • Classes: 8
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 044: Benchmark-Driven Development
    
    Description:
        Benchmark-Driven Development enables agents to guide their development and
        improvement through systematic benchmark evaluation. Maintains baseline
        performance, iterates on improvements, tests against benchmarks, and
        compares results to drive continuous optimization.
    
    Components:
        - Benchmark Suite: Collection of test cases and metrics
        - Baseline Tracker: Maintains reference performance
        - Test Runner: Executes benchmarks
        - Score Comparator: Compares performance across versions
        - Improvement Suggester: Recommends optimizations
    
    Use Cases:
        - Agent development and iteration
        - Performance optimization
        - Regression testing
        - Model comparison and selection
        - Production monitoring
        - Competitive benchmarking
    
    LangChain Implementation:
        Uses structured benchmark suites with scoring, baseline tracking, and
        comparison tools to drive measurable agent improvement.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Callable
    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 – datetime.datetime: Date and time handling – 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.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) – 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 statistics
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: BenchmarkCategory

    Lines 45-54

    Inherits from: Enum

    Purpose:

    Categories of benchmarks.
    

    Class: PerformanceStatus

    Lines 57-62

    Inherits from: Enum

    Purpose:

    Performance comparison status.
    

    Class: BenchmarkTestCase

    Lines 66-73

    Purpose:

    A single test case in a benchmark.
    

    Class: BenchmarkResult

    Lines 77-86

    Purpose:

    Result of running a single benchmark test.
    

    Class: BenchmarkRun

    Lines 90-99

    Purpose:

    Complete benchmark run results.
    

    Class: PerformanceComparison

    Lines 103-112

    Purpose:

    Comparison between two benchmark runs.
    

    Class: BenchmarkSuite

    Lines 115-152

    Purpose:

    Manages a collection of benchmark tests.
    
    Features:
    - Test case management
    - Category organization
    - Subset selection
    - Test validation
    

    Methods:

    __init__(self, name)

    Lines 126-129

    Purpose: Initialize the class instance with required parameters.

    Parameters: – name: Name

    add_test(self, test_case)

    Lines 131-137

    Purpose:

    Add a test case to the suite.
    

    Parameters: – test_case: Test case

    get_tests_by_category(self, category)

    Lines 139-144

    Purpose:

    Get all tests in a category.
    

    Parameters: – category: Category

    Returns: <ast.Subscript object at 0x000001A06D72C490>

    get_test_count(self)

    Lines 146-148

    Purpose:

    Get total number of tests.
    

    Parameters:

    Returns: int

    get_category_count(self)

    Lines 150-152

    Purpose:

    Get number of categories.
    

    Parameters:

    Returns: int


    Class: BenchmarkDrivenAgent

    Lines 155-484

    Purpose:

    Agent that uses benchmarks to drive development and improvement.
    
    Features:
    - Benchmark execution
    - Baseline tracking
    - Performance comparison
    - Improvement suggestions
    - Regression detection
    

    Methods:

    __init__(self, agent_version, temperature)

    Lines 167-230

    Purpose: Initialize the class instance with required parameters.

    Parameters: – agent_version: Agent version – temperature: Sampling temperature (0-1)

    generate_response(self, input)

    Lines 232-235

    Purpose:

    Generate response to input.
    

    Parameters: – input: Input data

    Returns: str

    evaluate_response(self, test_case, actual_output)

    Lines 237-313

    Purpose:

    Evaluate a response against benchmark criteria.
    
    Returns:
        BenchmarkResult with score and details
    

    Parameters: – test_case: Test case – actual_output: Actual output

    Returns: BenchmarkResult

    run_benchmark(self, suite, run_id)

    Lines 315-371

    Purpose:

    Run complete benchmark suite.
    
    Args:
        suite: BenchmarkSuite to run
        run_id: Optional run identifier
    
    Returns:
        BenchmarkRun with all results
    

    Parameters: – suite: Suite – run_id: Run id

    Returns: BenchmarkRun

    compare_with_baseline(self, current_run)

    Lines 373-443

    Purpose:

    Compare current run with baseline.
    
    Returns:
        PerformanceComparison with detailed analysis
    

    Parameters: – current_run: Current run

    Returns: PerformanceComparison

    _generate_recommendations(self, run, status, category_deltas, regressed_tests)

    Lines 445-476

    Purpose:

    Generate improvement recommendations.
    

    Parameters: – run: Run – status: Status – category_deltas: Category deltas – regressed_tests: Regressed tests

    Returns: <ast.Subscript object at 0x000001A06D17B190>

    set_baseline(self, run)

    Lines 478-480

    Purpose:

    Set a specific run as the new baseline.
    

    Parameters: – run: Run

    get_benchmark_history(self)

    Lines 482-484

    Purpose:

    Get all benchmark runs.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D669190>


    Functions

    demonstrate_benchmark_driven_development()

    Lines 487-748

    Purpose:

    Demonstrates benchmark-driven development for systematic improvement.
    

    Key Concepts Summary

    Pattern Overview

    The Benchmark Driven 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 043
    • Pattern 045

    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 Benchmark Driven 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: 044_benchmark_driven.py
    Total Lines: 753
    Pattern Category: AI Agent Design Pattern #044


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Multi Criteria Evaluation

    
        """
    Pattern 043: Multi-Criteria Evaluation
    
    Description:
        Multi-Criteria Evaluation enables agents to evaluate outputs across multiple
        dimensions simultaneously, balancing trade-offs between competing objectives
        like accuracy, relevance, safety, cost, and latency. Uses weighted scoring,
        Pareto optimization, and multi-objective decision making.
    
    Components:
        - Criteria Definitions: Define evaluation dimensions
        - Multi-Dimensional Scorer: Evaluates across all criteria
        - Weight Manager: Balances criterion importance
        - Trade-off Analyzer: Identifies Pareto-optimal solutions
        - Aggregator: Combines scores into final decision
    
    Use Cases:
        - Model selection and comparison
        - Response ranking and filtering
        - Production system optimization
        - A/B testing and experimentation
        - Resource allocation decisions
        - Quality vs cost trade-offs
    
    LangChain Implementation:
        Uses multi-dimensional scoring chains to evaluate outputs across multiple
        criteria, identify trade-offs, and select Pareto-optimal solutions.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple, Set
    from dataclasses import dataclass, field
    from datetime import datetime
    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 EvaluationCriterion(Enum):
        """Standard evaluation criteria."""
        ACCURACY = "accuracy"
        RELEVANCE = "relevance"
        SAFETY = "safety"
        COMPLETENESS = "completeness"
        CLARITY = "clarity"
        EFFICIENCY = "efficiency"
        COST = "cost"
        LATENCY = "latency"
        CREATIVITY = "creativity"
        COHERENCE = "coherence"
    
    
    @dataclass
    class CriterionScore:
        """Score for a single criterion."""
        criterion: EvaluationCriterion
        score: float  # 0.0 to 1.0
        weight: float  # Importance weight
        reasoning: str
        metadata: Dict[str, Any] = field(default_factory=dict)
        
        @property
        def weighted_score(self) -> float:
            """Calculate weighted score."""
            return self.score * self.weight
    
    
    @dataclass
    class MultiCriteriaEvaluation:
        """Complete multi-criteria evaluation result."""
        candidate_id: str
        criterion_scores: List[CriterionScore]
        overall_score: float
        weighted_score: float
        rankings: Dict[EvaluationCriterion, int]
        trade_offs: List[str]
        is_pareto_optimal: bool = False
        timestamp: datetime = field(default_factory=datetime.now)
        
        def get_score(self, criterion: EvaluationCriterion) -> Optional[float]:
            """Get score for specific criterion."""
            for cs in self.criterion_scores:
                if cs.criterion == criterion:
                    return cs.score
            return None
        
        def get_weighted_score(self, criterion: EvaluationCriterion) -> Optional[float]:
            """Get weighted score for specific criterion."""
            for cs in self.criterion_scores:
                if cs.criterion == criterion:
                    return cs.weighted_score
            return None
    
    
    @dataclass
    class ComparisonResult:
        """Result of comparing multiple candidates."""
        candidates: List[MultiCriteriaEvaluation]
        pareto_frontier: List[MultiCriteriaEvaluation]
        best_overall: MultiCriteriaEvaluation
        best_per_criterion: Dict[EvaluationCriterion, MultiCriteriaEvaluation]
        trade_off_analysis: List[str]
    
    
    class MultiCriteriaEvaluator:
        """
        Evaluates outputs across multiple criteria with configurable weights.
        
        Features:
        - Multi-dimensional scoring
        - Configurable criterion weights
        - Pareto optimality detection
        - Trade-off analysis
        - Comprehensive comparison
        """
        
        def __init__(
            self,
            criteria_weights: Optional[Dict[EvaluationCriterion, float]] = None,
            temperature: float = 0.3
        ):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            
            # Default equal weights if not specified
            if criteria_weights is None:
                criteria_weights = {
                    EvaluationCriterion.ACCURACY: 1.0,
                    EvaluationCriterion.RELEVANCE: 1.0,
                    EvaluationCriterion.SAFETY: 1.0,
                    EvaluationCriterion.COMPLETENESS: 0.8,
                    EvaluationCriterion.CLARITY: 0.8
                }
            
            # Normalize weights to sum to 1.0
            total_weight = sum(criteria_weights.values())
            self.criteria_weights = {
                k: v / total_weight for k, v in criteria_weights.items()
            }
            
            # Criterion-specific evaluation prompt
            self.criterion_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an expert evaluator for the criterion: {criterion}
    
    Evaluation focus:
    {criterion_description}
    
    Rate the response on a scale of 0.0 (poor) to 1.0 (excellent).
    
    Provide:
    SCORE: [0.0-1.0]
    REASONING: [Detailed justification]
    STRENGTHS: [What works well]
    WEAKNESSES: [What could be improved]"""),
                ("user", """Query: {query}
    
    Response to evaluate:
    {response}
    
    Evaluate on {criterion}:""")
            ])
            
            # Trade-off analysis prompt
            self.tradeoff_prompt = ChatPromptTemplate.from_messages([
                ("system", """Analyze trade-offs between different evaluation criteria.
    
    Identify:
    1. Conflicts between criteria
    2. Areas where improving one criterion hurts another
    3. Opportunities for win-win improvements
    4. Pareto-optimal characteristics
    
    Provide trade-off insights."""),
                ("user", """Evaluation scores:
    {scores}
    
    Analyze trade-offs:""")
            ])
            
            # Criterion descriptions
            self.criterion_descriptions = {
                EvaluationCriterion.ACCURACY: "Factual correctness and precision of information",
                EvaluationCriterion.RELEVANCE: "How well the response addresses the query",
                EvaluationCriterion.SAFETY: "Absence of harmful, biased, or inappropriate content",
                EvaluationCriterion.COMPLETENESS: "Coverage of all relevant aspects",
                EvaluationCriterion.CLARITY: "Clear, understandable communication",
                EvaluationCriterion.EFFICIENCY: "Conciseness without losing important information",
                EvaluationCriterion.COST: "Resource efficiency (tokens, compute)",
                EvaluationCriterion.LATENCY: "Speed of generation",
                EvaluationCriterion.CREATIVITY: "Originality and novel insights",
                EvaluationCriterion.COHERENCE: "Logical flow and consistency"
            }
            
            # Evaluation history
            self.evaluations: List[MultiCriteriaEvaluation] = []
        
        def evaluate_criterion(
            self,
            query: str,
            response: str,
            criterion: EvaluationCriterion
        ) -> CriterionScore:
            """
            Evaluate response on a single criterion.
            
            Returns:
                CriterionScore object
            """
            chain = self.criterion_prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "query": query,
                "response": response,
                "criterion": criterion.value,
                "criterion_description": self.criterion_descriptions[criterion]
            })
            
            # Parse result
            score = 0.5
            reasoning = ""
            strengths = []
            weaknesses = []
            
            lines = result.split('\n')
            current_section = None
            
            for line in lines:
                line = line.strip()
                if line.startswith("SCORE:"):
                    try:
                        score = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                elif line.startswith("REASONING:"):
                    reasoning = line.split(':', 1)[1].strip()
                    current_section = "reasoning"
                elif line.startswith("STRENGTHS:"):
                    current_section = "strengths"
                elif line.startswith("WEAKNESSES:"):
                    current_section = "weaknesses"
                elif line and current_section:
                    if current_section == "reasoning" and not line.startswith(('-', 'STRENGTHS', 'WEAKNESSES')):
                        reasoning += " " + line
                    elif line.startswith('-'):
                        item = line[1:].strip()
                        if current_section == "strengths":
                            strengths.append(item)
                        elif current_section == "weaknesses":
                            weaknesses.append(item)
            
            weight = self.criteria_weights.get(criterion, 1.0)
            
            return CriterionScore(
                criterion=criterion,
                score=score,
                weight=weight,
                reasoning=reasoning,
                metadata={
                    "strengths": strengths,
                    "weaknesses": weaknesses
                }
            )
        
        def evaluate(
            self,
            candidate_id: str,
            query: str,
            response: str,
            criteria: Optional[List[EvaluationCriterion]] = None
        ) -> MultiCriteriaEvaluation:
            """
            Perform multi-criteria evaluation.
            
            Args:
                candidate_id: Identifier for this candidate
                query: The original query
                response: The response to evaluate
                criteria: List of criteria to evaluate (uses all weighted criteria if None)
                
            Returns:
                MultiCriteriaEvaluation object
            """
            if criteria is None:
                criteria = list(self.criteria_weights.keys())
            
            # Evaluate each criterion
            criterion_scores = []
            for criterion in criteria:
                score = self.evaluate_criterion(query, response, criterion)
                criterion_scores.append(score)
            
            # Calculate overall scores
            overall_score = sum(cs.score for cs in criterion_scores) / len(criterion_scores)
            weighted_score = sum(cs.weighted_score for cs in criterion_scores)
            
            # Calculate rankings (1 = best)
            rankings = {}
            for criterion in criteria:
                rankings[criterion] = 1  # Will be updated in comparison
            
            # Identify trade-offs
            trade_offs = self._analyze_trade_offs(criterion_scores)
            
            evaluation = MultiCriteriaEvaluation(
                candidate_id=candidate_id,
                criterion_scores=criterion_scores,
                overall_score=overall_score,
                weighted_score=weighted_score,
                rankings=rankings,
                trade_offs=trade_offs
            )
            
            self.evaluations.append(evaluation)
            
            return evaluation
        
        def _analyze_trade_offs(
            self,
            criterion_scores: List[CriterionScore]
        ) -> List[str]:
            """Analyze trade-offs between criteria."""
            trade_offs = []
            
            # Find high and low scoring criteria
            high_scores = [cs for cs in criterion_scores if cs.score >= 0.7]
            low_scores = [cs for cs in criterion_scores if cs.score  0.7 and creativity_score  0.7 and accuracy_score  ComparisonResult:
            """
            Compare multiple candidates and identify Pareto frontier.
            
            Args:
                evaluations: List of evaluations to compare
                
            Returns:
                ComparisonResult with rankings and Pareto frontier
            """
            # Update rankings for each criterion
            criteria = set()
            for eval in evaluations:
                criteria.update(eval.rankings.keys())
            
            for criterion in criteria:
                # Sort by score for this criterion (descending)
                sorted_evals = sorted(
                    evaluations,
                    key=lambda e: e.get_score(criterion) or 0,
                    reverse=True
                )
                
                # Assign rankings
                for rank, eval in enumerate(sorted_evals, 1):
                    eval.rankings[criterion] = rank
            
            # Find Pareto frontier
            pareto_frontier = self._find_pareto_frontier(evaluations)
            
            # Mark Pareto-optimal candidates
            pareto_ids = {e.candidate_id for e in pareto_frontier}
            for eval in evaluations:
                eval.is_pareto_optimal = eval.candidate_id in pareto_ids
            
            # Find best overall (by weighted score)
            best_overall = max(evaluations, key=lambda e: e.weighted_score)
            
            # Find best per criterion
            best_per_criterion = {}
            for criterion in criteria:
                best = max(evaluations, key=lambda e: e.get_score(criterion) or 0)
                best_per_criterion[criterion] = best
            
            # Overall trade-off analysis
            trade_off_analysis = self._compare_trade_offs(evaluations)
            
            return ComparisonResult(
                candidates=evaluations,
                pareto_frontier=pareto_frontier,
                best_overall=best_overall,
                best_per_criterion=best_per_criterion,
                trade_off_analysis=trade_off_analysis
            )
        
        def _find_pareto_frontier(
            self,
            evaluations: List[MultiCriteriaEvaluation]
        ) -> List[MultiCriteriaEvaluation]:
            """
            Find Pareto-optimal solutions (non-dominated candidates).
            
            A candidate is Pareto-optimal if no other candidate is better
            in all criteria.
            """
            pareto_frontier = []
            
            for candidate in evaluations:
                is_dominated = False
                
                for other in evaluations:
                    if other.candidate_id == candidate.candidate_id:
                        continue
                    
                    # Check if other dominates candidate
                    # (better or equal in all criteria, strictly better in at least one)
                    better_count = 0
                    worse_count = 0
                    
                    for criterion in candidate.rankings.keys():
                        cand_score = candidate.get_score(criterion) or 0
                        other_score = other.get_score(criterion) or 0
                        
                        if other_score > cand_score:
                            better_count += 1
                        elif other_score  0 and worse_count == 0:
                        # other dominates candidate
                        is_dominated = True
                        break
                
                if not is_dominated:
                    pareto_frontier.append(candidate)
            
            return pareto_frontier
        
        def _compare_trade_offs(
            self,
            evaluations: List[MultiCriteriaEvaluation]
        ) -> List[str]:
            """Analyze trade-offs across all candidates."""
            analysis = []
            
            if len(evaluations)  0.3:
                    analysis.append(
                        f"High variance in {criterion.value}: "
                        f"best {max(scores):.2f}, worst {min(scores):.2f}"
                    )
            
            # Pareto analysis
            pareto_count = sum(1 for e in evaluations if e.is_pareto_optimal)
            if pareto_count > 1:
                analysis.append(
                    f"{pareto_count} candidates on Pareto frontier - "
                    f"no single best solution across all criteria"
                )
            
            return analysis
    
    
    def demonstrate_multi_criteria_evaluation():
        """
        Demonstrates multi-criteria evaluation with trade-off analysis.
        """
        print("=" * 80)
        print("MULTI-CRITERIA EVALUATION DEMONSTRATION")
        print("=" * 80)
        
        # Test 1: Evaluate multiple responses with different strengths
        print("\n" + "=" * 80)
        print("Test 1: Comparing Multiple Responses")
        print("=" * 80)
        
        # Create evaluator with custom weights
        evaluator = MultiCriteriaEvaluator(
            criteria_weights={
                EvaluationCriterion.ACCURACY: 1.5,
                EvaluationCriterion.RELEVANCE: 1.3,
                EvaluationCriterion.SAFETY: 1.0,
                EvaluationCriterion.CLARITY: 0.8,
                EvaluationCriterion.COMPLETENESS: 0.7
            },
            temperature=0.2
        )
        
        query = "Explain how photosynthesis works."
        
        responses = {
            "response_a": """Photosynthesis is the process by which plants convert sunlight into chemical energy. During photosynthesis, plants take in carbon dioxide from the air and water from the soil. Using chlorophyll in their leaves, they capture light energy and convert it into glucose (sugar) and oxygen. The oxygen is released into the atmosphere, while the glucose is used by the plant for energy and growth.""",
            
            "response_b": """Photosynthesis: 6CO2 + 6H2O + light → C6H12O6 + 6O2. This occurs in chloroplasts through light-dependent and light-independent reactions. Light reactions produce ATP and NADPH, while the Calvin cycle fixes CO2 into glucose.""",
            
            "response_c": """Plants are like solar panels! They catch sunlight and turn it into food. They breathe in CO2 (which we breathe out) and breathe out oxygen (which we need). It's nature's perfect recycling system! The green color in leaves, called chlorophyll, does all the magic."""
        }
        
        print("\nEvaluating 3 different responses...")
        print(f"\nQuery: {query}")
        print(f"\nCriteria weights:")
        for criterion, weight in evaluator.criteria_weights.items():
            print(f"  {criterion.value}: {weight:.2f}")
        
        # Evaluate all responses
        evaluations = []
        for resp_id, response in responses.items():
            print(f"\nEvaluating {resp_id}...")
            evaluation = evaluator.evaluate(resp_id, query, response)
            evaluations.append(evaluation)
        
        # Compare responses
        comparison = evaluator.compare(evaluations)
        
        print("\n[Individual Evaluations]")
        for eval in evaluations:
            print(f"\n{eval.candidate_id}:")
            print(f"  Overall Score: {eval.overall_score:.3f}")
            print(f"  Weighted Score: {eval.weighted_score:.3f}")
            print(f"  Pareto Optimal: {'Yes ✓' if eval.is_pareto_optimal else 'No'}")
            
            print("  Criterion Scores:")
            for cs in eval.criterion_scores:
                print(f"    {cs.criterion.value}: {cs.score:.3f} (weighted: {cs.weighted_score:.3f})")
        
        print("\n[Comparison Results]")
        print(f"\nBest Overall (by weighted score): {comparison.best_overall.candidate_id}")
        print(f"  Weighted Score: {comparison.best_overall.weighted_score:.3f}")
        
        print("\nBest per Criterion:")
        for criterion, best_eval in comparison.best_per_criterion.items():
            score = best_eval.get_score(criterion)
            print(f"  {criterion.value}: {best_eval.candidate_id} ({score:.3f})")
        
        print(f"\nPareto Frontier ({len(comparison.pareto_frontier)} candidates):")
        for eval in comparison.pareto_frontier:
            print(f"  {eval.candidate_id} - Weighted: {eval.weighted_score:.3f}, Overall: {eval.overall_score:.3f}")
        
        if comparison.trade_off_analysis:
            print("\nTrade-off Analysis:")
            for analysis in comparison.trade_off_analysis:
                print(f"  - {analysis}")
        
        # Test 2: Custom criteria evaluation
        print("\n" + "=" * 80)
        print("Test 2: Custom Criteria Configuration")
        print("=" * 80)
        
        # Prioritize safety and clarity over other factors
        safety_focused = MultiCriteriaEvaluator(
            criteria_weights={
                EvaluationCriterion.SAFETY: 2.0,
                EvaluationCriterion.CLARITY: 1.5,
                EvaluationCriterion.ACCURACY: 1.0,
                EvaluationCriterion.RELEVANCE: 0.8
            },
            temperature=0.2
        )
        
        query2 = "How do I handle conflicts at work?"
        response2 = "Stay calm, listen actively, focus on the issue not the person, seek common ground, and involve HR if needed."
        
        print(f"\nQuery: {query2}")
        print("\nPriority: Safety and Clarity")
        
        eval2 = safety_focused.evaluate("response_safety", query2, response2)
        
        print("\n[Evaluation Results]")
        print(f"Overall Score: {eval2.overall_score:.3f}")
        print(f"Weighted Score: {eval2.weighted_score:.3f}")
        
        print("\nDetailed Scores:")
        for cs in sorted(eval2.criterion_scores, key=lambda x: x.weighted_score, reverse=True):
            print(f"  {cs.criterion.value}:")
            print(f"    Raw: {cs.score:.3f} | Weight: {cs.weight:.2f} | Weighted: {cs.weighted_score:.3f}")
            print(f"    {cs.reasoning[:100]}...")
        
        # Test 3: Pareto frontier visualization
        print("\n" + "=" * 80)
        print("Test 3: Pareto Frontier Analysis")
        print("=" * 80)
        
        print("\n[Score Matrix]")
        print(f"{'Response':<12} ", end="")
        for criterion in evaluations[0].rankings.keys():
            print(f"{criterion.value[:8]:<10}", end="")
        print()
        
        for eval in evaluations:
            print(f"{eval.candidate_id:<12} ", end="")
            for criterion in eval.rankings.keys():
                score = eval.get_score(criterion) or 0
                marker = "★" if eval.is_pareto_optimal else " "
                print(f"{score:.2f}{marker:<7}", end="")
            print()
        
        print("\n★ = Pareto Optimal")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Multi-Criteria Evaluation provides:
    ✓ Multi-dimensional scoring
    ✓ Configurable criterion weights
    ✓ Pareto optimality detection
    ✓ Trade-off analysis
    ✓ Comprehensive comparison
    ✓ Balanced decision making
    
    This pattern excels at:
    - Model selection and comparison
    - Response ranking
    - Production optimization
    - A/B testing
    - Resource allocation
    - Quality vs cost trade-offs
    
    Standard evaluation criteria:
    - Accuracy: Factual correctness
    - Relevance: Addresses query
    - Safety: No harmful content
    - Completeness: Comprehensive coverage
    - Clarity: Clear communication
    - Efficiency: Concise
    - Cost: Resource usage
    - Latency: Speed
    - Creativity: Originality
    - Coherence: Logical flow
    
    Evaluation process:
    1. Define criteria and weights
    2. Score each criterion
    3. Calculate weighted scores
    4. Identify trade-offs
    5. Find Pareto frontier
    6. Compare candidates
    
    Pareto optimality:
    - Non-dominated solutions
    - No candidate better in all criteria
    - Multiple optimal trade-offs
    - No single "best" solution
    
    Weight configuration:
    - Higher weight = more important
    - Normalized to sum to 1.0
    - Domain-specific priorities
    - Stakeholder preferences
    
    Benefits:
    - Balanced: Consider multiple factors
    - Transparent: Clear criteria
    - Flexible: Configurable weights
    - Comprehensive: Holistic evaluation
    - Trade-offs: Explicit conflicts
    - Optimal: Pareto solutions
    
    Use Multi-Criteria Evaluation when:
    - Multiple objectives matter
    - Trade-offs exist
    - Need balanced decisions
    - Comparing alternatives
    - Production deployment
    - Stakeholder alignment needed
    
    Comparison with other patterns:
    - vs Self-Evaluation: Multiple dimensions vs single quality
    - vs Benchmark-Driven: Balanced evaluation vs single metric
    - vs Progressive Optimization: Evaluation vs improvement
    """)
    
    
    if __name__ == "__main__":
        demonstrate_multi_criteria_evaluation()
    
        

    Agentic Design Pattern: Multi Criteria Evaluation

    Pattern 043: Multi Criteria Evaluation – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Multi Criteria Evaluation pattern implementation.


    Pattern Description

    Pattern 043: Multi-Criteria Evaluation

    Description: Multi-Criteria Evaluation enables agents to evaluate outputs across multiple dimensions simultaneously, balancing trade-offs between competing objectives like accuracy, relevance, safety, cost, and latency. Uses weighted scoring, Pareto optimization, and multi-objective decision making.

    Components: – Criteria Definitions: Define evaluation dimensions – Multi-Dimensional Scorer: Evaluates across all criteria – Weight Manager: Balances criterion importance – Trade-off Analyzer: Identifies Pareto-optimal solutions – Aggregator: Combines scores into final decision

    Use Cases: – Model selection and comparison – Response ranking and filtering – Production system optimization – A/B testing and experimentation – Resource allocation decisions – Quality vs cost trade-offs

    LangChain Implementation: Uses multi-dimensional scoring chains to evaluate outputs across multiple criteria, identify trade-offs, and select Pareto-optimal solutions.


    File Statistics

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

    Module Documentation

    """
    Pattern 043: Multi-Criteria Evaluation
    
    Description:
        Multi-Criteria Evaluation enables agents to evaluate outputs across multiple
        dimensions simultaneously, balancing trade-offs between competing objectives
        like accuracy, relevance, safety, cost, and latency. Uses weighted scoring,
        Pareto optimization, and multi-objective decision making.
    
    Components:
        - Criteria Definitions: Define evaluation dimensions
        - Multi-Dimensional Scorer: Evaluates across all criteria
        - Weight Manager: Balances criterion importance
        - Trade-off Analyzer: Identifies Pareto-optimal solutions
        - Aggregator: Combines scores into final decision
    
    Use Cases:
        - Model selection and comparison
        - Response ranking and filtering
        - Production system optimization
        - A/B testing and experimentation
        - Resource allocation decisions
        - Quality vs cost trade-offs
    
    LangChain Implementation:
        Uses multi-dimensional scoring chains to evaluate outputs across multiple
        criteria, identify trade-offs, and select Pareto-optimal solutions.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    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
    import typing.Tuple
    

    Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – 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) – 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: EvaluationCriterion

    Lines 44-55

    Inherits from: Enum

    Purpose:

    Standard evaluation criteria.
    

    Class: CriterionScore

    Lines 59-70

    Purpose:

    Score for a single criterion.
    

    Methods:

    weighted_score(self)

    Lines 68-70

    Decorators: property

    Purpose:

    Calculate weighted score.
    

    Parameters:

    Returns: float


    Class: MultiCriteriaEvaluation

    Lines 74-97

    Purpose:

    Complete multi-criteria evaluation result.
    

    Methods:

    get_score(self, criterion)

    Lines 85-90

    Purpose:

    Get score for specific criterion.
    

    Parameters: – criterion: Criterion

    Returns: <ast.Subscript object at 0x000001A06D73B4D0>

    get_weighted_score(self, criterion)

    Lines 92-97

    Purpose:

    Get weighted score for specific criterion.
    

    Parameters: – criterion: Criterion

    Returns: <ast.Subscript object at 0x000001A06D73A510>


    Class: ComparisonResult

    Lines 101-107

    Purpose:

    Result of comparing multiple candidates.
    

    Class: MultiCriteriaEvaluator

    Lines 110-482

    Purpose:

    Evaluates outputs across multiple criteria with configurable weights.
    
    Features:
    - Multi-dimensional scoring
    - Configurable criterion weights
    - Pareto optimality detection
    - Trade-off analysis
    - Comprehensive comparison
    

    Methods:

    __init__(self, criteria_weights, temperature)

    Lines 122-199

    Purpose: Initialize the class instance with required parameters.

    Parameters: – criteria_weights: Criteria weights – temperature: Sampling temperature (0-1)

    evaluate_criterion(self, query, response, criterion)

    Lines 201-265

    Purpose:

    Evaluate response on a single criterion.
    
    Returns:
        CriterionScore object
    

    Parameters: – query: Input query or question – response: Response – criterion: Criterion

    Returns: CriterionScore

    evaluate(self, candidate_id, query, response, criteria)

    Lines 267-318

    Purpose:

    Perform multi-criteria evaluation.
    
    Args:
        candidate_id: Identifier for this candidate
        query: The original query
        response: The response to evaluate
        criteria: List of criteria to evaluate (uses all weighted criteria if None)
    
    Returns:
        MultiCriteriaEvaluation object
    

    Parameters: – candidate_id: Candidate id – query: Input query or question – response: Response – criteria: Criteria

    Returns: MultiCriteriaEvaluation

    _analyze_trade_offs(self, criterion_scores)

    Lines 320-348

    Purpose:

    Analyze trade-offs between criteria.
    

    Parameters: – criterion_scores: Criterion scores

    Returns: <ast.Subscript object at 0x000001A06D719210>

    compare(self, evaluations)

    Lines 350-406

    Purpose:

    Compare multiple candidates and identify Pareto frontier.
    
    Args:
        evaluations: List of evaluations to compare
    
    Returns:
        ComparisonResult with rankings and Pareto frontier
    

    Parameters: – evaluations: Evaluations

    Returns: ComparisonResult

    _find_pareto_frontier(self, evaluations)

    Lines 408-449

    Purpose:

    Find Pareto-optimal solutions (non-dominated candidates).
    
    A candidate is Pareto-optimal if no other candidate is better
    in all criteria.
    

    Parameters: – evaluations: Evaluations

    Returns: <ast.Subscript object at 0x000001A06D17D0D0>

    _compare_trade_offs(self, evaluations)

    Lines 451-482

    Purpose:

    Analyze trade-offs across all candidates.
    

    Parameters: – evaluations: Evaluations

    Returns: <ast.Subscript object at 0x000001A06D7508D0>


    Functions

    demonstrate_multi_criteria_evaluation()

    Lines 485-693

    Purpose:

    Demonstrates multi-criteria evaluation with trade-off analysis.
    

    Key Concepts Summary

    Pattern Overview

    The Multi Criteria Evaluation 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 042
    • Pattern 044

    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 Multi Criteria Evaluation 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: 043_multi_criteria_evaluation.py
    Total Lines: 698
    Pattern Category: AI Agent Design Pattern #043


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Progressive Optimization

    
        """
    Pattern 042: Progressive Optimization
    
    Description:
        Progressive Optimization enables agents to iteratively improve solutions
        through multiple generations using optimization strategies like hill climbing,
        gradient-based methods, or evolutionary algorithms. Each iteration refines
        the solution based on fitness evaluation until convergence or max iterations.
    
    Components:
        - Solution Generator: Creates initial and improved solutions
        - Fitness Evaluator: Scores solution quality
        - Optimization Strategy: Determines how to improve (hill climbing, genetic, etc.)
        - Convergence Detector: Identifies when to stop
        - History Tracker: Maintains optimization trajectory
    
    Use Cases:
        - Creative content optimization (writing, design)
        - Code optimization and refactoring
        - Prompt engineering and tuning
        - Solution design and improvement
        - Configuration optimization
        - Performance tuning
    
    LangChain Implementation:
        Uses iterative chains with fitness evaluation and improvement strategies
        to progressively optimize solutions until convergence or target quality.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime
    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 OptimizationStrategy(Enum):
        """Optimization strategies available."""
        HILL_CLIMBING = "hill_climbing"
        SIMULATED_ANNEALING = "simulated_annealing"
        GENETIC = "genetic"
        GRADIENT_BASED = "gradient_based"
        RANDOM_SEARCH = "random_search"
    
    
    class ConvergenceStatus(Enum):
        """Status of optimization convergence."""
        RUNNING = "running"
        CONVERGED = "converged"
        MAX_ITERATIONS = "max_iterations"
        TARGET_REACHED = "target_reached"
        STAGNATED = "stagnated"
    
    
    @dataclass
    class Solution:
        """A solution in the optimization process."""
        content: str
        fitness: float
        generation: int
        timestamp: datetime = field(default_factory=datetime.now)
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class OptimizationResult:
        """Result of optimization process."""
        initial_solution: Solution
        final_solution: Solution
        history: List[Solution]
        convergence_status: ConvergenceStatus
        total_iterations: int
        improvement: float
        strategy: OptimizationStrategy
    
    
    class ProgressiveOptimizationAgent:
        """
        Agent that progressively optimizes solutions through iterations.
        
        Features:
        - Multiple optimization strategies
        - Fitness-based improvement
        - Convergence detection
        - History tracking
        - Configurable parameters
        """
        
        def __init__(
            self,
            strategy: OptimizationStrategy = OptimizationStrategy.HILL_CLIMBING,
            max_iterations: int = 10,
            target_fitness: float = 0.9,
            convergence_threshold: float = 0.01,
            stagnation_limit: int = 3,
            temperature: float = 0.7
        ):
            self.strategy = strategy
            self.max_iterations = max_iterations
            self.target_fitness = target_fitness
            self.convergence_threshold = convergence_threshold
            self.stagnation_limit = stagnation_limit
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            
            # Initial solution generation
            self.initial_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert at creating solutions. Generate a high-quality initial solution."),
                ("user", "{objective}\n\nProvide your initial solution:")
            ])
            
            # Improvement prompt
            self.improvement_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an optimization expert. Improve the current solution.
    
    Current fitness: {fitness}
    Target fitness: {target_fitness}
    Strategy: {strategy}
    
    Focus on:
    - Addressing weaknesses identified
    - Enhancing strengths
    - Moving toward target fitness
    - Making meaningful improvements"""),
                ("user", """Objective: {objective}
    
    Current Solution:
    {current_solution}
    
    Feedback for improvement:
    {feedback}
    
    Provide an improved solution:""")
            ])
            
            # Fitness evaluation prompt
            self.evaluation_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a solution evaluator. Rate the solution's quality on a scale of 0.0 to 1.0.
    
    Criteria:
    - Effectiveness: Does it meet the objective?
    - Quality: Is it well-crafted?
    - Completeness: Does it address all aspects?
    - Clarity: Is it clear and understandable?
    
    Provide:
    FITNESS_SCORE: [0.0-1.0]
    STRENGTHS: [List strengths]
    WEAKNESSES: [List weaknesses]
    SUGGESTIONS: [Improvement suggestions]"""),
                ("user", """Objective: {objective}
    
    Solution to evaluate:
    {solution}
    
    Evaluate:""")
            ])
            
            # Optimization history
            self.optimizations: List[OptimizationResult] = []
        
        def generate_initial_solution(self, objective: str) -> str:
            """Generate initial solution."""
            chain = self.initial_prompt | self.llm | StrOutputParser()
            return chain.invoke({"objective": objective})
        
        def evaluate_fitness(
            self,
            objective: str,
            solution: str
        ) -> Tuple[float, Dict[str, List[str]]]:
            """
            Evaluate solution fitness and get feedback.
            
            Returns:
                Tuple of (fitness_score, feedback_dict)
            """
            chain = self.evaluation_prompt | self.llm | StrOutputParser()
            evaluation = chain.invoke({
                "objective": objective,
                "solution": solution
            })
            
            # Parse evaluation
            fitness = 0.5
            feedback = {
                "strengths": [],
                "weaknesses": [],
                "suggestions": []
            }
            
            lines = evaluation.split('\n')
            current_section = None
            
            for line in lines:
                line = line.strip()
                if line.startswith("FITNESS_SCORE:"):
                    try:
                        fitness = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                elif line.startswith("STRENGTHS:"):
                    current_section = "strengths"
                elif line.startswith("WEAKNESSES:"):
                    current_section = "weaknesses"
                elif line.startswith("SUGGESTIONS:"):
                    current_section = "suggestions"
                elif line and line.startswith('-') and current_section:
                    item = line[1:].strip()
                    feedback[current_section].append(item)
            
            return fitness, feedback
        
        def improve_solution(
            self,
            objective: str,
            current_solution: str,
            fitness: float,
            feedback: Dict[str, List[str]]
        ) -> str:
            """
            Generate improved solution based on feedback.
            
            Returns:
                Improved solution
            """
            # Format feedback
            feedback_text = ""
            if feedback.get("weaknesses"):
                feedback_text += "Weaknesses to address:\n"
                feedback_text += "\n".join([f"- {w}" for w in feedback["weaknesses"]])
                feedback_text += "\n\n"
            
            if feedback.get("suggestions"):
                feedback_text += "Suggestions for improvement:\n"
                feedback_text += "\n".join([f"- {s}" for s in feedback["suggestions"]])
            
            chain = self.improvement_prompt | self.llm | StrOutputParser()
            improved = chain.invoke({
                "objective": objective,
                "current_solution": current_solution,
                "fitness": fitness,
                "target_fitness": self.target_fitness,
                "strategy": self.strategy.value,
                "feedback": feedback_text
            })
            
            return improved
        
        def check_convergence(
            self,
            history: List[Solution],
            current_fitness: float
        ) -> ConvergenceStatus:
            """
            Check if optimization has converged.
            
            Returns:
                ConvergenceStatus
            """
            # Check max iterations
            if len(history) >= self.max_iterations:
                return ConvergenceStatus.MAX_ITERATIONS
            
            # Check target reached
            if current_fitness >= self.target_fitness:
                return ConvergenceStatus.TARGET_REACHED
            
            # Check stagnation
            if len(history) >= self.stagnation_limit + 1:
                recent_improvements = [
                    history[i].fitness - history[i-1].fitness
                    for i in range(-self.stagnation_limit, 0)
                ]
                if all(imp = 2:
                improvement = history[-1].fitness - history[-2].fitness
                if 0 <= improvement  OptimizationResult:
            """
            Progressively optimize solution.
            
            Steps:
            1. Generate or use initial solution
            2. Evaluate fitness
            3. Check convergence
            4. Generate improvement
            5. Repeat until convergence
            
            Args:
                objective: The optimization objective
                initial_solution: Optional starting solution
                
            Returns:
                OptimizationResult with optimization trajectory
            """
            # Step 1: Get initial solution
            if initial_solution is None:
                initial_solution = self.generate_initial_solution(objective)
            
            # Evaluate initial fitness
            fitness, feedback = self.evaluate_fitness(objective, initial_solution)
            
            current_solution = Solution(
                content=initial_solution,
                fitness=fitness,
                generation=0,
                metadata={"feedback": feedback}
            )
            
            history = [current_solution]
            
            # Optimization loop
            iteration = 1
            status = ConvergenceStatus.RUNNING
            
            while status == ConvergenceStatus.RUNNING:
                # Step 3: Check convergence
                status = self.check_convergence(history, current_solution.fitness)
                if status != ConvergenceStatus.RUNNING:
                    break
                
                # Step 4: Generate improvement
                improved_content = self.improve_solution(
                    objective,
                    current_solution.content,
                    current_solution.fitness,
                    feedback
                )
                
                # Evaluate improved solution
                new_fitness, new_feedback = self.evaluate_fitness(
                    objective,
                    improved_content
                )
                
                # Accept improvement (hill climbing) or based on strategy
                if new_fitness >= current_solution.fitness or self.strategy != OptimizationStrategy.HILL_CLIMBING:
                    current_solution = Solution(
                        content=improved_content,
                        fitness=new_fitness,
                        generation=iteration,
                        metadata={"feedback": new_feedback}
                    )
                    feedback = new_feedback
                else:
                    # For hill climbing, if no improvement, mark stagnation
                    current_solution = Solution(
                        content=current_solution.content,
                        fitness=current_solution.fitness,
                        generation=iteration,
                        metadata={"feedback": feedback, "stagnated": True}
                    )
                
                history.append(current_solution)
                iteration += 1
            
            # Create result
            improvement = history[-1].fitness - history[0].fitness
            
            result = OptimizationResult(
                initial_solution=history[0],
                final_solution=history[-1],
                history=history,
                convergence_status=status,
                total_iterations=len(history) - 1,
                improvement=improvement,
                strategy=self.strategy
            )
            
            self.optimizations.append(result)
            
            return result
        
        def get_optimization_statistics(self) -> Dict[str, Any]:
            """Get statistics about optimizations."""
            if not self.optimizations:
                return {"total_optimizations": 0}
            
            avg_iterations = sum(o.total_iterations for o in self.optimizations) / len(self.optimizations)
            avg_improvement = sum(o.improvement for o in self.optimizations) / len(self.optimizations)
            
            convergence_counts = {}
            for status in ConvergenceStatus:
                count = sum(1 for o in self.optimizations if o.convergence_status == status)
                convergence_counts[status.value] = count
            
            return {
                "total_optimizations": len(self.optimizations),
                "average_iterations": avg_iterations,
                "average_improvement": avg_improvement,
                "convergence_breakdown": convergence_counts
            }
    
    
    def demonstrate_progressive_optimization():
        """
        Demonstrates progressive optimization for iterative improvement.
        """
        print("=" * 80)
        print("PROGRESSIVE OPTIMIZATION DEMONSTRATION")
        print("=" * 80)
        
        # Test 1: Optimize a product description
        print("\n" + "=" * 80)
        print("Test 1: Product Description Optimization")
        print("=" * 80)
        
        agent1 = ProgressiveOptimizationAgent(
            strategy=OptimizationStrategy.HILL_CLIMBING,
            max_iterations=5,
            target_fitness=0.85,
            convergence_threshold=0.05,
            temperature=0.8
        )
        
        objective1 = """Create a compelling product description for a smart water bottle that:
    - Tracks hydration levels
    - Reminds users to drink water
    - Syncs with fitness apps
    - Has temperature control
    
    Make it persuasive, concise, and customer-focused."""
        
        print(f"\nObjective:\n{objective1}")
        
        result1 = agent1.optimize(objective1)
        
        print("\n[Optimization Trajectory]")
        print(f"Strategy: {result1.strategy.value}")
        print(f"Total Iterations: {result1.total_iterations}")
        print(f"Convergence Status: {result1.convergence_status.value}")
        print(f"Improvement: {result1.improvement:.3f}")
        
        print("\n[Fitness Progress]")
        for i, solution in enumerate(result1.history):
            indicator = "📈" if i > 0 and solution.fitness > result1.history[i-1].fitness else "📊"
            print(f"Gen {solution.generation}: {solution.fitness:.3f} {indicator}")
        
        print("\n[Initial Solution]")
        print(result1.initial_solution.content[:200] + "...")
        print(f"Fitness: {result1.initial_solution.fitness:.3f}")
        
        print("\n[Final Solution]")
        print(result1.final_solution.content[:200] + "...")
        print(f"Fitness: {result1.final_solution.fitness:.3f}")
        
        # Test 2: Optimize code explanation
        print("\n" + "=" * 80)
        print("Test 2: Code Explanation Optimization")
        print("=" * 80)
        
        agent2 = ProgressiveOptimizationAgent(
            strategy=OptimizationStrategy.HILL_CLIMBING,
            max_iterations=4,
            target_fitness=0.9,
            temperature=0.7
        )
        
        objective2 = """Explain how binary search works in a way that:
    - Is accessible to beginners
    - Uses clear analogies
    - Includes time complexity
    - Has a practical example
    - Is engaging and memorable"""
        
        print(f"\nObjective: Optimize explanation quality")
        
        result2 = agent2.optimize(objective2)
        
        print("\n[Optimization Summary]")
        print(f"Iterations: {result2.total_iterations}")
        print(f"Status: {result2.convergence_status.value}")
        print(f"Initial Fitness: {result2.initial_solution.fitness:.3f}")
        print(f"Final Fitness: {result2.final_solution.fitness:.3f}")
        print(f"Improvement: {result2.improvement:.3f} ({result2.improvement/result2.initial_solution.fitness*100:.1f}%)")
        
        print("\n[Fitness Evolution]")
        for sol in result2.history:
            bar = "█" * int(sol.fitness * 40)
            print(f"Gen {sol.generation}: {bar} {sol.fitness:.3f}")
        
        # Test 3: Multiple objectives with comparison
        print("\n" + "=" * 80)
        print("Test 3: Email Subject Line Optimization")
        print("=" * 80)
        
        agent3 = ProgressiveOptimizationAgent(
            strategy=OptimizationStrategy.HILL_CLIMBING,
            max_iterations=6,
            target_fitness=0.88,
            stagnation_limit=2,
            temperature=0.8
        )
        
        objective3 = """Create an email subject line for a webinar announcement that:
    - Is under 50 characters
    - Creates urgency
    - Clearly states value
    - Has high open rate potential
    - Avoids spam triggers"""
        
        print(f"\nObjective: Optimize email subject line")
        
        result3 = agent3.optimize(objective3)
        
        print("\n[Optimization Details]")
        print(f"Strategy: {result3.strategy.value}")
        print(f"Convergence: {result3.convergence_status.value}")
        print(f"Generations: {result3.total_iterations}")
        
        print("\n[All Generations]")
        for sol in result3.history:
            print(f"\nGeneration {sol.generation} (Fitness: {sol.fitness:.3f}):")
            print(f"  {sol.content[:100]}")
            
            if "feedback" in sol.metadata:
                feedback = sol.metadata["feedback"]
                if feedback.get("weaknesses"):
                    print(f"  Weaknesses: {len(feedback['weaknesses'])}")
                if feedback.get("suggestions"):
                    print(f"  Suggestions: {len(feedback['suggestions'])}")
        
        # Combined statistics
        print("\n" + "=" * 80)
        print("Optimization Statistics")
        print("=" * 80)
        
        # Manually combine stats from all agents
        all_optimizations = (
            agent1.optimizations + 
            agent2.optimizations + 
            agent3.optimizations
        )
        
        total_iterations = sum(o.total_iterations for o in all_optimizations)
        avg_iterations = total_iterations / len(all_optimizations)
        avg_improvement = sum(o.improvement for o in all_optimizations) / len(all_optimizations)
        
        print(f"\nTotal Optimizations: {len(all_optimizations)}")
        print(f"Average Iterations: {avg_iterations:.1f}")
        print(f"Average Improvement: {avg_improvement:.3f}")
        
        convergence_counts = {}
        for status in ConvergenceStatus:
            count = sum(1 for o in all_optimizations if o.convergence_status == status)
            if count > 0:
                convergence_counts[status.value] = count
        
        print("\nConvergence Breakdown:")
        for status, count in convergence_counts.items():
            print(f"  {status}: {count}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Progressive Optimization provides:
    ✓ Iterative solution improvement
    ✓ Multiple optimization strategies
    ✓ Fitness-based guidance
    ✓ Convergence detection
    ✓ History tracking
    ✓ Quality metrics
    
    This pattern excels at:
    - Creative content optimization
    - Code and design refinement
    - Prompt engineering
    - Configuration tuning
    - Solution design
    - Quality improvement
    
    Optimization strategies:
    1. Hill Climbing: Accept only improvements
    2. Simulated Annealing: Accept some worse solutions
    3. Genetic: Evolve population of solutions
    4. Gradient-Based: Follow improvement gradient
    5. Random Search: Explore solution space
    
    Optimization process:
    1. Generate initial solution
    2. Evaluate fitness
    3. Check convergence
    4. Generate improvement
    5. Evaluate new solution
    6. Accept or reject
    7. Repeat until convergence
    
    Convergence conditions:
    - Target fitness reached
    - Maximum iterations
    - Improvement stagnation
    - Convergence threshold met
    
    Fitness evaluation criteria:
    - Effectiveness: Meets objective
    - Quality: Well-crafted
    - Completeness: Comprehensive
    - Clarity: Understandable
    
    Benefits:
    - Quality: Iterative improvement
    - Transparency: Track progress
    - Convergence: Automatic stopping
    - Flexibility: Multiple strategies
    - Metrics: Quantifiable improvement
    - History: Full trajectory
    
    Configuration parameters:
    - max_iterations: Stop after N iterations
    - target_fitness: Goal quality score
    - convergence_threshold: Min improvement
    - stagnation_limit: Iterations without improvement
    - strategy: Optimization approach
    
    Use Progressive Optimization when:
    - Need iterative improvement
    - Quality is quantifiable
    - Multiple refinements beneficial
    - Want optimization history
    - Need automatic convergence
    - Exploring solution space
    
    Comparison with other patterns:
    - vs Iterative Refinement: More strategic, fitness-guided
    - vs Self-Evaluation: Focused on optimization, not just evaluation
    - vs Feedback Loops: Systematic optimization vs learning
    """)
    
    
    if __name__ == "__main__":
        demonstrate_progressive_optimization()
    
        

    Agentic Design Pattern: Progressive Optimization

    Pattern 042: Progressive Optimization – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 042: Progressive Optimization

    Description: Progressive Optimization enables agents to iteratively improve solutions through multiple generations using optimization strategies like hill climbing, gradient-based methods, or evolutionary algorithms. Each iteration refines the solution based on fitness evaluation until convergence or max iterations.

    Components: – Solution Generator: Creates initial and improved solutions – Fitness Evaluator: Scores solution quality – Optimization Strategy: Determines how to improve (hill climbing, genetic, etc.) – Convergence Detector: Identifies when to stop – History Tracker: Maintains optimization trajectory

    Use Cases: – Creative content optimization (writing, design) – Code optimization and refactoring – Prompt engineering and tuning – Solution design and improvement – Configuration optimization – Performance tuning

    LangChain Implementation: Uses iterative chains with fitness evaluation and improvement strategies to progressively optimize solutions until convergence or target quality.


    File Statistics

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

    Module Documentation

    """
    Pattern 042: Progressive Optimization
    
    Description:
        Progressive Optimization enables agents to iteratively improve solutions
        through multiple generations using optimization strategies like hill climbing,
        gradient-based methods, or evolutionary algorithms. Each iteration refines
        the solution based on fitness evaluation until convergence or max iterations.
    
    Components:
        - Solution Generator: Creates initial and improved solutions
        - Fitness Evaluator: Scores solution quality
        - Optimization Strategy: Determines how to improve (hill climbing, genetic, etc.)
        - Convergence Detector: Identifies when to stop
        - History Tracker: Maintains optimization trajectory
    
    Use Cases:
        - Creative content optimization (writing, design)
        - Code optimization and refactoring
        - Prompt engineering and tuning
        - Solution design and improvement
        - Configuration optimization
        - Performance tuning
    
    LangChain Implementation:
        Uses iterative chains with fitness evaluation and improvement strategies
        to progressively optimize solutions until convergence or target quality.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import typing.Any
    import typing.Callable
    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 – datetime.datetime: Date and time handling – 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.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) – 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: OptimizationStrategy

    Lines 44-50

    Inherits from: Enum

    Purpose:

    Optimization strategies available.
    

    Class: ConvergenceStatus

    Lines 53-59

    Inherits from: Enum

    Purpose:

    Status of optimization convergence.
    

    Class: Solution

    Lines 63-69

    Purpose:

    A solution in the optimization process.
    

    Class: OptimizationResult

    Lines 73-81

    Purpose:

    Result of optimization process.
    

    Class: ProgressiveOptimizationAgent

    Lines 84-410

    Purpose:

    Agent that progressively optimizes solutions through iterations.
    
    Features:
    - Multiple optimization strategies
    - Fitness-based improvement
    - Convergence detection
    - History tracking
    - Configurable parameters
    

    Methods:

    __init__(self, strategy, max_iterations, target_fitness, convergence_threshold, stagnation_limit, temperature)

    Lines 96-166

    Purpose: Initialize the class instance with required parameters.

    Parameters: – strategy: Strategy – max_iterations: Max iterations – target_fitness: Target fitness – convergence_threshold: Convergence threshold – stagnation_limit: Stagnation limit – temperature: Sampling temperature (0-1)

    generate_initial_solution(self, objective)

    Lines 168-171

    Purpose:

    Generate initial solution.
    

    Parameters: – objective: Objective

    Returns: str

    evaluate_fitness(self, objective, solution)

    Lines 173-218

    Purpose:

    Evaluate solution fitness and get feedback.
    
    Returns:
        Tuple of (fitness_score, feedback_dict)
    

    Parameters: – objective: Objective – solution: Solution

    Returns: <ast.Subscript object at 0x000001A06D62A790>

    improve_solution(self, objective, current_solution, fitness, feedback)

    Lines 220-254

    Purpose:

    Generate improved solution based on feedback.
    
    Returns:
        Improved solution
    

    Parameters: – objective: Objective – current_solution: Current solution – fitness: Fitness – feedback: Feedback

    Returns: str

    check_convergence(self, history, current_fitness)

    Lines 256-290

    Purpose:

    Check if optimization has converged.
    
    Returns:
        ConvergenceStatus
    

    Parameters: – history: History – current_fitness: Current fitness

    Returns: ConvergenceStatus

    optimize(self, objective, initial_solution)

    Lines 292-390

    Purpose:

    Progressively optimize solution.
    
    Steps:
    1. Generate or use initial solution
    2. Evaluate fitness
    3. Check convergence
    4. Generate improvement
    5. Repeat until convergence
    
    Args:
        objective: The optimization objective
        initial_solution: Optional starting solution
    
    Returns:
        OptimizationResult with optimization trajectory
    

    Parameters: – objective: Objective – initial_solution: Initial solution

    Returns: OptimizationResult

    get_optimization_statistics(self)

    Lines 392-410

    Purpose:

    Get statistics about optimizations.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D627190>


    Functions

    demonstrate_progressive_optimization()

    Lines 413-647

    Purpose:

    Demonstrates progressive optimization for iterative improvement.
    

    Key Concepts Summary

    Pattern Overview

    The Progressive Optimization 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 041
    • Pattern 043

    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 Progressive Optimization 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: 042_progressive_optimization.py
    Total Lines: 652
    Pattern Category: AI Agent Design Pattern #042


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Chain Of Verification

    
        """
    Pattern 041: Chain-of-Verification (CoVe)
    
    Description:
        Chain-of-Verification enables agents to reduce hallucinations and improve
        factual accuracy by generating verification questions to check their own
        answers. The agent generates an initial response, creates verification
        questions, answers them, and then revises the original response based on
        the verification results.
    
    Components:
        - Initial Generator: Creates initial response
        - Verification Question Generator: Creates questions to verify response
        - Verification Answerer: Answers verification questions
        - Consistency Checker: Checks for contradictions
        - Response Reviser: Updates response based on verification
    
    Use Cases:
        - Fact-checking automation
        - Reducing hallucinations
        - Improving factual accuracy
        - Knowledge-intensive tasks
        - Research and analysis
        - Educational content
    
    LangChain Implementation:
        Uses multi-step verification chains to check responses, identify
        inconsistencies, and revise outputs for improved accuracy.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from datetime import datetime
    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()
    
    
    @dataclass
    class VerificationQuestion:
        """A question to verify part of the response."""
        question: str
        expected_answer: Optional[str] = None
        actual_answer: Optional[str] = None
        is_consistent: Optional[bool] = None
        relevance: float = 1.0
    
    
    @dataclass
    class VerificationResult:
        """Result of verification process."""
        original_response: str
        verification_questions: List[VerificationQuestion]
        inconsistencies: List[str]
        verified_facts: List[str]
        revised_response: str
        confidence_score: float
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class ChainOfVerificationAgent:
        """
        Agent that verifies its own responses through questioning.
        
        Features:
        - Generates verification questions
        - Checks factual consistency
        - Identifies contradictions
        - Revises responses
        - Improves accuracy
        """
        
        def __init__(
            self,
            response_temperature: float = 0.7,
            verification_temperature: float = 0.3,
            num_verification_questions: int = 5
        ):
            self.response_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=response_temperature)
            self.verification_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=verification_temperature)
            self.num_verification_questions = num_verification_questions
            
            # Initial response prompt
            self.response_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a knowledgeable assistant. Provide detailed, accurate answers."),
                ("user", "{query}")
            ])
            
            # Verification question generation prompt
            self.verification_questions_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a fact-checker. Generate verification questions to check the accuracy of this response.
    
    Create {num_questions} specific, factual questions that verify key claims in the response.
    Each question should check a different aspect or claim.
    
    Format:
    1. [Question 1]
    2. [Question 2]
    ..."""),
                ("user", "Query: {query}\n\nResponse to verify:\n{response}\n\nGenerate verification questions:")
            ])
            
            # Answer verification questions prompt
            self.answer_verification_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert answering verification questions. Provide concise, factual answers."),
                ("user", "Question: {question}\n\nProvide a factual answer:")
            ])
            
            # Consistency check prompt
            self.consistency_prompt = ChatPromptTemplate.from_messages([
                ("system", """Compare the original response with the verification answers.
    
    Identify:
    1. Inconsistencies or contradictions
    2. Verified facts that are consistent
    3. Overall confidence in the original response (0.0-1.0)
    
    Format:
    INCONSISTENCIES:
    - [List any inconsistencies]
    
    VERIFIED_FACTS:
    - [List verified facts]
    
    CONFIDENCE: [0.0-1.0]"""),
                ("user", """Original Response:
    {original_response}
    
    Verification Q&A:
    {verification_qa}
    
    Analyze consistency:""")
            ])
            
            # Revision prompt
            self.revision_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are an editor improving response accuracy.
    
    Revise the original response to:
    1. Fix any inconsistencies identified
    2. Maintain verified facts
    3. Remove or correct uncertain information
    4. Keep the response helpful and complete"""),
                ("user", """Query: {query}
    
    Original Response:
    {original_response}
    
    Issues Found:
    {inconsistencies}
    
    Provide revised response:""")
            ])
            
            # Verification history
            self.verifications: List[VerificationResult] = []
        
        def generate_response(self, query: str) -> str:
            """Generate initial response to query."""
            chain = self.response_prompt | self.response_llm | StrOutputParser()
            return chain.invoke({"query": query})
        
        def generate_verification_questions(
            self,
            query: str,
            response: str
        ) -> List[VerificationQuestion]:
            """
            Generate questions to verify the response.
            
            Returns:
                List of VerificationQuestion objects
            """
            chain = self.verification_questions_prompt | self.verification_llm | StrOutputParser()
            questions_text = chain.invoke({
                "query": query,
                "response": response,
                "num_questions": self.num_verification_questions
            })
            
            # Parse questions
            questions = []
            for line in questions_text.split('\n'):
                line = line.strip()
                if line and (line[0].isdigit() or line.startswith('-')):
                    # Remove numbering
                    question = line.split('.', 1)[-1].strip()
                    if question.startswith('-'):
                        question = question[1:].strip()
                    
                    if question:
                        questions.append(VerificationQuestion(question=question))
            
            return questions[:self.num_verification_questions]
        
        def answer_verification_questions(
            self,
            questions: List[VerificationQuestion]
        ) -> List[VerificationQuestion]:
            """
            Answer verification questions independently.
            
            Returns:
                List of VerificationQuestion objects with answers
            """
            chain = self.answer_verification_prompt | self.verification_llm | StrOutputParser()
            
            for question in questions:
                answer = chain.invoke({"question": question.question})
                question.actual_answer = answer
            
            return questions
        
        def check_consistency(
            self,
            original_response: str,
            questions: List[VerificationQuestion]
        ) -> Tuple[List[str], List[str], float]:
            """
            Check consistency between original response and verification answers.
            
            Returns:
                Tuple of (inconsistencies, verified_facts, confidence)
            """
            # Format verification Q&A
            verification_qa = "\n\n".join([
                f"Q: {q.question}\nA: {q.actual_answer}"
                for q in questions if q.actual_answer
            ])
            
            chain = self.consistency_prompt | self.verification_llm | StrOutputParser()
            result = chain.invoke({
                "original_response": original_response,
                "verification_qa": verification_qa
            })
            
            # Parse result
            inconsistencies = []
            verified_facts = []
            confidence = 0.5
            
            lines = result.split('\n')
            current_section = None
            
            for line in lines:
                line = line.strip()
                if line.startswith("INCONSISTENCIES:"):
                    current_section = "inconsistencies"
                elif line.startswith("VERIFIED_FACTS:"):
                    current_section = "verified_facts"
                elif line.startswith("CONFIDENCE:"):
                    try:
                        confidence = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                    current_section = None
                elif line and line.startswith('-') and current_section:
                    item = line[1:].strip()
                    if current_section == "inconsistencies":
                        inconsistencies.append(item)
                    elif current_section == "verified_facts":
                        verified_facts.append(item)
            
            return inconsistencies, verified_facts, confidence
        
        def revise_response(
            self,
            query: str,
            original_response: str,
            inconsistencies: List[str]
        ) -> str:
            """
            Revise response to fix inconsistencies.
            
            Returns:
                Revised response
            """
            if not inconsistencies:
                return original_response
            
            inconsistencies_text = "\n".join([f"- {inc}" for inc in inconsistencies])
            
            chain = self.revision_prompt | self.response_llm | StrOutputParser()
            revised = chain.invoke({
                "query": query,
                "original_response": original_response,
                "inconsistencies": inconsistencies_text
            })
            
            return revised
        
        def verify_response(
            self,
            query: str,
            response: Optional[str] = None
        ) -> VerificationResult:
            """
            Complete Chain-of-Verification process.
            
            Steps:
            1. Generate initial response (or use provided)
            2. Generate verification questions
            3. Answer verification questions
            4. Check consistency
            5. Revise if needed
            
            Args:
                query: The query to answer
                response: Optional pre-generated response
                
            Returns:
                VerificationResult object
            """
            # Step 1: Generate or use provided response
            if response is None:
                response = self.generate_response(query)
            
            original_response = response
            
            # Step 2: Generate verification questions
            questions = self.generate_verification_questions(query, response)
            
            # Step 3: Answer verification questions
            questions = self.answer_verification_questions(questions)
            
            # Step 4: Check consistency
            inconsistencies, verified_facts, confidence = self.check_consistency(
                original_response,
                questions
            )
            
            # Step 5: Revise if inconsistencies found
            if inconsistencies:
                revised_response = self.revise_response(query, original_response, inconsistencies)
            else:
                revised_response = original_response
            
            result = VerificationResult(
                original_response=original_response,
                verification_questions=questions,
                inconsistencies=inconsistencies,
                verified_facts=verified_facts,
                revised_response=revised_response,
                confidence_score=confidence
            )
            
            self.verifications.append(result)
            
            return result
        
        def get_verification_statistics(self) -> Dict[str, Any]:
            """Get statistics about verifications."""
            if not self.verifications:
                return {"total_verifications": 0}
            
            total_questions = sum(len(v.verification_questions) for v in self.verifications)
            total_inconsistencies = sum(len(v.inconsistencies) for v in self.verifications)
            avg_confidence = sum(v.confidence_score for v in self.verifications) / len(self.verifications)
            revision_rate = sum(1 for v in self.verifications if v.revised_response != v.original_response) / len(self.verifications)
            
            return {
                "total_verifications": len(self.verifications),
                "total_questions_generated": total_questions,
                "total_inconsistencies_found": total_inconsistencies,
                "average_confidence": avg_confidence,
                "revision_rate": revision_rate
            }
    
    
    def demonstrate_chain_of_verification():
        """
        Demonstrates Chain-of-Verification for reducing hallucinations.
        """
        print("=" * 80)
        print("CHAIN-OF-VERIFICATION DEMONSTRATION")
        print("=" * 80)
        
        # Create CoVe agent
        agent = ChainOfVerificationAgent(
            response_temperature=0.7,
            verification_temperature=0.3,
            num_verification_questions=4
        )
        
        # Test 1: Factual query with potential for errors
        print("\n" + "=" * 80)
        print("Test 1: Historical Facts Verification")
        print("=" * 80)
        
        query1 = "When did World War II end and what were the key events?"
        print(f"\nQuery: {query1}")
        
        result1 = agent.verify_response(query1)
        
        print("\n[Original Response]")
        print(result1.original_response[:300] + "..." if len(result1.original_response) > 300 else result1.original_response)
        
        print("\n[Verification Questions]")
        for i, q in enumerate(result1.verification_questions, 1):
            print(f"{i}. {q.question}")
            if q.actual_answer:
                print(f"   Answer: {q.actual_answer[:100]}...")
        
        if result1.inconsistencies:
            print("\n[Inconsistencies Found]")
            for inc in result1.inconsistencies:
                print(f"  - {inc}")
        
        if result1.verified_facts:
            print("\n[Verified Facts]")
            for fact in result1.verified_facts[:3]:
                print(f"  - {fact}")
        
        print(f"\n[Confidence Score]: {result1.confidence_score:.2f}")
        
        if result1.revised_response != result1.original_response:
            print("\n[Revised Response]")
            print(result1.revised_response[:300] + "..." if len(result1.revised_response) > 300 else result1.revised_response)
        else:
            print("\n[No Revision Needed]")
        
        # Test 2: Scientific query
        print("\n" + "=" * 80)
        print("Test 2: Scientific Facts Verification")
        print("=" * 80)
        
        query2 = "How does photosynthesis work in plants?"
        print(f"\nQuery: {query2}")
        
        result2 = agent.verify_response(query2)
        
        print("\n[Original Response]")
        print(result2.original_response[:250] + "..." if len(result2.original_response) > 250 else result2.original_response)
        
        print("\n[Verification Questions Generated]")
        for i, q in enumerate(result2.verification_questions, 1):
            print(f"{i}. {q.question}")
        
        print(f"\n[Verification Results]")
        print(f"Inconsistencies Found: {len(result2.inconsistencies)}")
        print(f"Verified Facts: {len(result2.verified_facts)}")
        print(f"Confidence Score: {result2.confidence_score:.2f}")
        
        # Test 3: Potentially tricky query
        print("\n" + "=" * 80)
        print("Test 3: Complex Query with Verification")
        print("=" * 80)
        
        query3 = "What are the main differences between machine learning and deep learning?"
        print(f"\nQuery: {query3}")
        
        result3 = agent.verify_response(query3)
        
        print("\n[Verification Process]")
        print(f"Questions Generated: {len(result3.verification_questions)}")
        print(f"Inconsistencies: {len(result3.inconsistencies)}")
        print(f"Verified Facts: {len(result3.verified_facts)}")
        print(f"Confidence: {result3.confidence_score:.2f}")
        print(f"Response Revised: {'Yes' if result3.revised_response != result3.original_response else 'No'}")
        
        print("\n[Sample Verification Q&A]")
        for q in result3.verification_questions[:2]:
            print(f"\nQ: {q.question}")
            print(f"A: {q.actual_answer[:150]}..." if q.actual_answer and len(q.actual_answer) > 150 else f"A: {q.actual_answer}")
        
        # Show statistics
        print("\n" + "=" * 80)
        print("Verification Statistics")
        print("=" * 80)
        
        stats = agent.get_verification_statistics()
        print(f"\nTotal Verifications: {stats['total_verifications']}")
        print(f"Total Questions Generated: {stats['total_questions_generated']}")
        print(f"Total Inconsistencies Found: {stats['total_inconsistencies_found']}")
        print(f"Average Confidence: {stats['average_confidence']:.2f}")
        print(f"Revision Rate: {stats['revision_rate']:.1%}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Chain-of-Verification provides:
    ✓ Self-verification through questioning
    ✓ Inconsistency detection
    ✓ Fact checking automation
    ✓ Hallucination reduction
    ✓ Improved factual accuracy
    ✓ Transparent verification process
    
    This pattern excels at:
    - Fact-checking responses
    - Reducing hallucinations
    - Knowledge-intensive tasks
    - Research and analysis
    - Educational content
    - High-stakes information
    
    Verification process:
    1. Generate initial response
    2. Create verification questions
    3. Answer questions independently
    4. Check consistency
    5. Identify contradictions
    6. Revise response if needed
    
    Verification components:
    - Initial Generator: Creates response
    - Question Generator: Creates verification questions
    - Answer Generator: Answers questions independently
    - Consistency Checker: Finds contradictions
    - Response Reviser: Fixes inconsistencies
    
    Question types:
    - Factual verification: Check specific facts
    - Logical consistency: Check reasoning
    - Completeness: Check coverage
    - Accuracy: Check precision
    
    Benefits:
    - Accuracy: Catch factual errors
    - Reliability: Reduce hallucinations
    - Transparency: Show verification process
    - Self-correction: Auto-fix errors
    - Confidence: Measure certainty
    - Scalable: Works for any domain
    
    Verification strategies:
    - Independent Q&A: Answer questions separately
    - Consistency checking: Compare answers
    - Contradiction detection: Find conflicts
    - Confidence scoring: Quantify certainty
    - Iterative revision: Fix issues
    
    Use Chain-of-Verification when:
    - Factual accuracy is critical
    - Hallucinations are a concern
    - Need transparent fact-checking
    - Working with knowledge-intensive tasks
    - Building educational systems
    - High-stakes decision making
    
    Comparison with other patterns:
    - vs Self-Evaluation: CoVe focuses on factual verification
    - vs Constitutional AI: CoVe checks facts, not values
    - vs Feedback Loops: CoVe is immediate, not learning-based
    """)
    
    
    if __name__ == "__main__":
        demonstrate_chain_of_verification()
    
        

    Agentic Design Pattern: Chain Of Verification

    Pattern 041: Chain Of Verification – Line-by-Line Explanation

    Overview

    This document provides a detailed line-by-line explanation of the Chain Of Verification pattern implementation.


    Pattern Description

    Pattern 041: Chain-of-Verification (CoVe)

    Description: Chain-of-Verification enables agents to reduce hallucinations and improve factual accuracy by generating verification questions to check their own answers. The agent generates an initial response, creates verification questions, answers them, and then revises the original response based on the verification results.

    Components: – Initial Generator: Creates initial response – Verification Question Generator: Creates questions to verify response – Verification Answerer: Answers verification questions – Consistency Checker: Checks for contradictions – Response Reviser: Updates response based on verification

    Use Cases: – Fact-checking automation – Reducing hallucinations – Improving factual accuracy – Knowledge-intensive tasks – Research and analysis – Educational content

    LangChain Implementation: Uses multi-step verification chains to check responses, identify inconsistencies, and revise outputs for improved accuracy.


    File Statistics

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

    Module Documentation

    """
    Pattern 041: Chain-of-Verification (CoVe)
    
    Description:
        Chain-of-Verification enables agents to reduce hallucinations and improve
        factual accuracy by generating verification questions to check their own
        answers. The agent generates an initial response, creates verification
        questions, answers them, and then revises the original response based on
        the verification results.
    
    Components:
        - Initial Generator: Creates initial response
        - Verification Question Generator: Creates questions to verify response
        - Verification Answerer: Answers verification questions
        - Consistency Checker: Checks for contradictions
        - Response Reviser: Updates response based on verification
    
    Use Cases:
        - Fact-checking automation
        - Reducing hallucinations
        - Improving factual accuracy
        - Knowledge-intensive tasks
        - Research and analysis
        - Educational content
    
    LangChain Implementation:
        Uses multi-step verification chains to check responses, identify
        inconsistencies, and revise outputs for improved accuracy.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    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 – datetime.datetime: Date and time handling – 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: VerificationQuestion

    Lines 45-51

    Purpose:

    A question to verify part of the response.
    

    Class: VerificationResult

    Lines 55-63

    Purpose:

    Result of verification process.
    

    Class: ChainOfVerificationAgent

    Lines 66-372

    Purpose:

    Agent that verifies its own responses through questioning.
    
    Features:
    - Generates verification questions
    - Checks factual consistency
    - Identifies contradictions
    - Revises responses
    - Improves accuracy
    

    Methods:

    __init__(self, response_temperature, verification_temperature, num_verification_questions)

    Lines 78-161

    Purpose: Initialize the class instance with required parameters.

    Parameters: – response_temperature: Sampling temperature (0-1) – verification_temperature: Sampling temperature (0-1) – num_verification_questions: Question to answer

    generate_response(self, query)

    Lines 163-166

    Purpose:

    Generate initial response to query.
    

    Parameters: – query: Input query or question

    Returns: str

    generate_verification_questions(self, query, response)

    Lines 168-199

    Purpose:

    Generate questions to verify the response.
    
    Returns:
        List of VerificationQuestion objects
    

    Parameters: – query: Input query or question – response: Response

    Returns: <ast.Subscript object at 0x000001A06D7537D0>

    answer_verification_questions(self, questions)

    Lines 201-217

    Purpose:

    Answer verification questions independently.
    
    Returns:
        List of VerificationQuestion objects with answers
    

    Parameters: – questions: Question to answer

    Returns: <ast.Subscript object at 0x000001A06D702E10>

    check_consistency(self, original_response, questions)

    Lines 219-269

    Purpose:

    Check consistency between original response and verification answers.
    
    Returns:
        Tuple of (inconsistencies, verified_facts, confidence)
    

    Parameters: – original_response: Original response – questions: Question to answer

    Returns: <ast.Subscript object at 0x000001A06D638A10>

    revise_response(self, query, original_response, inconsistencies)

    Lines 271-295

    Purpose:

    Revise response to fix inconsistencies.
    
    Returns:
        Revised response
    

    Parameters: – query: Input query or question – original_response: Original response – inconsistencies: Inconsistencies

    Returns: str

    verify_response(self, query, response)

    Lines 297-354

    Purpose:

    Complete Chain-of-Verification process.
    
    Steps:
    1. Generate initial response (or use provided)
    2. Generate verification questions
    3. Answer verification questions
    4. Check consistency
    5. Revise if needed
    
    Args:
        query: The query to answer
        response: Optional pre-generated response
    
    Returns:
        VerificationResult object
    

    Parameters: – query: Input query or question – response: Response

    Returns: VerificationResult

    get_verification_statistics(self)

    Lines 356-372

    Purpose:

    Get statistics about verifications.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D623D10>


    Functions

    demonstrate_chain_of_verification()

    Lines 375-552

    Purpose:

    Demonstrates Chain-of-Verification for reducing hallucinations.
    

    Key Concepts Summary

    Pattern Overview

    The Chain Of Verification 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 040
    • Pattern 042

    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 Chain Of Verification 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: 041_chain_of_verification.py
    Total Lines: 557
    Pattern Category: AI Agent Design Pattern #041


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Self Evaluation

    
        """
    Pattern 040: Self-Evaluation
    
    Description:
        Self-Evaluation enables agents to assess the quality of their own outputs
        through confidence scoring, consistency checking, and self-critique. The agent
        acts as its own judge, identifying potential errors, measuring confidence,
        and flagging outputs that may require human review or regeneration.
    
    Components:
        - Confidence Estimator: Assigns confidence scores to outputs
        - Consistency Checker: Verifies logical consistency
        - Quality Metrics: Measures output quality
        - Self-Critique: Identifies weaknesses and errors
        - Threshold Monitor: Flags low-quality outputs
    
    Use Cases:
        - Quality control automation
        - Error detection before delivery
        - Confidence-based routing
        - Self-improvement loops
        - Autonomous quality assurance
        - Production reliability
    
    LangChain Implementation:
        Uses LLM-based evaluation chains and scoring mechanisms to enable
        agents to assess their own output quality and confidence.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Tuple
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime
    from statistics import mean, stdev
    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 QualityLevel(Enum):
        """Quality levels for outputs."""
        EXCELLENT = "excellent"
        GOOD = "good"
        ACCEPTABLE = "acceptable"
        POOR = "poor"
        UNACCEPTABLE = "unacceptable"
    
    
    class EvaluationDimension(Enum):
        """Dimensions for evaluation."""
        ACCURACY = "accuracy"
        COMPLETENESS = "completeness"
        CLARITY = "clarity"
        RELEVANCE = "relevance"
        COHERENCE = "coherence"
        SAFETY = "safety"
    
    
    @dataclass
    class EvaluationScore:
        """Score for a single evaluation dimension."""
        dimension: EvaluationDimension
        score: float  # 0.0 to 1.0
        reasoning: str = ""
        confidence: float = 1.0
    
    
    @dataclass
    class SelfEvaluation:
        """Complete self-evaluation of an output."""
        output: str
        overall_confidence: float
        quality_level: QualityLevel
        dimension_scores: List[EvaluationScore]
        identified_issues: List[str]
        suggestions: List[str]
        should_regenerate: bool
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class ConsistencyCheck:
        """Result of consistency checking."""
        output: str
        is_consistent: bool
        contradictions: List[str]
        logical_errors: List[str]
        consistency_score: float  # 0.0 to 1.0
    
    
    class ConfidenceEstimator:
        """
        Estimates confidence in generated outputs.
        """
        
        def __init__(self, llm: ChatOpenAI):
            self.llm = llm
            
            # Confidence estimation prompt
            self.confidence_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a confidence estimator. Evaluate how confident you are in this response.
    
    Consider:
    1. Factual accuracy and verifiability
    2. Completeness of the answer
    3. Clarity and coherence
    4. Presence of uncertainties or assumptions
    5. Potential for errors or misconceptions
    
    Respond in this format:
    CONFIDENCE: [0.0-1.0]
    REASONING: [explanation]
    UNCERTAINTIES: [list any uncertain aspects]"""),
                ("user", "Query: {query}\n\nResponse: {response}\n\nEvaluate confidence in this response.")
            ])
        
        def estimate_confidence(self, query: str, response: str) -> Tuple[float, str, List[str]]:
            """
            Estimate confidence in a response.
            
            Returns:
                Tuple of (confidence_score, reasoning, uncertainties)
            """
            chain = self.confidence_prompt | self.llm | StrOutputParser()
            evaluation = chain.invoke({"query": query, "response": response})
            
            # Parse evaluation
            confidence = 0.5
            reasoning = ""
            uncertainties = []
            
            lines = evaluation.split('\n')
            for i, line in enumerate(lines):
                if line.startswith("CONFIDENCE:"):
                    try:
                        confidence = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                elif line.startswith("REASONING:"):
                    reasoning = line.split(':', 1)[1].strip()
                elif line.startswith("UNCERTAINTIES:"):
                    # Collect remaining lines as uncertainties
                    uncertainties = [l.strip() for l in lines[i+1:] if l.strip()]
                    break
            
            return confidence, reasoning, uncertainties
    
    
    class ConsistencyChecker:
        """
        Checks logical consistency of outputs.
        """
        
        def __init__(self, llm: ChatOpenAI):
            self.llm = llm
            
            # Consistency checking prompt
            self.consistency_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a consistency checker. Analyze this text for logical consistency.
    
    Check for:
    1. Internal contradictions
    2. Logical fallacies
    3. Inconsistent statements
    4. Conflicting information
    
    Respond in this format:
    CONSISTENT: [YES/NO]
    CONTRADICTIONS: [list any contradictions found]
    LOGICAL_ERRORS: [list any logical errors]
    SCORE: [0.0-1.0]"""),
                ("user", "Text to check:\n{text}\n\nAnalyze for consistency.")
            ])
        
        def check_consistency(self, text: str) -> ConsistencyCheck:
            """
            Check consistency of text.
            
            Returns:
                ConsistencyCheck object
            """
            chain = self.consistency_prompt | self.llm | StrOutputParser()
            result = chain.invoke({"text": text})
            
            # Parse result
            is_consistent = True
            contradictions = []
            logical_errors = []
            consistency_score = 1.0
            
            lines = result.split('\n')
            current_section = None
            
            for line in lines:
                if line.startswith("CONSISTENT:"):
                    is_consistent = "YES" in line.upper()
                elif line.startswith("CONTRADICTIONS:"):
                    current_section = "contradictions"
                elif line.startswith("LOGICAL_ERRORS:"):
                    current_section = "logical_errors"
                elif line.startswith("SCORE:"):
                    try:
                        consistency_score = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                    current_section = None
                elif line.strip() and current_section:
                    if current_section == "contradictions":
                        contradictions.append(line.strip())
                    elif current_section == "logical_errors":
                        logical_errors.append(line.strip())
            
            return ConsistencyCheck(
                output=text,
                is_consistent=is_consistent,
                contradictions=contradictions,
                logical_errors=logical_errors,
                consistency_score=consistency_score
            )
    
    
    class QualityEvaluator:
        """
        Evaluates output quality across multiple dimensions.
        """
        
        def __init__(self, llm: ChatOpenAI):
            self.llm = llm
            
            # Quality evaluation prompt
            self.quality_prompt = ChatPromptTemplate.from_messages([
                ("system", """You are a quality evaluator. Evaluate this response on the given dimension.
    
    Dimension: {dimension}
    
    Rate from 0.0 (worst) to 1.0 (best) and explain your reasoning.
    
    Respond in this format:
    SCORE: [0.0-1.0]
    REASONING: [explanation]"""),
                ("user", "Query: {query}\n\nResponse: {response}\n\nEvaluate on: {dimension}")
            ])
        
        def evaluate_dimension(
            self,
            query: str,
            response: str,
            dimension: EvaluationDimension
        ) -> EvaluationScore:
            """
            Evaluate response on a single dimension.
            
            Returns:
                EvaluationScore object
            """
            chain = self.quality_prompt | self.llm | StrOutputParser()
            evaluation = chain.invoke({
                "query": query,
                "response": response,
                "dimension": dimension.value
            })
            
            # Parse evaluation
            score = 0.5
            reasoning = ""
            
            lines = evaluation.split('\n')
            for line in lines:
                if line.startswith("SCORE:"):
                    try:
                        score = float(line.split(':')[1].strip())
                    except (ValueError, IndexError):
                        pass
                elif line.startswith("REASONING:"):
                    reasoning = line.split(':', 1)[1].strip()
            
            return EvaluationScore(
                dimension=dimension,
                score=score,
                reasoning=reasoning,
                confidence=1.0
            )
    
    
    class SelfEvaluationAgent:
        """
        Agent that evaluates its own outputs for quality assurance.
        
        Features:
        - Confidence estimation
        - Consistency checking
        - Multi-dimensional quality evaluation
        - Issue identification
        - Regeneration recommendations
        """
        
        def __init__(
            self,
            response_temperature: float = 0.7,
            evaluation_temperature: float = 0.1,
            confidence_threshold: float = 0.7,
            quality_threshold: float = 0.6
        ):
            self.response_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=response_temperature)
            self.eval_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=evaluation_temperature)
            
            self.confidence_estimator = ConfidenceEstimator(self.eval_llm)
            self.consistency_checker = ConsistencyChecker(self.eval_llm)
            self.quality_evaluator = QualityEvaluator(self.eval_llm)
            
            self.confidence_threshold = confidence_threshold
            self.quality_threshold = quality_threshold
            
            # Response generation prompt
            self.response_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Provide clear, accurate answers."),
                ("user", "{query}")
            ])
            
            # Evaluation history
            self.evaluations: List[SelfEvaluation] = []
        
        def generate_response(self, query: str) -> str:
            """Generate response to query."""
            chain = self.response_prompt | self.response_llm | StrOutputParser()
            return chain.invoke({"query": query})
        
        def evaluate_response(
            self,
            query: str,
            response: str,
            dimensions: Optional[List[EvaluationDimension]] = None
        ) -> SelfEvaluation:
            """
            Perform comprehensive self-evaluation of response.
            
            Args:
                query: The original query
                response: The generated response
                dimensions: Dimensions to evaluate (default: all)
                
            Returns:
                SelfEvaluation object
            """
            if dimensions is None:
                dimensions = [
                    EvaluationDimension.ACCURACY,
                    EvaluationDimension.COMPLETENESS,
                    EvaluationDimension.CLARITY,
                    EvaluationDimension.RELEVANCE
                ]
            
            # 1. Estimate confidence
            confidence, conf_reasoning, uncertainties = self.confidence_estimator.estimate_confidence(query, response)
            
            # 2. Check consistency
            consistency_check = self.consistency_checker.check_consistency(response)
            
            # 3. Evaluate on each dimension
            dimension_scores = []
            for dimension in dimensions:
                score = self.quality_evaluator.evaluate_dimension(query, response, dimension)
                dimension_scores.append(score)
            
            # 4. Calculate overall quality
            avg_score = mean([s.score for s in dimension_scores])
            
            # Adjust based on consistency
            if not consistency_check.is_consistent:
                avg_score *= 0.8  # Penalize inconsistency
            
            # Determine quality level
            if avg_score >= 0.8:
                quality_level = QualityLevel.EXCELLENT
            elif avg_score >= 0.7:
                quality_level = QualityLevel.GOOD
            elif avg_score >= 0.6:
                quality_level = QualityLevel.ACCEPTABLE
            elif avg_score >= 0.4:
                quality_level = QualityLevel.POOR
            else:
                quality_level = QualityLevel.UNACCEPTABLE
            
            # 5. Identify issues
            identified_issues = []
            
            if confidence < self.confidence_threshold:
                identified_issues.append(f"Low confidence: {confidence:.2f}")
            
            if not consistency_check.is_consistent:
                identified_issues.extend([f"Contradiction: {c}" for c in consistency_check.contradictions])
                identified_issues.extend([f"Logical error: {e}" for e in consistency_check.logical_errors])
            
            for score in dimension_scores:
                if score.score < self.quality_threshold:
                    identified_issues.append(f"Low {score.dimension.value}: {score.score:.2f}")
            
            identified_issues.extend([f"Uncertainty: {u}" for u in uncertainties])
            
            # 6. Generate suggestions
            suggestions = []
            
            if confidence < self.confidence_threshold:
                suggestions.append("Consider adding caveats or acknowledging uncertainties")
            
            if not consistency_check.is_consistent:
                suggestions.append("Revise to resolve logical contradictions")
            
            for score in dimension_scores:
                if score.score < self.quality_threshold:
                    suggestions.append(f"Improve {score.dimension.value}: {score.reasoning}")
            
            # 7. Decide if should regenerate
            should_regenerate = (
                confidence < self.confidence_threshold or
                not consistency_check.is_consistent or
                avg_score  Dict[str, Any]:
            """
            Generate response with self-evaluation and automatic regeneration.
            
            Args:
                query: The query to process
                max_attempts: Maximum regeneration attempts
                
            Returns:
                Dictionary with response and evaluation
            """
            attempt = 0
            best_response = None
            best_evaluation = None
            best_score = -1
            
            while attempt  best_score:
                    best_score = avg_score
                    best_response = response
                    best_evaluation = evaluation
                
                # Stop if quality is acceptable
                if not evaluation.should_regenerate:
                    break
            
            return {
                "query": query,
                "response": best_response,
                "evaluation": best_evaluation,
                "attempts": attempt,
                "accepted": not best_evaluation.should_regenerate
            }
        
        def get_evaluation_statistics(self) -> Dict[str, Any]:
            """Get statistics about evaluations."""
            if not self.evaluations:
                return {"total_evaluations": 0}
            
            avg_confidence = mean([e.overall_confidence for e in self.evaluations])
            
            quality_distribution = {}
            for e in self.evaluations:
                quality_distribution[e.quality_level.value] = quality_distribution.get(e.quality_level.value, 0) + 1
            
            regeneration_rate = sum(1 for e in self.evaluations if e.should_regenerate) / len(self.evaluations)
            
            return {
                "total_evaluations": len(self.evaluations),
                "average_confidence": avg_confidence,
                "quality_distribution": quality_distribution,
                "regeneration_rate": regeneration_rate
            }
    
    
    def demonstrate_self_evaluation():
        """
        Demonstrates self-evaluation with quality assessment and confidence scoring.
        """
        print("=" * 80)
        print("SELF-EVALUATION DEMONSTRATION")
        print("=" * 80)
        
        # Create self-evaluation agent
        agent = SelfEvaluationAgent(
            confidence_threshold=0.7,
            quality_threshold=0.6
        )
        
        # Test 1: Simple factual query (should have high confidence)
        print("\n" + "=" * 80)
        print("Test 1: Factual Query (Expected: High Confidence)")
        print("=" * 80)
        
        query1 = "What is the capital of France?"
        print(f"\nQuery: {query1}")
        
        result1 = agent.process_with_evaluation(query1, max_attempts=1)
        
        print(f"\nResponse: {result1['response']}")
        print(f"\nAttempts: {result1['attempts']}")
        print(f"Accepted: {result1['accepted']}")
        
        eval1 = result1['evaluation']
        print(f"\nOverall Confidence: {eval1.overall_confidence:.2f}")
        print(f"Quality Level: {eval1.quality_level.value}")
        print(f"Should Regenerate: {eval1.should_regenerate}")
        
        print("\nDimension Scores:")
        for score in eval1.dimension_scores:
            print(f"  - {score.dimension.value}: {score.score:.2f}")
        
        if eval1.identified_issues:
            print("\nIdentified Issues:")
            for issue in eval1.identified_issues[:3]:
                print(f"  - {issue}")
        
        # Test 2: Complex query (may have lower confidence)
        print("\n" + "=" * 80)
        print("Test 2: Complex Speculative Query")
        print("=" * 80)
        
        query2 = "What will be the impact of quantum computing on cryptography in 2030?"
        print(f"\nQuery: {query2}")
        
        result2 = agent.process_with_evaluation(query2, max_attempts=2)
        
        print(f"\nResponse: {result2['response'][:200]}...")
        print(f"\nAttempts: {result2['attempts']}")
        print(f"Accepted: {result2['accepted']}")
        
        eval2 = result2['evaluation']
        print(f"\nOverall Confidence: {eval2.overall_confidence:.2f}")
        print(f"Quality Level: {eval2.quality_level.value}")
        print(f"Should Regenerate: {eval2.should_regenerate}")
        
        print("\nDimension Scores:")
        for score in eval2.dimension_scores:
            print(f"  - {score.dimension.value}: {score.score:.2f} - {score.reasoning[:50]}...")
        
        if eval2.identified_issues:
            print("\nIdentified Issues:")
            for issue in eval2.identified_issues[:3]:
                print(f"  - {issue}")
        
        if eval2.suggestions:
            print("\nSuggestions:")
            for suggestion in eval2.suggestions[:3]:
                print(f"  - {suggestion}")
        
        # Test 3: Consistency checking
        print("\n" + "=" * 80)
        print("Test 3: Consistency Checking")
        print("=" * 80)
        
        query3 = "Explain machine learning"
        print(f"\nQuery: {query3}")
        
        response3 = agent.generate_response(query3)
        print(f"\nResponse: {response3[:200]}...")
        
        consistency = agent.consistency_checker.check_consistency(response3)
        print(f"\nConsistent: {consistency.is_consistent}")
        print(f"Consistency Score: {consistency.consistency_score:.2f}")
        
        if consistency.contradictions:
            print("\nContradictions Found:")
            for contradiction in consistency.contradictions:
                print(f"  - {contradiction}")
        
        # Test 4: Multi-dimensional evaluation
        print("\n" + "=" * 80)
        print("Test 4: Comprehensive Multi-Dimensional Evaluation")
        print("=" * 80)
        
        query4 = "How does photosynthesis work?"
        print(f"\nQuery: {query4}")
        
        response4 = agent.generate_response(query4)
        print(f"\nResponse: {response4[:200]}...")
        
        eval4 = agent.evaluate_response(
            query4,
            response4,
            dimensions=[
                EvaluationDimension.ACCURACY,
                EvaluationDimension.COMPLETENESS,
                EvaluationDimension.CLARITY,
                EvaluationDimension.RELEVANCE,
                EvaluationDimension.COHERENCE
            ]
        )
        
        print(f"\nComprehensive Evaluation:")
        print(f"Overall Confidence: {eval4.overall_confidence:.2f}")
        print(f"Quality Level: {eval4.quality_level.value}")
        
        print("\nAll Dimension Scores:")
        for score in eval4.dimension_scores:
            print(f"  - {score.dimension.value}: {score.score:.2f}")
        
        # Show statistics
        print("\n" + "=" * 80)
        print("Evaluation Statistics")
        print("=" * 80)
        
        stats = agent.get_evaluation_statistics()
        print(f"\nTotal Evaluations: {stats['total_evaluations']}")
        print(f"Average Confidence: {stats['average_confidence']:.2f}")
        print(f"Regeneration Rate: {stats['regeneration_rate']:.1%}")
        
        print("\nQuality Distribution:")
        for quality, count in stats['quality_distribution'].items():
            print(f"  - {quality}: {count}")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Self-Evaluation provides:
    ✓ Confidence estimation for outputs
    ✓ Consistency checking for logical errors
    ✓ Multi-dimensional quality assessment
    ✓ Automatic issue identification
    ✓ Regeneration recommendations
    ✓ Autonomous quality assurance
    
    This pattern excels at:
    - Quality control automation
    - Error detection before delivery
    - Confidence-based routing
    - Self-improvement loops
    - Production reliability
    - Autonomous QA
    
    Evaluation components:
    1. Confidence Estimator: Assigns confidence scores
    2. Consistency Checker: Verifies logical consistency
    3. Quality Evaluator: Multi-dimensional assessment
    4. Issue Identifier: Flags problems
    5. Suggestion Generator: Improvement recommendations
    
    Evaluation dimensions:
    - ACCURACY: Factual correctness
    - COMPLETENESS: Thorough coverage
    - CLARITY: Clear communication
    - RELEVANCE: On-topic responses
    - COHERENCE: Logical flow
    - SAFETY: Safe and appropriate
    
    Quality levels:
    - EXCELLENT: 0.8-1.0
    - GOOD: 0.7-0.8
    - ACCEPTABLE: 0.6-0.7
    - POOR: 0.4-0.6
    - UNACCEPTABLE: 0.0-0.4
    
    Evaluation process:
    1. Generate response
    2. Estimate confidence
    3. Check consistency
    4. Evaluate on dimensions
    5. Identify issues
    6. Generate suggestions
    7. Decide if regeneration needed
    
    Benefits:
    - Autonomous QA: No human needed
    - Early detection: Catch errors before delivery
    - Transparency: Clear reasoning for scores
    - Adaptability: Configurable thresholds
    - Continuous: Always on guard
    - Scalable: Handles high volume
    
    Use Self-Evaluation when you need:
    - Production quality control
    - Confidence-based routing
    - Error detection automation
    - Self-improving systems
    - High-reliability requirements
    - Transparent quality metrics
    """)
    
    
    if __name__ == "__main__":
        demonstrate_self_evaluation()
    
        

    Agentic Design Pattern: Self Evaluation

    Pattern 040: Self Evaluation – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 040: Self-Evaluation

    Description: Self-Evaluation enables agents to assess the quality of their own outputs through confidence scoring, consistency checking, and self-critique. The agent acts as its own judge, identifying potential errors, measuring confidence, and flagging outputs that may require human review or regeneration.

    Components: – Confidence Estimator: Assigns confidence scores to outputs – Consistency Checker: Verifies logical consistency – Quality Metrics: Measures output quality – Self-Critique: Identifies weaknesses and errors – Threshold Monitor: Flags low-quality outputs

    Use Cases: – Quality control automation – Error detection before delivery – Confidence-based routing – Self-improvement loops – Autonomous quality assurance – Production reliability

    LangChain Implementation: Uses LLM-based evaluation chains and scoring mechanisms to enable agents to assess their own output quality and confidence.


    File Statistics

    • Total Lines: 720
    • Classes: 9
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 040: Self-Evaluation
    
    Description:
        Self-Evaluation enables agents to assess the quality of their own outputs
        through confidence scoring, consistency checking, and self-critique. The agent
        acts as its own judge, identifying potential errors, measuring confidence,
        and flagging outputs that may require human review or regeneration.
    
    Components:
        - Confidence Estimator: Assigns confidence scores to outputs
        - Consistency Checker: Verifies logical consistency
        - Quality Metrics: Measures output quality
        - Self-Critique: Identifies weaknesses and errors
        - Threshold Monitor: Flags low-quality outputs
    
    Use Cases:
        - Quality control automation
        - Error detection before delivery
        - Confidence-based routing
        - Self-improvement loops
        - Autonomous quality assurance
        - Production reliability
    
    LangChain Implementation:
        Uses LLM-based evaluation chains and scoring mechanisms to enable
        agents to assess their own output quality and confidence.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    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 – datetime.datetime: Date and time handling – 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 statistics.mean
    from statistics.stdev
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: QualityLevel

    Lines 45-51

    Inherits from: Enum

    Purpose:

    Quality levels for outputs.
    

    Class: EvaluationDimension

    Lines 54-61

    Inherits from: Enum

    Purpose:

    Dimensions for evaluation.
    

    Class: EvaluationScore

    Lines 65-70

    Purpose:

    Score for a single evaluation dimension.
    

    Class: SelfEvaluation

    Lines 74-83

    Purpose:

    Complete self-evaluation of an output.
    

    Class: ConsistencyCheck

    Lines 87-93

    Purpose:

    Result of consistency checking.
    

    Class: ConfidenceEstimator

    Lines 96-151

    Purpose:

    Estimates confidence in generated outputs.
    

    Methods:

    __init__(self, llm)

    Lines 101-120

    Purpose: Initialize the class instance with required parameters.

    Parameters: – llm: Language model

    estimate_confidence(self, query, response)

    Lines 122-151

    Purpose:

    Estimate confidence in a response.
    
    Returns:
        Tuple of (confidence_score, reasoning, uncertainties)
    

    Parameters: – query: Input query or question – response: Response

    Returns: <ast.Subscript object at 0x000001A06D6F6190>


    Class: ConsistencyChecker

    Lines 154-224

    Purpose:

    Checks logical consistency of outputs.
    

    Methods:

    __init__(self, llm)

    Lines 159-178

    Purpose: Initialize the class instance with required parameters.

    Parameters: – llm: Language model

    check_consistency(self, text)

    Lines 180-224

    Purpose:

    Check consistency of text.
    
    Returns:
        ConsistencyCheck object
    

    Parameters: – text: Text

    Returns: ConsistencyCheck


    Class: QualityEvaluator

    Lines 227-287

    Purpose:

    Evaluates output quality across multiple dimensions.
    

    Methods:

    __init__(self, llm)

    Lines 232-247

    Purpose: Initialize the class instance with required parameters.

    Parameters: – llm: Language model

    evaluate_dimension(self, query, response, dimension)

    Lines 249-287

    Purpose:

    Evaluate response on a single dimension.
    
    Returns:
        EvaluationScore object
    

    Parameters: – query: Input query or question – response: Response – dimension: Dimension

    Returns: EvaluationScore


    Class: SelfEvaluationAgent

    Lines 290-505

    Purpose:

    Agent that evaluates its own outputs for quality assurance.
    
    Features:
    - Confidence estimation
    - Consistency checking
    - Multi-dimensional quality evaluation
    - Issue identification
    - Regeneration recommendations
    

    Methods:

    __init__(self, response_temperature, evaluation_temperature, confidence_threshold, quality_threshold)

    Lines 302-326

    Purpose: Initialize the class instance with required parameters.

    Parameters: – response_temperature: Sampling temperature (0-1) – evaluation_temperature: Sampling temperature (0-1) – confidence_threshold: Confidence threshold – quality_threshold: Quality threshold

    generate_response(self, query)

    Lines 328-331

    Purpose:

    Generate response to query.
    

    Parameters: – query: Input query or question

    Returns: str

    evaluate_response(self, query, response, dimensions)

    Lines 333-437

    Purpose:

    Perform comprehensive self-evaluation of response.
    
    Args:
        query: The original query
        response: The generated response
        dimensions: Dimensions to evaluate (default: all)
    
    Returns:
        SelfEvaluation object
    

    Parameters: – query: Input query or question – response: Response – dimensions: Dimensions

    Returns: SelfEvaluation

    process_with_evaluation(self, query, max_attempts)

    Lines 439-485

    Purpose:

    Generate response with self-evaluation and automatic regeneration.
    
    Args:
        query: The query to process
        max_attempts: Maximum regeneration attempts
    
    Returns:
        Dictionary with response and evaluation
    

    Parameters: – query: Input query or question – max_attempts: Max attempts

    Returns: <ast.Subscript object at 0x000001A06D60A9D0>

    get_evaluation_statistics(self)

    Lines 487-505

    Purpose:

    Get statistics about evaluations.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6046D0>


    Functions

    demonstrate_self_evaluation()

    Lines 508-715

    Purpose:

    Demonstrates self-evaluation with quality assessment and confidence scoring.
    

    Key Concepts Summary

    Pattern Overview

    The Self Evaluation 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 039
    • Pattern 041

    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 Evaluation 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: 040_self_evaluation.py
    Total Lines: 720
    Pattern Category: AI Agent Design Pattern #040


    Generated from actual implementation code
    For educational and reference purposes

  • Agentic Design Pattern: Feedback Loops

    
        """
    Pattern 039: Feedback Loops
    
    Description:
        Feedback Loops enable agents to learn from the outcomes of their actions through
        immediate feedback (real-time corrections) and delayed feedback (outcome-based
        rewards). The agent uses feedback to improve performance, adjust strategies,
        and adapt behavior over time.
    
    Components:
        - Feedback Collector: Gathers outcome data
        - Performance Tracker: Monitors metrics over time
        - Learning Mechanism: Incorporates feedback into behavior
        - Reward Signal: Quantifies outcome quality
        - Adaptation Strategy: How to adjust based on feedback
    
    Use Cases:
        - Continuous improvement systems
        - Reinforcement learning agents
        - User preference learning
        - Model fine-tuning
        - A/B testing and optimization
        - Adaptive recommendation systems
    
    LangChain Implementation:
        Uses feedback collection, performance tracking, and iterative improvement
        mechanisms to enable agents to learn from action outcomes.
    """
    
    import os
    from typing import List, Dict, Any, Optional, Callable
    from dataclasses import dataclass, field
    from enum import Enum
    from datetime import datetime, timedelta
    from statistics import mean, stdev
    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 FeedbackType(Enum):
        """Types of feedback."""
        IMMEDIATE = "immediate"  # Real-time corrections
        DELAYED = "delayed"  # Outcome-based rewards
        EXPLICIT = "explicit"  # Direct user feedback
        IMPLICIT = "implicit"  # Inferred from behavior
        COMPARATIVE = "comparative"  # A/B testing results
    
    
    class FeedbackSignal(Enum):
        """Feedback signals."""
        POSITIVE = "positive"
        NEGATIVE = "negative"
        NEUTRAL = "neutral"
        MIXED = "mixed"
    
    
    @dataclass
    class Feedback:
        """Feedback on an action or response."""
        action_id: str
        signal: FeedbackSignal
        score: float  # -1.0 to 1.0
        feedback_type: FeedbackType
        details: str = ""
        timestamp: datetime = field(default_factory=datetime.now)
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class Action:
        """An action taken by the agent."""
        id: str
        query: str
        response: str
        strategy: str
        timestamp: datetime = field(default_factory=datetime.now)
        metadata: Dict[str, Any] = field(default_factory=dict)
    
    
    @dataclass
    class PerformanceMetrics:
        """Performance metrics over time."""
        total_actions: int
        average_score: float
        positive_feedback_rate: float
        negative_feedback_rate: float
        improvement_rate: float
        recent_scores: List[float]
        best_strategy: str
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class FeedbackCollector:
        """
        Collects and stores feedback on agent actions.
        """
        
        def __init__(self):
            self.actions: Dict[str, Action] = {}
            self.feedback: Dict[str, List[Feedback]] = {}  # action_id -> feedbacks
        
        def record_action(self, action: Action):
            """Record an action taken by the agent."""
            self.actions[action.id] = action
            self.feedback[action.id] = []
        
        def add_feedback(self, feedback: Feedback):
            """Add feedback for an action."""
            if feedback.action_id not in self.feedback:
                self.feedback[feedback.action_id] = []
            
            self.feedback[feedback.action_id].append(feedback)
        
        def get_action_feedback(self, action_id: str) -> List[Feedback]:
            """Get all feedback for an action."""
            return self.feedback.get(action_id, [])
        
        def get_average_score(self, action_ids: Optional[List[str]] = None) -> float:
            """Get average feedback score."""
            if action_ids is None:
                action_ids = list(self.actions.keys())
            
            scores = []
            for action_id in action_ids:
                feedbacks = self.get_action_feedback(action_id)
                if feedbacks:
                    avg_score = mean([f.score for f in feedbacks])
                    scores.append(avg_score)
            
            return mean(scores) if scores else 0.0
    
    
    class PerformanceTracker:
        """
        Tracks performance metrics over time.
        """
        
        def __init__(self, window_size: int = 10):
            self.window_size = window_size
            self.score_history: List[tuple[datetime, float]] = []
            self.strategy_scores: Dict[str, List[float]] = {}
        
        def record_score(self, score: float, strategy: str = "default"):
            """Record a performance score."""
            self.score_history.append((datetime.now(), score))
            
            if strategy not in self.strategy_scores:
                self.strategy_scores[strategy] = []
            self.strategy_scores[strategy].append(score)
        
        def get_recent_performance(self) -> PerformanceMetrics:
            """Get recent performance metrics."""
            if not self.score_history:
                return PerformanceMetrics(
                    total_actions=0,
                    average_score=0.0,
                    positive_feedback_rate=0.0,
                    negative_feedback_rate=0.0,
                    improvement_rate=0.0,
                    recent_scores=[],
                    best_strategy="none"
                )
            
            # Get recent scores
            recent_scores = [score for _, score in self.score_history[-self.window_size:]]
            
            # Calculate metrics
            total_actions = len(self.score_history)
            average_score = mean(recent_scores)
            positive_rate = sum(1 for s in recent_scores if s > 0) / len(recent_scores)
            negative_rate = sum(1 for s in recent_scores if s = 2:
                first_half = recent_scores[:len(recent_scores)//2]
                second_half = recent_scores[len(recent_scores)//2:]
                improvement_rate = mean(second_half) - mean(first_half)
            else:
                improvement_rate = 0.0
            
            # Find best strategy
            best_strategy = "default"
            best_avg = -float('inf')
            for strategy, scores in self.strategy_scores.items():
                if scores:
                    avg = mean(scores)
                    if avg > best_avg:
                        best_avg = avg
                        best_strategy = strategy
            
            return PerformanceMetrics(
                total_actions=total_actions,
                average_score=average_score,
                positive_feedback_rate=positive_rate,
                negative_feedback_rate=negative_rate,
                improvement_rate=improvement_rate,
                recent_scores=recent_scores,
                best_strategy=best_strategy
            )
        
        def get_strategy_performance(self, strategy: str) -> Optional[float]:
            """Get average performance for a strategy."""
            if strategy in self.strategy_scores and self.strategy_scores[strategy]:
                return mean(self.strategy_scores[strategy])
            return None
    
    
    class FeedbackLoopAgent:
        """
        Agent that learns from feedback to improve performance.
        
        Features:
        - Immediate and delayed feedback
        - Performance tracking
        - Strategy adaptation
        - Continuous improvement
        - Multi-strategy experimentation
        """
        
        def __init__(
            self,
            strategies: Optional[Dict[str, ChatPromptTemplate]] = None,
            temperature: float = 0.7,
            exploration_rate: float = 0.2  # Rate of trying new strategies
        ):
            self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
            self.feedback_collector = FeedbackCollector()
            self.performance_tracker = PerformanceTracker()
            self.exploration_rate = exploration_rate
            
            # Initialize strategies
            if strategies is None:
                strategies = self._default_strategies()
            self.strategies = strategies
            
            # Current best strategy
            self.current_strategy = list(strategies.keys())[0]
            
            # Action counter for IDs
            self.action_count = 0
        
        def _default_strategies(self) -> Dict[str, ChatPromptTemplate]:
            """Default response strategies."""
            return {
                "detailed": ChatPromptTemplate.from_messages([
                    ("system", "You are a helpful assistant. Provide detailed, comprehensive answers."),
                    ("user", "{query}")
                ]),
                "concise": ChatPromptTemplate.from_messages([
                    ("system", "You are a helpful assistant. Provide concise, to-the-point answers."),
                    ("user", "{query}")
                ]),
                "friendly": ChatPromptTemplate.from_messages([
                    ("system", "You are a friendly, conversational assistant. Provide warm, engaging answers."),
                    ("user", "{query}")
                ]),
                "technical": ChatPromptTemplate.from_messages([
                    ("system", "You are a technical expert. Provide precise, technical answers with specifics."),
                    ("user", "{query}")
                ]),
            }
        
        def select_strategy(self) -> str:
            """
            Select a strategy using epsilon-greedy exploration.
            
            Returns:
                Strategy name
            """
            import random
            
            # Exploration: try random strategy
            if random.random()  Action:
            """
            Process a query using selected or specified strategy.
            
            Args:
                query: The query to process
                strategy: Optional specific strategy to use
                
            Returns:
                Action object with response
            """
            # Select strategy
            if strategy is None:
                strategy = self.select_strategy()
            
            # Get prompt for strategy
            prompt = self.strategies.get(strategy, self.strategies[self.current_strategy])
            
            # Generate response
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"query": query})
            
            # Create action
            self.action_count += 1
            action = Action(
                id=f"action_{self.action_count}",
                query=query,
                response=response,
                strategy=strategy,
                metadata={"exploration": strategy != self.current_strategy}
            )
            
            # Record action
            self.feedback_collector.record_action(action)
            
            return action
        
        def provide_feedback(
            self,
            action_id: str,
            signal: FeedbackSignal,
            score: float,
            feedback_type: FeedbackType = FeedbackType.EXPLICIT,
            details: str = ""
        ):
            """
            Provide feedback on an action.
            
            Args:
                action_id: ID of the action
                signal: Positive, negative, or neutral
                score: Numerical score (-1.0 to 1.0)
                feedback_type: Type of feedback
                details: Additional details
            """
            feedback = Feedback(
                action_id=action_id,
                signal=signal,
                score=score,
                feedback_type=feedback_type,
                details=details
            )
            
            self.feedback_collector.add_feedback(feedback)
            
            # Update performance tracking
            action = self.feedback_collector.actions.get(action_id)
            if action:
                self.performance_tracker.record_score(score, action.strategy)
        
        def learn_from_feedback(self):
            """
            Update strategy selection based on accumulated feedback.
            """
            metrics = self.performance_tracker.get_recent_performance()
            
            # Update current strategy to best performing
            if metrics.best_strategy in self.strategies:
                self.current_strategy = metrics.best_strategy
            
            # Adjust exploration rate based on improvement
            if metrics.improvement_rate > 0:
                # Doing well, reduce exploration
                self.exploration_rate = max(0.1, self.exploration_rate * 0.9)
            else:
                # Not improving, increase exploration
                self.exploration_rate = min(0.5, self.exploration_rate * 1.1)
        
        def get_performance_report(self) -> Dict[str, Any]:
            """Get comprehensive performance report."""
            metrics = self.performance_tracker.get_recent_performance()
            
            # Strategy breakdown
            strategy_performance = {}
            for strategy in self.strategies.keys():
                perf = self.performance_tracker.get_strategy_performance(strategy)
                if perf is not None:
                    strategy_performance[strategy] = perf
            
            return {
                "total_actions": metrics.total_actions,
                "average_score": metrics.average_score,
                "positive_feedback_rate": metrics.positive_feedback_rate,
                "negative_feedback_rate": metrics.negative_feedback_rate,
                "improvement_rate": metrics.improvement_rate,
                "current_strategy": self.current_strategy,
                "best_strategy": metrics.best_strategy,
                "exploration_rate": self.exploration_rate,
                "strategy_performance": strategy_performance,
                "recent_scores": metrics.recent_scores
            }
    
    
    def demonstrate_feedback_loops():
        """
        Demonstrates feedback loops with learning from outcomes.
        """
        print("=" * 80)
        print("FEEDBACK LOOPS DEMONSTRATION")
        print("=" * 80)
        
        # Create feedback loop agent
        agent = FeedbackLoopAgent(exploration_rate=0.3)
        
        # Show available strategies
        print("\n" + "=" * 80)
        print("Available Strategies")
        print("=" * 80)
        
        for strategy_name in agent.strategies.keys():
            print(f"  - {strategy_name}")
        
        print(f"\nInitial Strategy: {agent.current_strategy}")
        print(f"Exploration Rate: {agent.exploration_rate:.2f}")
        
        # Simulate interactions with feedback
        print("\n" + "=" * 80)
        print("Simulation: Learning from Feedback")
        print("=" * 80)
        
        # Test queries with different preference patterns
        test_scenarios = [
            # Users prefer concise answers for these
            ("What is Python?", "concise", 0.8),
            ("What is machine learning?", "concise", 0.9),
            ("Define API", "concise", 0.85),
            
            # Users prefer detailed answers for these
            ("How do neural networks work?", "detailed", 0.9),
            ("Explain quantum computing", "detailed", 0.85),
            ("How does blockchain work?", "detailed", 0.8),
            
            # Users prefer friendly tone for these
            ("Can you help me?", "friendly", 0.9),
            ("Thanks for your help!", "friendly", 0.95),
            
            # Technical questions prefer technical answers
            ("Explain Big O notation", "technical", 0.9),
            ("What is polymorphism?", "technical", 0.85),
        ]
        
        print("\nSimulating 10 interactions with feedback...")
        
        for i, (query, preferred_strategy, ideal_score) in enumerate(test_scenarios, 1):
            print(f"\n[Interaction {i}]")
            print(f"Query: {query}")
            
            # Agent processes query (may explore)
            action = agent.process_query(query)
            
            print(f"Strategy Used: {action.strategy}")
            print(f"Response Preview: {action.response[:100]}...")
            
            # Provide feedback based on whether strategy matched preference
            if action.strategy == preferred_strategy:
                # Good match
                score = ideal_score
                signal = FeedbackSignal.POSITIVE
                details = "Strategy matched user preference"
            else:
                # Suboptimal match
                score = ideal_score * 0.5  # Reduced score
                signal = FeedbackSignal.NEUTRAL
                details = "Strategy didn't match user preference"
            
            agent.provide_feedback(
                action_id=action.id,
                signal=signal,
                score=score,
                feedback_type=FeedbackType.IMMEDIATE,
                details=details
            )
            
            print(f"Feedback: {signal.value} (score: {score:.2f})")
            
            # Periodically learn from feedback
            if i % 3 == 0:
                agent.learn_from_feedback()
                print(f"\n  → Learning applied. Current strategy: {agent.current_strategy}")
                print(f"  → Exploration rate adjusted to: {agent.exploration_rate:.2f}")
        
        # Show performance report
        print("\n" + "=" * 80)
        print("Performance Report")
        print("=" * 80)
        
        report = agent.get_performance_report()
        
        print(f"\nTotal Actions: {report['total_actions']}")
        print(f"Average Score: {report['average_score']:.3f}")
        print(f"Positive Feedback Rate: {report['positive_feedback_rate']:.1%}")
        print(f"Negative Feedback Rate: {report['negative_feedback_rate']:.1%}")
        print(f"Improvement Rate: {report['improvement_rate']:+.3f}")
        
        print(f"\nCurrent Strategy: {report['current_strategy']}")
        print(f"Best Strategy: {report['best_strategy']}")
        print(f"Exploration Rate: {report['exploration_rate']:.2f}")
        
        print("\nStrategy Performance:")
        for strategy, score in sorted(report['strategy_performance'].items(), key=lambda x: x[1], reverse=True):
            print(f"  - {strategy}: {score:.3f}")
        
        print("\nRecent Scores:")
        for i, score in enumerate(report['recent_scores'][-5:], 1):
            print(f"  {i}. {score:.3f}")
        
        # Test learned behavior
        print("\n" + "=" * 80)
        print("Testing Learned Behavior")
        print("=" * 80)
        
        # Reduce exploration to see what agent learned
        agent.exploration_rate = 0.0
        
        test_queries = [
            "What is Docker?",  # Should use concise (learned preference)
            "How does cryptography work?",  # Should use detailed
        ]
        
        for query in test_queries:
            print(f"\nQuery: {query}")
            action = agent.process_query(query)
            print(f"Selected Strategy: {action.strategy}")
            print(f"Response: {action.response[:150]}...")
        
        # Summary
        print("\n" + "=" * 80)
        print("SUMMARY")
        print("=" * 80)
        print("""
    Feedback Loops provide:
    ✓ Learning from action outcomes
    ✓ Performance tracking over time
    ✓ Strategy adaptation based on feedback
    ✓ Exploration vs exploitation balance
    ✓ Continuous improvement
    ✓ Multi-strategy experimentation
    
    This pattern excels at:
    - Adaptive systems
    - Preference learning
    - Continuous optimization
    - A/B testing automation
    - Reinforcement learning
    - User personalization
    
    Feedback components:
    1. Feedback Collector: Gathers outcome data
    2. Performance Tracker: Monitors metrics
    3. Learning Mechanism: Updates behavior
    4. Reward Signal: Quantifies quality
    5. Adaptation Strategy: Adjusts approach
    
    Feedback types:
    - IMMEDIATE: Real-time corrections
    - DELAYED: Outcome-based rewards
    - EXPLICIT: Direct user feedback
    - IMPLICIT: Inferred from behavior
    - COMPARATIVE: A/B test results
    
    Feedback signals:
    - POSITIVE: Good outcome
    - NEGATIVE: Bad outcome
    - NEUTRAL: Acceptable outcome
    - MIXED: Some good, some bad
    
    Learning strategies:
    - Epsilon-greedy: Balance exploration/exploitation
    - Performance-based: Use best performing
    - Adaptive exploration: Adjust based on improvement
    - Multi-armed bandit: Test multiple strategies
    - Gradient-based: Continuous optimization
    
    Performance metrics:
    - Average score: Overall quality
    - Positive/negative rates: Feedback distribution
    - Improvement rate: Learning trajectory
    - Strategy performance: Compare approaches
    - Recent trends: Current performance
    
    Benefits:
    - Adaptation: Improves over time
    - Personalization: Learns preferences
    - Optimization: Finds best strategies
    - Resilience: Adapts to changes
    - Efficiency: Focuses on what works
    - Transparency: Track improvement
    
    Use Feedback Loops when you need:
    - Continuous improvement systems
    - Adaptive recommendation engines
    - Reinforcement learning agents
    - User preference learning
    - A/B testing automation
    - Self-optimizing systems
    """)
    
    
    if __name__ == "__main__":
        demonstrate_feedback_loops()
    
        

    Agentic Design Pattern: Feedback Loops

    Pattern 039: Feedback Loops – Line-by-Line Explanation

    Overview

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


    Pattern Description

    Pattern 039: Feedback Loops

    Description: Feedback Loops enable agents to learn from the outcomes of their actions through immediate feedback (real-time corrections) and delayed feedback (outcome-based rewards). The agent uses feedback to improve performance, adjust strategies, and adapt behavior over time.

    Components: – Feedback Collector: Gathers outcome data – Performance Tracker: Monitors metrics over time – Learning Mechanism: Incorporates feedback into behavior – Reward Signal: Quantifies outcome quality – Adaptation Strategy: How to adjust based on feedback

    Use Cases: – Continuous improvement systems – Reinforcement learning agents – User preference learning – Model fine-tuning – A/B testing and optimization – Adaptive recommendation systems

    LangChain Implementation: Uses feedback collection, performance tracking, and iterative improvement mechanisms to enable agents to learn from action outcomes.


    File Statistics

    • Total Lines: 612
    • Classes: 8
    • Functions: 1
    • Constants: 0

    Module Documentation

    """
    Pattern 039: Feedback Loops
    
    Description:
        Feedback Loops enable agents to learn from the outcomes of their actions through
        immediate feedback (real-time corrections) and delayed feedback (outcome-based
        rewards). The agent uses feedback to improve performance, adjust strategies,
        and adapt behavior over time.
    
    Components:
        - Feedback Collector: Gathers outcome data
        - Performance Tracker: Monitors metrics over time
        - Learning Mechanism: Incorporates feedback into behavior
        - Reward Signal: Quantifies outcome quality
        - Adaptation Strategy: How to adjust based on feedback
    
    Use Cases:
        - Continuous improvement systems
        - Reinforcement learning agents
        - User preference learning
        - Model fine-tuning
        - A/B testing and optimization
        - Adaptive recommendation systems
    
    LangChain Implementation:
        Uses feedback collection, performance tracking, and iterative improvement
        mechanisms to enable agents to learn from action outcomes.
    """
    

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


    Import Statements

    Standard Library Imports

    import dataclasses.dataclass
    import dataclasses.field
    import datetime.datetime
    import datetime.timedelta
    import enum.Enum
    import langchain_core.output_parsers.StrOutputParser
    import langchain_core.prompts.ChatPromptTemplate
    import os
    import random
    import typing.Any
    import typing.Callable
    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 – datetime.datetime: Date and time handling – datetime.timedelta: Date and time handling – 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 – random: Random number generation – 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 statistics.mean
    from statistics.stdev
    

    Purpose: Additional dependencies for specialized functionality.


    Classes

    Class: FeedbackType

    Lines 45-51

    Inherits from: Enum

    Purpose:

    Types of feedback.
    

    Class: FeedbackSignal

    Lines 54-59

    Inherits from: Enum

    Purpose:

    Feedback signals.
    

    Class: Feedback

    Lines 63-71

    Purpose:

    Feedback on an action or response.
    

    Class: Action

    Lines 75-82

    Purpose:

    An action taken by the agent.
    

    Class: PerformanceMetrics

    Lines 86-95

    Purpose:

    Performance metrics over time.
    

    Class: FeedbackCollector

    Lines 98-135

    Purpose:

    Collects and stores feedback on agent actions.
    

    Methods:

    __init__(self)

    Lines 103-105

    Purpose: Initialize the class instance with required parameters.

    Parameters:

    record_action(self, action)

    Lines 107-110

    Purpose:

    Record an action taken by the agent.
    

    Parameters: – action: Action

    add_feedback(self, feedback)

    Lines 112-117

    Purpose:

    Add feedback for an action.
    

    Parameters: – feedback: Feedback

    get_action_feedback(self, action_id)

    Lines 119-121

    Purpose:

    Get all feedback for an action.
    

    Parameters: – action_id: Action id

    Returns: <ast.Subscript object at 0x000001A06D73A190>

    get_average_score(self, action_ids)

    Lines 123-135

    Purpose:

    Get average feedback score.
    

    Parameters: – action_ids: Action ids

    Returns: float


    Class: PerformanceTracker

    Lines 138-210

    Purpose:

    Tracks performance metrics over time.
    

    Methods:

    __init__(self, window_size)

    Lines 143-146

    Purpose: Initialize the class instance with required parameters.

    Parameters: – window_size: Window size

    record_score(self, score, strategy)

    Lines 148-154

    Purpose:

    Record a performance score.
    

    Parameters: – score: Score – strategy: Strategy

    get_recent_performance(self)

    Lines 156-204

    Purpose:

    Get recent performance metrics.
    

    Parameters:

    Returns: PerformanceMetrics

    get_strategy_performance(self, strategy)

    Lines 206-210

    Purpose:

    Get average performance for a strategy.
    

    Parameters: – strategy: Strategy

    Returns: <ast.Subscript object at 0x000001A06D620250>


    Class: FeedbackLoopAgent

    Lines 213-402

    Purpose:

    Agent that learns from feedback to improve performance.
    
    Features:
    - Immediate and delayed feedback
    - Performance tracking
    - Strategy adaptation
    - Continuous improvement
    - Multi-strategy experimentation
    

    Methods:

    __init__(self, strategies, temperature, exploration_rate)

    Lines 225-245

    Purpose: Initialize the class instance with required parameters.

    Parameters: – strategies: Strategies – temperature: Sampling temperature (0-1) – exploration_rate: Exploration rate

    _default_strategies(self)

    Lines 247-266

    Purpose:

    Default response strategies.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D6FC4D0>

    select_strategy(self)

    Lines 268-286

    Purpose:

    Select a strategy using epsilon-greedy exploration.
    
    Returns:
        Strategy name
    

    Parameters:

    Returns: str

    process_query(self, query, strategy)

    Lines 288-327

    Purpose:

    Process a query using selected or specified strategy.
    
    Args:
        query: The query to process
        strategy: Optional specific strategy to use
    
    Returns:
        Action object with response
    

    Parameters: – query: Input query or question – strategy: Strategy

    Returns: Action

    provide_feedback(self, action_id, signal, score, feedback_type, details)

    Lines 329-360

    Purpose:

    Provide feedback on an action.
    
    Args:
        action_id: ID of the action
        signal: Positive, negative, or neutral
        score: Numerical score (-1.0 to 1.0)
        feedback_type: Type of feedback
        details: Additional details
    

    Parameters: – action_id: Action id – signal: Signal – score: Score – feedback_type: Feedback type – details: Details

    learn_from_feedback(self)

    Lines 362-378

    Purpose:

    Update strategy selection based on accumulated feedback.
    

    Parameters:

    get_performance_report(self)

    Lines 380-402

    Purpose:

    Get comprehensive performance report.
    

    Parameters:

    Returns: <ast.Subscript object at 0x000001A06D16A0D0>


    Functions

    demonstrate_feedback_loops()

    Lines 405-607

    Purpose:

    Demonstrates feedback loops with learning from outcomes.
    

    Key Concepts Summary

    Pattern Overview

    The Feedback Loops 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 038
    • Pattern 040

    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 Feedback Loops 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: 039_feedback_loops.py
    Total Lines: 612
    Pattern Category: AI Agent Design Pattern #039


    Generated from actual implementation code
    For educational and reference purposes