aishastr.live

Agentic Architecture Pattern: Shadow Deployment

Written by

admin

in


    """
Pattern 274: Shadow Deployment

Purpose: Parallel production testing with traffic mirroring and result comparison
"""

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
import time


class ShadowMode(Enum):
    """Shadow deployment mode."""
    PASSIVE = "passive"  # Shadow runs but results not used
    COMPARATIVE = "comparative"  # Results compared for analysis
    VALIDATION = "validation"  # Results validated against production


class ComparisonStatus(Enum):
    """Status of result comparison."""
    MATCH = "match"
    MISMATCH = "mismatch"
    ERROR = "error"
    NOT_COMPARED = "not_compared"


@dataclass
class Request:
    """Represents a single request."""
    id: str
    data: Any
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Response:
    """Represents a response from a version."""
    request_id: str
    version: str
    result: Any
    latency_ms: float
    success: bool
    error_message: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Comparison:
    """Comparison between production and shadow responses."""
    request_id: str
    production_response: Response
    shadow_response: Response
    status: ComparisonStatus
    differences: List[str] = field(default_factory=list)
    similarity_score: float = 0.0  # 0-1 scale
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ShadowMetrics:
    """Metrics for shadow deployment."""
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    comparisons_match: int = 0
    comparisons_mismatch: int = 0
    comparisons_error: int = 0
    avg_latency_diff_ms: float = 0.0
    avg_similarity_score: float = 0.0


