aishastr.live

Agentic Architecture Pattern: Adaptive Agent

Written by

admin

in


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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts