aishastr.live

Agentic Architecture Pattern: Evolutionary Agent

Written by

admin

in


    """
Pattern 58: Evolutionary Agent

Purpose: Use genetic algorithms to evolve agent strategies and behaviors.
Selection, crossover, and mutation to optimize agent parameters over generations.
"""

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


class SelectionStrategy(Enum):
    """Selection strategies for evolution."""
    TOURNAMENT = "tournament"  # Tournament selection
    ROULETTE = "roulette"  # Roulette wheel (fitness-proportionate)
    RANK = "rank"  # Rank-based selection
    ELITISM = "elitism"  # Keep best individuals


@dataclass
class Gene:
    """A gene representing part of an agent's strategy."""
    name: str
    value: float
    min_value: float = 0.0
    max_value: float = 1.0
    mutation_rate: float = 0.1


@dataclass
class Genome:
    """A genome representing an agent's complete strategy."""
    id: str
    genes: Dict[str, Gene]
    fitness: float = 0.0
    generation: int = 0
    parent_ids: List[str] = field(default_factory=list)


@dataclass
class GenerationStats:
    """Statistics for a generation."""
    generation: int
    population_size: int
    best_fitness: float
    avg_fitness: float
    worst_fitness: float
    diversity: float
    timestamp: datetime = field(default_factory=datetime.now)


