"""
Pattern 59: Adaptive Agent
Purpose: Dynamically adapt strategies and behaviors based on environmental feedback.
Monitor performance, detect changes, and adjust strategies in real-time.
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from collections import deque
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class AdaptationTrigger(Enum):
"""Triggers for adaptation."""
PERFORMANCE_DROP = "performance_drop" # Performance below threshold
ENVIRONMENT_CHANGE = "environment_change" # Detected environment shift
TIME_BASED = "time_based" # Periodic adaptation
USER_FEEDBACK = "user_feedback" # Based on user input
class StrategyType(Enum):
"""Types of strategies available."""
CONSERVATIVE = "conservative" # Low risk, stable
AGGRESSIVE = "aggressive" # High risk, high reward
BALANCED = "balanced" # Middle ground
EXPLORATORY = "exploratory" # Exploration focused
EXPLOITATIVE = "exploitative" # Exploitation focused
@dataclass
class EnvironmentState:
"""Current state of the environment."""
difficulty: float # 0.0 to 1.0
volatility: float # 0.0 to 1.0 (how much it changes)
resource_availability: float # 0.0 to 1.0
competition_level: float # 0.0 to 1.0
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class PerformanceMetrics:
"""Performance metrics for adaptation decisions."""
success_rate: float
avg_response_time: float
error_rate: float
efficiency: float
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class AdaptationEvent:
"""Record of an adaptation event."""
trigger: AdaptationTrigger
old_strategy: StrategyType
new_strategy: StrategyType
reason: str
performance_before: float
performance_after: Optional[float] = None
timestamp: datetime = field(default_factory=datetime.now)
class AdaptiveAgent:
"""
Adaptive Agent - Dynamically adapt to changing environments.
Monitors performance, detects environmental changes, and adjusts
strategies in real-time to maintain optimal performance.
"""
def __init__(self,
name: str,
performance_threshold: float = 0.6,
adaptation_cooldown: int = 5,
window_size: int = 20):
self.name = name
self.performance_threshold = performance_threshold
self.adaptation_cooldown = adaptation_cooldown # Min actions between adaptations
self.window_size = window_size
self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
# Current state
self.current_strategy = StrategyType.BALANCED
self.environment = EnvironmentState(
difficulty=0.5,
volatility=0.3,
resource_availability=0.8,
competition_level=0.5
)
# Performance tracking
self.performance_history: deque = deque(maxlen=window_size)
self.recent_actions: deque = deque(maxlen=window_size)
self.actions_since_adaptation = 0
# Adaptation history
self.adaptation_events: List[AdaptationEvent] = []
self.strategy_performance: Dict[StrategyType, List[float]] = {
strategy: [] for strategy in StrategyType
}
# Statistics
self.statistics: Dict[str, Any] = {
"total_actions": 0,
"total_adaptations": 0,
"avg_performance": 0.0,
"best_strategy": None
}
def act(self, task: str, environment: Optional[EnvironmentState] = None) -> Dict[str, Any]:
"""Perform action with current strategy."""
if environment:
self.environment = environment
# Strategy-specific behavior
strategy_params = self._get_strategy_params()
# Simulate action performance (in practice, execute real task)
performance = self._simulate_action(task, strategy_params)
# Record action
self.recent_actions.append({
"task": task,
"strategy": self.current_strategy,
"performance": performance,
"timestamp": datetime.now()
})
self.performance_history.append(performance)
self.strategy_performance[self.current_strategy].append(performance)
self.actions_since_adaptation += 1
self.statistics["total_actions"] += 1
# Check if adaptation needed
adaptation_needed, trigger, reason = self._check_adaptation_needed()
if adaptation_needed:
self._adapt(trigger, reason)
return {
"task": task,
"strategy": self.current_strategy.value,
"performance": performance,
"adapted": adaptation_needed
}
def _get_strategy_params(self) -> Dict[str, float]:
"""Get parameters for current strategy."""
if self.current_strategy == StrategyType.CONSERVATIVE:
return {"risk": 0.2, "exploration": 0.1, "speed": 0.5}
elif self.current_strategy == StrategyType.AGGRESSIVE:
return {"risk": 0.9, "exploration": 0.3, "speed": 0.9}
elif self.current_strategy == StrategyType.EXPLORATORY:
return {"risk": 0.5, "exploration": 0.9, "speed": 0.6}
elif self.current_strategy == StrategyType.EXPLOITATIVE:
return {"risk": 0.3, "exploration": 0.1, "speed": 0.8}
else: # BALANCED
return {"risk": 0.5, "exploration": 0.5, "speed": 0.7}
def _simulate_action(self, task: str, params: Dict[str, float]) -> float:
"""Simulate action performance."""
# Base performance from strategy params
base_perf = (params["risk"] * 0.3 +
params["exploration"] * 0.3 +
params["speed"] * 0.4)
# Adjust for environment
env_factor = (1.0 - self.environment.difficulty * 0.5 +
self.environment.resource_availability * 0.3)
# Add some noise
import random
noise = random.uniform(-0.1, 0.1)
performance = base_perf * env_factor + noise
return max(0.0, min(1.0, performance))
def _check_adaptation_needed(self) -> Tuple[bool, AdaptationTrigger, str]:
"""Check if adaptation is needed."""
# Must wait for cooldown
if self.actions_since_adaptation < self.adaptation_cooldown:
return False, AdaptationTrigger.PERFORMANCE_DROP, ""
# Need minimum history
if len(self.performance_history) < 5:
return False, AdaptationTrigger.PERFORMANCE_DROP, ""
# Check performance drop
recent_perf = sum(list(self.performance_history)[-5:]) / 5
if recent_perf = 10:
older_perf = sum(list(self.performance_history)[-10:-5]) / 5
if abs(recent_perf - older_perf) > 0.15:
return True, AdaptationTrigger.ENVIRONMENT_CHANGE, \
f"Performance shift detected: {older_perf:.3f} -> {recent_perf:.3f}"
# Time-based adaptation
if self.actions_since_adaptation >= self.window_size:
return True, AdaptationTrigger.TIME_BASED, \
f"Periodic adaptation after {self.actions_since_adaptation} actions"
return False, AdaptationTrigger.PERFORMANCE_DROP, ""
def _adapt(self, trigger: AdaptationTrigger, reason: str):
"""Adapt strategy based on trigger."""
old_strategy = self.current_strategy
old_performance = sum(self.performance_history) / len(self.performance_history)
# Select new strategy
new_strategy = self._select_new_strategy(trigger)
# Apply adaptation
self.current_strategy = new_strategy
self.actions_since_adaptation = 0
# Record event
event = AdaptationEvent(
trigger=trigger,
old_strategy=old_strategy,
new_strategy=new_strategy,
reason=reason,
performance_before=old_performance
)
self.adaptation_events.append(event)
self.statistics["total_adaptations"] += 1
print(f" {self.name}: Adapted {old_strategy.value} -> {new_strategy.value}")
print(f" Trigger: {trigger.value}")
print(f" Reason: {reason}")
def _select_new_strategy(self, trigger: AdaptationTrigger) -> StrategyType:
"""Select new strategy based on situation."""
# Get performance stats for each strategy
strategy_scores = {}
for strategy, perfs in self.strategy_performance.items():
if perfs:
strategy_scores[strategy] = sum(perfs) / len(perfs)
else:
strategy_scores[strategy] = 0.5 # Default score
# Remove current strategy
available_strategies = [s for s in StrategyType if s != self.current_strategy]
if trigger == AdaptationTrigger.PERFORMANCE_DROP:
# Switch to best performing alternative
best_strategy = max(available_strategies,
key=lambda s: strategy_scores.get(s, 0.5))
return best_strategy
elif trigger == AdaptationTrigger.ENVIRONMENT_CHANGE:
# Try exploratory to learn about new environment
if self.current_strategy != StrategyType.EXPLORATORY:
return StrategyType.EXPLORATORY
else:
return StrategyType.BALANCED
elif trigger == AdaptationTrigger.TIME_BASED:
# Rotate to untested or underperforming strategies
least_tested = min(available_strategies,
key=lambda s: len(self.strategy_performance[s]))
return least_tested
else: # USER_FEEDBACK
return StrategyType.BALANCED
def update_environment(self, environment: EnvironmentState):
"""Update environment state."""
self.environment = environment
print(f" {self.name}: Environment updated")
print(f" Difficulty: {environment.difficulty:.2f}")
print(f" Volatility: {environment.volatility:.2f}")
def get_adaptation_summary(self) -> Dict[str, Any]:
"""Get summary of adaptations."""
if not self.adaptation_events:
return {"total_adaptations": 0}
trigger_counts = {}
for event in self.adaptation_events:
trigger_counts[event.trigger.value] = \
trigger_counts.get(event.trigger.value, 0) + 1
strategy_usage = {}
for action in self.recent_actions:
strategy = action["strategy"].value
strategy_usage[strategy] = strategy_usage.get(strategy, 0) + 1
return {
"total_adaptations": len(self.adaptation_events),
"trigger_counts": trigger_counts,
"current_strategy": self.current_strategy.value,
"strategy_usage": strategy_usage,
"recent_performance": sum(self.performance_history) / len(self.performance_history) \
if self.performance_history else 0.0
}
def get_statistics(self) -> Dict[str, Any]:
"""Get comprehensive statistics."""
# Calculate best strategy
best_strategy = None
best_avg = 0.0
for strategy, perfs in self.strategy_performance.items():
if perfs:
avg = sum(perfs) / len(perfs)
if avg > best_avg:
best_avg = avg
best_strategy = strategy
self.statistics["best_strategy"] = best_strategy.value if best_strategy else None
self.statistics["avg_performance"] = \
sum(self.performance_history) / len(self.performance_history) \
if self.performance_history else 0.0
return self.statistics
class AdaptivePattern:
"""
Adaptive Pattern - Coordinate multiple adaptive agents.
Manages agents that dynamically adjust to changing conditions.
"""
def __init__(self):
self.agents: Dict[str, AdaptiveAgent] = {}
def register_agent(self, agent: AdaptiveAgent):
"""Register adaptive agent."""
self.agents[agent.name] = agent
print(f" Registered agent: {agent.name}")
def run(self, tasks: List[str], environment_changes: List[EnvironmentState]) -> Dict[str, Any]:
"""Run adaptive agents through tasks."""
print("=== Adaptive Agent Pattern Demo ===\n")
print("1. Running agents through tasks...")
results = []
for i, task in enumerate(tasks):
# Update environment if applicable
if i < len(environment_changes):
for agent in self.agents.values():
agent.update_environment(environment_changes[i])
# Execute task with each agent
for agent in self.agents.values():
result = agent.act(task)
results.append(result)
# Get summaries
print("\n2. Adaptation summaries:")
summaries = {}
for name, agent in self.agents.items():
summary = agent.get_adaptation_summary()
summaries[name] = summary
print(f"\n {name}:")
print(f" Total adaptations: {summary['total_adaptations']}")
print(f" Current strategy: {summary['current_strategy']}")
print(f" Recent performance: {summary['recent_performance']:.3f}")
return {
"pattern": "Adaptive",
"agents": len(self.agents),
"tasks": len(tasks),
"total_actions": sum(agent.statistics["total_actions"]
for agent in self.agents.values()),
"summaries": summaries
}
# Example usage
if __name__ == "__main__":
print("=== Adaptive Agent Pattern Demo ===\n")
# Create pattern
pattern = AdaptivePattern()
# Create agents
print("1. Creating adaptive agents...")
agent1 = AdaptiveAgent(name="Agent1", performance_threshold=0.55, adaptation_cooldown=3)
agent2 = AdaptiveAgent(name="Agent2", performance_threshold=0.60, adaptation_cooldown=5)
pattern.register_agent(agent1)
pattern.register_agent(agent2)
# Define tasks
tasks = [
"Process easy task",
"Handle medium task",
"Solve complex task",
"Quick response needed",
"Detailed analysis",
"Creative solution",
"Optimize performance",
"Handle uncertainty",
"Resource management",
"Competitive scenario",
"Adapt to change",
"Maintain stability",
"Explore options",
"Exploit knowledge",
"Balance tradeoffs"
]
# Define environment changes
environments = [
EnvironmentState(difficulty=0.3, volatility=0.2, resource_availability=0.9, competition_level=0.3),
EnvironmentState(difficulty=0.5, volatility=0.4, resource_availability=0.7, competition_level=0.5),
EnvironmentState(difficulty=0.7, volatility=0.6, resource_availability=0.5, competition_level=0.7),
EnvironmentState(difficulty=0.4, volatility=0.3, resource_availability=0.8, competition_level=0.4),
]
# Run pattern
print("\n2. Running tasks with environmental changes...")
summary = pattern.run(tasks, environments)
print(f"\n3. Overall Results:")
print(f" Pattern: {summary['pattern']}")
print(f" Agents: {summary['agents']}")
print(f" Tasks: {summary['tasks']}")
print(f" Total actions: {summary['total_actions']}")
print(f"\n4. Agent Statistics:")
for name, agent in pattern.agents.items():
stats = agent.get_statistics()
print(f"\n {name}:")
for key, value in stats.items():
if isinstance(value, float):
print(f" {key}: {value:.4f}")
else:
print(f" {key}: {value}")
print("\nā Agents successfully adapted to changing environments!")
Agentic Architecture Pattern: Adaptive Agent
Pattern 59: 59 Adaptive Agent – Line-by-Line Explanation
Overview
Pattern 59: Adaptive Agent
Purpose: Dynamically adapt strategies and behaviors based on environmental feedback. Monitor performance, detect changes, and adjust strategies in real-time.
Code Structure Analysis
File Information
- Pattern Number: 59
- Pattern Name: 59 Adaptive Agent
- File Path:
patterns\learning_adaptation\59_adaptive_agent.py - Total Lines: 442
Detailed Code Explanation
Module Docstring
"""
Pattern 59: Adaptive Agent
Purpose: Dynamically adapt strategies and behaviors based on environmental feedback.
Monitor performance, detect changes, and adjust strategies in real-time.
"""
Explanation: This module docstring describes the purpose and key features of the 59 Adaptive Agent pattern.
Imports Section
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from collections import deque
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import random # 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, timedelta: 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
AdaptationTrigger (Line 17)
class AdaptationTrigger(Enum):
PERFORMANCE_DROP
ENVIRONMENT_CHANGE
TIME_BASED
USER_FEEDBACK
Purpose: Triggers for adaptation.
Values:
– PERFORMANCE_DROP: Represents performance drop
– ENVIRONMENT_CHANGE: Represents environment change
– TIME_BASED: Represents time based
– USER_FEEDBACK: Represents user feedback
StrategyType (Line 25)
class StrategyType(Enum):
CONSERVATIVE
AGGRESSIVE
BALANCED
EXPLORATORY
EXPLOITATIVE
Purpose: Types of strategies available.
Values:
– CONSERVATIVE: Represents conservative
– AGGRESSIVE: Represents aggressive
– BALANCED: Represents balanced
– EXPLORATORY: Represents exploratory
– EXPLOITATIVE: Represents exploitative
Classes
AdaptationTrigger (Line 17)
Description: Triggers for adaptation.
StrategyType (Line 25)
Description: Types of strategies available.
EnvironmentState (Line 35)
Description: Current state of the environment.
PerformanceMetrics (Line 45)
Description: Performance metrics for adaptation decisions.
AdaptationEvent (Line 55)
Description: Record of an adaptation event.
AdaptiveAgent (Line 66)
Description: Adaptive Agent – Dynamically adapt to changing environments.
Monitors performance, detects environmental changes, and adjusts strategies in real-time to maintain optimal performance.
Methods:
__init__ (Line 74)
def __init__(self, name, performance_threshold, adaptation_cooldown, window_size):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– performance_threshold: Parameter for performance_threshold
– adaptation_cooldown: Parameter for adaptation_cooldown
– window_size: Parameter for window_size
act (Line 113)
def act(self, task, environment):
"""Perform action with current strategy."""
Purpose: Perform action with current strategy.
Parameters:
– self: Instance reference
– task: Parameter for task
– environment: Parameter for environment
_get_strategy_params (Line 150)
def _get_strategy_params(self):
"""Get parameters for current strategy."""
Purpose: Get parameters for current strategy.
Parameters:
– self: Instance reference
_simulate_action (Line 163)
def _simulate_action(self, task, params):
"""Simulate action performance."""
Purpose: Simulate action performance.
Parameters:
– self: Instance reference
– task: Parameter for task
– params: Parameter for params
_check_adaptation_needed (Line 181)
def _check_adaptation_needed(self):
"""Check if adaptation is needed."""
Purpose: Check if adaptation is needed.
Parameters:
– self: Instance reference
_adapt (Line 211)
def _adapt(self, trigger, reason):
"""Adapt strategy based on trigger."""
Purpose: Adapt strategy based on trigger.
Parameters:
– self: Instance reference
– trigger: Parameter for trigger
– reason: Parameter for reason
_select_new_strategy (Line 239)
def _select_new_strategy(self, trigger):
"""Select new strategy based on situation."""
Purpose: Select new strategy based on situation.
Parameters:
– self: Instance reference
– trigger: Parameter for trigger
update_environment (Line 274)
def update_environment(self, environment):
"""Update environment state."""
Purpose: Update environment state.
Parameters:
– self: Instance reference
– environment: Parameter for environment
get_adaptation_summary (Line 281)
def get_adaptation_summary(self):
"""Get summary of adaptations."""
Purpose: Get summary of adaptations.
Parameters:
– self: Instance reference
get_statistics (Line 305)
def get_statistics(self):
"""Get comprehensive statistics."""
Purpose: Get comprehensive statistics.
Parameters:
– self: Instance reference
AdaptivePattern (Line 325)
Description: Adaptive Pattern – Coordinate multiple adaptive agents.
Manages agents that dynamically adjust to changing conditions.
Methods:
__init__ (Line 332)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
register_agent (Line 335)
def register_agent(self, agent):
"""Register adaptive agent."""
Purpose: Register adaptive agent.
Parameters:
– self: Instance reference
– agent: Parameter for agent
run (Line 340)
def run(self, tasks, environment_changes):
"""Run adaptive agents through tasks."""
Purpose: Run adaptive agents through tasks.
Parameters:
– self: Instance reference
– tasks: Parameter for tasks
– environment_changes: Parameter for environment_changes
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 = AdaptationTrigger(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