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

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

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


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


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


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


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


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


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


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

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

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

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


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


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

    

Agentic Architecture Pattern: Test Double

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

Overview

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

Code Explanation

Lines 1-17: Imports

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

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

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

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

Explanation: Standard imports plus json for parsing LLM responses.

Lines 19-30: Enum Definitions

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

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

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

Explanation: Tracks whether mock expectations were verified.

Lines 32-71: Dataclasses

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

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

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

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

Explanation: Records each method invocation for spies and verification.

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

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

Lines 73-85: TypedDict State

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

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

Lines 87-137: IntelligentTestDouble Initialization

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

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

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

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

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

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

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

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

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

Explanation: Initialize call tracking, workflow, and statistics.

Lines 139-157: Workflow Building

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

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

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

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

    return workflow.compile()

Explanation: Define linear flow through all nodes.

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

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

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

Explanation: Extract parameters and update call counter.

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

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

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

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

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

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

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

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

    return state

Lines 218-230: Configure Behavior Node

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

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

    return state

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

Lines 232-291: Execute Call Node

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

    result = None

Explanation: Extract parameters and initialize result.

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

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

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

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

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

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

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

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

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

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

    state["result"] = result

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

    return state

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

Lines 293-346: Verify Expectations Node

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

    expectations_met = True

Explanation: Initialize verification – only meaningful for mocks.

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

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

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

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

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

Explanation: Verify arguments match expectations if specified.

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

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

    return state

Explanation: Set verification status and update statistics.

Lines 348-361: Generate Report Node

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

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

    return state

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

Lines 363-403: Public Configuration Methods

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

    if implementation:
        self.fake_implementations[method_name] = implementation

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

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

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

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

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

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

Lines 428-475: TestDoublePattern Coordinator

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

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

Explanation: Coordinator managing multiple test doubles.

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

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

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

Lines 477-675: Demo Code

Scenario 1: Dummy Double

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

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

Scenario 2: Stub Double

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

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

Scenario 3: Fake Double

fake = pattern.create_fake("FakeEmailService")

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

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

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

Scenario 4: Spy Double

spy = pattern.create_spy("MetricsSpy")

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

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

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

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

Scenario 5: Mock Double

mock = pattern.create_mock("PaymentMock")

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

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

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

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

Test Double Comparison

When to Use Each

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

Key Design Patterns

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

Performance Considerations

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

Extension Points

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

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

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

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


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


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


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


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


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


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


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

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

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

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


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


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

    

Agentic Architecture Pattern: Stub

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

Overview

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

Code Explanation

Lines 1-17: Imports and Setup

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

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

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

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

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

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

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

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

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

Lines 19-32: Enum Definitions

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

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

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

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

Lines 34-62: Dataclasses

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

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

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

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

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

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

Lines 64-77: TypedDict State

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

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

Lines 79-129: IntelligentStubManager Initialization

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

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

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

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

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

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

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

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

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

Explanation: Build workflow and initialize statistics tracking.

Lines 131-149: Workflow Building

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

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

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

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

    return workflow.compile()

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

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

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

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

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

Explanation: Extract call details, retrieve any existing configuration, update call count.

    # If no stub configured, use LLM to analyze
    if not stub_config:
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in test stub design and testing patterns."),
            ("user", """Analyze this method call and recommend an appropriate stub type:

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

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

Provide response as JSON with keys: recommended_type, complexity, reasoning, alternatives""")
        ])

Explanation: If method not configured, create prompt asking LLM to analyze the call and recommend stub type.

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

Explanation: Invoke LLM with method details.

        try:
            analysis_data = json.loads(response.content)
            state["analysis"] = StubAnalysis(
                recommended_type=StubType(analysis_data.get("recommended_type", "simple")),
                complexity=StubComplexity(analysis_data.get("complexity", "simple")),
                reasoning=analysis_data.get("reasoning", "No reasoning provided"),
                alternative_approaches=analysis_data.get("alternatives", [])
            )
        except:
            # Fallback
            state["analysis"] = StubAnalysis(
                recommended_type=StubType.SIMPLE,
                complexity=StubComplexity.SIMPLE,
                reasoning="Default simple stub",
                alternative_approaches=[]
            )

Explanation: Parse JSON response into StubAnalysis object. If parsing fails, use simple stub as fallback.

    return state

Lines 208-226: Select Stub Type Node

def _select_stub_type_node(self, state: StubState) -> StubState:
    """Select appropriate stub type."""
    stub_config = state["stub_config"]

    if stub_config:
        # Use configured stub type
        state["recommended_type"] = stub_config.stub_type
        state["complexity"] = stub_config.complexity
    else:
        # Use analyzed recommendation
        analysis = state.get("analysis")
        if analysis:
            state["recommended_type"] = analysis.recommended_type
            state["complexity"] = analysis.complexity
        else:
            state["recommended_type"] = StubType.SIMPLE
            state["complexity"] = StubComplexity.SIMPLE

    return state

Explanation: Decide stub type: use configured type if available, otherwise use LLM recommendation, or default to SIMPLE.

Lines 228-310: Generate Response Node

def _generate_response_node(self, state: StubState) -> StubState:
    """Generate stub response."""
    stub_type = state["recommended_type"]
    stub_config = state["stub_config"]
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]

    result = None

Explanation: Extract parameters and initialize result.

    try:
        if stub_type == StubType.EXCEPTION:
            if stub_config and stub_config.exception:
                raise stub_config.exception
            else:
                raise Exception(f"Stub exception for {method_name}")

Explanation: EXCEPTION type: raise configured exception or generic exception.

        elif stub_type == StubType.NULL:
            result = None

Explanation: NULL type: always return None.

        elif stub_type == StubType.SIMPLE:
            if stub_config:
                result = stub_config.return_value
            else:
                result = {"status": "stubbed", "method": method_name}

Explanation: SIMPLE type: return configured value or generic stub response.

        elif stub_type == StubType.SEQUENCE:
            if stub_config and stub_config.return_sequence:
                idx = self.sequence_indices[method_name]
                if idx < len(stub_config.return_sequence):
                    result = stub_config.return_sequence[idx]
                    self.sequence_indices[method_name] += 1
                else:
                    result = stub_config.return_sequence[-1] if stub_config.return_sequence else None
            else:
                result = {"status": "sequence_stub", "index": self.sequence_indices[method_name]}
                self.sequence_indices[method_name] += 1

Explanation: SEQUENCE type: return next value in sequence, track index. After sequence exhausted, return last value.

        elif stub_type == StubType.CONDITIONAL:
            if stub_config and stub_config.condition:
                if stub_config.condition(*args, **kwargs):
                    result = stub_config.return_value
                else:
                    result = None
            else:
                # Default conditional behavior
                result = {"status": "conditional", "condition_met": True}

Explanation: CONDITIONAL type: evaluate condition function with arguments, return value if true, None if false.

        elif stub_type == StubType.DYNAMIC:
            # Use LLM to generate dynamic response
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are generating a realistic stub response for testing."),
                ("user", """Generate a realistic stub response for this method:

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

Generate appropriate test data as JSON.""")
            ])

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

            try:
                result = json.loads(response.content)
            except:
                result = {"status": "dynamic_stub", "content": response.content}

Explanation: DYNAMIC type: use LLM to generate realistic test data based on method name and arguments.

    except Exception as e:
        state["insights"] = state.get("insights", [])
        state["insights"].append(f"Exception during response generation: {str(e)}")
        result = {"error": str(e)}

    state["result"] = result
    return state

Explanation: Catch exceptions, record as insight, return error dict.

Lines 312-333: Validate Response Node

def _validate_response_node(self, state: StubState) -> StubState:
    """Validate generated response."""
    result = state["result"]
    stub_type = state["recommended_type"]

    # Basic validation
    validation_passed = True

    if stub_type == StubType.NULL:
        validation_passed = result is None
    elif stub_type == StubType.EXCEPTION:
        validation_passed = isinstance(result, dict) and "error" in result
    elif stub_type == StubType.SIMPLE:
        validation_passed = result is not None

Explanation: Validate that result matches expected type: – NULL should be None – EXCEPTION should be error dict – SIMPLE should not be None

    state["validation_passed"] = validation_passed

    if not validation_passed:
        state["insights"] = state.get("insights", [])
        state["insights"].append(f"Validation failed for {stub_type.value} stub")

    return state

Explanation: Store validation result, add insight if failed.

Lines 335-356: Record Call Node

def _record_call_node(self, state: StubState) -> StubState:
    """Record the stub call."""
    call = StubCall(
        method_name=state["method_name"],
        args=state["args"],
        kwargs=state["kwargs"],
        result=state["result"],
        stub_type=state["recommended_type"],
        call_number=state["call_number"]
    )

    self.call_history.append(call)
    self.stats["total_calls"] += 1
    self.stats["stub_types"][state["recommended_type"].value] += 1

    return state

Explanation: Create call record and add to history, update statistics.

Lines 358-384: Public Methods

def register_stub(
    self,
    method_name: str,
    stub_type: StubType = StubType.SIMPLE,
    return_value: Any = None,
    return_sequence: List[Any] = None,
    condition: Optional[Callable] = None,
    exception: Optional[Exception] = None,
    complexity: StubComplexity = StubComplexity.SIMPLE
):
    """Register a stub configuration."""
    stub = StubResponse(
        method_name=method_name,
        stub_type=stub_type,
        return_value=return_value,
        return_sequence=return_sequence or [],
        condition=condition,
        exception=exception,
        complexity=complexity
    )

    self.stubs[method_name] = stub
    self.stats["methods_stubbed"] = len(self.stubs)

Explanation: Register stub configuration for a method. Called by users to configure stub behavior.

def call(self, method_name: str, *args, **kwargs) -> Any:
    """Execute stub method call through workflow."""
    initial_state: StubState = {
        "method_name": method_name,
        "args": args,
        "kwargs": kwargs,
        "stub_config": None,
        "recommended_type": None,
        "complexity": None,
        "analysis": None,
        "result": None,
        "validation_passed": False,
        "insights": [],
        "call_number": 0
    }

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

Explanation: Main entry point – create initial state, run workflow, return result. This is called when test code invokes the stubbed method.

def get_call_history(self, method_name: str = None) -> List[StubCall]:
    """Get call history."""
    if method_name:
        return [call for call in self.call_history if call.method_name == method_name]
    return self.call_history

Explanation: Retrieve call history, optionally filtered by method name.

Lines 408-445: StubPattern Coordinator

class StubPattern:
    """
    Comprehensive Stub Pattern coordinator with LangChain/LangGraph.
    """

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

Explanation: Coordinator managing multiple stub managers (e.g., one for database, one for API).

    def create_manager(self, name: str, target_class: str = "") -> IntelligentStubManager:
        """Create a stub manager."""
        manager = IntelligentStubManager(name, model_name=self.model_name)
        self.managers[name] = manager
        return manager

Explanation: Factory method to create new stub manager.

    def call(self, manager_name: str, method_name: str, *args, **kwargs) -> Dict[str, Any]:
        """Execute stub call."""
        if manager_name not in self.managers:
            return {"error": f"Manager {manager_name} not found"}

        manager = self.managers[manager_name]
        result = manager.call(method_name, *args, **kwargs)

        return {
            "manager": manager_name,
            "method": method_name,
            "result": result,
            "call_count": manager.call_counts[method_name]
        }

Explanation: Route call to appropriate manager, return result with metadata.

Lines 447-572: Demo Code

The demo demonstrates 5 scenarios:

Scenario 1: Database Stub with Sequence

db_manager.register_stub(
    "get_users",
    stub_type=StubType.SEQUENCE,
    return_sequence=[
        [{"id": 1, "name": "Alice"}],
        [{"id": 2, "name": "Bob"}],
        [{"id": 3, "name": "Charlie"}]
    ],
    complexity=StubComplexity.SIMPLE
)

Explanation: Register sequence stub that returns different users on each call. Useful for testing pagination or iteration.

Scenario 2: API Stub with Conditional Logic

api_manager.register_stub(
    "fetch_data",
    stub_type=StubType.CONDITIONAL,
    return_value={"status": "success", "data": "test_data"},
    condition=lambda *args, **kwargs: kwargs.get("authenticated", False),
    complexity=StubComplexity.MODERATE
)

Explanation: Register conditional stub that returns success only if authenticated=True in kwargs. Tests authentication logic.

Scenario 3: Dynamic Stub with LLM Generation

result3 = pattern.call("ServiceStub", "process_payment", 
                      amount=100.0, currency="USD", user_id=123)

Explanation: Call unconfigured method – LLM generates realistic payment response based on method name and arguments.

Scenario 4: Exception Stub

error_manager.register_stub(
    "dangerous_operation",
    stub_type=StubType.EXCEPTION,
    exception=Exception("Operation not allowed in test mode"),
    complexity=StubComplexity.TRIVIAL
)

Explanation: Register exception stub to test error handling code.

Scenario 5: Simple Fixed Response

config_manager.register_stub(
    "get_config",
    stub_type=StubType.SIMPLE,
    return_value={"timeout": 30, "retries": 3, "debug": True},
    complexity=StubComplexity.TRIVIAL
)

Explanation: Register simple stub returning fixed configuration. Most common stub type.

Key Design Patterns

1. Strategy Pattern: Different stub types implement different response strategies
2. Factory Pattern: StubPattern creates and manages stub managers
3. Chain of Responsibility: Workflow nodes process requests sequentially
4. Observer Pattern: Call history records all invocations
5. State Pattern: Workflow progresses through defined states

Testing Best Practices

1. Use SIMPLE stubs for deterministic, repeatable tests
2. Use SEQUENCE stubs for testing iteration and pagination
3. Use CONDITIONAL stubs for testing branching logic
4. Use DYNAMIC stubs when you need varied, realistic test data
5. Use EXCEPTION stubs to test error handling paths
6. Track call history to verify methods were called correctly

Performance Considerations

• SIMPLE/NULL/EXCEPTION: O(1) – instant response
• SEQUENCE: O(1) – array lookup
• CONDITIONAL: O(1) – single function call
• DYNAMIC: O(LLM) – requires LLM call (~1-3 seconds)
• Memory: O(calls) – stores full call history

Extension Points

1. Add new stub types by adding to StubType enum and handling in _generate_response_node
2. Enhance validation logic in _validate_response_node
3. Add stub recording/replay capabilities
4. Implement stub verification assertions
5. Add timing constraints for performance testing

    """
Pattern 265: Mock Agent

Purpose: Mock object creation with behavior simulation and interaction verification.
Creates mock agents for testing with configurable responses and call tracking.
"""

from typing import Dict, Any, List, Optional, Callable
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


class MockBehavior(Enum):
    """Mock behavior types."""
    RETURN_VALUE = "return_value"  # Return fixed value
    RAISE_ERROR = "raise_error"  # Raise exception
    CALL_REAL = "call_real"  # Call real implementation
    RECORD_ONLY = "record_only"  # Just record call
    CONDITIONAL = "conditional"  # Based on input


@dataclass
class MockCall:
    """Record of a mock call."""
    method_name: str
    args: tuple = field(default_factory=tuple)
    kwargs: Dict[str, Any] = field(default_factory=dict)
    return_value: Any = None
    timestamp: datetime = field(default_factory=datetime.now)
    call_count: int = 1


@dataclass
class MockConfiguration:
    """Configuration for mock method."""
    method_name: str
    behavior: MockBehavior
    return_value: Any = None
    side_effect: Optional[Callable] = None
    raise_exception: Optional[Exception] = None


