"""
Pattern 28: Swarm Intelligence Pattern

Emergent collective behavior from simple agent interactions.
Inspired by natural swarms (bees, ants, birds) - local interactions lead to global solutions.
"""

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


@dataclass
class SwarmAgent:
    """Individual agent in the swarm."""
    agent_id: str
    position: Tuple[float, float]  # 2D position
    velocity: Tuple[float, float]  # Current velocity
    best_position: Tuple[float, float]  # Personal best
    best_score: float = float('-inf')
    neighbors: List[str] = field(default_factory=list)


@dataclass
class SwarmProblem:
    """Problem to be solved by swarm."""
    problem_id: str
    description: str
    search_space: Tuple[Tuple[float, float], Tuple[float, float]]  # ((x_min, x_max), (y_min, y_max))
    goal: str


@dataclass
class SwarmSolution:
    """Solution found by swarm."""
    problem_id: str
    best_position: Tuple[float, float]
    best_score: float
    iterations: int
    convergence_history: List[float]


class SwarmIntelligencePattern:
    """
    Swarm Intelligence Pattern - Emergent collective optimization.
    
    Multiple simple agents follow local rules and interact with neighbors.
    Global optimal solutions emerge from collective behavior.
    Inspired by Particle Swarm Optimization (PSO).
    """
    
    def __init__(self, num_agents: int = 20, inertia: float = 0.7, 
                 cognitive: float = 1.5, social: float = 1.5, model_name: str = "gpt-4"):
        self.num_agents = num_agents
        self.inertia = inertia  # Weight for current velocity
        self.cognitive = cognitive  # Weight for personal best
        self.social = social  # Weight for global best
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        self.swarm: List[SwarmAgent] = []
        self.global_best_position: Optional[Tuple[float, float]] = None
        self.global_best_score: float = float('-inf')
    
    def initialize_swarm(self, search_space: Tuple[Tuple[float, float], Tuple[float, float]]):
        """Initialize swarm agents in search space."""
        self.swarm = []
        (x_min, x_max), (y_min, y_max) = search_space
        
        for i in range(self.num_agents):
            position = (
                random.uniform(x_min, x_max),
                random.uniform(y_min, y_max)
            )
            velocity = (
                random.uniform(-1, 1),
                random.uniform(-1, 1)
            )
            
            agent = SwarmAgent(
                agent_id=f"agent_{i}",
                position=position,
                velocity=velocity,
                best_position=position
            )
            self.swarm.append(agent)
    
    def evaluate_position(self, position: Tuple[float, float], problem: SwarmProblem) -> float:
        """
        Evaluate fitness of position for problem.
        Uses LLM to score position quality.
        """
        x, y = position
        
        # For demo: simple optimization function (can be replaced with LLM evaluation)
        # Maximize: -(x^2 + y^2) (peak at origin)
        score = -(x**2 + y**2)
        
        return score
    
    def update_velocity(self, agent: SwarmAgent) -> Tuple[float, float]:
        """Update agent velocity based on personal and global best."""
        if not self.global_best_position:
            return agent.velocity
        
        vx, vy = agent.velocity
        x, y = agent.position
        pbx, pby = agent.best_position
        gbx, gby = self.global_best_position
        
        # PSO velocity update formula
        r1, r2 = random.random(), random.random()
        
        new_vx = (self.inertia * vx + 
                  self.cognitive * r1 * (pbx - x) + 
                  self.social * r2 * (gbx - x))
        
        new_vy = (self.inertia * vy + 
                  self.cognitive * r1 * (pby - y) + 
                  self.social * r2 * (gby - y))
        
        # Limit velocity
        max_velocity = 2.0
        new_vx = max(-max_velocity, min(max_velocity, new_vx))
        new_vy = max(-max_velocity, min(max_velocity, new_vy))
        
        return (new_vx, new_vy)
    
    def update_position(self, agent: SwarmAgent, search_space: Tuple[Tuple[float, float], Tuple[float, float]]):
        """Update agent position based on velocity."""
        (x_min, x_max), (y_min, y_max) = search_space
        
        x, y = agent.position
        vx, vy = agent.velocity
        
        new_x = x + vx
        new_y = y + vy
        
        # Keep in bounds
        new_x = max(x_min, min(x_max, new_x))
        new_y = max(y_min, min(y_max, new_y))
        
        agent.position = (new_x, new_y)
    
    def solve(self, problem: SwarmProblem, max_iterations: int = 30) -> SwarmSolution:
        """Solve problem using swarm intelligence."""
        print(f"\n   Problem: {problem.description}")
        print(f"   Swarm size: {self.num_agents} agents")
        print(f"   Search space: {problem.search_space}")
        
        # Initialize swarm
        self.initialize_swarm(problem.search_space)
        convergence_history = []
        
        # Optimize over iterations
        for iteration in range(max_iterations):
            # Evaluate all agents
            for agent in self.swarm:
                score = self.evaluate_position(agent.position, problem)
                
                # Update personal best
                if score > agent.best_score:
                    agent.best_score = score
                    agent.best_position = agent.position
                
                # Update global best
                if score > self.global_best_score:
                    self.global_best_score = score
                    self.global_best_position = agent.position
            
            # Update velocities and positions
            for agent in self.swarm:
                agent.velocity = self.update_velocity(agent)
                self.update_position(agent, problem.search_space)
            
            convergence_history.append(self.global_best_score)
            
            if iteration % 10 == 0:
                print(f"   Iteration {iteration}: Best score = {self.global_best_score:.4f}")
        
        print(f"   ✓ Final best position: {self.global_best_position}")
        print(f"   ✓ Final best score: {self.global_best_score:.4f}")
        
        solution = SwarmSolution(
            problem_id=problem.problem_id,
            best_position=self.global_best_position or (0, 0),
            best_score=self.global_best_score,
            iterations=max_iterations,
            convergence_history=convergence_history
        )
        
        return solution
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute swarm intelligence pattern."""
        return {
            "pattern": "SwarmIntelligence",
            "swarm_size": self.num_agents,
            "inertia": self.inertia,
            "cognitive_weight": self.cognitive,
            "social_weight": self.social,
            "global_best_score": self.global_best_score,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Swarm Intelligence Pattern Demo ===\n")
    
    # Create swarm pattern
    print("1. Creating swarm...")
    pattern = SwarmIntelligencePattern(
        num_agents=25,
        inertia=0.7,
        cognitive=1.5,
        social=1.5
    )
    
    # Define optimization problem
    print("\n2. Defining optimization problem...")
    problem = SwarmProblem(
        problem_id="optimize1",
        description="Find maximum point in 2D space (minimize distance from origin)",
        search_space=((-10, 10), (-10, 10)),
        goal="Find position closest to origin (0, 0)"
    )
    
    # Solve using swarm
    print("\n3. Solving with swarm intelligence...")
    solution = pattern.solve(problem, max_iterations=30)
    
    # Summary
    print("\n4. Solution Summary:")
    print(f"   Best Position: ({solution.best_position[0]:.4f}, {solution.best_position[1]:.4f})")
    print(f"   Best Score: {solution.best_score:.4f}")
    print(f"   Iterations: {solution.iterations}")
    print(f"   Improvement: {solution.convergence_history[-1] - solution.convergence_history[0]:.4f}")
    
    summary = pattern.run("2D optimization via swarm")
    print(f"\n5. Pattern Configuration:")
    print(f"   Swarm Size: {summary['swarm_size']}")
    print(f"   Inertia: {summary['inertia']}")
    print(f"   Cognitive: {summary['cognitive_weight']}")
    print(f"   Social: {summary['social_weight']}")
    
    print("\n✓ Swarm intelligence found optimal solution through emergent behavior!")

    

Agentic Architecture Pattern: Swarm Intelligence

Pattern 28: 28 Swarm Intelligence – Line-by-Line Explanation

Overview

Pattern 28: Swarm Intelligence Pattern

Emergent collective behavior from simple agent interactions. Inspired by natural swarms (bees, ants, birds) – local interactions lead to global solutions.

Code Structure Analysis

File Information

• Pattern Number: 28
• Pattern Name: 28 Swarm Intelligence
• File Path: patterns\collaboration\28_swarm_intelligence.py
• Total Lines: 250

Detailed Code Explanation

Module Docstring

"""
Pattern 28: Swarm Intelligence Pattern

