aishastr.live

Agentic Architecture Pattern: Elastic Agent

Written by

admin

in


    """
Pattern 163: Elastic Agent Pattern

Purpose: Dynamically scale resources both vertically and horizontally based on demand.
Combines both scaling strategies for optimal resource utilization.
"""

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


class ScalingDirection(Enum):
    """Direction of scaling."""
    SCALE_UP_VERTICAL = "scale_up_vertical"
    SCALE_DOWN_VERTICAL = "scale_down_vertical"
    SCALE_OUT_HORIZONTAL = "scale_out_horizontal"
    SCALE_IN_HORIZONTAL = "scale_in_horizontal"
    NO_CHANGE = "no_change"


@dataclass
class ResourceMetrics:
    """Current resource utilization metrics."""
    cpu_usage: float  # 0.0 to 1.0
    memory_usage: float  # 0.0 to 1.0
    throughput: float  # tasks per second
    response_time: float  # seconds
    queue_length: int
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ElasticPolicy:
    """Elasticity policy configuration."""
    cpu_scale_up_threshold: float = 0.8
    cpu_scale_down_threshold: float = 0.3
    queue_scale_out_threshold: int = 5
    queue_scale_in_threshold: int = 0
    cooldown_period: float = 60.0  # seconds
    min_instances: int = 1
    max_instances: int = 10


@dataclass
class ScalingDecision:
    """Record of scaling decision."""
    direction: ScalingDirection
    reason: str
    metrics: ResourceMetrics
    timestamp: datetime = field(default_factory=datetime.now)


class ElasticAgent:
    """Agent with elastic scaling capabilities."""
    
    def __init__(self, agent_id: str, tier: str = "standard"):
        self.agent_id = agent_id
        self.tier = tier
        self.llm = ChatOpenAI(model="gpt-4" if tier == "premium" else "gpt-3.5-turbo", 
                             temperature=0.7)
        self.tasks_processed = 0
        self.is_active = True
    
    def process_task(self, task: str) -> str:
        """Process a task."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are elastic agent {self.agent_id} (tier: {self.tier})"),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        self.tasks_processed += 1
        
        return str(response.content)


class ElasticAgentPattern:
    """
    Elastic Agent Pattern - Dynamic bidirectional scaling.
    
    Features:
    - Vertical scaling: Upgrade/downgrade agent tiers
    - Horizontal scaling: Add/remove agent instances
    - Policy-based decisions
    - Cooldown periods to prevent thrashing
    - Metrics tracking
    """
    
    def __init__(self, policy: Optional[ElasticPolicy] = None):
        self.policy = policy or ElasticPolicy()
        self.agents: List[ElasticAgent] = []
        self.task_queue: List[str] = []
        self.metrics_history: List[ResourceMetrics] = []
        self.scaling_decisions: List[ScalingDecision] = []
        self.last_scaling_time = datetime.now()
        
        # Initialize with minimum instances
        for i in range(self.policy.min_instances):
            self.agents.append(ElasticAgent(f"agent-{i}"))
    
    def collect_metrics(self) -> ResourceMetrics:
        """Collect current resource metrics."""
        active_agents = [a for a in self.agents if a.is_active]
        
        # Simulate metrics (in real system, would measure actual resources)
        metrics = ResourceMetrics(
            cpu_usage=len(self.task_queue) / (len(active_agents) * 10 + 1),
            memory_usage=len(active_agents) / self.policy.max_instances,
            throughput=sum(a.tasks_processed for a in active_agents) / max(1, len(self.metrics_history)),
            response_time=1.0 + (len(self.task_queue) * 0.1),
            queue_length=len(self.task_queue)
        )
        
        self.metrics_history.append(metrics)
        return metrics
    
    def make_scaling_decision(self, metrics: ResourceMetrics) -> ScalingDecision:
        """Decide on scaling action based on metrics."""
        # Check cooldown period
        time_since_last = (datetime.now() - self.last_scaling_time).total_seconds()
        if time_since_last  self.policy.queue_scale_out_threshold:
            if len(self.agents)  self.policy.cpu_scale_up_threshold:
            standard_agents = [a for a in self.agents if a.tier == "standard"]
            if standard_agents:
                return ScalingDecision(
                    ScalingDirection.SCALE_UP_VERTICAL,
                    f"CPU usage {metrics.cpu_usage:.2f} exceeds threshold",
                    metrics
                )
        
        # Check for horizontal scale in (remove instances)
        if metrics.queue_length  self.policy.min_instances:
                return ScalingDecision(
                    ScalingDirection.SCALE_IN_HORIZONTAL,
                    f"Queue empty and excess capacity",
                    metrics
                )
        
        # Check for vertical scale down (downgrade tiers)
        if metrics.cpu_usage  self.policy.min_instances:
                removed = self.agents.pop()
                removed.is_active = False
                print(f"➖ Scaled in: Removed {removed.agent_id} (Total: {len(self.agents)})")
        
        elif decision.direction == ScalingDirection.SCALE_UP_VERTICAL:
            # Upgrade an agent to premium tier
            for agent in self.agents:
                if agent.tier == "standard":
                    agent.tier = "premium"
                    agent.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
                    print(f"⬆️ Scaled up: Upgraded {agent.agent_id} to premium")
                    break
        
        elif decision.direction == ScalingDirection.SCALE_DOWN_VERTICAL:
            # Downgrade an agent to standard tier
            for agent in self.agents:
                if agent.tier == "premium":
                    agent.tier = "standard"
                    agent.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
                    print(f"⬇️ Scaled down: Downgraded {agent.agent_id} to standard")
                    break
    
    def add_tasks(self, tasks: List[str]):
        """Add tasks to queue."""
        self.task_queue.extend(tasks)
    
    def process_tasks(self):
        """Process tasks with current agent pool."""
        while self.task_queue and any(a.is_active for a in self.agents):
            # Get available agent
            for agent in self.agents:
                if agent.is_active and self.task_queue:
                    task = self.task_queue.pop(0)
                    agent.process_task(task)
            
            # Check if we should scale
            metrics = self.collect_metrics()
            decision = self.make_scaling_decision(metrics)
            self.execute_scaling(decision)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get elasticity statistics."""
        premium_count = sum(1 for a in self.agents if a.tier == "premium")
        standard_count = len(self.agents) - premium_count
        
        return {
            "total_agents": len(self.agents),
            "premium_agents": premium_count,
            "standard_agents": standard_count,
            "tasks_in_queue": len(self.task_queue),
            "total_scaling_actions": len(self.scaling_decisions),
            "metrics_collected": len(self.metrics_history)
        }


# Example usage
if __name__ == "__main__":
    print("=== Elastic Agent Pattern Demo ===\n")
    
    # Create elastic pattern with custom policy
    policy = ElasticPolicy(
        cpu_scale_up_threshold=0.7,
        queue_scale_out_threshold=3,
        min_instances=2,
        max_instances=6,
        cooldown_period=5.0  # Short for demo
    )
    
    pattern = ElasticAgentPattern(policy)
    
    print(f"Initial agents: {len(pattern.agents)}\n")
    
    # Simulate varying workload
    print("Phase 1: Light load (2 tasks)")
    pattern.add_tasks(["Task A", "Task B"])
    pattern.process_tasks()
    
    print(f"\nPhase 2: Heavy load (10 tasks)")
    pattern.add_tasks([f"Task-{i}" for i in range(10)])
    time.sleep(6)  # Wait for cooldown
    pattern.process_tasks()
    
    print(f"\nPhase 3: Processing...")
    time.sleep(6)
    pattern.process_tasks()
    
    # Statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")
    
    # Scaling history
    print(f"\n=== Scaling History ===")
    for decision in pattern.scaling_decisions:
        print(f"{decision.direction.value}: {decision.reason}")

Agentic Architecture Pattern: Elastic Agent

Pattern 163: 163 Elastic Agent – Line-by-Line Explanation

Overview

Pattern 163: Elastic Agent Pattern

Purpose: Dynamically scale resources both vertically and horizontally based on demand. Combines both scaling strategies for optimal resource utilization.

Code Structure Analysis

File Information

  • Pattern Number: 163
  • Pattern Name: 163 Elastic Agent
  • File Path: patterns\scaling\163_elastic_agent.py
  • Total Lines: 283

Detailed Code Explanation

Module Docstring

"""
Pattern 163: Elastic Agent Pattern

Purpose: Dynamically scale resources both vertically and horizontally based on demand.
Combines both scaling strategies for optimal resource utilization.
"""

Explanation: This module docstring describes the purpose and key features of the 163 Elastic Agent pattern.

Imports Section

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

ScalingDirection (Line 17)

class ScalingDirection(Enum):
    SCALE_UP_VERTICAL
    SCALE_DOWN_VERTICAL
    SCALE_OUT_HORIZONTAL
    SCALE_IN_HORIZONTAL
    NO_CHANGE

Purpose: Direction of scaling.

Values: – SCALE_UP_VERTICAL: Represents scale up vertical – SCALE_DOWN_VERTICAL: Represents scale down vertical – SCALE_OUT_HORIZONTAL: Represents scale out horizontal – SCALE_IN_HORIZONTAL: Represents scale in horizontal – NO_CHANGE: Represents no change

Classes

ScalingDirection (Line 17)

Description: Direction of scaling.

ResourceMetrics (Line 27)

Description: Current resource utilization metrics.

ElasticPolicy (Line 38)

Description: Elasticity policy configuration.

ScalingDecision (Line 50)

Description: Record of scaling decision.

ElasticAgent (Line 58)

Description: Agent with elastic scaling capabilities.

Methods:

__init__ (Line 61)
def __init__(self, agent_id, tier):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – tier: Parameter for tier

process_task (Line 69)
def process_task(self, task):
    """Process a task."""

Purpose: Process a task.

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

ElasticAgentPattern (Line 83)

Description: Elastic Agent Pattern – Dynamic bidirectional scaling.

Features: – Vertical scaling: Upgrade/downgrade agent tiers – Horizontal scaling: Add/remove agent instances – Policy-based decisions – Cooldown periods to prevent thrashing – Metrics tracking

Methods:

__init__ (Line 95)
def __init__(self, policy):
    """No method docstring."""

Purpose: No method docstring.

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

collect_metrics (Line 107)
def collect_metrics(self):
    """Collect current resource metrics."""

Purpose: Collect current resource metrics.

Parameters: – self: Instance reference

make_scaling_decision (Line 123)
def make_scaling_decision(self, metrics):
    """Decide on scaling action based on metrics."""

Purpose: Decide on scaling action based on metrics.

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

execute_scaling (Line 170)
def execute_scaling(self, decision):
    """Execute the scaling decision."""

Purpose: Execute the scaling decision.

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

add_tasks (Line 209)
def add_tasks(self, tasks):
    """Add tasks to queue."""

Purpose: Add tasks to queue.

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

process_tasks (Line 213)
def process_tasks(self):
    """Process tasks with current agent pool."""

Purpose: Process tasks with current agent pool.

Parameters: – self: Instance reference

get_statistics (Line 227)
def get_statistics(self):
    """Get elasticity statistics."""

Purpose: Get elasticity 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 = ScalingDirection(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