class MockAgent:
    """
    Mock Agent - Mock object creation and verification.
    
    Features:
    - Configurable mock behaviors
    - Call recording and verification
    - Return value configuration
    - Exception simulation
    - Call count tracking
    - Argument verification
    - Spy functionality
    - Reset capabilities
    """
    
    def __init__(self, name: str, target_class: str = ""):
        self.name = name
        self.target_class = target_class
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Mock configurations
        self.configurations: Dict[str, MockConfiguration] = {}
        
        # Call tracking
        self.calls: List[MockCall] = []
        self.call_counts: Dict[str, int] = defaultdict(int)
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_calls": 0,
            "methods_mocked": 0,
            "behaviors": {b.value: 0 for b in MockBehavior}
        }
    
    def configure_method(
        self,
        method_name: str,
        behavior: MockBehavior = MockBehavior.RETURN_VALUE,
        return_value: Any = None,
        side_effect: Optional[Callable] = None,
        raise_exception: Optional[Exception] = None
    ):
        """Configure behavior for a mocked method."""
        
        config = MockConfiguration(
            method_name=method_name,
            behavior=behavior,
            return_value=return_value,
            side_effect=side_effect,
            raise_exception=raise_exception
        )
        
        self.configurations[method_name] = config
        self.statistics["methods_mocked"] = len(self.configurations)
    
    def call(
        self,
        method_name: str,
        *args,
        **kwargs
    ) -> Any:
        """Execute a mocked method call."""
        
        # Record the call
        call_record = MockCall(
            method_name=method_name,
            args=args,
            kwargs=kwargs,
            call_count=self.call_counts[method_name] + 1
        )
        
        self.calls.append(call_record)
        self.call_counts[method_name] += 1
        self.statistics["total_calls"] += 1
        
        # Check if method is configured
        if method_name not in self.configurations:
            # Default behavior: return None
            call_record.return_value = None
            return None
        
        config = self.configurations[method_name]
        self.statistics["behaviors"][config.behavior.value] += 1
        
        # Execute based on behavior
        if config.behavior == MockBehavior.RAISE_ERROR:
            if config.raise_exception:
                raise config.raise_exception
            raise Exception(f"Mock error for {method_name}")
        
        elif config.behavior == MockBehavior.RETURN_VALUE:
            call_record.return_value = config.return_value
            return config.return_value
        
        elif config.behavior == MockBehavior.CALL_REAL:
            # Would call real implementation if available
            call_record.return_value = f"Real implementation of {method_name}"
            return call_record.return_value
        
        elif config.behavior == MockBehavior.CONDITIONAL:
            # Execute side effect if provided
            if config.side_effect:
                result = config.side_effect(*args, **kwargs)
                call_record.return_value = result
                return result
            call_record.return_value = config.return_value
            return config.return_value
        
        # Default: record only
        call_record.return_value = None
        return None
    
    def was_called(self, method_name: str) -> bool:
        """Check if method was called."""
        return self.call_counts[method_name] > 0
    
    def call_count(self, method_name: str) -> int:
        """Get call count for method."""
        return self.call_counts[method_name]
    
    def was_called_with(
        self,
        method_name: str,
        *args,
        **kwargs
    ) -> bool:
        """Verify method was called with specific arguments."""
        
        for call in self.calls:
            if call.method_name == method_name:
                if call.args == args and call.kwargs == kwargs:
                    return True
        
        return False
    
    def get_calls(self, method_name: str = None) -> List[MockCall]:
        """Get all calls or calls for specific method."""
        
        if method_name:
            return [c for c in self.calls if c.method_name == method_name]
        
        return self.calls
    
    def reset(self):
        """Reset all call records."""
        self.calls.clear()
        self.call_counts.clear()
        self.statistics["total_calls"] = 0
        self.statistics["behaviors"] = {b.value: 0 for b in MockBehavior}
    
    def verify_expectations(self) -> Dict[str, Any]:
        """Verify mock expectations."""
        
        verification = {
            "all_configured_called": True,
            "uncalled_methods": [],
            "total_calls": self.statistics["total_calls"]
        }
        
        for method_name in self.configurations:
            if not self.was_called(method_name):
                verification["all_configured_called"] = False
                verification["uncalled_methods"].append(method_name)
        
        return verification
    
    def generate_mock_response(
        self,
        method_name: str,
        expected_type: str = "dict"
    ) -> Any:
        """Generate realistic mock response using LLM."""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Generate a realistic mock response for testing purposes."),
            ("human", f"Method: {method_name}\n"
                     f"Expected Type: {expected_type}\n"
                     f"Generate mock response:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse response (simplified)
        return {"generated": str(response.content)[:200]}
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get mock statistics."""
        return {
            "name": self.name,
            "target_class": self.target_class,
            **self.statistics
        }


class MockAgentPattern:
    """
    Mock Agent Pattern - Manage mock agents.
    
    Coordinates multiple mock agents for testing.
    """
    
    def __init__(self):
        self.agents: Dict[str, MockAgent] = {}
    
    def create_agent(
        self,
        name: str,
        target_class: str = ""
    ) -> MockAgent:
        """Create a new mock agent."""
        agent = MockAgent(name, target_class)
        self.agents[name] = agent
        print(f"   Created mock agent: {name} (mocking {target_class or 'generic'})")
        return agent
    
    def run(
        self,
        agent_name: str,
        method_name: str,
        *args,
        **kwargs
    ) -> Dict[str, Any]:
        """Execute mock method call."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        
        try:
            result = agent.call(method_name, *args, **kwargs)
            
            print(f"\n=== Mock Call by {agent_name} ===")
            print(f"Method: {method_name}")
            print(f"Args: {args}")
            print(f"Kwargs: {kwargs}")
            print(f"Result: {result}")
            print(f"Call Count: {agent.call_count(method_name)}")
            
            return {
                "agent": agent_name,
                "method": method_name,
                "result": result,
                "call_count": agent.call_count(method_name),
                "statistics": agent.get_statistics()
            }
        
        except Exception as e:
            print(f"\n=== Mock Error in {agent_name} ===")
            print(f"Method: {method_name}")
            print(f"Error: {str(e)}")
            
            return {
                "agent": agent_name,
                "method": method_name,
                "error": str(e)
            }


# Example usage
if __name__ == "__main__":
    print("=== Mock Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = MockAgentPattern()
    
    # Create mock agents
    print("1. Setting up mock agents...")
    
    user_service_mock = pattern.create_agent("UserServiceMock", "UserService")
    payment_mock = pattern.create_agent("PaymentMock", "PaymentGateway")
    
    # Configure mock behaviors
    print("\n2. Configuring mock behaviors...")
    
    user_service_mock.configure_method(
        "get_user",
        behavior=MockBehavior.RETURN_VALUE,
        return_value={"id": 1, "name": "John Doe", "email": "john@example.com"}
    )
    
    user_service_mock.configure_method(
        "create_user",
        behavior=MockBehavior.RETURN_VALUE,
        return_value={"id": 2, "status": "created"}
    )
    
    payment_mock.configure_method(
        "process_payment",
        behavior=MockBehavior.RETURN_VALUE,
        return_value={"transaction_id": "TXN123", "status": "approved"}
    )
    
    payment_mock.configure_method(
        "refund",
        behavior=MockBehavior.RAISE_ERROR,
        raise_exception=Exception("Refund not available")
    )
    
    # Execute mock calls
    print("\n3. Executing mock calls...")
    
    result1 = pattern.run("UserServiceMock", "get_user", user_id=1)
    
    print("\n" + "="*60 + "\n")
    
    result2 = pattern.run("UserServiceMock", "get_user", user_id=1)  # Second call
    
    print("\n" + "="*60 + "\n")
    
    result3 = pattern.run("PaymentMock", "process_payment", amount=100.0)
    
    print("\n" + "="*60 + "\n")
    
    # Try error case
    result4 = pattern.run("PaymentMock", "refund", transaction_id="TXN123")
    
    # Verify expectations
    print("\n" + "="*60)
    print("\n4. Verifying mock expectations...")
    
    verification = user_service_mock.verify_expectations()
    print(f"All configured methods called: {verification['all_configured_called']}")
    print(f"Uncalled methods: {verification['uncalled_methods']}")
    
    # Print final statistics
    print("\n5. Final Statistics:")
    stats = result3["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Comprehensive mock agent with behavior simulation complete!")

    

Agentic Architecture Pattern: Mock Agent

Pattern 265: 265 Mock Agent – Line-by-Line Explanation

Overview

Pattern 265: Mock Agent

Purpose: Mock object creation with behavior simulation and interaction verification. Creates mock agents for testing with configurable responses and call tracking.

Code Structure Analysis

File Information

• Pattern Number: 265
• Pattern Name: 265 Mock Agent
• File Path: patterns\testing_validation\265_mock_agent.py
• Total Lines: 379

Detailed Code Explanation

Module Docstring

"""
Pattern 265: Mock Agent

Purpose: Mock object creation with behavior simulation and interaction verification.
Creates mock agents for testing with configurable responses and call tracking.
"""

Explanation: This module docstring describes the purpose and key features of the 265 Mock Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
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  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

MockBehavior (Line 17)

class MockBehavior(Enum):
    RETURN_VALUE
    RAISE_ERROR
    CALL_REAL
    RECORD_ONLY
    CONDITIONAL

Purpose: Mock behavior types.

Values: – RETURN_VALUE: Represents return value – RAISE_ERROR: Represents raise error – CALL_REAL: Represents call real – RECORD_ONLY: Represents record only – CONDITIONAL: Represents conditional

Classes

MockBehavior (Line 17)

Description: Mock behavior types.

MockCall (Line 27)

Description: Record of a mock call.

MockConfiguration (Line 38)

Description: Configuration for mock method.

MockAgent (Line 47)

Description: Mock Agent – Mock object creation and verification.

Features: – Configurable mock behaviors – Call recording and verification – Return value configuration – Exception simulation – Call count tracking – Argument verification – Spy functionality – Reset capabilities

Methods:

__init__ (Line 62)

def __init__(self, name, target_class):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – target_class: Parameter for target_class

configure_method (Line 81)

def configure_method(self, method_name, behavior, return_value, side_effect, raise_exception):
    """Configure behavior for a mocked method."""

Purpose: Configure behavior for a mocked method.

Parameters: – self: Instance reference – method_name: Parameter for method_name – behavior: Parameter for behavior – return_value: Parameter for return_value – side_effect: Parameter for side_effect – raise_exception: Parameter for raise_exception

call (Line 102)

def call(self, method_name):
    """Execute a mocked method call."""

Purpose: Execute a mocked method call.

Parameters: – self: Instance reference – method_name: Parameter for method_name

was_called (Line 159)

def was_called(self, method_name):
    """Check if method was called."""

Purpose: Check if method was called.

Parameters: – self: Instance reference – method_name: Parameter for method_name

call_count (Line 163)

def call_count(self, method_name):
    """Get call count for method."""

Purpose: Get call count for method.

Parameters: – self: Instance reference – method_name: Parameter for method_name

was_called_with (Line 167)

def was_called_with(self, method_name):
    """Verify method was called with specific arguments."""

Purpose: Verify method was called with specific arguments.

Parameters: – self: Instance reference – method_name: Parameter for method_name

get_calls (Line 182)

def get_calls(self, method_name):
    """Get all calls or calls for specific method."""

Purpose: Get all calls or calls for specific method.

Parameters: – self: Instance reference – method_name: Parameter for method_name

reset (Line 190)

def reset(self):
    """Reset all call records."""

Purpose: Reset all call records.

Parameters: – self: Instance reference

verify_expectations (Line 197)

def verify_expectations(self):
    """Verify mock expectations."""

Purpose: Verify mock expectations.

Parameters: – self: Instance reference

generate_mock_response (Line 213)

def generate_mock_response(self, method_name, expected_type):
    """Generate realistic mock response using LLM."""

Purpose: Generate realistic mock response using LLM.

Parameters: – self: Instance reference – method_name: Parameter for method_name – expected_type: Parameter for expected_type

get_statistics (Line 233)

def get_statistics(self):
    """Get mock statistics."""

Purpose: Get mock statistics.

Parameters: – self: Instance reference

MockAgentPattern (Line 242)

Description: Mock Agent Pattern – Manage mock agents.

Coordinates multiple mock agents for testing.

Methods:

__init__ (Line 249)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 252)

def create_agent(self, name, target_class):
    """Create a new mock agent."""

Purpose: Create a new mock agent.

Parameters: – self: Instance reference – name: Identifier or label – target_class: Parameter for target_class

run (Line 263)

def run(self, agent_name, method_name):
    """Execute mock method call."""

Purpose: Execute mock method call.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – method_name: Parameter for method_name

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = MockBehavior(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 264: Emulation

Purpose: System emulation with behavior replication and performance testing.
Emulates external systems, APIs, and services for testing purposes.
"""

from typing import Dict, Any, List, Optional, Callable
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
import time
import random


class EmulationType(Enum):
    """Types of emulation."""
    API = "api"  # API emulation
    DATABASE = "database"  # Database emulation
    SERVICE = "service"  # Service emulation
    HARDWARE = "hardware"  # Hardware emulation
    NETWORK = "network"  # Network conditions


@dataclass
class EmulatedSystem:
    """Emulated system configuration."""
    name: str
    emulation_type: EmulationType
    behavior: Dict[str, Any] = field(default_factory=dict)
    latency_ms: float = 10.0
    failure_rate: float = 0.0  # 0.0 to 1.0
    state: Dict[str, Any] = field(default_factory=dict)


@dataclass
class EmulationRequest:
    """Request to emulated system."""
    system_name: str
    operation: str
    parameters: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class EmulationResponse:
    """Response from emulated system."""
    request: EmulationRequest
    success: bool
    data: Any = None
    error: str = ""
    latency_ms: float = 0.0
    timestamp: datetime = field(default_factory=datetime.now)


class EmulationAgent:
    """
    Emulation Agent - System and service emulation.
    
    Features:
    - API emulation
    - Database emulation
    - Service behavior replication
    - Latency simulation
    - Failure injection
    - State management
    - Performance testing
    - Response generation
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.5)
        
        # Emulated systems
        self.systems: Dict[str, EmulatedSystem] = {}
        self.request_history: List[EmulationRequest] = []
        self.response_history: List[EmulationResponse] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "emulation_types": {t.value: 0 for t in EmulationType},
            "avg_latency_ms": 0.0
        }
    
    def register_system(
        self,
        name: str,
        emulation_type: EmulationType,
        behavior: Dict[str, Any] = None,
        latency_ms: float = 10.0,
        failure_rate: float = 0.0
    ) -> EmulatedSystem:
        """Register an emulated system."""
        
        system = EmulatedSystem(
            name=name,
            emulation_type=emulation_type,
            behavior=behavior or {},
            latency_ms=latency_ms,
            failure_rate=failure_rate
        )
        
        self.systems[name] = system
        
        return system
    
    def emulate_request(
        self,
        request: EmulationRequest
    ) -> EmulationResponse:
        """Emulate a request to a system."""
        
        start_time = time.time()
        
        # Check if system exists
        if request.system_name not in self.systems:
            return EmulationResponse(
                request=request,
                success=False,
                error=f"System {request.system_name} not found",
                latency_ms=0.0
            )
        
        system = self.systems[request.system_name]
        
        # Simulate latency
        time.sleep(system.latency_ms / 1000.0)
        
        # Simulate failure
        if random.random()  Dict[str, Any]:
        """Emulate API response."""
        
        # Check behavior configuration
        if request.operation in system.behavior:
            return system.behavior[request.operation]
        
        # Default API response
        return {
            "status": "success",
            "operation": request.operation,
            "parameters": request.parameters,
            "timestamp": datetime.now().isoformat()
        }
    
    def _emulate_database(
        self,
        system: EmulatedSystem,
        request: EmulationRequest
    ) -> Dict[str, Any]:
        """Emulate database operation."""
        
        operation = request.operation.lower()
        
        if operation == "select":
            # Return mock data
            return {
                "rows": [
                    {"id": 1, "data": "value1"},
                    {"id": 2, "data": "value2"}
                ],
                "count": 2
            }
        elif operation == "insert":
            # Simulate insert
            return {
                "inserted_id": random.randint(1, 1000),
                "affected_rows": 1
            }
        elif operation == "update":
            # Simulate update
            return {
                "affected_rows": random.randint(0, 10)
            }
        elif operation == "delete":
            # Simulate delete
            return {
                "affected_rows": random.randint(0, 5)
            }
        
        return {"status": "executed", "operation": operation}
    
    def _emulate_service(
        self,
        system: EmulatedSystem,
        request: EmulationRequest
    ) -> Dict[str, Any]:
        """Emulate service call."""
        
        # Use LLM to generate realistic response
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are emulating a {request.operation} service. "
                      f"Generate a realistic JSON response."),
            ("human", f"Parameters: {request.parameters}\n\n"
                     f"Generate response (JSON format):")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse response (simplified)
        return {
            "service_response": str(response.content)[:200],
            "status": "completed"
        }
    
    def set_system_state(
        self,
        system_name: str,
        state: Dict[str, Any]
    ):
        """Set state for emulated system."""
        if system_name in self.systems:
            self.systems[system_name].state.update(state)
    
    def configure_network_conditions(
        self,
        system_name: str,
        latency_ms: float,
        failure_rate: float
    ):
        """Configure network conditions for emulation."""
        if system_name in self.systems:
            self.systems[system_name].latency_ms = latency_ms
            self.systems[system_name].failure_rate = failure_rate
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get emulation statistics."""
        
        success_rate = 0.0
        if self.statistics["total_requests"] > 0:
            success_rate = (
                self.statistics["successful_requests"] / 
                self.statistics["total_requests"]
            ) * 100
        
        return {
            "name": self.name,
            **self.statistics,
            "success_rate": f"{success_rate:.1f}%"
        }


class EmulationPattern:
    """
    Emulation Pattern - Manage emulation agents.
    
    Coordinates emulation of multiple systems.
    """
    
    def __init__(self):
        self.agents: Dict[str, EmulationAgent] = {}
    
    def create_agent(self, name: str) -> EmulationAgent:
        """Create a new emulation agent."""
        agent = EmulationAgent(name)
        self.agents[name] = agent
        print(f"   Created emulation agent: {name}")
        return agent
    
    def run(
        self,
        agent_name: str,
        request: EmulationRequest
    ) -> Dict[str, Any]:
        """Execute emulation request."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        response = agent.emulate_request(request)
        
        print(f"\n=== Emulation by {agent_name} ===")
        print(f"System: {request.system_name}")
        print(f"Operation: {request.operation}")
        print(f"Success: {response.success}")
        print(f"Latency: {response.latency_ms:.2f}ms")
        
        if response.success:
            print(f"Data: {str(response.data)[:100]}")
        else:
            print(f"Error: {response.error}")
        
        return {
            "agent": agent_name,
            "success": response.success,
            "latency_ms": response.latency_ms,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Emulation Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = EmulationPattern()
    
    # Create agent
    print("1. Setting up emulation agent...")
    emulator = pattern.create_agent("SystemEmulator")
    
    # Register emulated systems
    print("\n2. Registering emulated systems...")
    
    emulator.register_system(
        "PaymentAPI",
        EmulationType.API,
        behavior={
            "process_payment": {"status": "approved", "transaction_id": "TXN123"}
        },
        latency_ms=50.0,
        failure_rate=0.1
    )
    
    emulator.register_system(
        "UserDB",
        EmulationType.DATABASE,
        latency_ms=20.0,
        failure_rate=0.05
    )
    
    # Execute emulations
    print("\n3. Executing emulation requests...")
    
    request1 = EmulationRequest(
        system_name="PaymentAPI",
        operation="process_payment",
        parameters={"amount": 100.0, "currency": "USD"}
    )
    
    result1 = pattern.run("SystemEmulator", request1)
    
    print("\n" + "="*60 + "\n")
    
    request2 = EmulationRequest(
        system_name="UserDB",
        operation="select",
        parameters={"query": "SELECT * FROM users WHERE id=1"}
    )
    
    result2 = pattern.run("SystemEmulator", request2)
    
    # Print final statistics
    print("\n" + "="*60)
    print("\n4. Final Statistics:")
    stats = result2["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Comprehensive system emulation with behavior replication complete!")

    

Agentic Architecture Pattern: Emulation

Pattern 264: 264 Emulation – Line-by-Line Explanation

Overview

Pattern 264: Emulation

Purpose: System emulation with behavior replication and performance testing. Emulates external systems, APIs, and services for testing purposes.

Code Structure Analysis

File Information

• Pattern Number: 264
• Pattern Name: 264 Emulation
• File Path: patterns\testing_validation\264_emulation.py
• Total Lines: 413

Detailed Code Explanation

Module Docstring

"""
Pattern 264: Emulation

Purpose: System emulation with behavior replication and performance testing.
Emulates external systems, APIs, and services for testing purposes.
"""

Explanation: This module docstring describes the purpose and key features of the 264 Emulation pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
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
import time
import random  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

EmulationType (Line 19)

class EmulationType(Enum):
    API
    DATABASE
    SERVICE
    HARDWARE
    NETWORK

Purpose: Types of emulation.

Values: – API: Represents api – DATABASE: Represents database – SERVICE: Represents service – HARDWARE: Represents hardware – NETWORK: Represents network

Classes

EmulationType (Line 19)

Description: Types of emulation.

EmulatedSystem (Line 29)

Description: Emulated system configuration.

EmulationRequest (Line 40)

Description: Request to emulated system.

EmulationResponse (Line 49)

Description: Response from emulated system.

EmulationAgent (Line 59)

Description: Emulation Agent – System and service emulation.

Features: – API emulation – Database emulation – Service behavior replication – Latency simulation – Failure injection – State management – Performance testing – Response generation

Methods:

__init__ (Line 74)

def __init__(self, name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

register_system (Line 92)

def register_system(self, name, emulation_type, behavior, latency_ms, failure_rate):
    """Register an emulated system."""

Purpose: Register an emulated system.

Parameters: – self: Instance reference – name: Identifier or label – emulation_type: Parameter for emulation_type – behavior: Parameter for behavior – latency_ms: Parameter for latency_ms – failure_rate: Parameter for failure_rate

emulate_request (Line 114)

def emulate_request(self, request):
    """Emulate a request to a system."""

Purpose: Emulate a request to a system.

Parameters: – self: Instance reference – request: Parameter for request

_emulate_api (Line 188)

def _emulate_api(self, system, request):
    """Emulate API response."""

Purpose: Emulate API response.

Parameters: – self: Instance reference – system: Parameter for system – request: Parameter for request

_emulate_database (Line 207)

def _emulate_database(self, system, request):
    """Emulate database operation."""

Purpose: Emulate database operation.

Parameters: – self: Instance reference – system: Parameter for system – request: Parameter for request

_emulate_service (Line 244)

def _emulate_service(self, system, request):
    """Emulate service call."""

Purpose: Emulate service call.

Parameters: – self: Instance reference – system: Parameter for system – request: Parameter for request

set_system_state (Line 268)

def set_system_state(self, system_name, state):
    """Set state for emulated system."""

Purpose: Set state for emulated system.

Parameters: – self: Instance reference – system_name: Parameter for system_name – state: Current workflow state

configure_network_conditions (Line 277)

def configure_network_conditions(self, system_name, latency_ms, failure_rate):
    """Configure network conditions for emulation."""

Purpose: Configure network conditions for emulation.

Parameters: – self: Instance reference – system_name: Parameter for system_name – latency_ms: Parameter for latency_ms – failure_rate: Parameter for failure_rate

get_statistics (Line 288)

def get_statistics(self):
    """Get emulation statistics."""

Purpose: Get emulation statistics.

Parameters: – self: Instance reference

EmulationPattern (Line 305)

Description: Emulation Pattern – Manage emulation agents.

Coordinates emulation of multiple systems.

Methods:

__init__ (Line 312)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 315)

def create_agent(self, name):
    """Create a new emulation agent."""

Purpose: Create a new emulation agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 322)

