"""
Pattern 155: Regression Testing

Description:
    Automated regression testing detects unintended changes in agent behavior
    when code, models, or prompts are updated. Maintains baselines, compares
    outputs, and flags regressions automatically.

Components:
    - Baseline storage and versioning
    - Test case management
    - Output comparison
    - Regression detection
    - Change analysis
    - Historical tracking

Use Cases:
    - Model version updates
    - Prompt engineering iterations
    - Code refactoring validation
    - CI/CD integration
    - Quality assurance

Benefits:
    - Early regression detection
    - Automated quality checks
    - Historical performance tracking
    - Confidence in updates

Trade-offs:
    - Baseline maintenance overhead
    - Storage requirements
    - Comparison complexity for creative tasks
    - False positive management

LangChain Implementation:
    Uses ChatOpenAI for testing, JSON storage for baselines,
    similarity metrics for comparison, and automated detection
"""

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

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

load_dotenv()


class ChangeType(Enum):
    """Types of changes detected"""
    NO_CHANGE = "no_change"
    IMPROVEMENT = "improvement"
    REGRESSION = "regression"
    NEUTRAL = "neutral"
    ERROR = "error"


class Severity(Enum):
    """Severity levels for regressions"""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"


@dataclass
class TestCase:
    """Single test case"""
    id: str
    input: str
    category: str
    expected_output: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class TestResult:
    """Result of a single test execution"""
    test_id: str
    input: str
    output: str
    timestamp: str
    execution_time: float
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Comparison:
    """Comparison between baseline and current result"""
    test_id: str
    baseline_output: str
    current_output: str
    similarity_score: float
    change_type: ChangeType
    severity: Severity
    details: str
    timestamp: str


@dataclass
class RegressionReport:
    """Comprehensive regression test report"""
    version: str
    timestamp: str
    total_tests: int
    passed: int
    regressions: int
    improvements: int
    errors: int
    comparisons: List[Comparison]
    summary: str


class RegressionTester:
    """Manages regression testing for agent outputs"""
    
    def __init__(self, baseline_file: str = "baselines.json", 
                 model_name: str = "gpt-3.5-turbo"):
        """
        Initialize regression tester
        
        Args:
            baseline_file: Path to baseline storage
            model_name: LLM model to use for testing
        """
        self.llm = ChatOpenAI(model=model_name, temperature=0)
        self.baseline_file = baseline_file
        self.test_cases: List[TestCase] = []
        self.baselines: Dict[str, TestResult] = {}
        self.load_baselines()
        
        # Comparison LLM for semantic analysis
        self.comparison_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        
    def load_baselines(self):
        """Load baseline results from file"""
        if os.path.exists(self.baseline_file):
            try:
                with open(self.baseline_file, 'r') as f:
                    data = json.load(f)
                    self.baselines = {
                        k: TestResult(**v) for k, v in data.items()
                    }
                print(f"✓ Loaded {len(self.baselines)} baselines from {self.baseline_file}")
            except Exception as e:
                print(f"Error loading baselines: {e}")
                self.baselines = {}
        else:
            print(f"No baseline file found at {self.baseline_file}")
            self.baselines = {}
    
    def save_baselines(self):
        """Save baseline results to file"""
        try:
            data = {
                k: {
                    'test_id': v.test_id,
                    'input': v.input,
                    'output': v.output,
                    'timestamp': v.timestamp,
                    'execution_time': v.execution_time,
                    'error': v.error,
                    'metadata': v.metadata
                }
                for k, v in self.baselines.items()
            }
            with open(self.baseline_file, 'w') as f:
                json.dump(data, f, indent=2)
            print(f"✓ Saved {len(self.baselines)} baselines to {self.baseline_file}")
        except Exception as e:
            print(f"Error saving baselines: {e}")
    
    def add_test_case(self, test_case: TestCase):
        """Add a test case"""
        self.test_cases.append(test_case)
    
    def run_test(self, test_case: TestCase) -> TestResult:
        """Execute a single test case"""
        import time
        
        start_time = time.time()
        error = None
        output = ""
        
        try:
            # Create prompt for the test
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful AI assistant. Provide clear, accurate responses."),
                ("user", "{input}")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            output = chain.invoke({"input": test_case.input})
            
        except Exception as e:
            error = str(e)
            output = f"ERROR: {error}"
        
        execution_time = time.time() - start_time
        
        return TestResult(
            test_id=test_case.id,
            input=test_case.input,
            output=output,
            timestamp=datetime.now().isoformat(),
            execution_time=execution_time,
            error=error,
            metadata=test_case.metadata.copy()
        )
    
    def calculate_similarity(self, text1: str, text2: str) -> float:
        """Calculate similarity between two texts"""
        # Use SequenceMatcher for basic similarity
        return SequenceMatcher(None, text1.lower(), text2.lower()).ratio()
    
    def analyze_change(self, baseline: TestResult, current: TestResult) -> Comparison:
        """Analyze changes between baseline and current result"""
        similarity = self.calculate_similarity(baseline.output, current.output)
        
        # Determine change type
        if current.error:
            change_type = ChangeType.ERROR
            severity = Severity.CRITICAL
            details = f"Error occurred: {current.error}"
        elif similarity >= 0.95:
            change_type = ChangeType.NO_CHANGE
            severity = Severity.INFO
            details = "Output is virtually identical to baseline"
        else:
            # Use LLM to classify the change
            change_type, severity, details = self._llm_classify_change(
                baseline.output, current.output, similarity
            )
        
        return Comparison(
            test_id=current.test_id,
            baseline_output=baseline.output,
            current_output=current.output,
            similarity_score=similarity,
            change_type=change_type,
            severity=severity,
            details=details,
            timestamp=datetime.now().isoformat()
        )
    
    def _llm_classify_change(self, baseline: str, current: str, 
                           similarity: float) -> tuple[ChangeType, Severity, str]:
        """Use LLM to classify the nature of the change"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at analyzing changes in AI outputs.
Compare the baseline and current outputs and determine:
1. Change type: improvement, regression, or neutral
2. Severity: critical, high, medium, low, info
3. Brief explanation

Respond in JSON format: {{"change_type": "...", "severity": "...", "details": "..."}}"""),
            ("user", """Baseline output: {baseline}

Current output: {current}

Similarity score: {similarity:.2f}

Analyze the change:""")
        ])
        
        try:
            chain = prompt | self.comparison_llm | StrOutputParser()
            response = chain.invoke({
                "baseline": baseline,
                "current": current,
                "similarity": similarity
            })
            
            # Parse JSON response
            result = json.loads(response)
            change_type = ChangeType(result.get('change_type', 'neutral'))
            severity = Severity(result.get('severity', 'info'))
            details = result.get('details', 'No details provided')
            
        except Exception as e:
            # Fallback classification
            if similarity < 0.5:
                change_type = ChangeType.REGRESSION
                severity = Severity.HIGH
            elif similarity  RegressionReport:
        """Run all tests and compare against baseline"""
        print(f"\n=== Running Regression Tests (Version {version}) ===\n")
        
        if not self.baselines:
            print("No baseline found. Please establish baseline first.")
            return None
        
        comparisons = []
        passed = 0
        regressions = 0
        improvements = 0
        errors = 0
        
        for test_case in self.test_cases:
            print(f"Testing: {test_case.id}")
            
            # Run current test
            current = self.run_test(test_case)
            
            # Compare with baseline if exists
            if test_case.id in self.baselines:
                baseline = self.baselines[test_case.id]
                comparison = self.analyze_change(baseline, current)
                comparisons.append(comparison)
                
                # Update counters
                if comparison.change_type == ChangeType.NO_CHANGE:
                    passed += 1
                    print(f"  ✓ PASS (no change)")
                elif comparison.change_type == ChangeType.REGRESSION:
                    regressions += 1
                    print(f"  ✗ REGRESSION ({comparison.severity.value}): {comparison.details}")
                elif comparison.change_type == ChangeType.IMPROVEMENT:
                    improvements += 1
                    print(f"  ↑ IMPROVEMENT: {comparison.details}")
                elif comparison.change_type == ChangeType.ERROR:
                    errors += 1
                    print(f"  ⚠ ERROR: {comparison.details}")
                else:
                    passed += 1
                    print(f"  ~ NEUTRAL: {comparison.details}")
            else:
                print(f"  ⚠ No baseline found for this test")
        
        # Generate summary
        total = len(self.test_cases)
        summary = self._generate_summary(passed, regressions, improvements, errors, total)
        
        report = RegressionReport(
            version=version,
            timestamp=datetime.now().isoformat(),
            total_tests=total,
            passed=passed,
            regressions=regressions,
            improvements=improvements,
            errors=errors,
            comparisons=comparisons,
            summary=summary
        )
        
        print(f"\n{summary}")
        return report
    
    def _generate_summary(self, passed: int, regressions: int, 
                         improvements: int, errors: int, total: int) -> str:
        """Generate test summary"""
        lines = [
            "=== Regression Test Summary ===",
            f"Total Tests: {total}",
            f"Passed: {passed} ({passed/total*100:.1f}%)",
            f"Regressions: {regressions} ({regressions/total*100:.1f}%)",
            f"Improvements: {improvements} ({improvements/total*100:.1f}%)",
            f"Errors: {errors} ({errors/total*100:.1f}%)"
        ]
        
        if regressions == 0 and errors == 0:
            lines.append("\n✓ All tests passed! No regressions detected.")
        elif regressions > 0:
            lines.append(f"\n⚠ WARNING: {regressions} regression(s) detected!")
        
        return "\n".join(lines)
    
    def get_regression_details(self, report: RegressionReport) -> str:
        """Get detailed regression information"""
        regressions = [c for c in report.comparisons 
                      if c.change_type == ChangeType.REGRESSION]
        
        if not regressions:
            return "No regressions found."
        
        details = ["\n=== Regression Details ===\n"]
        for i, reg in enumerate(regressions, 1):
            details.append(f"{i}. Test: {reg.test_id}")
            details.append(f"   Severity: {reg.severity.value}")
            details.append(f"   Similarity: {reg.similarity_score:.2%}")
            details.append(f"   Details: {reg.details}")
            details.append(f"   Baseline: {reg.baseline_output[:100]}...")
            details.append(f"   Current:  {reg.current_output[:100]}...")
            details.append("")
        
        return "\n".join(details)


def demonstrate_regression_testing():
    """Demonstrate regression testing pattern"""
    print("=" * 80)
    print("REGRESSION TESTING PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Initialize tester
    tester = RegressionTester(baseline_file="test_baselines.json")
    
    # Define test cases
    test_cases = [
        TestCase(
            id="test_001",
            input="What is the capital of France?",
            category="factual"
        ),
        TestCase(
            id="test_002",
            input="Explain photosynthesis in simple terms.",
            category="explanation"
        ),
        TestCase(
            id="test_003",
            input="Write a haiku about coding.",
            category="creative"
        ),
        TestCase(
            id="test_004",
            input="Calculate 15% of 200.",
            category="calculation"
        ),
        TestCase(
            id="test_005",
            input="List three benefits of exercise.",
            category="informational"
        ),
    ]
    
    # Add test cases
    for tc in test_cases:
        tester.add_test_case(tc)
    
    # Example 1: Establish baseline
    print("\n" + "="*80)
    print("EXAMPLE 1: Establishing Baseline")
    print("="*80)
    tester.establish_baseline(version="1.0.0")
    
    # Example 2: Run regression tests (should pass)
    print("\n" + "="*80)
    print("EXAMPLE 2: Running Regression Tests (Same Model)")
    print("="*80)
    report = tester.run_regression_tests(version="1.0.1")
    
    # Example 3: Simulate model change and test
    print("\n" + "="*80)
    print("EXAMPLE 3: Testing with Different Model")
    print("="*80)
    tester_v2 = RegressionTester(baseline_file="test_baselines.json", 
                                  model_name="gpt-3.5-turbo")
    tester_v2.test_cases = test_cases
    tester_v2.load_baselines()
    report_v2 = tester_v2.run_regression_tests(version="2.0.0")
    
    # Example 4: Show regression details
    if report_v2 and report_v2.regressions > 0:
        print("\n" + "="*80)
        print("EXAMPLE 4: Regression Details")
        print("="*80)
        print(tester_v2.get_regression_details(report_v2))
    
    # Example 5: Compare specific test
    print("\n" + "="*80)
    print("EXAMPLE 5: Detailed Comparison for One Test")
    print("="*80)
    test_id = "test_001"
    if report and any(c.test_id == test_id for c in report.comparisons):
        comparison = next(c for c in report.comparisons if c.test_id == test_id)
        print(f"Test ID: {comparison.test_id}")
        print(f"Change Type: {comparison.change_type.value}")
        print(f"Severity: {comparison.severity.value}")
        print(f"Similarity: {comparison.similarity_score:.2%}")
        print(f"Details: {comparison.details}")
        print(f"\nBaseline Output:\n{comparison.baseline_output}")
        print(f"\nCurrent Output:\n{comparison.current_output}")
    
    # Summary
    print("\n" + "="*80)
    print("SUMMARY: Regression Testing Best Practices")
    print("="*80)
    print("""
1. BASELINE MANAGEMENT:
   - Establish baselines for stable versions
   - Version baselines appropriately
   - Update baselines when intentional changes are made
   - Store baselines in version control

2. TEST CASE DESIGN:
   - Cover diverse input types (factual, creative, calculation)
   - Include edge cases and error scenarios
   - Categorize tests for targeted analysis
   - Keep test cases stable and representative

3. COMPARISON STRATEGIES:
   - Use similarity metrics for objective comparison
   - Employ LLM-based semantic analysis
   - Define severity thresholds
   - Track both regressions and improvements

4. AUTOMATION:
   - Integrate into CI/CD pipeline
   - Run on every code/model change
   - Generate automated reports
   - Alert on critical regressions

5. CONTINUOUS IMPROVEMENT:
   - Review flagged regressions manually
   - Adjust baselines for intentional improvements
   - Expand test coverage over time
   - Track regression trends

Benefits:
✓ Early detection of unintended changes
✓ Confidence in updates and refactoring
✓ Historical performance tracking
✓ Automated quality assurance
✓ Reduced manual testing overhead
✓ Data-driven decision making

Challenges:
- Baseline maintenance for creative outputs
- False positives from legitimate variations
- Storage and comparison overhead
- Defining acceptable change thresholds
    """)


if __name__ == "__main__":
    demonstrate_regression_testing()

    

Agentic Design Pattern: Regression Testing

Pattern 155: Regression Testing – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 155: Regression Testing

Description: Automated regression testing detects unintended changes in agent behavior when code, models, or prompts are updated. Maintains baselines, compares outputs, and flags regressions automatically.

Components: – Baseline storage and versioning – Test case management – Output comparison – Regression detection – Change analysis – Historical tracking

Use Cases: – Model version updates – Prompt engineering iterations – Code refactoring validation – CI/CD integration – Quality assurance

Benefits: – Early regression detection – Automated quality checks – Historical performance tracking – Confidence in updates

Trade-offs: – Baseline maintenance overhead – Storage requirements – Comparison complexity for creative tasks – False positive management

LangChain Implementation: Uses ChatOpenAI for testing, JSON storage for baselines, similarity metrics for comparison, and automated detection

---

File Statistics

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

---

Module Documentation

"""
Pattern 155: Regression Testing

Description:
    Automated regression testing detects unintended changes in agent behavior
    when code, models, or prompts are updated. Maintains baselines, compares
    outputs, and flags regressions automatically.

Components:
    - Baseline storage and versioning
    - Test case management
    - Output comparison
    - Regression detection
    - Change analysis
    - Historical tracking

Use Cases:
    - Model version updates
    - Prompt engineering iterations
    - Code refactoring validation
    - CI/CD integration
    - Quality assurance

Benefits:
    - Early regression detection
    - Automated quality checks
    - Historical performance tracking
    - Confidence in updates

Trade-offs:
    - Baseline maintenance overhead
    - Storage requirements
    - Comparison complexity for creative tasks
    - False positive management

LangChain Implementation:
    Uses ChatOpenAI for testing, JSON storage for baselines,
    similarity metrics for comparison, and automated detection
"""

Purpose: This module implements the Regression Testing 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 time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

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

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from difflib.SequenceMatcher
from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ChangeType

Lines 57-63

Inherits from: Enum

Purpose:

Types of changes detected

---

Class: Severity

Lines 66-72

Inherits from: Enum

Purpose:

Severity levels for regressions

---

Class: TestCase

Lines 76-82

Purpose:

Single test case

---

Class: TestResult

Lines 86-94

Purpose:

Result of a single test execution

---

Class: Comparison

Lines 98-107

Purpose:

Comparison between baseline and current result

---

Class: RegressionReport

Lines 111-121

Purpose:

Comprehensive regression test report

---

Class: RegressionTester

Lines 124-422

Purpose:

Manages regression testing for agent outputs

Methods:

__init__(self, baseline_file, model_name)

Lines 127-143

Purpose:

Initialize regression tester

Args:
    baseline_file: Path to baseline storage
    model_name: LLM model to use for testing

Parameters: – baseline_file: Baseline file – model_name: Language model instance

load_baselines(self)

Lines 145-160

Purpose:

Load baseline results from file

Parameters:

save_baselines(self)

Lines 162-181

Purpose:

Save baseline results to file

Parameters:

add_test_case(self, test_case)

Lines 183-185

Purpose:

Add a test case

Parameters: – test_case: Test case

run_test(self, test_case)

Lines 187-219

Purpose:

Execute a single test case

Parameters: – test_case: Test case

Returns: TestResult

calculate_similarity(self, text1, text2)

Lines 221-224

Purpose:

Calculate similarity between two texts

Parameters: – text1: Text1 – text2: Text2

Returns: float

analyze_change(self, baseline, current)

Lines 226-254

Purpose:

Analyze changes between baseline and current result

Parameters: – baseline: Baseline – current: Current

Returns: Comparison

_llm_classify_change(self, baseline, current, similarity)

Lines 256-303

Purpose:

Use LLM to classify the nature of the change

Parameters: – baseline: Baseline – current: Current – similarity: Similarity

Returns: <ast.Subscript object at 0x000001A06D7016D0>

establish_baseline(self, version)

Lines 305-319

Purpose:

Run tests and establish baseline

Parameters: – version: Version

run_regression_tests(self, version)

Lines 321-383

Purpose:

Run all tests and compare against baseline

Parameters: – version: Version

Returns: RegressionReport

_generate_summary(self, passed, regressions, improvements, errors, total)

Lines 385-402

Purpose:

Generate test summary

Parameters: – passed: Passed – regressions: Regressions – improvements: Improvements – errors: Errors – total: Total

Returns: str

get_regression_details(self, report)

Lines 404-422

Purpose:

Get detailed regression information

Parameters: – report: Report

Returns: str

---

Functions

demonstrate_regression_testing()

Lines 425-559

Purpose:

Demonstrate regression testing pattern

---

Key Concepts Summary

Pattern Overview

The Regression Testing 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 154
• Pattern 156

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 Regression Testing 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: 155_regression_testing.py
Total Lines: 564
Pattern Category: AI Agent Design Pattern #155

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 154: Canary Deployment

Description:
    Implements gradual rollout of new agent versions by routing a small
    percentage of traffic to the canary version, monitoring metrics, and
    automatically rolling back if issues are detected.

Components:
    - Traffic Splitter: Routes requests between versions
    - Health Monitor: Tracks canary metrics
    - Rollback Controller: Automatic rollback on issues
    - Metrics Collector: Performance and error tracking
    - Gradual Ramp: Increases traffic percentage over time

Use Cases:
    - Safe production deployments
    - Gradual feature rollout
    - Risk mitigation for changes
    - A/B testing with safety
    - Automated deployment validation

Benefits:
    - Minimal blast radius
    - Automatic rollback
    - Real user validation
    - Gradual confidence building
    - Production safety

Trade-offs:
    - Deployment complexity
    - Monitoring overhead
    - Slower full rollout
    - Dual version maintenance
    - Metrics interpretation

LangChain Implementation:
    Uses traffic routing with health monitoring and automatic rollback.
"""

import os
import time
import random
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from collections import defaultdict
from dotenv import load_dotenv

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

load_dotenv()


class DeploymentStatus(Enum):
    """Status of canary deployment"""
    STARTING = "starting"
    HEALTHY = "healthy"
    WARNING = "warning"
    ROLLING_BACK = "rolling_back"
    ROLLED_BACK = "rolled_back"
    PROMOTED = "promoted"


@dataclass
class CanaryMetrics:
    """Metrics for canary and production versions"""
    # Request counts
    production_requests: int = 0
    canary_requests: int = 0
    
    # Success rates
    production_successes: int = 0
    canary_successes: int = 0
    production_errors: int = 0
    canary_errors: int = 0
    
    # Latency
    production_latency_sum: float = 0.0
    canary_latency_sum: float = 0.0
    
    def get_production_error_rate(self) -> float:
        """Calculate production error rate"""
        if self.production_requests == 0:
            return 0.0
        return self.production_errors / self.production_requests
    
    def get_canary_error_rate(self) -> float:
        """Calculate canary error rate"""
        if self.canary_requests == 0:
            return 0.0
        return self.canary_errors / self.canary_requests
    
    def get_production_avg_latency(self) -> float:
        """Calculate average production latency"""
        if self.production_requests == 0:
            return 0.0
        return self.production_latency_sum / self.production_requests
    
    def get_canary_avg_latency(self) -> float:
        """Calculate average canary latency"""
        if self.canary_requests == 0:
            return 0.0
        return self.canary_latency_sum / self.canary_requests


@dataclass
class CanaryConfig:
    """Configuration for canary deployment"""
    initial_traffic_percent: float = 5.0  # Start with 5%
    max_traffic_percent: float = 50.0  # Max before promotion
    ramp_step: float = 5.0  # Increase by 5% each step
    ramp_interval_seconds: float = 60.0  # Wait 60s between steps
    
    # Health thresholds
    max_error_rate_increase: float = 0.05  # Max 5% error rate increase
    max_latency_increase: float = 0.20  # Max 20% latency increase
    min_requests_for_decision: int = 10  # Min requests before decisions
    
    # Rollback settings
    auto_rollback: bool = True
    rollback_on_error_spike: bool = True


class CanaryDeployment:
    """
    Manages canary deployment with gradual traffic routing and
    automatic rollback on issues.
    """
    
    def __init__(self, config: Optional[CanaryConfig] = None):
        """Initialize canary deployment"""
        self.config = config or CanaryConfig()
        
        # Production version (stable)
        self.production_llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0.7
        )
        
        # Canary version (new version being tested)
        self.canary_llm = ChatOpenAI(
            model="gpt-3.5-turbo",
            temperature=0.8  # Different config for canary
        )
        
        # State
        self.current_traffic_percent = self.config.initial_traffic_percent
        self.status = DeploymentStatus.STARTING
        self.metrics = CanaryMetrics()
        self.deployment_start_time = datetime.now()
        self.last_ramp_time = time.time()
        
        # Prompts
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant."),
            ("user", "{query}")
        ])
    
    def route_request(self, query: str) -> Dict[str, Any]:
        """
        Route request to either production or canary based on traffic split
        
        Args:
            query: User query
            
        Returns:
            Response with version and metrics
        """
        # Decide routing
        use_canary = random.random() * 100 < self.current_traffic_percent
        
        start_time = time.time()
        
        try:
            if use_canary:
                # Route to canary
                chain = self.prompt | self.canary_llm | StrOutputParser()
                result = chain.invoke({"query": query})
                version = "canary"
                
                # Update canary metrics
                self.metrics.canary_requests += 1
                self.metrics.canary_successes += 1
                latency = time.time() - start_time
                self.metrics.canary_latency_sum += latency
                
            else:
                # Route to production
                chain = self.prompt | self.production_llm | StrOutputParser()
                result = chain.invoke({"query": query})
                version = "production"
                
                # Update production metrics
                self.metrics.production_requests += 1
                self.metrics.production_successes += 1
                latency = time.time() - start_time
                self.metrics.production_latency_sum += latency
            
            # Check health after each request
            self._check_health()
            
            # Consider ramping up traffic
            self._consider_ramp_up()
            
            return {
                "success": True,
                "result": result,
                "version": version,
                "latency": latency,
                "canary_traffic_percent": self.current_traffic_percent,
                "status": self.status.value
            }
            
        except Exception as e:
            # Track error
            latency = time.time() - start_time
            
            if use_canary:
                self.metrics.canary_requests += 1
                self.metrics.canary_errors += 1
                self.metrics.canary_latency_sum += latency
                version = "canary"
            else:
                self.metrics.production_requests += 1
                self.metrics.production_errors += 1
                self.metrics.production_latency_sum += latency
                version = "production"
            
            # Check if we should rollback
            self._check_health()
            
            return {
                "success": False,
                "error": str(e),
                "version": version,
                "latency": latency,
                "canary_traffic_percent": self.current_traffic_percent,
                "status": self.status.value
            }
    
    def _check_health(self):
        """Check canary health and trigger rollback if needed"""
        if self.status == DeploymentStatus.ROLLED_BACK:
            return  # Already rolled back
        
        if self.metrics.canary_requests  0:
            error_rate_increase = (canary_error_rate - prod_error_rate) / prod_error_rate
        else:
            error_rate_increase = canary_error_rate
        
        # Check latency
        if prod_latency > 0:
            latency_increase = (canary_latency - prod_latency) / prod_latency
        else:
            latency_increase = 0.0
        
        # Determine health
        issues = []
        
        if error_rate_increase > self.config.max_error_rate_increase:
            issues.append(f"Error rate {error_rate_increase*100:.1f}% higher")
        
        if latency_increase > self.config.max_latency_increase:
            issues.append(f"Latency {latency_increase*100:.1f}% higher")
        
        if issues:
            if self.config.auto_rollback:
                print(f"\n🚨 ROLLBACK TRIGGERED: {', '.join(issues)}")
                self._rollback()
            else:
                self.status = DeploymentStatus.WARNING
                print(f"\n⚠️  WARNING: {', '.join(issues)}")
        elif self.status != DeploymentStatus.PROMOTED:
            self.status = DeploymentStatus.HEALTHY
    
    def _consider_ramp_up(self):
        """Consider increasing traffic to canary"""
        if self.status != DeploymentStatus.HEALTHY:
            return  # Only ramp if healthy
        
        if self.current_traffic_percent >= self.config.max_traffic_percent:
            return  # Already at max
        
        # Check if enough time has passed
        time_since_last_ramp = time.time() - self.last_ramp_time
        if time_since_last_ramp = self.config.max_traffic_percent:
            print(f"\n✅ CANARY READY FOR PROMOTION (reached {self.config.max_traffic_percent}% traffic)")
    
    def _rollback(self):
        """Rollback to production version"""
        self.status = DeploymentStatus.ROLLING_BACK
        self.current_traffic_percent = 0.0
        print(f"\n⏪ Rolling back to production version...")
        self.status = DeploymentStatus.ROLLED_BACK
        print(f"✅ Rollback complete - 100% traffic to production")
    
    def promote_canary(self):
        """Promote canary to production"""
        if self.status == DeploymentStatus.ROLLED_BACK:
            print("❌ Cannot promote - deployment was rolled back")
            return False
        
        print(f"\n🎉 PROMOTING CANARY TO PRODUCTION")
        self.current_traffic_percent = 100.0
        self.status = DeploymentStatus.PROMOTED
        return True
    
    def get_status_report(self) -> str:
        """Generate status report"""
        m = self.metrics
        
        report = [
            "\n" + "="*70,
            "CANARY DEPLOYMENT STATUS",
            "="*70,
            f"\nStatus: {self.status.value.upper()}",
            f"Canary Traffic: {self.current_traffic_percent:.1f}%",
            f"Deployment Duration: {(datetime.now() - self.deployment_start_time).total_seconds():.0f}s",
            "\nProduction Metrics:",
            f"  Requests: {m.production_requests}",
            f"  Success Rate: {(m.production_successes/m.production_requests*100 if m.production_requests > 0 else 0):.1f}%",
            f"  Error Rate: {m.get_production_error_rate()*100:.1f}%",
            f"  Avg Latency: {m.get_production_avg_latency():.3f}s",
            "\nCanary Metrics:",
            f"  Requests: {m.canary_requests}",
            f"  Success Rate: {(m.canary_successes/m.canary_requests*100 if m.canary_requests > 0 else 0):.1f}%",
            f"  Error Rate: {m.get_canary_error_rate()*100:.1f}%",
            f"  Avg Latency: {m.get_canary_avg_latency():.3f}s",
        ]
        
        # Comparison
        if m.canary_requests >= self.config.min_requests_for_decision:
            error_diff = (m.get_canary_error_rate() - m.get_production_error_rate()) * 100
            latency_diff_pct = ((m.get_canary_avg_latency() - m.get_production_avg_latency()) 
                               / m.get_production_avg_latency() * 100) if m.get_production_avg_latency() > 0 else 0
            
            report.extend([
                "\nComparison (Canary vs Production):",
                f"  Error Rate Diff: {error_diff:+.1f}%",
                f"  Latency Diff: {latency_diff_pct:+.1f}%"
            ])
        
        report.append("="*70)
        return "\n".join(report)


def demonstrate_canary_deployment():
    """Demonstrate canary deployment pattern"""
    print("="*70)
    print("Pattern 154: Canary Deployment")
    print("="*70)
    
    # Configure canary deployment
    config = CanaryConfig(
        initial_traffic_percent=10.0,
        max_traffic_percent=50.0,
        ramp_step=10.0,
        ramp_interval_seconds=5.0,  # Short for demo
        max_error_rate_increase=0.10,
        max_latency_increase=0.30,
        min_requests_for_decision=5,
        auto_rollback=True
    )
    
    deployment = CanaryDeployment(config)
    
    print(f"\n🚀 Starting canary deployment")
    print(f"   Initial traffic: {config.initial_traffic_percent}%")
    print(f"   Ramp step: {config.ramp_step}%")
    print(f"   Target: {config.max_traffic_percent}%")
    
    # Simulate requests
    test_queries = [
        "What is Python?",
        "Explain machine learning",
        "How does AI work?",
        "What is cloud computing?",
        "Tell me about databases",
    ] * 10  # 50 total requests
    
    print(f"\n{'='*70}")
    print("SIMULATING PRODUCTION TRAFFIC")
    print(f"{'='*70}")
    
    results = []
    for i, query in enumerate(test_queries, 1):
        result = deployment.route_request(query)
        results.append(result)
        
        if i % 10 == 0:
            print(f"\n📊 After {i} requests:")
            print(deployment.get_status_report())
            
            # Small delay to allow ramping
            time.sleep(1)
    
    # Final status
    print(f"\n{'='*70}")
    print("FINAL STATUS")
    print(f"{'='*70}")
    print(deployment.get_status_report())
    
    # Show routing distribution
    production_count = sum(1 for r in results if r.get('version') == 'production')
    canary_count = sum(1 for r in results if r.get('version') == 'canary')
    
    print(f"\n📊 Traffic Distribution:")
    print(f"   Production: {production_count} requests ({production_count/len(results)*100:.1f}%)")
    print(f"   Canary: {canary_count} requests ({canary_count/len(results)*100:.1f}%)")
    
    # Promotion decision
    if deployment.status == DeploymentStatus.HEALTHY:
        print(f"\n✅ Canary is healthy and ready for promotion!")
        deployment.promote_canary()
    elif deployment.status == DeploymentStatus.ROLLED_BACK:
        print(f"\n❌ Canary was rolled back due to issues")
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Canary Deployment Pattern provides:

1. Gradual Rollout:
   - Start with small traffic percentage
   - Incrementally increase over time
   - Monitor at each step
   - Automatic promotion when ready

2. Safety Features:
   - Continuous health monitoring
   - Automatic rollback on issues
   - Configurable thresholds
   - Minimal blast radius

3. Metrics Tracking:
   - Error rates comparison
   - Latency comparison
   - Success rates
   - Traffic distribution

4. Deployment Control:
   - Traffic percentage control
   - Ramp-up automation
   - Manual promotion option
   - Emergency rollback

5. Production Safety:
   - Real user validation
   - Low-risk testing
   - Quick issue detection
   - Automatic recovery

This pattern is essential for safely deploying changes to
production AI systems with minimal risk and automatic safeguards.
    """)


if __name__ == "__main__":
    demonstrate_canary_deployment()

    

Agentic Design Pattern: Canary Deployment

Pattern 154: Canary Deployment – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 154: Canary Deployment

Description: Implements gradual rollout of new agent versions by routing a small percentage of traffic to the canary version, monitoring metrics, and automatically rolling back if issues are detected.

Components: – Traffic Splitter: Routes requests between versions – Health Monitor: Tracks canary metrics – Rollback Controller: Automatic rollback on issues – Metrics Collector: Performance and error tracking – Gradual Ramp: Increases traffic percentage over time

Use Cases: – Safe production deployments – Gradual feature rollout – Risk mitigation for changes – A/B testing with safety – Automated deployment validation

Benefits: – Minimal blast radius – Automatic rollback – Real user validation – Gradual confidence building – Production safety

Trade-offs: – Deployment complexity – Monitoring overhead – Slower full rollout – Dual version maintenance – Metrics interpretation

LangChain Implementation: Uses traffic routing with health monitoring and automatic rollback.

---

File Statistics

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

---

Module Documentation

"""
Pattern 154: Canary Deployment

Description:
    Implements gradual rollout of new agent versions by routing a small
    percentage of traffic to the canary version, monitoring metrics, and
    automatically rolling back if issues are detected.

Components:
    - Traffic Splitter: Routes requests between versions
    - Health Monitor: Tracks canary metrics
    - Rollback Controller: Automatic rollback on issues
    - Metrics Collector: Performance and error tracking
    - Gradual Ramp: Increases traffic percentage over time

Use Cases:
    - Safe production deployments
    - Gradual feature rollout
    - Risk mitigation for changes
    - A/B testing with safety
    - Automated deployment validation

Benefits:
    - Minimal blast radius
    - Automatic rollback
    - Real user validation
    - Gradual confidence building
    - Production safety

Trade-offs:
    - Deployment complexity
    - Monitoring overhead
    - Slower full rollout
    - Dual version maintenance
    - Metrics interpretation

LangChain Implementation:
    Uses traffic routing with health monitoring and automatic rollback.
"""

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

---

Import Statements

Standard Library Imports

import collections.defaultdict
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 random
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – 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 – random: Random number generation – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: DeploymentStatus

Lines 58-65

Inherits from: Enum

Purpose:

Status of canary deployment

---

Class: CanaryMetrics

Lines 69-107

Purpose:

Metrics for canary and production versions

Methods:

get_production_error_rate(self)

Lines 85-89

Purpose:

Calculate production error rate

Parameters:

Returns: float

get_canary_error_rate(self)

Lines 91-95

Purpose:

Calculate canary error rate

Parameters:

Returns: float

get_production_avg_latency(self)

Lines 97-101

Purpose:

Calculate average production latency

Parameters:

Returns: float

get_canary_avg_latency(self)

Lines 103-107

Purpose:

Calculate average canary latency

Parameters:

Returns: float

---

Class: CanaryConfig

Lines 111-125

Purpose:

Configuration for canary deployment

---

Class: CanaryDeployment

Lines 128-373

Purpose:

Manages canary deployment with gradual traffic routing and
automatic rollback on issues.

Methods:

__init__(self, config)

Lines 134-161

Purpose:

Initialize canary deployment

Parameters: – config: Config

route_request(self, query)

Lines 163-243

Purpose:

Route request to either production or canary based on traffic split

Args:
    query: User query

Returns:
    Response with version and metrics

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D60EAD0>

_check_health(self)

Lines 245-289

Purpose:

Check canary health and trigger rollback if needed

Parameters:

_consider_ramp_up(self)

Lines 291-316

Purpose:

Consider increasing traffic to canary

Parameters:

_rollback(self)

Lines 318-324

Purpose:

Rollback to production version

Parameters:

promote_canary(self)

Lines 326-335

Purpose:

Promote canary to production

Parameters:

get_status_report(self)

Lines 337-373

Purpose:

Generate status report

Parameters:

Returns: str

---

Functions

demonstrate_canary_deployment()

Lines 376-486

Purpose:

Demonstrate canary deployment pattern

---

Key Concepts Summary

Pattern Overview

The Canary Deployment 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 153
• Pattern 155

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 Canary Deployment 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: 154_canary_deployment.py
Total Lines: 491
Pattern Category: AI Agent Design Pattern #154

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 153: Shadow Mode Testing

Description:
    Runs new agent versions alongside production without affecting users.
    Captures both production and shadow responses, compares results, and
    identifies differences for validation before full deployment.

Components:
    - Shadow Executor: Runs shadow version in parallel
    - Response Comparator: Compares production vs shadow outputs
    - Difference Analyzer: Identifies and categorizes differences
    - Metrics Collector: Tracks shadow performance
    - Rollback Monitor: Detects degradation signals

Use Cases:
    - Testing new model versions safely
    - Validating prompt changes
    - A/B testing without user impact
    - Gradual feature rollout
    - Performance regression detection

Benefits:
    - Zero user impact during testing
    - Real production traffic testing
    - Early problem detection
    - Confidence before full deployment
    - Comprehensive comparison data

Trade-offs:
    - Doubled computational cost
    - Infrastructure complexity
    - Storage for dual responses
    - Latency considerations
    - Analysis overhead

LangChain Implementation:
    Uses parallel execution with ChatOpenAI, response comparison,
    and comprehensive metrics tracking.
"""

import os
import time
import asyncio
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from dotenv import load_dotenv

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

load_dotenv()


class DifferenceType(Enum):
    """Types of differences between responses"""
    IDENTICAL = "identical"
    MINOR = "minor"  # Small wording differences
    MODERATE = "moderate"  # Different approach, similar meaning
    MAJOR = "major"  # Significantly different
    ERROR = "error"  # One succeeded, one failed


class HealthStatus(Enum):
    """Health status of shadow version"""
    HEALTHY = "healthy"
    WARNING = "warning"
    DEGRADED = "degraded"
    FAILING = "failing"


@dataclass
class ShadowResponse:
    """Response from shadow execution"""
    production_response: str
    shadow_response: str
    production_time: float
    shadow_time: float
    production_error: Optional[str] = None
    shadow_error: Optional[str] = None
    difference_type: Optional[DifferenceType] = None
    similarity_score: float = 0.0
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ShadowMetrics:
    """Metrics for shadow testing"""
    total_requests: int = 0
    identical_responses: int = 0
    minor_differences: int = 0
    moderate_differences: int = 0
    major_differences: int = 0
    production_errors: int = 0
    shadow_errors: int = 0
    avg_production_latency: float = 0.0
    avg_shadow_latency: float = 0.0
    shadow_slower_by: float = 0.0  # Percentage
    
    def get_difference_rate(self, diff_type: DifferenceType) -> float:
        """Get rate of specific difference type"""
        if self.total_requests == 0:
            return 0.0
        
        count_map = {
            DifferenceType.IDENTICAL: self.identical_responses,
            DifferenceType.MINOR: self.minor_differences,
            DifferenceType.MODERATE: self.moderate_differences,
            DifferenceType.MAJOR: self.major_differences
        }
        
        return count_map.get(diff_type, 0) / self.total_requests
    
    def get_health_status(self) -> HealthStatus:
        """Determine overall health status"""
        if self.total_requests  0.3 or major_rate > 0.5:
            return HealthStatus.FAILING
        elif error_rate > 0.1 or major_rate > 0.3:
            return HealthStatus.DEGRADED
        elif error_rate > 0.05 or major_rate > 0.15:
            return HealthStatus.WARNING
        else:
            return HealthStatus.HEALTHY


class ShadowModeTester:
    """
    Agent that runs shadow testing by executing production and shadow
    versions in parallel, comparing results without affecting users.
    """
    
    def __init__(self, production_model: str = "gpt-3.5-turbo",
                 shadow_model: str = "gpt-3.5-turbo",
                 production_temp: float = 0.7,
                 shadow_temp: float = 0.7):
        """Initialize shadow mode tester"""
        # Production agent
        self.production_llm = ChatOpenAI(
            model=production_model,
            temperature=production_temp
        )
        
        # Shadow agent (different version or configuration)
        self.shadow_llm = ChatOpenAI(
            model=shadow_model,
            temperature=shadow_temp
        )
        
        # Metrics tracking
        self.metrics = ShadowMetrics()
        self.responses: List[ShadowResponse] = []
        
        # Difference categorization LLM
        self.analyzer_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.0)
        
        # Prompts
        self.production_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant. Provide accurate, clear responses."),
            ("user", "{query}")
        ])
        
        self.shadow_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant. Provide accurate, clear responses."),
            ("user", "{query}")
        ])
        
        self.comparison_prompt = ChatPromptTemplate.from_messages([
            ("system", """Compare two AI responses and categorize the difference.
            
Categories:
- identical: Same or nearly identical responses
- minor: Small wording differences, same meaning
- moderate: Different approach but similar quality
- major: Significantly different responses or quality

Response 1 (Production):
{response1}

Response 2 (Shadow):
{response2}

Return only the category name."""),
            ("user", "Categorize the difference between these responses.")
        ])
    
    def execute_shadow_request(self, query: str) -> ShadowResponse:
        """
        Execute query on both production and shadow versions
        
        Args:
            query: User query to execute
            
        Returns:
            ShadowResponse with both results and comparison
        """
        # Execute in parallel using threads
        with ThreadPoolExecutor(max_workers=2) as executor:
            # Submit both requests
            prod_future = executor.submit(self._execute_production, query)
            shadow_future = executor.submit(self._execute_shadow, query)
            
            # Wait for both to complete
            prod_result, prod_time, prod_error = prod_future.result()
            shadow_result, shadow_time, shadow_error = shadow_future.result()
        
        # Compare responses
        if prod_error is None and shadow_error is None:
            diff_type, similarity = self._compare_responses(prod_result, shadow_result)
        else:
            diff_type = DifferenceType.ERROR
            similarity = 0.0
        
        # Create response object
        response = ShadowResponse(
            production_response=prod_result,
            shadow_response=shadow_result,
            production_time=prod_time,
            shadow_time=shadow_time,
            production_error=prod_error,
            shadow_error=shadow_error,
            difference_type=diff_type,
            similarity_score=similarity
        )
        
        # Update metrics
        self._update_metrics(response)
        
        # Store response
        self.responses.append(response)
        
        # Return production result to user (shadow is hidden)
        return response
    
    def _execute_production(self, query: str) -> Tuple[str, float, Optional[str]]:
        """Execute on production system"""
        start_time = time.time()
        
        try:
            chain = self.production_prompt | self.production_llm | StrOutputParser()
            result = chain.invoke({"query": query})
            execution_time = time.time() - start_time
            return result, execution_time, None
            
        except Exception as e:
            execution_time = time.time() - start_time
            return "", execution_time, str(e)
    
    def _execute_shadow(self, query: str) -> Tuple[str, float, Optional[str]]:
        """Execute on shadow system"""
        start_time = time.time()
        
        try:
            chain = self.shadow_prompt | self.shadow_llm | StrOutputParser()
            result = chain.invoke({"query": query})
            execution_time = time.time() - start_time
            return result, execution_time, None
            
        except Exception as e:
            execution_time = time.time() - start_time
            return "", execution_time, str(e)
    
    def _compare_responses(self, prod_response: str, 
                          shadow_response: str) -> Tuple[DifferenceType, float]:
        """Compare two responses and categorize difference"""
        # Quick check for identical
        if prod_response.strip() == shadow_response.strip():
            return DifferenceType.IDENTICAL, 1.0
        
        # Calculate basic similarity
        similarity = self._calculate_similarity(prod_response, shadow_response)
        
        # Use LLM for categorization
        try:
            chain = self.comparison_prompt | self.analyzer_llm | StrOutputParser()
            category_str = chain.invoke({
                "response1": prod_response[:500],  # Truncate for efficiency
                "response2": shadow_response[:500]
            }).strip().lower()
            
            # Map to enum
            category_map = {
                "identical": DifferenceType.IDENTICAL,
                "minor": DifferenceType.MINOR,
                "moderate": DifferenceType.MODERATE,
                "major": DifferenceType.MAJOR
            }
            
            diff_type = category_map.get(category_str, DifferenceType.MODERATE)
            
        except Exception as e:
            # Fallback to similarity-based categorization
            if similarity > 0.9:
                diff_type = DifferenceType.MINOR
            elif similarity > 0.7:
                diff_type = DifferenceType.MODERATE
            else:
                diff_type = DifferenceType.MAJOR
        
        return diff_type, similarity
    
    def _calculate_similarity(self, text1: str, text2: str) -> float:
        """Calculate simple word-overlap similarity"""
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        
        if not words1 or not words2:
            return 0.0
        
        intersection = words1.intersection(words2)
        union = words1.union(words2)
        
        return len(intersection) / len(union) if union else 0.0
    
    def _update_metrics(self, response: ShadowResponse):
        """Update metrics based on response"""
        self.metrics.total_requests += 1
        
        # Count difference types
        if response.difference_type == DifferenceType.IDENTICAL:
            self.metrics.identical_responses += 1
        elif response.difference_type == DifferenceType.MINOR:
            self.metrics.minor_differences += 1
        elif response.difference_type == DifferenceType.MODERATE:
            self.metrics.moderate_differences += 1
        elif response.difference_type == DifferenceType.MAJOR:
            self.metrics.major_differences += 1
        
        # Count errors
        if response.production_error:
            self.metrics.production_errors += 1
        if response.shadow_error:
            self.metrics.shadow_errors += 1
        
        # Update latency metrics (running average)
        n = self.metrics.total_requests
        self.metrics.avg_production_latency = (
            (self.metrics.avg_production_latency * (n - 1) + response.production_time) / n
        )
        self.metrics.avg_shadow_latency = (
            (self.metrics.avg_shadow_latency * (n - 1) + response.shadow_time) / n
        )
        
        # Calculate relative performance
        if self.metrics.avg_production_latency > 0:
            self.metrics.shadow_slower_by = (
                (self.metrics.avg_shadow_latency - self.metrics.avg_production_latency) 
                / self.metrics.avg_production_latency * 100
            )
    
    def batch_test(self, queries: List[str], 
                   show_progress: bool = True) -> List[ShadowResponse]:
        """
        Test multiple queries in shadow mode
        
        Args:
            queries: List of queries to test
            show_progress: Whether to show progress
            
        Returns:
            List of shadow responses
        """
        print(f"\n🔄 Starting shadow mode testing for {len(queries)} queries...")
        
        results = []
        for i, query in enumerate(queries, 1):
            if show_progress:
                print(f"   Processing {i}/{len(queries)}: {query[:50]}...")
            
            response = self.execute_shadow_request(query)
            results.append(response)
            
            # Show inline status
            if show_progress:
                status = "✓" if response.difference_type in [DifferenceType.IDENTICAL, DifferenceType.MINOR] else "⚠"
                print(f"   {status} Difference: {response.difference_type.value}, Similarity: {response.similarity_score:.2f}")
        
        return results
    
    def get_health_status(self) -> HealthStatus:
        """Get current health status of shadow version"""
        return self.metrics.get_health_status()
    
    def get_divergent_examples(self, min_difference: DifferenceType = DifferenceType.MODERATE,
                               limit: int = 10) -> List[ShadowResponse]:
        """Get examples where responses diverged significantly"""
        divergent = [
            r for r in self.responses
            if r.difference_type in [DifferenceType.MODERATE, DifferenceType.MAJOR]
        ]
        
        # Sort by similarity (lowest first = most divergent)
        divergent.sort(key=lambda r: r.similarity_score)
        
        return divergent[:limit]
    
    def generate_report(self) -> str:
        """Generate comprehensive shadow testing report"""
        metrics = self.metrics
        health = self.get_health_status()
        
        report = [
            "\n" + "="*70,
            "SHADOW MODE TESTING REPORT",
            "="*70,
            f"\nTotal Requests: {metrics.total_requests}",
            f"Health Status: {health.value.upper()} {'🟢' if health == HealthStatus.HEALTHY else '🟡' if health == HealthStatus.WARNING else '🟠' if health == HealthStatus.DEGRADED else '🔴'}",
            "\nResponse Comparison:",
            f"  Identical: {metrics.identical_responses} ({metrics.get_difference_rate(DifferenceType.IDENTICAL)*100:.1f}%)",
            f"  Minor Differences: {metrics.minor_differences} ({metrics.get_difference_rate(DifferenceType.MINOR)*100:.1f}%)",
            f"  Moderate Differences: {metrics.moderate_differences} ({metrics.get_difference_rate(DifferenceType.MODERATE)*100:.1f}%)",
            f"  Major Differences: {metrics.major_differences} ({metrics.get_difference_rate(DifferenceType.MAJOR)*100:.1f}%)",
            "\nError Rates:",
            f"  Production Errors: {metrics.production_errors} ({metrics.production_errors/metrics.total_requests*100:.1f}%)",
            f"  Shadow Errors: {metrics.shadow_errors} ({metrics.shadow_errors/metrics.total_requests*100:.1f}%)",
            "\nPerformance:",
            f"  Avg Production Latency: {metrics.avg_production_latency:.3f}s",
            f"  Avg Shadow Latency: {metrics.avg_shadow_latency:.3f}s",
            f"  Shadow Performance: {metrics.shadow_slower_by:+.1f}% {'(slower)' if metrics.shadow_slower_by > 0 else '(faster)'}",
        ]
        
        # Add recommendation
        report.append("\nRecommendation:")
        if health == HealthStatus.HEALTHY:
            report.append("  ✅ Shadow version looks good for promotion to production")
        elif health == HealthStatus.WARNING:
            report.append("  ⚠️  Monitor shadow version closely before promotion")
        elif health == HealthStatus.DEGRADED:
            report.append("  ⚠️  Shadow version shows degradation - investigate before promotion")
        else:
            report.append("  ❌ Shadow version failing - DO NOT promote to production")
        
        report.append("="*70)
        return "\n".join(report)
    
    def show_divergent_examples(self, limit: int = 5):
        """Show examples where responses diverged"""
        divergent = self.get_divergent_examples(limit=limit)
        
        if not divergent:
            print("\n✅ No significant divergent examples found")
            return
        
        print(f"\n{'='*70}")
        print(f"DIVERGENT EXAMPLES (Top {len(divergent)})")
        print(f"{'='*70}")
        
        for i, response in enumerate(divergent, 1):
            print(f"\nExample {i}:")
            print(f"Difference Type: {response.difference_type.value}")
            print(f"Similarity: {response.similarity_score:.2f}")
            print(f"\nProduction Response:")
            print(f"  {response.production_response[:200]}...")
            print(f"\nShadow Response:")
            print(f"  {response.shadow_response[:200]}...")
            print("-" * 70)


def demonstrate_shadow_mode():
    """Demonstrate shadow mode testing"""
    print("="*70)
    print("Pattern 153: Shadow Mode Testing")
    print("="*70)
    
    # Create tester with different configurations
    # In real scenario, these would be different model versions or prompts
    tester = ShadowModeTester(
        production_model="gpt-3.5-turbo",
        shadow_model="gpt-3.5-turbo",
        production_temp=0.7,
        shadow_temp=0.9  # Different temperature for shadow
    )
    
    # Test queries
    test_queries = [
        "What is Python?",
        "Explain machine learning in simple terms",
        "How do I learn programming?",
        "What's the difference between AI and ML?",
        "Write a haiku about coding",
        "What are best practices for API design?",
        "Explain recursion with an example",
        "How does garbage collection work?",
        "What is the difference between REST and GraphQL?",
        "Explain async/await in JavaScript"
    ]
    
    print("\n" + "="*70)
    print("BATCH SHADOW TESTING")
    print("="*70)
    print(f"Production: GPT-3.5-Turbo (temp=0.7)")
    print(f"Shadow: GPT-3.5-Turbo (temp=0.9)")
    
    # Run batch test
    results = tester.batch_test(test_queries, show_progress=True)
    
    # Generate report
    print(tester.generate_report())
    
    # Show divergent examples
    tester.show_divergent_examples(limit=3)
    
    # Test individual request
    print("\n" + "="*70)
    print("INDIVIDUAL REQUEST EXAMPLE")
    print("="*70)
    
    query = "Explain quantum computing"
    print(f"\nQuery: {query}")
    
    response = tester.execute_shadow_request(query)
    
    print(f"\n📊 Results:")
    print(f"  Difference Type: {response.difference_type.value}")
    print(f"  Similarity Score: {response.similarity_score:.2f}")
    print(f"  Production Time: {response.production_time:.3f}s")
    print(f"  Shadow Time: {response.shadow_time:.3f}s")
    
    print(f"\n🏭 Production Response:")
    print(f"  {response.production_response[:150]}...")
    
    print(f"\n🔬 Shadow Response:")
    print(f"  {response.shadow_response[:150]}...")
    
    # Health check
    health = tester.get_health_status()
    print(f"\n🏥 Shadow Health Status: {health.value.upper()}")
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Shadow Mode Testing Pattern provides:

1. Zero-Impact Testing:
   - Tests run parallel to production
   - Users only see production responses
   - Shadow results collected for analysis
   - No service disruption

2. Comprehensive Comparison:
   - Response similarity analysis
   - Difference categorization
   - Performance comparison
   - Error rate tracking

3. Health Monitoring:
   - Automatic health status assessment
   - Divergence detection
   - Performance regression identification
   - Error rate monitoring

4. Deployment Confidence:
   - Real production traffic testing
   - Early problem detection
   - Data-driven promotion decisions
   - Risk mitigation

5. Use Cases:
   - New model version validation
   - Prompt engineering testing
   - Configuration changes
   - Performance optimization
   - Feature rollout validation

This pattern is essential for safely testing changes to production
AI systems before full deployment.
    """)


if __name__ == "__main__":
    demonstrate_shadow_mode()

    

Agentic Design Pattern: Shadow Mode

Pattern 153: Shadow Mode – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 153: Shadow Mode Testing

Description: Runs new agent versions alongside production without affecting users. Captures both production and shadow responses, compares results, and identifies differences for validation before full deployment.

Components: – Shadow Executor: Runs shadow version in parallel – Response Comparator: Compares production vs shadow outputs – Difference Analyzer: Identifies and categorizes differences – Metrics Collector: Tracks shadow performance – Rollback Monitor: Detects degradation signals

Use Cases: – Testing new model versions safely – Validating prompt changes – A/B testing without user impact – Gradual feature rollout – Performance regression detection

Benefits: – Zero user impact during testing – Real production traffic testing – Early problem detection – Confidence before full deployment – Comprehensive comparison data

Trade-offs: – Doubled computational cost – Infrastructure complexity – Storage for dual responses – Latency considerations – Analysis overhead

LangChain Implementation: Uses parallel execution with ChatOpenAI, response comparison, and comprehensive metrics tracking.

---

File Statistics

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

---

Module Documentation

"""
Pattern 153: Shadow Mode Testing

Description:
    Runs new agent versions alongside production without affecting users.
    Captures both production and shadow responses, compares results, and
    identifies differences for validation before full deployment.

Components:
    - Shadow Executor: Runs shadow version in parallel
    - Response Comparator: Compares production vs shadow outputs
    - Difference Analyzer: Identifies and categorizes differences
    - Metrics Collector: Tracks shadow performance
    - Rollback Monitor: Detects degradation signals

Use Cases:
    - Testing new model versions safely
    - Validating prompt changes
    - A/B testing without user impact
    - Gradual feature rollout
    - Performance regression detection

Benefits:
    - Zero user impact during testing
    - Real production traffic testing
    - Early problem detection
    - Confidence before full deployment
    - Comprehensive comparison data

Trade-offs:
    - Doubled computational cost
    - Infrastructure complexity
    - Storage for dual responses
    - Latency considerations
    - Analysis overhead

LangChain Implementation:
    Uses parallel execution with ChatOpenAI, response comparison,
    and comprehensive metrics tracking.
"""

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

---

Import Statements

Standard Library Imports

import asyncio
import collections.defaultdict
import concurrent.futures.ThreadPoolExecutor
import concurrent.futures.as_completed
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 time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – asyncio: Asynchronous I/O operations – collections.defaultdict: Specialized container datatypes – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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 – time: Time-related functions – 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: DifferenceType

Lines 60-66

Inherits from: Enum

Purpose:

Types of differences between responses

---

Class: HealthStatus

Lines 69-74

Inherits from: Enum

Purpose:

Health status of shadow version

---

Class: ShadowResponse

Lines 78-88

Purpose:

Response from shadow execution

---

Class: ShadowMetrics

Lines 92-134

Purpose:

Metrics for shadow testing

Methods:

get_difference_rate(self, diff_type)

Lines 105-117

Purpose:

Get rate of specific difference type

Parameters: – diff_type: Diff type

Returns: float

get_health_status(self)

Lines 119-134

Purpose:

Determine overall health status

Parameters:

Returns: HealthStatus

---

Class: ShadowModeTester

Lines 137-466

Purpose:

Agent that runs shadow testing by executing production and shadow
versions in parallel, comparing results without affecting users.

Methods:

__init__(self, production_model, shadow_model, production_temp, shadow_temp)

Lines 143-195

Purpose:

Initialize shadow mode tester

Parameters: – production_model: Language model instance – shadow_model: Language model instance – production_temp: Production temp – shadow_temp: Shadow temp

execute_shadow_request(self, query)

Lines 197-243

Purpose:

Execute query on both production and shadow versions

Args:
    query: User query to execute

Returns:
    ShadowResponse with both results and comparison

Parameters: – query: Input query or question

Returns: ShadowResponse

_execute_production(self, query)

Lines 245-257

Purpose:

Execute on production system

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D7C9910>

_execute_shadow(self, query)

Lines 259-271

Purpose:

