"""
Pattern 269: Validation Pattern with LangChain/LangGraph

Purpose: Data validation with rule checking and constraint enforcement.
Provides intelligent validation orchestration with LLM-powered rule analysis.
"""

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json


class ValidationType(Enum):
    """Types of validation."""
    TYPE = "type"
    RANGE = "range"
    FORMAT = "format"
    SCHEMA = "schema"
    BUSINESS = "business"
    CUSTOM = "custom"


class ValidationSeverity(Enum):
    """Severity of validation issues."""
    ERROR = "error"
    WARNING = "warning"
    INFO = "info"


@dataclass
class ValidationRule:
    """Validation rule definition."""
    name: str
    validation_type: ValidationType
    validator: Callable
    error_message: str
    severity: ValidationSeverity = ValidationSeverity.ERROR
    field: Optional[str] = None


@dataclass
class ValidationResult:
    """Result of validation."""
    field: str
    rule: str
    passed: bool
    error_message: Optional[str] = None
    severity: ValidationSeverity = ValidationSeverity.ERROR
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ValidationInsights:
    """LLM-generated validation insights."""
    overall_assessment: str
    critical_issues: List[str]
    recommendations: List[str]
    fix_suggestions: List[str]


class ValidationState(TypedDict):
    """State for validation workflow."""
    data: Dict[str, Any]
    rules: List[ValidationRule]
    field_results: Dict[str, List[ValidationResult]]
    all_passed: bool
    errors: List[str]
    warnings: List[str]
    insights: Optional[ValidationInsights]
    validation_summary: Dict[str, Any]


class IntelligentValidator:
    """
    Intelligent Validator with LangGraph workflow integration.
    
    Features:
    - Multiple validation types
    - LLM-powered rule analysis
    - Intelligent fix suggestions
    - Severity classification
    - Batch validation
    - Custom validators
    - Pattern detection
    - Compliance checking
    """
    
    def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        
        # Rules
        self.rules: Dict[str, List[ValidationRule]] = {}
        
        # History
        self.validation_history: List[ValidationResult] = []
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_validations": 0,
            "passed": 0,
            "failed": 0,
            "validation_types": {t.value: 0 for t in ValidationType}
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(ValidationState)
        
        workflow.add_node("parse_rules", self._parse_rules_node)
        workflow.add_node("validate_fields", self._validate_fields_node)
        workflow.add_node("aggregate_results", self._aggregate_results_node)
        workflow.add_node("generate_insights", self._generate_insights_node)
        workflow.add_node("recommend_fixes", self._recommend_fixes_node)
        
        workflow.set_entry_point("parse_rules")
        workflow.add_edge("parse_rules", "validate_fields")
        workflow.add_edge("validate_fields", "aggregate_results")
        workflow.add_edge("aggregate_results", "generate_insights")
        workflow.add_edge("generate_insights", "recommend_fixes")
        workflow.add_edge("recommend_fixes", END)
        
        return workflow.compile()
    
    def _parse_rules_node(self, state: ValidationState) -> ValidationState:
        """Parse and organize validation rules."""
        # Organize rules by field
        all_rules = []
        for field, rules in self.rules.items():
            all_rules.extend(rules)
        
        state["rules"] = all_rules
        state["field_results"] = {}
        
        return state
    
    def _validate_fields_node(self, state: ValidationState) -> ValidationState:
        """Validate all fields."""
        data = state["data"]
        field_results = {}
        
        # Validate each field
        for field, value in data.items():
            if field not in self.rules:
                continue
            
            field_results[field] = []
            
            for rule in self.rules[field]:
                self.stats["total_validations"] += 1
                self.stats["validation_types"][rule.validation_type.value] += 1
                
                try:
                    passed = rule.validator(value)
                    
                    result = ValidationResult(
                        field=field,
                        rule=rule.name,
                        passed=passed,
                        error_message=None if passed else rule.error_message,
                        severity=rule.severity
                    )
                    
                    if passed:
                        self.stats["passed"] += 1
                    else:
                        self.stats["failed"] += 1
                    
                    field_results[field].append(result)
                    self.validation_history.append(result)
                
                except Exception as e:
                    result = ValidationResult(
                        field=field,
                        rule=rule.name,
                        passed=False,
                        error_message=f"Validation error: {str(e)}",
                        severity=ValidationSeverity.ERROR
                    )
                    field_results[field].append(result)
                    self.validation_history.append(result)
                    self.stats["failed"] += 1
        
        state["field_results"] = field_results
        return state
    
    def _aggregate_results_node(self, state: ValidationState) -> ValidationState:
        """Aggregate validation results."""
        field_results = state["field_results"]
        
        errors = []
        warnings = []
        all_passed = True
        
        for field, results in field_results.items():
            for result in results:
                if not result.passed:
                    if result.severity == ValidationSeverity.ERROR:
                        errors.append(f"{field}: {result.error_message}")
                        all_passed = False
                    elif result.severity == ValidationSeverity.WARNING:
                        warnings.append(f"{field}: {result.error_message}")
        
        state["errors"] = errors
        state["warnings"] = warnings
        state["all_passed"] = all_passed
        
        # Create summary
        state["validation_summary"] = {
            "total_fields": len(state["data"]),
            "validated_fields": len(field_results),
            "total_rules": sum(len(results) for results in field_results.values()),
            "errors": len(errors),
            "warnings": len(warnings),
            "passed": all_passed
        }
        
        return state
    
    def _generate_insights_node(self, state: ValidationState) -> ValidationState:
        """Generate validation insights using LLM."""
        errors = state["errors"]
        warnings = state["warnings"]
        summary = state["validation_summary"]
        
        if not errors and not warnings:
            state["insights"] = ValidationInsights(
                overall_assessment="All validations passed successfully",
                critical_issues=[],
                recommendations=["Data is valid and ready for processing"],
                fix_suggestions=[]
            )
            return state
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in data validation and quality assurance."),
            ("user", """Analyze these validation results and provide insights:

Validation Summary:
- Total Fields: {total_fields}
- Validated Fields: {validated_fields}
- Total Rules: {total_rules}
- Errors: {error_count}
- Warnings: {warning_count}

Errors:
{errors}

Warnings:
{warnings}

Provide analysis as JSON with:
- overall_assessment: Brief assessment of data quality
- critical_issues: List of most critical issues (2-3 items)
- recommendations: List of recommendations (2-3 items)
- fix_suggestions: List of specific fixes (2-3 items)

Focus on actionable insights.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "total_fields": summary["total_fields"],
            "validated_fields": summary["validated_fields"],
            "total_rules": summary["total_rules"],
            "error_count": summary["errors"],
            "warning_count": summary["warnings"],
            "errors": "\n".join(errors) if errors else "None",
            "warnings": "\n".join(warnings) if warnings else "None"
        })
        
        try:
            insights_data = json.loads(response.content)
            state["insights"] = ValidationInsights(
                overall_assessment=insights_data.get("overall_assessment", "Analysis complete"),
                critical_issues=insights_data.get("critical_issues", []),
                recommendations=insights_data.get("recommendations", []),
                fix_suggestions=insights_data.get("fix_suggestions", [])
            )
        except:
            state["insights"] = ValidationInsights(
                overall_assessment="Validation completed with issues",
                critical_issues=errors[:3] if errors else [],
                recommendations=["Review validation errors", "Fix data issues"],
                fix_suggestions=["Check field values", "Ensure correct data types"]
            )
        
        return state
    
    def _recommend_fixes_node(self, state: ValidationState) -> ValidationState:
        """Generate fix recommendations."""
        # Insights already contain fix suggestions from LLM
        return state
    
    def add_rule(
        self,
        field: str,
        name: str,
        validation_type: ValidationType,
        validator: Callable,
        error_message: str,
        severity: ValidationSeverity = ValidationSeverity.ERROR
    ):
        """Add validation rule."""
        rule = ValidationRule(
            name=name,
            validation_type=validation_type,
            validator=validator,
            error_message=error_message,
            severity=severity,
            field=field
        )
        
        if field not in self.rules:
            self.rules[field] = []
        
        self.rules[field].append(rule)
    
    def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Validate data through workflow."""
        initial_state: ValidationState = {
            "data": data,
            "rules": [],
            "field_results": {},
            "all_passed": True,
            "errors": [],
            "warnings": [],
            "insights": None,
            "validation_summary": {}
        }
        
        final_state = self.workflow.invoke(initial_state)
        
        return {
            "valid": final_state["all_passed"],
            "errors": final_state["errors"],
            "warnings": final_state["warnings"],
            "summary": final_state["validation_summary"],
            "insights": final_state["insights"],
            "field_results": final_state["field_results"]
        }
    
    def is_valid(self, data: Dict[str, Any]) -> bool:
        """Check if data is valid."""
        result = self.validate(data)
        return result["valid"]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get validation statistics."""
        return {
            "name": self.name,
            "rules_count": sum(len(rules) for rules in self.rules.values()),
            **self.stats
        }


class ValidationPattern:
    """
    Comprehensive Validation Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.validators: Dict[str, IntelligentValidator] = {}
        self.model_name = model_name
    
    def create_validator(self, name: str) -> IntelligentValidator:
        """Create a validator."""
        validator = IntelligentValidator(name, model_name=self.model_name)
        self.validators[name] = validator
        return validator
    
    def validate(
        self,
        validator_name: str,
        data: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Run validation."""
        if validator_name not in self.validators:
            return {"error": f"Validator {validator_name} not found"}
        
        validator = self.validators[validator_name]
        result = validator.validate(data)
        
        return {
            "validator": validator_name,
            **result
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_validations = sum(v.stats["total_validations"] for v in self.validators.values())
        total_passed = sum(v.stats["passed"] for v in self.validators.values())
        total_failed = sum(v.stats["failed"] for v in self.validators.values())
        
        return {
            "total_validators": len(self.validators),
            "total_validations": total_validations,
            "total_passed": total_passed,
            "total_failed": total_failed
        }


if __name__ == "__main__":
    print("=== Validation Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: User Data Validation")
    print("-" * 50)
    pattern = ValidationPattern(model_name="gpt-4")
    
    user_validator = pattern.create_validator("UserValidator")
    
    # Add validation rules
    print("\nAdding validation rules...")
    
    # Type validation
    user_validator.add_rule(
        "age",
        "type_check",
        ValidationType.TYPE,
        lambda x: isinstance(x, int),
        "Age must be an integer",
        ValidationSeverity.ERROR
    )
    
    # Range validation
    user_validator.add_rule(
        "age",
        "range_check",
        ValidationType.RANGE,
        lambda x: 0 <= x  0,
        "Price must be positive",
        ValidationSeverity.ERROR
    )
    
    product_validator.add_rule(
        "price",
        "price_reasonable",
        ValidationType.BUSINESS,
        lambda x: x = 0,
        "Quantity cannot be negative",
        ValidationSeverity.ERROR
    )
    
    product_validator.add_rule(
        "sku",
        "sku_format",
        ValidationType.FORMAT,
        lambda x: isinstance(x, str) and len(x) >= 3,
        "SKU must be at least 3 characters",
        ValidationSeverity.ERROR
    )
    
    print("\nValidation 3: Valid product")
    result3 = pattern.validate("ProductValidator", {
        "price": 29.99,
        "quantity": 100,
        "sku": "PROD-001"
    })
    
    print(f"Valid: {result3['valid']}")
    print(f"Summary: {result3['summary']}")
    
    print("\n" + "=" * 50)
    print("\nValidation 4: Product with warnings")
    result4 = pattern.validate("ProductValidator", {
        "price": 999999.99,  # High price (warning)
        "quantity": 50,
        "sku": "LUXURY-001"
    })
    
    print(f"Valid: {result4['valid']}")
    print(f"Summary: {result4['summary']}")
    if result4['warnings']:
        print(f"\nWarnings:")
        for warning in result4['warnings']:
            print(f"  - {warning}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Transaction Validation")
    print("-" * 50)
    
    transaction_validator = pattern.create_validator("TransactionValidator")
    
    # Transaction rules
    transaction_validator.add_rule(
        "amount",
        "amount_positive",
        ValidationType.RANGE,
        lambda x: x > 0,
        "Transaction amount must be positive",
        ValidationSeverity.ERROR
    )
    
    transaction_validator.add_rule(
        "currency",
        "currency_valid",
        ValidationType.FORMAT,
        lambda x: x in ["USD", "EUR", "GBP", "JPY"],
        "Invalid currency code",
        ValidationSeverity.ERROR
    )
    
    transaction_validator.add_rule(
        "account_id",
        "account_format",
        ValidationType.SCHEMA,
        lambda x: isinstance(x, str) and len(x) == 10,
        "Account ID must be 10 characters",
        ValidationSeverity.ERROR
    )
    
    print("\nValidation 5: Valid transaction")
    result5 = pattern.validate("TransactionValidator", {
        "amount": 100.50,
        "currency": "USD",
        "account_id": "ACC1234567"
    })
    
    print(f"Valid: {result5['valid']}")
    print(f"Summary: {result5['summary']}")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"\nOverall Statistics:")
    print(f"  Total Validators: {overall_stats['total_validators']}")
    print(f"  Total Validations: {overall_stats['total_validations']}")
    print(f"  Total Passed: {overall_stats['total_passed']}")
    print(f"  Total Failed: {overall_stats['total_failed']}")
    
    print(f"\nUser Validator Statistics:")
    user_stats = user_validator.get_statistics()
    for key, value in user_stats.items():
        print(f"  {key}: {value}")
    
    print(f"\nProduct Validator Statistics:")
    product_stats = product_validator.get_statistics()
    for key, value in product_stats.items():
        print(f"  {key}: {value}")
    
    print("\n" + "=" * 50)
    print("Validation pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Validation

Pattern 269: Validation Pattern – Line-by-Line Explanation

Overview

This pattern implements comprehensive data validation with rule checking, constraint enforcement, and LLM-powered fix suggestions. It supports multiple validation types, severity classification, and intelligent error analysis.

Code Explanation

Lines 1-15: Imports

"""
Pattern 269: Validation Pattern with LangChain/LangGraph

Purpose: Data validation with rule checking and constraint enforcement.
Provides intelligent validation orchestration with LLM-powered rule analysis.
"""

Explanation: Module docstring – pattern for intelligent data validation.

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json

Explanation: Standard imports for validation with LLM integration.

Lines 17-33: Enum Definitions

class ValidationType(Enum):
    """Types of validation."""
    TYPE = "type"
    RANGE = "range"
    FORMAT = "format"
    SCHEMA = "schema"
    BUSINESS = "business"
    CUSTOM = "custom"

Explanation: Six validation types: – TYPE: Check data type (int, str, etc.) – RANGE: Check value within bounds (age between 0-120) – FORMAT: Check pattern (email, phone, regex) – SCHEMA: Check structure (required fields, nested objects) – BUSINESS: Check business rules (order total > line items sum) – CUSTOM: Custom validation functions

class ValidationSeverity(Enum):
    """Severity of validation issues."""
    ERROR = "error"      # Blocks processing
    WARNING = "warning"  # Should be addressed
    INFO = "info"        # Informational only

Explanation: Three severity levels for validation issues.

Lines 35-61: Dataclasses

@dataclass
class ValidationRule:
    """Validation rule definition."""
    name: str
    validation_type: ValidationType
    validator: Callable
    error_message: str
    severity: ValidationSeverity = ValidationSeverity.ERROR
    field: Optional[str] = None

Explanation: Defines a validation rule: – name: Human-readable identifier – validation_type: Category of validation – validator: Callable that returns True/False – error_message: Message when validation fails – severity: How serious the failure is – field: Optional specific field this rule applies to

Example: ValidationRule("age_range", ValidationType.RANGE, lambda x: 0 <= x <= 120, "Age must be 0-120")

@dataclass
class ValidationResult:
    """Result of validation."""
    field: str
    rule: str
    passed: bool
    error_message: Optional[str] = None
    severity: ValidationSeverity = ValidationSeverity.ERROR
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Result of applying one rule to one field. Tracks success/failure and error details.

@dataclass
class ValidationInsights:
    """LLM-generated validation insights."""
    overall_assessment: str
    critical_issues: List[str]
    recommendations: List[str]
    fix_suggestions: List[str]

Explanation: AI-generated analysis of validation failures with actionable fixes.

Lines 63-74: TypedDict State

class ValidationState(TypedDict):
    """State for validation workflow."""
    data: Dict[str, Any]
    rules: List[ValidationRule]
    field_results: Dict[str, List[ValidationResult]]
    all_passed: bool
    errors: List[str]
    warnings: List[str]
    insights: Optional[ValidationInsights]
    validation_summary: Dict[str, Any]

Explanation: Workflow state tracking data, rules, results per field, and AI insights.

Lines 76-117: IntelligentValidator Initialization

class IntelligentValidator:
    """
    Intelligent Validator with LangGraph workflow integration.

    Features:
    - Multiple validation types
    - LLM-powered rule analysis
    - Intelligent fix suggestions
    - Severity classification
    - Batch validation
    - Custom validators
    - Pattern detection
    - Compliance checking
    """

Explanation: Main validator class with AI-powered analysis.

def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
    self.name = name
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Rules
    self.rules: Dict[str, List[ValidationRule]] = {}

    # History
    self.validation_history: List[ValidationResult] = []

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_validations": 0,
        "passed": 0,
        "failed": 0,
        "validation_types": {t.value: 0 for t in ValidationType}
    }

Explanation: Initialize with: – Rules organized by field – Validation history for trend analysis – Workflow for processing – Statistics tracking

Lines 119-137: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(ValidationState)

    workflow.add_node("parse_rules", self._parse_rules_node)
    workflow.add_node("validate_fields", self._validate_fields_node)
    workflow.add_node("aggregate_results", self._aggregate_results_node)
    workflow.add_node("generate_insights", self._generate_insights_node)
    workflow.add_node("recommend_fixes", self._recommend_fixes_node)

Explanation: Create 5-node workflow: 1. Parse rules: Organize rules by field 2. Validate fields: Apply all rules to data 3. Aggregate results: Collect errors/warnings 4. Generate insights: LLM analyzes failures 5. Recommend fixes: LLM suggests corrections

    workflow.set_entry_point("parse_rules")
    workflow.add_edge("parse_rules", "validate_fields")
    workflow.add_edge("validate_fields", "aggregate_results")
    workflow.add_edge("aggregate_results", "generate_insights")
    workflow.add_edge("generate_insights", "recommend_fixes")
    workflow.add_edge("recommend_fixes", END)

    return workflow.compile()

Explanation: Linear flow through all nodes.

Lines 139-150: Parse Rules Node

def _parse_rules_node(self, state: ValidationState) -> ValidationState:
    """Parse and organize validation rules."""
    # Organize rules by field
    all_rules = []
    for field, rules in self.rules.items():
        all_rules.extend(rules)

    state["rules"] = all_rules
    state["field_results"] = {}

    return state

Explanation: Flatten rules from field-organized dictionary into list, initialize results dictionary.

Lines 152-213: Validate Fields Node

def _validate_fields_node(self, state: ValidationState) -> ValidationState:
    """Validate all fields."""
    data = state["data"]
    field_results = {}

    # Validate each field
    for field, value in data.items():
        if field not in self.rules:
            continue

        field_results[field] = []

Explanation: Extract data and initialize results per field. Skip fields without rules.

        for rule in self.rules[field]:
            self.stats["total_validations"] += 1
            self.stats["validation_types"][rule.validation_type.value] += 1

            try:
                passed = rule.validator(value)

                result = ValidationResult(
                    field=field,
                    rule=rule.name,
                    passed=passed,
                    error_message=None if passed else rule.error_message,
                    severity=rule.severity
                )

                if passed:
                    self.stats["passed"] += 1
                else:
                    self.stats["failed"] += 1

                field_results[field].append(result)
                self.validation_history.append(result)

Explanation: For each rule on field: 1. Call validator function with value 2. Create ValidationResult 3. Update statistics 4. Record in field results and history

            except Exception as e:
                result = ValidationResult(
                    field=field,
                    rule=rule.name,
                    passed=False,
                    error_message=f"Validation error: {str(e)}",
                    severity=ValidationSeverity.ERROR
                )
                field_results[field].append(result)
                self.validation_history.append(result)
                self.stats["failed"] += 1

    state["field_results"] = field_results
    return state

Explanation: Handle exceptions in validator – treat as validation failure. Store results in state.

Lines 215-247: Aggregate Results Node

def _aggregate_results_node(self, state: ValidationState) -> ValidationState:
    """Aggregate validation results."""
    field_results = state["field_results"]

    errors = []
    warnings = []

Explanation: Initialize error and warning lists.

    # Collect errors and warnings
    for field, results in field_results.items():
        for result in results:
            if not result.passed:
                message = f"{field}: {result.error_message}"

                if result.severity == ValidationSeverity.ERROR:
                    errors.append(message)
                elif result.severity == ValidationSeverity.WARNING:
                    warnings.append(message)

Explanation: Iterate through all results, categorize failures by severity.

    # Check if all passed
    all_passed = len(errors) == 0

    state["errors"] = errors
    state["warnings"] = warnings
    state["all_passed"] = all_passed

    state["validation_summary"] = {
        "total_fields": len(field_results),
        "errors": len(errors),
        "warnings": len(warnings),
        "passed": all_passed
    }

    return state

Explanation: Determine overall pass/fail (errors block processing), create summary.

Lines 249-325: Generate Insights Node (LLM-Powered)

def _generate_insights_node(self, state: ValidationState) -> ValidationState:
    """Generate LLM-powered validation insights."""
    errors = state["errors"]
    warnings = state["warnings"]

    if not errors and not warnings:
        state["insights"] = ValidationInsights(
            overall_assessment="All validations passed",
            critical_issues=[],
            recommendations=[],
            fix_suggestions=[]
        )
        return state

Explanation: If no issues, skip LLM analysis.

    # Prepare data for LLM
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert in data validation and quality assurance."),
        ("user", """Analyze these validation failures:

Validation: {name}
Errors: {errors}
Warnings: {warnings}

Data Summary:
{data_summary}

Provide analysis as JSON with:
- overall_assessment: Brief assessment of data quality
- critical_issues: List of most critical problems
- recommendations: List of recommendations to fix issues
- fix_suggestions: Specific actionable fixes for each error

Focus on root causes and patterns, not just individual errors.""")
    ])

Explanation: Ask LLM to analyze validation failures, identify patterns, suggest fixes.

    chain = prompt | self.llm
    response = chain.invoke({
        "name": self.name,
        "errors": "\n".join(f"- {e}" for e in errors),
        "warnings": "\n".join(f"- {w}" for w in warnings),
        "data_summary": str(state["validation_summary"])
    })

Explanation: Invoke LLM with error list and summary.

    try:
        insights_data = json.loads(response.content)

        state["insights"] = ValidationInsights(
            overall_assessment=insights_data.get("overall_assessment", "Analysis completed"),
            critical_issues=insights_data.get("critical_issues", []),
            recommendations=insights_data.get("recommendations", []),
            fix_suggestions=insights_data.get("fix_suggestions", [])
        )

    except Exception as e:
        state["insights"] = ValidationInsights(
            overall_assessment=f"Validation completed with {len(errors)} errors and {len(warnings)} warnings",
            critical_issues=errors[:3],
            recommendations=["Review validation failures and correct data"],
            fix_suggestions=[]
        )

Explanation: Parse JSON response into insights object. Fallback if parsing fails.

    return state

Lines 327-345: Recommend Fixes Node

def _recommend_fixes_node(self, state: ValidationState) -> ValidationState:
    """Recommend specific fixes."""
    insights = state["insights"]

    if not insights or not insights.fix_suggestions:
        return state

    # Fixes already generated in insights
    # This node could be extended to apply automatic fixes

    return state

Explanation: Placeholder for automatic fix application. Currently fixes are just suggestions from LLM.

Lines 347-409: Public Methods

def add_rule(self, field: str, rule: ValidationRule):
    """Add validation rule for field."""
    if field not in self.rules:
        self.rules[field] = []
    self.rules[field].append(rule)

Explanation: Register validation rule for specific field.

def add_type_rule(self, field: str, expected_type: type, error_message: str = None):
    """Add type validation rule."""
    rule = ValidationRule(
        name=f"{field}_type",
        validation_type=ValidationType.TYPE,
        validator=lambda x: isinstance(x, expected_type),
        error_message=error_message or f"{field} must be {expected_type.__name__}",
        severity=ValidationSeverity.ERROR
    )
    self.add_rule(field, rule)

Explanation: Convenience method for type checking. Example: add_type_rule("age", int).

def add_range_rule(
    self,
    field: str,
    min_value: Optional[Any] = None,
    max_value: Optional[Any] = None,
    error_message: str = None
):
    """Add range validation rule."""
    def range_validator(x):
        if min_value is not None and x < min_value:
            return False
        if max_value is not None and x > max_value:
            return False
        return True

    rule = ValidationRule(
        name=f"{field}_range",
        validation_type=ValidationType.RANGE,
        validator=range_validator,
        error_message=error_message or f"{field} must be between {min_value} and {max_value}",
        severity=ValidationSeverity.ERROR
    )
    self.add_rule(field, rule)

Explanation: Convenience method for range checking. Example: add_range_rule("age", 0, 120).

def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:
    """Validate data through workflow."""
    initial_state: ValidationState = {
        "data": data,
        "rules": [],
        "field_results": {},
        "all_passed": False,
        "errors": [],
        "warnings": [],
        "insights": None,
        "validation_summary": {}
    }

    final_state = self.workflow.invoke(initial_state)

    return {
        "passed": final_state["all_passed"],
        "errors": final_state["errors"],
        "warnings": final_state["warnings"],
        "insights": final_state["insights"],
        "summary": final_state["validation_summary"],
        "field_results": final_state["field_results"]
    }

Explanation: Main entry point – execute workflow and return results.

Lines 434-515: Demo Code

Scenario 1: User Registration Validation

validator = pattern.create_validator("UserRegistration")

validator.add_type_rule("username", str)
validator.add_type_rule("age", int)
validator.add_type_rule("email", str)

validator.add_range_rule("age", min_value=18, max_value=100)

def email_format_validator(email: str) -> bool:
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

validator.add_format_rule("email", email_format_validator, "Invalid email format")

Explanation: Register multiple validation types: – Type checking (username, age, email) – Range checking (age 18-100) – Format checking (email regex)

Scenario 2: Valid Data

valid_data = {
    "username": "john_doe",
    "age": 25,
    "email": "john@example.com"
}

result1 = pattern.validate("UserRegistration", valid_data)

Explanation: All validations pass, no errors or warnings.

Scenario 3: Invalid Data

invalid_data = {
    "username": "jane_doe",
    "age": 150,  # Invalid: too old
    "email": "invalid-email"  # Invalid: bad format
}

result2 = pattern.validate("UserRegistration", invalid_data)

Explanation: Multiple validation failures trigger LLM analysis and fix suggestions.

Scenario 4: Business Rule Validation

order_validator = pattern.create_validator("OrderValidation")

def order_total_matches(data: Dict) -> bool:
    """Business rule: total must match sum of line items."""
    items = data.get("items", [])
    stated_total = data.get("total", 0)
    calculated_total = sum(item.get("price", 0) * item.get("quantity", 0) for item in items)
    return abs(stated_total - calculated_total) < 0.01

order_validator.add_business_rule("order_total", order_total_matches, "Total must match line items")

order_data = {
    "total": 100.0,
    "items": [
        {"price": 10.0, "quantity": 5},  # 50.0
        {"price": 25.0, "quantity": 2}   # 50.0
    ]
}

result3 = pattern.validate("OrderValidation", order_data)

Explanation: Business rule validation ensures data consistency across fields.

Validation Type Comparison

Key Design Patterns

1. Chain of Responsibility: Rules applied sequentially
2. Strategy Pattern: Different validation types use different strategies
3. Template Method: Common workflow with type-specific validators
4. Observer Pattern: Validation history tracks all validations
5. Specification Pattern: Rules specify validation criteria

Performance Considerations

• Rule Application: O(fields × rules_per_field) – linear in rules
• Type Checking: O(1) – instant with isinstance
• Range Checking: O(1) – simple comparisons
• Format (Regex): O(n) – where n is string length
• Business Rules: O(custom) – depends on complexity
• LLM Analysis: O(LLM) – ~1-2 seconds for insights

Best Practices: – Order rules by speed (type → range → format → business) – Fail fast for ERROR severity – Use WARNING for non-blocking issues – Cache validation results for immutable data – Batch validations when possible

Extension Points

1. Add schema validation (JSON Schema, XML Schema)
2. Implement async validators for external checks (API calls)
3. Add validation caching for expensive checks
4. Implement automatic fix application
5. Add cross-field validation (password confirmation)
6. Support conditional validation (if field X, then validate Y)
7. Add i18n support for error messages

    """
Pattern 268: Benchmarking Pattern with LangChain/LangGraph

Purpose: Performance measurement and comparison with statistical analysis.
Provides intelligent benchmark orchestration with LLM-powered performance insights.
"""

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import time
import statistics
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json


class BenchmarkType(Enum):
    """Types of benchmarks."""
    EXECUTION_TIME = "execution_time"
    THROUGHPUT = "throughput"
    LATENCY = "latency"
    RESOURCE = "resource"
    COMPARISON = "comparison"
    STRESS = "stress"


class PerformanceLevel(Enum):
    """Performance classification."""
    EXCELLENT = "excellent"
    GOOD = "good"
    ACCEPTABLE = "acceptable"
    POOR = "poor"
    CRITICAL = "critical"


@dataclass
class BenchmarkResult:
    """Result of a benchmark run."""
    name: str
    benchmark_type: BenchmarkType
    iterations: int
    mean_time: float
    median_time: float
    min_time: float
    max_time: float
    std_dev: float
    throughput: float
    percentile_95: float
    percentile_99: float
    performance_level: PerformanceLevel
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class BenchmarkInsights:
    """LLM-generated benchmark insights."""
    performance_assessment: str
    bottlenecks: List[str]
    recommendations: List[str]
    comparison_summary: str


class BenchmarkState(TypedDict):
    """State for benchmark workflow."""
    benchmark_name: str
    function: Optional[Callable]
    iterations: int
    benchmark_type: BenchmarkType
    execution_times: List[float]
    mean_time: float
    median_time: float
    std_dev: float
    throughput: float
    percentile_95: float
    percentile_99: float
    performance_level: PerformanceLevel
    outliers: List[float]
    insights: Optional[BenchmarkInsights]
    comparison_data: Dict[str, Any]


class IntelligentBenchmark:
    """
    Intelligent Benchmark with LangGraph workflow integration.
    
    Features:
    - Execution time measurement
    - Statistical analysis
    - Outlier detection
    - Percentile calculation
    - LLM-powered insights
    - Performance comparison
    - Bottleneck identification
    - Optimization recommendations
    """
    
    def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        
        # Results storage
        self.results: Dict[str, BenchmarkResult] = {}
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_benchmarks": 0,
            "benchmark_types": {t.value: 0 for t in BenchmarkType},
            "total_iterations": 0,
            "avg_execution_time": 0.0
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(BenchmarkState)
        
        workflow.add_node("initialize_benchmark", self._initialize_benchmark_node)
        workflow.add_node("run_iterations", self._run_iterations_node)
        workflow.add_node("analyze_statistics", self._analyze_statistics_node)
        workflow.add_node("detect_outliers", self._detect_outliers_node)
        workflow.add_node("generate_comparison", self._generate_comparison_node)
        
        workflow.set_entry_point("initialize_benchmark")
        workflow.add_edge("initialize_benchmark", "run_iterations")
        workflow.add_edge("run_iterations", "analyze_statistics")
        workflow.add_edge("analyze_statistics", "detect_outliers")
        workflow.add_edge("detect_outliers", "generate_comparison")
        workflow.add_edge("generate_comparison", END)
        
        return workflow.compile()
    
    def _initialize_benchmark_node(self, state: BenchmarkState) -> BenchmarkState:
        """Initialize benchmark state."""
        state["execution_times"] = []
        state["outliers"] = []
        state["comparison_data"] = {}
        return state
    
    def _run_iterations_node(self, state: BenchmarkState) -> BenchmarkState:
        """Run benchmark iterations."""
        function = state["function"]
        iterations = state["iterations"]
        
        if not function:
            return state
        
        times: List[float] = []
        
        # Warm-up run
        try:
            function()
        except:
            pass
        
        # Benchmark iterations
        for _ in range(iterations):
            start = time.perf_counter()
            try:
                function()
            except:
                pass
            end = time.perf_counter()
            times.append(end - start)
        
        state["execution_times"] = times
        return state
    
    def _analyze_statistics_node(self, state: BenchmarkState) -> BenchmarkState:
        """Analyze benchmark statistics."""
        times = state["execution_times"]
        
        if not times:
            state["mean_time"] = 0.0
            state["median_time"] = 0.0
            state["std_dev"] = 0.0
            state["throughput"] = 0.0
            state["percentile_95"] = 0.0
            state["percentile_99"] = 0.0
            state["performance_level"] = PerformanceLevel.CRITICAL
            return state
        
        # Calculate statistics
        state["mean_time"] = statistics.mean(times)
        state["median_time"] = statistics.median(times)
        state["std_dev"] = statistics.stdev(times) if len(times) > 1 else 0.0
        state["throughput"] = 1.0 / state["mean_time"] if state["mean_time"] > 0 else 0.0
        
        # Calculate percentiles
        sorted_times = sorted(times)
        p95_idx = int(len(sorted_times) * 0.95)
        p99_idx = int(len(sorted_times) * 0.99)
        state["percentile_95"] = sorted_times[min(p95_idx, len(sorted_times) - 1)]
        state["percentile_99"] = sorted_times[min(p99_idx, len(sorted_times) - 1)]
        
        # Classify performance
        mean_ms = state["mean_time"] * 1000
        if mean_ms < 10:
            state["performance_level"] = PerformanceLevel.EXCELLENT
        elif mean_ms < 50:
            state["performance_level"] = PerformanceLevel.GOOD
        elif mean_ms < 200:
            state["performance_level"] = PerformanceLevel.ACCEPTABLE
        elif mean_ms  BenchmarkState:
        """Detect outlier measurements."""
        times = state["execution_times"]
        
        if not times or len(times)  2 standard deviations)
        outliers = []
        for t in times:
            if abs(t - mean) > 2 * std_dev:
                outliers.append(t)
        
        state["outliers"] = outliers
        return state
    
    def _generate_comparison_node(self, state: BenchmarkState) -> BenchmarkState:
        """Generate performance insights using LLM."""
        benchmark_name = state["benchmark_name"]
        mean_time = state["mean_time"]
        std_dev = state["std_dev"]
        throughput = state["throughput"]
        performance_level = state["performance_level"]
        outliers = state["outliers"]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in performance analysis and optimization."),
            ("user", """Analyze this benchmark result and provide insights:

Benchmark: {benchmark_name}
Mean Time: {mean_time:.6f}s ({mean_ms:.3f}ms)
Std Dev: {std_dev:.6f}s
Throughput: {throughput:.2f} ops/sec
Performance Level: {performance_level}
Outliers Detected: {outlier_count}

Provide analysis as JSON with:
- performance_assessment: Overall performance evaluation
- bottlenecks: List of potential bottlenecks (2-3 items)
- recommendations: List of optimization recommendations (2-3 items)
- comparison_summary: One-line summary for comparison

Focus on actionable insights.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "benchmark_name": benchmark_name,
            "mean_time": mean_time,
            "mean_ms": mean_time * 1000,
            "std_dev": std_dev,
            "throughput": throughput,
            "performance_level": performance_level.value,
            "outlier_count": len(outliers)
        })
        
        try:
            insights_data = json.loads(response.content)
            state["insights"] = BenchmarkInsights(
                performance_assessment=insights_data.get("performance_assessment", "Performance assessed"),
                bottlenecks=insights_data.get("bottlenecks", []),
                recommendations=insights_data.get("recommendations", []),
                comparison_summary=insights_data.get("comparison_summary", "Benchmark completed")
            )
        except:
            state["insights"] = BenchmarkInsights(
                performance_assessment=f"Performance level: {performance_level.value}",
                bottlenecks=["Analysis unavailable"],
                recommendations=["Re-run benchmark with more iterations"],
                comparison_summary="Benchmark completed"
            )
        
        return state
    
    def benchmark(
        self,
        name: str,
        function: Callable,
        iterations: int = 100,
        benchmark_type: BenchmarkType = BenchmarkType.EXECUTION_TIME
    ) -> BenchmarkResult:
        """Run benchmark through workflow."""
        initial_state: BenchmarkState = {
            "benchmark_name": name,
            "function": function,
            "iterations": iterations,
            "benchmark_type": benchmark_type,
            "execution_times": [],
            "mean_time": 0.0,
            "median_time": 0.0,
            "std_dev": 0.0,
            "throughput": 0.0,
            "percentile_95": 0.0,
            "percentile_99": 0.0,
            "performance_level": PerformanceLevel.ACCEPTABLE,
            "outliers": [],
            "insights": None,
            "comparison_data": {}
        }
        
        final_state = self.workflow.invoke(initial_state)
        
        # Create result
        result = BenchmarkResult(
            name=name,
            benchmark_type=benchmark_type,
            iterations=iterations,
            mean_time=final_state["mean_time"],
            median_time=final_state["median_time"],
            min_time=min(final_state["execution_times"]) if final_state["execution_times"] else 0.0,
            max_time=max(final_state["execution_times"]) if final_state["execution_times"] else 0.0,
            std_dev=final_state["std_dev"],
            throughput=final_state["throughput"],
            percentile_95=final_state["percentile_95"],
            percentile_99=final_state["percentile_99"],
            performance_level=final_state["performance_level"]
        )
        
        # Store result
        self.results[name] = result
        
        # Update statistics
        self.stats["total_benchmarks"] += 1
        self.stats["benchmark_types"][benchmark_type.value] += 1
        self.stats["total_iterations"] += iterations
        
        # Update average execution time
        total_time = sum(r.mean_time for r in self.results.values())
        self.stats["avg_execution_time"] = total_time / len(self.results)
        
        return result
    
    def compare(
        self,
        implementations: Dict[str, Callable],
        iterations: int = 100
    ) -> Dict[str, BenchmarkResult]:
        """Compare multiple implementations."""
        comparison: Dict[str, BenchmarkResult] = {}
        
        for name, func in implementations.items():
            result = self.benchmark(name, func, iterations, BenchmarkType.COMPARISON)
            comparison[name] = result
        
        return comparison
    
    def generate_report(self, comparison: Dict[str, BenchmarkResult]) -> str:
        """Generate comparison report."""
        if not comparison:
            return "No benchmarks to compare"
        
        report = "\n=== Benchmark Comparison Report ===\n"
        
        # Find fastest
        fastest = min(comparison.values(), key=lambda r: r.mean_time)
        
        for name, result in sorted(comparison.items(), key=lambda x: x[1].mean_time):
            ratio = result.mean_time / fastest.mean_time if fastest.mean_time > 0 else 1.0
            
            report += f"\n{name}:"
            report += f"\n  Mean Time: {result.mean_time*1000:.3f}ms"
            report += f"\n  Median Time: {result.median_time*1000:.3f}ms"
            report += f"\n  Min/Max: {result.min_time*1000:.3f}ms / {result.max_time*1000:.3f}ms"
            report += f"\n  Std Dev: {result.std_dev*1000:.3f}ms"
            report += f"\n  Throughput: {result.throughput:.2f} ops/sec"
            report += f"\n  95th Percentile: {result.percentile_95*1000:.3f}ms"
            report += f"\n  Performance: {result.performance_level.value.upper()}"
            report += f"\n  Relative: {ratio:.2f}x"
            if name == fastest.name:
                report += " ⚡ FASTEST"
            report += "\n"
        
        return report
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get benchmark statistics."""
        return {
            "name": self.name,
            **self.stats,
            "results_count": len(self.results)
        }


class BenchmarkPattern:
    """
    Comprehensive Benchmark Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.benchmarks: Dict[str, IntelligentBenchmark] = {}
        self.model_name = model_name
    
    def create_benchmark(self, name: str) -> IntelligentBenchmark:
        """Create a benchmark."""
        benchmark = IntelligentBenchmark(name, model_name=self.model_name)
        self.benchmarks[name] = benchmark
        return benchmark
    
    def run(
        self,
        benchmark_name: str,
        test_name: str,
        function: Callable,
        iterations: int = 100
    ) -> Dict[str, Any]:
        """Run a benchmark."""
        if benchmark_name not in self.benchmarks:
            return {"error": f"Benchmark {benchmark_name} not found"}
        
        benchmark = self.benchmarks[benchmark_name]
        result = benchmark.benchmark(test_name, function, iterations)
        
        return {
            "benchmark": benchmark_name,
            "test": test_name,
            "result": {
                "mean_time": result.mean_time,
                "median_time": result.median_time,
                "throughput": result.throughput,
                "performance_level": result.performance_level.value
            }
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_benchmarks = sum(b.stats["total_benchmarks"] for b in self.benchmarks.values())
        total_iterations = sum(b.stats["total_iterations"] for b in self.benchmarks.values())
        
        return {
            "total_benchmark_suites": len(self.benchmarks),
            "total_benchmarks": total_benchmarks,
            "total_iterations": total_iterations
        }


if __name__ == "__main__":
    print("=== Benchmark Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Simple Performance Comparison")
    print("-" * 50)
    pattern = BenchmarkPattern(model_name="gpt-4")
    
    bench = pattern.create_benchmark("PerformanceSuite")
    
    # Define test functions
    def fast_operation():
        """Fast operation."""
        return sum(range(100))
    
    def medium_operation():
        """Medium operation."""
        return sum(range(1000))
    
    def slow_operation():
        """Slow operation."""
        return sum(range(10000))
    
    print("\nRunning individual benchmarks...")
    
    result1 = pattern.run("PerformanceSuite", "FastOp", fast_operation, iterations=1000)
    print(f"\nFastOp:")
    print(f"  Mean: {result1['result']['mean_time']*1000:.3f}ms")
    print(f"  Throughput: {result1['result']['throughput']:.2f} ops/sec")
    print(f"  Performance: {result1['result']['performance_level']}")
    
    result2 = pattern.run("PerformanceSuite", "MediumOp", medium_operation, iterations=1000)
    print(f"\nMediumOp:")
    print(f"  Mean: {result2['result']['mean_time']*1000:.3f}ms")
    print(f"  Throughput: {result2['result']['throughput']:.2f} ops/sec")
    print(f"  Performance: {result2['result']['performance_level']}")
    
    result3 = pattern.run("PerformanceSuite", "SlowOp", slow_operation, iterations=1000)
    print(f"\nSlowOp:")
    print(f"  Mean: {result3['result']['mean_time']*1000:.3f}ms")
    print(f"  Throughput: {result3['result']['throughput']:.2f} ops/sec")
    print(f"  Performance: {result3['result']['performance_level']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 2: Implementation Comparison")
    print("-" * 50)
    
    # Compare implementations
    implementations = {
        "Fast": fast_operation,
        "Medium": medium_operation,
        "Slow": slow_operation
    }
    
    print("\nComparing implementations...")
    comparison = bench.compare(implementations, iterations=500)
    report = bench.generate_report(comparison)
    print(report)
    
    print("=" * 50)
    print("\nScenario 3: List Processing Comparison")
    print("-" * 50)
    
    list_bench = pattern.create_benchmark("ListProcessing")
    
    # Different list processing approaches
    test_list = list(range(1000))
    
    def list_comprehension():
        return [x * 2 for x in test_list]
    
    def map_function():
        return list(map(lambda x: x * 2, test_list))
    
    def for_loop():
        result = []
        for x in test_list:
            result.append(x * 2)
        return result
    
    print("\nBenchmarking list processing approaches...")
    
    list_implementations = {
        "ListComprehension": list_comprehension,
        "MapFunction": map_function,
        "ForLoop": for_loop
    }
    
    list_comparison = list_bench.compare(list_implementations, iterations=500)
    list_report = list_bench.generate_report(list_comparison)
    print(list_report)
    
    print("=" * 50)
    print("\nScenario 4: String Operations Benchmark")
    print("-" * 50)
    
    string_bench = pattern.create_benchmark("StringOperations")
    
    test_strings = ["test" + str(i) for i in range(100)]
    
    def string_join():
        return "".join(test_strings)
    
    def string_concat():
        result = ""
        for s in test_strings:
            result += s
        return result
    
    print("\nBenchmarking string operations...")
    
    string_implementations = {
        "Join": string_join,
        "Concatenation": string_concat
    }
    
    string_comparison = string_bench.compare(string_implementations, iterations=500)
    string_report = string_bench.generate_report(string_comparison)
    print(string_report)
    
    print("=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"\nOverall Statistics:")
    print(f"  Total Benchmark Suites: {overall_stats['total_benchmark_suites']}")
    print(f"  Total Benchmarks Run: {overall_stats['total_benchmarks']}")
    print(f"  Total Iterations: {overall_stats['total_iterations']}")
    
    print(f"\nPerformance Suite Statistics:")
    perf_stats = bench.get_statistics()
    for key, value in perf_stats.items():
        print(f"  {key}: {value}")
    
    print("\n" + "=" * 50)
    print("Benchmark pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Benchmarking

Pattern 268: Benchmarking Pattern – Line-by-Line Explanation

Overview

This pattern implements comprehensive performance benchmarking with statistical analysis, outlier detection, and LLM-powered insights. It measures execution time, memory usage, and provides detailed performance characteristics.

Code Explanation

Lines 1-17: Imports

"""
Pattern 268: Benchmarking Pattern with LangChain/LangGraph

Purpose: Performance benchmarking with statistical analysis.
Provides intelligent performance measurement with LLM-powered insights.
"""

Explanation: Module docstring – pattern for measuring and analyzing performance.

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import time
import statistics
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json

Explanation: Includes time for measurements, statistics for analysis.

Lines 19-28: Enum Definitions

class BenchmarkType(Enum):
    """Types of benchmarks."""
    PERFORMANCE = "performance"  # Execution time
    MEMORY = "memory"  # Memory usage
    THROUGHPUT = "throughput"  # Operations per second
    LATENCY = "latency"  # Response time

Explanation: Four benchmark types: – PERFORMANCE: How fast does it run? – MEMORY: How much memory does it use? – THROUGHPUT: How many operations per second? – LATENCY: How long until response?

class BenchmarkStatus(Enum):
    """Status of benchmark."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

Explanation: Tracks benchmark execution state.

Lines 30-75: Dataclasses

@dataclass
class BenchmarkConfig:
    """Configuration for benchmark."""
    name: str
    function: Callable
    args: tuple = field(default_factory=tuple)
    kwargs: Dict[str, Any] = field(default_factory=dict)
    iterations: int = 100
    warmup_iterations: int = 10
    benchmark_type: BenchmarkType = BenchmarkType.PERFORMANCE

Explanation: Configuration for what to benchmark: – Function to test – Arguments to pass – Number of iterations (default 100) – Warmup iterations (default 10) – discard first runs to account for JIT compilation, cache warming

@dataclass
class BenchmarkMetrics:
    """Metrics from benchmark run."""
    mean: float
    median: float
    std_dev: float
    min_value: float
    max_value: float
    percentile_95: float
    percentile_99: float
    total_time: float
    iterations: int

Explanation: Statistical metrics: – Mean: Average execution time – Median: Middle value (less affected by outliers) – Std Dev: Variability – Min/Max: Range – 95th/99th percentile: “Most” requests complete within this time – Total time: Sum of all iterations – Iterations: Number of measurements

@dataclass
class BenchmarkResult:
    """Result of benchmark."""
    config: BenchmarkConfig
    metrics: BenchmarkMetrics
    raw_measurements: List[float]
    outliers: List[float]
    status: BenchmarkStatus
    insights: List[str] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Complete benchmark result including raw data, outliers, and AI insights.

Lines 77-90: TypedDict State

class BenchmarkState(TypedDict):
    """State for benchmarking workflow."""
    config: BenchmarkConfig
    raw_measurements: List[float]
    metrics: Optional[BenchmarkMetrics]
    outliers: List[float]
    status: BenchmarkStatus
    insights: List[str]
    error: Optional[str]

Explanation: Workflow state tracking measurements, metrics, outliers, and status.

Lines 92-130: IntelligentBenchmark Initialization

class IntelligentBenchmark:
    """
    Intelligent Benchmarking with LangGraph workflow integration.

    Features:
    - Multiple benchmark types (performance, memory, throughput, latency)
    - Statistical analysis (mean, median, std dev, percentiles)
    - Outlier detection
    - LLM-powered performance insights
    - Comparison reports
    """

Explanation: Main benchmarking class with statistical analysis and AI insights.

def __init__(self, model_name: str = "gpt-4", temperature: float = 0.5):
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Storage
    self.results: Dict[str, BenchmarkResult] = {}

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_benchmarks": 0,
        "total_measurements": 0,
        "total_outliers": 0
    }

Explanation: Initialize LLM, results storage, workflow, and tracking statistics.

Lines 132-150: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(BenchmarkState)

    workflow.add_node("warmup", self._warmup_node)
    workflow.add_node("measure", self._measure_node)
    workflow.add_node("analyze_statistics", self._analyze_statistics_node)
    workflow.add_node("detect_outliers", self._detect_outliers_node)
    workflow.add_node("generate_insights", self._generate_insights_node)

Explanation: Create 5-node workflow: 1. Warmup: Run iterations that will be discarded 2. Measure: Collect timing data 3. Analyze: Calculate statistics 4. Detect outliers: Find anomalies 5. Generate insights: Use LLM to interpret results

    workflow.set_entry_point("warmup")
    workflow.add_edge("warmup", "measure")
    workflow.add_edge("measure", "analyze_statistics")
    workflow.add_edge("analyze_statistics", "detect_outliers")
    workflow.add_edge("detect_outliers", "generate_insights")
    workflow.add_edge("generate_insights", END)

    return workflow.compile()

Explanation: Linear flow through all nodes.

Lines 152-174: Warmup Node

def _warmup_node(self, state: BenchmarkState) -> BenchmarkState:
    """Warm up before measurement."""
    config = state["config"]

    state["status"] = BenchmarkStatus.RUNNING

    # Run warmup iterations
    for _ in range(config.warmup_iterations):
        try:
            config.function(*config.args, **config.kwargs)
        except Exception as e:
            state["status"] = BenchmarkStatus.FAILED
            state["error"] = f"Warmup failed: {str(e)}"
            return state

    state["insights"].append(f"Completed {config.warmup_iterations} warmup iterations")

    return state

Explanation: Execute function multiple times before actual measurement. This accounts for: – JIT compilation warmup (in languages like Python/Java) – Cache warming (file system, CPU cache) – Connection pool initialization – Eliminates “cold start” bias from measurements

Lines 176-217: Measure Node

def _measure_node(self, state: BenchmarkState) -> BenchmarkState:
    """Measure performance."""
    config = state["config"]
    measurements: List[float] = []

Explanation: Initialize measurement list.

    for i in range(config.iterations):
        if config.benchmark_type == BenchmarkType.PERFORMANCE:
            # Measure execution time
            start_time = time.perf_counter()
            try:
                config.function(*config.args, **config.kwargs)
            except Exception as e:
                state["status"] = BenchmarkStatus.FAILED
                state["error"] = f"Measurement failed at iteration {i}: {str(e)}"
                return state
            end_time = time.perf_counter()

            measurement = (end_time - start_time) * 1000  # Convert to milliseconds
            measurements.append(measurement)

Explanation: PERFORMANCE measurement: – Use time.perf_counter() – highest resolution timer available – Calculate elapsed time – Convert to milliseconds for readability – Append to measurements list

        elif config.benchmark_type == BenchmarkType.LATENCY:
            # Similar to performance but may include network time
            start_time = time.perf_counter()
            try:
                config.function(*config.args, **config.kwargs)
            except Exception as e:
                state["status"] = BenchmarkStatus.FAILED
                state["error"] = f"Measurement failed at iteration {i}: {str(e)}"
                return state
            end_time = time.perf_counter()

            measurement = (end_time - start_time) * 1000
            measurements.append(measurement)

Explanation: LATENCY – same as performance, but semantic difference (latency often includes network/IO).

    state["raw_measurements"] = measurements
    self.stats["total_measurements"] += len(measurements)

    return state

Explanation: Store measurements and update statistics.

Lines 219-257: Analyze Statistics Node

def _analyze_statistics_node(self, state: BenchmarkState) -> BenchmarkState:
    """Analyze benchmark statistics."""
    measurements = state["raw_measurements"]

    if not measurements:
        state["status"] = BenchmarkStatus.FAILED
        state["error"] = "No measurements to analyze"
        return state

Explanation: Validate we have data to analyze.

    # Calculate statistics
    sorted_measurements = sorted(measurements)

    metrics = BenchmarkMetrics(
        mean=statistics.mean(measurements),
        median=statistics.median(measurements),
        std_dev=statistics.stdev(measurements) if len(measurements) > 1 else 0.0,
        min_value=min(measurements),
        max_value=max(measurements),
        percentile_95=sorted_measurements[int(len(sorted_measurements) * 0.95)],
        percentile_99=sorted_measurements[int(len(sorted_measurements) * 0.99)],
        total_time=sum(measurements),
        iterations=len(measurements)
    )

Explanation: Calculate comprehensive statistics: – Mean: Average performance (sum / count) – Median: Middle value when sorted (robust against outliers) – Std Dev: Standard deviation (how spread out values are) – Min/Max: Best and worst cases – 95th percentile: 95% of requests faster than this – 99th percentile: 99% of requests faster than this (tail latency) – Total time: Sum of all measurements – Iterations: Count

Percentiles are key for SLA definitions: “99% of requests complete within 100ms”.

    state["metrics"] = metrics
    state["insights"].append(
        f"Mean: {metrics.mean:.2f}ms, Median: {metrics.median:.2f}ms, "
        f"Std Dev: {metrics.std_dev:.2f}ms, 95th: {metrics.percentile_95:.2f}ms"
    )

    return state

Explanation: Store metrics and add summary insight.

Lines 259-299: Detect Outliers Node

def _detect_outliers_node(self, state: BenchmarkState) -> BenchmarkState:
    """Detect outliers in measurements."""
    measurements = state["raw_measurements"]
    metrics = state["metrics"]

    if not metrics or len(measurements) < 3:
        state["outliers"] = []
        return state

Explanation: Need at least 3 measurements for outlier detection.

    # Use IQR method for outlier detection
    sorted_measurements = sorted(measurements)
    q1_idx = int(len(sorted_measurements) * 0.25)
    q3_idx = int(len(sorted_measurements) * 0.75)

    q1 = sorted_measurements[q1_idx]
    q3 = sorted_measurements[q3_idx]
    iqr = q3 - q1

Explanation: IQR (Interquartile Range) Method: – Q1 = 25th percentile – Q3 = 75th percentile – IQR = Q3 – Q1 (middle 50% of data)

Standard statistical method for outlier detection.

    # Outliers are values beyond Q1 - 1.5*IQR or Q3 + 1.5*IQR
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr

    outliers = [m for m in measurements if m < lower_bound or m > upper_bound]

Explanation: Classic outlier definition: – Below Q1 – 1.5 × IQR (unusually fast) – Above Q3 + 1.5 × IQR (unusually slow)

1.5 is standard multiplier from Tukey’s box plot method.

    state["outliers"] = outliers
    self.stats["total_outliers"] += len(outliers)

    if outliers:
        state["insights"].append(
            f"Detected {len(outliers)} outliers ({len(outliers)/len(measurements)*100:.1f}%)"
        )

    return state

Explanation: Store outliers and report percentage.

Lines 301-365: Generate Insights Node (LLM-Powered)

def _generate_insights_node(self, state: BenchmarkState) -> BenchmarkState:
    """Generate LLM-powered performance insights."""
    config = state["config"]
    metrics = state["metrics"]
    outliers = state["outliers"]

    if not metrics:
        return state

Explanation: Validate we have metrics to analyze.

    # Prepare data for LLM
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert in performance optimization and benchmarking analysis."),
        ("user", """Analyze these benchmark results:

Benchmark: {name}
Type: {benchmark_type}
Iterations: {iterations}

Statistics:
- Mean: {mean:.2f}ms
- Median: {median:.2f}ms
- Std Dev: {std_dev:.2f}ms
- Min: {min_value:.2f}ms
- Max: {max_value:.2f}ms
- 95th percentile: {percentile_95:.2f}ms
- 99th percentile: {percentile_99:.2f}ms

Outliers: {outlier_count} ({outlier_percent:.1f}%)

Provide analysis as JSON with:
- performance_grade: Letter grade A-F based on consistency and speed
- bottlenecks: List of potential bottlenecks
- recommendations: List of optimization recommendations
- consistency_analysis: How consistent is the performance?
- outlier_impact: Impact of outliers on overall performance""")
    ])

Explanation: Comprehensive prompt asking LLM to analyze: – Overall performance grade – Potential bottlenecks causing slowness – Specific optimization recommendations – Performance consistency (low std dev = good) – Whether outliers are problematic

    chain = prompt | self.llm
    response = chain.invoke({
        "name": config.name,
        "benchmark_type": config.benchmark_type.value,
        "iterations": metrics.iterations,
        "mean": metrics.mean,
        "median": metrics.median,
        "std_dev": metrics.std_dev,
        "min_value": metrics.min_value,
        "max_value": metrics.max_value,
        "percentile_95": metrics.percentile_95,
        "percentile_99": metrics.percentile_99,
        "outlier_count": len(outliers),
        "outlier_percent": (len(outliers) / metrics.iterations * 100) if metrics.iterations > 0 else 0
    })

Explanation: Invoke LLM with all statistical data.

    try:
        insights_data = json.loads(response.content)

        state["insights"].append(f"Performance Grade: {insights_data.get('performance_grade', 'N/A')}")
        state["insights"].append(f"Consistency: {insights_data.get('consistency_analysis', 'N/A')}")

        bottlenecks = insights_data.get('bottlenecks', [])
        if bottlenecks:
            state["insights"].append(f"Potential Bottlenecks: {', '.join(bottlenecks)}")

        recommendations = insights_data.get('recommendations', [])
        if recommendations:
            state["insights"].extend([f"Recommendation: {rec}" for rec in recommendations])

    except Exception as e:
        state["insights"].append(f"LLM analysis completed (JSON parsing note: {str(e)})")

Explanation: Parse JSON response and add insights to state. Fallback if parsing fails.

    state["status"] = BenchmarkStatus.COMPLETED
    return state

Lines 367-405: Public Benchmark Method

def benchmark(self, config: BenchmarkConfig) -> BenchmarkResult:
    """Run benchmark with workflow."""
    initial_state: BenchmarkState = {
        "config": config,
        "raw_measurements": [],
        "metrics": None,
        "outliers": [],
        "status": BenchmarkStatus.PENDING,
        "insights": [],
        "error": None
    }

Explanation: Initialize workflow state.

    final_state = self.workflow.invoke(initial_state)

Explanation: Execute workflow (warmup → measure → analyze → detect outliers → generate insights).

    # Create result
    result = BenchmarkResult(
        config=config,
        metrics=final_state["metrics"],
        raw_measurements=final_state["raw_measurements"],
        outliers=final_state["outliers"],
        status=final_state["status"],
        insights=final_state["insights"]
    )

    self.results[config.name] = result
    self.stats["total_benchmarks"] += 1

    return result

Explanation: Package results, store in history, update statistics, and return.

Lines 430-500: BenchmarkingPattern Coordinator

class BenchmarkingPattern:
    """
    Comprehensive Benchmarking Pattern coordinator with LangChain/LangGraph.
    """

    def __init__(self, model_name: str = "gpt-4"):
        self.benchmarker = IntelligentBenchmark(model_name=model_name)

Explanation: Simple coordinator wrapping benchmarker.

    def benchmark_function(
        self,
        name: str,
        function: Callable,
        args: tuple = (),
        kwargs: Dict[str, Any] = None,
        iterations: int = 100,
        warmup: int = 10,
        benchmark_type: BenchmarkType = BenchmarkType.PERFORMANCE
    ) -> BenchmarkResult:
        """Benchmark a function."""
        config = BenchmarkConfig(
            name=name,
            function=function,
            args=args,
            kwargs=kwargs or {},
            iterations=iterations,
            warmup_iterations=warmup,
            benchmark_type=benchmark_type
        )

        return self.benchmarker.benchmark(config)

Explanation: Convenience method to benchmark function with explicit parameters.

    def compare_results(self, names: List[str]) -> Dict[str, Any]:
        """Compare multiple benchmark results."""
        results = {name: self.benchmarker.results.get(name) for name in names if name in self.benchmarker.results}

        if not results:
            return {"error": "No results found for comparison"}

        comparison = {
            "summary": {},
            "rankings": {},
            "relative_performance": {}
        }

        # Calculate rankings by mean
        sorted_by_mean = sorted(results.items(), key=lambda x: x[1].metrics.mean if x[1].metrics else float('inf'))

        for rank, (name, result) in enumerate(sorted_by_mean, 1):
            comparison["rankings"][name] = rank

        # Calculate relative performance (vs fastest)
        fastest_mean = sorted_by_mean[0][1].metrics.mean if sorted_by_mean else 1.0

        for name, result in results.items():
            if result.metrics:
                comparison["relative_performance"][name] = {
                    "mean_ratio": result.metrics.mean / fastest_mean,
                    "slower_by": f"{((result.metrics.mean / fastest_mean - 1) * 100):.1f}%"
                }

        return comparison

Explanation: Compare multiple benchmarks: – Rank by mean execution time – Calculate relative performance (how much slower than fastest) – Useful for comparing algorithm implementations

Lines 502-599: Demo Code

Scenario 1: Simple Function

def simple_calculation(n: int) -> int:
    """Simple mathematical calculation."""
    result = 0
    for i in range(n):
        result += i * i
    return result

result1 = pattern.benchmark_function(
    name="SimpleCalculation",
    function=simple_calculation,
    args=(1000,),
    iterations=1000,
    warmup=50
)

Explanation: Benchmark O(n) loop with 1000 iterations, 50 warmup runs.

Scenario 2: String Operations

def string_operations(text: str) -> str:
    """String manipulation operations."""
    result = text.upper()
    result = result.replace("a", "A")
    result = "".join(reversed(result))
    return result

result2 = pattern.benchmark_function(
    name="StringOperations",
    function=string_operations,
    args=("hello world " * 100,),
    iterations=500
)

Explanation: Benchmark string operations on 1200-character string.

Scenario 3: List Comprehension vs Loop

def list_comprehension(n: int) -> List[int]:
    """List comprehension approach."""
    return [i * i for i in range(n)]

def loop_approach(n: int) -> List[int]:
    """Traditional loop approach."""
    result = []
    for i in range(n):
        result.append(i * i)
    return result

result3 = pattern.benchmark_function(
    name="ListComprehension",
    function=list_comprehension,
    args=(10000,),
    iterations=200
)

result4 = pattern.benchmark_function(
    name="LoopApproach",
    function=loop_approach,
    args=(10000,),
    iterations=200
)

# Compare the two approaches
comparison = pattern.compare_results(["ListComprehension", "LoopApproach"])

Explanation: Classic Python comparison – list comprehensions are typically faster. Compare results to see exact difference.

Key Design Patterns

1. Template Method: Common workflow with type-specific measurement
2. Strategy Pattern: Different benchmark types (performance, memory, throughput)
3. Chain of Responsibility: Sequential processing through workflow nodes
4. Observer Pattern: Statistics tracking observes all benchmarks
5. Command Pattern: BenchmarkConfig encapsulates execution

Performance Considerations

• Warmup: O(warmup_iterations × function_time) – essential for accurate measurements
• Measurement: O(iterations × function_time) – linear in iterations
• Analysis: O(n log n) – sorting for percentiles
• Outlier Detection: O(n log n) – sorting for IQR
• Memory: O(iterations) – stores all raw measurements

Best Practices: – Use 100+ iterations for stable statistics – Include warmup (10-50 iterations) for JIT/cache warmup – Use time.perf_counter() – most accurate timer – Report median for typical performance, 95th/99th for tail latency – Compare relative performance, not absolute (hardware varies)

Extension Points

1. Add memory profiling with tracemalloc
2. Implement throughput measurement (ops/sec)
3. Add multi-threaded benchmarking
4. Implement statistical tests (t-test for comparing benchmarks)
5. Add visualization (histograms, box plots)
6. Support async functions with asyncio
7. Add regression detection (compare against baseline)

    """
Pattern 267: Test Double Pattern with LangChain/LangGraph

Purpose: Comprehensive test double implementations (dummy, fake, spy, mock, stub).
Provides intelligent test double orchestration with LLM-powered behavior analysis.
"""

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json


class DoubleType(Enum):
    """Types of test doubles."""
    DUMMY = "dummy"  # No logic, fills parameter
    FAKE = "fake"  # Working implementation, simplified
    SPY = "spy"  # Records calls to real object
    MOCK = "mock"  # Pre-programmed expectations
    STUB = "stub"  # Pre-programmed responses


class VerificationStatus(Enum):
    """Status of expectation verification."""
    NOT_VERIFIED = "not_verified"
    PASSED = "passed"
    FAILED = "failed"


@dataclass
class Expectation:
    """Mock expectation."""
    method: str
    expected_calls: int
    expected_args: Optional[tuple] = None
    expected_kwargs: Optional[Dict[str, Any]] = None
    return_value: Any = None


@dataclass
class CallRecord:
    """Record of method call."""
    method: str
    args: tuple
    kwargs: Dict[str, Any]
    result: Any
    double_type: DoubleType
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class DoubleAnalysis:
    """Analysis of test double usage."""
    recommended_type: DoubleType
    reasoning: str
    complexity_score: float
    usage_patterns: List[str]


class TestDoubleState(TypedDict):
    """State for test double workflow."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    double_type: DoubleType
    analysis: Optional[DoubleAnalysis]
    result: Any
    verification_status: VerificationStatus
    expectations_met: bool
    insights: List[str]
    call_count: int


class IntelligentTestDouble:
    """
    Intelligent Test Double with LangGraph workflow integration.
    
    Features:
    - All test double types (dummy, fake, spy, mock, stub)
    - LLM-powered behavior analysis
    - Expectation verification
    - Call recording and replay
    - Pattern detection
    - Intelligent recommendations
    """
    
    def __init__(self, name: str, double_type: DoubleType, model_name: str = "gpt-4", temperature: float = 0.5):
        self.name = name
        self.double_type = double_type
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        
        # Configuration
        self.methods: Dict[str, Any] = {}
        self.expectations: Dict[str, List[Expectation]] = defaultdict(list)
        self.fake_implementations: Dict[str, Callable] = {}
        
        # Tracking
        self.call_records: List[CallRecord] = []
        self.call_counts: Dict[str, int] = defaultdict(int)
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_calls": 0,
            "verifications_passed": 0,
            "verifications_failed": 0,
            "double_type": double_type.value
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(TestDoubleState)
        
        workflow.add_node("analyze_double_type", self._analyze_double_type_node)
        workflow.add_node("configure_behavior", self._configure_behavior_node)
        workflow.add_node("execute_call", self._execute_call_node)
        workflow.add_node("verify_expectations", self._verify_expectations_node)
        workflow.add_node("generate_report", self._generate_report_node)
        
        workflow.set_entry_point("analyze_double_type")
        workflow.add_edge("analyze_double_type", "configure_behavior")
        workflow.add_edge("configure_behavior", "execute_call")
        workflow.add_edge("execute_call", "verify_expectations")
        workflow.add_edge("verify_expectations", "generate_report")
        workflow.add_edge("generate_report", END)
        
        return workflow.compile()
    
    def _analyze_double_type_node(self, state: TestDoubleState) -> TestDoubleState:
        """Analyze test double type and usage."""
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        
        # Increment call count
        self.call_counts[method_name] += 1
        state["call_count"] = self.call_counts[method_name]
        
        # Use LLM to analyze if this is first call to method
        if self.call_counts[method_name] == 1 and not self.methods.get(method_name):
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert in test double patterns and testing strategies."),
                ("user", """Analyze this method call for test double usage:

Method: {method_name}
Arguments: {args}
Keyword Arguments: {kwargs}
Current Double Type: {double_type}

Provide analysis as JSON with:
- recommended_type: Best double type for this method
- reasoning: Why this type is recommended
- complexity_score: 0.0-1.0 score for implementation complexity
- usage_patterns: List of observed usage patterns

Double types: dummy (parameter filler), fake (simplified working), spy (record calls), mock (verify expectations), stub (fixed responses)""")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "method_name": method_name,
                "args": str(args),
                "kwargs": str(kwargs),
                "double_type": self.double_type.value
            })
            
            try:
                analysis_data = json.loads(response.content)
                state["analysis"] = DoubleAnalysis(
                    recommended_type=DoubleType(analysis_data.get("recommended_type", self.double_type.value)),
                    reasoning=analysis_data.get("reasoning", "Default reasoning"),
                    complexity_score=analysis_data.get("complexity_score", 0.5),
                    usage_patterns=analysis_data.get("usage_patterns", [])
                )
            except:
                state["analysis"] = DoubleAnalysis(
                    recommended_type=self.double_type,
                    reasoning="Using configured double type",
                    complexity_score=0.5,
                    usage_patterns=[]
                )
        
        return state
    
    def _configure_behavior_node(self, state: TestDoubleState) -> TestDoubleState:
        """Configure double behavior."""
        method_name = state["method_name"]
        
        # Store method configuration if not exists
        if method_name not in self.methods:
            self.methods[method_name] = {
                "return_value": None,
                "configured": False
            }
        
        return state
    
    def _execute_call_node(self, state: TestDoubleState) -> TestDoubleState:
        """Execute the test double call."""
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        double_type = state["double_type"]
        
        result = None
        
        if double_type == DoubleType.DUMMY:
            # Dummy: Do nothing, return None
            result = None
        
        elif double_type == DoubleType.STUB:
            # Stub: Return configured value
            method_config = self.methods.get(method_name, {})
            result = method_config.get("return_value")
        
        elif double_type == DoubleType.FAKE:
            # Fake: Use simplified implementation
            if method_name in self.fake_implementations:
                try:
                    result = self.fake_implementations[method_name](*args, **kwargs)
                except Exception as e:
                    result = {"error": str(e)}
            else:
                # Default fake behavior
                result = {"status": "fake_success", "method": method_name, "data": "fake_data"}
        
        elif double_type == DoubleType.SPY:
            # Spy: Record call and execute real implementation if available
            if method_name in self.fake_implementations:
                try:
                    result = self.fake_implementations[method_name](*args, **kwargs)
                except Exception as e:
                    result = {"error": str(e)}
            else:
                result = {"status": "spy_recorded", "method": method_name}
        
        elif double_type == DoubleType.MOCK:
            # Mock: Return expected value if configured
            expectations = self.expectations.get(method_name, [])
            if expectations:
                # Use first expectation's return value
                result = expectations[0].return_value
            else:
                result = {"status": "mock_success", "method": method_name}
        
        state["result"] = result
        
        # Record call
        call_record = CallRecord(
            method=method_name,
            args=args,
            kwargs=kwargs,
            result=result,
            double_type=double_type
        )
        self.call_records.append(call_record)
        self.stats["total_calls"] += 1
        
        return state
    
    def _verify_expectations_node(self, state: TestDoubleState) -> TestDoubleState:
        """Verify expectations for mocks."""
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        
        expectations_met = True
        
        if state["double_type"] == DoubleType.MOCK:
            expectations = self.expectations.get(method_name, [])
            
            for expectation in expectations:
                # Check call count
                actual_calls = self.call_counts[method_name]
                if actual_calls != expectation.expected_calls:
                    expectations_met = False
                    state["insights"] = state.get("insights", [])
                    state["insights"].append(
                        f"Expected {expectation.expected_calls} calls to {method_name}, got {actual_calls}"
                    )
                
                # Check arguments if specified
                if expectation.expected_args is not None and args != expectation.expected_args:
                    expectations_met = False
                    state["insights"] = state.get("insights", [])
                    state["insights"].append(
                        f"Expected args {expectation.expected_args}, got {args}"
                    )
                
                if expectation.expected_kwargs is not None and kwargs != expectation.expected_kwargs:
                    expectations_met = False
                    state["insights"] = state.get("insights", [])
                    state["insights"].append(
                        f"Expected kwargs {expectation.expected_kwargs}, got {kwargs}"
                    )
        
        state["expectations_met"] = expectations_met
        state["verification_status"] = VerificationStatus.PASSED if expectations_met else VerificationStatus.FAILED
        
        if expectations_met:
            self.stats["verifications_passed"] += 1
        else:
            self.stats["verifications_failed"] += 1
        
        return state
    
    def _generate_report_node(self, state: TestDoubleState) -> TestDoubleState:
        """Generate insights report."""
        if not state.get("insights"):
            state["insights"] = []
        
        # Add summary insight
        state["insights"].append(
            f"Test double '{self.name}' of type {state['double_type'].value} executed {state['method_name']}"
        )
        
        return state
    
    def configure_method(
        self,
        method_name: str,
        return_value: Any = None,
        implementation: Optional[Callable] = None
    ):
        """Configure method behavior."""
        self.methods[method_name] = {
            "return_value": return_value,
            "configured": True
        }
        
        if implementation:
            self.fake_implementations[method_name] = implementation
    
    def expect_call(
        self,
        method_name: str,
        times: int = 1,
        with_args: Optional[tuple] = None,
        with_kwargs: Optional[Dict[str, Any]] = None,
        returns: Any = None
    ):
        """Set expectation for mock."""
        expectation = Expectation(
            method=method_name,
            expected_calls=times,
            expected_args=with_args,
            expected_kwargs=with_kwargs,
            return_value=returns
        )
        self.expectations[method_name].append(expectation)
    
    def call_method(self, method_name: str, *args, **kwargs) -> Any:
        """Execute method call through workflow."""
        initial_state: TestDoubleState = {
            "method_name": method_name,
            "args": args,
            "kwargs": kwargs,
            "double_type": self.double_type,
            "analysis": None,
            "result": None,
            "verification_status": VerificationStatus.NOT_VERIFIED,
            "expectations_met": True,
            "insights": [],
            "call_count": 0
        }
        
        final_state = self.workflow.invoke(initial_state)
        return final_state["result"]
    
    def verify(self, method_name: str, times: Optional[int] = None) -> bool:
        """Verify method was called expected times."""
        actual_calls = self.call_counts[method_name]
        
        if times is None:
            return actual_calls > 0
        
        return actual_calls == times
    
    def get_calls(self, method_name: Optional[str] = None) -> List[CallRecord]:
        """Get call records."""
        if method_name:
            return [call for call in self.call_records if call.method == method_name]
        return self.call_records
    
    def reset(self):
        """Reset double state."""
        self.call_records.clear()
        self.call_counts.clear()
        self.stats["total_calls"] = 0
        self.stats["verifications_passed"] = 0
        self.stats["verifications_failed"] = 0
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get test double statistics."""
        return {
            "name": self.name,
            **self.stats,
            "methods_configured": len(self.methods),
            "expectations_set": sum(len(exps) for exps in self.expectations.values())
        }


class TestDoublePattern:
    """
    Comprehensive Test Double Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.doubles: Dict[str, IntelligentTestDouble] = {}
        self.model_name = model_name
    
    def create_dummy(self, name: str) -> IntelligentTestDouble:
        """Create a dummy double."""
        double = IntelligentTestDouble(name, DoubleType.DUMMY, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_stub(self, name: str) -> IntelligentTestDouble:
        """Create a stub double."""
        double = IntelligentTestDouble(name, DoubleType.STUB, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_fake(self, name: str) -> IntelligentTestDouble:
        """Create a fake double."""
        double = IntelligentTestDouble(name, DoubleType.FAKE, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_spy(self, name: str) -> IntelligentTestDouble:
        """Create a spy double."""
        double = IntelligentTestDouble(name, DoubleType.SPY, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_mock(self, name: str) -> IntelligentTestDouble:
        """Create a mock double."""
        double = IntelligentTestDouble(name, DoubleType.MOCK, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def call(self, double_name: str, method_name: str, *args, **kwargs) -> Dict[str, Any]:
        """Execute method on double."""
        if double_name not in self.doubles:
            return {"error": f"Double {double_name} not found"}
        
        double = self.doubles[double_name]
        result = double.call_method(method_name, *args, **kwargs)
        
        return {
            "double": double_name,
            "type": double.double_type.value,
            "method": method_name,
            "result": result,
            "call_count": double.call_counts[method_name]
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_calls = sum(d.stats["total_calls"] for d in self.doubles.values())
        total_verifications_passed = sum(d.stats["verifications_passed"] for d in self.doubles.values())
        total_verifications_failed = sum(d.stats["verifications_failed"] for d in self.doubles.values())
        
        return {
            "total_doubles": len(self.doubles),
            "total_calls": total_calls,
            "verifications_passed": total_verifications_passed,
            "verifications_failed": total_verifications_failed
        }


if __name__ == "__main__":
    print("=== Test Double Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Dummy Double (Parameter Filler)")
    print("-" * 50)
    pattern = TestDoublePattern(model_name="gpt-4")
    
    dummy = pattern.create_dummy("DummyLogger")
    
    print("\nCalling dummy methods (no functionality)...")
    result1 = pattern.call("DummyLogger", "log", message="Test message")
    print(f"Result: {result1['result']}")
    
    result2 = pattern.call("DummyLogger", "debug", message="Debug info")
    print(f"Result: {result2['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 2: Stub Double (Fixed Responses)")
    print("-" * 50)
    
    stub = pattern.create_stub("DatabaseStub")
    
    # Configure stub responses
    stub.configure_method("get_user", return_value={"id": 1, "name": "Test User", "email": "test@example.com"})
    stub.configure_method("get_count", return_value=42)
    
    print("\nCalling stub methods...")
    result3 = pattern.call("DatabaseStub", "get_user", user_id=1)
    print(f"get_user result: {result3['result']}")
    
    result4 = pattern.call("DatabaseStub", "get_count")
    print(f"get_count result: {result4['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Fake Double (Simplified Implementation)")
    print("-" * 50)
    
    fake = pattern.create_fake("FakeEmailService")
    
    # Configure fake implementation
    def fake_send_email(to: str, subject: str, body: str = "") -> Dict[str, Any]:
        return {
            "status": "sent",
            "message_id": f"FAKE_{hash(to + subject)}",
            "to": to,
            "subject": subject
        }
    
    fake.configure_method("send_email", implementation=fake_send_email)
    
    print("\nCalling fake email service...")
    result5 = pattern.call("FakeEmailService", "send_email", 
                          to="user@example.com", 
                          subject="Test Email",
                          body="This is a test")
    print(f"Result: {result5['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 4: Spy Double (Record Calls)")
    print("-" * 50)
    
    spy = pattern.create_spy("MetricsSpy")
    
    # Configure spy implementation
    def track_metric(metric_name: str, value: float) -> Dict[str, Any]:
        return {
            "tracked": True,
            "metric": metric_name,
            "value": value,
            "timestamp": datetime.now().isoformat()
        }
    
    spy.configure_method("track", implementation=track_metric)
    
    print("\nCalling spy methods (recording calls)...")
    result6 = pattern.call("MetricsSpy", "track", metric_name="page_views", value=100.0)
    print(f"Result: {result6['result']}")
    
    result7 = pattern.call("MetricsSpy", "track", metric_name="conversions", value=5.0)
    print(f"Result: {result7['result']}")
    
    # Show recorded calls
    print(f"\nRecorded calls to 'track': {spy.call_counts['track']}")
    calls = spy.get_calls("track")
    for i, call in enumerate(calls, 1):
        print(f"  Call {i}: args={call.args}, kwargs={call.kwargs}")
    
    print("\n" + "=" * 50)
    print("\nScenario 5: Mock Double (Expectation Verification)")
    print("-" * 50)
    
    mock = pattern.create_mock("PaymentMock")
    
    # Set expectations
    mock.expect_call(
        "process_payment",
        times=2,
        returns={"status": "success", "transaction_id": "TXN123"}
    )
    
    print("\nCalling mock methods (with expectations)...")
    result8 = pattern.call("PaymentMock", "process_payment", amount=100.0)
    print(f"Result: {result8['result']}")
    
    result9 = pattern.call("PaymentMock", "process_payment", amount=50.0)
    print(f"Result: {result9['result']}")
    
    # Verify expectations
    print(f"\nVerifying 'process_payment' called 2 times: {mock.verify('process_payment', times=2)}")
    
    print("\n" + "=" * 50)
    print("\nScenario 6: Multiple Calls and Pattern Detection")
    print("-" * 50)
    
    multi_stub = pattern.create_stub("ConfigStub")
    multi_stub.configure_method("get_setting", return_value={"timeout": 30})
    
    print("\nMaking multiple calls to detect patterns...")
    for i in range(3):
        result = pattern.call("ConfigStub", "get_setting", key="timeout")
        print(f"Call {i+1}: {result['result']}")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"\nOverall Statistics:")
    print(f"  Total Doubles: {overall_stats['total_doubles']}")
    print(f"  Total Calls: {overall_stats['total_calls']}")
    print(f"  Verifications Passed: {overall_stats['verifications_passed']}")
    print(f"  Verifications Failed: {overall_stats['verifications_failed']}")
    
    print(f"\nIndividual Double Statistics:")
    for name, double in pattern.doubles.items():
        stats = double.get_statistics()
        print(f"\n{name} ({stats['double_type']}):")
        print(f"  Total Calls: {stats['total_calls']}")
        print(f"  Methods Configured: {stats['methods_configured']}")
        print(f"  Expectations Set: {stats['expectations_set']}")
    
    print("\n" + "=" * 50)
    print("Test double pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Test Double

Pattern 267: Test Double Pattern – Line-by-Line Explanation

Overview

This pattern implements comprehensive test double implementations (dummy, fake, spy, mock, stub) with intelligent orchestration and LLM-powered behavior analysis. Test doubles are objects that stand in for real objects during testing.

Code Explanation

Lines 1-17: Imports

"""
Pattern 267: Test Double Pattern with LangChain/LangGraph

Purpose: Comprehensive test double implementations (dummy, fake, spy, mock, stub).
Provides intelligent test double orchestration with LLM-powered behavior analysis.
"""

Explanation: Module docstring – this pattern provides all 5 types of test doubles from Martin Fowler’s testing taxonomy.

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json

Explanation: Standard imports plus json for parsing LLM responses.

Lines 19-30: Enum Definitions

class DoubleType(Enum):
    """Types of test doubles."""
    DUMMY = "dummy"  # No logic, fills parameter
    FAKE = "fake"  # Working implementation, simplified
    SPY = "spy"  # Records calls to real object
    MOCK = "mock"  # Pre-programmed expectations
    STUB = "stub"  # Pre-programmed responses

Explanation: Five test double types: – DUMMY: Fills parameter slots, never actually used (e.g., null object) – FAKE: Simplified working implementation (e.g., in-memory database) – SPY: Wraps real object and records calls for verification – MOCK: Has expectations about how it should be called – STUB: Returns canned responses (like pattern 266)

class VerificationStatus(Enum):
    """Status of expectation verification."""
    NOT_VERIFIED = "not_verified"
    PASSED = "passed"
    FAILED = "failed"

Explanation: Tracks whether mock expectations were verified.

Lines 32-71: Dataclasses

@dataclass
class Expectation:
    """Mock expectation."""
    method: str
    expected_calls: int
    expected_args: Optional[tuple] = None
    expected_kwargs: Optional[Dict[str, Any]] = None
    return_value: Any = None

Explanation: Defines what a mock expects: – How many times method should be called – What arguments it should receive – What it should return

Example: Expect process_payment called exactly 2 times with amount > 0.

@dataclass
class CallRecord:
    """Record of method call."""
    method: str
    args: tuple
    kwargs: Dict[str, Any]
    result: Any
    double_type: DoubleType
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Records each method invocation for spies and verification.

@dataclass
class DoubleAnalysis:
    """Analysis of test double usage."""
    recommended_type: DoubleType
    reasoning: str
    complexity_score: float
    usage_patterns: List[str]

Explanation: LLM analysis recommending best double type based on usage patterns.

Lines 73-85: TypedDict State

class TestDoubleState(TypedDict):
    """State for test double workflow."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    double_type: DoubleType
    analysis: Optional[DoubleAnalysis]
    result: Any
    verification_status: VerificationStatus
    expectations_met: bool
    insights: List[str]
    call_count: int

Explanation: Workflow state tracking method call, analysis, result, and verification status.

Lines 87-137: IntelligentTestDouble Initialization

class IntelligentTestDouble:
    """
    Intelligent Test Double with LangGraph workflow integration.

    Features:
    - All test double types (dummy, fake, spy, mock, stub)
    - LLM-powered behavior analysis
    - Expectation verification
    - Call recording and replay
    - Pattern detection
    - Intelligent recommendations
    """

Explanation: Main class implementing all 5 double types with AI analysis.

def __init__(self, name: str, double_type: DoubleType, model_name: str = "gpt-4", temperature: float = 0.5):
    self.name = name
    self.double_type = double_type
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Configuration
    self.methods: Dict[str, Any] = {}
    self.expectations: Dict[str, List[Expectation]] = defaultdict(list)
    self.fake_implementations: Dict[str, Callable] = {}

Explanation: Initialize with: – Name and type of double – LLM for analysis – Method configurations – Mock expectations – Fake implementations (actual functions)

    # Tracking
    self.call_records: List[CallRecord] = []
    self.call_counts: Dict[str, int] = defaultdict(int)

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_calls": 0,
        "verifications_passed": 0,
        "verifications_failed": 0,
        "double_type": double_type.value
    }

Explanation: Initialize call tracking, workflow, and statistics.

Lines 139-157: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(TestDoubleState)

    workflow.add_node("analyze_double_type", self._analyze_double_type_node)
    workflow.add_node("configure_behavior", self._configure_behavior_node)
    workflow.add_node("execute_call", self._execute_call_node)
    workflow.add_node("verify_expectations", self._verify_expectations_node)
    workflow.add_node("generate_report", self._generate_report_node)

Explanation: Create 5-node workflow: 1. Analyze what type of double is needed 2. Configure behavior based on type 3. Execute the call 4. Verify expectations (for mocks) 5. Generate report

    workflow.set_entry_point("analyze_double_type")
    workflow.add_edge("analyze_double_type", "configure_behavior")
    workflow.add_edge("configure_behavior", "execute_call")
    workflow.add_edge("execute_call", "verify_expectations")
    workflow.add_edge("verify_expectations", "generate_report")
    workflow.add_edge("generate_report", END)

    return workflow.compile()

Explanation: Define linear flow through all nodes.

Lines 159-216: Analyze Double Type Node (LLM-Powered)

def _analyze_double_type_node(self, state: TestDoubleState) -> TestDoubleState:
    """Analyze test double type and usage."""
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]

    # Increment call count
    self.call_counts[method_name] += 1
    state["call_count"] = self.call_counts[method_name]

Explanation: Extract parameters and update call counter.

    # Use LLM to analyze if this is first call to method
    if self.call_counts[method_name] == 1 and not self.methods.get(method_name):
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in test double patterns and testing strategies."),
            ("user", """Analyze this method call for test double usage:

Method: {method_name}
Arguments: {args}
Keyword Arguments: {kwargs}
Current Double Type: {double_type}

Provide analysis as JSON with:
- recommended_type: Best double type for this method
- reasoning: Why this type is recommended
- complexity_score: 0.0-1.0 score for implementation complexity
- usage_patterns: List of observed usage patterns

Double types: dummy (parameter filler), fake (simplified working), spy (record calls), mock (verify expectations), stub (fixed responses)""")
        ])

Explanation: On first call, ask LLM to analyze method and recommend best double type.

        chain = prompt | self.llm
        response = chain.invoke({
            "method_name": method_name,
            "args": str(args),
            "kwargs": str(kwargs),
            "double_type": self.double_type.value
        })

        try:
            analysis_data = json.loads(response.content)
            state["analysis"] = DoubleAnalysis(
                recommended_type=DoubleType(analysis_data.get("recommended_type", self.double_type.value)),
                reasoning=analysis_data.get("reasoning", "Default reasoning"),
                complexity_score=analysis_data.get("complexity_score", 0.5),
                usage_patterns=analysis_data.get("usage_patterns", [])
            )
        except:
            state["analysis"] = DoubleAnalysis(
                recommended_type=self.double_type,
                reasoning="Using configured double type",
                complexity_score=0.5,
                usage_patterns=[]
            )

Explanation: Invoke LLM and parse JSON response into analysis object. Fallback to current type if parsing fails.

    return state

Lines 218-230: Configure Behavior Node

def _configure_behavior_node(self, state: TestDoubleState) -> TestDoubleState:
    """Configure double behavior."""
    method_name = state["method_name"]

    # Store method configuration if not exists
    if method_name not in self.methods:
        self.methods[method_name] = {
            "return_value": None,
            "configured": False
        }

    return state

Explanation: Ensure method has configuration entry. Simple node that prepares for execution.

Lines 232-291: Execute Call Node

def _execute_call_node(self, state: TestDoubleState) -> TestDoubleState:
    """Execute the test double call."""
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]
    double_type = state["double_type"]

    result = None

Explanation: Extract parameters and initialize result.

    if double_type == DoubleType.DUMMY:
        # Dummy: Do nothing, return None
        result = None

Explanation: DUMMY – simplest double, does nothing. Used when object is required but never actually used.

    elif double_type == DoubleType.STUB:
        # Stub: Return configured value
        method_config = self.methods.get(method_name, {})
        result = method_config.get("return_value")

Explanation: STUB – returns pre-configured value. No logic, just returns what you told it to.

    elif double_type == DoubleType.FAKE:
        # Fake: Use simplified implementation
        if method_name in self.fake_implementations:
            try:
                result = self.fake_implementations[method_name](*args, **kwargs)
            except Exception as e:
                result = {"error": str(e)}
        else:
            # Default fake behavior
            result = {"status": "fake_success", "method": method_name, "data": "fake_data"}

Explanation: FAKE – has working implementation, but simplified (e.g., in-memory instead of real database). Actually executes logic.

    elif double_type == DoubleType.SPY:
        # Spy: Record call and execute real implementation if available
        if method_name in self.fake_implementations:
            try:
                result = self.fake_implementations[method_name](*args, **kwargs)
            except Exception as e:
                result = {"error": str(e)}
        else:
            result = {"status": "spy_recorded", "method": method_name}

Explanation: SPY – like fake but emphasizes call recording. Executes real code while tracking calls.

    elif double_type == DoubleType.MOCK:
        # Mock: Return expected value if configured
        expectations = self.expectations.get(method_name, [])
        if expectations:
            # Use first expectation's return value
            result = expectations[0].return_value
        else:
            result = {"status": "mock_success", "method": method_name}

Explanation: MOCK – returns value from expectations. Unlike stub, mock will be verified later.

    state["result"] = result

    # Record call
    call_record = CallRecord(
        method=method_name,
        args=args,
        kwargs=kwargs,
        result=result,
        double_type=double_type
    )
    self.call_records.append(call_record)
    self.stats["total_calls"] += 1

    return state

Explanation: Store result, record call in history, update statistics.

Lines 293-346: Verify Expectations Node

def _verify_expectations_node(self, state: TestDoubleState) -> TestDoubleState:
    """Verify expectations for mocks."""
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]

    expectations_met = True

Explanation: Initialize verification – only meaningful for mocks.

    if state["double_type"] == DoubleType.MOCK:
        expectations = self.expectations.get(method_name, [])

        for expectation in expectations:
            # Check call count
            actual_calls = self.call_counts[method_name]
            if actual_calls != expectation.expected_calls:
                expectations_met = False
                state["insights"] = state.get("insights", [])
                state["insights"].append(
                    f"Expected {expectation.expected_calls} calls to {method_name}, got {actual_calls}"
                )

Explanation: Verify mock was called expected number of times. This is the key difference between mocks and stubs – mocks verify behavior.

            # Check arguments if specified
            if expectation.expected_args is not None and args != expectation.expected_args:
                expectations_met = False
                state["insights"] = state.get("insights", [])
                state["insights"].append(
                    f"Expected args {expectation.expected_args}, got {args}"
                )

            if expectation.expected_kwargs is not None and kwargs != expectation.expected_kwargs:
                expectations_met = False
                state["insights"] = state.get("insights", [])
                state["insights"].append(
                    f"Expected kwargs {expectation.expected_kwargs}, got {kwargs}"
                )

Explanation: Verify arguments match expectations if specified.

    state["expectations_met"] = expectations_met
    state["verification_status"] = VerificationStatus.PASSED if expectations_met else VerificationStatus.FAILED

    if expectations_met:
        self.stats["verifications_passed"] += 1
    else:
        self.stats["verifications_failed"] += 1

    return state

Explanation: Set verification status and update statistics.

Lines 348-361: Generate Report Node

def _generate_report_node(self, state: TestDoubleState) -> TestDoubleState:
    """Generate insights report."""
    if not state.get("insights"):
        state["insights"] = []

    # Add summary insight
    state["insights"].append(
        f"Test double '{self.name}' of type {state['double_type'].value} executed {state['method_name']}"
    )

    return state

Explanation: Add summary insight to state. Simple reporting node.

Lines 363-403: Public Configuration Methods

def configure_method(
    self,
    method_name: str,
    return_value: Any = None,
    implementation: Optional[Callable] = None
):
    """Configure method behavior."""
    self.methods[method_name] = {
        "return_value": return_value,
        "configured": True
    }

    if implementation:
        self.fake_implementations[method_name] = implementation

Explanation: Configure what method should return (for stubs) or how it should behave (for fakes – actual function).

def expect_call(
    self,
    method_name: str,
    times: int = 1,
    with_args: Optional[tuple] = None,
    with_kwargs: Optional[Dict[str, Any]] = None,
    returns: Any = None
):
    """Set expectation for mock."""
    expectation = Expectation(
        method=method_name,
        expected_calls=times,
        expected_args=with_args,
        expected_kwargs=with_kwargs,
        return_value=returns
    )
    self.expectations[method_name].append(expectation)

Explanation: Set expectations for mocks. Used in test setup phase. Example: mock.expect_call("save", times=1, with_args=("data",)).

def call_method(self, method_name: str, *args, **kwargs) -> Any:
    """Execute method call through workflow."""
    initial_state: TestDoubleState = {
        "method_name": method_name,
        "args": args,
        "kwargs": kwargs,
        "double_type": self.double_type,
        "analysis": None,
        "result": None,
        "verification_status": VerificationStatus.NOT_VERIFIED,
        "expectations_met": True,
        "insights": [],
        "call_count": 0
    }

    final_state = self.workflow.invoke(initial_state)
    return final_state["result"]

Explanation: Main entry point – execute workflow and return result.

Lines 428-475: TestDoublePattern Coordinator

class TestDoublePattern:
    """
    Comprehensive Test Double Pattern coordinator with LangChain/LangGraph.
    """

    def __init__(self, model_name: str = "gpt-4"):
        self.doubles: Dict[str, IntelligentTestDouble] = {}
        self.model_name = model_name

Explanation: Coordinator managing multiple test doubles.

    def create_dummy(self, name: str) -> IntelligentTestDouble:
        """Create a dummy double."""
        double = IntelligentTestDouble(name, DoubleType.DUMMY, model_name=self.model_name)
        self.doubles[name] = double
        return double

Explanation: Factory methods for each double type. Clear, explicit API.

Similar factory methods for stub, fake, spy, and mock types follow the same pattern.

Lines 477-675: Demo Code

Scenario 1: Dummy Double

dummy = pattern.create_dummy("DummyLogger")
result1 = pattern.call("DummyLogger", "log", message="Test message")

Explanation: Dummy does nothing, returns None. Used when parameter is required but not used (e.g., logger in test).

Scenario 2: Stub Double

stub = pattern.create_stub("DatabaseStub")
stub.configure_method("get_user", return_value={"id": 1, "name": "Test User", "email": "test@example.com"})

Explanation: Stub returns fixed value. Good for tests that just need data, don’t care about interactions.

Scenario 3: Fake Double

fake = pattern.create_fake("FakeEmailService")

def fake_send_email(to: str, subject: str, body: str = "") -> Dict[str, Any]:
    return {
        "status": "sent",
        "message_id": f"FAKE_{hash(to + subject)}",
        "to": to,
        "subject": subject
    }

fake.configure_method("send_email", implementation=fake_send_email)

Explanation: Fake has working implementation (simplified). Actually processes inputs, but doesn’t hit real email server.

Scenario 4: Spy Double

spy = pattern.create_spy("MetricsSpy")

def track_metric(metric_name: str, value: float) -> Dict[str, Any]:
    return {
        "tracked": True,
        "metric": metric_name,
        "value": value,
        "timestamp": datetime.now().isoformat()
    }

spy.configure_method("track", implementation=track_metric)

# Later, verify calls were made
calls = spy.get_calls("track")
for i, call in enumerate(calls, 1):
    print(f"  Call {i}: args={call.args}, kwargs={call.kwargs}")

Explanation: Spy executes real code AND records calls. Used to verify side effects.

Scenario 5: Mock Double

mock = pattern.create_mock("PaymentMock")

# Set expectations
mock.expect_call(
    "process_payment",
    times=2,
    returns={"status": "success", "transaction_id": "TXN123"}
)

# Call twice
result8 = pattern.call("PaymentMock", "process_payment", amount=100.0)
result9 = pattern.call("PaymentMock", "process_payment", amount=50.0)

# Verify expectations
print(f"\nVerifying 'process_payment' called 2 times: {mock.verify('process_payment', times=2)}")

Explanation: Mock has expectations that must be met. Test fails if method not called expected times. Key difference from stub – verification is mandatory.

Test Double Comparison

When to Use Each

• Dummy: Parameter required but not used (null object pattern)
• Stub: Need data, don’t care about interactions
• Fake: Need working implementation but real one is too slow/complex
• Spy: Need to verify side effects while using real code
• Mock: Need to verify specific interactions occurred

Key Design Patterns

1. Strategy Pattern: Different double types implement different strategies
2. Template Method: Common workflow with type-specific execution
3. Observer Pattern: Call recording observes all invocations
4. Proxy Pattern: Doubles proxy real objects
5. Factory Pattern: Coordinator creates appropriate doubles

Performance Considerations

• Dummy/Stub/Mock: O(1) – instant
• Fake/Spy: O(implementation) – depends on fake logic
• Verification: O(calls × expectations) – linear in recorded calls
• Memory: O(calls) – stores all call records

Extension Points

1. Add partial mocks (some methods real, some mocked)
2. Implement argument matchers (any(), contains(), etc.)
3. Add sequence verification (call A before call B)
4. Implement call chains for fluent APIs
5. Add async/await support for async methods

    """
Pattern 266: Stub Pattern with LangChain/LangGraph

Purpose: Minimal test implementations with controlled, predetermined responses.
Provides intelligent stub selection and response generation with LLM analysis.
"""

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json


class StubType(Enum):
    """Stub implementation types."""
    SIMPLE = "simple"  # Return fixed value
    SEQUENCE = "sequence"  # Return values in sequence
    CONDITIONAL = "conditional"  # Return based on input
    EXCEPTION = "exception"  # Raise exception
    NULL = "null"  # Return None
    DYNAMIC = "dynamic"  # LLM-generated response


class StubComplexity(Enum):
    """Complexity of stub response."""
    TRIVIAL = "trivial"
    SIMPLE = "simple"
    MODERATE = "moderate"
    COMPLEX = "complex"


@dataclass
class StubResponse:
    """Stub response configuration."""
    method_name: str
    stub_type: StubType
    return_value: Any = None
    return_sequence: List[Any] = field(default_factory=list)
    condition: Optional[Callable] = None
    exception: Optional[Exception] = None
    complexity: StubComplexity = StubComplexity.SIMPLE


@dataclass
class StubCall:
    """Record of stub call."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    result: Any
    stub_type: StubType
    call_number: int
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class StubAnalysis:
    """Analysis of stub behavior."""
    recommended_type: StubType
    complexity: StubComplexity
    reasoning: str
    alternative_approaches: List[str]


class StubState(TypedDict):
    """State for stub workflow."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    stub_config: Optional[StubResponse]
    recommended_type: Optional[StubType]
    complexity: Optional[StubComplexity]
    analysis: Optional[StubAnalysis]
    result: Any
    validation_passed: bool
    insights: List[str]
    call_number: int


class IntelligentStubManager:
    """
    Intelligent Stub Manager with LangGraph workflow integration.
    
    Features:
    - Intelligent stub type selection
    - LLM-powered response generation
    - Dynamic stub behavior
    - Call analysis and insights
    - Complexity assessment
    - Pattern detection
    """
    
    def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        
        # Stub configurations
        self.stubs: Dict[str, StubResponse] = {}
        
        # Call history
        self.call_history: List[StubCall] = []
        self.call_counts: Dict[str, int] = defaultdict(int)
        self.sequence_indices: Dict[str, int] = defaultdict(int)
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_calls": 0,
            "stub_types": {t.value: 0 for t in StubType},
            "methods_stubbed": 0
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(StubState)
        
        workflow.add_node("analyze_request", self._analyze_request_node)
        workflow.add_node("select_stub_type", self._select_stub_type_node)
        workflow.add_node("generate_response", self._generate_response_node)
        workflow.add_node("validate_response", self._validate_response_node)
        workflow.add_node("record_call", self._record_call_node)
        
        workflow.set_entry_point("analyze_request")
        workflow.add_edge("analyze_request", "select_stub_type")
        workflow.add_edge("select_stub_type", "generate_response")
        workflow.add_edge("generate_response", "validate_response")
        workflow.add_edge("validate_response", "record_call")
        workflow.add_edge("record_call", END)
        
        return workflow.compile()
    
    def _analyze_request_node(self, state: StubState) -> StubState:
        """Analyze the stub request."""
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        
        # Get stub configuration if exists
        stub_config = self.stubs.get(method_name)
        state["stub_config"] = stub_config
        
        # Increment call count
        self.call_counts[method_name] += 1
        state["call_number"] = self.call_counts[method_name]
        
        # If no stub configured, use LLM to analyze
        if not stub_config:
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert in test stub design and testing patterns."),
                ("user", """Analyze this method call and recommend an appropriate stub type:

Method: {method_name}
Arguments: {args}
Keyword Arguments: {kwargs}

Recommend:
1. Best stub type (simple, sequence, conditional, dynamic, null, exception)
2. Complexity level (trivial, simple, moderate, complex)
3. Reasoning for recommendation
4. Alternative approaches

Provide response as JSON with keys: recommended_type, complexity, reasoning, alternatives""")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "method_name": method_name,
                "args": str(args),
                "kwargs": str(kwargs)
            })
            
            try:
                analysis_data = json.loads(response.content)
                state["analysis"] = StubAnalysis(
                    recommended_type=StubType(analysis_data.get("recommended_type", "simple")),
                    complexity=StubComplexity(analysis_data.get("complexity", "simple")),
                    reasoning=analysis_data.get("reasoning", "No reasoning provided"),
                    alternative_approaches=analysis_data.get("alternatives", [])
                )
            except:
                # Fallback
                state["analysis"] = StubAnalysis(
                    recommended_type=StubType.SIMPLE,
                    complexity=StubComplexity.SIMPLE,
                    reasoning="Default simple stub",
                    alternative_approaches=[]
                )
        
        return state
    
    def _select_stub_type_node(self, state: StubState) -> StubState:
        """Select appropriate stub type."""
        stub_config = state["stub_config"]
        
        if stub_config:
            # Use configured stub type
            state["recommended_type"] = stub_config.stub_type
            state["complexity"] = stub_config.complexity
        else:
            # Use analyzed recommendation
            analysis = state.get("analysis")
            if analysis:
                state["recommended_type"] = analysis.recommended_type
                state["complexity"] = analysis.complexity
            else:
                state["recommended_type"] = StubType.SIMPLE
                state["complexity"] = StubComplexity.SIMPLE
        
        return state
    
    def _generate_response_node(self, state: StubState) -> StubState:
        """Generate stub response."""
        stub_type = state["recommended_type"]
        stub_config = state["stub_config"]
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        
        result = None
        
        try:
            if stub_type == StubType.EXCEPTION:
                if stub_config and stub_config.exception:
                    raise stub_config.exception
                else:
                    raise Exception(f"Stub exception for {method_name}")
            
            elif stub_type == StubType.NULL:
                result = None
            
            elif stub_type == StubType.SIMPLE:
                if stub_config:
                    result = stub_config.return_value
                else:
                    result = {"status": "stubbed", "method": method_name}
            
            elif stub_type == StubType.SEQUENCE:
                if stub_config and stub_config.return_sequence:
                    idx = self.sequence_indices[method_name]
                    if idx  StubState:
        """Validate generated response."""
        result = state["result"]
        stub_type = state["recommended_type"]
        
        # Basic validation
        validation_passed = True
        
        if stub_type == StubType.NULL:
            validation_passed = result is None
        elif stub_type == StubType.EXCEPTION:
            validation_passed = isinstance(result, dict) and "error" in result
        elif stub_type == StubType.SIMPLE:
            validation_passed = result is not None
        
        state["validation_passed"] = validation_passed
        
        if not validation_passed:
            state["insights"] = state.get("insights", [])
            state["insights"].append(f"Validation failed for {stub_type.value} stub")
        
        return state
    
    def _record_call_node(self, state: StubState) -> StubState:
        """Record the stub call."""
        call = StubCall(
            method_name=state["method_name"],
            args=state["args"],
            kwargs=state["kwargs"],
            result=state["result"],
            stub_type=state["recommended_type"],
            call_number=state["call_number"]
        )
        
        self.call_history.append(call)
        self.stats["total_calls"] += 1
        self.stats["stub_types"][state["recommended_type"].value] += 1
        
        return state
    
    def register_stub(
        self,
        method_name: str,
        stub_type: StubType = StubType.SIMPLE,
        return_value: Any = None,
        return_sequence: List[Any] = None,
        condition: Optional[Callable] = None,
        exception: Optional[Exception] = None,
        complexity: StubComplexity = StubComplexity.SIMPLE
    ):
        """Register a stub configuration."""
        stub = StubResponse(
            method_name=method_name,
            stub_type=stub_type,
            return_value=return_value,
            return_sequence=return_sequence or [],
            condition=condition,
            exception=exception,
            complexity=complexity
        )
        
        self.stubs[method_name] = stub
        self.stats["methods_stubbed"] = len(self.stubs)
    
    def call(self, method_name: str, *args, **kwargs) -> Any:
        """Execute stub method call through workflow."""
        initial_state: StubState = {
            "method_name": method_name,
            "args": args,
            "kwargs": kwargs,
            "stub_config": None,
            "recommended_type": None,
            "complexity": None,
            "analysis": None,
            "result": None,
            "validation_passed": False,
            "insights": [],
            "call_number": 0
        }
        
        final_state = self.workflow.invoke(initial_state)
        return final_state["result"]
    
    def get_call_history(self, method_name: str = None) -> List[StubCall]:
        """Get call history."""
        if method_name:
            return [call for call in self.call_history if call.method_name == method_name]
        return self.call_history
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get stub statistics."""
        return {
            "name": self.name,
            **self.stats,
            "unique_methods": len(set(call.method_name for call in self.call_history))
        }
    
    def reset(self):
        """Reset stub state."""
        self.call_history.clear()
        self.call_counts.clear()
        self.sequence_indices.clear()
        self.stats["total_calls"] = 0
        self.stats["stub_types"] = {t.value: 0 for t in StubType}


class StubPattern:
    """
    Comprehensive Stub Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.managers: Dict[str, IntelligentStubManager] = {}
        self.model_name = model_name
    
    def create_manager(self, name: str, target_class: str = "") -> IntelligentStubManager:
        """Create a stub manager."""
        manager = IntelligentStubManager(name, model_name=self.model_name)
        self.managers[name] = manager
        return manager
    
    def call(self, manager_name: str, method_name: str, *args, **kwargs) -> Dict[str, Any]:
        """Execute stub call."""
        if manager_name not in self.managers:
            return {"error": f"Manager {manager_name} not found"}
        
        manager = self.managers[manager_name]
        result = manager.call(method_name, *args, **kwargs)
        
        return {
            "manager": manager_name,
            "method": method_name,
            "result": result,
            "call_count": manager.call_counts[method_name]
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_calls = sum(m.stats["total_calls"] for m in self.managers.values())
        total_methods = sum(m.stats["methods_stubbed"] for m in self.managers.values())
        
        return {
            "total_managers": len(self.managers),
            "total_calls": total_calls,
            "total_methods_stubbed": total_methods
        }


if __name__ == "__main__":
    print("=== Stub Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Database Stub with Sequence")
    print("-" * 50)
    pattern = StubPattern(model_name="gpt-4")
    
    db_manager = pattern.create_manager("DatabaseStub", "Database")
    
    # Register sequence stub
    db_manager.register_stub(
        "get_users",
        stub_type=StubType.SEQUENCE,
        return_sequence=[
            [{"id": 1, "name": "Alice"}],
            [{"id": 2, "name": "Bob"}],
            [{"id": 3, "name": "Charlie"}]
        ],
        complexity=StubComplexity.SIMPLE
    )
    
    print("\nExecuting sequence stub calls...")
    for i in range(3):
        result = pattern.call("DatabaseStub", "get_users")
        print(f"Call {i+1}: {result['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 2: API Stub with Conditional Logic")
    print("-" * 50)
    
    api_manager = pattern.create_manager("APIStub", "ExternalAPI")
    
    # Register conditional stub
    api_manager.register_stub(
        "fetch_data",
        stub_type=StubType.CONDITIONAL,
        return_value={"status": "success", "data": "test_data"},
        condition=lambda *args, **kwargs: kwargs.get("authenticated", False),
        complexity=StubComplexity.MODERATE
    )
    
    print("\nAuthenticated call:")
    result1 = pattern.call("APIStub", "fetch_data", authenticated=True)
    print(f"Result: {result1['result']}")
    
    print("\nUnauthenticated call:")
    result2 = pattern.call("APIStub", "fetch_data", authenticated=False)
    print(f"Result: {result2['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Dynamic Stub with LLM Generation")
    print("-" * 50)
    
    service_manager = pattern.create_manager("ServiceStub", "PaymentService")
    
    # Register dynamic stub (no configuration - uses LLM)
    print("\nCalling unconfigured method (LLM will generate response)...")
    result3 = pattern.call("ServiceStub", "process_payment", 
                          amount=100.0, currency="USD", user_id=123)
    print(f"Result: {result3['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 4: Exception Stub")
    print("-" * 50)
    
    error_manager = pattern.create_manager("ErrorStub", "FailingService")
    
    # Register exception stub
    error_manager.register_stub(
        "dangerous_operation",
        stub_type=StubType.EXCEPTION,
        exception=Exception("Operation not allowed in test mode"),
        complexity=StubComplexity.TRIVIAL
    )
    
    print("\nCalling exception stub...")
    result4 = pattern.call("ErrorStub", "dangerous_operation")
    print(f"Result: {result4['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 5: Simple Fixed Response Stub")
    print("-" * 50)
    
    config_manager = pattern.create_manager("ConfigStub", "ConfigService")
    
    # Register simple stub
    config_manager.register_stub(
        "get_config",
        stub_type=StubType.SIMPLE,
        return_value={"timeout": 30, "retries": 3, "debug": True},
        complexity=StubComplexity.TRIVIAL
    )
    
    print("\nCalling simple stub...")
    result5 = pattern.call("ConfigStub", "get_config")
    print(f"Result: {result5['result']}")
    
    print("\n" + "=" * 50)
    print("\nCall History Analysis")
    print("-" * 50)
    
    # Show call history for database stub
    db_history = db_manager.get_call_history("get_users")
    print(f"\nDatabase 'get_users' call history ({len(db_history)} calls):")
    for call in db_history:
        print(f"  Call #{call.call_number}: Type={call.stub_type.value}, Result={call.result}")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"Overall Statistics:")
    print(f"  Total Managers: {overall_stats['total_managers']}")
    print(f"  Total Calls: {overall_stats['total_calls']}")
    print(f"  Total Methods Stubbed: {overall_stats['total_methods_stubbed']}")
    
    print(f"\nDatabase Stub Statistics:")
    db_stats = db_manager.get_statistics()
    for key, value in db_stats.items():
        print(f"  {key}: {value}")
    
    print(f"\nAPI Stub Statistics:")
    api_stats = api_manager.get_statistics()
    for key, value in api_stats.items():
        print(f"  {key}: {value}")
    
    print("\n" + "=" * 50)
    print("Stub pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Stub

Pattern 266: Stub Pattern – Line-by-Line Explanation

Overview

This pattern implements minimal test implementations with controlled, predetermined responses. It provides intelligent stub selection and response generation with LLM analysis for testing without real dependencies.

Code Explanation

Lines 1-17: Imports and Setup

"""
Pattern 266: Stub Pattern with LangChain/LangGraph

Purpose: Minimal test implementations with controlled, predetermined responses.
Provides intelligent stub selection and response generation with LLM analysis.
"""

Explanation: Module docstring describing the pattern’s purpose – creating test stubs that replace real implementations with controlled, predictable responses.

from typing import Dict, Any, List, Optional, Callable, TypedDict

Explanation: Import type hints. Callable is used for function types (validators, conditions).

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict

Explanation: Import dataclasses, datetime for timestamps, Enum for type-safe enumerations, and defaultdict for automatic dictionary initialization.

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json

Explanation: Import LangChain/LangGraph components and json for parsing LLM responses.

Lines 19-32: Enum Definitions

class StubType(Enum):
    """Stub implementation types."""
    SIMPLE = "simple"  # Return fixed value
    SEQUENCE = "sequence"  # Return values in sequence
    CONDITIONAL = "conditional"  # Return based on input
    EXCEPTION = "exception"  # Raise exception
    NULL = "null"  # Return None
    DYNAMIC = "dynamic"  # LLM-generated response

Explanation: Enum defining 6 stub types: – SIMPLE: Always returns same fixed value – SEQUENCE: Returns different values on successive calls – CONDITIONAL: Returns value based on input conditions – EXCEPTION: Simulates error by raising exception – NULL: Always returns None (for optional parameters) – DYNAMIC: Uses LLM to generate realistic responses

class StubComplexity(Enum):
    """Complexity of stub response."""
    TRIVIAL = "trivial"
    SIMPLE = "simple"
    MODERATE = "moderate"
    COMPLEX = "complex"

Explanation: Enum classifying stub implementation complexity from trivial (basic return) to complex (conditional logic).

Lines 34-62: Dataclasses

@dataclass
class StubResponse:
    """Stub response configuration."""
    method_name: str
    stub_type: StubType
    return_value: Any = None
    return_sequence: List[Any] = field(default_factory=list)
    condition: Optional[Callable] = None
    exception: Optional[Exception] = None
    complexity: StubComplexity = StubComplexity.SIMPLE

Explanation: Configuration for a stub method: – method_name: Method being stubbed (e.g., “get_user”) – stub_type: Type from StubType enum – return_value: Value to return (for SIMPLE/CONDITIONAL) – return_sequence: List of values (for SEQUENCE) – condition: Function to evaluate (for CONDITIONAL) – exception: Exception to raise (for EXCEPTION) – complexity: Implementation complexity level

@dataclass
class StubCall:
    """Record of stub call."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    result: Any
    stub_type: StubType
    call_number: int
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Record of each stub invocation with: – Arguments passed – Result returned – Type of stub used – Call sequence number – Timestamp for tracking

@dataclass
class StubAnalysis:
    """Analysis of stub behavior."""
    recommended_type: StubType
    complexity: StubComplexity
    reasoning: str
    alternative_approaches: List[str]

Explanation: LLM-generated analysis recommending best stub type, complexity assessment, reasoning, and alternative approaches.

Lines 64-77: TypedDict State

class StubState(TypedDict):
    """State for stub workflow."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    stub_config: Optional[StubResponse]
    recommended_type: Optional[StubType]
    complexity: Optional[StubComplexity]
    analysis: Optional[StubAnalysis]
    result: Any
    validation_passed: bool
    insights: List[str]
    call_number: int

Explanation: Workflow state containing: – Method call details (name, args, kwargs) – Configuration and recommendations – Analysis results – Generated result – Validation status – Call tracking

Lines 79-129: IntelligentStubManager Initialization

class IntelligentStubManager:
    """
    Intelligent Stub Manager with LangGraph workflow integration.

    Features:
    - Intelligent stub type selection
    - LLM-powered response generation
    - Dynamic stub behavior
    - Call analysis and insights
    - Complexity assessment
    - Pattern detection
    """

Explanation: Main class for managing stubs with AI-powered selection and generation.

def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
    self.name = name
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Stub configurations
    self.stubs: Dict[str, StubResponse] = {}

    # Call history
    self.call_history: List[StubCall] = []
    self.call_counts: Dict[str, int] = defaultdict(int)
    self.sequence_indices: Dict[str, int] = defaultdict(int)

Explanation: Initialize with: – Name for identification – LLM for intelligent analysis – Stub configurations dictionary – Call tracking (history, counts, sequence indices)

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_calls": 0,
        "stub_types": {t.value: 0 for t in StubType},
        "methods_stubbed": 0
    }

Explanation: Build workflow and initialize statistics tracking.

Lines 131-149: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(StubState)

    workflow.add_node("analyze_request", self._analyze_request_node)
    workflow.add_node("select_stub_type", self._select_stub_type_node)
    workflow.add_node("generate_response", self._generate_response_node)
    workflow.add_node("validate_response", self._validate_response_node)
    workflow.add_node("record_call", self._record_call_node)

Explanation: Create 5-node workflow for processing stub calls.

    workflow.set_entry_point("analyze_request")
    workflow.add_edge("analyze_request", "select_stub_type")
    workflow.add_edge("select_stub_type", "generate_response")
    workflow.add_edge("generate_response", "validate_response")
    workflow.add_edge("validate_response", "record_call")
    workflow.add_edge("record_call", END)

    return workflow.compile()

Explanation: Define linear flow: analyze → select → generate → validate → record → end.

Lines 151-206: Analyze Request Node (LLM-Powered)

def _analyze_request_node(self, state: StubState) -> StubState:
    """Analyze the stub request."""
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]

    # Get stub configuration if exists
    stub_config = self.stubs.get(method_name)
    state["stub_config"] = stub_config

    # Increment call count
    self.call_counts[method_name] += 1
    state["call_number"] = self.call_counts[method_name]