def run(self, agent_name, request):
    """Execute emulation request."""

Purpose: Execute emulation request.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – request: Parameter for request

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = EmulationType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 263: Simulation Pattern with LangChain/LangGraph

Purpose: Environment modeling and scenario simulation with intelligent analysis.
Provides Monte Carlo, discrete event, agent-based simulations with LLM insights.
"""

from typing import Dict, Any, List, Optional, 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 random
import statistics
import time


class SimulationType(Enum):
    """Types of simulations."""
    MONTE_CARLO = "monte_carlo"
    DISCRETE_EVENT = "discrete_event"
    AGENT_BASED = "agent_based"
    SYSTEM_DYNAMICS = "system_dynamics"


class SimulationStatus(Enum):
    """Simulation execution status."""
    INITIALIZED = "initialized"
    RUNNING = "running"
    ANALYZING = "analyzing"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class SimulationScenario:
    """Simulation scenario configuration."""
    scenario_id: str
    name: str
    simulation_type: SimulationType
    initial_state: Dict[str, float]
    steps: int = 100
    iterations: int = 10
    parameters: Dict[str, Any] = field(default_factory=dict)


@dataclass
class SimulationMetrics:
    """Metrics from simulation run."""
    execution_time: float
    total_steps: int
    total_iterations: int
    convergence_score: float
    variance: Dict[str, float]
    stability_index: float


@dataclass
class SimulationResult:
    """Complete simulation result."""
    scenario: SimulationScenario
    final_state: Dict[str, float]
    state_history: List[Dict[str, float]]
    metrics: SimulationMetrics
    insights: List[str]
    anomalies: List[str]
    recommendations: List[str]


class SimulationState(TypedDict):
    """State for simulation workflow."""
    scenario_id: str
    simulation_type: str
    initial_state: Dict[str, float]
    steps: int
    iterations: int
    parameters: Dict[str, Any]
    final_state: Dict[str, float]
    state_history: List[Dict[str, float]]
    metrics: Dict[str, Any]
    insights: List[str]
    anomalies: List[str]
    recommendations: List[str]


class IntelligentSimulator:
    """
    Intelligent Simulator with LangGraph workflow integration.
    
    Features:
    - Monte Carlo simulation
    - Discrete event simulation
    - Agent-based modeling
    - LLM-powered pattern detection
    - Anomaly detection
    - Predictive insights
    """
    
    def __init__(self, model_name: str = "gpt-4", temperature: float = 0.5):
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.results: Dict[str, SimulationResult] = {}
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_simulations": 0,
            "simulations_by_type": {t.value: 0 for t in SimulationType},
            "total_steps": 0,
            "total_iterations": 0,
            "avg_execution_time": 0.0
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(SimulationState)
        
        workflow.add_node("initialize", self._initialize_node)
        workflow.add_node("run_simulation", self._run_simulation_node)
        workflow.add_node("calculate_metrics", self._calculate_metrics_node)
        workflow.add_node("detect_anomalies", self._detect_anomalies_node)
        workflow.add_node("generate_insights", self._generate_insights_node)
        
        workflow.set_entry_point("initialize")
        workflow.add_edge("initialize", "run_simulation")
        workflow.add_edge("run_simulation", "calculate_metrics")
        workflow.add_edge("calculate_metrics", "detect_anomalies")
        workflow.add_edge("detect_anomalies", "generate_insights")
        workflow.add_edge("generate_insights", END)
        
        return workflow.compile()
    
    def _initialize_node(self, state: SimulationState) -> SimulationState:
        """Initialize simulation state."""
        state["final_state"] = {}
        state["state_history"] = []
        state["metrics"] = {}
        state["insights"] = []
        state["anomalies"] = []
        state["recommendations"] = []
        return state
    
    def _run_simulation_node(self, state: SimulationState) -> SimulationState:
        """Run the simulation."""
        sim_type = state["simulation_type"]
        
        if sim_type == SimulationType.MONTE_CARLO.value:
            result = self._run_monte_carlo(state)
        elif sim_type == SimulationType.AGENT_BASED.value:
            result = self._run_agent_based(state)
        elif sim_type == SimulationType.DISCRETE_EVENT.value:
            result = self._run_discrete_event(state)
        else:
            result = self._run_monte_carlo(state)
        
        state["final_state"] = result["final_state"]
        state["state_history"] = result["state_history"]
        
        return state
    
    def _run_monte_carlo(self, state: SimulationState) -> Dict[str, Any]:
        """Run Monte Carlo simulation."""
        initial = state["initial_state"]
        steps = state["steps"]
        iterations = state["iterations"]
        
        all_final_states = []
        state_history = []
        
        for _ in range(iterations):
            current_state = initial.copy()
            iteration_history = [current_state.copy()]
            
            for step in range(steps):
                # Apply random variations
                for key in current_state:
                    variation = random.gauss(0, 0.05)  # 5% std dev
                    current_state[key] = max(0, current_state[key] * (1 + variation))
                
                iteration_history.append(current_state.copy())
            
            all_final_states.append(current_state.copy())
            if not state_history:
                state_history = iteration_history
        
        # Calculate average final state
        final_state = {}
        for key in initial.keys():
            values = [s[key] for s in all_final_states]
            final_state[key] = statistics.mean(values)
        
        return {
            "final_state": final_state,
            "state_history": state_history,
            "all_final_states": all_final_states
        }
    
    def _run_agent_based(self, state: SimulationState) -> Dict[str, Any]:
        """Run agent-based simulation."""
        initial = state["initial_state"]
        steps = state["steps"]
        num_agents = state["parameters"].get("num_agents", 10)
        
        # Initialize agents
        agents = [initial.copy() for _ in range(num_agents)]
        state_history = []
        
        for step in range(steps):
            # Agent interactions
            for i in range(len(agents)):
                # Influence from neighbors
                if i > 0 and i  Dict[str, Any]:
        """Run discrete event simulation."""
        current_state = state["initial_state"].copy()
        events = state["parameters"].get("events", [])
        state_history = [current_state.copy()]
        
        # Sort events by time
        sorted_events = sorted(events, key=lambda e: e.get("time", 0))
        
        for event in sorted_events:
            # Apply event effects
            if "effects" in event:
                for key, delta in event["effects"].items():
                    if key in current_state:
                        current_state[key] += delta
            
            state_history.append(current_state.copy())
        
        return {
            "final_state": current_state,
            "state_history": state_history
        }
    
    def _calculate_metrics_node(self, state: SimulationState) -> SimulationState:
        """Calculate simulation metrics."""
        history = state["state_history"]
        
        if not history:
            return state
        
        # Calculate variance
        variance = {}
        for key in state["initial_state"].keys():
            values = [s[key] for s in history if key in s]
            if values:
                variance[key] = statistics.variance(values) if len(values) > 1 else 0.0
        
        # Calculate convergence (lower variance at end = higher convergence)
        if len(history) > 10:
            recent_variance = {}
            for key in state["initial_state"].keys():
                recent_values = [s[key] for s in history[-10:] if key in s]
                if recent_values:
                    recent_variance[key] = statistics.variance(recent_values) if len(recent_values) > 1 else 0.0
            
            avg_variance = statistics.mean(recent_variance.values()) if recent_variance else 0
            convergence = max(0, 1 - min(avg_variance, 1.0))
        else:
            convergence = 0.5
        
        # Stability index
        stability = convergence
        
        state["metrics"] = {
            "variance": variance,
            "convergence_score": convergence,
            "stability_index": stability,
            "total_steps": len(history),
            "total_iterations": state["iterations"]
        }
        
        return state
    
    def _detect_anomalies_node(self, state: SimulationState) -> SimulationState:
        """Detect anomalies in simulation results."""
        history = state["state_history"]
        anomalies = []
        
        if len(history)  0.5:  # 50% change
                    anomalies.append(f"Sudden change in {key} at step {i}: {change_pct*100:.1f}%")
        
        # Detect extreme values
        for key in state["initial_state"].keys():
            values = [s[key] for s in history if key in s]
            if values:
                mean_val = statistics.mean(values)
                std_val = statistics.stdev(values) if len(values) > 1 else 0
                
                for i, val in enumerate(values):
                    if std_val > 0 and abs(val - mean_val) > 3 * std_val:
                        anomalies.append(f"Outlier in {key} at step {i}: {val:.2f}")
        
        state["anomalies"] = anomalies[:5]  # Limit to 5
        return state
    
    def _generate_insights_node(self, state: SimulationState) -> SimulationState:
        """Generate AI insights from simulation."""
        final_state = state["final_state"]
        initial_state = state["initial_state"]
        metrics = state["metrics"]
        anomalies = state["anomalies"]
        
        # Use LLM for insights
        prompt = ChatPromptTemplate.from_template(
            "Analyze this simulation result:\n"
            "Type: {sim_type}\n"
            "Initial State: {initial}\n"
            "Final State: {final}\n"
            "Convergence: {convergence:.2f}\n"
            "Anomalies: {anomalies}\n\n"
            "Provide 3 key insights about the simulation behavior."
        )
        
        chain = prompt | self.llm
        response = chain.invoke({
            "sim_type": state["simulation_type"],
            "initial": str(initial_state),
            "final": str(final_state),
            "convergence": metrics.get("convergence_score", 0),
            "anomalies": str(anomalies)
        })
        
        insights_text = str(response.content)
        insights = [insights_text[:200]]
        
        # Generate recommendations
        recommendations = []
        
        if metrics.get("convergence_score", 0)  3:
            recommendations.append("High number of anomalies detected - review simulation parameters")
        
        for key, var in metrics.get("variance", {}).items():
            if var > 1.0:
                recommendations.append(f"High variance in {key} - consider stabilization measures")
        
        state["insights"] = insights
        state["recommendations"] = recommendations
        
        return state
    
    def simulate(self, scenario: SimulationScenario) -> SimulationResult:
        """Run simulation with workflow."""
        start_time = time.time()
        
        initial_state: SimulationState = {
            "scenario_id": scenario.scenario_id,
            "simulation_type": scenario.simulation_type.value,
            "initial_state": scenario.initial_state,
            "steps": scenario.steps,
            "iterations": scenario.iterations,
            "parameters": scenario.parameters,
            "final_state": {},
            "state_history": [],
            "metrics": {},
            "insights": [],
            "anomalies": [],
            "recommendations": []
        }
        
        # Run workflow
        result_state = self.workflow.invoke(initial_state)
        
        execution_time = time.time() - start_time
        
        # Create result
        metrics = SimulationMetrics(
            execution_time=execution_time,
            total_steps=result_state["metrics"].get("total_steps", 0),
            total_iterations=result_state["metrics"].get("total_iterations", 0),
            convergence_score=result_state["metrics"].get("convergence_score", 0.0),
            variance=result_state["metrics"].get("variance", {}),
            stability_index=result_state["metrics"].get("stability_index", 0.0)
        )
        
        result = SimulationResult(
            scenario=scenario,
            final_state=result_state["final_state"],
            state_history=result_state["state_history"],
            metrics=metrics,
            insights=result_state["insights"],
            anomalies=result_state["anomalies"],
            recommendations=result_state["recommendations"]
        )
        
        self.results[scenario.scenario_id] = result
        
        # Update stats
        self.stats["total_simulations"] += 1
        self.stats["simulations_by_type"][scenario.simulation_type.value] += 1
        self.stats["total_steps"] += scenario.steps
        self.stats["total_iterations"] += scenario.iterations
        
        if self.stats["total_simulations"] > 0:
            total_time = self.stats["avg_execution_time"] * (self.stats["total_simulations"] - 1) + execution_time
            self.stats["avg_execution_time"] = total_time / self.stats["total_simulations"]
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get simulation statistics."""
        return {
            **self.stats,
            "total_results_stored": len(self.results)
        }