class ShadowDeployment:
    """
    Implements shadow deployment pattern.
    
    Features:
    - Parallel execution with production
    - Traffic mirroring to shadow version
    - Result comparison and analysis
    - Performance metrics collection
    - No impact on production traffic
    - Confidence building for new versions
    """
    
    def __init__(
        self,
        production_version: str = "1.0.0",
        shadow_version: Optional[str] = None,
        mode: ShadowMode = ShadowMode.COMPARATIVE,
        traffic_mirror_percentage: float = 100.0
    ):
        self.production_version = production_version
        self.shadow_version = shadow_version
        self.mode = mode
        self.traffic_mirror_percentage = traffic_mirror_percentage
        
        self.is_active = False
        self.production_handler: Optional[Callable] = None
        self.shadow_handler: Optional[Callable] = None
        
        self.requests: Dict[str, Request] = {}
        self.production_responses: Dict[str, Response] = {}
        self.shadow_responses: Dict[str, Response] = {}
        self.comparisons: List[Comparison] = []
        
        self.metrics = ShadowMetrics()
        
        # Statistics
        self.statistics = {
            "total_deployments": 0,
            "active_deployments": 0,
            "total_requests_processed": 0,
            "current_version": production_version,
            "shadow_version": shadow_version
        }
    
    def start_shadow(
        self,
        shadow_version: str,
        production_handler: Callable,
        shadow_handler: Callable
    ) -> Dict[str, Any]:
        """
        Start shadow deployment.
        
        Args:
            shadow_version: Version of shadow deployment
            production_handler: Function to handle production requests
            shadow_handler: Function to handle shadow requests
            
        Returns:
            Status of shadow start
        """
        if self.is_active:
            return {
                "success": False,
                "message": "Shadow deployment already active"
            }
        
        self.shadow_version = shadow_version
        self.production_handler = production_handler
        self.shadow_handler = shadow_handler
        self.is_active = True
        
        # Reset metrics
        self.metrics = ShadowMetrics()
        self.requests = {}
        self.production_responses = {}
        self.shadow_responses = {}
        self.comparisons = []
        
        self.statistics["total_deployments"] += 1
        self.statistics["active_deployments"] += 1
        self.statistics["shadow_version"] = shadow_version
        
        return {
            "success": True,
            "shadow_version": shadow_version,
            "mode": self.mode.value,
            "traffic_mirror_percentage": self.traffic_mirror_percentage,
            "message": f"Shadow deployment started for version {shadow_version}"
        }
    
    def process_request(self, request_id: str, request_data: Any) -> Dict[str, Any]:
        """
        Process request through both production and shadow.
        
        Args:
            request_id: Unique request identifier
            request_data: Request data to process
            
        Returns:
            Production response (shadow runs in parallel)
        """
        if not self.is_active or not self.production_handler or not self.shadow_handler:
            return {
                "success": False,
                "message": "Shadow deployment not properly configured"
            }
        
        # Store request
        request = Request(id=request_id, data=request_data)
        self.requests[request_id] = request
        self.metrics.total_requests += 1
        self.statistics["total_requests_processed"] += 1
        
        # Determine if this request should be mirrored
        import random
        should_mirror = random.random() * 100  Comparison:
        """Compare production and shadow responses."""
        differences = []
        
        # Check success status
        if prod_response.success != shadow_response.success:
            differences.append(f"Success status differs: prod={prod_response.success}, shadow={shadow_response.success}")
        
        # Check results (basic comparison)
        if prod_response.result != shadow_response.result:
            differences.append(f"Results differ: prod={prod_response.result}, shadow={shadow_response.result}")
        
        # Check latency difference
        latency_diff = abs(shadow_response.latency_ms - prod_response.latency_ms)
        if latency_diff > 100:  # More than 100ms difference
            differences.append(f"Significant latency difference: {latency_diff:.2f}ms")
        
        # Calculate similarity score (simple version)
        similarity_score = 1.0
        if differences:
            # Reduce score based on number of differences
            similarity_score = max(0.0, 1.0 - (len(differences) * 0.2))
        
        # Determine status
        if not prod_response.success or not shadow_response.success:
            status = ComparisonStatus.ERROR
        elif not differences:
            status = ComparisonStatus.MATCH
        else:
            status = ComparisonStatus.MISMATCH
        
        return Comparison(
            request_id=prod_response.request_id,
            production_response=prod_response,
            shadow_response=shadow_response,
            status=status,
            differences=differences,
            similarity_score=similarity_score
        )
    
    def _update_comparison_metrics(self, comparison: Comparison):
        """Update metrics based on comparison."""
        if comparison.status == ComparisonStatus.MATCH:
            self.metrics.comparisons_match += 1
        elif comparison.status == ComparisonStatus.MISMATCH:
            self.metrics.comparisons_mismatch += 1
        elif comparison.status == ComparisonStatus.ERROR:
            self.metrics.comparisons_error += 1
        
        # Update average similarity
        total_comparisons = len(self.comparisons)
        if total_comparisons > 0:
            self.metrics.avg_similarity_score = sum(c.similarity_score for c in self.comparisons) / total_comparisons
        
        # Update average latency difference
        if comparison.production_response.success and comparison.shadow_response.success:
            latency_diff = abs(
                comparison.shadow_response.latency_ms - comparison.production_response.latency_ms
            )
            
            comparison_count = self.metrics.comparisons_match + self.metrics.comparisons_mismatch
            if comparison_count > 0:
                current_avg = self.metrics.avg_latency_diff_ms
                self.metrics.avg_latency_diff_ms = (
                    (current_avg * (comparison_count - 1) + latency_diff) / comparison_count
                )
    
    def stop_shadow(self) -> Dict[str, Any]:
        """Stop shadow deployment and return final analysis."""
        if not self.is_active:
            return {
                "success": False,
                "message": "No active shadow deployment"
            }
        
        self.is_active = False
        self.statistics["active_deployments"] -= 1
        
        analysis = self.get_analysis()
        
        return {
            "success": True,
            "message": f"Shadow deployment stopped for version {self.shadow_version}",
            "analysis": analysis
        }
    
    def get_analysis(self) -> Dict[str, Any]:
        """Get detailed analysis of shadow deployment."""
        total_comparisons = len(self.comparisons)
        
        match_rate = 0.0
        if total_comparisons > 0:
            match_rate = (self.metrics.comparisons_match / total_comparisons) * 100
        
        return {
            "production_version": self.production_version,
            "shadow_version": self.shadow_version,
            "mode": self.mode.value,
            "requests": {
                "total": self.metrics.total_requests,
                "successful": self.metrics.successful_requests,
                "failed": self.metrics.failed_requests,
                "success_rate": (self.metrics.successful_requests / self.metrics.total_requests * 100) if self.metrics.total_requests > 0 else 0
            },
            "comparisons": {
                "total": total_comparisons,
                "matches": self.metrics.comparisons_match,
                "mismatches": self.metrics.comparisons_mismatch,
                "errors": self.metrics.comparisons_error,
                "match_rate": match_rate
            },
            "performance": {
                "avg_latency_diff_ms": self.metrics.avg_latency_diff_ms,
                "avg_similarity_score": self.metrics.avg_similarity_score
            },
            "recommendation": self._generate_recommendation(match_rate)
        }
    
    def _generate_recommendation(self, match_rate: float) -> str:
        """Generate deployment recommendation based on analysis."""
        if match_rate >= 95:
            return "Strong confidence: Shadow version performs identically to production. Safe to deploy."
        elif match_rate >= 85:
            return "Good confidence: Minor differences detected. Review mismatches before deploying."
        elif match_rate >= 70:
            return "Moderate confidence: Significant differences detected. Investigate before deploying."
        else:
            return "Low confidence: Major differences detected. Do not deploy without investigation."
    
    def get_status(self) -> Dict[str, Any]:
        """Get current shadow deployment status."""
        return {
            "is_active": self.is_active,
            "production_version": self.production_version,
            "shadow_version": self.shadow_version,
            "mode": self.mode.value,
            "traffic_mirror_percentage": self.traffic_mirror_percentage,
            "metrics": {
                "total_requests": self.metrics.total_requests,
                "successful_requests": self.metrics.successful_requests,
                "failed_requests": self.metrics.failed_requests,
                "comparisons": len(self.comparisons),
                "match_rate": (self.metrics.comparisons_match / len(self.comparisons) * 100) if self.comparisons else 0
            }
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get deployment statistics."""
        return {
            "total_deployments": self.statistics["total_deployments"],
            "active_deployments": self.statistics["active_deployments"],
            "total_requests_processed": self.statistics["total_requests_processed"],
            "current_version": self.statistics["current_version"],
            "shadow_version": self.statistics["shadow_version"]
        }


class ShadowDeploymentPattern:
    """
    Coordinator for shadow deployments.
    
    Manages shadow deployment lifecycle with traffic mirroring and analysis.
    """
    
    def __init__(
        self,
        production_version: str = "1.0.0",
        mode: ShadowMode = ShadowMode.COMPARATIVE,
        traffic_mirror_percentage: float = 100.0
    ):
        self.deployment = ShadowDeployment(
            production_version=production_version,
            mode=mode,
            traffic_mirror_percentage=traffic_mirror_percentage
        )
    
    def start_shadow(
        self,
        shadow_version: str,
        production_handler: Callable,
        shadow_handler: Callable
    ) -> Dict[str, Any]:
        """Start shadow deployment."""
        return self.deployment.start_shadow(shadow_version, production_handler, shadow_handler)
    
    def process_request(self, request_id: str, request_data: Any) -> Dict[str, Any]:
        """Process request through production and shadow."""
        return self.deployment.process_request(request_id, request_data)
    
    def stop_shadow(self) -> Dict[str, Any]:
        """Stop shadow deployment."""
        return self.deployment.stop_shadow()
    
    def get_status(self) -> Dict[str, Any]:
        """Get deployment status."""
        return self.deployment.get_status()
    
    def get_analysis(self) -> Dict[str, Any]:
        """Get deployment analysis."""
        return self.deployment.get_analysis()
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get deployment statistics."""
        return self.deployment.get_statistics()


# Example usage
if __name__ == "__main__":
    print("=== Shadow Deployment Demo ===\n")
    
    # Define production and shadow handlers
    def production_handler(data: Any) -> Any:
        """Simulated production handler."""
        time.sleep(0.01)  # Simulate processing
        return f"Production result for: {data}"
    
    def shadow_handler(data: Any) -> Any:
        """Simulated shadow handler (slightly different)."""
        time.sleep(0.012)  # Slightly slower
        # Simulate 90% match rate
        import random
        if random.random() < 0.9:
            return f"Production result for: {data}"
        else:
            return f"Shadow result for: {data}"  # Different result
    
    # Create shadow deployment
    pattern = ShadowDeploymentPattern(
        production_version="1.0.0",
        mode=ShadowMode.COMPARATIVE,
        traffic_mirror_percentage=100.0
    )
    
    print("1. Starting shadow deployment...")
    result = pattern.start_shadow("2.0.0", production_handler, shadow_handler)
    print(f"  Status: {result['message']}")
    print(f"  Shadow version: {result['shadow_version']}")
    print(f"  Mode: {result['mode']}")
    
    # Process requests
    print("\n2. Processing requests through production and shadow...")
    for i in range(20):
        request_result = pattern.process_request(f"req-{i}", f"data-{i}")
        if i < 3:  # Show first few
            print(f"  Request {i}: success={request_result['success']}, latency={request_result['latency_ms']:.2f}ms, shadow_executed={request_result['shadow_executed']}")
    print(f"  ... processed total of 20 requests")
    
    # Check status
    print("\n3. Current deployment status:")
    status = pattern.get_status()
    print(f"  Active: {status['is_active']}")
    print(f"  Total requests: {status['metrics']['total_requests']}")
    print(f"  Successful: {status['metrics']['successful_requests']}")
    print(f"  Comparisons: {status['metrics']['comparisons']}")
    print(f"  Match rate: {status['metrics']['match_rate']:.1f}%")
    
    # Stop shadow and get analysis
    print("\n4. Stopping shadow deployment...")
    stop_result = pattern.stop_shadow()
    print(f"  Status: {stop_result['message']}")
    
    # Show detailed analysis
    print("\n5. Shadow deployment analysis:")
    analysis = stop_result['analysis']
    print(f"  Production version: {analysis['production_version']}")
    print(f"  Shadow version: {analysis['shadow_version']}")
    print(f"\n  Requests:")
    print(f"    Total: {analysis['requests']['total']}")
    print(f"    Success rate: {analysis['requests']['success_rate']:.1f}%")
    print(f"\n  Comparisons:")
    print(f"    Total: {analysis['comparisons']['total']}")
    print(f"    Matches: {analysis['comparisons']['matches']}")
    print(f"    Mismatches: {analysis['comparisons']['mismatches']}")
    print(f"    Match rate: {analysis['comparisons']['match_rate']:.1f}%")
    print(f"\n  Performance:")
    print(f"    Avg latency diff: {analysis['performance']['avg_latency_diff_ms']:.2f}ms")
    print(f"    Avg similarity: {analysis['performance']['avg_similarity_score']:.2f}")
    print(f"\n  Recommendation:")
    print(f"    {analysis['recommendation']}")
    
    # Statistics
    print("\n6. Deployment statistics:")
    stats = pattern.get_statistics()
    print(f"  Total deployments: {stats['total_deployments']}")
    print(f"  Total requests processed: {stats['total_requests_processed']}")
    
    print("\nāœ“ Comprehensive shadow deployment with traffic mirroring complete!")

Agentic Architecture Pattern: Shadow Deployment

Pattern 274: 274 Shadow Deployment – Line-by-Line Explanation

Overview

Pattern 274: Shadow Deployment

Purpose: Parallel production testing with traffic mirroring and result comparison

Code Structure Analysis

File Information

  • Pattern Number: 274
  • Pattern Name: 274 Shadow Deployment
  • File Path: patterns\deployment\274_shadow_deployment.py
  • Total Lines: 556

Detailed Code Explanation

Module Docstring

"""
Pattern 274: Shadow Deployment

Purpose: Parallel production testing with traffic mirroring and result comparison
"""

Explanation: This module docstring describes the purpose and key features of the 274 Shadow Deployment 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
import time
        import random
        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

Enumerations

ShadowMode (Line 15)

class ShadowMode(Enum):
    PASSIVE
    COMPARATIVE
    VALIDATION

Purpose: Shadow deployment mode.

Values: – PASSIVE: Represents passive – COMPARATIVE: Represents comparative – VALIDATION: Represents validation

ComparisonStatus (Line 22)

class ComparisonStatus(Enum):
    MATCH
    MISMATCH
    ERROR
    NOT_COMPARED

Purpose: Status of result comparison.

Values: – MATCH: Represents match – MISMATCH: Represents mismatch – ERROR: Error severity level – NOT_COMPARED: Represents not compared

Classes

ShadowMode (Line 15)

Description: Shadow deployment mode.

ComparisonStatus (Line 22)

Description: Status of result comparison.

Request (Line 31)

Description: Represents a single request.

Response (Line 39)

Description: Represents a response from a version.

Comparison (Line 51)

Description: Comparison between production and shadow responses.

ShadowMetrics (Line 63)

Description: Metrics for shadow deployment.

ShadowDeployment (Line 75)

Description: Implements shadow deployment pattern.

Features: – Parallel execution with production – Traffic mirroring to shadow version – Result comparison and analysis – Performance metrics collection – No impact on production traffic – Confidence building for new versions

Methods:

__init__ (Line 88)
def __init__(self, production_version, shadow_version, mode, traffic_mirror_percentage):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – production_version: Parameter for production_version – shadow_version: Parameter for shadow_version – mode: Parameter for mode – traffic_mirror_percentage: Parameter for traffic_mirror_percentage

start_shadow (Line 120)
def start_shadow(self, shadow_version, production_handler, shadow_handler):
    """Start shadow deployment.

Args:
    shadow_version: Version of shadow deployment
    production_handler: Function to handle production requests
    shadow_handler: Function to handle shadow requests

Returns:
    Status of shadow start"""

Purpose: Start shadow deployment.

Args: shadow_version: Version of shadow deployment production_handler: Function to handle production requests shadow_handler: Function to handle shadow requests

Returns: Status of shadow start

Parameters: – self: Instance reference – shadow_version: Parameter for shadow_version – production_handler: Parameter for production_handler – shadow_handler: Parameter for shadow_handler

process_request (Line 167)
def process_request(self, request_id, request_data):
    """Process request through both production and shadow.

Args:
    request_id: Unique request identifier
    request_data: Request data to process

Returns:
    Production response (shadow runs in parallel)"""

Purpose: Process request through both production and shadow.

Args: request_id: Unique request identifier request_data: Request data to process

Returns: Production response (shadow runs in parallel)

Parameters: – self: Instance reference – request_id: Parameter for request_id – request_data: Parameter for request_data

_compare_responses (Line 265)
def _compare_responses(self, prod_response, shadow_response):
    """Compare production and shadow responses."""

Purpose: Compare production and shadow responses.

Parameters: – self: Instance reference – prod_response: Parameter for prod_response – shadow_response: Parameter for shadow_response

_update_comparison_metrics (Line 309)
def _update_comparison_metrics(self, comparison):
    """Update metrics based on comparison."""

Purpose: Update metrics based on comparison.

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

stop_shadow (Line 336)
def stop_shadow(self):
    """Stop shadow deployment and return final analysis."""

Purpose: Stop shadow deployment and return final analysis.

Parameters: – self: Instance reference

get_analysis (Line 355)
def get_analysis(self):
    """Get detailed analysis of shadow deployment."""

Purpose: Get detailed analysis of shadow deployment.

Parameters: – self: Instance reference

_generate_recommendation (Line 387)
def _generate_recommendation(self, match_rate):
    """Generate deployment recommendation based on analysis."""

Purpose: Generate deployment recommendation based on analysis.

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

get_status (Line 398)
def get_status(self):
    """Get current shadow deployment status."""

Purpose: Get current shadow deployment status.

Parameters: – self: Instance reference

get_statistics (Line 415)
def get_statistics(self):
    """Get deployment statistics."""

Purpose: Get deployment statistics.

Parameters: – self: Instance reference

ShadowDeploymentPattern (Line 426)

Description: Coordinator for shadow deployments.

Manages shadow deployment lifecycle with traffic mirroring and analysis.

Methods:

__init__ (Line 433)
def __init__(self, production_version, mode, traffic_mirror_percentage):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – production_version: Parameter for production_version – mode: Parameter for mode – traffic_mirror_percentage: Parameter for traffic_mirror_percentage

start_shadow (Line 445)
def start_shadow(self, shadow_version, production_handler, shadow_handler):
    """Start shadow deployment."""

Purpose: Start shadow deployment.

Parameters: – self: Instance reference – shadow_version: Parameter for shadow_version – production_handler: Parameter for production_handler – shadow_handler: Parameter for shadow_handler

process_request (Line 454)
def process_request(self, request_id, request_data):
    """Process request through production and shadow."""

Purpose: Process request through production and shadow.

Parameters: – self: Instance reference – request_id: Parameter for request_id – request_data: Parameter for request_data

stop_shadow (Line 458)
def stop_shadow(self):
    """Stop shadow deployment."""

Purpose: Stop shadow deployment.

Parameters: – self: Instance reference

get_status (Line 462)
def get_status(self):
    """Get deployment status."""

Purpose: Get deployment status.

Parameters: – self: Instance reference

get_analysis (Line 466)
def get_analysis(self):
    """Get deployment analysis."""

Purpose: Get deployment analysis.

Parameters: – self: Instance reference

get_statistics (Line 470)
def get_statistics(self):
    """Get deployment statistics."""

Purpose: Get deployment statistics.

Parameters: – self: Instance reference

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 = ShadowMode(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 *

More posts