"""
Pattern 270: Verification Pattern with LangChain/LangGraph

Purpose: Correctness verification and property checking with formal methods.
Provides intelligent verification orchestration with LLM-powered proof 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 VerificationType(Enum):
    """Types of verification."""
    PROPERTY = "property"
    INVARIANT = "invariant"
    PRECONDITION = "precondition"
    POSTCONDITION = "postcondition"
    ASSERTION = "assertion"
    CONTRACT = "contract"


class VerificationSeverity(Enum):
    """Severity of verification issues."""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@dataclass
class VerificationCondition:
    """Verification condition definition."""
    name: str
    verification_type: VerificationType
    condition: Callable
    description: str
    severity: VerificationSeverity = VerificationSeverity.CRITICAL


@dataclass
class VerificationResult:
    """Result of verification."""
    condition_name: str
    verification_type: VerificationType
    passed: bool
    message: str
    severity: VerificationSeverity
    evidence: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class VerificationProof:
    """LLM-generated verification proof."""
    formal_statement: str
    proof_sketch: str
    assumptions: List[str]
    conclusions: List[str]
    confidence: float


class VerificationState(TypedDict):
    """State for verification workflow."""
    context: Dict[str, Any]
    conditions: List[VerificationCondition]
    results: List[VerificationResult]
    all_passed: bool
    failures: List[VerificationResult]
    proof: Optional[VerificationProof]
    recommendations: List[str]
    verification_summary: Dict[str, Any]


class IntelligentVerifier:
    """
    Intelligent Verifier with LangGraph workflow integration.
    
    Features:
    - Property verification
    - Invariant checking
    - Pre/postcondition verification
    - Contract checking
    - LLM-powered proof generation
    - Evidence collection
    - Formal reasoning support
    - Counterexample 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)
        
        # Conditions
        self.conditions: List[VerificationCondition] = []
        
        # History
        self.verification_history: List[VerificationResult] = []
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_verifications": 0,
            "passed": 0,
            "failed": 0,
            "verification_types": {t.value: 0 for t in VerificationType}
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(VerificationState)
        
        workflow.add_node("analyze_conditions", self._analyze_conditions_node)
        workflow.add_node("verify_properties", self._verify_properties_node)
        workflow.add_node("check_invariants", self._check_invariants_node)
        workflow.add_node("collect_evidence", self._collect_evidence_node)
        workflow.add_node("generate_proof", self._generate_proof_node)
        
        workflow.set_entry_point("analyze_conditions")
        workflow.add_edge("analyze_conditions", "verify_properties")
        workflow.add_edge("verify_properties", "check_invariants")
        workflow.add_edge("check_invariants", "collect_evidence")
        workflow.add_edge("collect_evidence", "generate_proof")
        workflow.add_edge("generate_proof", END)
        
        return workflow.compile()
    
    def _analyze_conditions_node(self, state: VerificationState) -> VerificationState:
        """Analyze verification conditions."""
        state["conditions"] = self.conditions
        state["results"] = []
        state["failures"] = []
        
        return state
    
    def _verify_properties_node(self, state: VerificationState) -> VerificationState:
        """Verify properties and conditions."""
        context = state["context"]
        conditions = state["conditions"]
        results = []
        
        for condition in conditions:
            self.stats["total_verifications"] += 1
            self.stats["verification_types"][condition.verification_type.value] += 1
            
            try:
                passed = condition.condition(context)
                
                message = (
                    f"✓ {condition.description}" if passed
                    else f"✗ {condition.description} FAILED"
                )
                
                result = VerificationResult(
                    condition_name=condition.name,
                    verification_type=condition.verification_type,
                    passed=passed,
                    message=message,
                    severity=condition.severity,
                    evidence=context.copy()
                )
                
                if passed:
                    self.stats["passed"] += 1
                else:
                    self.stats["failed"] += 1
                    state["failures"].append(result)
                
                results.append(result)
                self.verification_history.append(result)
            
            except Exception as e:
                result = VerificationResult(
                    condition_name=condition.name,
                    verification_type=condition.verification_type,
                    passed=False,
                    message=f"Verification error: {str(e)}",
                    severity=VerificationSeverity.CRITICAL,
                    evidence={"error": str(e)}
                )
                results.append(result)
                state["failures"].append(result)
                self.verification_history.append(result)
                self.stats["failed"] += 1
        
        state["results"] = results
        state["all_passed"] = len(state["failures"]) == 0
        
        return state
    
    def _check_invariants_node(self, state: VerificationState) -> VerificationState:
        """Check system invariants."""
        # Additional invariant checks already handled in verify_properties
        return state
    
    def _collect_evidence_node(self, state: VerificationState) -> VerificationState:
        """Collect verification evidence."""
        # Create verification summary
        results = state["results"]
        failures = state["failures"]
        
        state["verification_summary"] = {
            "total_conditions": len(results),
            "passed": sum(1 for r in results if r.passed),
            "failed": len(failures),
            "critical_failures": sum(1 for f in failures if f.severity == VerificationSeverity.CRITICAL),
            "verification_types": {}
        }
        
        # Count by type
        for vtype in VerificationType:
            count = sum(1 for r in results if r.verification_type == vtype)
            if count > 0:
                state["verification_summary"]["verification_types"][vtype.value] = count
        
        return state
    
    def _generate_proof_node(self, state: VerificationState) -> VerificationState:
        """Generate verification proof using LLM."""
        results = state["results"]
        failures = state["failures"]
        summary = state["verification_summary"]
        context = state["context"]
        
        if state["all_passed"]:
            # Generate proof for successful verification
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert in formal verification and mathematical proof."),
                ("user", """Generate a verification proof for this successful verification:

Context: {context}
Conditions Verified: {total_conditions}
Verification Types: {verification_types}

Provide proof as JSON with:
- formal_statement: Formal statement of what was verified
- proof_sketch: Brief proof sketch
- assumptions: List of assumptions (2-3 items)
- conclusions: List of conclusions (2-3 items)
- confidence: Confidence score 0.0-1.0

Focus on formal correctness.""")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "context": str(context),
                "total_conditions": summary["total_conditions"],
                "verification_types": str(summary["verification_types"])
            })
            
            try:
                proof_data = json.loads(response.content)
                state["proof"] = VerificationProof(
                    formal_statement=proof_data.get("formal_statement", "System properties verified"),
                    proof_sketch=proof_data.get("proof_sketch", "All conditions satisfied"),
                    assumptions=proof_data.get("assumptions", []),
                    conclusions=proof_data.get("conclusions", ["System is correct"]),
                    confidence=proof_data.get("confidence", 0.95)
                )
                state["recommendations"] = ["System verification passed", "Properties hold"]
            except:
                state["proof"] = VerificationProof(
                    formal_statement="System properties verified",
                    proof_sketch="All verification conditions passed",
                    assumptions=["Input conditions valid"],
                    conclusions=["System behaves correctly"],
                    confidence=0.90
                )
                state["recommendations"] = ["Verification successful"]
        else:
            # Generate analysis for failures
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert in formal verification and debugging."),
                ("user", """Analyze these verification failures:

Context: {context}
Total Failures: {failure_count}
Critical Failures: {critical_count}

Failed Conditions:
{failures}

Provide analysis as JSON with:
- formal_statement: What failed to verify
- proof_sketch: Why verification failed
- assumptions: Violated assumptions (2-3 items)
- conclusions: Failure implications (2-3 items)
- confidence: Confidence in analysis 0.0-1.0

Also suggest fixes.""")
            ])
            
            chain = prompt | self.llm
            failure_details = "\n".join([f"- {f.message}" for f in failures])
            
            response = chain.invoke({
                "context": str(context),
                "failure_count": summary["failed"],
                "critical_count": summary["critical_failures"],
                "failures": failure_details
            })
            
            try:
                proof_data = json.loads(response.content)
                state["proof"] = VerificationProof(
                    formal_statement=proof_data.get("formal_statement", "Verification failed"),
                    proof_sketch=proof_data.get("proof_sketch", "Conditions not satisfied"),
                    assumptions=proof_data.get("assumptions", []),
                    conclusions=proof_data.get("conclusions", ["System has errors"]),
                    confidence=proof_data.get("confidence", 0.85)
                )
                state["recommendations"] = [
                    "Fix critical verification failures",
                    "Review violated conditions",
                    "Update system implementation"
                ]
            except:
                state["proof"] = VerificationProof(
                    formal_statement="Verification failed",
                    proof_sketch="Some conditions not satisfied",
                    assumptions=["Conditions were violated"],
                    conclusions=["System needs correction"],
                    confidence=0.80
                )
                state["recommendations"] = ["Fix verification failures"]
        
        return state
    
    def add_condition(
        self,
        name: str,
        verification_type: VerificationType,
        condition: Callable,
        description: str,
        severity: VerificationSeverity = VerificationSeverity.CRITICAL
    ):
        """Add verification condition."""
        vc = VerificationCondition(
            name=name,
            verification_type=verification_type,
            condition=condition,
            description=description,
            severity=severity
        )
        
        self.conditions.append(vc)
    
    def verify(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """Verify system through workflow."""
        initial_state: VerificationState = {
            "context": context,
            "conditions": [],
            "results": [],
            "all_passed": True,
            "failures": [],
            "proof": None,
            "recommendations": [],
            "verification_summary": {}
        }
        
        final_state = self.workflow.invoke(initial_state)
        
        return {
            "passed": final_state["all_passed"],
            "results": final_state["results"],
            "failures": final_state["failures"],
            "summary": final_state["verification_summary"],
            "proof": final_state["proof"],
            "recommendations": final_state["recommendations"]
        }
    
    def verify_property(
        self,
        property_name: str,
        property_check: Callable,
        context: Dict[str, Any]
    ) -> bool:
        """Verify specific property."""
        self.add_condition(
            property_name,
            VerificationType.PROPERTY,
            property_check,
            f"Property: {property_name}",
            VerificationSeverity.CRITICAL
        )
        
        result = self.verify(context)
        return result["passed"]
    
    def check_invariant(
        self,
        invariant_name: str,
        invariant_check: Callable,
        context: Dict[str, Any]
    ) -> bool:
        """Check invariant."""
        self.add_condition(
            invariant_name,
            VerificationType.INVARIANT,
            invariant_check,
            f"Invariant: {invariant_name}",
            VerificationSeverity.CRITICAL
        )
        
        result = self.verify(context)
        return result["passed"]
    
    def get_failures(self) -> List[VerificationResult]:
        """Get all failed verifications."""
        return [r for r in self.verification_history if not r.passed]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get verification statistics."""
        return {
            "name": self.name,
            "conditions_count": len(self.conditions),
            **self.stats
        }


class VerificationPattern:
    """
    Comprehensive Verification Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.verifiers: Dict[str, IntelligentVerifier] = {}
        self.model_name = model_name
    
    def create_verifier(self, name: str) -> IntelligentVerifier:
        """Create a verifier."""
        verifier = IntelligentVerifier(name, model_name=self.model_name)
        self.verifiers[name] = verifier
        return verifier
    
    def verify(
        self,
        verifier_name: str,
        context: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Run verification."""
        if verifier_name not in self.verifiers:
            return {"error": f"Verifier {verifier_name} not found"}
        
        verifier = self.verifiers[verifier_name]
        result = verifier.verify(context)
        
        return {
            "verifier": verifier_name,
            **result
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_verifications = sum(v.stats["total_verifications"] for v in self.verifiers.values())
        total_passed = sum(v.stats["passed"] for v in self.verifiers.values())
        total_failed = sum(v.stats["failed"] for v in self.verifiers.values())
        
        return {
            "total_verifiers": len(self.verifiers),
            "total_verifications": total_verifications,
            "total_passed": total_passed,
            "total_failed": total_failed
        }


if __name__ == "__main__":
    print("=== Verification Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Banking System Verification")
    print("-" * 50)
    pattern = VerificationPattern(model_name="gpt-4")
    
    banking_verifier = pattern.create_verifier("BankingVerifier")
    
    # Add verification conditions
    print("\nAdding verification conditions...")
    
    # Property: Balance non-negative
    banking_verifier.add_condition(
        "balance_non_negative",
        VerificationType.PROPERTY,
        lambda ctx: ctx.get("balance", 0) >= 0,
        "Account balance must be non-negative",
        VerificationSeverity.CRITICAL
    )
    
    # Invariant: Total equals sum
    banking_verifier.add_condition(
        "total_equals_sum",
        VerificationType.INVARIANT,
        lambda ctx: abs(ctx.get("total", 0) - sum(ctx.get("transactions", [])))  0,
        "Order total must be positive",
        VerificationSeverity.CRITICAL
    )
    
    order_verifier.add_condition(
        "items_not_empty",
        VerificationType.INVARIANT,
        lambda ctx: len(ctx.get("items", [])) > 0,
        "Order must contain items",
        VerificationSeverity.CRITICAL
    )
    
    order_verifier.add_condition(
        "payment_authorized",
        VerificationType.PRECONDITION,
        lambda ctx: ctx.get("payment_status") == "authorized",
        "Payment must be authorized",
        VerificationSeverity.HIGH
    )
    
    order_verifier.add_condition(
        "shipping_address_valid",
        VerificationType.ASSERTION,
        lambda ctx: ctx.get("shipping_address") is not None and len(ctx.get("shipping_address", "")) > 0,
        "Shipping address must be provided",
        VerificationSeverity.HIGH
    )
    
    print("\nVerification 3: Valid order")
    result3 = pattern.verify("OrderVerifier", {
        "order_total": 150.99,
        "items": ["item1", "item2", "item3"],
        "payment_status": "authorized",
        "shipping_address": "123 Main St, City, State 12345"
    })
    
    print(f"Passed: {result3['passed']}")
    print(f"Summary: {result3['summary']}")
    
    print("\n" + "=" * 50)
    print("\nVerification 4: Order with issues")
    result4 = pattern.verify("OrderVerifier", {
        "order_total": 0.0,  # Invalid: zero total
        "items": [],  # Invalid: no items
        "payment_status": "pending",  # Invalid: not authorized
        "shipping_address": ""  # Invalid: empty address
    })
    
    print(f"Passed: {result4['passed']}")
    print(f"Summary: {result4['summary']}")
    print(f"\nFailures:")
    for failure in result4['failures']:
        print(f"  - {failure.message}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Data Structure Invariants")
    print("-" * 50)
    
    structure_verifier = pattern.create_verifier("StructureVerifier")
    
    # Data structure invariants
    structure_verifier.add_condition(
        "array_sorted",
        VerificationType.INVARIANT,
        lambda ctx: all(ctx["array"][i]  1 else True,
        "Array must be sorted",
        VerificationSeverity.MEDIUM
    )
    
    structure_verifier.add_condition(
        "no_duplicates",
        VerificationType.PROPERTY,
        lambda ctx: len(ctx.get("array", [])) == len(set(ctx.get("array", []))),
        "Array must not contain duplicates",
        VerificationSeverity.LOW
    )
    
    structure_verifier.add_condition(
        "size_within_bounds",
        VerificationType.ASSERTION,
        lambda ctx: 0 <= len(ctx.get("array", [])) <= ctx.get("max_size", 1000),
        "Array size must be within bounds",
        VerificationSeverity.HIGH
    )
    
    print("\nVerification 5: Valid sorted array")
    result5 = pattern.verify("StructureVerifier", {
        "array": [1, 2, 3, 5, 8, 13],
        "max_size": 100
    })
    
    print(f"Passed: {result5['passed']}")
    print(f"Summary: {result5['summary']}")
    
    print("\n" + "=" * 50)
    print("\nVerification 6: Unsorted array with duplicates")
    result6 = pattern.verify("StructureVerifier", {
        "array": [5, 2, 8, 2, 1],  # Unsorted and has duplicates
        "max_size": 100
    })
    
    print(f"Passed: {result6['passed']}")
    print(f"Summary: {result6['summary']}")
    if result6['failures']:
        print(f"\nFailures:")
        for failure in result6['failures']:
            print(f"  - [{failure.severity.value}] {failure.message}")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"\nOverall Statistics:")
    print(f"  Total Verifiers: {overall_stats['total_verifiers']}")
    print(f"  Total Verifications: {overall_stats['total_verifications']}")
    print(f"  Total Passed: {overall_stats['total_passed']}")
    print(f"  Total Failed: {overall_stats['total_failed']}")
    
    print(f"\nBanking Verifier Statistics:")
    banking_stats = banking_verifier.get_statistics()
    for key, value in banking_stats.items():
        print(f"  {key}: {value}")
    
    print(f"\nOrder Verifier Statistics:")
    order_stats = order_verifier.get_statistics()
    for key, value in order_stats.items():
        print(f"  {key}: {value}")
    
    print("\n" + "=" * 50)
    print("Verification pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Verification

Pattern 270: Verification Pattern – Line-by-Line Explanation

Overview

This pattern implements formal verification with property checking, invariant validation, and LLM-powered proof generation. It ensures system correctness through mathematical reasoning and formal methods.

Code Explanation

Lines 1-15: Imports

"""
Pattern 270: Verification Pattern with LangChain/LangGraph

Purpose: Formal verification with property checking and proof generation.
Provides intelligent verification orchestration with LLM-powered formal reasoning.
"""

Explanation: Module docstring – pattern for formal correctness verification.

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 verification with LLM integration.

Lines 17-35: Enum Definitions

class VerificationType(Enum):
    """Types of verification."""
    PROPERTY = "property"          # General properties
    INVARIANT = "invariant"        # Always true conditions
    PRECONDITION = "precondition"  # Must be true before execution
    POSTCONDITION = "postcondition"  # Must be true after execution
    ASSERTION = "assertion"        # Point-in-time checks
    CONTRACT = "contract"          # Interface contracts

Explanation: Six verification types based on formal methods: – PROPERTY: General system properties (e.g., “balance never negative”) – INVARIANT: Conditions that must always hold (e.g., “list is always sorted”) – PRECONDITION: Requirements before function execution (e.g., “input not null”) – POSTCONDITION: Guarantees after function execution (e.g., “result is positive”) – ASSERTION: Runtime checks at specific points (e.g., “index in bounds”) – CONTRACT: Interface guarantees (e.g., “implements all required methods”)

class VerificationResult(Enum):
    """Result of verification."""
    VERIFIED = "verified"      # Proof successful
    VIOLATED = "violated"      # Proof failed
    UNKNOWN = "unknown"        # Cannot determine

Explanation: Three possible verification outcomes.

Lines 37-79: Dataclasses

@dataclass
class VerificationProperty:
    """Property to verify."""
    name: str
    verification_type: VerificationType
    checker: Callable
    description: str
    expected_result: bool = True

Explanation: Defines a property to verify: – name: Identifier – verification_type: Category – checker: Function that checks property – description: Human-readable explanation – expected_result: What checker should return (usually True)

@dataclass
class VerificationEvidence:
    """Evidence for verification."""
    property_name: str
    result: VerificationResult
    evidence: List[str]
    counterexample: Optional[Dict[str, Any]] = None
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Evidence supporting verification: – result: Verified, violated, or unknown – evidence: List of proof steps or test results – counterexample: Example input that violates property (if violated)

Example counterexample: {"input": -5, "output": None, "expected": "positive number"}

@dataclass
class ProofStep:
    """Step in verification proof."""
    step_number: int
    statement: str
    justification: str
    proven: bool

Explanation: Individual step in formal proof. Tracks logical progression.

@dataclass
class VerificationInsights:
    """LLM-generated verification insights."""
    summary: str
    proof_quality: str
    gaps: List[str]
    suggestions: List[str]
    formal_proof: List[ProofStep]

Explanation: AI-generated formal reasoning: – summary: Overall verification status – proof_quality: Assessment of proof completeness – gaps: Missing proof steps or assumptions – suggestions: How to strengthen proof – formal_proof: Step-by-step logical proof

Lines 81-94: TypedDict State

class VerificationState(TypedDict):
    """State for verification workflow."""
    properties: List[VerificationProperty]
    evidence: Dict[str, VerificationEvidence]
    system_state: Dict[str, Any]
    verification_results: Dict[str, VerificationResult]
    insights: Optional[VerificationInsights]
    proof_complete: bool

Explanation: Workflow state tracking properties, evidence, and formal proofs.

Lines 96-142: IntelligentVerifier Initialization

class IntelligentVerifier:
    """
    Intelligent Verifier with LangGraph workflow integration.

    Features:
    - Multiple verification types (property, invariant, contract, etc.)
    - LLM-powered proof generation
    - Formal reasoning
    - Counterexample generation
    - Evidence collection
    - Symbolic execution
    - Theorem proving assistance
    """

Explanation: Main verifier class with formal methods and AI assistance.

def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.3):
    self.name = name
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)  # Lower temp for formal reasoning

    # Properties
    self.properties: List[VerificationProperty] = []

    # Evidence
    self.evidence_history: List[VerificationEvidence] = []

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

    # Statistics
    self.stats = {
        "total_verifications": 0,
        "verified": 0,
        "violated": 0,
        "unknown": 0
    }

Explanation: Initialize with lower temperature (0.3) for more deterministic formal reasoning.

Lines 144-162: Workflow Building

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

    workflow.add_node("check_properties", self._check_properties_node)
    workflow.add_node("collect_evidence", self._collect_evidence_node)
    workflow.add_node("generate_proof", self._generate_proof_node)
    workflow.add_node("validate_proof", self._validate_proof_node)
    workflow.add_node("synthesize_insights", self._synthesize_insights_node)

Explanation: Create 5-node workflow: 1. Check properties: Execute checkers 2. Collect evidence: Gather proof steps 3. Generate proof: LLM creates formal proof 4. Validate proof: Verify logical soundness 5. Synthesize insights: LLM analyzes completeness

    workflow.set_entry_point("check_properties")
    workflow.add_edge("check_properties", "collect_evidence")
    workflow.add_edge("collect_evidence", "generate_proof")
    workflow.add_edge("generate_proof", "validate_proof")
    workflow.add_edge("validate_proof", "synthesize_insights")
    workflow.add_edge("synthesize_insights", END)

    return workflow.compile()

Explanation: Linear flow through verification stages.

Lines 164-227: Check Properties Node

def _check_properties_node(self, state: VerificationState) -> VerificationState:
    """Check all properties."""
    system_state = state["system_state"]
    verification_results = {}
    evidence = {}

Explanation: Initialize results and evidence dictionaries.

    for prop in self.properties:
        self.stats["total_verifications"] += 1

        try:
            # Execute checker
            result = prop.checker(system_state)

            if result == prop.expected_result:
                verification_results[prop.name] = VerificationResult.VERIFIED
                self.stats["verified"] += 1
                evidence_list = [f"Property '{prop.name}' holds", f"Checker returned {result}"]
            else:
                verification_results[prop.name] = VerificationResult.VIOLATED
                self.stats["violated"] += 1
                evidence_list = [
                    f"Property '{prop.name}' violated",
                    f"Expected {prop.expected_result}, got {result}"
                ]

Explanation: Execute each checker: – If returns expected result → VERIFIED – If returns unexpected result → VIOLATED – Record evidence for both cases

            # Try to find counterexample if violated
            counterexample = None
            if verification_results[prop.name] == VerificationResult.VIOLATED:
                counterexample = {
                    "system_state": system_state,
                    "checker_result": result,
                    "expected": prop.expected_result
                }

Explanation: For violations, capture system state as counterexample. This helps diagnose why property failed.

            evidence[prop.name] = VerificationEvidence(
                property_name=prop.name,
                result=verification_results[prop.name],
                evidence=evidence_list,
                counterexample=counterexample
            )

        except Exception as e:
            verification_results[prop.name] = VerificationResult.UNKNOWN
            self.stats["unknown"] += 1
            evidence[prop.name] = VerificationEvidence(
                property_name=prop.name,
                result=VerificationResult.UNKNOWN,
                evidence=[f"Verification error: {str(e)}"]
            )

Explanation: Handle exceptions as UNKNOWN result. Cannot prove or disprove.

    state["verification_results"] = verification_results
    state["evidence"] = evidence

    return state

Lines 229-251: Collect Evidence Node

def _collect_evidence_node(self, state: VerificationState) -> VerificationState:
    """Collect additional evidence for verification."""
    evidence = state["evidence"]

    # Add evidence to history
    for prop_name, evidence_obj in evidence.items():
        self.evidence_history.append(evidence_obj)

    # Collect counterexamples if any
    counterexamples = []
    for prop_name, evidence_obj in evidence.items():
        if evidence_obj.counterexample:
            counterexamples.append({
                "property": prop_name,
                "counterexample": evidence_obj.counterexample
            })

    if counterexamples:
        state["insights"] = VerificationInsights(
            summary=f"Found {len(counterexamples)} violations",
            proof_quality="Incomplete",
            gaps=[f"Property {ce['property']} violated" for ce in counterexamples],
            suggestions=["Address counterexamples to complete verification"],
            formal_proof=[]
        )

    return state

Explanation: Collect evidence into history, identify counterexamples for further analysis.

Lines 253-338: Generate Proof Node (LLM-Powered)

def _generate_proof_node(self, state: VerificationState) -> VerificationState:
    """Generate formal proof using LLM."""
    properties = state["properties"]
    evidence = state["evidence"]
    verification_results = state["verification_results"]

    # Prepare properties summary
    properties_summary = []
    for prop in properties:
        result = verification_results.get(prop.name, VerificationResult.UNKNOWN)
        properties_summary.append(f"- {prop.name}: {prop.description} [{result.value}]")

Explanation: Prepare property list with current verification status.

    # Prepare evidence summary
    evidence_summary = []
    for prop_name, evidence_obj in evidence.items():
        evidence_summary.append(f"Property: {prop_name}")
        evidence_summary.append(f"  Result: {evidence_obj.result.value}")
        evidence_summary.extend([f"  - {e}" for e in evidence_obj.evidence])
        if evidence_obj.counterexample:
            evidence_summary.append(f"  Counterexample: {evidence_obj.counterexample}")

Explanation: Prepare evidence list with counterexamples if any.

    # Use LLM to generate formal proof
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert in formal verification and mathematical proof generation."),
        ("user", """Generate a formal verification proof:

System: {name}

Properties to Verify:
{properties}

Evidence Collected:
{evidence}

Provide analysis as JSON with:
- summary: Brief summary of verification status
- proof_quality: Assessment (Complete/Partial/Incomplete)
- gaps: List of gaps in current proof
- suggestions: How to complete verification
- formal_proof: Array of proof steps with:
  - step_number: Sequential step number
  - statement: Logical statement
  - justification: Why this step is valid
  - proven: true/false

Focus on logical rigor and formal correctness.""")
    ])

Explanation: Ask LLM to generate formal proof with step-by-step logical reasoning.

    chain = prompt | self.llm
    response = chain.invoke({
        "name": self.name,
        "properties": "\n".join(properties_summary),
        "evidence": "\n".join(evidence_summary)
    })

Explanation: Invoke LLM with properties and evidence.

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

        # Parse proof steps
        formal_proof = []
        for step_data in insights_data.get("formal_proof", []):
            formal_proof.append(ProofStep(
                step_number=step_data.get("step_number", 0),
                statement=step_data.get("statement", ""),
                justification=step_data.get("justification", ""),
                proven=step_data.get("proven", False)
            ))

        state["insights"] = VerificationInsights(
            summary=insights_data.get("summary", "Verification analysis completed"),
            proof_quality=insights_data.get("proof_quality", "Unknown"),
            gaps=insights_data.get("gaps", []),
            suggestions=insights_data.get("suggestions", []),
            formal_proof=formal_proof
        )

    except Exception as e:
        # Fallback insights
        state["insights"] = VerificationInsights(
            summary="Verification completed",
            proof_quality="Partial",
            gaps=["LLM proof generation encountered an error"],
            suggestions=["Manual verification may be needed"],
            formal_proof=[]
        )

Explanation: Parse JSON response into proof steps. Fallback if parsing fails.

    return state

Lines 340-368: Validate Proof Node

def _validate_proof_node(self, state: VerificationState) -> VerificationState:
    """Validate logical soundness of proof."""
    insights = state["insights"]

    if not insights or not insights.formal_proof:
        state["proof_complete"] = False
        return state

    # Check if all proof steps are proven
    all_proven = all(step.proven for step in insights.formal_proof)

    # Check if there are no critical gaps
    critical_gaps = [g for g in insights.gaps if "violated" in g.lower() or "failed" in g.lower()]

    proof_complete = all_proven and len(critical_gaps) == 0

    state["proof_complete"] = proof_complete

    return state

Explanation: Validate proof completeness: – All steps must be proven – No critical gaps (violations/failures) – Sets proof_complete flag

Lines 370-392: Synthesize Insights Node

def _synthesize_insights_node(self, state: VerificationState) -> VerificationState:
    """Synthesize final verification insights."""
    insights = state["insights"]

    if not insights:
        return state

    # Add summary based on results
    verified_count = sum(1 for r in state["verification_results"].values() if r == VerificationResult.VERIFIED)
    violated_count = sum(1 for r in state["verification_results"].values() if r == VerificationResult.VIOLATED)
    unknown_count = sum(1 for r in state["verification_results"].values() if r == VerificationResult.UNKNOWN)

    insights.summary += f" | Verified: {verified_count}, Violated: {violated_count}, Unknown: {unknown_count}"

    return state

Explanation: Add verification statistics to insights summary.

Lines 394-460: Public Methods

def add_property(self, prop: VerificationProperty):
    """Add verification property."""
    self.properties.append(prop)

Explanation: Register property to verify.

def add_invariant(self, name: str, checker: Callable, description: str):
    """Add invariant property."""
    prop = VerificationProperty(
        name=name,
        verification_type=VerificationType.INVARIANT,
        checker=checker,
        description=description
    )
    self.add_property(prop)

Explanation: Convenience method for invariants. Example: add_invariant("balance_non_negative", lambda s: s["balance"] >= 0, "Balance is always non-negative").

def verify(self, system_state: Dict[str, Any]) -> Dict[str, Any]:
    """Verify system with workflow."""
    initial_state: VerificationState = {
        "properties": self.properties,
        "evidence": {},
        "system_state": system_state,
        "verification_results": {},
        "insights": None,
        "proof_complete": False
    }

    final_state = self.workflow.invoke(initial_state)

    return {
        "results": final_state["verification_results"],
        "evidence": final_state["evidence"],
        "insights": final_state["insights"],
        "proof_complete": final_state["proof_complete"]
    }

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

Lines 485-630: Demo Code

Scenario 1: Banking System Verification

verifier = pattern.create_verifier("BankingSystem")

# Invariant: balance never negative
def balance_non_negative(state: Dict) -> bool:
    return state.get("balance", 0) >= 0

verifier.add_invariant("balance_non_negative", balance_non_negative, "Balance must be non-negative")

# Precondition: withdrawal amount valid
def withdrawal_valid(state: Dict) -> bool:
    withdrawal = state.get("withdrawal_amount", 0)
    balance = state.get("balance", 0)
    return withdrawal > 0 and withdrawal <= balance

verifier.add_precondition("withdrawal_valid", withdrawal_valid, "Withdrawal must be positive and not exceed balance")

# Valid state
valid_state = {"balance": 1000, "withdrawal_amount": 100}
result1 = pattern.verify("BankingSystem", valid_state)

Explanation: Define invariants and preconditions for banking system. Valid state passes verification.

Scenario 2: Contract Violation

# Invalid state (negative balance)
invalid_state = {"balance": -500, "withdrawal_amount": 100}
result2 = pattern.verify("BankingSystem", invalid_state)

Explanation: Negative balance violates invariant, triggers proof generation with counterexample.

Scenario 3: E-commerce Order Verification

order_verifier = pattern.create_verifier("OrderProcessing")

def total_matches_items(state: Dict) -> bool:
    """Postcondition: order total matches sum of items."""
    items = state.get("items", [])
    total = state.get("total", 0)
    calculated = sum(item.get("price", 0) * item.get("quantity", 0) for item in items)
    return abs(total - calculated) < 0.01

order_verifier.add_postcondition("total_matches", total_matches_items, "Order total must equal sum of line items")

order_state = {
    "total": 150.0,
    "items": [
        {"price": 50.0, "quantity": 2},  # 100
        {"price": 25.0, "quantity": 2}   # 50
    ]
}

result3 = pattern.verify("OrderProcessing", order_state)

Explanation: Postcondition verification ensures business logic correctness.

Verification Type Comparison

Key Design Patterns

1. Specification Pattern: Properties specify correctness criteria
2. Chain of Responsibility: Sequential proof step validation
3. Template Method: Common workflow with type-specific checkers
4. Strategy Pattern: Different verification strategies per type
5. Observer Pattern: Evidence collection observes all checks

Performance Considerations

• Property Checking: O(properties × checker_complexity) – depends on checker
• Evidence Collection: O(properties) – linear
• Proof Generation: O(LLM) – ~2-4 seconds for complex proofs
• Proof Validation: O(proof_steps) – linear in proof length
• Memory: O(properties + evidence) – stores all results

Best Practices: – Use fast checkers (avoid heavy computation) – Order properties by likelihood of violation (fail fast) – Cache verification results for immutable states – Use symbolic execution for exhaustive checking – Combine runtime checks with static analysis

Extension Points

1. Add symbolic execution for path exploration
2. Implement SMT solver integration (Z3, CVC4)
3. Add model checking for state space exploration
4. Support temporal logic (LTL, CTL)
5. Add automated test generation from counterexamples
6. Implement proof tactics library
7. Add visualization of proof trees

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