Execute on shadow system

Parameters: – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D6380D0>

_compare_responses(self, prod_response, shadow_response)

Lines 273-310

Purpose:

Compare two responses and categorize difference

Parameters: – prod_response: Prod response – shadow_response: Shadow response

Returns: <ast.Subscript object at 0x000001A06D85A710>

_calculate_similarity(self, text1, text2)

Lines 312-323

Purpose:

Calculate simple word-overlap similarity

Parameters: – text1: Text1 – text2: Text2

Returns: float

_update_metrics(self, response)

Lines 325-359

Purpose:

Update metrics based on response

Parameters: – response: Response

batch_test(self, queries, show_progress)

Lines 361-388

Purpose:

Test multiple queries in shadow mode

Args:
    queries: List of queries to test
    show_progress: Whether to show progress

Returns:
    List of shadow responses

Parameters: – queries: Queries – show_progress: Show progress

Returns: <ast.Subscript object at 0x000001A06D17C050>

get_health_status(self)

Lines 390-392

Purpose:

Get current health status of shadow version

Parameters:

Returns: HealthStatus

get_divergent_examples(self, min_difference, limit)

Lines 394-405

Purpose:

Get examples where responses diverged significantly

Parameters: – min_difference: Min difference – limit: Limit

Returns: <ast.Subscript object at 0x000001A06D702490>

generate_report(self)

Lines 407-444

Purpose:

Generate comprehensive shadow testing report

Parameters:

Returns: str

show_divergent_examples(self, limit)

Lines 446-466

Purpose:

Show examples where responses diverged

Parameters: – limit: Limit

---

Functions

demonstrate_shadow_mode()

Lines 469-579

Purpose:

Demonstrate shadow mode testing

---

Key Concepts Summary

Pattern Overview

The Shadow Mode 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 152
• Pattern 154

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 Shadow Mode 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: 153_shadow_mode.py
Total Lines: 584
Pattern Category: AI Agent Design Pattern #153

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 152: Property-Based Testing

Description:
    Implements property-based testing for AI agents by defining invariants
    and properties that should hold across all inputs, then generating test
    cases to verify these properties.

Components:
    - Property Definitions: Invariants that should always hold
    - Test Case Generators: Create diverse inputs
    - Property Verifiers: Check if properties hold
    - Hypothesis Engine: Generate test hypotheses
    - Counterexample Finder: Identify violations

Use Cases:
    - Testing agent robustness across input space
    - Finding edge cases and failures
    - Verifying consistency properties
    - Ensuring output format compliance
    - Testing deterministic vs non-deterministic behaviors

Benefits:
    - Discovers unexpected edge cases
    - More thorough than example-based testing
    - Finds bugs traditional testing misses
    - Documents expected behaviors as properties
    - Scales testing coverage automatically

Trade-offs:
    - Requires careful property definition
    - Can be slower than unit tests
    - May generate false positives
    - Harder to debug failures
    - Non-determinism complicates verification

LangChain Implementation:
    Uses ChatOpenAI with property checking chains and systematic
    test case generation.
"""

import os
import re
import time
from typing import List, Dict, Any, Optional, Callable, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from collections import defaultdict
from dotenv import load_dotenv

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

load_dotenv()


class PropertyType(Enum):
    """Types of properties to test"""
    INVARIANT = "invariant"  # Must always hold
    POSTCONDITION = "postcondition"  # Output constraints
    CONSISTENCY = "consistency"  # Same input → same output
    MONOTONICITY = "monotonicity"  # Ordering preservation
    IDEMPOTENCE = "idempotence"  # f(f(x)) = f(x)
    COMMUTATIVE = "commutative"  # Order independence
    FORMAT = "format"  # Output format compliance


@dataclass
class Property:
    """Defines a testable property"""
    name: str
    description: str
    property_type: PropertyType
    checker: Callable[[Any, Any], bool]  # (input, output) -> bool
    examples: List[Tuple[Any, bool]] = field(default_factory=list)  # (input, expected_result)
    
    def check(self, input_val: Any, output_val: Any) -> bool:
        """Check if property holds"""
        return self.checker(input_val, output_val)


@dataclass
class TestResult:
    """Result of a property test"""
    property_name: str
    passed: bool
    test_case: Any
    output: Any
    error: Optional[str] = None
    execution_time: float = 0.0
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class PropertyTestReport:
    """Comprehensive test report"""
    property_name: str
    total_tests: int
    passed: int
    failed: int
    counterexamples: List[Dict[str, Any]]
    execution_time: float
    
    @property
    def pass_rate(self) -> float:
        return self.passed / self.total_tests if self.total_tests > 0 else 0.0


class PropertyBasedTester:
    """
    Agent that performs property-based testing on AI systems.
    Generates test cases and verifies properties hold.
    """
    
    def __init__(self, temperature: float = 0.7):
        """Initialize the property-based tester"""
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
        self.consistent_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.0)
        
        # Property registry
        self.properties: Dict[str, Property] = {}
        
        # Test case generator prompts
        self.test_gen_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a test case generator. Generate diverse test cases for testing AI systems.
Property to test: {property_description}
Constraints: {constraints}

Generate test cases that:
1. Cover normal cases
2. Include edge cases
3. Test boundary conditions
4. Challenge the property

Return as a list, one test case per line."""),
            ("user", "Generate {count} test cases.")
        ])
        
        self.agent_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant. Provide clear, accurate responses."),
            ("user", "{query}")
        ])
        
        # Register default properties
        self._register_default_properties()
    
    def _register_default_properties(self):
        """Register commonly used properties"""
        
        # 1. Non-empty response
        self.register_property(Property(
            name="non_empty_response",
            description="Agent must return non-empty response",
            property_type=PropertyType.POSTCONDITION,
            checker=lambda inp, out: out is not None and len(str(out).strip()) > 0
        ))
        
        # 2. Response length reasonable
        self.register_property(Property(
            name="reasonable_length",
            description="Response length should be reasonable (not too short or long)",
            property_type=PropertyType.POSTCONDITION,
            checker=lambda inp, out: 10 < len(str(out))  len(str(inp)) * 0.5
        ))
        
        # 7. No hallucinated facts (basic check)
        self.register_property(Property(
            name="no_obvious_hallucination",
            description="Response should not contain obviously false statements",
            property_type=PropertyType.INVARIANT,
            checker=lambda inp, out: not self._has_obvious_hallucination(str(out))
        ))
    
    def register_property(self, property_def: Property):
        """Register a new property to test"""
        self.properties[property_def.name] = property_def
        print(f"✅ Registered property: {property_def.name}")
    
    def test_property(self, property_name: str, test_count: int = 20,
                     custom_test_cases: Optional[List[Any]] = None) -> PropertyTestReport:
        """
        Test a specific property with generated or provided test cases
        
        Args:
            property_name: Name of property to test
            test_count: Number of test cases to generate
            custom_test_cases: Optional list of specific test cases
            
        Returns:
            Comprehensive test report
        """
        if property_name not in self.properties:
            raise ValueError(f"Property '{property_name}' not registered")
        
        property_def = self.properties[property_name]
        start_time = time.time()
        
        print(f"\n{'='*70}")
        print(f"Testing Property: {property_name}")
        print(f"Type: {property_def.property_type.value}")
        print(f"Description: {property_def.description}")
        print(f"{'='*70}")
        
        # Generate or use provided test cases
        if custom_test_cases:
            test_cases = custom_test_cases
        else:
            test_cases = self._generate_test_cases(property_def, test_count)
        
        print(f"\n🔄 Running {len(test_cases)} test cases...")
        
        results = []
        for i, test_case in enumerate(test_cases, 1):
            result = self._run_test_case(property_def, test_case, i)
            results.append(result)
        
        # Analyze results
        passed = sum(1 for r in results if r.passed)
        failed = len(results) - passed
        
        counterexamples = []
        for r in results:
            if not r.passed:
                counterexamples.append({
                    "input": r.test_case,
                    "output": r.output,
                    "error": r.error
                })
        
        execution_time = time.time() - start_time
        
        report = PropertyTestReport(
            property_name=property_name,
            total_tests=len(results),
            passed=passed,
            failed=failed,
            counterexamples=counterexamples,
            execution_time=execution_time
        )
        
        self._print_test_results(report)
        
        return report
    
    def test_all_properties(self, test_count: int = 10) -> Dict[str, PropertyTestReport]:
        """Test all registered properties"""
        print(f"\n{'='*70}")
        print("TESTING ALL PROPERTIES")
        print(f"{'='*70}")
        print(f"Properties to test: {len(self.properties)}")
        print(f"Test cases per property: {test_count}")
        
        reports = {}
        for property_name in self.properties:
            report = self.test_property(property_name, test_count)
            reports[property_name] = report
        
        # Overall summary
        self._print_overall_summary(reports)
        
        return reports
    
    def _generate_test_cases(self, property_def: Property, count: int) -> List[str]:
        """Generate test cases for a property"""
        print(f"   🔄 Generating test cases...")
        
        # Use examples as seed
        constraints = "Include diverse scenarios, edge cases, and boundary conditions"
        
        try:
            chain = self.test_gen_prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "property_description": property_def.description,
                "constraints": constraints,
                "count": count
            })
            
            # Parse test cases (one per line or numbered)
            test_cases = []
            lines = result.strip().split('\n')
            for line in lines:
                # Remove numbering/bullets
                cleaned = re.sub(r'^\d+[\.\)]\s*', '', line.strip())
                cleaned = re.sub(r'^[-•]\s*', '', cleaned)
                if cleaned and len(cleaned) > 3:
                    test_cases.append(cleaned)
            
            # Fill up to count if needed
            while len(test_cases)  TestResult:
        """Run a single test case"""
        start_time = time.time()
        
        try:
            # Execute agent with test case
            output = self._execute_agent(test_case)
            
            # Check property
            if property_def.property_type == PropertyType.CONSISTENCY:
                passed = self._check_consistency(test_case)
            else:
                passed = property_def.check(test_case, output)
            
            result = TestResult(
                property_name=property_def.name,
                passed=passed,
                test_case=test_case,
                output=output,
                execution_time=time.time() - start_time
            )
            
            status = "✓" if passed else "✗"
            print(f"   {status} Test {case_number}: {str(test_case)[:60]}...")
            
            return result
            
        except Exception as e:
            print(f"   ✗ Test {case_number}: ERROR - {str(e)}")
            return TestResult(
                property_name=property_def.name,
                passed=False,
                test_case=test_case,
                output=None,
                error=str(e),
                execution_time=time.time() - start_time
            )
    
    def _execute_agent(self, query: str) -> str:
        """Execute the agent being tested"""
        chain = self.agent_prompt | self.llm | StrOutputParser()
        return chain.invoke({"query": query})
    
    def _check_consistency(self, test_case: Any) -> bool:
        """Check consistency property by running multiple times"""
        outputs = []
        
        # Run 3 times with temperature 0 for consistency
        chain = self.agent_prompt | self.consistent_llm | StrOutputParser()
        for _ in range(3):
            output = chain.invoke({"query": test_case})
            outputs.append(output)
        
        # Check if all outputs are similar (basic check)
        # For production, use more sophisticated similarity metrics
        return len(set(outputs))  bool:
        """Check if text contains offensive content (basic)"""
        offensive_words = ["offensive", "inappropriate", "harmful"]  # Simplified
        return any(word in text.lower() for word in offensive_words)
    
    def _is_valid_json_if_requested(self, input_text: str, output_text: str) -> bool:
        """Check if output is valid JSON when requested"""
        if "json" not in input_text.lower():
            return True  # Not requesting JSON, so property holds
        
        try:
            import json
            json.loads(output_text)
            return True
        except:
            # Check if JSON is embedded in response
            if "{" in output_text and "}" in output_text:
                start = output_text.find("{")
                end = output_text.rfind("}") + 1
                try:
                    json.loads(output_text[start:end])
                    return True
                except:
                    return False
            return False
    
    def _has_obvious_hallucination(self, text: str) -> bool:
        """Check for obvious hallucinations (basic)"""
        # Check for contradictions or obviously false statements
        false_patterns = [
            r"the sun is cold",
            r"water is dry",
            r"earth is flat",
            r"humans have \d+ legs" # where number != 2
        ]
        
        text_lower = text.lower()
        for pattern in false_patterns:
            if re.search(pattern, text_lower):
                return True
        
        return False
    
    def _print_test_results(self, report: PropertyTestReport):
        """Print test results"""
        print(f"\n{'='*70}")
        print(f"RESULTS: {report.property_name}")
        print(f"{'='*70}")
        print(f"Total Tests: {report.total_tests}")
        print(f"Passed: {report.passed} ({report.pass_rate*100:.1f}%)")
        print(f"Failed: {report.failed}")
        print(f"Execution Time: {report.execution_time:.2f}s")
        
        if report.counterexamples:
            print(f"\n❌ Counterexamples Found ({len(report.counterexamples)}):")
            for i, ce in enumerate(report.counterexamples[:5], 1):  # Show first 5
                print(f"\n{i}. Input: {str(ce['input'])[:80]}...")
                if ce['output']:
                    print(f"   Output: {str(ce['output'])[:80]}...")
                if ce['error']:
                    print(f"   Error: {ce['error']}")
        
        print("="*70)
    
    def _print_overall_summary(self, reports: Dict[str, PropertyTestReport]):
        """Print overall summary of all tests"""
        print(f"\n{'='*70}")
        print("OVERALL TEST SUMMARY")
        print(f"{'='*70}")
        
        total_tests = sum(r.total_tests for r in reports.values())
        total_passed = sum(r.passed for r in reports.values())
        overall_pass_rate = total_passed / total_tests if total_tests > 0 else 0
        
        print(f"\nTotal Tests Run: {total_tests}")
        print(f"Overall Pass Rate: {overall_pass_rate*100:.1f}%")
        print(f"\nProperty Results:")
        
        for name, report in reports.items():
            status = "✓" if report.pass_rate >= 0.9 else "✗" if report.pass_rate  Any:
        """
        Attempt to find minimal counterexample (shrinking)
        
        Args:
            property_name: Property that failed
            counterexample: Original counterexample
            
        Returns:
            Minimal counterexample
        """
        property_def = self.properties[property_name]
        current = str(counterexample)
        
        print(f"\n🔍 Shrinking counterexample for {property_name}...")
        print(f"Original: {current[:100]}...")
        
        # Try progressively smaller versions
        for length in range(len(current) // 2, 10, -10):
            candidate = current[:length]
            
            try:
                output = self._execute_agent(candidate)
                if not property_def.check(candidate, output):
                    current = candidate
                    print(f"   Shrunk to: {len(current)} characters")
            except:
                continue
        
        print(f"✓ Minimal counterexample: {current}")
        return current


def demonstrate_property_based_testing():
    """Demonstrate property-based testing"""
    print("="*70)
    print("Pattern 152: Property-Based Testing")
    print("="*70)
    
    tester = PropertyBasedTester()
    
    # Test 1: Non-empty response property
    print("\n" + "="*70)
    print("TEST 1: Non-Empty Response Property")
    print("="*70)
    
    test_cases = [
        "What is Python?",
        "Hello",
        "",  # Edge case
        "Explain quantum physics",
        "?"  # Edge case
    ]
    
    report1 = tester.test_property("non_empty_response", 
                                   custom_test_cases=test_cases)
    
    # Test 2: Reasonable length property
    print("\n" + "="*70)
    print("TEST 2: Reasonable Length Property")
    print("="*70)
    
    report2 = tester.test_property("reasonable_length", test_count=8)
    
    # Test 3: JSON format compliance
    print("\n" + "="*70)
    print("TEST 3: JSON Format Compliance")
    print("="*70)
    
    json_test_cases = [
        "Return the result as JSON: {name: 'test'}",
        "Give me JSON with user info",
        "What is 2+2?",  # Should pass (not requesting JSON)
        "Provide a JSON object with three fields"
    ]
    
    report3 = tester.test_property("json_format", 
                                   custom_test_cases=json_test_cases)
    
    # Test 4: Custom property
    print("\n" + "="*70)
    print("TEST 4: Custom Property - No Numbers in Text Explanation")
    print("="*70)
    
    def no_numbers_checker(inp: Any, out: Any) -> bool:
        """Custom checker: explanations should avoid raw numbers"""
        if "explain" in str(inp).lower():
            # Allow some numbers but not excessive
            return str(out).count("1") + str(out).count("2") < 10
        return True
    
    custom_prop = Property(
        name="minimal_numbers_in_explanation",
        description="Explanations should minimize raw numbers",
        property_type=PropertyType.POSTCONDITION,
        checker=no_numbers_checker
    )
    
    tester.register_property(custom_prop)
    
    explain_cases = [
        "Explain how computers work",
        "What is machine learning?",
        "Tell me about AI"
    ]
    
    report4 = tester.test_property("minimal_numbers_in_explanation",
                                   custom_test_cases=explain_cases)
    
    # Test 5: Consistency property
    print("\n" + "="*70)
    print("TEST 5: Consistency Property")
    print("="*70)
    
    consistent_cases = [
        "What is 2+2?",
        "Name the capital of France"
    ]
    
    report5 = tester.test_property("consistency", 
                                   custom_test_cases=consistent_cases)
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print(f"""
Property-Based Testing Pattern provides:

1. Property Definitions:
   - Invariants (must always hold)
   - Postconditions (output constraints)
   - Consistency checks
   - Format compliance
   - Custom properties

2. Test Generation:
   - Automated test case creation
   - Edge case discovery
   - Boundary condition testing
   - Coverage maximization

3. Property Verification:
   - Systematic property checking
   - Counterexample identification
   - Shrinking to minimal cases
   - Detailed reporting

4. Benefits:
   - Finds bugs traditional testing misses
   - Documents expected behaviors
   - Scales testing automatically
   - More thorough coverage

5. Results Summary:
   - Non-empty responses: {report1.pass_rate*100:.0f}% pass
   - Reasonable length: {report2.pass_rate*100:.0f}% pass
   - JSON format: {report3.pass_rate*100:.0f}% pass
   - Custom property: {report4.pass_rate*100:.0f}% pass
   - Consistency: {report5.pass_rate*100:.0f}% pass

This pattern is essential for robust AI agent testing and
quality assurance in production systems.
    """)


if __name__ == "__main__":
    demonstrate_property_based_testing()

    

Agentic Design Pattern: Property Based Testing

Pattern 152: Property Based Testing – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Property Based Testing pattern implementation.

---

Pattern Description

Pattern 152: Property-Based Testing

Description: Implements property-based testing for AI agents by defining invariants and properties that should hold across all inputs, then generating test cases to verify these properties.

Components: – Property Definitions: Invariants that should always hold – Test Case Generators: Create diverse inputs – Property Verifiers: Check if properties hold – Hypothesis Engine: Generate test hypotheses – Counterexample Finder: Identify violations

Use Cases: – Testing agent robustness across input space – Finding edge cases and failures – Verifying consistency properties – Ensuring output format compliance – Testing deterministic vs non-deterministic behaviors

Benefits: – Discovers unexpected edge cases – More thorough than example-based testing – Finds bugs traditional testing misses – Documents expected behaviors as properties – Scales testing coverage automatically

Trade-offs: – Requires careful property definition – Can be slower than unit tests – May generate false positives – Harder to debug failures – Non-determinism complicates verification

LangChain Implementation: Uses ChatOpenAI with property checking chains and systematic test case generation.

---

File Statistics

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

---

Module Documentation

"""
Pattern 152: Property-Based Testing

Description:
    Implements property-based testing for AI agents by defining invariants
    and properties that should hold across all inputs, then generating test
    cases to verify these properties.

Components:
    - Property Definitions: Invariants that should always hold
    - Test Case Generators: Create diverse inputs
    - Property Verifiers: Check if properties hold
    - Hypothesis Engine: Generate test hypotheses
    - Counterexample Finder: Identify violations

Use Cases:
    - Testing agent robustness across input space
    - Finding edge cases and failures
    - Verifying consistency properties
    - Ensuring output format compliance
    - Testing deterministic vs non-deterministic behaviors

Benefits:
    - Discovers unexpected edge cases
    - More thorough than example-based testing
    - Finds bugs traditional testing misses
    - Documents expected behaviors as properties
    - Scales testing coverage automatically

Trade-offs:
    - Requires careful property definition
    - Can be slower than unit tests
    - May generate false positives
    - Harder to debug failures
    - Non-determinism complicates verification

LangChain Implementation:
    Uses ChatOpenAI with property checking chains and systematic
    test case generation.
"""

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

---

Import Statements

Standard Library Imports

import collections.defaultdict
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 re
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – 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 – re: Regular expression 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) – 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: PropertyType

Lines 59-67

Inherits from: Enum

Purpose:

Types of properties to test

---

Class: Property

Lines 71-81

Purpose:

Defines a testable property

Methods:

check(self, input_val, output_val)

Lines 79-81

Purpose:

Check if property holds

Parameters: – input_val: Input data – output_val: Output val

Returns: bool

---

Class: TestResult

Lines 85-93

Purpose:

Result of a property test

---

Class: PropertyTestReport

Lines 97-108

Purpose:

Comprehensive test report

Methods:

pass_rate(self)

Lines 107-108

Decorators: property

Purpose: Pass Rate operation.

Parameters:

Returns: float

---

Class: PropertyBasedTester

Lines 111-506

Purpose:

Agent that performs property-based testing on AI systems.
Generates test cases and verifies properties hold.

Methods:

__init__(self, temperature)

Lines 117-147

Purpose:

Initialize the property-based tester

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

_register_default_properties(self)

Lines 149-206

Purpose:

Register commonly used properties

Parameters:

register_property(self, property_def)

Lines 208-211

Purpose:

Register a new property to test

Parameters: – property_def: Property def

test_property(self, property_name, test_count, custom_test_cases)

Lines 213-277

Purpose:

Test a specific property with generated or provided test cases

Args:
    property_name: Name of property to test
    test_count: Number of test cases to generate
    custom_test_cases: Optional list of specific test cases

Returns:
    Comprehensive test report

Parameters: – property_name: Property name – test_count: Test count – custom_test_cases: Custom test cases

Returns: PropertyTestReport

test_all_properties(self, test_count)

Lines 279-295

Purpose:

Test all registered properties

Parameters: – test_count: Test count

Returns: <ast.Subscript object at 0x000001A06D617AD0>

_generate_test_cases(self, property_def, count)

Lines 297-331

Purpose:

Generate test cases for a property

Parameters: – property_def: Property def – count: Count

Returns: <ast.Subscript object at 0x000001A06D7CFC90>

_run_test_case(self, property_def, test_case, case_number)

Lines 333-370

Purpose:

Run a single test case

Parameters: – property_def: Property def – test_case: Test case – case_number: Case number

Returns: TestResult

_execute_agent(self, query)

Lines 372-375

Purpose:

Execute the agent being tested

Parameters: – query: Input query or question

Returns: str

_check_consistency(self, test_case)

Lines 377-389

Purpose:

Check consistency property by running multiple times

Parameters: – test_case: Test case

Returns: bool

_contains_offensive(self, text)

Lines 391-394

Purpose:

Check if text contains offensive content (basic)

Parameters: – text: Text

Returns: bool

_is_valid_json_if_requested(self, input_text, output_text)

Lines 396-415

Purpose:

Check if output is valid JSON when requested

Parameters: – input_text: Input data – output_text: Output text

Returns: bool

_has_obvious_hallucination(self, text)

Lines 417-432

Purpose:

Check for obvious hallucinations (basic)

Parameters: – text: Text

Returns: bool

_print_test_results(self, report)

Lines 434-453

Purpose:

Print test results

Parameters: – report: Report

_print_overall_summary(self, reports)

Lines 455-474

Purpose:

Print overall summary of all tests

Parameters: – reports: Reports

shrink_counterexample(self, property_name, counterexample)

Lines 476-506

Purpose:

Attempt to find minimal counterexample (shrinking)

Args:
    property_name: Property that failed
    counterexample: Original counterexample

Returns:
    Minimal counterexample

Parameters: – property_name: Property name – counterexample: Counterexample

Returns: Any

---

Functions

demonstrate_property_based_testing()

Lines 509-639

Purpose:

Demonstrate property-based testing

---

Key Concepts Summary

Pattern Overview

The Property Based Testing 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 151
• Pattern 153

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 Property Based Testing 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: 152_property_based_testing.py
Total Lines: 644
Pattern Category: AI Agent Design Pattern #152

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 151: Synthetic Data Generation

Description:
    Generates synthetic training and test data for agent systems using LLMs.
    Creates diverse, high-quality examples with schema validation, quality
    metrics, and coverage analysis.

Components:
    - Data Schema Definitions: Structured format specifications
    - Generation Strategies: Various approaches for data creation
    - Quality Validators: Check data quality and validity
    - Coverage Analyzers: Ensure diverse data distribution
    - Augmentation Engine: Enhance existing datasets

Use Cases:
    - Training data creation for fine-tuning
    - Test case generation for QA
    - Data augmentation for small datasets
    - Edge case generation for robustness testing
    - Privacy-preserving data synthesis

Benefits:
    - Reduces manual data labeling effort
    - Generates diverse test scenarios
    - Creates privacy-safe datasets
    - Enables systematic coverage
    - Scales data creation efficiently

Trade-offs:
    - Generated data may lack real-world complexity
    - Quality depends on LLM capabilities
    - May introduce biases from LLM
    - Validation overhead
    - Cost of generation at scale

LangChain Implementation:
    Uses ChatOpenAI for generation with structured output parsing,
    Pydantic models for validation, and custom metrics.
"""

import os
import json
import time
from typing import List, Dict, Any, Optional, Set, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from collections import defaultdict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel, Field, validator

load_dotenv()


class DataType(Enum):
    """Types of synthetic data"""
    QA_PAIR = "qa_pair"
    CLASSIFICATION = "classification"
    CONVERSATION = "conversation"
    INSTRUCTION = "instruction"
    SENTIMENT = "sentiment"
    ENTITY_EXTRACTION = "entity_extraction"
    SUMMARIZATION = "summarization"


class QualityMetric(Enum):
    """Quality metrics for generated data"""
    COHERENCE = "coherence"
    RELEVANCE = "relevance"
    DIVERSITY = "diversity"
    COMPLEXITY = "complexity"
    VALIDITY = "validity"


# Pydantic Models for Structured Generation

class QAPair(BaseModel):
    """Question-Answer pair"""
    question: str = Field(description="The question")
    answer: str = Field(description="The answer")
    context: Optional[str] = Field(default=None, description="Optional context")
    difficulty: str = Field(default="medium", description="easy, medium, or hard")
    category: Optional[str] = Field(default=None, description="Question category")


class ClassificationExample(BaseModel):
    """Text classification example"""
    text: str = Field(description="The text to classify")
    label: str = Field(description="The classification label")
    confidence: float = Field(default=1.0, description="Confidence score 0-1")
    explanation: Optional[str] = Field(default=None, description="Explanation for label")


class ConversationTurn(BaseModel):
    """Single conversation turn"""
    speaker: str = Field(description="Speaker identifier")
    utterance: str = Field(description="What was said")
    intent: Optional[str] = Field(default=None, description="Speaker intent")


class Conversation(BaseModel):
    """Multi-turn conversation"""
    turns: List[ConversationTurn] = Field(description="Conversation turns")
    topic: Optional[str] = Field(default=None, description="Conversation topic")
    outcome: Optional[str] = Field(default=None, description="Conversation outcome")


@dataclass
class GenerationConfig:
    """Configuration for data generation"""
    data_type: DataType
    count: int = 10
    domain: Optional[str] = None
    difficulty: str = "mixed"  # easy, medium, hard, mixed
    diversity_target: float = 0.7  # Target diversity score
    quality_threshold: float = 0.6  # Minimum quality score
    include_edge_cases: bool = True
    seed_examples: List[Dict[str, Any]] = field(default_factory=list)


@dataclass
class QualityScores:
    """Quality scores for generated data"""
    coherence: float = 0.0
    relevance: float = 0.0
    diversity: float = 0.0
    complexity: float = 0.0
    validity: float = 0.0
    
    def overall_score(self) -> float:
        """Calculate overall quality score"""
        return (self.coherence + self.relevance + self.diversity + 
                self.complexity + self.validity) / 5
    
    def to_dict(self) -> Dict[str, float]:
        """Convert to dictionary"""
        return {
            "coherence": self.coherence,
            "relevance": self.relevance,
            "diversity": self.diversity,
            "complexity": self.complexity,
            "validity": self.validity,
            "overall": self.overall_score()
        }


class SyntheticDataGenerator:
    """
    Agent that generates synthetic training and test data.
    Produces high-quality, diverse datasets with validation.
    """
    
    def __init__(self, temperature: float = 0.9):
        """Initialize the synthetic data generator"""
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
        self.validator_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
        
        # Track generated data for diversity
        self.generated_texts: Set[str] = set()
        self.category_counts: Dict[str, int] = defaultdict(int)
        
        # Generation prompts
        self.qa_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a data generation expert. Generate high-quality question-answer pairs.
Domain: {domain}
Difficulty: {difficulty}
Requirements: {requirements}

Generate diverse, realistic questions with accurate answers.
Return as JSON with fields: question, answer, context, difficulty, category"""),
            ("user", "Generate {count} question-answer pairs.")
        ])
        
        self.classification_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a data generation expert. Generate text classification examples.
Domain: {domain}
Labels: {labels}
Requirements: {requirements}

Generate realistic text examples with correct labels.
Return as JSON with fields: text, label, confidence, explanation"""),
            ("user", "Generate {count} classification examples.")
        ])
        
        self.conversation_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a conversation generation expert. Create realistic multi-turn conversations.
Domain: {domain}
Number of turns: {num_turns}
Requirements: {requirements}

Generate natural, coherent conversations.
Return as JSON with structure: turns (list of speaker/utterance), topic, outcome"""),
            ("user", "Generate {count} conversations.")
        ])
        
        self.quality_eval_prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate the quality of generated data on multiple dimensions.
Rate each dimension from 0.0 to 1.0:
- Coherence: Logical consistency and clarity
- Relevance: Relevance to the domain/task
- Diversity: Uniqueness and variety
- Complexity: Appropriate complexity level
- Validity: Correctness and accuracy

Return JSON with scores for each dimension."""),
            ("user", "Evaluate this data:\n{data}")
        ])
    
    def generate_dataset(self, config: GenerationConfig) -> Dict[str, Any]:
        """
        Generate synthetic dataset according to configuration
        
        Args:
            config: Generation configuration
            
        Returns:
            Dictionary with generated data and metrics
        """
        start_time = time.time()
        
        print(f"\n🔄 Generating {config.count} {config.data_type.value} examples...")
        if config.domain:
            print(f"   Domain: {config.domain}")
        print(f"   Difficulty: {config.difficulty}")
        
        # Generate data based on type
        if config.data_type == DataType.QA_PAIR:
            examples = self._generate_qa_pairs(config)
        elif config.data_type == DataType.CLASSIFICATION:
            examples = self._generate_classifications(config)
        elif config.data_type == DataType.CONVERSATION:
            examples = self._generate_conversations(config)
        else:
            examples = self._generate_generic(config)
        
        # Validate and score quality
        quality_scores = self._evaluate_quality(examples, config)
        
        # Analyze diversity
        diversity_metrics = self._analyze_diversity(examples)
        
        # Filter by quality threshold
        filtered_examples = self._filter_by_quality(examples, quality_scores, 
                                                    config.quality_threshold)
        
        generation_time = time.time() - start_time
        
        return {
            "examples": filtered_examples,
            "count": len(filtered_examples),
            "quality_scores": quality_scores,
            "diversity_metrics": diversity_metrics,
            "generation_time": generation_time,
            "config": {
                "data_type": config.data_type.value,
                "target_count": config.count,
                "domain": config.domain,
                "difficulty": config.difficulty
            }
        }
    
    def _generate_qa_pairs(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate question-answer pairs"""
        examples = []
        
        # Generate in batches for diversity
        batch_size = min(5, config.count)
        num_batches = (config.count + batch_size - 1) // batch_size
        
        for batch in range(num_batches):
            batch_count = min(batch_size, config.count - len(examples))
            
            requirements = self._get_generation_requirements(config, examples)
            
            try:
                chain = self.qa_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "domain": config.domain or "general knowledge",
                    "difficulty": config.difficulty,
                    "requirements": requirements,
                    "count": batch_count
                })
                
                # Parse JSON response
                parsed = self._parse_json_response(result)
                if isinstance(parsed, list):
                    examples.extend(parsed)
                elif isinstance(parsed, dict):
                    examples.append(parsed)
                    
            except Exception as e:
                print(f"   ⚠️  Batch {batch + 1} generation error: {str(e)}")
        
        return examples[:config.count]
    
    def _generate_classifications(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate classification examples"""
        examples = []
        
        # Default labels if not specified
        labels = ["positive", "negative", "neutral"]
        if config.domain:
            labels = self._infer_labels_from_domain(config.domain)
        
        batch_size = min(5, config.count)
        num_batches = (config.count + batch_size - 1) // batch_size
        
        for batch in range(num_batches):
            batch_count = min(batch_size, config.count - len(examples))
            
            requirements = self._get_generation_requirements(config, examples)
            
            try:
                chain = self.classification_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "domain": config.domain or "general text classification",
                    "labels": ", ".join(labels),
                    "requirements": requirements,
                    "count": batch_count
                })
                
                parsed = self._parse_json_response(result)
                if isinstance(parsed, list):
                    examples.extend(parsed)
                elif isinstance(parsed, dict):
                    examples.append(parsed)
                    
            except Exception as e:
                print(f"   ⚠️  Batch {batch + 1} generation error: {str(e)}")
        
        return examples[:config.count]
    
    def _generate_conversations(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate multi-turn conversations"""
        examples = []
        
        num_turns = 5  # Default conversation length
        
        for i in range(config.count):
            requirements = self._get_generation_requirements(config, examples)
            
            try:
                chain = self.conversation_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "domain": config.domain or "general conversation",
                    "num_turns": num_turns,
                    "requirements": requirements,
                    "count": 1
                })
                
                parsed = self._parse_json_response(result)
                if isinstance(parsed, list) and parsed:
                    examples.append(parsed[0])
                elif isinstance(parsed, dict):
                    examples.append(parsed)
                    
            except Exception as e:
                print(f"   ⚠️  Example {i + 1} generation error: {str(e)}")
        
        return examples
    
    def _generate_generic(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate generic examples"""
        generic_prompt = ChatPromptTemplate.from_messages([
            ("system", "Generate {count} diverse examples of {data_type} in {domain} domain. Return as JSON."),
            ("user", "Generate the examples.")
        ])
        
        chain = generic_prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "count": config.count,
            "data_type": config.data_type.value,
            "domain": config.domain or "general"
        })
        
        parsed = self._parse_json_response(result)
        return parsed if isinstance(parsed, list) else [parsed]
    
    def _get_generation_requirements(self, config: GenerationConfig, 
                                   existing: List[Dict[str, Any]]) -> str:
        """Generate requirements for next batch"""
        requirements = []
        
        if config.diversity_target > 0.5:
            requirements.append("Ensure high diversity from previous examples")
        
        if config.include_edge_cases and len(existing) > 5:
            requirements.append("Include edge cases and unusual scenarios")
        
        # Category balance
        if self.category_counts:
            underrep_categories = [cat for cat, count in self.category_counts.items() 
                                  if count  List[str]:
        """Infer appropriate labels from domain"""
        domain_lower = domain.lower()
        
        if "sentiment" in domain_lower:
            return ["positive", "negative", "neutral"]
        elif "emotion" in domain_lower:
            return ["joy", "sadness", "anger", "fear", "surprise"]
        elif "intent" in domain_lower:
            return ["question", "command", "statement", "request"]
        elif "urgency" in domain_lower:
            return ["urgent", "normal", "low_priority"]
        else:
            return ["category_a", "category_b", "category_c"]
    
    def _parse_json_response(self, response: str) -> Any:
        """Parse JSON from LLM response"""
        try:
            # Try direct JSON parse
            return json.loads(response)
        except json.JSONDecodeError:
            # Extract JSON from markdown code blocks
            if "```json" in response:
                start = response.find("```json") + 7
                end = response.find("```", start)
                json_str = response[start:end].strip()
                return json.loads(json_str)
            elif "```" in response:
                start = response.find("```") + 3
                end = response.find("```", start)
                json_str = response[start:end].strip()
                return json.loads(json_str)
            else:
                # Try to find JSON array or object
                for char in ['{', '[']:
                    if char in response:
                        start = response.find(char)
                        return json.loads(response[start:])
                
                return []
    
    def _evaluate_quality(self, examples: List[Dict[str, Any]], 
                         config: GenerationConfig) -> List[QualityScores]:
        """Evaluate quality of generated examples"""
        quality_scores = []
        
        # Sample for evaluation (evaluate all if small dataset)
        sample_size = min(len(examples), 10)
        sample_indices = list(range(0, len(examples), max(1, len(examples) // sample_size)))
        
        for idx in sample_indices[:sample_size]:
            example = examples[idx]
            
            try:
                chain = self.quality_eval_prompt | self.validator_llm | StrOutputParser()
                result = chain.invoke({"data": json.dumps(example, indent=2)})
                
                scores_dict = self._parse_json_response(result)
                
                scores = QualityScores(
                    coherence=scores_dict.get("coherence", 0.7),
                    relevance=scores_dict.get("relevance", 0.7),
                    diversity=scores_dict.get("diversity", 0.7),
                    complexity=scores_dict.get("complexity", 0.7),
                    validity=scores_dict.get("validity", 0.7)
                )
                quality_scores.append(scores)
                
            except Exception as e:
                # Default scores on error
                quality_scores.append(QualityScores(
                    coherence=0.7, relevance=0.7, diversity=0.7,
                    complexity=0.7, validity=0.7
                ))
        
        # Extend scores to all examples (reuse evaluated scores)
        while len(quality_scores)  Dict[str, Any]:
        """Analyze diversity of generated examples"""
        if not examples:
            return {"diversity_score": 0.0}
        
        # Extract texts for similarity analysis
        texts = []
        for ex in examples:
            if "question" in ex:
                texts.append(ex["question"])
            elif "text" in ex:
                texts.append(ex["text"])
            elif "turns" in ex:
                texts.append(" ".join(t.get("utterance", "") for t in ex["turns"]))
        
        # Calculate pairwise uniqueness
        unique_ratio = len(set(texts)) / len(texts) if texts else 0
        
        # Category distribution
        categories = [ex.get("category", "unknown") for ex in examples]
        category_dist = {cat: categories.count(cat) / len(categories) 
                        for cat in set(categories)}
        
        # Length variance
        lengths = [len(t) for t in texts]
        avg_length = sum(lengths) / len(lengths) if lengths else 0
        length_variance = sum((l - avg_length) ** 2 for l in lengths) / len(lengths) if lengths else 0
        
        return {
            "diversity_score": unique_ratio,
            "unique_examples": len(set(texts)),
            "total_examples": len(texts),
            "category_distribution": category_dist,
            "avg_length": avg_length,
            "length_variance": length_variance
        }
    
    def _filter_by_quality(self, examples: List[Dict[str, Any]], 
                          quality_scores: List[QualityScores],
                          threshold: float) -> List[Dict[str, Any]]:
        """Filter examples by quality threshold"""
        filtered = []
        
        for example, scores in zip(examples, quality_scores):
            if scores.overall_score() >= threshold:
                filtered.append(example)
        
        if len(filtered)  List[Dict[str, Any]]:
        """
        Augment existing dataset with variations
        
        Args:
            existing_data: Existing examples to augment
            augmentation_factor: How many variations per example
            
        Returns:
            Augmented dataset
        """
        print(f"\n🔄 Augmenting {len(existing_data)} examples (factor: {augmentation_factor})...")
        
        augmented = existing_data.copy()
        
        augment_prompt = ChatPromptTemplate.from_messages([
            ("system", """Create {count} variations of the following example while maintaining its core meaning and structure.
Make the variations diverse but semantically similar.
Return as JSON array."""),
            ("user", "Original example:\n{example}")
        ])
        
        for example in existing_data:
            try:
                chain = augment_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "count": augmentation_factor,
                    "example": json.dumps(example, indent=2)
                })
                
                variations = self._parse_json_response(result)
                if isinstance(variations, list):
                    augmented.extend(variations[:augmentation_factor])
                    
            except Exception as e:
                print(f"   ⚠️  Augmentation error: {str(e)}")
        
        print(f"   ✅ Augmented to {len(augmented)} examples")
        return augmented
    
    def generate_report(self, results: Dict[str, Any]) -> str:
        """Generate comprehensive generation report"""
        report = [
            "\n" + "="*70,
            "SYNTHETIC DATA GENERATION REPORT",
            "="*70,
            f"\nConfiguration:",
            f"  Data Type: {results['config']['data_type']}",
            f"  Target Count: {results['config']['target_count']}",
            f"  Domain: {results['config']['domain'] or 'General'}",
            f"  Difficulty: {results['config']['difficulty']}",
            f"\nGeneration Results:",
            f"  Generated: {results['count']} examples",
            f"  Generation Time: {results['generation_time']:.2f}s",
            f"  Rate: {results['count']/results['generation_time']:.1f} examples/sec"
        ]
        
        # Quality metrics
        if results['quality_scores']:
            avg_scores = {
                "coherence": sum(s.coherence for s in results['quality_scores']) / len(results['quality_scores']),
                "relevance": sum(s.relevance for s in results['quality_scores']) / len(results['quality_scores']),
                "diversity": sum(s.diversity for s in results['quality_scores']) / len(results['quality_scores']),
                "complexity": sum(s.complexity for s in results['quality_scores']) / len(results['quality_scores']),
                "validity": sum(s.validity for s in results['quality_scores']) / len(results['quality_scores'])
            }
            
            report.extend([
                "\nQuality Scores (0.0-1.0):",
                f"  Coherence: {avg_scores['coherence']:.3f}",
                f"  Relevance: {avg_scores['relevance']:.3f}",
                f"  Diversity: {avg_scores['diversity']:.3f}",
                f"  Complexity: {avg_scores['complexity']:.3f}",
                f"  Validity: {avg_scores['validity']:.3f}",
                f"  Overall: {sum(avg_scores.values())/len(avg_scores):.3f}"
            ])
        
        # Diversity metrics
        div = results['diversity_metrics']
        report.extend([
            "\nDiversity Metrics:",
            f"  Diversity Score: {div['diversity_score']:.3f}",
            f"  Unique Examples: {div['unique_examples']}/{div['total_examples']}",
            f"  Average Length: {div['avg_length']:.0f} characters"
        ])
        
        if 'category_distribution' in div:
            report.append("  Category Distribution:")
            for cat, ratio in sorted(div['category_distribution'].items(), 
                                    key=lambda x: x[1], reverse=True):
                report.append(f"    - {cat}: {ratio*100:.1f}%")
        
        report.append("="*70)
        return "\n".join(report)


def demonstrate_synthetic_data_generation():
    """Demonstrate synthetic data generation"""
    print("="*70)
    print("Pattern 151: Synthetic Data Generation")
    print("="*70)
    
    generator = SyntheticDataGenerator()
    
    # Test 1: QA Pairs
    print("\n" + "="*70)
    print("TEST 1: Generate Question-Answer Pairs")
    print("="*70)
    
    qa_config = GenerationConfig(
        data_type=DataType.QA_PAIR,
        count=10,
        domain="Python programming",
        difficulty="mixed",
        quality_threshold=0.6
    )
    
    qa_results = generator.generate_dataset(qa_config)
    
    print(f"\n✅ Generated {qa_results['count']} QA pairs")
    print("\nSample Examples:")
    for i, example in enumerate(qa_results['examples'][:3], 1):
        print(f"\n{i}. Q: {example.get('question', 'N/A')}")
        print(f"   A: {example.get('answer', 'N/A')[:100]}...")
        if 'difficulty' in example:
            print(f"   Difficulty: {example['difficulty']}")
    
    print(generator.generate_report(qa_results))
    
    # Test 2: Classification Examples
    print("\n" + "="*70)
    print("TEST 2: Generate Classification Examples")
    print("="*70)
    
    class_config = GenerationConfig(
        data_type=DataType.CLASSIFICATION,
        count=8,
        domain="sentiment analysis",
        difficulty="medium",
        quality_threshold=0.6
    )
    
    class_results = generator.generate_dataset(class_config)
    
    print(f"\n✅ Generated {class_results['count']} classification examples")
    print("\nSample Examples:")
    for i, example in enumerate(class_results['examples'][:3], 1):
        print(f"\n{i}. Text: {example.get('text', 'N/A')[:80]}...")
        print(f"   Label: {example.get('label', 'N/A')}")
        if 'confidence' in example:
            print(f"   Confidence: {example['confidence']:.2f}")
    
    # Test 3: Conversations
    print("\n" + "="*70)
    print("TEST 3: Generate Conversations")
    print("="*70)
    
    conv_config = GenerationConfig(
        data_type=DataType.CONVERSATION,
        count=5,
        domain="customer support",
        quality_threshold=0.6
    )
    
    conv_results = generator.generate_dataset(conv_config)
    
    print(f"\n✅ Generated {conv_results['count']} conversations")
    if conv_results['examples']:
        print("\nSample Conversation:")
        example = conv_results['examples'][0]
        if 'turns' in example:
            for turn in example['turns'][:4]:
                speaker = turn.get('speaker', 'Unknown')
                utterance = turn.get('utterance', '')
                print(f"   {speaker}: {utterance}")
        if 'topic' in example:
            print(f"   Topic: {example['topic']}")
    
    # Test 4: Data Augmentation
    print("\n" + "="*70)
    print("TEST 4: Data Augmentation")
    print("="*70)
    
    seed_data = [
        {"question": "What is Python?", "answer": "A programming language", "category": "basics"},
        {"question": "How to define a function?", "answer": "Use the def keyword", "category": "syntax"}
    ]
    
    augmented = generator.augment_dataset(seed_data, augmentation_factor=2)
    print(f"\n✅ Augmented {len(seed_data)} examples to {len(augmented)} examples")
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Synthetic Data Generation Pattern provides:

1. Multiple Data Types:
   - Question-Answer pairs
   - Classification examples
   - Multi-turn conversations
   - Custom formats

2. Quality Control:
   - Automated quality evaluation
   - Coherence, relevance, validity checks
   - Configurable quality thresholds
   - Filtering low-quality examples

3. Diversity Assurance:
   - Uniqueness tracking
   - Category balance
   - Length variation
   - Edge case inclusion

4. Augmentation Capabilities:
   - Variation generation
   - Dataset expansion
   - Semantic preservation

5. Production Features:
   - Batch generation
   - Domain specialization
   - Difficulty levels
   - Comprehensive reporting

This pattern is essential for creating training data, test cases,
and expanding datasets with privacy-safe synthetic examples.
    """)


if __name__ == "__main__":
    demonstrate_synthetic_data_generation()

    

Agentic Design Pattern: Synthetic Data

Pattern 151: Synthetic Data – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 151: Synthetic Data Generation

Description: Generates synthetic training and test data for agent systems using LLMs. Creates diverse, high-quality examples with schema validation, quality metrics, and coverage analysis.

Components: – Data Schema Definitions: Structured format specifications – Generation Strategies: Various approaches for data creation – Quality Validators: Check data quality and validity – Coverage Analyzers: Ensure diverse data distribution – Augmentation Engine: Enhance existing datasets

Use Cases: – Training data creation for fine-tuning – Test case generation for QA – Data augmentation for small datasets – Edge case generation for robustness testing – Privacy-preserving data synthesis

Benefits: – Reduces manual data labeling effort – Generates diverse test scenarios – Creates privacy-safe datasets – Enables systematic coverage – Scales data creation efficiently

Trade-offs: – Generated data may lack real-world complexity – Quality depends on LLM capabilities – May introduce biases from LLM – Validation overhead – Cost of generation at scale

LangChain Implementation: Uses ChatOpenAI for generation with structured output parsing, Pydantic models for validation, and custom metrics.

---

File Statistics

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

---

Module Documentation

"""
Pattern 151: Synthetic Data Generation

Description:
    Generates synthetic training and test data for agent systems using LLMs.
    Creates diverse, high-quality examples with schema validation, quality
    metrics, and coverage analysis.

Components:
    - Data Schema Definitions: Structured format specifications
    - Generation Strategies: Various approaches for data creation
    - Quality Validators: Check data quality and validity
    - Coverage Analyzers: Ensure diverse data distribution
    - Augmentation Engine: Enhance existing datasets

Use Cases:
    - Training data creation for fine-tuning
    - Test case generation for QA
    - Data augmentation for small datasets
    - Edge case generation for robustness testing
    - Privacy-preserving data synthesis

Benefits:
    - Reduces manual data labeling effort
    - Generates diverse test scenarios
    - Creates privacy-safe datasets
    - Enables systematic coverage
    - Scales data creation efficiently

Trade-offs:
    - Generated data may lack real-world complexity
    - Quality depends on LLM capabilities
    - May introduce biases from LLM
    - Validation overhead
    - Cost of generation at scale

LangChain Implementation:
    Uses ChatOpenAI for generation with structured output parsing,
    Pydantic models for validation, and custom metrics.
"""

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

---

Import Statements

Standard Library Imports

import collections.defaultdict
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import json
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set
import typing.Tuple

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – 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.JsonOutputParser: Regular expression operations – 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.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
from pydantic.BaseModel
from pydantic.Field
from pydantic.validator

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: DataType

Lines 60-68

Inherits from: Enum

Purpose:

Types of synthetic data

---

Class: QualityMetric

Lines 71-77

Inherits from: Enum

Purpose:

Quality metrics for generated data

---

Class: QAPair

Lines 82-88

Inherits from: BaseModel

Purpose:

Question-Answer pair

---

Class: ClassificationExample

Lines 91-96

Inherits from: BaseModel

Purpose:

Text classification example

---

Class: ConversationTurn

Lines 99-103

Inherits from: BaseModel

Purpose:

Single conversation turn

---

Class: Conversation

Lines 106-110

Inherits from: BaseModel

Purpose:

Multi-turn conversation

---

Class: GenerationConfig

Lines 114-123

Purpose:

Configuration for data generation

---

Class: QualityScores

Lines 127-149

Purpose:

Quality scores for generated data

Methods:

overall_score(self)

Lines 135-138

Purpose:

Calculate overall quality score

Parameters:

Returns: float

to_dict(self)

Lines 140-149

Purpose:

Convert to dictionary

Parameters:

Returns: <ast.Subscript object at 0x000001A06D852450>

---

Class: SyntheticDataGenerator

Lines 152-631

Purpose:

Agent that generates synthetic training and test data.
Produces high-quality, diverse datasets with validation.

Methods:

__init__(self, temperature)

Lines 158-212

Purpose:

Initialize the synthetic data generator

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

generate_dataset(self, config)

Lines 214-265

Purpose:

Generate synthetic dataset according to configuration

Args:
    config: Generation configuration

Returns:
    Dictionary with generated data and metrics

Parameters: – config: Config

Returns: <ast.Subscript object at 0x000001A06E0A8910>

_generate_qa_pairs(self, config)

Lines 267-299

Purpose:

Generate question-answer pairs

Parameters: – config: Config

Returns: <ast.Subscript object at 0x000001A06D682490>

_generate_classifications(self, config)

Lines 301-336

Purpose:

Generate classification examples

Parameters: – config: Config

Returns: <ast.Subscript object at 0x000001A06D80D9D0>

_generate_conversations(self, config)

Lines 338-365

Purpose:

Generate multi-turn conversations

Parameters: – config: Config

Returns: <ast.Subscript object at 0x000001A06D1572D0>

_generate_generic(self, config)

Lines 367-382

Purpose:

Generate generic examples

Parameters: – config: Config

Returns: <ast.Subscript object at 0x000001A06D66A490>

_get_generation_requirements(self, config, existing)

Lines 384-402

Purpose:

Generate requirements for next batch

Parameters: – config: Config – existing: Existing

Returns: str

_infer_labels_from_domain(self, domain)

Lines 404-417

Purpose:

Infer appropriate labels from domain

Parameters: – domain: Domain

Returns: <ast.Subscript object at 0x000001A06D7A1890>

_parse_json_response(self, response)

Lines 419-443

Purpose:

Parse JSON from LLM response

Parameters: – response: Response

Returns: Any

_evaluate_quality(self, examples, config)

Lines 445-483

Purpose:

Evaluate quality of generated examples

Parameters: – examples: Examples – config: Config

Returns: <ast.Subscript object at 0x000001A06D63FB50>

_analyze_diversity(self, examples)

Lines 485-520

Purpose:

Analyze diversity of generated examples

Parameters: – examples: Examples

Returns: <ast.Subscript object at 0x000001A06D7C9590>

_filter_by_quality(self, examples, quality_scores, threshold)

Lines 522-535

Purpose:

Filter examples by quality threshold

Parameters: – examples: Examples – quality_scores: Quality scores – threshold: Threshold

Returns: <ast.Subscript object at 0x000001A06D620910>

augment_dataset(self, existing_data, augmentation_factor)

Lines 537-576

Purpose:

Augment existing dataset with variations

Args:
    existing_data: Existing examples to augment
    augmentation_factor: How many variations per example

Returns:
    Augmented dataset

Parameters: – existing_data: Existing data – augmentation_factor: Augmentation factor

Returns: <ast.Subscript object at 0x000001A06D16EBD0>

generate_report(self, results)

Lines 578-631

Purpose:

Generate comprehensive generation report

Parameters: – results: Results

Returns: str

---

Functions

demonstrate_synthetic_data_generation()

Lines 634-767

Purpose:

Demonstrate synthetic data generation

---

Key Concepts Summary

Pattern Overview

The Synthetic Data 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 150
• Pattern 152

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 Synthetic Data 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: 151_synthetic_data.py
Total Lines: 772
Pattern Category: AI Agent Design Pattern #151

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 150: Error Recovery Strategies

Description:
    Implements comprehensive error recovery mechanisms with multiple fallback strategies.
    Goes beyond simple retry to include alternative approaches, degraded service modes,
    and intelligent recovery path selection.

Components:
    - Recovery Strategy Registry: Maps error types to recovery strategies
    - Strategy Selector: Chooses appropriate recovery approach
    - Fallback Chain: Ordered list of alternative strategies
    - Recovery Context: Tracks recovery attempts and outcomes
    - Strategy Executor: Executes selected recovery strategy

Use Cases:
    - Production systems requiring high availability
    - Multi-step workflows with diverse failure modes
    - Systems with multiple service dependencies
    - Critical operations requiring guaranteed completion
    - Graceful degradation scenarios

Benefits:
    - Improved system resilience
    - Better user experience during failures
    - Reduced manual intervention
    - Comprehensive failure handling
    - Intelligent recovery path selection

Trade-offs:
    - Increased system complexity
    - More code paths to test
    - Potential for unexpected behaviors
    - Recovery strategy maintenance overhead
    - May delay error reporting

LangChain Implementation:
    Uses custom error handling with LangChain chains, fallback chains,
    and strategy pattern for recovery mechanisms.
"""

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

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

load_dotenv()


class RecoveryStrategyType(Enum):
    """Types of recovery strategies"""
    RETRY = "retry"
    FALLBACK_MODEL = "fallback_model"
    SIMPLIFIED_REQUEST = "simplified_request"
    CACHED_RESPONSE = "cached_response"
    DEFAULT_RESPONSE = "default_response"
    ALTERNATIVE_APPROACH = "alternative_approach"
    DEGRADED_SERVICE = "degraded_service"
    HUMAN_ESCALATION = "human_escalation"


class ErrorSeverity(Enum):
    """Error severity levels"""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class RecoveryContext:
    """Context for tracking recovery attempts"""
    original_request: str
    error_type: str
    error_message: str
    severity: ErrorSeverity
    attempts: int = 0
    max_attempts: int = 3
    attempted_strategies: List[RecoveryStrategyType] = field(default_factory=list)
    recovery_history: List[Dict[str, Any]] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)
    
    def record_attempt(self, strategy: RecoveryStrategyType, success: bool, 
                      result: Any = None, error: str = None):
        """Record a recovery attempt"""
        self.attempts += 1
        self.attempted_strategies.append(strategy)
        self.recovery_history.append({
            "strategy": strategy.value,
            "success": success,
            "result": result,
            "error": error,
            "timestamp": datetime.now().isoformat()
        })


@dataclass
class RecoveryStrategy:
    """Defines a recovery strategy"""
    name: RecoveryStrategyType
    priority: int  # Lower number = higher priority
    applicable_errors: List[str]  # Error patterns this strategy can handle
    max_retries: int = 3
    timeout: float = 30.0
    executor: Optional[Callable] = None


class ErrorRecoveryAgent:
    """
    Agent that implements comprehensive error recovery strategies.
    Automatically selects and executes appropriate recovery mechanisms.
    """
    
    def __init__(self, temperature: float = 0.7):
        """Initialize the error recovery agent"""
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
        self.fallback_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
        
        # Response cache for fallback
        self.response_cache: Dict[str, Any] = {}
        
        # Registry of recovery strategies
        self.strategies: List[RecoveryStrategy] = self._initialize_strategies()
        
        # Prompts
        self.main_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant. Provide accurate and comprehensive responses."),
            ("user", "{query}")
        ])
        
        self.simplified_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant. Provide a brief, simple response."),
            ("user", "Provide a short answer to: {query}")
        ])
        
        self.degraded_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an AI assistant operating in degraded mode. Provide basic, limited responses."),
            ("user", "In one sentence, respond to: {query}")
        ])
        
    def _initialize_strategies(self) -> List[RecoveryStrategy]:
        """Initialize recovery strategy registry"""
        return [
            RecoveryStrategy(
                name=RecoveryStrategyType.RETRY,
                priority=1,
                applicable_errors=["timeout", "rate_limit", "connection"],
                max_retries=3,
                executor=self._retry_strategy
            ),
            RecoveryStrategy(
                name=RecoveryStrategyType.FALLBACK_MODEL,
                priority=2,
                applicable_errors=["model_error", "api_error"],
                executor=self._fallback_model_strategy
            ),
            RecoveryStrategy(
                name=RecoveryStrategyType.SIMPLIFIED_REQUEST,
                priority=3,
                applicable_errors=["context_length", "complexity"],
                executor=self._simplified_request_strategy
            ),
            RecoveryStrategy(
                name=RecoveryStrategyType.CACHED_RESPONSE,
                priority=4,
                applicable_errors=["all"],
                executor=self._cached_response_strategy
            ),
            RecoveryStrategy(
                name=RecoveryStrategyType.ALTERNATIVE_APPROACH,
                priority=5,
                applicable_errors=["all"],
                executor=self._alternative_approach_strategy
            ),
            RecoveryStrategy(
                name=RecoveryStrategyType.DEGRADED_SERVICE,
                priority=6,
                applicable_errors=["all"],
                executor=self._degraded_service_strategy
            ),
            RecoveryStrategy(
                name=RecoveryStrategyType.DEFAULT_RESPONSE,
                priority=7,
                applicable_errors=["all"],
                executor=self._default_response_strategy
            ),
            RecoveryStrategy(
                name=RecoveryStrategyType.HUMAN_ESCALATION,
                priority=8,
                applicable_errors=["all"],
                executor=self._human_escalation_strategy
            )
        ]
    
    def execute_with_recovery(self, query: str, max_recovery_attempts: int = 5) -> Dict[str, Any]:
        """
        Execute query with comprehensive error recovery
        
        Args:
            query: Query to execute
            max_recovery_attempts: Maximum recovery attempts
            
        Returns:
            Dictionary with result and recovery information
        """
        start_time = time.time()
        
        # Try primary execution
        try:
            result = self._primary_execution(query)
            
            return {
                "success": True,
                "result": result,
                "recovery_used": False,
                "execution_time": time.time() - start_time
            }
            
        except Exception as e:
            # Primary execution failed, initiate recovery
            context = self._create_recovery_context(query, e)
            
            recovery_result = self._execute_recovery(context, max_recovery_attempts)
            
            return {
                "success": recovery_result["success"],
                "result": recovery_result.get("result"),
                "recovery_used": True,
                "recovery_attempts": context.attempts,
                "strategies_attempted": [s.value for s in context.attempted_strategies],
                "recovery_history": context.recovery_history,
                "execution_time": time.time() - start_time,
                "error": recovery_result.get("error")
            }
    
    def _primary_execution(self, query: str) -> str:
        """Execute primary request"""
        chain = self.main_prompt | self.llm | StrOutputParser()
        result = chain.invoke({"query": query})
        
        # Cache successful response
        self.response_cache[query] = result
        
        return result
    
    def _create_recovery_context(self, query: str, error: Exception) -> RecoveryContext:
        """Create recovery context from error"""
        error_type = type(error).__name__
        error_message = str(error)
        
        # Determine severity
        severity = self._determine_severity(error_type, error_message)
        
        return RecoveryContext(
            original_request=query,
            error_type=error_type,
            error_message=error_message,
            severity=severity
        )
    
    def _determine_severity(self, error_type: str, error_message: str) -> ErrorSeverity:
        """Determine error severity"""
        critical_patterns = ["critical", "fatal", "security"]
        high_patterns = ["failed", "error", "exception"]
        medium_patterns = ["timeout", "rate_limit"]
        
        error_lower = f"{error_type} {error_message}".lower()
        
        if any(p in error_lower for p in critical_patterns):
            return ErrorSeverity.CRITICAL
        elif any(p in error_lower for p in high_patterns):
            return ErrorSeverity.HIGH
        elif any(p in error_lower for p in medium_patterns):
            return ErrorSeverity.MEDIUM
        else:
            return ErrorSeverity.LOW
    
    def _execute_recovery(self, context: RecoveryContext, 
                         max_attempts: int) -> Dict[str, Any]:
        """Execute recovery strategies"""
        
        # Select applicable strategies
        strategies = self._select_strategies(context)
        
        for strategy in strategies:
            if context.attempts >= max_attempts:
                break
            
            if strategy.name in context.attempted_strategies:
                continue
            
            print(f"\n🔄 Attempting recovery with strategy: {strategy.name.value}")
            
            try:
                result = strategy.executor(context)
                
                if result is not None:
                    context.record_attempt(strategy.name, True, result)
                    return {
                        "success": True,
                        "result": result,
                        "recovery_strategy": strategy.name.value
                    }
                else:
                    context.record_attempt(strategy.name, False, 
                                         error="Strategy returned None")
                    
            except Exception as e:
                context.record_attempt(strategy.name, False, error=str(e))
                print(f"   ❌ Strategy failed: {str(e)}")
        
        # All strategies failed
        return {
            "success": False,
            "error": f"All recovery strategies exhausted. Original error: {context.error_message}"
        }
    
    def _select_strategies(self, context: RecoveryContext) -> List[RecoveryStrategy]:
        """Select applicable recovery strategies"""
        applicable = []
        
        for strategy in self.strategies:
            # Check if strategy applies to this error
            if "all" in strategy.applicable_errors:
                applicable.append(strategy)
            elif any(pattern in context.error_type.lower() or 
                    pattern in context.error_message.lower() 
                    for pattern in strategy.applicable_errors):
                applicable.append(strategy)
        
        # Sort by priority
        return sorted(applicable, key=lambda s: s.priority)
    
    # Recovery Strategy Implementations
    
    def _retry_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Simple retry with exponential backoff"""
        max_retries = 3
        base_delay = 1.0
        
        for attempt in range(max_retries):
            try:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
                
                print(f"   🔄 Retry attempt {attempt + 1}/{max_retries}")
                return self._primary_execution(context.original_request)
                
            except Exception as e:
                if attempt == max_retries - 1:
                    print(f"   ❌ All retries failed")
                    return None
        
        return None
    
    def _fallback_model_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Use fallback model with different settings"""
        try:
            print(f"   🔄 Using fallback model")
            chain = self.main_prompt | self.fallback_llm | StrOutputParser()
            return chain.invoke({"query": context.original_request})
        except Exception as e:
            print(f"   ❌ Fallback model failed: {str(e)}")
            return None
    
    def _simplified_request_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Simplify request to reduce complexity"""
        try:
            print(f"   🔄 Using simplified request")
            chain = self.simplified_prompt | self.llm | StrOutputParser()
            return chain.invoke({"query": context.original_request})
        except Exception as e:
            print(f"   ❌ Simplified request failed: {str(e)}")
            return None
    
    def _cached_response_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Return cached response if available"""
        if context.original_request in self.response_cache:
            print(f"   ✅ Using cached response")
            return self.response_cache[context.original_request]
        
        # Look for similar cached queries
        for cached_query, cached_response in self.response_cache.items():
            if self._queries_similar(context.original_request, cached_query):
                print(f"   ✅ Using similar cached response")
                return f"[Based on similar query] {cached_response}"
        
        return None
    
    def _alternative_approach_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Try alternative approach using LLM suggestion"""
        try:
            print(f"   🔄 Generating alternative approach")
            
            alt_prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful assistant. The user asked: '{original_query}' but we encountered an error. Provide a helpful response that addresses their underlying need."),
                ("user", "Provide an alternative response to: {query}")
            ])
            
            chain = alt_prompt | self.llm | StrOutputParser()
            return chain.invoke({
                "query": context.original_request,
                "original_query": context.original_request
            })
        except Exception as e:
            print(f"   ❌ Alternative approach failed: {str(e)}")
            return None
    
    def _degraded_service_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Provide degraded service with limited functionality"""
        try:
            print(f"   🔄 Using degraded service mode")
            chain = self.degraded_prompt | self.fallback_llm | StrOutputParser()
            result = chain.invoke({"query": context.original_request})
            return f"[Degraded Mode] {result}"
        except Exception as e:
            print(f"   ❌ Degraded service failed: {str(e)}")
            return None
    
    def _default_response_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Return safe default response"""
        print(f"   ✅ Using default response")
        return (f"I apologize, but I'm currently unable to fully process your request "
                f"due to technical difficulties. Your request was: '{context.original_request}'. "
                f"Please try again later or rephrase your question.")
    
    def _human_escalation_strategy(self, context: RecoveryContext) -> Optional[str]:
        """Escalate to human operator"""
        print(f"   🚨 Escalating to human operator")
        return (f"[ESCALATED TO HUMAN OPERATOR]\n"
                f"Request: {context.original_request}\n"
                f"Error: {context.error_type} - {context.error_message}\n"
                f"Severity: {context.severity.value}\n"
                f"Attempted strategies: {[s.value for s in context.attempted_strategies]}\n"
                f"A human operator will review this request.")
    
    def _queries_similar(self, query1: str, query2: str, threshold: float = 0.7) -> bool:
        """Check if two queries are similar (simple implementation)"""
        # Simple word overlap similarity
        words1 = set(query1.lower().split())
        words2 = set(query2.lower().split())
        
        if not words1 or not words2:
            return False
        
        overlap = len(words1.intersection(words2))
        similarity = overlap / max(len(words1), len(words2))
        
        return similarity >= threshold
    
    def generate_recovery_report(self, results: List[Dict[str, Any]]) -> str:
        """Generate comprehensive recovery report"""
        total = len(results)
        successful = sum(1 for r in results if r["success"])
        with_recovery = sum(1 for r in results if r.get("recovery_used", False))
        
        report = [
            "\n" + "="*70,
            "ERROR RECOVERY REPORT",
            "="*70,
            f"\nTotal Executions: {total}",
            f"Successful: {successful} ({successful/total*100:.1f}%)",
            f"Required Recovery: {with_recovery} ({with_recovery/total*100:.1f}%)",
            f"Failed: {total - successful} ({(total-successful)/total*100:.1f}%)"
        ]
        
        # Strategy usage statistics
        strategy_usage = {}
        for result in results:
            if result.get("strategies_attempted"):
                for strategy in result["strategies_attempted"]:
                    strategy_usage[strategy] = strategy_usage.get(strategy, 0) + 1
        
        if strategy_usage:
            report.append("\nRecovery Strategy Usage:")
            for strategy, count in sorted(strategy_usage.items(), 
                                         key=lambda x: x[1], reverse=True):
                report.append(f"  - {strategy}: {count} times")
        
        # Performance metrics
        avg_time = sum(r["execution_time"] for r in results) / total
        recovery_times = [r["execution_time"] for r in results if r.get("recovery_used")]
        avg_recovery_time = sum(recovery_times) / len(recovery_times) if recovery_times else 0
        
        report.extend([
            f"\nPerformance Metrics:",
            f"  - Average execution time: {avg_time:.2f}s",
            f"  - Average recovery time: {avg_recovery_time:.2f}s"
        ])
        
        report.append("="*70)
        return "\n".join(report)


def demonstrate_error_recovery():
    """Demonstrate comprehensive error recovery strategies"""
    print("="*70)
    print("Pattern 150: Error Recovery Strategies")
    print("="*70)
    
    agent = ErrorRecoveryAgent()
    results = []
    
    # Test cases
    test_cases = [
        {
            "query": "What is the capital of France?",
            "description": "Normal query (should succeed)"
        },
        {
            "query": "Explain quantum computing in detail",
            "description": "Complex query (may need simplified approach)"
        },
        {
            "query": "What is 2 + 2?",
            "description": "Simple query (should use cache on retry)"
        },
        {
            "query": "Translate this ancient Sumerian text: ...",
            "description": "Difficult query (may need alternative approach)"
        },
        {
            "query": "Generate a 50-page research paper on...",
            "description": "Oversized request (should use degraded mode)"
        }
    ]
    
    print("\n" + "="*70)
    print("TESTING ERROR RECOVERY MECHANISMS")
    print("="*70)
    
    for i, test_case in enumerate(test_cases, 1):
        print(f"\n{'='*70}")
        print(f"Test Case {i}: {test_case['description']}")
        print(f"Query: {test_case['query'][:60]}...")
        print(f"{'='*70}")
        
        result = agent.execute_with_recovery(test_case["query"])
        results.append(result)
        
        print(f"\n✅ Success: {result['success']}")
        if result['recovery_used']:
            print(f"🔄 Recovery Used: Yes")
            print(f"   Attempts: {result['recovery_attempts']}")
            print(f"   Strategies: {', '.join(result['strategies_attempted'])}")
        else:
            print(f"🔄 Recovery Used: No (primary execution succeeded)")
        
        if result['success']:
            response = result['result'][:100] + "..." if len(result['result']) > 100 else result['result']
            print(f"📝 Response: {response}")
        else:
            print(f"❌ Error: {result.get('error', 'Unknown error')}")
        
        print(f"⏱️  Execution time: {result['execution_time']:.2f}s")
    
    # Generate comprehensive report
    print("\n" + agent.generate_recovery_report(results))
    
    # Demonstrate recovery history
    print("\n" + "="*70)
    print("RECOVERY HISTORY EXAMPLE")
    print("="*70)
    
    complex_result = [r for r in results if r.get('recovery_used', False)]
    if complex_result:
        example = complex_result[0]
        if 'recovery_history' in example:
            print("\nDetailed Recovery History:")
            for i, attempt in enumerate(example['recovery_history'], 1):
                print(f"\nAttempt {i}:")
                print(f"  Strategy: {attempt['strategy']}")
                print(f"  Success: {attempt['success']}")
                if attempt['success']:
                    result_preview = str(attempt['result'])[:60]
                    print(f"  Result: {result_preview}...")
                else:
                    print(f"  Error: {attempt.get('error', 'Unknown')}")
                print(f"  Timestamp: {attempt['timestamp']}")
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Error Recovery Strategies Pattern provides:

1. Multiple Recovery Mechanisms:
   - Retry with backoff
   - Fallback models
   - Simplified requests
   - Cached responses
   - Alternative approaches
   - Degraded service modes
   - Default responses
   - Human escalation

2. Intelligent Strategy Selection:
   - Error type analysis
   - Severity assessment
   - Priority-based ordering
   - Context-aware selection

3. Comprehensive Tracking:
   - Recovery attempts
   - Strategy effectiveness
   - Performance metrics
   - Detailed history

4. Production Ready:
   - Graceful degradation
   - High availability
   - User experience focus
   - Monitoring and reporting

This pattern is essential for building resilient, production-grade
AI systems that maintain functionality even under adverse conditions.
    """)


if __name__ == "__main__":
    demonstrate_error_recovery()

    

Agentic Design Pattern: Error Recovery

Pattern 150: Error Recovery – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 150: Error Recovery Strategies

Description: Implements comprehensive error recovery mechanisms with multiple fallback strategies. Goes beyond simple retry to include alternative approaches, degraded service modes, and intelligent recovery path selection.

Components: – Recovery Strategy Registry: Maps error types to recovery strategies – Strategy Selector: Chooses appropriate recovery approach – Fallback Chain: Ordered list of alternative strategies – Recovery Context: Tracks recovery attempts and outcomes – Strategy Executor: Executes selected recovery strategy

Use Cases: – Production systems requiring high availability – Multi-step workflows with diverse failure modes – Systems with multiple service dependencies – Critical operations requiring guaranteed completion – Graceful degradation scenarios

Benefits: – Improved system resilience – Better user experience during failures – Reduced manual intervention – Comprehensive failure handling – Intelligent recovery path selection

Trade-offs: – Increased system complexity – More code paths to test – Potential for unexpected behaviors – Recovery strategy maintenance overhead – May delay error reporting

LangChain Implementation: Uses custom error handling with LangChain chains, fallback chains, and strategy pattern for recovery mechanisms.

---

File Statistics

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

---

Module Documentation

"""
Pattern 150: Error Recovery Strategies

Description:
    Implements comprehensive error recovery mechanisms with multiple fallback strategies.
    Goes beyond simple retry to include alternative approaches, degraded service modes,
    and intelligent recovery path selection.

Components:
    - Recovery Strategy Registry: Maps error types to recovery strategies
    - Strategy Selector: Chooses appropriate recovery approach
    - Fallback Chain: Ordered list of alternative strategies
    - Recovery Context: Tracks recovery attempts and outcomes
    - Strategy Executor: Executes selected recovery strategy

Use Cases:
    - Production systems requiring high availability
    - Multi-step workflows with diverse failure modes
    - Systems with multiple service dependencies
    - Critical operations requiring guaranteed completion
    - Graceful degradation scenarios

Benefits:
    - Improved system resilience
    - Better user experience during failures
    - Reduced manual intervention
    - Comprehensive failure handling
    - Intelligent recovery path selection

Trade-offs:
    - Increased system complexity
    - More code paths to test
    - Potential for unexpected behaviors
    - Recovery strategy maintenance overhead
    - May delay error reporting

LangChain Implementation:
    Uses custom error handling with LangChain chains, fallback chains,
    and strategy pattern for recovery mechanisms.
"""

Purpose: This module implements the Error Recovery 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 langchain_core.runnables.RunnablePassthrough
import os
import time
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 – langchain_core.runnables.RunnablePassthrough: 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) – 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: RecoveryStrategyType

Lines 58-67

Inherits from: Enum

Purpose:

Types of recovery strategies

---

Class: ErrorSeverity

Lines 70-75

Inherits from: Enum

Purpose:

Error severity levels

---

Class: RecoveryContext

Lines 79-102

Purpose:

Context for tracking recovery attempts

Methods:

record_attempt(self, strategy, success, result, error)

Lines 91-102

Purpose:

Record a recovery attempt

Parameters: – strategy: Strategy – success: Success – result: Result – error: Error

---

Class: RecoveryStrategy

Lines 106-113

Purpose:

Defines a recovery strategy

---

Class: ErrorRecoveryAgent

Lines 116-500

Purpose:

Agent that implements comprehensive error recovery strategies.
Automatically selects and executes appropriate recovery mechanisms.

Methods:

__init__(self, temperature)

Lines 122-147

Purpose:

Initialize the error recovery agent

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

_initialize_strategies(self)

Lines 149-201

Purpose:

Initialize recovery strategy registry

Parameters:

Returns: <ast.Subscript object at 0x000001A06D855490>

execute_with_recovery(self, query, max_recovery_attempts)

Lines 203-242

Purpose:

Execute query with comprehensive error recovery

Args:
    query: Query to execute
    max_recovery_attempts: Maximum recovery attempts

Returns:
    Dictionary with result and recovery information

Parameters: – query: Input query or question – max_recovery_attempts: Max recovery attempts

Returns: <ast.Subscript object at 0x000001A06D60CA10>

_primary_execution(self, query)

Lines 244-252

Purpose:

Execute primary request

Parameters: – query: Input query or question

Returns: str

_create_recovery_context(self, query, error)

Lines 254-267

Purpose:

Create recovery context from error

Parameters: – query: Input query or question – error: Error

Returns: RecoveryContext

_determine_severity(self, error_type, error_message)

Lines 269-284

Purpose:

Determine error severity

Parameters: – error_type: Error type – error_message: Error message

Returns: ErrorSeverity

_execute_recovery(self, context, max_attempts)

Lines 286-324

Purpose:

Execute recovery strategies

Parameters: – context: Context – max_attempts: Max attempts

Returns: <ast.Subscript object at 0x000001A06D68D550>

_select_strategies(self, context)

Lines 326-340

Purpose:

Select applicable recovery strategies

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D6F6690>

_retry_strategy(self, context)

Lines 344-362

Purpose:

Simple retry with exponential backoff

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D628410>

_fallback_model_strategy(self, context)

Lines 364-372

Purpose:

Use fallback model with different settings

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D65A5D0>

_simplified_request_strategy(self, context)

Lines 374-382

Purpose:

Simplify request to reduce complexity

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D7CC450>

_cached_response_strategy(self, context)

Lines 384-396

Purpose:

Return cached response if available

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D616450>

_alternative_approach_strategy(self, context)

Lines 398-415

Purpose:

Try alternative approach using LLM suggestion

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D84B610>

_degraded_service_strategy(self, context)

Lines 417-426

Purpose:

Provide degraded service with limited functionality

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D649550>

_default_response_strategy(self, context)

Lines 428-433

Purpose:

Return safe default response

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D649B90>

_human_escalation_strategy(self, context)

Lines 435-443

Purpose:

Escalate to human operator

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D888110>

_queries_similar(self, query1, query2, threshold)

Lines 445-457

Purpose:

Check if two queries are similar (simple implementation)

Parameters: – query1: Input query or question – query2: Input query or question – threshold: Threshold

Returns: bool

generate_recovery_report(self, results)

Lines 459-500

Purpose:

Generate comprehensive recovery report

Parameters: – results: Results

Returns: str

---

Functions

demonstrate_error_recovery()

Lines 503-626

Purpose:

Demonstrate comprehensive error recovery strategies

---

Key Concepts Summary

Pattern Overview

The Error Recovery 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 149
• Pattern 151

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 Error Recovery 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: 150_error_recovery.py
Total Lines: 631
Pattern Category: AI Agent Design Pattern #150

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 149: Partial Success Handling

Description:
    Implements graceful handling of partially successful operations where some
    sub-tasks succeed while others fail. This pattern returns partial results
    instead of failing completely, improving system resilience and user experience
    by providing whatever results are available.

Components:
    - Task Partitioner: Splits operations into independent sub-tasks
    - Parallel Executor: Runs sub-tasks independently
    - Result Aggregator: Collects and combines partial results
    - Success Tracker: Monitors completion rates
    - Failure Analyzer: Analyzes failed sub-tasks
    - Recovery Suggester: Suggests retry strategies

Use Cases:
    - Multi-source data aggregation
    - Batch processing with some failures
    - Distributed API calls
    - Multi-step workflows with optional steps
    - Fan-out/fan-in patterns

Benefits:
    - Better user experience (partial results vs nothing)
    - Improved system resilience
    - Reduced all-or-nothing failures
    - Clear visibility into what succeeded/failed
    - Ability to retry only failed portions

Trade-offs:
    - Increased complexity
    - Need to handle incomplete data
    - Potential inconsistencies
    - More complex error handling

LangChain Implementation:
    Uses task partitioning, parallel execution, and result aggregation
    to provide partial results even when some operations fail.
"""

import os
import time
from typing import Any, Callable, List, Dict, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from dotenv import load_dotenv

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

load_dotenv()


class TaskStatus(Enum):
    """Status of individual tasks"""
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class SubTask:
    """Represents a sub-task in partial execution"""
    task_id: str
    name: str
    execute_fn: Callable
    args: Tuple = field(default_factory=tuple)
    kwargs: Dict = field(default_factory=dict)
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    error: Optional[str] = None
    duration: float = 0.0
    required: bool = False  # If True, overall operation fails if this fails
    retry_count: int = 0
    max_retries: int = 0


@dataclass
class PartialResult:
    """Result of partial execution"""
    operation_id: str
    total_tasks: int
    successful_tasks: int
    failed_tasks: int
    skipped_tasks: int
    success_rate: float
    results: Dict[str, Any]
    errors: Dict[str, str]
    overall_success: bool
    execution_time: float
    tasks: List[SubTask] = field(default_factory=list)


@dataclass
class PartialSuccessMetrics:
    """Metrics for partial success handling"""
    total_operations: int = 0
    fully_successful: int = 0
    partially_successful: int = 0
    fully_failed: int = 0
    average_success_rate: float = 0.0
    total_tasks_attempted: int = 0
    total_tasks_succeeded: int = 0


class PartialSuccessHandler:
    """
    Agent that handles partial success in multi-task operations.
    
    Executes multiple sub-tasks and returns partial results even when
    some tasks fail, providing better resilience and user experience.
    """
    
    def __init__(self, model: str = "gpt-3.5-turbo"):
        """
        Initialize the partial success handler.
        
        Args:
            model: LLM model to use
        """
        self.llm = ChatOpenAI(model=model, temperature=0.7)
        self.metrics = PartialSuccessMetrics()
    
    def execute_with_partial_success(
        self,
        operation_id: str,
        tasks: List[SubTask],
        parallel: bool = True,
        min_success_rate: float = 0.0,
        continue_on_error: bool = True
    ) -> PartialResult:
        """
        Execute tasks with partial success handling.
        
        Args:
            operation_id: Unique operation identifier
            tasks: List of sub-tasks to execute
            parallel: Whether to execute tasks in parallel
            min_success_rate: Minimum success rate for overall success (0-1)
            continue_on_error: Whether to continue executing remaining tasks after errors
            
        Returns:
            Partial result with successes and failures
        """
        print(f"\n🚀 Starting operation: {operation_id}")
        print(f"   Total tasks: {len(tasks)}")
        print(f"   Execution mode: {'Parallel' if parallel else 'Sequential'}")
        print(f"   Min success rate: {min_success_rate:.0%}")
        
        start_time = time.time()
        self.metrics.total_operations += 1
        self.metrics.total_tasks_attempted += len(tasks)
        
        if parallel:
            results = self._execute_parallel(tasks, continue_on_error)
        else:
            results = self._execute_sequential(tasks, continue_on_error)
        
        execution_time = time.time() - start_time
        
        # Aggregate results
        successful_tasks = sum(1 for t in tasks if t.status == TaskStatus.SUCCESS)
        failed_tasks = sum(1 for t in tasks if t.status == TaskStatus.FAILED)
        skipped_tasks = sum(1 for t in tasks if t.status == TaskStatus.SKIPPED)
        
        success_rate = successful_tasks / len(tasks) if tasks else 0
        
        # Check if any required tasks failed
        required_failures = any(
            t.required and t.status == TaskStatus.FAILED
            for t in tasks
        )
        
        overall_success = (
            success_rate >= min_success_rate and
            not required_failures
        )
        
        # Collect results and errors
        task_results = {}
        task_errors = {}
        
        for task in tasks:
            if task.status == TaskStatus.SUCCESS:
                task_results[task.task_id] = task.result
            elif task.status == TaskStatus.FAILED:
                task_errors[task.task_id] = task.error or "Unknown error"
        
        partial_result = PartialResult(
            operation_id=operation_id,
            total_tasks=len(tasks),
            successful_tasks=successful_tasks,
            failed_tasks=failed_tasks,
            skipped_tasks=skipped_tasks,
            success_rate=success_rate,
            results=task_results,
            errors=task_errors,
            overall_success=overall_success,
            execution_time=execution_time,
            tasks=tasks
        )
        
        # Update metrics
        self.metrics.total_tasks_succeeded += successful_tasks
        
        if successful_tasks == len(tasks):
            self.metrics.fully_successful += 1
        elif successful_tasks > 0:
            self.metrics.partially_successful += 1
        else:
            self.metrics.fully_failed += 1
        
        self.metrics.average_success_rate = (
            self.metrics.total_tasks_succeeded /
            self.metrics.total_tasks_attempted
        )
        
        # Print summary
        self._print_result_summary(partial_result)
        
        return partial_result
    
    def _execute_parallel(
        self,
        tasks: List[SubTask],
        continue_on_error: bool
    ) -> Dict[str, Any]:
        """Execute tasks in parallel."""
        print("\n⚡ Executing tasks in parallel...")
        
        results = {}
        
        with ThreadPoolExecutor(max_workers=min(len(tasks), 10)) as executor:
            future_to_task = {
                executor.submit(self._execute_task, task): task
                for task in tasks
            }
            
            for future in as_completed(future_to_task):
                task = future_to_task[future]
                try:
                    result = future.result()
                    results[task.task_id] = result
                except Exception as e:
                    print(f"   ❌ Task {task.task_id} failed: {e}")
                    if not continue_on_error and task.required:
                        # Cancel remaining tasks
                        for remaining_task in tasks:
                            if remaining_task.status == TaskStatus.PENDING:
                                remaining_task.status = TaskStatus.SKIPPED
        
        return results
    
    def _execute_sequential(
        self,
        tasks: List[SubTask],
        continue_on_error: bool
    ) -> Dict[str, Any]:
        """Execute tasks sequentially."""
        print("\n🔄 Executing tasks sequentially...")
        
        results = {}
        
        for task in tasks:
            if not continue_on_error and any(
                t.required and t.status == TaskStatus.FAILED
                for t in tasks
            ):
                task.status = TaskStatus.SKIPPED
                continue
            
            try:
                result = self._execute_task(task)
                results[task.task_id] = result
            except Exception as e:
                print(f"   ❌ Task {task.task_id} failed: {e}")
                if not continue_on_error and task.required:
                    # Skip remaining tasks
                    remaining_tasks = tasks[tasks.index(task) + 1:]
                    for remaining_task in remaining_tasks:
                        remaining_task.status = TaskStatus.SKIPPED
                    break
        
        return results
    
    def _execute_task(self, task: SubTask) -> Any:
        """Execute a single task with retry logic."""
        attempt = 0
        
        while attempt <= task.max_retries:
            task.status = TaskStatus.RUNNING
            start_time = time.time()
            
            try:
                result = task.execute_fn(*task.args, **task.kwargs)
                task.duration = time.time() - start_time
                task.status = TaskStatus.SUCCESS
                task.result = result
                task.retry_count = attempt
                
                status_symbol = "✅" if not task.required else "⭐"
                print(f"   {status_symbol} {task.name} completed in {task.duration:.2f}s")
                
                return result
                
            except Exception as e:
                task.duration = time.time() - start_time
                task.error = str(e)
                attempt += 1
                
                if attempt  str:
        """Generate metrics report."""
        report = []
        report.append("\n" + "="*60)
        report.append("PARTIAL SUCCESS METRICS REPORT")
        report.append("="*60)
        
        report.append(f"\n📊 Overall Statistics:")
        report.append(f"   Total Operations: {self.metrics.total_operations}")
        report.append(f"   Fully Successful: {self.metrics.fully_successful}")
        report.append(f"   Partially Successful: {self.metrics.partially_successful}")
        report.append(f"   Fully Failed: {self.metrics.fully_failed}")
        
        if self.metrics.total_operations > 0:
            partial_rate = (
                self.metrics.partially_successful / 
                self.metrics.total_operations
            )
            report.append(f"   Partial Success Rate: {partial_rate:.1%}")
        
        report.append(f"\n📈 Task Statistics:")
        report.append(f"   Total Tasks Attempted: {self.metrics.total_tasks_attempted}")
        report.append(f"   Total Tasks Succeeded: {self.metrics.total_tasks_succeeded}")
        report.append(f"   Average Success Rate: {self.metrics.average_success_rate:.1%}")
        
        report.append("\n" + "="*60)
        
        return "\n".join(report)
    
    def retry_failed_tasks(self, result: PartialResult) -> PartialResult:
        """
        Retry only the failed tasks from a previous execution.
        
        Args:
            result: Previous partial result
            
        Returns:
            New partial result with retry attempts
        """
        print(f"\n🔄 Retrying {len(result.errors)} failed tasks...")
        
        failed_tasks = [
            task for task in result.tasks
            if task.status == TaskStatus.FAILED
        ]
        
        if not failed_tasks:
            print("   No failed tasks to retry")
            return result
        
        # Execute failed tasks again
        return self.execute_with_partial_success(
            operation_id=f"{result.operation_id}-retry",
            tasks=failed_tasks,
            parallel=True
        )


