aishastr.live

Agentic Architecture Pattern: Vertical Scaling

Written by

admin

in


    """
Pattern 161: Vertical Scaling Pattern

Purpose: Scale agent capabilities by upgrading to more powerful models/resources
within a single instance (CPU, memory, model size).
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class ResourceTier(Enum):
    """Resource tiers for vertical scaling."""
    BASIC = "gpt-3.5-turbo"
    STANDARD = "gpt-4"
    PREMIUM = "gpt-4-turbo-preview"
    ENTERPRISE = "gpt-4-32k"


@dataclass
class ScalingMetrics:
    """Metrics for scaling decisions."""
    load: float  # 0.0 to 1.0
    response_time: float  # seconds
    success_rate: float  # 0.0 to 1.0
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ScalingEvent:
    """Record of scaling event."""
    from_tier: ResourceTier
    to_tier: ResourceTier
    reason: str
    metrics: ScalingMetrics
    timestamp: datetime = field(default_factory=datetime.now)


class VerticalScalingAgent:
    """Agent with vertical scaling capabilities."""
    
    def __init__(self, name: str, initial_tier: ResourceTier = ResourceTier.BASIC):
        self.name = name
        self.current_tier = initial_tier
        self.llm = ChatOpenAI(model=initial_tier.value, temperature=0.7)
        self.metrics_history: List[ScalingMetrics] = []
        self.scaling_events: List[ScalingEvent] = []
        
    def scale_up(self) -> bool:
        """Scale up to next tier."""
        tiers = [ResourceTier.BASIC, ResourceTier.STANDARD, 
                ResourceTier.PREMIUM, ResourceTier.ENTERPRISE]
        current_idx = tiers.index(self.current_tier)
        
        if current_idx  bool:
        """Scale down to previous tier."""
        tiers = [ResourceTier.BASIC, ResourceTier.STANDARD, 
                ResourceTier.PREMIUM, ResourceTier.ENTERPRISE]
        current_idx = tiers.index(self.current_tier)
        
        if current_idx > 0:
            old_tier = self.current_tier
            self.current_tier = tiers[current_idx - 1]
            self.llm = ChatOpenAI(model=self.current_tier.value, temperature=0.7)
            
            event = ScalingEvent(
                from_tier=old_tier,
                to_tier=self.current_tier,
                reason="Manual scale down",
                metrics=self.metrics_history[-1] if self.metrics_history else ScalingMetrics(0, 0, 0)
            )
            self.scaling_events.append(event)
            return True
        return False
    
    def process(self, task: str) -> Dict[str, Any]:
        """Process task with current tier."""
        start_time = datetime.now()
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} running on {self.current_tier.value}"),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        
        end_time = datetime.now()
        response_time = (end_time - start_time).total_seconds()
        
        # Record metrics
        metrics = ScalingMetrics(
            load=0.5,  # Simulated
            response_time=response_time,
            success_rate=1.0
        )
        self.metrics_history.append(metrics)
        
        return {
            "agent": self.name,
            "tier": self.current_tier.name,
            "result": response.content,
            "response_time": response_time
        }


class VerticalScalingPattern:
    """
    Vertical Scaling Pattern - Scale by upgrading resources/capabilities.
    
    Auto-scales based on load, response time, and success rate thresholds.
    """
    
    def __init__(self):
        self.agents: Dict[str, VerticalScalingAgent] = {}
        self.scale_up_threshold = 0.8  # Load threshold to scale up
        self.scale_down_threshold = 0.3  # Load threshold to scale down
        self.response_time_threshold = 5.0  # seconds
        
    def register_agent(self, agent: VerticalScalingAgent):
        """Register an agent for scaling management."""
        self.agents[agent.name] = agent
    
    def check_and_scale(self, agent_name: str) -> Optional[str]:
        """Check metrics and auto-scale if needed."""
        agent = self.agents.get(agent_name)
        if not agent or not agent.metrics_history:
            return None
        
        recent_metrics = agent.metrics_history[-5:]  # Last 5 metrics
        avg_load = sum(m.load for m in recent_metrics) / len(recent_metrics)
        avg_response_time = sum(m.response_time for m in recent_metrics) / len(recent_metrics)
        avg_success_rate = sum(m.success_rate for m in recent_metrics) / len(recent_metrics)
        
        # Scale up if overloaded or slow
        if avg_load > self.scale_up_threshold or avg_response_time > self.response_time_threshold:
            if agent.scale_up():
                return f"Scaled up {agent_name} to {agent.current_tier.name}"
        
        # Scale down if underutilized
        elif avg_load  0.95:
            if agent.scale_down():
                return f"Scaled down {agent_name} to {agent.current_tier.name}"
        
        return None
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get scaling statistics."""
        stats = {}
        for name, agent in self.agents.items():
            stats[name] = {
                "current_tier": agent.current_tier.name,
                "scaling_events": len(agent.scaling_events),
                "metrics_collected": len(agent.metrics_history),
                "recent_response_time": agent.metrics_history[-1].response_time if agent.metrics_history else 0
            }
        return stats