Emergent collective behavior from simple agent interactions.
Inspired by natural swarms (bees, ants, birds) - local interactions lead to global solutions.
"""

Explanation: This module docstring describes the purpose and key features of the 28 Swarm Intelligence pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

SwarmAgent (Line 17)

Description: Individual agent in the swarm.

SwarmProblem (Line 28)

Description: Problem to be solved by swarm.

SwarmSolution (Line 37)

Description: Solution found by swarm.

SwarmIntelligencePattern (Line 46)

Description: Swarm Intelligence Pattern – Emergent collective optimization.

Multiple simple agents follow local rules and interact with neighbors. Global optimal solutions emerge from collective behavior. Inspired by Particle Swarm Optimization (PSO).

Methods:

__init__ (Line 55)

def __init__(self, num_agents, inertia, cognitive, social, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – num_agents: Parameter for num_agents – inertia: Parameter for inertia – cognitive: Parameter for cognitive – social: Parameter for social – model_name: LLM model to use (e.g., “gpt-4”)

initialize_swarm (Line 67)

def initialize_swarm(self, search_space):
    """Initialize swarm agents in search space."""

Purpose: Initialize swarm agents in search space.

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

evaluate_position (Line 90)

def evaluate_position(self, position, problem):
    """Evaluate fitness of position for problem.
Uses LLM to score position quality."""

Purpose: Evaluate fitness of position for problem. Uses LLM to score position quality.

Parameters: – self: Instance reference – position: Parameter for position – problem: Parameter for problem

update_velocity (Line 103)

def update_velocity(self, agent):
    """Update agent velocity based on personal and global best."""

Purpose: Update agent velocity based on personal and global best.

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

update_position (Line 131)

def update_position(self, agent, search_space):
    """Update agent position based on velocity."""

Purpose: Update agent position based on velocity.

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

solve (Line 147)

def solve(self, problem, max_iterations):
    """Solve problem using swarm intelligence."""

Purpose: Solve problem using swarm intelligence.

Parameters: – self: Instance reference – problem: Parameter for problem – max_iterations: Parameter for max_iterations

run (Line 196)

def run(self, scenario):
    """Execute swarm intelligence pattern."""

Purpose: Execute swarm intelligence pattern.

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

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