class SimulationPattern:
    """
    Comprehensive Simulation Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.simulator = IntelligentSimulator(model_name=model_name)
    
    def create_scenario(
        self,
        scenario_id: str,
        name: str,
        simulation_type: SimulationType,
        initial_state: Dict[str, float],
        steps: int = 100,
        iterations: int = 10,
        parameters: Optional[Dict[str, Any]] = None
    ) -> SimulationScenario:
        """Create simulation scenario."""
        return SimulationScenario(
            scenario_id=scenario_id,
            name=name,
            simulation_type=simulation_type,
            initial_state=initial_state,
            steps=steps,
            iterations=iterations,
            parameters=parameters or {}
        )
    
    def run_simulation(self, scenario: SimulationScenario) -> SimulationResult:
        """Run simulation."""
        return self.simulator.simulate(scenario)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics."""
        return self.simulator.get_statistics()


if __name__ == "__main__":
    print("=== Simulation Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Monte Carlo Simulation")
    print("-" * 50)
    pattern = SimulationPattern(model_name="gpt-4")
    
    # Create Monte Carlo scenario
    mc_scenario = pattern.create_scenario(
        scenario_id="mc-001",
        name="Market Price Simulation",
        simulation_type=SimulationType.MONTE_CARLO,
        initial_state={"price": 100.0, "volume": 1000.0, "volatility": 0.2},
        steps=50,
        iterations=100
    )
    
    print(f"Running {mc_scenario.name}...")
    result1 = pattern.run_simulation(mc_scenario)
    
    print(f"\nFinal State:")
    for key, value in result1.final_state.items():
        print(f"  {key}: {value:.2f}")
    
    print(f"\nMetrics:")
    print(f"  Execution Time: {result1.metrics.execution_time:.3f}s")
    print(f"  Convergence Score: {result1.metrics.convergence_score:.3f}")
    print(f"  Stability Index: {result1.metrics.stability_index:.3f}")
    
    if result1.anomalies:
        print(f"\nAnomalies Detected: {len(result1.anomalies)}")
        for anomaly in result1.anomalies[:3]:
            print(f"  - {anomaly}")
    
    if result1.insights:
        print(f"\nAI Insights:")
        for insight in result1.insights:
            print(f"  {insight}")
    
    if result1.recommendations:
        print(f"\nRecommendations:")
        for rec in result1.recommendations:
            print(f"  - {rec}")
    
    print("\n" + "=" * 50)
    print("\nScenario 2: Agent-Based Simulation")
    print("-" * 50)
    
    ab_scenario = pattern.create_scenario(
        scenario_id="ab-001",
        name="Opinion Dynamics",
        simulation_type=SimulationType.AGENT_BASED,
        initial_state={"opinion": 0.5, "confidence": 0.8},
        steps=30,
        iterations=1,
        parameters={"num_agents": 20}
    )
    
    print(f"Running {ab_scenario.name}...")
    result2 = pattern.run_simulation(ab_scenario)
    
    print(f"\nFinal State:")
    for key, value in result2.final_state.items():
        print(f"  {key}: {value:.2f}")
    
    print(f"\nMetrics:")
    print(f"  Execution Time: {result2.metrics.execution_time:.3f}s")
    print(f"  Total Steps: {result2.metrics.total_steps}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Discrete Event Simulation")
    print("-" * 50)
    
    events = [
        {"time": 5, "effects": {"queue_length": 10}},
        {"time": 10, "effects": {"queue_length": -5}},
        {"time": 15, "effects": {"queue_length": 8}},
        {"time": 20, "effects": {"queue_length": -12}}
    ]
    
    de_scenario = pattern.create_scenario(
        scenario_id="de-001",
        name="Queue Management",
        simulation_type=SimulationType.DISCRETE_EVENT,
        initial_state={"queue_length": 0.0, "throughput": 100.0},
        steps=25,
        iterations=1,
        parameters={"events": events}
    )
    
    print(f"Running {de_scenario.name}...")
    result3 = pattern.run_simulation(de_scenario)
    
    print(f"\nFinal State:")
    for key, value in result3.final_state.items():
        print(f"  {key}: {value:.2f}")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    stats = pattern.get_statistics()
    print(f"Total Simulations: {stats['total_simulations']}")
    print(f"Total Steps Simulated: {stats['total_steps']}")
    print(f"Total Iterations: {stats['total_iterations']}")
    print(f"Average Execution Time: {stats['avg_execution_time']:.3f}s")
    
    print(f"\nSimulations by Type:")
    for sim_type, count in stats['simulations_by_type'].items():
        if count > 0:
            print(f"  {sim_type}: {count}")
    
    print("\n" + "=" * 50)
    print("Simulation pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Simulation

Pattern 263: Simulation Pattern – Line-by-Line Explanation

Overview

This pattern implements environment modeling and scenario simulation with intelligent analysis using LangChain/LangGraph. It supports Monte Carlo, discrete event, and agent-based simulations with LLM-powered insights.

Code Explanation

Lines 1-18: Imports and Setup

"""
Pattern 263: Simulation Pattern with LangChain/LangGraph

Purpose: Environment modeling and scenario simulation with intelligent analysis.
Provides Monte Carlo, discrete event, agent-based simulations with LLM insights.
"""

Explanation: Module docstring describing the pattern’s purpose – providing comprehensive simulation capabilities with AI-powered analysis.

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

Explanation: Import type hints for better code documentation and IDE support. TypedDict is crucial for LangGraph state definitions.

from dataclasses import dataclass, field

Explanation: Import dataclass decorators for creating structured data classes with automatic __init__, __repr__, etc.

from datetime import datetime

Explanation: Import datetime for timestamping simulation results.

from enum import Enum

Explanation: Import Enum for creating type-safe enumeration classes.

from langchain_core.prompts import ChatPromptTemplate

Explanation: Import LangChain’s prompt template system for structured LLM interactions.

from langchain_openai import ChatOpenAI

Explanation: Import OpenAI chat model integration from LangChain.

from langgraph.graph import StateGraph, END

Explanation: Import LangGraph’s state graph workflow system. StateGraph creates workflows, END marks terminal nodes.

import random
import numpy as np

Explanation: Import libraries for random number generation and numerical operations (Monte Carlo simulations).

Lines 20-36: Enums Definition

class SimulationType(Enum):
    """Types of simulations."""
    MONTE_CARLO = "monte_carlo"
    DISCRETE_EVENT = "discrete_event"
    AGENT_BASED = "agent_based"
    SYSTEM_DYNAMICS = "system_dynamics"

Explanation: Enum defining four simulation types: – MONTE_CARLO: Random sampling-based simulations – DISCRETE_EVENT: Event-driven state transitions – AGENT_BASED: Multi-agent interaction modeling – SYSTEM_DYNAMICS: Continuous system behavior modeling

class SimulationStatus(Enum):
    """Simulation execution status."""
    INITIALIZED = "initialized"
    RUNNING = "running"
    ANALYZING = "analyzing"
    COMPLETED = "completed"
    FAILED = "failed"

Explanation: Enum tracking simulation lifecycle states from initialization through completion or failure.

Lines 38-72: Dataclasses

@dataclass
class SimulationScenario:
    """Simulation scenario configuration."""
    scenario_id: str
    name: str
    simulation_type: SimulationType
    initial_state: Dict[str, float]
    steps: int = 100
    iterations: int = 10
    parameters: Dict[str, Any] = field(default_factory=dict)

Explanation: Dataclass defining simulation scenario configuration: – scenario_id: Unique identifier – name: Human-readable name – simulation_type: Type from SimulationType enum – initial_state: Starting values (e.g., {“price”: 100.0}) – steps: Number of simulation steps per iteration – iterations: How many times to run the simulation – parameters: Additional configuration (default empty dict)

@dataclass
class SimulationMetrics:
    """Metrics from simulation run."""
    execution_time: float
    total_steps: int
    total_iterations: int
    convergence_score: float
    variance: Dict[str, float]
    stability_index: float

Explanation: Dataclass storing simulation performance metrics: – execution_time: How long simulation took (seconds) – convergence_score: 0.0-1.0 score indicating if simulation converged – variance: Variability of each state variable – stability_index: Overall stability measure

@dataclass
class SimulationResult:
    """Complete simulation result."""
    scenario: SimulationScenario
    final_state: Dict[str, float]
    state_history: List[Dict[str, float]]
    metrics: SimulationMetrics
    insights: List[str]
    anomalies: List[str]
    recommendations: List[str]

Explanation: Complete result package containing scenario, final state, full history, metrics, and LLM-generated insights.

Lines 74-87: TypedDict State Definition

class SimulationState(TypedDict):
    """State for simulation workflow."""
    scenario_id: str
    simulation_type: str
    initial_state: Dict[str, float]
    steps: int
    iterations: int
    parameters: Dict[str, Any]
    final_state: Dict[str, float]
    state_history: List[Dict[str, float]]
    metrics: Dict[str, Any]
    insights: List[str]
    anomalies: List[str]
    recommendations: List[str]

Explanation: TypedDict defining the structure of state passed between workflow nodes. This is required by LangGraph to type-check workflow state transitions. Contains all information needed throughout the simulation workflow.

Lines 89-116: IntelligentSimulator Class Initialization

class IntelligentSimulator:
    """
    Intelligent Simulator with LangGraph workflow integration.

    Features:
    - Monte Carlo simulation
    - Discrete event simulation
    - Agent-based modeling
    - LLM-powered pattern detection
    - Anomaly detection
    - Predictive insights
    """

Explanation: Main simulator class docstring listing all capabilities.

def __init__(self, model_name: str = "gpt-4", temperature: float = 0.5):
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)
    self.results: Dict[str, SimulationResult] = {}
    self.workflow = self._build_workflow()