# Example usage
if __name__ == "__main__":
    print("=== Vertical Scaling Pattern Demo ===\n")
    
    # Create pattern
    pattern = VerticalScalingPattern()
    
    # Create agent starting at BASIC tier
    agent = VerticalScalingAgent("TaskAgent", ResourceTier.BASIC)
    pattern.register_agent(agent)
    
    print(f"Initial tier: {agent.current_tier.name}\n")
    
    # Process tasks
    tasks = [
        "Summarize this text",
        "Analyze complex data patterns",
        "Generate creative content"
    ]
    
    for i, task in enumerate(tasks, 1):
        print(f"Task {i}: {task}")
        result = agent.process(task)
        print(f"  Tier: {result['tier']}")
        print(f"  Response time: {result['response_time']:.2f}s")
        
        # Simulate increasing load
        if agent.metrics_history:
            agent.metrics_history[-1].load = 0.5 + (i * 0.2)
        
        # Check for auto-scaling
        scale_action = pattern.check_and_scale("TaskAgent")
        if scale_action:
            print(f"  ⚡ {scale_action}")
        print()
    
    # Show statistics
    print("\n=== Scaling Statistics ===")
    stats = pattern.get_statistics()
    for agent_name, agent_stats in stats.items():
        print(f"\n{agent_name}:")
        for key, value in agent_stats.items():
            print(f"  {key}: {value}")
    
    # Show scaling events
    print(f"\n=== Scaling Events ===")
    for event in agent.scaling_events:
        print(f"{event.from_tier.name} → {event.to_tier.name}: {event.reason}")

Agentic Architecture Pattern: Vertical Scaling

Pattern 161: 161 Vertical Scaling – Line-by-Line Explanation

Overview

Pattern 161: Vertical Scaling Pattern

Purpose: Scale agent capabilities by upgrading to more powerful models/resources within a single instance (CPU, memory, model size).

Code Structure Analysis

File Information

  • Pattern Number: 161
  • Pattern Name: 161 Vertical Scaling
  • File Path: patterns\scaling\161_vertical_scaling.py
  • Total Lines: 227

Detailed Code Explanation

Module Docstring

"""
Pattern 161: Vertical Scaling Pattern

Purpose: Scale agent capabilities by upgrading to more powerful models/resources
within a single instance (CPU, memory, model size).
"""

Explanation: This module docstring describes the purpose and key features of the 161 Vertical Scaling pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
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: 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

ResourceTier (Line 16)

class ResourceTier(Enum):
    BASIC
    STANDARD
    PREMIUM
    ENTERPRISE

Purpose: Resource tiers for vertical scaling.

Values: – BASIC: Represents basic – STANDARD: Represents standard – PREMIUM: Represents premium – ENTERPRISE: Represents enterprise

Classes

ResourceTier (Line 16)

Description: Resource tiers for vertical scaling.

ScalingMetrics (Line 25)

Description: Metrics for scaling decisions.

ScalingEvent (Line 34)

Description: Record of scaling event.

VerticalScalingAgent (Line 43)

Description: Agent with vertical scaling capabilities.

Methods:

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

Purpose: No method docstring.

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

scale_up (Line 53)
def scale_up(self):
    """Scale up to next tier."""

Purpose: Scale up to next tier.

Parameters: – self: Instance reference

scale_down (Line 74)
def scale_down(self):
    """Scale down to previous tier."""

Purpose: Scale down to previous tier.

Parameters: – self: Instance reference

process (Line 95)
def process(self, task):
    """Process task with current tier."""

Purpose: Process task with current tier.

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

VerticalScalingPattern (Line 126)

Description: Vertical Scaling Pattern – Scale by upgrading resources/capabilities.

Auto-scales based on load, response time, and success rate thresholds.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 139)
def register_agent(self, agent):
    """Register an agent for scaling management."""

Purpose: Register an agent for scaling management.

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

check_and_scale (Line 143)
def check_and_scale(self, agent_name):
    """Check metrics and auto-scale if needed."""

Purpose: Check metrics and auto-scale if needed.

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

get_statistics (Line 166)
def get_statistics(self):
    """Get scaling statistics."""

Purpose: Get scaling 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 = ResourceTier(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