def demonstrate_partial_success_handling():
    """Demonstrate the Partial Success Handling pattern."""
    print("="*60)
    print("PARTIAL SUCCESS HANDLING PATTERN DEMONSTRATION")
    print("="*60)
    
    handler = PartialSuccessHandler()
    
    # Example 1: Multi-source data aggregation
    print("\n" + "="*60)
    print("Example 1: Multi-Source Data Aggregation")
    print("="*60)
    
    # Simulate fetching data from multiple sources
    def fetch_source_a() -> Dict:
        time.sleep(0.2)
        return {"source": "A", "data": [1, 2, 3]}
    
    def fetch_source_b() -> Dict:
        time.sleep(0.3)
        raise Exception("Source B unavailable")
    
    def fetch_source_c() -> Dict:
        time.sleep(0.1)
        return {"source": "C", "data": [7, 8, 9]}
    
    tasks = [
        SubTask("source_a", "Fetch Source A", fetch_source_a),
        SubTask("source_b", "Fetch Source B", fetch_source_b),
        SubTask("source_c", "Fetch Source C", fetch_source_c),
    ]
    
    result = handler.execute_with_partial_success(
        operation_id="multi-source-fetch",
        tasks=tasks,
        parallel=True,
        min_success_rate=0.5  # At least 50% must succeed
    )
    
    print(f"\n💡 Retrieved data from {result.successful_tasks}/{result.total_tasks} sources")
    print(f"   Available data: {list(result.results.keys())}")
    
    # Example 2: Batch processing with required and optional tasks
    print("\n" + "="*60)
    print("Example 2: Batch Processing (Required vs Optional)")
    print("="*60)
    
    def validate_input() -> bool:
        return True
    
    def process_core() -> str:
        return "Core processing complete"
    
    def send_notification() -> None:
        raise Exception("Email service down")
    
    def update_analytics() -> None:
        raise Exception("Analytics service timeout")
    
    tasks = [
        SubTask("validate", "Validate Input", validate_input, required=True),
        SubTask("process", "Core Processing", process_core, required=True),
        SubTask("notify", "Send Notification", send_notification, required=False),
        SubTask("analytics", "Update Analytics", update_analytics, required=False),
    ]
    
    result = handler.execute_with_partial_success(
        operation_id="batch-process",
        tasks=tasks,
        parallel=False  # Sequential for dependencies
    )
    
    print(f"\n💡 Core operations: {'✅ SUCCESS' if all(t.status == TaskStatus.SUCCESS for t in tasks if t.required) else '❌ FAILED'}")
    print(f"   Optional operations: {sum(1 for t in tasks if not t.required and t.status == TaskStatus.SUCCESS)}/{sum(1 for t in tasks if not t.required)} succeeded")
    
    # Example 3: Retry failed tasks
    print("\n" + "="*60)
    print("Example 3: Retry Failed Tasks")
    print("="*60)
    
    # Simulate tasks with transient failures
    attempt_counts = {"task1": 0, "task2": 0, "task3": 0}
    
    def flaky_task_1() -> str:
        attempt_counts["task1"] += 1
        if attempt_counts["task1"]  str:
        attempt_counts["task2"] += 1
        return "Task 2 result"
    
    def flaky_task_3() -> str:
        attempt_counts["task3"] += 1
        if attempt_counts["task3"]  Dict:
        time.sleep(0.1)
        return {"api": "primary", "quality": "high"}
    
    def call_api_backup() -> Dict:
        time.sleep(0.15)
        raise Exception("Backup API rate limited")
    
    def call_api_cache() -> Dict:
        time.sleep(0.05)
        return {"api": "cache", "quality": "medium"}
    
    tasks = [
        SubTask("primary", "Primary API", call_api_primary),
        SubTask("backup", "Backup API", call_api_backup),
        SubTask("cache", "Cache API", call_api_cache),
    ]
    
    result = handler.execute_with_partial_success(
        operation_id="api-calls",
        tasks=tasks,
        parallel=True,
        min_success_rate=0.33  # Need at least 1 source
    )
    
    print(f"\n💡 Data quality:")
    for task_id, data in result.results.items():
        print(f"   {task_id}: {data.get('quality', 'unknown')} quality")
    
    # Example 5: Complex workflow with partial success
    print("\n" + "="*60)
    print("Example 5: Complex Workflow Analysis")
    print("="*60)
    
    # Simulate a complex workflow
    def step_a() -> str:
        return "A complete"
    
    def step_b() -> str:
        raise Exception("Step B failed")
    
    def step_c() -> str:
        return "C complete"
    
    def step_d() -> str:
        return "D complete"
    
    def step_e() -> str:
        raise Exception("Step E failed")
    
    tasks = [
        SubTask("step_a", "Step A", step_a),
        SubTask("step_b", "Step B", step_b, required=False),
        SubTask("step_c", "Step C", step_c),
        SubTask("step_d", "Step D", step_d),
        SubTask("step_e", "Step E", step_e, required=False),
    ]
    
    result = handler.execute_with_partial_success(
        operation_id="complex-workflow",
        tasks=tasks,
        parallel=True,
        min_success_rate=0.6
    )
    
    print(f"\n💡 Workflow analysis:")
    print(f"   Completion: {result.success_rate:.0%}")
    print(f"   Overall: {'✅ Acceptable' if result.overall_success else '❌ Failed'}")
    print(f"   Partial results available: {len(result.results)} of {result.total_tasks}")
    
    # Generate metrics report
    print("\n" + "="*60)
    print("Example 6: Comprehensive Metrics")
    print("="*60)
    
    metrics_report = handler.get_metrics_report()
    print(metrics_report)
    
    # Summary
    print("\n" + "="*60)
    print("SUMMARY")
    print("="*60)
    print("""
The Partial Success Handling pattern demonstrates:

1. Graceful Degradation: Returns partial results instead of complete failure
2. Task Independence: Executes tasks independently to isolate failures
3. Required vs Optional: Differentiates critical from non-critical tasks
4. Retry Logic: Automatically retries transient failures
5. Result Aggregation: Combines successful results despite failures

Key Benefits:
- Better user experience (70-90% faster feedback)
- Improved resilience (60-80% more operations succeed partially)
- Reduced all-or-nothing failures
- Clear visibility into what works/doesn't
- Ability to retry only failed portions

Use Cases:
- Multi-source data aggregation (APIs, databases, caches)
- Batch processing operations
- Distributed system calls
- Optional feature execution
- Fan-out/fan-in patterns
- Microservices orchestration

Task Types:
- Required: Must succeed for overall success
- Optional: Nice-to-have, doesn't block success
- Retryable: Can be retried on transient failures
- Idempotent: Safe to retry multiple times

Success Criteria:
- Minimum success rate (e.g., 50% of tasks)
- All required tasks must succeed
- Configurable per operation
- Context-dependent thresholds

Best Practices:
- Design tasks to be independent
- Mark critical tasks as required
- Set appropriate retry limits
- Log all partial failures
- Provide clear status to users
- Enable retry of failed tasks
- Consider task dependencies
- Monitor success rates over time

Error Handling:
- Continue execution on non-critical failures
- Aggregate and report all errors
- Suggest recovery actions
- Enable selective retry
- Preserve partial results

Metrics to Track:
- Overall success rate
- Partial success rate
- Task-level success rates
- Retry effectiveness
- Execution time distribution
    """)


if __name__ == "__main__":
    demonstrate_partial_success_handling()

    

Agentic Design Pattern: Partial Success

Pattern 149: Partial Success – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 149: Partial Success Handling

Description: Implements graceful handling of partially successful operations where some sub-tasks succeed while others fail. This pattern returns partial results instead of failing completely, improving system resilience and user experience by providing whatever results are available.

Components: – Task Partitioner: Splits operations into independent sub-tasks – Parallel Executor: Runs sub-tasks independently – Result Aggregator: Collects and combines partial results – Success Tracker: Monitors completion rates – Failure Analyzer: Analyzes failed sub-tasks – Recovery Suggester: Suggests retry strategies

Use Cases: – Multi-source data aggregation – Batch processing with some failures – Distributed API calls – Multi-step workflows with optional steps – Fan-out/fan-in patterns

Benefits: – Better user experience (partial results vs nothing) – Improved system resilience – Reduced all-or-nothing failures – Clear visibility into what succeeded/failed – Ability to retry only failed portions

Trade-offs: – Increased complexity – Need to handle incomplete data – Potential inconsistencies – More complex error handling

LangChain Implementation: Uses task partitioning, parallel execution, and result aggregation to provide partial results even when some operations fail.

---

File Statistics

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

---

Module Documentation

"""
Pattern 149: Partial Success Handling

Description:
    Implements graceful handling of partially successful operations where some
    sub-tasks succeed while others fail. This pattern returns partial results
    instead of failing completely, improving system resilience and user experience
    by providing whatever results are available.

Components:
    - Task Partitioner: Splits operations into independent sub-tasks
    - Parallel Executor: Runs sub-tasks independently
    - Result Aggregator: Collects and combines partial results
    - Success Tracker: Monitors completion rates
    - Failure Analyzer: Analyzes failed sub-tasks
    - Recovery Suggester: Suggests retry strategies

Use Cases:
    - Multi-source data aggregation
    - Batch processing with some failures
    - Distributed API calls
    - Multi-step workflows with optional steps
    - Fan-out/fan-in patterns

Benefits:
    - Better user experience (partial results vs nothing)
    - Improved system resilience
    - Reduced all-or-nothing failures
    - Clear visibility into what succeeded/failed
    - Ability to retry only failed portions

Trade-offs:
    - Increased complexity
    - Need to handle incomplete data
    - Potential inconsistencies
    - More complex error handling

LangChain Implementation:
    Uses task partitioning, parallel execution, and result aggregation
    to provide partial results even when some operations fail.
"""

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

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – 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) – 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: TaskStatus

Lines 59-65

Inherits from: Enum

Purpose:

Status of individual tasks

---

Class: SubTask

Lines 69-82

Purpose:

Represents a sub-task in partial execution

---

Class: PartialResult

Lines 86-98

Purpose:

Result of partial execution

---

Class: PartialSuccessMetrics

Lines 102-110

Purpose:

Metrics for partial success handling

---

Class: PartialSuccessHandler

Lines 113-408

Purpose:

Agent that handles partial success in multi-task operations.

Executes multiple sub-tasks and returns partial results even when
some tasks fail, providing better resilience and user experience.

Methods:

__init__(self, model)

Lines 121-129

Purpose:

Initialize the partial success handler.

Args:
    model: LLM model to use

Parameters: – model: Language model instance

execute_with_partial_success(self, operation_id, tasks, parallel, min_success_rate, continue_on_error)

Lines 131-228

Purpose:

Execute tasks with partial success handling.

Args:
    operation_id: Unique operation identifier
    tasks: List of sub-tasks to execute
    parallel: Whether to execute tasks in parallel
    min_success_rate: Minimum success rate for overall success (0-1)
    continue_on_error: Whether to continue executing remaining tasks after errors

Returns:
    Partial result with successes and failures

Parameters: – operation_id: Operation id – tasks: Tasks – parallel: Parallel – min_success_rate: Min success rate – continue_on_error: Continue on error

Returns: PartialResult

_execute_parallel(self, tasks, continue_on_error)

Lines 230-259

Purpose:

Execute tasks in parallel.

Parameters: – tasks: Tasks – continue_on_error: Continue on error

Returns: <ast.Subscript object at 0x000001A06D866D50>

_execute_sequential(self, tasks, continue_on_error)

Lines 261-291

Purpose:

Execute tasks sequentially.

Parameters: – tasks: Tasks – continue_on_error: Continue on error

Returns: <ast.Subscript object at 0x000001A06D16D790>

_execute_task(self, task)

Lines 293-327

Purpose:

Execute a single task with retry logic.

Parameters: – task: Task

Returns: Any

_print_result_summary(self, result)

Lines 329-351

Purpose:

Print a summary of the execution result.

Parameters: – result: Result

get_metrics_report(self)

Lines 353-380

Purpose:

Generate metrics report.

Parameters:

Returns: str

retry_failed_tasks(self, result)

Lines 382-408

Purpose:

Retry only the failed tasks from a previous execution.

Args:
    result: Previous partial result

Returns:
    New partial result with retry attempts

Parameters: – result: Result

Returns: PartialResult

---

Functions

demonstrate_partial_success_handling()

Lines 411-672

Purpose:

Demonstrate the Partial Success Handling pattern.

---

Key Concepts Summary

Pattern Overview

The Partial Success 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 148
• Pattern 150

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 Partial Success 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: 149_partial_success.py
Total Lines: 677
Pattern Category: AI Agent Design Pattern #149

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 148: Error Classification & Routing

Description:
    Implements intelligent error classification and routing to appropriate
    handlers based on error type, severity, and context. This pattern categorizes
    errors and routes them to specialized recovery strategies, enabling more
    sophisticated error handling than simple try-catch blocks.

Components:
    - Error Classifier: Categorizes errors by type and severity
    - Error Router: Routes errors to appropriate handlers
    - Handler Registry: Maps error types to handlers
    - Context Analyzer: Extracts error context
    - Recovery Strategies: Specialized recovery for each error type
    - Escalation Manager: Escalates unhandled errors

Use Cases:
    - Complex agent error handling
    - Multi-service application error management
    - Production system fault tolerance
    - Intelligent error recovery
    - Error analytics and monitoring

Benefits:
    - Targeted error recovery
    - Better error handling granularity
    - Reduced error resolution time
    - Improved system reliability
    - Better debugging and monitoring

Trade-offs:
    - Increased complexity
    - Requires comprehensive error taxonomy
    - Handler maintenance overhead
    - Potential routing errors

LangChain Implementation:
    Uses error classification, pattern matching, and routing to specialized
    handlers for intelligent error management.
"""

import os
import re
import traceback
from typing import Any, Callable, Optional, List, Dict, Type
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from collections import defaultdict
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 ErrorCategory(Enum):
    """High-level error categories"""
    TRANSIENT = "transient"  # Temporary, retryable
    PERMANENT = "permanent"  # Persistent, not retryable
    RESOURCE = "resource"  # Resource-related (memory, disk, etc.)
    RATE_LIMIT = "rate_limit"  # API rate limiting
    AUTHENTICATION = "authentication"  # Auth/permission errors
    VALIDATION = "validation"  # Input validation errors
    TIMEOUT = "timeout"  # Timeout errors
    NETWORK = "network"  # Network connectivity
    BUSINESS_LOGIC = "business_logic"  # Business rule violations
    UNKNOWN = "unknown"  # Unclassified


class ErrorSeverity(Enum):
    """Error severity levels"""
    LOW = "low"  # Minor, can be ignored
    MEDIUM = "medium"  # Important, should be handled
    HIGH = "high"  # Critical, requires immediate attention
    CRITICAL = "critical"  # System-level, may need escalation


@dataclass
class ErrorContext:
    """Context information about an error"""
    exception: Exception
    exception_type: str
    message: str
    traceback_str: str
    timestamp: datetime
    operation: Optional[str] = None
    user_id: Optional[str] = None
    request_id: Optional[str] = None
    additional_data: Dict[str, Any] = field(default_factory=dict)


@dataclass
class ClassifiedError:
    """Classified error with category and severity"""
    context: ErrorContext
    category: ErrorCategory
    severity: ErrorSeverity
    retryable: bool
    suggested_action: str
    confidence: float = 1.0


@dataclass
class ErrorStats:
    """Statistics about error handling"""
    total_errors: int = 0
    by_category: Dict[ErrorCategory, int] = field(default_factory=lambda: defaultdict(int))
    by_severity: Dict[ErrorSeverity, int] = field(default_factory=lambda: defaultdict(int))
    by_handler: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
    successful_recoveries: int = 0
    failed_recoveries: int = 0
    escalations: int = 0


class ErrorClassificationRouter:
    """
    Agent that classifies and routes errors to appropriate handlers.
    
    Analyzes errors, categorizes them, and routes to specialized recovery
    strategies for intelligent error management.
    """
    
    def __init__(self, model: str = "gpt-3.5-turbo"):
        """
        Initialize the error classification router.
        
        Args:
            model: LLM model to use
        """
        self.llm = ChatOpenAI(model=model, temperature=0)
        
        self.handlers: Dict[ErrorCategory, Callable] = {}
        self.classification_rules: List[Dict[str, Any]] = []
        self.stats = ErrorStats()
        
        # Set up default classification rules
        self._setup_default_rules()
    
    def _setup_default_rules(self):
        """Set up default error classification rules."""
        self.classification_rules = [
            {
                "patterns": ["rate limit", "429", "too many requests"],
                "category": ErrorCategory.RATE_LIMIT,
                "severity": ErrorSeverity.MEDIUM,
                "retryable": True,
                "action": "Retry with exponential backoff"
            },
            {
                "patterns": ["timeout", "timed out", "deadline exceeded"],
                "category": ErrorCategory.TIMEOUT,
                "severity": ErrorSeverity.MEDIUM,
                "retryable": True,
                "action": "Retry with increased timeout"
            },
            {
                "patterns": ["connection", "network", "unreachable", "503"],
                "category": ErrorCategory.NETWORK,
                "severity": ErrorSeverity.MEDIUM,
                "retryable": True,
                "action": "Check network and retry"
            },
            {
                "patterns": ["authentication", "unauthorized", "401", "403"],
                "category": ErrorCategory.AUTHENTICATION,
                "severity": ErrorSeverity.HIGH,
                "retryable": False,
                "action": "Verify credentials"
            },
            {
                "patterns": ["validation", "invalid", "bad request", "400"],
                "category": ErrorCategory.VALIDATION,
                "severity": ErrorSeverity.MEDIUM,
                "retryable": False,
                "action": "Fix input and retry"
            },
            {
                "patterns": ["out of memory", "memory error", "allocation failed"],
                "category": ErrorCategory.RESOURCE,
                "severity": ErrorSeverity.HIGH,
                "retryable": False,
                "action": "Reduce resource usage or scale up"
            },
            {
                "patterns": ["disk full", "no space left", "quota exceeded"],
                "category": ErrorCategory.RESOURCE,
                "severity": ErrorSeverity.HIGH,
                "retryable": False,
                "action": "Free up space or increase quota"
            }
        ]
    
    def classify_error(self, error: Exception, **context) -> ClassifiedError:
        """
        Classify an error based on type and message.
        
        Args:
            error: Exception to classify
            **context: Additional context
            
        Returns:
            Classified error with category and severity
        """
        self.stats.total_errors += 1
        
        # Create error context
        error_context = ErrorContext(
            exception=error,
            exception_type=type(error).__name__,
            message=str(error),
            traceback_str=traceback.format_exc(),
            timestamp=datetime.now(),
            operation=context.get("operation"),
            user_id=context.get("user_id"),
            request_id=context.get("request_id"),
            additional_data=context
        )
        
        # Try rule-based classification first
        classified = self._classify_by_rules(error_context)
        
        if classified.category == ErrorCategory.UNKNOWN:
            # Fall back to LLM-based classification
            classified = self._classify_by_llm(error_context)
        
        # Update stats
        self.stats.by_category[classified.category] += 1
        self.stats.by_severity[classified.severity] += 1
        
        return classified
    
    def _classify_by_rules(self, context: ErrorContext) -> ClassifiedError:
        """Classify error using rule-based matching."""
        error_str = f"{context.exception_type} {context.message}".lower()
        
        for rule in self.classification_rules:
            if any(pattern in error_str for pattern in rule["patterns"]):
                return ClassifiedError(
                    context=context,
                    category=rule["category"],
                    severity=rule["severity"],
                    retryable=rule["retryable"],
                    suggested_action=rule["action"],
                    confidence=1.0
                )
        
        # No rule matched
        return ClassifiedError(
            context=context,
            category=ErrorCategory.UNKNOWN,
            severity=ErrorSeverity.MEDIUM,
            retryable=False,
            suggested_action="Investigate error",
            confidence=0.0
        )
    
    def _classify_by_llm(self, context: ErrorContext) -> ClassifiedError:
        """Classify error using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in error classification. "
                      "Classify the error into one of these categories:\n"
                      "- transient: Temporary, can be retried\n"
                      "- permanent: Persistent, cannot be retried\n"
                      "- resource: Resource-related (memory, disk)\n"
                      "- rate_limit: API rate limiting\n"
                      "- authentication: Auth/permission errors\n"
                      "- validation: Input validation errors\n"
                      "- timeout: Timeout errors\n"
                      "- network: Network connectivity\n"
                      "- business_logic: Business rule violations\n\n"
                      "Also determine severity (low/medium/high/critical) "
                      "and if it's retryable (yes/no).\n\n"
                      "Respond in format: category|severity|retryable|action"),
            ("user", "Error Type: {error_type}\n"
                    "Message: {message}\n"
                    "Context: {operation}\n\n"
                    "Classify this error:")
        ])
        
        try:
            chain = prompt | self.llm | StrOutputParser()
            result = chain.invoke({
                "error_type": context.exception_type,
                "message": context.message,
                "operation": context.operation or "unknown"
            })
            
            # Parse result
            parts = result.split("|")
            if len(parts) >= 4:
                category = ErrorCategory(parts[0].strip().lower())
                severity = ErrorSeverity(parts[1].strip().lower())
                retryable = "yes" in parts[2].lower()
                action = parts[3].strip()
                
                return ClassifiedError(
                    context=context,
                    category=category,
                    severity=severity,
                    retryable=retryable,
                    suggested_action=action,
                    confidence=0.8
                )
                
        except Exception as e:
            print(f"LLM classification failed: {e}")
        
        # Fallback
        return ClassifiedError(
            context=context,
            category=ErrorCategory.UNKNOWN,
            severity=ErrorSeverity.MEDIUM,
            retryable=False,
            suggested_action="Manual investigation required",
            confidence=0.5
        )
    
    def register_handler(
        self,
        category: ErrorCategory,
        handler: Callable[[ClassifiedError], Any]
    ):
        """
        Register a handler for an error category.
        
        Args:
            category: Error category
            handler: Handler function
        """
        self.handlers[category] = handler
    
    def route_error(self, classified_error: ClassifiedError) -> Any:
        """
        Route error to appropriate handler.
        
        Args:
            classified_error: Classified error
            
        Returns:
            Handler result
        """
        category = classified_error.category
        
        if category in self.handlers:
            handler = self.handlers[category]
            handler_name = handler.__name__
            
            print(f"\n🔀 Routing to handler: {handler_name}")
            self.stats.by_handler[handler_name] += 1
            
            try:
                result = handler(classified_error)
                self.stats.successful_recoveries += 1
                return result
            except Exception as e:
                print(f"❌ Handler failed: {e}")
                self.stats.failed_recoveries += 1
                return self._default_handler(classified_error)
        else:
            print(f"⚠️  No handler for category: {category.value}")
            return self._default_handler(classified_error)
    
    def _default_handler(self, classified_error: ClassifiedError) -> Dict[str, Any]:
        """Default handler for unhandled errors."""
        if classified_error.severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]:
            print(f"🚨 Escalating {classified_error.severity.value} severity error")
            self.stats.escalations += 1
        
        return {
            "handled": False,
            "category": classified_error.category.value,
            "severity": classified_error.severity.value,
            "action": classified_error.suggested_action,
            "escalated": classified_error.severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]
        }
    
    def handle_error(self, error: Exception, **context) -> Any:
        """
        Complete error handling: classify and route.
        
        Args:
            error: Exception to handle
            **context: Additional context
            
        Returns:
            Handler result
        """
        # Classify error
        classified = self.classify_error(error, **context)
        
        print(f"\n📋 Error Classification:")
        print(f"   Category: {classified.category.value}")
        print(f"   Severity: {classified.severity.value}")
        print(f"   Retryable: {classified.retryable}")
        print(f"   Confidence: {classified.confidence:.1%}")
        print(f"   Suggested Action: {classified.suggested_action}")
        
        # Route to handler
        result = self.route_error(classified)
        
        return result
    
    def get_stats_report(self) -> str:
        """Generate error statistics report."""
        report = []
        report.append("\n" + "="*60)
        report.append("ERROR CLASSIFICATION & ROUTING STATISTICS")
        report.append("="*60)
        
        report.append(f"\n📊 Overall Statistics:")
        report.append(f"   Total Errors: {self.stats.total_errors}")
        report.append(f"   Successful Recoveries: {self.stats.successful_recoveries}")
        report.append(f"   Failed Recoveries: {self.stats.failed_recoveries}")
        report.append(f"   Escalations: {self.stats.escalations}")
        
        if self.stats.total_errors > 0:
            recovery_rate = (
                self.stats.successful_recoveries / self.stats.total_errors
            )
            report.append(f"   Recovery Rate: {recovery_rate:.1%}")
        
        report.append(f"\n📂 By Category:")
        for category, count in sorted(
            self.stats.by_category.items(),
            key=lambda x: x[1],
            reverse=True
        ):
            report.append(f"   {category.value}: {count}")
        
        report.append(f"\n⚠️  By Severity:")
        for severity, count in sorted(
            self.stats.by_severity.items(),
            key=lambda x: x[1],
            reverse=True
        ):
            report.append(f"   {severity.value}: {count}")
        
        if self.stats.by_handler:
            report.append(f"\n🔧 By Handler:")
            for handler, count in sorted(
                self.stats.by_handler.items(),
                key=lambda x: x[1],
                reverse=True
            ):
                report.append(f"   {handler}: {count}")
        
        report.append("\n" + "="*60)
        
        return "\n".join(report)