Explanation: Initialize simulator with: – LLM for generating insights (temperature 0.5 balances creativity and consistency) – Results dictionary to store simulation outputs – Compiled workflow graph

    # Statistics
    self.stats = {
        "total_simulations": 0,
        "simulations_by_type": {t.value: 0 for t in SimulationType},
        "total_steps": 0,
        "total_iterations": 0,
        "avg_execution_time": 0.0
    }

Explanation: Initialize statistics tracking dictionary with counters for each simulation type and aggregate metrics.

Lines 118-136: Workflow Building

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

Explanation: Create a new StateGraph with SimulationState as the state type. This enforces type safety.

    workflow.add_node("initialize", self._initialize_node)
    workflow.add_node("run_simulation", self._run_simulation_node)
    workflow.add_node("calculate_metrics", self._calculate_metrics_node)
    workflow.add_node("detect_anomalies", self._detect_anomalies_node)
    workflow.add_node("generate_insights", self._generate_insights_node)

Explanation: Add 5 nodes to the workflow graph, each pointing to a method that processes the state.

    workflow.set_entry_point("initialize")
    workflow.add_edge("initialize", "run_simulation")
    workflow.add_edge("run_simulation", "calculate_metrics")
    workflow.add_edge("calculate_metrics", "detect_anomalies")
    workflow.add_edge("detect_anomalies", "generate_insights")
    workflow.add_edge("generate_insights", END)

Explanation: Define workflow execution order: 1. Start at “initialize” 2. Flow through run → metrics → anomalies → insights 3. End after insights generation

    return workflow.compile()

Explanation: Compile the workflow into an executable graph. This validates the graph structure.

Lines 138-146: Initialize Node

def _initialize_node(self, state: SimulationState) -> SimulationState:
    """Initialize simulation state."""
    state["final_state"] = {}
    state["state_history"] = []
    state["metrics"] = {}
    state["insights"] = []
    state["anomalies"] = []
    state["recommendations"] = []
    return state

Explanation: First workflow node – initializes empty containers for results. Takes state dict, modifies it, returns modified state. This pattern is standard for all nodes.

Lines 148-162: Run Simulation Node (Dispatcher)

def _run_simulation_node(self, state: SimulationState) -> SimulationState:
    """Run the simulation."""
    sim_type = state["simulation_type"]

    if sim_type == "monte_carlo":
        result = self._run_monte_carlo(state)
    elif sim_type == "agent_based":
        result = self._run_agent_based(state)
    elif sim_type == "discrete_event":
        result = self._run_discrete_event(state)
    else:
        result = {"final_state": state["initial_state"], "history": []}

Explanation: Dispatcher node that routes to appropriate simulation method based on type. Returns simulation result dict.

    state["final_state"] = result["final_state"]
    state["state_history"] = result["history"]
    return state