class EvolutionaryAgent:
    """
    Evolutionary Agent - Evolve strategies using genetic algorithms.
    
    Uses selection, crossover, and mutation to optimize agent behaviors
    over multiple generations. LLM assists in evaluating fitness.
    """
    
    def __init__(self,
                 name: str,
                 gene_names: List[str],
                 population_size: int = 20,
                 selection_strategy: SelectionStrategy = SelectionStrategy.TOURNAMENT,
                 mutation_rate: float = 0.1,
                 crossover_rate: float = 0.7,
                 elitism_count: int = 2):
        self.name = name
        self.gene_names = gene_names
        self.population_size = population_size
        self.selection_strategy = selection_strategy
        self.mutation_rate = mutation_rate
        self.crossover_rate = crossover_rate
        self.elitism_count = elitism_count
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Population
        self.population: List[Genome] = []
        self.current_generation: int = 0
        self.best_genome: Optional[Genome] = None
        
        # History
        self.generation_stats: List[GenerationStats] = []
        self.evolution_history: List[Genome] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_generations": 0,
            "total_evaluations": 0,
            "best_fitness_ever": 0.0,
            "avg_fitness_improvement": 0.0
        }
    
    def initialize_population(self):
        """Initialize random population."""
        print(f"   {self.name}: Initializing population of {self.population_size}...")
        
        self.population = []
        for i in range(self.population_size):
            genes = {}
            for gene_name in self.gene_names:
                genes[gene_name] = Gene(
                    name=gene_name,
                    value=random.uniform(0.0, 1.0),
                    mutation_rate=self.mutation_rate
                )
            
            genome = Genome(
                id=f"gen0_ind{i}",
                genes=genes,
                generation=0
            )
            self.population.append(genome)
        
        print(f"   Population initialized with {len(self.population)} individuals")
    
    def evaluate_fitness(self, genome: Genome, task: str) -> float:
        """Evaluate fitness of a genome for a task."""
        # Simulate fitness based on gene values
        # In practice, this would run the agent with these parameters
        
        # Simple fitness function: weighted sum of genes
        fitness = 0.0
        weights = {"learning_rate": 0.3, "exploration": 0.2, "memory_size": 0.2,
                  "attention": 0.15, "creativity": 0.15}
        
        for gene_name, gene in genome.genes.items():
            weight = weights.get(gene_name, 1.0 / len(genome.genes))
            fitness += gene.value * weight
        
        # Add some randomness to simulate noisy evaluation
        fitness *= random.uniform(0.8, 1.2)
        fitness = max(0.0, min(1.0, fitness))
        
        self.statistics["total_evaluations"] += 1
        
        return fitness
    
    def evaluate_population(self, task: str):
        """Evaluate fitness for entire population."""
        for genome in self.population:
            genome.fitness = self.evaluate_fitness(genome, task)
        
        # Sort by fitness
        self.population.sort(key=lambda g: g.fitness, reverse=True)
        
        # Track best genome
        if not self.best_genome or self.population[0].fitness > self.best_genome.fitness:
            self.best_genome = self.population[0]
            self.statistics["best_fitness_ever"] = self.best_genome.fitness
    
    def select_parents(self) -> Tuple[Genome, Genome]:
        """Select two parents for reproduction."""
        if self.selection_strategy == SelectionStrategy.TOURNAMENT:
            return self._tournament_selection(), self._tournament_selection()
        elif self.selection_strategy == SelectionStrategy.ROULETTE:
            return self._roulette_selection(), self._roulette_selection()
        elif self.selection_strategy == SelectionStrategy.RANK:
            return self._rank_selection(), self._rank_selection()
        else:  # ELITISM
            # Select from top half
            top_half = self.population[:len(self.population)//2]
            return random.choice(top_half), random.choice(top_half)
    
    def _tournament_selection(self, tournament_size: int = 3) -> Genome:
        """Tournament selection."""
        tournament = random.sample(self.population, min(tournament_size, len(self.population)))
        return max(tournament, key=lambda g: g.fitness)
    
    def _roulette_selection(self) -> Genome:
        """Roulette wheel selection."""
        total_fitness = sum(g.fitness for g in self.population)
        if total_fitness == 0:
            return random.choice(self.population)
        
        pick = random.uniform(0, total_fitness)
        current = 0
        for genome in self.population:
            current += genome.fitness
            if current > pick:
                return genome
        return self.population[-1]
    
    def _rank_selection(self) -> Genome:
        """Rank-based selection."""
        # Already sorted by fitness
        ranks = list(range(len(self.population), 0, -1))
        total_rank = sum(ranks)
        pick = random.uniform(0, total_rank)
        
        current = 0
        for i, genome in enumerate(self.population):
            current += ranks[i]
            if current > pick:
                return genome
        return self.population[-1]
    
    def crossover(self, parent1: Genome, parent2: Genome) -> Tuple[Genome, Genome]:
        """Perform crossover between two parents."""
        child1_genes = {}
        child2_genes = {}
        
        for gene_name in self.gene_names:
            if random.random() < self.crossover_rate:
                # Crossover: swap genes
                child1_genes[gene_name] = Gene(
                    name=gene_name,
                    value=parent2.genes[gene_name].value,
                    mutation_rate=self.mutation_rate
                )
                child2_genes[gene_name] = Gene(
                    name=gene_name,
                    value=parent1.genes[gene_name].value,
                    mutation_rate=self.mutation_rate
                )
            else:
                # No crossover: copy parent genes
                child1_genes[gene_name] = Gene(
                    name=gene_name,
                    value=parent1.genes[gene_name].value,
                    mutation_rate=self.mutation_rate
                )
                child2_genes[gene_name] = Gene(
                    name=gene_name,
                    value=parent2.genes[gene_name].value,
                    mutation_rate=self.mutation_rate
                )
        
        child1 = Genome(
            id=f"gen{self.current_generation + 1}_child",
            genes=child1_genes,
            generation=self.current_generation + 1,
            parent_ids=[parent1.id, parent2.id]
        )
        
        child2 = Genome(
            id=f"gen{self.current_generation + 1}_child",
            genes=child2_genes,
            generation=self.current_generation + 1,
            parent_ids=[parent1.id, parent2.id]
        )
        
        return child1, child2
    
    def mutate(self, genome: Genome):
        """Mutate a genome."""
        for gene in genome.genes.values():
            if random.random() < gene.mutation_rate:
                # Gaussian mutation
                mutation = random.gauss(0, 0.1)
                gene.value = max(gene.min_value, min(gene.max_value, gene.value + mutation))
    
    def evolve_generation(self, task: str):
        """Evolve one generation."""
        self.current_generation += 1
        
        # Evaluate current population
        self.evaluate_population(task)
        
        # Record statistics
        fitnesses = [g.fitness for g in self.population]
        diversity = self._calculate_diversity()
        
        stats = GenerationStats(
            generation=self.current_generation,
            population_size=len(self.population),
            best_fitness=max(fitnesses),
            avg_fitness=sum(fitnesses) / len(fitnesses),
            worst_fitness=min(fitnesses),
            diversity=diversity
        )
        self.generation_stats.append(stats)
        
        # Create new population
        new_population = []
        
        # Elitism: keep best individuals
        new_population.extend(self.population[:self.elitism_count])
        
        # Generate offspring
        while len(new_population) < self.population_size:
            parent1, parent2 = self.select_parents()
            child1, child2 = self.crossover(parent1, parent2)
            
            self.mutate(child1)
            self.mutate(child2)
            
            new_population.append(child1)
            if len(new_population)  1:
            improvement = (self.generation_stats[-1].avg_fitness - 
                          self.generation_stats[-2].avg_fitness)
            self.statistics["avg_fitness_improvement"] = improvement
    
    def _calculate_diversity(self) -> float:
        """Calculate population diversity."""
        if len(self.population)  0 else 0.0
    
    def run_evolution(self, task: str, generations: int):
        """Run evolution for multiple generations."""
        print(f"\n   {self.name}: Running evolution for {generations} generations...")
        
        if not self.population:
            self.initialize_population()
        
        for gen in range(generations):
            self.evolve_generation(task)
            
            if (gen + 1) % max(1, generations // 5) == 0:
                stats = self.generation_stats[-1]
                print(f"   Generation {self.current_generation}: "
                      f"Best={stats.best_fitness:.3f}, "
                      f"Avg={stats.avg_fitness:.3f}, "
                      f"Diversity={stats.diversity:.3f}")
        
        print(f"   Evolution complete! Best fitness: {self.best_genome.fitness:.3f}")
    
    def get_best_strategy(self) -> Dict[str, float]:
        """Get the best evolved strategy."""
        if not self.best_genome:
            return {}
        
        return {gene_name: gene.value 
                for gene_name, gene in self.best_genome.genes.items()}
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get comprehensive statistics."""
        return {
            "name": self.name,
            "selection_strategy": self.selection_strategy.value,
            "population_size": self.population_size,
            "current_generation": self.current_generation,
            "best_fitness": self.best_genome.fitness if self.best_genome else 0.0,
            **self.statistics
        }


class EvolutionaryPattern:
    """
    Evolutionary Pattern - Evolve agent strategies via genetic algorithms.
    
    Coordinates multiple evolutionary agents optimizing different strategies.
    """
    
    def __init__(self):
        self.agents: Dict[str, EvolutionaryAgent] = {}
        self.tasks: List[str] = []
    
    def register_agent(self, agent: EvolutionaryAgent):
        """Register evolutionary agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name} ({agent.selection_strategy.value})")
    
    def add_task(self, task: str):
        """Add optimization task."""
        self.tasks.append(task)
    
    def evolve_all(self, generations: int):
        """Evolve all agents."""
        print("\n   Evolving all agents...")
        
        for agent in self.agents.values():
            for task in self.tasks:
                agent.run_evolution(task, generations)
    
    def run(self, tasks: List[str], generations: int = 20) -> Dict[str, Any]:
        """Run evolutionary optimization."""
        print("=== Evolutionary Agent Pattern Demo ===\n")
        
        # Add tasks
        print("1. Adding tasks...")
        for task in tasks:
            self.add_task(task)
        
        # Evolve all agents
        print("\n2. Running evolution...")
        self.evolve_all(generations)
        
        # Get best strategies
        print("\n3. Extracting best strategies...")
        best_strategies = {name: agent.get_best_strategy() 
                          for name, agent in self.agents.items()}
        
        # Get statistics
        stats = {name: agent.get_statistics() 
                for name, agent in self.agents.items()}
        
        return {
            "pattern": "Evolutionary",
            "agents": len(self.agents),
            "tasks": len(tasks),
            "generations": generations,
            "best_strategies": best_strategies,
            "statistics": stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Evolutionary Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = EvolutionaryPattern()
    
    # Define gene names (strategy parameters to evolve)
    gene_names = ["learning_rate", "exploration", "memory_size", "attention", "creativity"]
    
    # Create agents with different selection strategies
    print("1. Creating evolutionary agents...")
    tournament_agent = EvolutionaryAgent(
        name="TournamentAgent",
        gene_names=gene_names,
        population_size=30,
        selection_strategy=SelectionStrategy.TOURNAMENT,
        mutation_rate=0.1
    )
    pattern.register_agent(tournament_agent)
    
    roulette_agent = EvolutionaryAgent(
        name="RouletteAgent",
        gene_names=gene_names,
        population_size=30,
        selection_strategy=SelectionStrategy.ROULETTE,
        mutation_rate=0.15
    )
    pattern.register_agent(roulette_agent)
    
    # Define tasks
    tasks = ["optimize_learning_strategy", "maximize_performance"]
    
    # Run evolution
    print("\n2. Running evolutionary optimization...")
    summary = pattern.run(tasks, generations=30)
    
    print(f"\n3. Results:")
    print(f"   Agents: {summary['agents']}")
    print(f"   Tasks: {summary['tasks']}")
    print(f"   Generations: {summary['generations']}")
    
    print(f"\n4. Best Evolved Strategies:")
    for agent_name, strategy in summary['best_strategies'].items():
        print(f"\n   {agent_name}:")
        for param, value in strategy.items():
            print(f"     {param}: {value:.3f}")
    
    print(f"\n5. Evolution Statistics:")
    for agent_name, stats in summary['statistics'].items():
        print(f"\n   {agent_name}:")
        for key, value in stats.items():
            if key != "name":
                if isinstance(value, float):
                    print(f"     {key}: {value:.4f}")
                else:
                    print(f"     {key}: {value}")
    
    print("\nāœ“ Evolutionary algorithms optimized agent strategies!")

Agentic Architecture Pattern: Evolutionary Agent

Pattern 58: 58 Evolutionary Agent – Line-by-Line Explanation

Overview

Pattern 58: Evolutionary Agent

Purpose: Use genetic algorithms to evolve agent strategies and behaviors. Selection, crossover, and mutation to optimize agent parameters over generations.

Code Structure Analysis

File Information

  • Pattern Number: 58
  • Pattern Name: 58 Evolutionary Agent
  • File Path: patterns\learning_adaptation\58_evolutionary_agent.py
  • Total Lines: 486

Detailed Code Explanation

Module Docstring

"""
Pattern 58: Evolutionary Agent

Purpose: Use genetic algorithms to evolve agent strategies and behaviors.
Selection, crossover, and mutation to optimize agent parameters over generations.
"""

Explanation: This module docstring describes the purpose and key features of the 58 Evolutionary Agent pattern.

Imports Section

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

SelectionStrategy (Line 17)

class SelectionStrategy(Enum):
    TOURNAMENT
    ROULETTE
    RANK
    ELITISM

Purpose: Selection strategies for evolution.

Values: – TOURNAMENT: Represents tournament – ROULETTE: Represents roulette – RANK: Represents rank – ELITISM: Represents elitism

Classes

SelectionStrategy (Line 17)

Description: Selection strategies for evolution.

Gene (Line 26)

Description: A gene representing part of an agent’s strategy.

Genome (Line 36)

Description: A genome representing an agent’s complete strategy.

GenerationStats (Line 46)

Description: Statistics for a generation.

EvolutionaryAgent (Line 57)

Description: Evolutionary Agent – Evolve strategies using genetic algorithms.

Uses selection, crossover, and mutation to optimize agent behaviors over multiple generations. LLM assists in evaluating fitness.

Methods:

__init__ (Line 65)
def __init__(self, name, gene_names, population_size, selection_strategy, mutation_rate, crossover_rate, elitism_count):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – gene_names: Parameter for gene_names – population_size: Parameter for population_size – selection_strategy: Parameter for selection_strategy – mutation_rate: Parameter for mutation_rate – crossover_rate: Parameter for crossover_rate – elitism_count: Parameter for elitism_count

initialize_population (Line 99)
def initialize_population(self):
    """Initialize random population."""

Purpose: Initialize random population.

Parameters: – self: Instance reference

evaluate_fitness (Line 122)
def evaluate_fitness(self, genome, task):
    """Evaluate fitness of a genome for a task."""

Purpose: Evaluate fitness of a genome for a task.

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

evaluate_population (Line 144)
def evaluate_population(self, task):
    """Evaluate fitness for entire population."""

Purpose: Evaluate fitness for entire population.

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

select_parents (Line 157)
def select_parents(self):
    """Select two parents for reproduction."""

Purpose: Select two parents for reproduction.

Parameters: – self: Instance reference

_tournament_selection (Line 170)
def _tournament_selection(self, tournament_size):
    """Tournament selection."""

Purpose: Tournament selection.

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

_roulette_selection (Line 175)
def _roulette_selection(self):
    """Roulette wheel selection."""

Purpose: Roulette wheel selection.

Parameters: – self: Instance reference

_rank_selection (Line 189)
def _rank_selection(self):
    """Rank-based selection."""

Purpose: Rank-based selection.

Parameters: – self: Instance reference

crossover (Line 203)
def crossover(self, parent1, parent2):
    """Perform crossover between two parents."""

Purpose: Perform crossover between two parents.

Parameters: – self: Instance reference – parent1: Parameter for parent1 – parent2: Parameter for parent2

mutate (Line 250)
def mutate(self, genome):
    """Mutate a genome."""

Purpose: Mutate a genome.

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

evolve_generation (Line 258)
def evolve_generation(self, task):
    """Evolve one generation."""

Purpose: Evolve one generation.

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

_calculate_diversity (Line 306)
def _calculate_diversity(self):
    """Calculate population diversity."""

Purpose: Calculate population diversity.

Parameters: – self: Instance reference

run_evolution (Line 328)
def run_evolution(self, task, generations):
    """Run evolution for multiple generations."""

Purpose: Run evolution for multiple generations.

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

get_best_strategy (Line 347)
def get_best_strategy(self):
    """Get the best evolved strategy."""

Purpose: Get the best evolved strategy.

Parameters: – self: Instance reference

get_statistics (Line 355)
def get_statistics(self):
    """Get comprehensive statistics."""

Purpose: Get comprehensive statistics.

Parameters: – self: Instance reference

EvolutionaryPattern (Line 367)

Description: Evolutionary Pattern – Evolve agent strategies via genetic algorithms.

Coordinates multiple evolutionary agents optimizing different strategies.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 378)
def register_agent(self, agent):
    """Register evolutionary agent."""

Purpose: Register evolutionary agent.

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

add_task (Line 383)
def add_task(self, task):
    """Add optimization task."""

Purpose: Add optimization task.

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

evolve_all (Line 387)
def evolve_all(self, generations):
    """Evolve all agents."""

Purpose: Evolve all agents.

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

run (Line 395)
def run(self, tasks, generations):
    """Run evolutionary optimization."""

Purpose: Run evolutionary optimization.

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

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 = SelectionStrategy(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