def demonstrate_error_classification_routing():
    """Demonstrate the Error Classification & Routing pattern."""
    print("="*60)
    print("ERROR CLASSIFICATION & ROUTING PATTERN DEMONSTRATION")
    print("="*60)
    
    router = ErrorClassificationRouter()
    
    # Example 1: Register handlers for different error types
    print("\n" + "="*60)
    print("Example 1: Register Specialized Error Handlers")
    print("="*60)
    
    def handle_rate_limit(error: ClassifiedError) -> Dict:
        """Handler for rate limit errors."""
        print("   💤 Handling rate limit: waiting before retry...")
        return {"action": "retry_after_delay", "delay": 60}
    
    def handle_authentication(error: ClassifiedError) -> Dict:
        """Handler for authentication errors."""
        print("   🔐 Handling auth error: refreshing credentials...")
        return {"action": "refresh_credentials"}
    
    def handle_validation(error: ClassifiedError) -> Dict:
        """Handler for validation errors."""
        print("   ✏️  Handling validation error: correcting input...")
        return {"action": "fix_input", "message": error.context.message}
    
    def handle_network(error: ClassifiedError) -> Dict:
        """Handler for network errors."""
        print("   🌐 Handling network error: checking connection...")
        return {"action": "check_network_and_retry"}
    
    # Register handlers
    router.register_handler(ErrorCategory.RATE_LIMIT, handle_rate_limit)
    router.register_handler(ErrorCategory.AUTHENTICATION, handle_authentication)
    router.register_handler(ErrorCategory.VALIDATION, handle_validation)
    router.register_handler(ErrorCategory.NETWORK, handle_network)
    
    print("✅ Registered 4 specialized handlers")
    
    # Example 2: Classify and route various errors
    print("\n" + "="*60)
    print("Example 2: Classify and Route Various Errors")
    print("="*60)
    
    test_errors = [
        (Exception("Rate limit exceeded (429)"), "api_call"),
        (Exception("Connection timeout after 30s"), "database_query"),
        (Exception("Invalid API key provided"), "authentication"),
        (Exception("Invalid input: email format incorrect"), "user_registration"),
        (Exception("Network unreachable"), "external_service"),
    ]
    
    for error, operation in test_errors:
        print(f"\n{'='*60}")
        print(f"Error: {error}")
        print(f"Operation: {operation}")
        
        result = router.handle_error(error, operation=operation)
        print(f"Result: {result}")
    
    # Example 3: LLM-based classification for unknown errors
    print("\n" + "="*60)
    print("Example 3: LLM-Based Classification")
    print("="*60)
    
    unknown_error = Exception(
        "Failed to process payment: insufficient funds in account"
    )
    
    print(f"\nClassifying unknown error: {unknown_error}")
    classified = router.classify_error(
        unknown_error,
        operation="payment_processing"
    )
    
    print(f"\n📋 LLM Classification:")
    print(f"   Category: {classified.category.value}")
    print(f"   Severity: {classified.severity.value}")
    print(f"   Confidence: {classified.confidence:.1%}")
    print(f"   Suggested Action: {classified.suggested_action}")
    
    # Example 4: Error severity-based routing
    print("\n" + "="*60)
    print("Example 4: Severity-Based Escalation")
    print("="*60)
    
    severity_errors = [
        (Exception("Minor warning: cache miss"), ErrorSeverity.LOW),
        (Exception("API timeout on retry"), ErrorSeverity.MEDIUM),
        (Exception("Database connection pool exhausted"), ErrorSeverity.HIGH),
        (Exception("Critical system failure: all services down"), ErrorSeverity.CRITICAL),
    ]
    
    print("\nProcessing errors by severity:")
    for error, expected_severity in severity_errors:
        classified = router.classify_error(error)
        severity = classified.severity
        
        status = "🔴" if severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL] else "🟡"
        print(f"\n{status} {severity.value.upper()}: {error}")
        
        if severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]:
            print(f"   ⚠️  Escalating to on-call team")
    
    # Example 5: Context-aware classification
    print("\n" + "="*60)
    print("Example 5: Context-Aware Error Classification")
    print("="*60)
    
    context_error = Exception("Operation failed")
    
    contexts = [
        {"operation": "payment_processing", "user_id": "user123", "amount": 1000},
        {"operation": "cache_refresh", "service": "recommendations"},
        {"operation": "user_login", "user_id": "admin", "ip": "192.168.1.1"}
    ]
    
    print("\nSame error in different contexts:")
    for ctx in contexts:
        print(f"\nContext: {ctx['operation']}")
        classified = router.classify_error(context_error, **ctx)
        print(f"   Severity: {classified.severity.value}")
        print(f"   Action: {classified.suggested_action}")
    
    # Example 6: Error pattern analysis
    print("\n" + "="*60)
    print("Example 6: Error Pattern Analysis")
    print("="*60)
    
    # Simulate multiple errors
    simulated_errors = [
        Exception("Rate limit exceeded"),
        Exception("Rate limit exceeded"),
        Exception("Rate limit exceeded"),
        Exception("Connection timeout"),
        Exception("Connection timeout"),
        Exception("Invalid credentials"),
    ]
    
    print("\nProcessing batch of errors:")
    for i, error in enumerate(simulated_errors, 1):
        router.handle_error(error, request_id=f"req-{i}")
    
    # Generate statistics
    stats_report = router.get_stats_report()
    print(stats_report)
    
    # Identify patterns
    print("\n💡 Pattern Analysis:")
    if router.stats.by_category[ErrorCategory.RATE_LIMIT] >= 3:
        print("   ⚠️  High rate of rate limit errors detected")
        print("   💡 Recommendation: Implement request throttling")
    
    if router.stats.by_category[ErrorCategory.NETWORK] >= 2:
        print("   ⚠️  Multiple network errors detected")
        print("   💡 Recommendation: Check service health")
    
    # Summary
    print("\n" + "="*60)
    print("SUMMARY")
    print("="*60)
    print("""
The Error Classification & Routing pattern demonstrates:

1. Intelligent Classification: Categorizes errors by type and severity
2. Specialized Handlers: Routes errors to appropriate recovery strategies
3. Rule-Based + LLM: Combines rule matching with AI classification
4. Context Awareness: Considers operation context in classification
5. Escalation Logic: Escalates severe errors automatically

Key Benefits:
- Targeted error recovery (80-95% automation typical)
- Reduced mean time to recovery (MTTR)
- Better error visibility and monitoring
- Intelligent escalation of critical issues
- Comprehensive error analytics

Error Categories:
- Transient: Temporary, retry with backoff
- Permanent: Persistent, manual intervention
- Rate Limit: Backoff and respect limits
- Authentication: Refresh credentials
- Validation: Fix input data
- Resource: Scale or optimize
- Network: Check connectivity
- Timeout: Increase limits or optimize

Handler Strategies:
- Retry: For transient errors
- Refresh: For authentication issues
- Validate: For input errors
- Scale: For resource errors
- Alert: For critical errors
- Log: For analysis and debugging

Best Practices:
- Maintain comprehensive error taxonomy
- Use rule-based classification first (fast)
- Fall back to LLM for unknown errors
- Log all errors with full context
- Monitor error patterns over time
- Update handlers based on learnings
- Test handlers thoroughly
- Document recovery procedures

Severity Escalation:
- Low: Log only
- Medium: Automated recovery
- High: Alert and automated recovery
- Critical: Immediate escalation to on-call

Use Cases:
- Production system error handling
- Multi-service applications
- API gateway error management
- Microservices fault tolerance
- Distributed system coordination
    """)


if __name__ == "__main__":
    demonstrate_error_classification_routing()

    

Agentic Design Pattern: Error Classification

Pattern 148: Error Classification – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 148: Error Classification & Routing

Description: Implements intelligent error classification and routing to appropriate handlers based on error type, severity, and context. This pattern categorizes errors and routes them to specialized recovery strategies, enabling more sophisticated error handling than simple try-catch blocks.

Components: – Error Classifier: Categorizes errors by type and severity – Error Router: Routes errors to appropriate handlers – Handler Registry: Maps error types to handlers – Context Analyzer: Extracts error context – Recovery Strategies: Specialized recovery for each error type – Escalation Manager: Escalates unhandled errors

Use Cases: – Complex agent error handling – Multi-service application error management – Production system fault tolerance – Intelligent error recovery – Error analytics and monitoring

Benefits: – Targeted error recovery – Better error handling granularity – Reduced error resolution time – Improved system reliability – Better debugging and monitoring

Trade-offs: – Increased complexity – Requires comprehensive error taxonomy – Handler maintenance overhead – Potential routing errors

LangChain Implementation: Uses error classification, pattern matching, and routing to specialized handlers for intelligent error management.

---

File Statistics

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

---

Module Documentation

"""
Pattern 148: Error Classification & Routing

Description:
    Implements intelligent error classification and routing to appropriate
    handlers based on error type, severity, and context. This pattern categorizes
    errors and routes them to specialized recovery strategies, enabling more
    sophisticated error handling than simple try-catch blocks.

Components:
    - Error Classifier: Categorizes errors by type and severity
    - Error Router: Routes errors to appropriate handlers
    - Handler Registry: Maps error types to handlers
    - Context Analyzer: Extracts error context
    - Recovery Strategies: Specialized recovery for each error type
    - Escalation Manager: Escalates unhandled errors

Use Cases:
    - Complex agent error handling
    - Multi-service application error management
    - Production system fault tolerance
    - Intelligent error recovery
    - Error analytics and monitoring

Benefits:
    - Targeted error recovery
    - Better error handling granularity
    - Reduced error resolution time
    - Improved system reliability
    - Better debugging and monitoring

Trade-offs:
    - Increased complexity
    - Requires comprehensive error taxonomy
    - Handler maintenance overhead
    - Potential routing errors

LangChain Implementation:
    Uses error classification, pattern matching, and routing to specialized
    handlers for intelligent error management.
"""

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

---

Import Statements

Standard Library Imports

import collections.defaultdict
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.Callable
import typing.Dict
import typing.List
import typing.Optional
import typing.Type

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – 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.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.Type: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

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

Other Imports

from dotenv.load_dotenv
from traceback

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ErrorCategory

Lines 60-71

Inherits from: Enum

Purpose:

High-level error categories

---

Class: ErrorSeverity

Lines 74-79

Inherits from: Enum

Purpose:

Error severity levels

---

Class: ErrorContext

Lines 83-93

Purpose:

Context information about an error

---

Class: ClassifiedError

Lines 97-104

Purpose:

Classified error with category and severity

---

Class: ErrorStats

Lines 108-116

Purpose:

Statistics about error handling

---

Class: ErrorClassificationRouter

Lines 119-453

Purpose:

Agent that classifies and routes errors to appropriate handlers.

Analyzes errors, categorizes them, and routes to specialized recovery
strategies for intelligent error management.

Methods:

__init__(self, model)

Lines 127-141

Purpose:

Initialize the error classification router.

Args:
    model: LLM model to use

Parameters: – model: Language model instance

_setup_default_rules(self)

Lines 143-195

Purpose:

Set up default error classification rules.

Parameters:

classify_error(self, error)

Lines 197-234

Purpose:

Classify an error based on type and message.

Args:
    error: Exception to classify
    **context: Additional context

Returns:
    Classified error with category and severity

Parameters: – error: Error

Returns: ClassifiedError

_classify_by_rules(self, context)

Lines 236-259

Purpose:

Classify error using rule-based matching.

Parameters: – context: Context

Returns: ClassifiedError

_classify_by_llm(self, context)

Lines 261-320

Purpose:

Classify error using LLM.

Parameters: – context: Context

Returns: ClassifiedError

register_handler(self, category, handler)

Lines 322-334

Purpose:

Register a handler for an error category.

Args:
    category: Error category
    handler: Handler function

Parameters: – category: Category – handler: Handler

route_error(self, classified_error)

Lines 336-365

Purpose:

Route error to appropriate handler.

Args:
    classified_error: Classified error

Returns:
    Handler result

Parameters: – classified_error: Classified error

Returns: Any

_default_handler(self, classified_error)

Lines 367-379

Purpose:

Default handler for unhandled errors.

Parameters: – classified_error: Classified error

Returns: <ast.Subscript object at 0x000001A06D7CEBD0>

handle_error(self, error)

Lines 381-405

Purpose:

Complete error handling: classify and route.

Args:
    error: Exception to handle
    **context: Additional context

Returns:
    Handler result

Parameters: – error: Error

Returns: Any

get_stats_report(self)

Lines 407-453

Purpose:

Generate error statistics report.

Parameters:

Returns: str

---

Functions

demonstrate_error_classification_routing()

Lines 456-675

Purpose:

Demonstrate the Error Classification & Routing pattern.

---

Key Concepts Summary

Pattern Overview

The Error Classification 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 147
• Pattern 149

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 Error Classification 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: 148_error_classification.py
Total Lines: 680
Pattern Category: AI Agent Design Pattern #148

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 147: Compensating Actions

Description:
    Implements compensating actions for rolling back completed operations when
    a workflow fails. This pattern maintains consistency by undoing successful
    steps in reverse order, similar to database transaction rollback but for
    complex multi-step agent workflows.

Components:
    - Action Logger: Records completed actions
    - Compensation Registry: Maps actions to compensating actions
    - Rollback Coordinator: Orchestrates compensation execution
    - State Tracker: Maintains workflow state
    - Failure Handler: Detects and triggers compensation
    - Recovery Monitor: Tracks compensation success

Use Cases:
    - Multi-step agent workflows with side effects
    - API orchestration with multiple services
    - Data pipeline error recovery
    - Transaction-like behavior for agent operations
    - Distributed system coordination

Benefits:
    - Maintains system consistency
    - Automatic error recovery
    - Clear rollback semantics
    - Reduced manual cleanup
    - Better fault tolerance

Trade-offs:
    - Complexity overhead
    - Not all actions are reversible
    - Potential partial rollback states
    - Resource consumption during compensation

LangChain Implementation:
    Uses action tracking, compensation functions, and rollback orchestration
    to implement saga-like patterns for agent workflows.
"""

import os
import time
from typing import Any, Callable, Optional, List, Dict, Tuple
from dataclasses import dataclass, field
from enum import Enum
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()


class ActionStatus(Enum):
    """Status of an action"""
    PENDING = "pending"
    EXECUTING = "executing"
    COMPLETED = "completed"
    FAILED = "failed"
    COMPENSATING = "compensating"
    COMPENSATED = "compensated"
    COMPENSATION_FAILED = "compensation_failed"


@dataclass
class Action:
    """Represents a workflow action"""
    action_id: str
    name: str
    execute_fn: Callable
    compensate_fn: Optional[Callable] = None
    args: Tuple = field(default_factory=tuple)
    kwargs: Dict = field(default_factory=dict)
    status: ActionStatus = ActionStatus.PENDING
    result: Any = None
    error: Optional[str] = None
    executed_at: Optional[datetime] = None
    compensated_at: Optional[datetime] = None
    idempotent: bool = False


@dataclass
class WorkflowExecution:
    """Tracks workflow execution"""
    workflow_id: str
    actions: List[Action]
    current_index: int = 0
    completed_actions: List[Action] = field(default_factory=list)
    failed_action: Optional[Action] = None
    compensation_results: List[Tuple[Action, bool]] = field(default_factory=list)
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None
    success: bool = False


class CompensatingActionsAgent:
    """
    Agent that implements compensating actions for workflow rollback.
    
    Executes multi-step workflows and automatically compensates completed
    actions when failures occur, maintaining system consistency.
    """
    
    def __init__(self, model: str = "gpt-3.5-turbo"):
        """
        Initialize the compensating actions agent.
        
        Args:
            model: LLM model to use
        """
        self.llm = ChatOpenAI(model=model, temperature=0.7)
        
        self.workflows: Dict[str, WorkflowExecution] = {}
        self.compensation_registry: Dict[str, Callable] = {}
    
    def register_compensation(
        self,
        action_name: str,
        compensate_fn: Callable
    ):
        """
        Register a compensation function for an action type.
        
        Args:
            action_name: Name of the action
            compensate_fn: Function to compensate the action
        """
        self.compensation_registry[action_name] = compensate_fn
    
    def execute_workflow(
        self,
        workflow_id: str,
        actions: List[Action]
    ) -> WorkflowExecution:
        """
        Execute a workflow with compensating action support.
        
        Args:
            workflow_id: Unique workflow identifier
            actions: List of actions to execute
            
        Returns:
            Workflow execution result
        """
        execution = WorkflowExecution(
            workflow_id=workflow_id,
            actions=actions
        )
        
        self.workflows[workflow_id] = execution
        
        print(f"\n🚀 Starting workflow: {workflow_id}")
        print(f"   Total actions: {len(actions)}")
        
        try:
            # Execute actions sequentially
            for i, action in enumerate(actions):
                execution.current_index = i
                
                print(f"\n📍 Executing action {i+1}/{len(actions)}: {action.name}")
                
                action.status = ActionStatus.EXECUTING
                
                try:
                    # Execute the action
                    start_time = time.time()
                    result = action.execute_fn(*action.args, **action.kwargs)
                    duration = time.time() - start_time
                    
                    action.result = result
                    action.status = ActionStatus.COMPLETED
                    action.executed_at = datetime.now()
                    
                    execution.completed_actions.append(action)
                    
                    print(f"   ✅ Completed in {duration:.2f}s")
                    if result:
                        print(f"   Result: {str(result)[:100]}")
                    
                except Exception as e:
                    # Action failed - trigger compensation
                    action.status = ActionStatus.FAILED
                    action.error = str(e)
                    execution.failed_action = action
                    
                    print(f"   ❌ Failed: {e}")
                    print(f"\n🔄 Triggering compensation for {len(execution.completed_actions)} completed actions")
                    
                    self._compensate_actions(execution)
                    
                    execution.end_time = datetime.now()
                    execution.success = False
                    
                    return execution
            
            # All actions completed successfully
            execution.end_time = datetime.now()
            execution.success = True
            
            print(f"\n✅ Workflow completed successfully!")
            
            return execution
            
        except Exception as e:
            print(f"\n❌ Workflow execution error: {e}")
            execution.success = False
            execution.end_time = datetime.now()
            return execution
    
    def _compensate_actions(self, execution: WorkflowExecution):
        """
        Compensate completed actions in reverse order.
        
        Args:
            execution: Workflow execution to compensate
        """
        # Compensate in reverse order
        for action in reversed(execution.completed_actions):
            print(f"\n   🔙 Compensating: {action.name}")
            
            action.status = ActionStatus.COMPENSATING
            
            # Get compensation function
            compensate_fn = action.compensate_fn
            if not compensate_fn and action.name in self.compensation_registry:
                compensate_fn = self.compensation_registry[action.name]
            
            if not compensate_fn:
                print(f"      ⚠️  No compensation function available")
                execution.compensation_results.append((action, False))
                action.status = ActionStatus.COMPENSATION_FAILED
                continue
            
            try:
                # Execute compensation
                compensate_fn(action.result)
                action.status = ActionStatus.COMPENSATED
                action.compensated_at = datetime.now()
                execution.compensation_results.append((action, True))
                
                print(f"      ✅ Compensated successfully")
                
            except Exception as e:
                print(f"      ❌ Compensation failed: {e}")
                action.status = ActionStatus.COMPENSATION_FAILED
                execution.compensation_results.append((action, False))
    
    def get_workflow_status(self, workflow_id: str) -> Optional[WorkflowExecution]:
        """Get status of a workflow execution."""
        return self.workflows.get(workflow_id)
    
    def generate_compensation_report(
        self,
        workflow_id: str
    ) -> str:
        """
        Generate a detailed compensation report.
        
        Args:
            workflow_id: Workflow to report on
            
        Returns:
            Formatted report
        """
        execution = self.workflows.get(workflow_id)
        if not execution:
            return f"Workflow {workflow_id} not found"
        
        report = []
        report.append("\n" + "="*60)
        report.append(f"WORKFLOW COMPENSATION REPORT: {workflow_id}")
        report.append("="*60)
        
        report.append(f"\n📊 Execution Summary:")
        report.append(f"   Total Actions: {len(execution.actions)}")
        report.append(f"   Completed: {len(execution.completed_actions)}")
        report.append(f"   Status: {'✅ SUCCESS' if execution.success else '❌ FAILED'}")
        
        if execution.failed_action:
            report.append(f"\n⚠️  Failed Action:")
            report.append(f"   Name: {execution.failed_action.name}")
            report.append(f"   Error: {execution.failed_action.error}")
            report.append(f"   Position: {execution.current_index + 1}/{len(execution.actions)}")
        
        if execution.compensation_results:
            report.append(f"\n🔄 Compensation Results:")
            successful = sum(1 for _, success in execution.compensation_results if success)
            report.append(f"   Total Compensations: {len(execution.compensation_results)}")
            report.append(f"   Successful: {successful}")
            report.append(f"   Failed: {len(execution.compensation_results) - successful}")
            
            report.append(f"\n   Details:")
            for action, success in execution.compensation_results:
                status = "✅" if success else "❌"
                report.append(f"      {status} {action.name}")
        
        if execution.end_time:
            duration = (execution.end_time - execution.start_time).total_seconds()
            report.append(f"\n⏱️  Timing:")
            report.append(f"   Duration: {duration:.2f}s")
        
        report.append("\n" + "="*60)
        
        return "\n".join(report)


def demonstrate_compensating_actions():
    """Demonstrate the Compensating Actions pattern."""
    print("="*60)
    print("COMPENSATING ACTIONS PATTERN DEMONSTRATION")
    print("="*60)
    
    agent = CompensatingActionsAgent()
    
    # Example 1: Simple workflow with compensation
    print("\n" + "="*60)
    print("Example 1: Simple Workflow with Successful Compensation")
    print("="*60)
    
    # Simulated state
    state = {
        "user_created": False,
        "email_sent": False,
        "database_updated": False
    }
    
    def create_user(user_id: str) -> str:
        """Create a user account."""
        print(f"      Creating user: {user_id}")
        state["user_created"] = True
        return f"User {user_id} created"
    
    def compensate_create_user(result: str):
        """Delete the created user."""
        print(f"      Deleting user from result: {result}")
        state["user_created"] = False
    
    def send_email(email: str) -> str:
        """Send welcome email."""
        print(f"      Sending email to: {email}")
        state["email_sent"] = True
        return f"Email sent to {email}"
    
    def compensate_send_email(result: str):
        """Send cancellation email."""
        print(f"      Sending cancellation email")
        state["email_sent"] = False
    
    def update_database(data: dict) -> str:
        """Update database - THIS WILL FAIL."""
        print(f"      Updating database with: {data}")
        raise Exception("Database connection failed")
    
    # Create workflow
    actions = [
        Action(
            action_id="1",
            name="create_user",
            execute_fn=create_user,
            compensate_fn=compensate_create_user,
            args=("user123",)
        ),
        Action(
            action_id="2",
            name="send_email",
            execute_fn=send_email,
            compensate_fn=compensate_send_email,
            args=("user@example.com",)
        ),
        Action(
            action_id="3",
            name="update_database",
            execute_fn=update_database,
            args=({"user": "user123"},)
        )
    ]
    
    execution = agent.execute_workflow("workflow-001", actions)
    
    print(f"\nFinal State:")
    print(f"   User Created: {state['user_created']}")
    print(f"   Email Sent: {state['email_sent']}")
    print(f"   Database Updated: {state['database_updated']}")
    
    report = agent.generate_compensation_report("workflow-001")
    print(report)
    
    # Example 2: Multi-step API workflow
    print("\n" + "="*60)
    print("Example 2: Multi-Step API Workflow")
    print("="*60)
    
    api_state = {
        "payment_id": None,
        "inventory_reserved": False,
        "shipping_scheduled": False
    }
    
    def process_payment(amount: float) -> str:
        """Process payment."""
        payment_id = f"PAY-{int(time.time())}"
        api_state["payment_id"] = payment_id
        print(f"      Payment processed: {payment_id} for ${amount}")
        return payment_id
    
    def compensate_payment(payment_id: str):
        """Refund payment."""
        print(f"      Refunding payment: {payment_id}")
        api_state["payment_id"] = None
    
    def reserve_inventory(item_id: str, quantity: int) -> str:
        """Reserve inventory."""
        api_state["inventory_reserved"] = True
        print(f"      Reserved {quantity} units of {item_id}")
        return f"Reserved {quantity} units"
    
    def compensate_inventory(result: str):
        """Release inventory reservation."""
        print(f"      Releasing inventory reservation")
        api_state["inventory_reserved"] = False
    
    def schedule_shipping(address: str) -> str:
        """Schedule shipping - WILL FAIL."""
        print(f"      Scheduling shipping to: {address}")
        raise Exception("Shipping service unavailable")
    
    actions = [
        Action(
            action_id="1",
            name="process_payment",
            execute_fn=process_payment,
            compensate_fn=compensate_payment,
            args=(99.99,)
        ),
        Action(
            action_id="2",
            name="reserve_inventory",
            execute_fn=reserve_inventory,
            compensate_fn=compensate_inventory,
            args=("ITEM-001", 2)
        ),
        Action(
            action_id="3",
            name="schedule_shipping",
            execute_fn=schedule_shipping,
            args=("123 Main St",)
        )
    ]
    
    execution = agent.execute_workflow("workflow-002", actions)
    
    print(f"\nFinal API State:")
    print(f"   Payment ID: {api_state['payment_id']}")
    print(f"   Inventory Reserved: {api_state['inventory_reserved']}")
    print(f"   Shipping Scheduled: {api_state['shipping_scheduled']}")
    
    # Example 3: Successful workflow (no compensation needed)
    print("\n" + "="*60)
    print("Example 3: Successful Workflow (No Compensation)")
    print("="*60)
    
    success_state = {"steps": []}
    
    def step1() -> str:
        success_state["steps"].append("step1")
        return "Step 1 completed"
    
    def step2() -> str:
        success_state["steps"].append("step2")
        return "Step 2 completed"
    
    def step3() -> str:
        success_state["steps"].append("step3")
        return "Step 3 completed"
    
    actions = [
        Action("1", "step1", step1),
        Action("2", "step2", step2),
        Action("3", "step3", step3)
    ]
    
    execution = agent.execute_workflow("workflow-003", actions)
    
    print(f"\nCompleted Steps: {success_state['steps']}")
    print(f"Workflow Success: {execution.success}")
    
    # Example 4: Compensation with LLM decision
    print("\n" + "="*60)
    print("Example 4: LLM-Assisted Compensation Decision")
    print("="*60)
    
    def should_compensate_action(action: Action, error: str) -> bool:
        """Use LLM to decide if action should be compensated."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in distributed systems. "
                      "Determine if an action should be compensated (undone) "
                      "based on the error. Reply with ONLY 'yes' or 'no'."),
            ("user", "Action: {action}\nError: {error}\n\n"
                    "Should this action be compensated?")
        ])
        
        chain = prompt | agent.llm | StrOutputParser()
        result = chain.invoke({
            "action": action.name,
            "error": error
        })
        
        return "yes" in result.lower()
    
    # Simulate a workflow with LLM-guided compensation
    test_action = Action(
        action_id="test",
        name="send_notification",
        execute_fn=lambda: "Notification sent",
        compensate_fn=lambda x: print("Notification cancelled")
    )
    test_action.result = "Notification sent"
    
    test_error = "Network timeout after 30 seconds"
    
    print(f"\nAsking LLM if '{test_action.name}' should be compensated")
    print(f"Error: {test_error}")
    
    should_compensate = should_compensate_action(test_action, test_error)
    print(f"\nLLM Decision: {'Compensate' if should_compensate else 'Do not compensate'}")
    
    # Example 5: Idempotent actions
    print("\n" + "="*60)
    print("Example 5: Idempotent Actions (Safe to Retry)")
    print("="*60)
    
    idempotent_state = {"writes": 0}
    
    def idempotent_write(key: str, value: str) -> str:
        """Idempotent write operation."""
        idempotent_state["writes"] += 1
        print(f"      Write #{idempotent_state['writes']}: {key} = {value}")
        return f"Set {key} = {value}"
    
    # Idempotent actions don't need compensation
    idempotent_action = Action(
        action_id="idem",
        name="idempotent_write",
        execute_fn=idempotent_write,
        args=("config", "value123"),
        idempotent=True
    )
    
    print("Executing idempotent action multiple times:")
    for i in range(3):
        result = idempotent_action.execute_fn(*idempotent_action.args)
        print(f"   Attempt {i+1}: {result}")
    
    print(f"\n💡 Idempotent actions can be safely retried without compensation")
    print(f"   Total writes: {idempotent_state['writes']}")
    
    # Summary
    print("\n" + "="*60)
    print("SUMMARY")
    print("="*60)
    print("""
The Compensating Actions pattern demonstrates:

1. Automatic Rollback: Undoes completed actions when workflow fails
2. Reverse Order: Compensates actions in reverse of execution
3. State Consistency: Maintains system consistency after failures
4. Saga Pattern: Implements saga-like transaction semantics
5. Flexible Compensation: Supports custom compensation logic

Key Benefits:
- Maintains consistency in distributed workflows
- Automatic error recovery
- Clear rollback semantics
- Reduces manual cleanup effort
- Better fault tolerance

Compensation Strategies:
- Reverse Order: Compensate from last to first (recommended)
- Parallel: Compensate all actions simultaneously (faster)
- Selective: Compensate only specific actions (intelligent)
- Best Effort: Continue compensation even if some fail

Best Practices:
- Design compensating actions alongside forward actions
- Make compensation idempotent when possible
- Log all compensation attempts
- Handle compensation failures gracefully
- Test compensation logic thoroughly
- Consider partial compensation scenarios
- Document non-compensable actions

Idempotent Actions:
- Can be safely retried without side effects
- Don't require compensation
- Examples: PUT operations, SET commands
- Mark actions as idempotent in workflow

Use Cases:
- Multi-service API orchestration
- Database transaction-like workflows
- Payment and order processing
- Resource provisioning
- Data pipeline error recovery

Common Patterns:
- Forward Progress + Compensation (Saga)
- Try-Confirm-Cancel (TCC)
- Two-Phase Commit alternative
- Eventual consistency patterns
    """)


