Agentic Architecture Pattern: Mock Agent

Written by

in


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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *