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