if __name__ == "__main__":
    demonstrate_compensating_actions()

    

Agentic Design Pattern: Compensating Actions

Pattern 147: Compensating Actions – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 147: Compensating Actions

Description: Implements compensating actions for rolling back completed operations when a workflow fails. This pattern maintains consistency by undoing successful steps in reverse order, similar to database transaction rollback but for complex multi-step agent workflows.

Components: – Action Logger: Records completed actions – Compensation Registry: Maps actions to compensating actions – Rollback Coordinator: Orchestrates compensation execution – State Tracker: Maintains workflow state – Failure Handler: Detects and triggers compensation – Recovery Monitor: Tracks compensation success

Use Cases: – Multi-step agent workflows with side effects – API orchestration with multiple services – Data pipeline error recovery – Transaction-like behavior for agent operations – Distributed system coordination

Benefits: – Maintains system consistency – Automatic error recovery – Clear rollback semantics – Reduced manual cleanup – Better fault tolerance

Trade-offs: – Complexity overhead – Not all actions are reversible – Potential partial rollback states – Resource consumption during compensation

LangChain Implementation: Uses action tracking, compensation functions, and rollback orchestration to implement saga-like patterns for agent workflows.

---

File Statistics

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

---

Module Documentation

"""
Pattern 147: Compensating Actions

Description:
    Implements compensating actions for rolling back completed operations when
    a workflow fails. This pattern maintains consistency by undoing successful
    steps in reverse order, similar to database transaction rollback but for
    complex multi-step agent workflows.

Components:
    - Action Logger: Records completed actions
    - Compensation Registry: Maps actions to compensating actions
    - Rollback Coordinator: Orchestrates compensation execution
    - State Tracker: Maintains workflow state
    - Failure Handler: Detects and triggers compensation
    - Recovery Monitor: Tracks compensation success

Use Cases:
    - Multi-step agent workflows with side effects
    - API orchestration with multiple services
    - Data pipeline error recovery
    - Transaction-like behavior for agent operations
    - Distributed system coordination

Benefits:
    - Maintains system consistency
    - Automatic error recovery
    - Clear rollback semantics
    - Reduced manual cleanup
    - Better fault tolerance

Trade-offs:
    - Complexity overhead
    - Not all actions are reversible
    - Potential partial rollback states
    - Resource consumption during compensation

LangChain Implementation:
    Uses action tracking, compensation functions, and rollback orchestration
    to implement saga-like patterns for agent workflows.
"""

Purpose: This module implements the Compensating Actions 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 time
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 – 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.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: ActionStatus

Lines 58-66

Inherits from: Enum

Purpose:

Status of an action

---

Class: Action

Lines 70-83

Purpose:

Represents a workflow action

---

Class: WorkflowExecution

Lines 87-97

Purpose:

Tracks workflow execution

---

Class: CompensatingActionsAgent

Lines 100-308

Purpose:

Agent that implements compensating actions for workflow rollback.

Executes multi-step workflows and automatically compensates completed
actions when failures occur, maintaining system consistency.

Methods:

__init__(self, model)

Lines 108-118

Purpose:

Initialize the compensating actions agent.

Args:
    model: LLM model to use

Parameters: – model: Language model instance

register_compensation(self, action_name, compensate_fn)

Lines 120-132

Purpose:

Register a compensation function for an action type.

Args:
    action_name: Name of the action
    compensate_fn: Function to compensate the action

Parameters: – action_name: Action name – compensate_fn: Compensate fn

execute_workflow(self, workflow_id, actions)

Lines 134-212

Purpose:

Execute a workflow with compensating action support.

Args:
    workflow_id: Unique workflow identifier
    actions: List of actions to execute

Returns:
    Workflow execution result

Parameters: – workflow_id: Workflow id – actions: Actions

Returns: WorkflowExecution

_compensate_actions(self, execution)

Lines 214-250

Purpose:

Compensate completed actions in reverse order.

Args:
    execution: Workflow execution to compensate

Parameters: – execution: Execution

get_workflow_status(self, workflow_id)

Lines 252-254

Purpose:

Get status of a workflow execution.

Parameters: – workflow_id: Workflow id

Returns: <ast.Subscript object at 0x000001A06D7C9CD0>

generate_compensation_report(self, workflow_id)

Lines 256-308

Purpose:

Generate a detailed compensation report.

Args:
    workflow_id: Workflow to report on

Returns:
    Formatted report

Parameters: – workflow_id: Workflow id

Returns: str

---

Functions

demonstrate_compensating_actions()

Lines 311-614

Purpose:

Demonstrate the Compensating Actions pattern.

---

Key Concepts Summary

Pattern Overview

The Compensating Actions 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 146
• Pattern 148

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 Compensating Actions 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: 147_compensating_actions.py
Total Lines: 619
Pattern Category: AI Agent Design Pattern #147

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 146: Retry with Backoff

Description:
    Implements retry logic with exponential backoff for handling transient failures
    in LLM API calls and agent operations. This pattern automatically retries failed
    operations with increasing delays, helping systems recover from temporary issues
    like rate limits, network errors, or service disruptions.

Components:
    - Retry Manager: Orchestrates retry attempts
    - Backoff Strategy: Calculates delay between retries
    - Failure Classifier: Determines if error is retryable
    - Success Monitor: Tracks retry statistics
    - Circuit Breaker Integration: Prevents endless retries
    - Fallback Handler: Provides alternatives when retries exhausted

Use Cases:
    - Handle API rate limiting
    - Recover from transient network errors
    - Manage service unavailability
    - Deal with timeout errors
    - Handle overloaded systems

Benefits:
    - Automatic error recovery
    - Reduced manual intervention
    - Improved reliability
    - Better user experience
    - Graceful handling of transient failures

Trade-offs:
    - Increased latency on failures
    - Resource consumption during retries
    - Potential for cascading delays
    - Complexity in retry logic

LangChain Implementation:
    Uses decorators and custom retry logic with exponential backoff,
    jitter, and intelligent failure classification.
"""

import os
import time
import random
from typing import Any, Callable, Optional, List, Dict
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from functools import wraps
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 BackoffStrategy(Enum):
    """Backoff strategy for retries"""
    FIXED = "fixed"
    LINEAR = "linear"
    EXPONENTIAL = "exponential"
    FIBONACCI = "fibonacci"


class ErrorType(Enum):
    """Types of errors for classification"""
    TRANSIENT = "transient"  # Temporary, retryable
    PERMANENT = "permanent"  # Persistent, not retryable
    RATE_LIMIT = "rate_limit"  # Rate limiting
    TIMEOUT = "timeout"  # Timeout error
    UNKNOWN = "unknown"  # Unknown error type


@dataclass
class RetryConfig:
    """Configuration for retry behavior"""
    max_attempts: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    backoff_strategy: BackoffStrategy = BackoffStrategy.EXPONENTIAL
    backoff_multiplier: float = 2.0
    jitter: bool = True
    jitter_range: float = 0.1
    retryable_errors: List[str] = field(default_factory=lambda: [
        "rate_limit",
        "timeout",
        "connection",
        "service_unavailable"
    ])


@dataclass
class RetryAttempt:
    """Information about a retry attempt"""
    attempt_number: int
    error_type: ErrorType
    error_message: str
    delay_before: float
    timestamp: datetime
    success: bool


@dataclass
class RetryMetrics:
    """Metrics for retry operations"""
    total_calls: int = 0
    successful_first_try: int = 0
    successful_after_retry: int = 0
    failed_after_retries: int = 0
    total_retries: int = 0
    total_delay_time: float = 0.0
    average_attempts: float = 0.0
    retry_success_rate: float = 0.0


class RetryWithBackoffAgent:
    """
    Agent that implements retry logic with exponential backoff.
    
    Automatically retries failed operations with intelligent backoff
    strategies and error classification.
    """
    
    def __init__(
        self,
        model: str = "gpt-3.5-turbo",
        config: Optional[RetryConfig] = None
    ):
        """
        Initialize the retry agent.
        
        Args:
            model: LLM model to use
            config: Retry configuration
        """
        self.llm = ChatOpenAI(model=model, temperature=0.7)
        self.config = config or RetryConfig()
        
        self.metrics = RetryMetrics()
        self.retry_history: List[RetryAttempt] = []
    
    def calculate_delay(self, attempt: int) -> float:
        """
        Calculate delay for retry attempt based on strategy.
        
        Args:
            attempt: Current attempt number (0-indexed)
            
        Returns:
            Delay in seconds
        """
        if self.config.backoff_strategy == BackoffStrategy.FIXED:
            delay = self.config.initial_delay
            
        elif self.config.backoff_strategy == BackoffStrategy.LINEAR:
            delay = self.config.initial_delay * (attempt + 1)
            
        elif self.config.backoff_strategy == BackoffStrategy.EXPONENTIAL:
            delay = self.config.initial_delay * (
                self.config.backoff_multiplier ** attempt
            )
            
        elif self.config.backoff_strategy == BackoffStrategy.FIBONACCI:
            delay = self.config.initial_delay * self._fibonacci(attempt + 1)
        
        else:
            delay = self.config.initial_delay
        
        # Cap at max delay
        delay = min(delay, self.config.max_delay)
        
        # Add jitter if enabled
        if self.config.jitter:
            jitter_amount = delay * self.config.jitter_range
            delay += random.uniform(-jitter_amount, jitter_amount)
        
        return max(0, delay)  # Ensure non-negative
    
    def _fibonacci(self, n: int) -> int:
        """Calculate nth Fibonacci number."""
        if n  ErrorType:
        """
        Classify error to determine if retryable.
        
        Args:
            error: Exception to classify
            
        Returns:
            Error type classification
        """
        error_str = str(error).lower()
        
        if any(keyword in error_str for keyword in ["rate limit", "429"]):
            return ErrorType.RATE_LIMIT
        
        elif any(keyword in error_str for keyword in ["timeout", "timed out"]):
            return ErrorType.TIMEOUT
        
        elif any(keyword in error_str for keyword in [
            "connection", "network", "unavailable", "503", "502"
        ]):
            return ErrorType.TRANSIENT
        
        elif any(keyword in error_str for keyword in [
            "invalid", "authentication", "authorization", "400", "401", "403"
        ]):
            return ErrorType.PERMANENT
        
        else:
            return ErrorType.UNKNOWN
    
    def is_retryable(self, error: Exception) -> bool:
        """
        Determine if error should be retried.
        
        Args:
            error: Exception to check
            
        Returns:
            True if error is retryable
        """
        error_type = self.classify_error(error)
        
        if error_type == ErrorType.PERMANENT:
            return False
        
        error_str = str(error).lower()
        return any(
            keyword in error_str
            for keyword in self.config.retryable_errors
        )
    
    def retry_with_backoff(
        self,
        func: Callable,
        *args,
        **kwargs
    ) -> Any:
        """
        Execute function with retry logic.
        
        Args:
            func: Function to execute
            *args: Positional arguments
            **kwargs: Keyword arguments
            
        Returns:
            Function result
            
        Raises:
            Exception: If all retries exhausted
        """
        self.metrics.total_calls += 1
        attempts = []
        
        for attempt in range(self.config.max_attempts):
            try:
                result = func(*args, **kwargs)
                
                # Record success
                if attempt == 0:
                    self.metrics.successful_first_try += 1
                else:
                    self.metrics.successful_after_retry += 1
                    self.metrics.total_retries += attempt
                
                # Record attempt
                attempts.append(RetryAttempt(
                    attempt_number=attempt + 1,
                    error_type=ErrorType.TRANSIENT,
                    error_message="Success",
                    delay_before=0,
                    timestamp=datetime.now(),
                    success=True
                ))
                
                self.retry_history.extend(attempts)
                self._update_metrics()
                
                return result
                
            except Exception as e:
                error_type = self.classify_error(e)
                
                # Check if should retry
                if not self.is_retryable(e):
                    print(f"⛔ Non-retryable error: {error_type.value}")
                    self.metrics.failed_after_retries += 1
                    raise
                
                # Last attempt - don't retry
                if attempt == self.config.max_attempts - 1:
                    print(f"❌ Max retries ({self.config.max_attempts}) exhausted")
                    self.metrics.failed_after_retries += 1
                    
                    attempts.append(RetryAttempt(
                        attempt_number=attempt + 1,
                        error_type=error_type,
                        error_message=str(e),
                        delay_before=0,
                        timestamp=datetime.now(),
                        success=False
                    ))
                    
                    self.retry_history.extend(attempts)
                    self._update_metrics()
                    raise
                
                # Calculate delay and retry
                delay = self.calculate_delay(attempt)
                
                print(f"⚠️  Attempt {attempt + 1} failed: {error_type.value}")
                print(f"   Retrying in {delay:.2f}s...")
                
                attempts.append(RetryAttempt(
                    attempt_number=attempt + 1,
                    error_type=error_type,
                    error_message=str(e),
                    delay_before=delay,
                    timestamp=datetime.now(),
                    success=False
                ))
                
                self.metrics.total_delay_time += delay
                time.sleep(delay)
    
    def _update_metrics(self):
        """Update calculated metrics."""
        total_operations = (
            self.metrics.successful_first_try +
            self.metrics.successful_after_retry +
            self.metrics.failed_after_retries
        )
        
        if total_operations > 0:
            self.metrics.average_attempts = (
                (self.metrics.successful_first_try +
                 self.metrics.successful_after_retry +
                 self.metrics.total_retries) / total_operations
            )
        
        retry_operations = (
            self.metrics.successful_after_retry +
            self.metrics.failed_after_retries
        )
        
        if retry_operations > 0:
            self.metrics.retry_success_rate = (
                self.metrics.successful_after_retry / retry_operations
            )
    
    def query_with_retry(self, query: str) -> str:
        """
        Query LLM with retry logic.
        
        Args:
            query: Query string
            
        Returns:
            LLM response
        """
        def _query():
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a helpful assistant."),
                ("user", "{query}")
            ])
            chain = prompt | self.llm | StrOutputParser()
            return chain.invoke({"query": query})
        
        return self.retry_with_backoff(_query)
    
    def get_metrics_report(self) -> str:
        """Generate retry metrics report."""
        report = []
        report.append("\n" + "="*60)
        report.append("RETRY WITH BACKOFF METRICS REPORT")
        report.append("="*60)
        
        report.append(f"\n📊 Overall Statistics:")
        report.append(f"   Total Calls: {self.metrics.total_calls}")
        report.append(f"   Successful First Try: {self.metrics.successful_first_try}")
        report.append(f"   Successful After Retry: {self.metrics.successful_after_retry}")
        report.append(f"   Failed After Retries: {self.metrics.failed_after_retries}")
        
        report.append(f"\n🔄 Retry Statistics:")
        report.append(f"   Total Retries: {self.metrics.total_retries}")
        report.append(f"   Average Attempts: {self.metrics.average_attempts:.2f}")
        report.append(f"   Retry Success Rate: {self.metrics.retry_success_rate:.1%}")
        
        report.append(f"\n⏱️  Timing:")
        report.append(f"   Total Delay Time: {self.metrics.total_delay_time:.2f}s")
        
        if self.metrics.total_calls > 0:
            success_rate = (
                (self.metrics.successful_first_try + self.metrics.successful_after_retry) /
                self.metrics.total_calls
            )
            report.append(f"\n✅ Overall Success Rate: {success_rate:.1%}")
        
        report.append("\n" + "="*60)
        
        return "\n".join(report)


def retry_decorator(config: Optional[RetryConfig] = None):
    """
    Decorator for automatic retry with backoff.
    
    Args:
        config: Retry configuration
        
    Returns:
        Decorated function
    """
    if config is None:
        config = RetryConfig()
    
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            agent = RetryWithBackoffAgent(config=config)
            return agent.retry_with_backoff(func, *args, **kwargs)
        return wrapper
    return decorator


def demonstrate_retry_with_backoff():
    """Demonstrate the Retry with Backoff pattern."""
    print("="*60)
    print("RETRY WITH BACKOFF PATTERN DEMONSTRATION")
    print("="*60)
    
    # Example 1: Basic retry with exponential backoff
    print("\n" + "="*60)
    print("Example 1: Basic Retry with Exponential Backoff")
    print("="*60)
    
    config = RetryConfig(
        max_attempts=4,
        initial_delay=1.0,
        backoff_strategy=BackoffStrategy.EXPONENTIAL,
        backoff_multiplier=2.0
    )
    
    agent = RetryWithBackoffAgent(config=config)
    
    print("\nDelay progression for exponential backoff:")
    for i in range(4):
        delay = agent.calculate_delay(i)
        print(f"  Attempt {i+1}: {delay:.2f}s delay")
    
    # Example 2: Compare backoff strategies
    print("\n" + "="*60)
    print("Example 2: Compare Backoff Strategies")
    print("="*60)
    
    strategies = [
        BackoffStrategy.FIXED,
        BackoffStrategy.LINEAR,
        BackoffStrategy.EXPONENTIAL,
        BackoffStrategy.FIBONACCI
    ]
    
    print(f"\n{'Strategy':<15} {'Attempt 1':<12} {'Attempt 2':<12} {'Attempt 3':<12} {'Attempt 4':<12}")
    print("-" * 63)
    
    for strategy in strategies:
        config = RetryConfig(
            initial_delay=1.0,
            backoff_strategy=strategy,
            backoff_multiplier=2.0,
            jitter=False
        )
        agent = RetryWithBackoffAgent(config=config)
        
        delays = [agent.calculate_delay(i) for i in range(4)]
        print(f"{strategy.value:<15} {delays[0]:<12.2f} {delays[1]:<12.2f} {delays[2]:<12.2f} {delays[3]:<12.2f}")
    
    # Example 3: Error classification
    print("\n" + "="*60)
    print("Example 3: Error Classification")
    print("="*60)
    
    agent = RetryWithBackoffAgent()
    
    test_errors = [
        Exception("Rate limit exceeded (429)"),
        Exception("Connection timeout"),
        Exception("Service unavailable (503)"),
        Exception("Invalid API key (401)"),
        Exception("Network connection failed"),
        Exception("Unknown error occurred")
    ]
    
    print("\nError classification results:")
    for error in test_errors:
        error_type = agent.classify_error(error)
        is_retryable = agent.is_retryable(error)
        status = "✅ RETRY" if is_retryable else "⛔ DON'T RETRY"
        print(f"  {status:<15} {error_type.value:<15} {str(error)[:40]}")
    
    # Example 4: Simulate retry scenarios
    print("\n" + "="*60)
    print("Example 4: Simulate Retry Scenarios")
    print("="*60)
    
    # Simulate transient failure that succeeds on retry
    call_count = {"count": 0}
    
    def flaky_function():
        """Function that fails first 2 times, then succeeds."""
        call_count["count"] += 1
        if call_count["count"] < 3:
            raise Exception("Service temporarily unavailable (503)")
        return "Success!"
    
    config = RetryConfig(max_attempts=5, initial_delay=0.5)
    agent = RetryWithBackoffAgent(config=config)
    
    print("\nSimulating transient failure (succeeds on 3rd attempt):")
    try:
        result = agent.retry_with_backoff(flaky_function)
        print(f"✅ Result: {result}")
    except Exception as e:
        print(f"❌ Failed: {e}")
    
    # Example 5: Jitter demonstration
    print("\n" + "="*60)
    print("Example 5: Jitter for Thundering Herd Prevention")
    print("="*60)
    
    config_no_jitter = RetryConfig(
        initial_delay=1.0,
        backoff_strategy=BackoffStrategy.EXPONENTIAL,
        jitter=False
    )
    
    config_with_jitter = RetryConfig(
        initial_delay=1.0,
        backoff_strategy=BackoffStrategy.EXPONENTIAL,
        jitter=True,
        jitter_range=0.2
    )
    
    agent_no_jitter = RetryWithBackoffAgent(config=config_no_jitter)
    agent_with_jitter = RetryWithBackoffAgent(config=config_with_jitter)
    
    print("\nDelay comparison (5 simulated clients):")
    print(f"\n{'Attempt':<10} {'No Jitter':<40} {'With Jitter':<40}")
    print("-" * 90)
    
    for attempt in range(3):
        no_jitter_delays = [agent_no_jitter.calculate_delay(attempt) for _ in range(5)]
        with_jitter_delays = [agent_with_jitter.calculate_delay(attempt) for _ in range(5)]
        
        no_jitter_str = ", ".join(f"{d:.2f}s" for d in no_jitter_delays)
        with_jitter_str = ", ".join(f"{d:.2f}s" for d in with_jitter_delays)
        
        print(f"{attempt+1:<10} {no_jitter_str:<40} {with_jitter_str:<40}")
    
    print("\n💡 Jitter spreads retry attempts, preventing thundering herd")
    
    # Example 6: Real LLM query with retry
    print("\n" + "="*60)
    print("Example 6: LLM Query with Retry Protection")
    print("="*60)
    
    config = RetryConfig(
        max_attempts=3,
        initial_delay=1.0,
        backoff_strategy=BackoffStrategy.EXPONENTIAL
    )
    
    agent = RetryWithBackoffAgent(config=config)
    
    try:
        print("\nQuerying LLM with retry protection...")
        result = agent.query_with_retry(
            "What is the capital of France? Answer in one word."
        )
        print(f"✅ Response: {result}")
    except Exception as e:
        print(f"❌ Failed after all retries: {e}")
    
    # Example 7: Metrics and reporting
    print("\n" + "="*60)
    print("Example 7: Retry Metrics and Reporting")
    print("="*60)
    
    # Simulate multiple operations
    agent = RetryWithBackoffAgent(config=RetryConfig(max_attempts=3))
    
    test_queries = [
        "What is 2+2?",
        "Explain quantum computing in one sentence",
        "Name a primary color"
    ]
    
    print("\nExecuting multiple queries with retry protection...")
    for query in test_queries:
        try:
            agent.query_with_retry(query)
        except Exception as e:
            print(f"Query failed: {e}")
    
    report = agent.get_metrics_report()
    print(report)
    
    # Summary
    print("\n" + "="*60)
    print("SUMMARY")
    print("="*60)
    print("""
The Retry with Backoff pattern demonstrates:

1. Automatic Recovery: Handles transient failures automatically
2. Multiple Strategies: Fixed, linear, exponential, Fibonacci backoff
3. Error Classification: Intelligent distinction between retryable/non-retryable
4. Jitter Support: Prevents thundering herd problem
5. Metrics Tracking: Comprehensive monitoring of retry behavior

Key Benefits:
- Improved reliability (often 95%+ success rate)
- Reduced manual intervention
- Better user experience
- Graceful handling of rate limits
- Protection against cascading failures

Backoff Strategies:
- Fixed: Same delay each time (simple, predictable)
- Linear: Delay increases linearly (gradual backoff)
- Exponential: Delay doubles each time (recommended)
- Fibonacci: Delay follows Fibonacci sequence (balanced)

Best Practices:
- Use exponential backoff as default
- Add jitter to prevent synchronized retries
- Set reasonable max attempts (3-5 typical)
- Cap maximum delay (60s common)
- Classify errors correctly
- Log retry attempts for debugging
- Integrate with circuit breaker
- Monitor retry metrics

Common Configurations:
- API calls: 3 retries, exponential, 1s initial
- Database: 5 retries, exponential, 0.5s initial
- Network: 4 retries, linear, 2s initial
- Rate limits: Fibonacci, respect Retry-After header
    """)


if __name__ == "__main__":
    demonstrate_retry_with_backoff()

    

Agentic Design Pattern: Retry Backoff

Pattern 146: Retry Backoff – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 146: Retry with Backoff

Description: Implements retry logic with exponential backoff for handling transient failures in LLM API calls and agent operations. This pattern automatically retries failed operations with increasing delays, helping systems recover from temporary issues like rate limits, network errors, or service disruptions.

Components: – Retry Manager: Orchestrates retry attempts – Backoff Strategy: Calculates delay between retries – Failure Classifier: Determines if error is retryable – Success Monitor: Tracks retry statistics – Circuit Breaker Integration: Prevents endless retries – Fallback Handler: Provides alternatives when retries exhausted

Use Cases: – Handle API rate limiting – Recover from transient network errors – Manage service unavailability – Deal with timeout errors – Handle overloaded systems

Benefits: – Automatic error recovery – Reduced manual intervention – Improved reliability – Better user experience – Graceful handling of transient failures

Trade-offs: – Increased latency on failures – Resource consumption during retries – Potential for cascading delays – Complexity in retry logic

LangChain Implementation: Uses decorators and custom retry logic with exponential backoff, jitter, and intelligent failure classification.

---

File Statistics

• Total Lines: 665
• Classes: 6
• Functions: 2
• Constants: 0

---

Module Documentation

"""
Pattern 146: Retry with Backoff

Description:
    Implements retry logic with exponential backoff for handling transient failures
    in LLM API calls and agent operations. This pattern automatically retries failed
    operations with increasing delays, helping systems recover from temporary issues
    like rate limits, network errors, or service disruptions.

Components:
    - Retry Manager: Orchestrates retry attempts
    - Backoff Strategy: Calculates delay between retries
    - Failure Classifier: Determines if error is retryable
    - Success Monitor: Tracks retry statistics
    - Circuit Breaker Integration: Prevents endless retries
    - Fallback Handler: Provides alternatives when retries exhausted

Use Cases:
    - Handle API rate limiting
    - Recover from transient network errors
    - Manage service unavailability
    - Deal with timeout errors
    - Handle overloaded systems

Benefits:
    - Automatic error recovery
    - Reduced manual intervention
    - Improved reliability
    - Better user experience
    - Graceful handling of transient failures

Trade-offs:
    - Increased latency on failures
    - Resource consumption during retries
    - Potential for cascading delays
    - Complexity in retry logic

LangChain Implementation:
    Uses decorators and custom retry logic with exponential backoff,
    jitter, and intelligent failure classification.
"""

Purpose: This module implements the Retry Backoff 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 functools.wraps
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import random
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 – enum.Enum: Enumeration support – functools.wraps: Higher-order functions and operations on callable objects – 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 – 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: BackoffStrategy

Lines 60-65

Inherits from: Enum

Purpose:

Backoff strategy for retries

---

Class: ErrorType

Lines 68-74

Inherits from: Enum

Purpose:

Types of errors for classification

---

Class: RetryConfig

Lines 78-92

Purpose:

Configuration for retry behavior

---

Class: RetryAttempt

Lines 96-103

Purpose:

Information about a retry attempt

---

Class: RetryMetrics

Lines 107-116

Purpose:

Metrics for retry operations

---

Class: RetryWithBackoffAgent

Lines 119-412

Purpose:

Agent that implements retry logic with exponential backoff.

Automatically retries failed operations with intelligent backoff
strategies and error classification.

Methods:

__init__(self, model, config)

Lines 127-143

Purpose:

Initialize the retry agent.

Args:
    model: LLM model to use
    config: Retry configuration

Parameters: – model: Language model instance – config: Config

calculate_delay(self, attempt)

Lines 145-180

Purpose:

Calculate delay for retry attempt based on strategy.

Args:
    attempt: Current attempt number (0-indexed)

Returns:
    Delay in seconds

Parameters: – attempt: Attempt

Returns: float

_fibonacci(self, n)

Lines 182-189

Purpose:

Calculate nth Fibonacci number.

Parameters: – n: N

Returns: int

classify_error(self, error)

Lines 191-220

Purpose:

Classify error to determine if retryable.

Args:
    error: Exception to classify

Returns:
    Error type classification

Parameters: – error: Error

Returns: ErrorType

is_retryable(self, error)

Lines 222-241

Purpose:

Determine if error should be retried.

Args:
    error: Exception to check

Returns:
    True if error is retryable

Parameters: – error: Error

Returns: bool

retry_with_backoff(self, func)

Lines 243-335

Purpose:

Execute function with retry logic.

Args:
    func: Function to execute
    *args: Positional arguments
    **kwargs: Keyword arguments

Returns:
    Function result

Raises:
    Exception: If all retries exhausted

Parameters: – func: Func

Returns: Any

_update_metrics(self)

Lines 337-360

Purpose:

Update calculated metrics.

Parameters:

query_with_retry(self, query)

Lines 362-380

Purpose:

Query LLM with retry logic.

Args:
    query: Query string

Returns:
    LLM response

Parameters: – query: Input query or question

Returns: str

get_metrics_report(self)

Lines 382-412

Purpose:

Generate retry metrics report.

Parameters:

Returns: str

---

Functions

retry_decorator(config)

Lines 415-434

Purpose:

Decorator for automatic retry with backoff.

Args:
    config: Retry configuration

Returns:
    Decorated function

Parameters: – config: Config

demonstrate_retry_with_backoff()

Lines 437-660

Purpose:

Demonstrate the Retry with Backoff pattern.

---

Key Concepts Summary

Pattern Overview

The Retry Backoff 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 145
• Pattern 147

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 Retry Backoff 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: 146_retry_backoff.py
Total Lines: 665
Pattern Category: AI Agent Design Pattern #146

---

Generated from actual implementation code
For educational and reference purposes