Explanation: Update state with simulation results and return.

Lines 164-198: Monte Carlo Simulation

def _run_monte_carlo(self, state: SimulationState) -> Dict[str, Any]:
    """Run Monte Carlo simulation."""
    initial = state["initial_state"]
    steps = state["steps"]
    iterations = state["iterations"]

Explanation: Extract parameters from state.

    all_histories = []

    for _ in range(iterations):
        current = initial.copy()
        history = [current.copy()]

Explanation: Run multiple iterations, each starting from initial state. Track history for each iteration.

        for _ in range(steps):
            new_state = {}
            for key, value in current.items():
                # Random walk with small variance
                change = random.gauss(0, value * 0.02)
                new_state[key] = max(0, value + change)

            current = new_state
            history.append(current.copy())

Explanation: For each step, apply random changes to each state variable using Gaussian distribution (mean=0, std=2% of current value). Ensure non-negative values.

        all_histories.append(history)

Explanation: Store this iteration’s history.

    # Average across all iterations
    final_state = {}
    for key in initial.keys():
        values = [h[-1][key] for h in all_histories]
        final_state[key] = sum(values) / len(values)

Explanation: Calculate final state as average of all iterations’ final values.

    # Use history from first iteration for display
    return {
        "final_state": final_state,
        "history": all_histories[0]
    }

Explanation: Return averaged final state and first iteration’s history for visualization.

Lines 200-230: Agent-Based Simulation

def _run_agent_based(self, state: SimulationState) -> Dict[str, Any]:
    """Run agent-based simulation."""
    initial = state["initial_state"]
    steps = state["steps"]
    num_agents = state["parameters"].get("num_agents", 10)

Explanation: Extract parameters, including number of agents from parameters dict (default 10).

    # Initialize agents
    agents = [{key: value for key, value in initial.items()} for _ in range(num_agents)]

Explanation: Create list of agents, each with a copy of initial state.

    history = [initial.copy()]

    for _ in range(steps):
        # Agents interact and influence each other
        for i, agent in enumerate(agents):
            for key in agent.keys():
                # Average with neighbors
                neighbors = [agents[j][key] for j in range(num_agents) if j != i]
                if neighbors:
                    influence = sum(neighbors) / len(neighbors)
                    agent[key] = 0.7 * agent[key] + 0.3 * influence

Explanation: Each step, each agent updates its state by averaging with all other agents (70% own state, 30% average of others). This simulates social influence or opinion dynamics.

        # Aggregate state
        aggregate = {}
        for key in initial.keys():
            aggregate[key] = sum(a[key] for a in agents) / num_agents

        history.append(aggregate.copy())

Explanation: Calculate aggregate state across all agents and add to history.

    return {
        "final_state": history[-1],
        "history": history
    }

Explanation: Return final aggregate state and full history.

Lines 232-253: Discrete Event Simulation

def _run_discrete_event(self, state: SimulationState) -> Dict[str, Any]:
    """Run discrete event simulation."""
    current = state["initial_state"].copy()
    steps = state["steps"]
    events = state["parameters"].get("events", [])

Explanation: Extract events list from parameters. Events are predefined changes at specific times.

    history = [current.copy()]

    for step in range(steps):
        # Apply events at this step
        for event in events:
            if event.get("time") == step:
                effects = event.get("effects", {})
                for key, value in effects.items():
                    if key in current:
                        current[key] = max(0, current[key] + value)

Explanation: For each step, check if any events occur. If so, apply their effects by modifying state values.

        history.append(current.copy())

Explanation: Record state after events applied.

    return {
        "final_state": current,
        "history": history
    }

Explanation: Return final state and complete history.

Lines 255-293: Calculate Metrics Node

def _calculate_metrics_node(self, state: SimulationState) -> SimulationState:
    """Calculate simulation metrics."""
    history = state["state_history"]

    if not history:
        return state

Explanation: Extract history, return early if empty.

    # Calculate variance for each variable
    variance = {}
    for key in state["initial_state"].keys():
        values = [h[key] for h in history if key in h]
        if len(values) > 1:
            mean_val = sum(values) / len(values)
            variance[key] = sum((v - mean_val) ** 2 for v in values) / len(values)

Explanation: Calculate variance for each state variable across all history steps. Variance measures how much values fluctuate.

    # Convergence score (how stable is final state)
    if len(history) >= 10:
        recent = history[-10:]
        recent_variance = {}
        for key in state["initial_state"].keys():
            values = [h[key] for h in recent if key in h]
            if len(values) > 1:
                mean_val = sum(values) / len(values)
                recent_variance[key] = sum((v - mean_val) ** 2 for v in values) / len(values)

        # Lower recent variance = higher convergence
        avg_recent_var = sum(recent_variance.values()) / len(recent_variance) if recent_variance else 1.0
        convergence_score = 1.0 / (1.0 + avg_recent_var)
    else:
        convergence_score = 0.5

Explanation: Calculate convergence score (0.0-1.0) by looking at last 10 steps. Lower variance in recent history = higher convergence (system is stabilizing).

    # Stability index
    avg_variance = sum(variance.values()) / len(variance) if variance else 0.0
    stability_index = 1.0 / (1.0 + avg_variance)

Explanation: Calculate overall stability as inverse of average variance. Higher stability = lower variance.

    state["metrics"] = {
        "variance": variance,
        "convergence_score": convergence_score,
        "stability_index": stability_index
    }

    return state

Explanation: Store calculated metrics in state and return.

Lines 295-325: Detect Anomalies Node

def _detect_anomalies_node(self, state: SimulationState) -> SimulationState:
    """Detect anomalies in simulation results."""
    history = state["state_history"]
    anomalies = []

    if len(history) < 2:
        state["anomalies"] = anomalies
        return state

Explanation: Initialize anomalies list, return early if insufficient history.

    # Detect sudden changes
    for i in range(1, len(history)):
        prev = history[i-1]
        curr = history[i]

        for key in curr.keys():
            if key in prev:
                change_pct = abs(curr[key] - prev[key]) / (prev[key] + 1e-10)
                if change_pct > 0.5:  # 50% change
                    anomalies.append(
                        f"Large change in {key} at step {i}: {prev[key]:.2f} → {curr[key]:.2f}"
                    )

Explanation: Detect anomalies by checking for large percentage changes (>50%) between consecutive steps. The 1e-10 prevents division by zero.

    # Detect outliers in final state
    for key, value in state["final_state"].items():
        initial_value = state["initial_state"].get(key, 0)
        if initial_value > 0:
            final_ratio = value / initial_value
            if final_ratio > 10 or final_ratio < 0.1:
                anomalies.append(
                    f"Extreme final value for {key}: {initial_value:.2f} → {value:.2f}"
                )

Explanation: Detect outliers by comparing final values to initial values. Flag if final is >10x or <0.1x initial.

    state["anomalies"] = anomalies
    return state

Explanation: Store detected anomalies and return state.

Lines 327-373: Generate Insights Node (LLM-Powered)

def _generate_insights_node(self, state: SimulationState) -> SimulationState:
    """Generate insights using LLM."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert in simulation analysis and predictive modeling."),
        ("user", """Analyze this simulation result:

Simulation Type: {sim_type}
Initial State: {initial}
Final State: {final}
Convergence Score: {convergence}
Stability Index: {stability}
Anomalies: {anomalies}

Provide 3-5 key insights about the simulation behavior, patterns, and recommendations.""")
    ])