"""
Pattern 60: Self-Organizing Agent

Purpose: Agents spontaneously organize into emergent structures without central control.
Bottom-up organization through local interactions and simple rules leading to complex behaviors.
"""

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


class OrganizationRule(Enum):
    """Rules for self-organization."""
    PROXIMITY = "proximity"  # Organize by nearness
    SIMILARITY = "similarity"  # Organize by similarity
    COMPLEMENTARITY = "complementarity"  # Organize by complementary skills
    WORKLOAD_BALANCE = "workload_balance"  # Balance workload


class AgentRole(Enum):
    """Emergent roles agents can take."""
    COORDINATOR = "coordinator"  # Coordinates group activities
    SPECIALIST = "specialist"  # Specialized in specific tasks
    GENERALIST = "generalist"  # Handles diverse tasks
    CONNECTOR = "connector"  # Bridges between groups


@dataclass
class AgentProfile:
    """Profile of an autonomous agent."""
    id: str
    name: str
    skills: Set[str]
    capacity: float  # 0.0 to 1.0
    workload: float = 0.0
    connections: Set[str] = field(default_factory=set)
    position: tuple = (0.0, 0.0)  # 2D position for proximity-based organization
    role: Optional[AgentRole] = None


@dataclass
class Task:
    """Task to be organized."""
    id: str
    name: str
    required_skills: Set[str]
    complexity: float  # 0.0 to 1.0
    assigned_to: Optional[str] = None


@dataclass
class OrganizationCluster:
    """Emergent cluster of agents."""
    id: str
    agents: Set[str]
    centroid: tuple = (0.0, 0.0)
    shared_skills: Set[str] = field(default_factory=set)
    total_capacity: float = 0.0


class SelfOrganizingAgent:
    """
    Self-Organizing Agent - Spontaneous organization through local interactions.
    
    Agents organize themselves into emergent structures without central control,
    following simple rules that lead to complex adaptive behaviors.
    """
    
    def __init__(self,
                 organization_rule: OrganizationRule = OrganizationRule.SIMILARITY,
                 proximity_threshold: float = 5.0,
                 similarity_threshold: float = 0.3,
                 reorg_interval: int = 10):
        self.organization_rule = organization_rule
        self.proximity_threshold = proximity_threshold
        self.similarity_threshold = similarity_threshold
        self.reorg_interval = reorg_interval
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Agent population
        self.agents: Dict[str, AgentProfile] = {}
        self.tasks: Dict[str, Task] = {}
        
        # Emergent structures
        self.clusters: Dict[str, OrganizationCluster] = {}
        self.interaction_count = 0
        self.reorganization_count = 0
        
        # History
        self.organization_history: List[Dict[str, Any]] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_agents": 0,
            "total_tasks": 0,
            "total_clusters": 0,
            "reorganizations": 0,
            "avg_cluster_size": 0.0
        }
    
    def add_agent(self, agent: AgentProfile):
        """Add autonomous agent to population."""
        self.agents[agent.id] = agent
        self.statistics["total_agents"] = len(self.agents)
        print(f"   Added agent: {agent.name} with skills {agent.skills}")
    
    def add_task(self, task: Task):
        """Add task to be organized."""
        self.tasks[task.id] = task
        self.statistics["total_tasks"] = len(self.tasks)
    
    def organize(self):
        """Trigger self-organization based on current rule."""
        print(f"\n   Self-organizing using {self.organization_rule.value} rule...")
        
        if self.organization_rule == OrganizationRule.PROXIMITY:
            self._organize_by_proximity()
        elif self.organization_rule == OrganizationRule.SIMILARITY:
            self._organize_by_similarity()
        elif self.organization_rule == OrganizationRule.COMPLEMENTARITY:
            self._organize_by_complementarity()
        else:  # WORKLOAD_BALANCE
            self._organize_by_workload()
        
        self._assign_roles()
        self._assign_tasks()
        self.reorganization_count += 1
        self.statistics["reorganizations"] = self.reorganization_count
        
        # Record organization state
        self.organization_history.append({
            "reorganization": self.reorganization_count,
            "clusters": len(self.clusters),
            "avg_cluster_size": self._calculate_avg_cluster_size(),
            "timestamp": datetime.now()
        })
    
    def _organize_by_proximity(self):
        """Organize agents by spatial proximity."""
        self.clusters = {}
        unclustered = set(self.agents.keys())
        cluster_id = 0
        
        while unclustered:
            # Start new cluster with random agent
            seed_id = random.choice(list(unclustered))
            seed_agent = self.agents[seed_id]
            
            cluster_agents = {seed_id}
            unclustered.remove(seed_id)
            
            # Add nearby agents
            for agent_id in list(unclustered):
                agent = self.agents[agent_id]
                distance = self._calculate_distance(seed_agent.position, agent.position)
                
                if distance = self.similarity_threshold:
                    cluster_agents.add(agent_id)
                    unclustered.remove(agent_id)
            
            # Calculate shared skills
            shared_skills = set.intersection(*[self.agents[aid].skills 
                                              for aid in cluster_agents])
            
            cluster = OrganizationCluster(
                id=f"cluster_{cluster_id}",
                agents=cluster_agents,
                shared_skills=shared_skills
            )
            self.clusters[cluster.id] = cluster
            cluster_id += 1
        
        print(f"   Formed {len(self.clusters)} similarity-based clusters")
    
    def _organize_by_complementarity(self):
        """Organize agents with complementary skills."""
        self.clusters = {}
        unclustered = set(self.agents.keys())
        cluster_id = 0
        
        # Collect all skills
        all_skills = set()
        for agent in self.agents.values():
            all_skills.update(agent.skills)
        
        while unclustered and len(unclustered) >= 2:
            # Start with random agent
            seed_id = random.choice(list(unclustered))
            cluster_agents = {seed_id}
            unclustered.remove(seed_id)
            
            cluster_skills = set(self.agents[seed_id].skills)
            
            # Add agents with complementary skills
            for agent_id in list(unclustered):
                agent = self.agents[agent_id]
                # Complementary if they have different skills
                overlap = len(cluster_skills & agent.skills)
                if overlap = 4:  # Limit cluster size
                    break
            
            cluster = OrganizationCluster(
                id=f"cluster_{cluster_id}",
                agents=cluster_agents,
                shared_skills=cluster_skills
            )
            self.clusters[cluster.id] = cluster
            cluster_id += 1
        
        # Add remaining agents to existing clusters
        for agent_id in unclustered:
            if self.clusters:
                # Add to smallest cluster
                smallest_cluster = min(self.clusters.values(), key=lambda c: len(c.agents))
                smallest_cluster.agents.add(agent_id)
        
        print(f"   Formed {len(self.clusters)} complementarity-based clusters")
    
    def _organize_by_workload(self):
        """Organize to balance workload."""
        # Target cluster count
        target_clusters = max(1, len(self.agents) // 4)
        
        self.clusters = {}
        for i in range(target_clusters):
            self.clusters[f"cluster_{i}"] = OrganizationCluster(
                id=f"cluster_{i}",
                agents=set()
            )
        
        # Sort agents by workload (descending)
        sorted_agents = sorted(self.agents.items(), 
                              key=lambda x: x[1].workload, reverse=True)
        
        # Distribute agents to balance load
        cluster_loads = {cid: 0.0 for cid in self.clusters.keys()}
        
        for agent_id, agent in sorted_agents:
            # Assign to least loaded cluster
            min_cluster = min(cluster_loads.items(), key=lambda x: x[1])
            cluster_id = min_cluster[0]
            
            self.clusters[cluster_id].agents.add(agent_id)
            cluster_loads[cluster_id] += agent.workload
        
        print(f"   Formed {len(self.clusters)} workload-balanced clusters")
    
    def _assign_roles(self):
        """Assign emergent roles to agents based on their position in structure."""
        for cluster in self.clusters.values():
            if len(cluster.agents) == 0:
                continue
            
            # Coordinator: agent with most connections
            max_connections = 0
            coordinator = None
            for agent_id in cluster.agents:
                agent = self.agents[agent_id]
                conn_count = len(agent.connections & cluster.agents)
                if conn_count > max_connections:
                    max_connections = conn_count
                    coordinator = agent_id
            
            if coordinator:
                self.agents[coordinator].role = AgentRole.COORDINATOR
            
            # Specialists: agents with unique skills
            for agent_id in cluster.agents:
                if agent_id == coordinator:
                    continue
                agent = self.agents[agent_id]
                # Check if has unique skills
                others_skills = set()
                for other_id in cluster.agents:
                    if other_id != agent_id:
                        others_skills.update(self.agents[other_id].skills)
                
                unique_skills = agent.skills - others_skills
                if unique_skills:
                    agent.role = AgentRole.SPECIALIST
                else:
                    agent.role = AgentRole.GENERALIST
    
    def _assign_tasks(self):
        """Assign tasks to agents based on organization."""
        for task in self.tasks.values():
            if task.assigned_to:
                continue  # Already assigned
            
            # Find best cluster for task
            best_cluster = None
            best_score = 0.0
            
            for cluster in self.clusters.values():
                # Calculate cluster's match with task
                cluster_skills = set()
                total_capacity = 0.0
                for agent_id in cluster.agents:
                    agent = self.agents[agent_id]
                    cluster_skills.update(agent.skills)
                    total_capacity += (agent.capacity - agent.workload)
                
                skill_match = len(task.required_skills & cluster_skills) / len(task.required_skills) \
                             if task.required_skills else 1.0
                capacity_score = min(1.0, total_capacity / task.complexity)
                score = skill_match * 0.7 + capacity_score * 0.3
                
                if score > best_score:
                    best_score = score
                    best_cluster = cluster
            
            # Assign to best agent in cluster
            if best_cluster and best_cluster.agents:
                best_agent = None
                best_agent_score = 0.0
                
                for agent_id in best_cluster.agents:
                    agent = self.agents[agent_id]
                    skill_match = len(task.required_skills & agent.skills) / len(task.required_skills) \
                                 if task.required_skills else 1.0
                    capacity_available = agent.capacity - agent.workload
                    agent_score = skill_match * 0.8 + min(1.0, capacity_available / task.complexity) * 0.2
                    
                    if agent_score > best_agent_score and capacity_available > 0:
                        best_agent_score = agent_score
                        best_agent = agent_id
                
                if best_agent:
                    task.assigned_to = best_agent
                    self.agents[best_agent].workload += task.complexity
    
    def _calculate_distance(self, pos1: tuple, pos2: tuple) -> float:
        """Calculate Euclidean distance."""
        return math.sqrt((pos1[0] - pos2[0])**2 + (pos1[1] - pos2[1])**2)
    
    def _calculate_centroid(self, agent_ids: Set[str]) -> tuple:
        """Calculate centroid of agent positions."""
        if not agent_ids:
            return (0.0, 0.0)
        
        x_sum = sum(self.agents[aid].position[0] for aid in agent_ids)
        y_sum = sum(self.agents[aid].position[1] for aid in agent_ids)
        count = len(agent_ids)
        
        return (x_sum / count, y_sum / count)
    
    def _calculate_skill_similarity(self, skills1: Set[str], skills2: Set[str]) -> float:
        """Calculate Jaccard similarity between skill sets."""
        if not skills1 or not skills2:
            return 0.0
        
        intersection = len(skills1 & skills2)
        union = len(skills1 | skills2)
        
        return intersection / union if union > 0 else 0.0
    
    def _calculate_avg_cluster_size(self) -> float:
        """Calculate average cluster size."""
        if not self.clusters:
            return 0.0
        
        total_agents = sum(len(c.agents) for c in self.clusters.values())
        return total_agents / len(self.clusters)
    
    def get_organization_summary(self) -> Dict[str, Any]:
        """Get summary of current organization."""
        cluster_info = []
        for cluster in self.clusters.values():
            cluster_info.append({
                "id": cluster.id,
                "size": len(cluster.agents),
                "skills": list(cluster.shared_skills),
                "agents": [self.agents[aid].name for aid in cluster.agents]
            })
        
        role_distribution = {}
        for agent in self.agents.values():
            role = agent.role.value if agent.role else "unassigned"
            role_distribution[role] = role_distribution.get(role, 0) + 1
        
        return {
            "total_clusters": len(self.clusters),
            "avg_cluster_size": self._calculate_avg_cluster_size(),
            "clusters": cluster_info,
            "role_distribution": role_distribution,
            "tasks_assigned": sum(1 for t in self.tasks.values() if t.assigned_to)
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get comprehensive statistics."""
        self.statistics["total_clusters"] = len(self.clusters)
        self.statistics["avg_cluster_size"] = self._calculate_avg_cluster_size()
        
        return self.statistics


class SelfOrganizingPattern:
    """
    Self-Organizing Pattern - Coordinate emergent organization.
    
    Manages self-organizing systems where complex structures emerge from simple rules.
    """
    
    def __init__(self):
        self.systems: Dict[str, SelfOrganizingAgent] = {}
    
    def register_system(self, system: SelfOrganizingAgent, name: str):
        """Register self-organizing system."""
        self.systems[name] = system
        print(f"   Registered system: {name}")
    
    def run(self) -> Dict[str, Any]:
        """Run self-organization."""
        print("=== Self-Organizing Agent Pattern Demo ===\n")
        
        print("1. Organizing systems...")
        for name, system in self.systems.items():
            print(f"\n   System: {name}")
            system.organize()
        
        print("\n2. Getting organization summaries...")
        summaries = {}
        for name, system in self.systems.items():
            summaries[name] = system.get_organization_summary()
        
        return {
            "pattern": "Self-Organizing",
            "systems": len(self.systems),
            "summaries": summaries
        }


# Example usage
if __name__ == "__main__":
    print("=== Self-Organizing Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = SelfOrganizingPattern()
    
    # Create self-organizing system with similarity rule
    print("1. Creating self-organizing system...")
    system = SelfOrganizingAgent(
        organization_rule=OrganizationRule.SIMILARITY,
        similarity_threshold=0.25
    )
    
    # Add agents with different skills
    print("\n2. Adding autonomous agents...")
    agents_data = [
        ("A1", "Alice", {"python", "ml", "data_analysis"}),
        ("A2", "Bob", {"python", "web", "backend"}),
        ("A3", "Carol", {"ml", "data_analysis", "statistics"}),
        ("A4", "Dave", {"web", "frontend", "design"}),
        ("A5", "Eve", {"python", "ml", "nlp"}),
        ("A6", "Frank", {"backend", "database", "devops"}),
        ("A7", "Grace", {"data_analysis", "statistics", "visualization"}),
        ("A8", "Hank", {"frontend", "design", "ux"}),
    ]
    
    for agent_id, name, skills in agents_data:
        agent = AgentProfile(
            id=agent_id,
            name=name,
            skills=skills,
            capacity=random.uniform(0.7, 1.0),
            position=(random.uniform(0, 10), random.uniform(0, 10))
        )
        system.add_agent(agent)
    
    # Add tasks
    print("\n3. Adding tasks...")
    tasks_data = [
        ("T1", "ML Pipeline", {"python", "ml", "data_analysis"}),
        ("T2", "Web API", {"python", "web", "backend"}),
        ("T3", "Data Dashboard", {"data_analysis", "visualization", "frontend"}),
        ("T4", "NLP Model", {"python", "ml", "nlp"}),
    ]
    
    for task_id, name, skills in tasks_data:
        task = Task(
            id=task_id,
            name=name,
            required_skills=skills,
            complexity=random.uniform(0.3, 0.8)
        )
        system.add_task(task)
    
    # Register and run
    pattern.register_system(system, "SkillBasedOrganization")
    
    print("\n4. Running self-organization...")
    summary = pattern.run()
    
    print(f"\n5. Organization Results:")
    print(f"   Pattern: {summary['pattern']}")
    print(f"   Systems: {summary['systems']}")
    
    for system_name, system_summary in summary['summaries'].items():
        print(f"\n   System: {system_name}")
        print(f"      Total clusters: {system_summary['total_clusters']}")
        print(f"      Avg cluster size: {system_summary['avg_cluster_size']:.1f}")
        print(f"      Tasks assigned: {system_summary['tasks_assigned']}/{len(tasks_data)}")
        
        print(f"\n      Clusters:")
        for cluster in system_summary['clusters']:
            print(f"         {cluster['id']}: {cluster['size']} agents")
            print(f"            Skills: {cluster['skills']}")
            print(f"            Agents: {cluster['agents']}")
        
        print(f"\n      Role distribution:")
        for role, count in system_summary['role_distribution'].items():
            print(f"         {role}: {count}")
    
    print(f"\n6. Statistics:")
    stats = system.get_statistics()
    for key, value in stats.items():
        if isinstance(value, float):
            print(f"   {key}: {value:.2f}")
        else:
            print(f"   {key}: {value}")
    
    print("\n✓ Agents self-organized into emergent structures!")

    

Agentic Architecture Pattern: Self Organizing

Pattern 60: 60 Self Organizing – Line-by-Line Explanation

Overview

Pattern 60: Self-Organizing Agent

Purpose: Agents spontaneously organize into emergent structures without central control. Bottom-up organization through local interactions and simple rules leading to complex behaviors.

Code Structure Analysis

File Information

• Pattern Number: 60
• Pattern Name: 60 Self Organizing
• File Path: patterns\learning_adaptation\60_self_organizing.py
• Total Lines: 568

Detailed Code Explanation

Module Docstring

"""
Pattern 60: Self-Organizing Agent

Purpose: Agents spontaneously organize into emergent structures without central control.
Bottom-up organization through local interactions and simple rules leading to complex behaviors.
"""

Explanation: This module docstring describes the purpose and key features of the 60 Self Organizing pattern.

Imports Section

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

OrganizationRule (Line 18)

class OrganizationRule(Enum):
    PROXIMITY
    SIMILARITY
    COMPLEMENTARITY
    WORKLOAD_BALANCE

Purpose: Rules for self-organization.

Values: – PROXIMITY: Represents proximity – SIMILARITY: Represents similarity – COMPLEMENTARITY: Represents complementarity – WORKLOAD_BALANCE: Represents workload balance

AgentRole (Line 26)

class AgentRole(Enum):
    COORDINATOR
    SPECIALIST
    GENERALIST
    CONNECTOR

Purpose: Emergent roles agents can take.

Values: – COORDINATOR: Represents coordinator – SPECIALIST: Represents specialist – GENERALIST: Represents generalist – CONNECTOR: Represents connector

Classes

OrganizationRule (Line 18)

Description: Rules for self-organization.

AgentRole (Line 26)

Description: Emergent roles agents can take.

AgentProfile (Line 35)

Description: Profile of an autonomous agent.

Task (Line 48)

Description: Task to be organized.

OrganizationCluster (Line 58)

Description: Emergent cluster of agents.

SelfOrganizingAgent (Line 67)

Description: Self-Organizing Agent – Spontaneous organization through local interactions.

Agents organize themselves into emergent structures without central control, following simple rules that lead to complex adaptive behaviors.

Methods:

__init__ (Line 75)

def __init__(self, organization_rule, proximity_threshold, similarity_threshold, reorg_interval):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – organization_rule: Parameter for organization_rule – proximity_threshold: Parameter for proximity_threshold – similarity_threshold: Parameter for similarity_threshold – reorg_interval: Parameter for reorg_interval

add_agent (Line 107)

def add_agent(self, agent):
    """Add autonomous agent to population."""

Purpose: Add autonomous agent to population.

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

add_task (Line 113)

def add_task(self, task):
    """Add task to be organized."""

Purpose: Add task to be organized.

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

organize (Line 118)

def organize(self):
    """Trigger self-organization based on current rule."""

Purpose: Trigger self-organization based on current rule.

Parameters: – self: Instance reference

_organize_by_proximity (Line 144)

def _organize_by_proximity(self):
    """Organize agents by spatial proximity."""

Purpose: Organize agents by spatial proximity.

Parameters: – self: Instance reference

_organize_by_similarity (Line 178)

def _organize_by_similarity(self):
    """Organize agents by skill similarity."""

Purpose: Organize agents by skill similarity.

Parameters: – self: Instance reference

_organize_by_complementarity (Line 215)

def _organize_by_complementarity(self):
    """Organize agents with complementary skills."""

Purpose: Organize agents with complementary skills.

Parameters: – self: Instance reference

_organize_by_workload (Line 264)

def _organize_by_workload(self):
    """Organize to balance workload."""

Purpose: Organize to balance workload.

Parameters: – self: Instance reference

_assign_roles (Line 293)

def _assign_roles(self):
    """Assign emergent roles to agents based on their position in structure."""

Purpose: Assign emergent roles to agents based on their position in structure.

Parameters: – self: Instance reference

_assign_tasks (Line 329)

def _assign_tasks(self):
    """Assign tasks to agents based on organization."""

Purpose: Assign tasks to agents based on organization.

Parameters: – self: Instance reference

_calculate_distance (Line 377)

def _calculate_distance(self, pos1, pos2):
    """Calculate Euclidean distance."""

Purpose: Calculate Euclidean distance.

Parameters: – self: Instance reference – pos1: Parameter for pos1 – pos2: Parameter for pos2

_calculate_centroid (Line 381)

def _calculate_centroid(self, agent_ids):
    """Calculate centroid of agent positions."""

Purpose: Calculate centroid of agent positions.

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

_calculate_skill_similarity (Line 392)

def _calculate_skill_similarity(self, skills1, skills2):
    """Calculate Jaccard similarity between skill sets."""

Purpose: Calculate Jaccard similarity between skill sets.

Parameters: – self: Instance reference – skills1: Parameter for skills1 – skills2: Parameter for skills2

_calculate_avg_cluster_size (Line 402)

def _calculate_avg_cluster_size(self):
    """Calculate average cluster size."""

Purpose: Calculate average cluster size.

Parameters: – self: Instance reference

get_organization_summary (Line 410)

def get_organization_summary(self):
    """Get summary of current organization."""

Purpose: Get summary of current organization.

Parameters: – self: Instance reference

…and 1 more methods

SelfOrganizingPattern (Line 442)

Description: Self-Organizing Pattern – Coordinate emergent organization.

Manages self-organizing systems where complex structures emerge from simple rules.

Methods:

__init__ (Line 449)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_system (Line 452)

def register_system(self, system, name):
    """Register self-organizing system."""

Purpose: Register self-organizing system.

Parameters: – self: Instance reference – system: Parameter for system – name: Identifier or label

run (Line 457)

def run(self):
    """Run self-organization."""

Purpose: Run self-organization.

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

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

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

    """
Pattern 57: Meta-Learning

Purpose: Learn how to learn - optimize learning algorithms and strategies themselves.
"Learning to learn" by improving the learning process across multiple tasks.
"""

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 MetaStrategy(Enum):
    """Meta-learning strategies."""
    MAML = "maml"  # Model-Agnostic Meta-Learning
    REPTILE = "reptile"  # Simplified MAML
    LEARNING_TO_COMPARE = "learning_to_compare"  # Siamese networks
    OPTIMIZATION_AS_MODEL = "optimization_as_model"  # Learn optimizer


@dataclass
class Task:
    """A learning task."""
    id: str
    name: str
    domain: str
    examples: List[Dict[str, Any]] = field(default_factory=list)
    difficulty: float = 0.5
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class LearningExperience:
    """Experience from learning a task."""
    task_id: str
    strategy_used: str
    initial_performance: float
    final_performance: float
    learning_rate: float
    steps_taken: int
    insights: List[str] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class MetaKnowledge:
    """Meta-level knowledge about learning."""
    strategy: str
    effectiveness: float
    best_domains: List[str]
    optimal_learning_rate: float
    optimal_batch_size: int
    learned_from_tasks: int
    confidence: float


class MetaLearningAgent:
    """
    Meta-Learning Agent - Learn how to learn.
    
    Optimizes the learning process itself by learning from experience
    across multiple tasks. Discovers optimal learning strategies,
    hyperparameters, and approaches.
    """
    
    def __init__(self, 
                 name: str,
                 strategy: MetaStrategy = MetaStrategy.MAML):
        self.name = name
        self.strategy = strategy
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Meta-knowledge base
        self.learning_experiences: List[LearningExperience] = []
        self.meta_knowledge: Dict[str, MetaKnowledge] = {}
        
        # Learned meta-parameters
        self.optimal_learning_rate: float = 0.01
        self.optimal_batch_size: int = 32
        self.adaptation_steps: int = 5
        
        # Performance tracking
        self.tasks_learned: int = 0
        self.meta_updates: int = 0
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "avg_initial_performance": 0.0,
            "avg_final_performance": 0.0,
            "avg_improvement": 0.0,
            "best_strategy": None
        }
    
    def meta_train(self, tasks: List[Task]):
        """Meta-train on multiple tasks to learn how to learn."""
        print(f"\n   {self.name}: Meta-training on {len(tasks)} tasks...")
        
        for task in tasks:
            # Learn the task
            experience = self._learn_task(task)
            self.learning_experiences.append(experience)
            self.tasks_learned += 1
            
            # Update meta-knowledge
            self._update_meta_knowledge(experience)
            self.meta_updates += 1
            
            if self.tasks_learned % 5 == 0:
                print(f"   Meta-trained on {self.tasks_learned} tasks...")
        
        # Extract meta-level insights
        self._extract_meta_insights()
        
        print(f"   Meta-training complete! Learned from {self.tasks_learned} tasks")
    
    def _learn_task(self, task: Task) -> LearningExperience:
        """Learn a single task and record the experience."""
        # Simulate initial performance
        initial_perf = random.uniform(0.3, 0.5)
        
        # Apply current meta-knowledge
        learning_rate = self.optimal_learning_rate
        batch_size = self.optimal_batch_size
        
        # Simulate learning with current strategy
        final_perf = initial_perf
        steps = 0
        
        for step in range(self.adaptation_steps):
            # Simulate improvement based on strategy effectiveness
            if self.strategy == MetaStrategy.MAML:
                improvement = 0.08 * (1.0 - final_perf)  # Fast adaptation
            elif self.strategy == MetaStrategy.REPTILE:
                improvement = 0.06 * (1.0 - final_perf)
            elif self.strategy == MetaStrategy.LEARNING_TO_COMPARE:
                improvement = 0.07 * (1.0 - final_perf)
            else:  # OPTIMIZATION_AS_MODEL
                improvement = 0.09 * (1.0 - final_perf)
            
            # Add some randomness
            improvement *= random.uniform(0.8, 1.2)
            final_perf = min(0.95, final_perf + improvement)
            steps += 1
            
            # Early stopping if good enough
            if final_perf > 0.85:
                break
        
        # Generate insights using LLM
        insights = self._generate_learning_insights(task, initial_perf, final_perf, steps)
        
        return LearningExperience(
            task_id=task.id,
            strategy_used=self.strategy.value,
            initial_performance=initial_perf,
            final_performance=final_perf,
            learning_rate=learning_rate,
            steps_taken=steps,
            insights=insights
        )
    
    def _generate_learning_insights(self, 
                                    task: Task,
                                    initial_perf: float,
                                    final_perf: float,
                                    steps: int) -> List[str]:
        """Generate insights about the learning process."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a meta-learning expert analyzing learning experiences. "
                      "Provide 2-3 brief insights about what made this learning process effective or ineffective."),
            ("human", f"Task: {task.name} (domain: {task.domain})\n"
                     f"Strategy: {self.strategy.value}\n"
                     f"Initial Performance: {initial_perf:.2%}\n"
                     f"Final Performance: {final_perf:.2%}\n"
                     f"Steps: {steps}\n"
                     f"Improvement: {final_perf - initial_perf:.2%}\n\n"
                     f"Provide 2-3 brief insights (one per line):")
        ])
        
        chain = prompt | self.llm
        insights = []
        
        try:
            response = chain.invoke({})
            content = response.content if isinstance(response.content, str) else str(response.content)
            lines = content.strip().split('\n')
            insights = [line.strip('- ').strip() for line in lines if line.strip()][:3]
        except:
            insights = [
                f"Improved from {initial_perf:.2%} to {final_perf:.2%}",
                f"Required {steps} adaptation steps",
                f"Strategy {self.strategy.value} was effective"
            ]
        
        return insights
    
    def _update_meta_knowledge(self, experience: LearningExperience):
        """Update meta-knowledge based on learning experience."""
        strategy_key = experience.strategy_used
        
        if strategy_key not in self.meta_knowledge:
            self.meta_knowledge[strategy_key] = MetaKnowledge(
                strategy=strategy_key,
                effectiveness=0.0,
                best_domains=[],
                optimal_learning_rate=self.optimal_learning_rate,
                optimal_batch_size=self.optimal_batch_size,
                learned_from_tasks=0,
                confidence=0.0
            )
        
        meta = self.meta_knowledge[strategy_key]
        
        # Update effectiveness (moving average)
        improvement = experience.final_performance - experience.initial_performance
        meta.effectiveness = (
            0.9 * meta.effectiveness + 0.1 * improvement
        ) if meta.learned_from_tasks > 0 else improvement
        
        meta.learned_from_tasks += 1
        meta.confidence = min(1.0, meta.learned_from_tasks / 20.0)
        
        # Optimize learning rate based on experience
        if improvement > 0.3:
            # Good improvement, slightly increase LR
            self.optimal_learning_rate = min(0.1, self.optimal_learning_rate * 1.05)
        elif improvement  Dict[str, Any]:
        """Rapidly adapt to a new task using meta-learned knowledge."""
        if not self.meta_knowledge:
            print(f"   {self.name}: No meta-knowledge yet. Learning from scratch...")
            experience = self._learn_task(task)
            return {
                "task": task.name,
                "initial_performance": experience.initial_performance,
                "final_performance": experience.final_performance,
                "steps": experience.steps_taken,
                "used_meta_knowledge": False
            }
        
        print(f"\n   {self.name}: Adapting to new task '{task.name}' using meta-knowledge...")
        
        # Use meta-knowledge for faster adaptation
        best_meta = max(self.meta_knowledge.values(), key=lambda m: m.effectiveness)
        
        # Apply optimal hyperparameters
        old_lr = self.optimal_learning_rate
        old_steps = self.adaptation_steps
        
        self.optimal_learning_rate = best_meta.optimal_learning_rate
        self.adaptation_steps = max(2, self.adaptation_steps // 2)  # Faster adaptation
        
        # Learn with meta-knowledge
        experience = self._learn_task(task)
        
        # Restore
        self.optimal_learning_rate = old_lr
        self.adaptation_steps = old_steps
        
        print(f"   Adapted in {experience.steps_taken} steps "
              f"(improvement: {experience.final_performance - experience.initial_performance:.2%})")
        
        return {
            "task": task.name,
            "initial_performance": experience.initial_performance,
            "final_performance": experience.final_performance,
            "steps": experience.steps_taken,
            "used_meta_knowledge": True,
            "strategy": best_meta.strategy,
            "insights": experience.insights
        }
    
    def get_meta_insights(self) -> str:
        """Get human-readable meta-learning insights."""
        if not self.learning_experiences:
            return "No learning experiences yet."
        
        insights = []
        insights.append(f"Meta-learned from {self.tasks_learned} tasks using {self.strategy.value}")
        insights.append(f"Average improvement: {self.statistics['avg_improvement']:.2%}")
        insights.append(f"Optimal learning rate: {self.optimal_learning_rate:.4f}")
        insights.append(f"Optimal adaptation steps: {self.adaptation_steps}")
        
        if self.meta_knowledge:
            insights.append(f"\nMeta-knowledge acquired:")
            for strategy, meta in self.meta_knowledge.items():
                insights.append(f"  - {strategy}: effectiveness {meta.effectiveness:.2%} "
                              f"(confidence: {meta.confidence:.2%})")
        
        return "\n".join(insights)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get comprehensive statistics."""
        return {
            "name": self.name,
            "strategy": self.strategy.value,
            "tasks_learned": self.tasks_learned,
            "meta_updates": self.meta_updates,
            "optimal_learning_rate": self.optimal_learning_rate,
            "optimal_batch_size": self.optimal_batch_size,
            "adaptation_steps": self.adaptation_steps,
            **self.statistics
        }


class MetaLearningPattern:
    """
    Meta-Learning Pattern - Learn how to learn.
    
    Coordinates meta-learning agents to optimize learning strategies
    across multiple tasks.
    """
    
    def __init__(self):
        self.agents: Dict[str, MetaLearningAgent] = {}
        self.training_tasks: List[Task] = []
        self.test_tasks: List[Task] = []
    
    def register_agent(self, agent: MetaLearningAgent):
        """Register meta-learning agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name} ({agent.strategy.value})")
    
    def add_training_tasks(self, tasks: List[Task]):
        """Add tasks for meta-training."""
        self.training_tasks.extend(tasks)
        print(f"   Added {len(tasks)} training tasks (total: {len(self.training_tasks)})")
    
    def add_test_tasks(self, tasks: List[Task]):
        """Add tasks for testing adaptation."""
        self.test_tasks.extend(tasks)
        print(f"   Added {len(tasks)} test tasks")
    
    def meta_train_all(self):
        """Meta-train all agents."""
        print("\n   Meta-training all agents...")
        for agent in self.agents.values():
            agent.meta_train(self.training_tasks)
    
    def test_adaptation(self) -> Dict[str, List[Dict[str, Any]]]:
        """Test rapid adaptation on new tasks."""
        print("\n   Testing adaptation on new tasks...")
        
        results = {}
        for agent_name, agent in self.agents.items():
            agent_results = []
            for task in self.test_tasks:
                result = agent.adapt_to_new_task(task)
                agent_results.append(result)
            results[agent_name] = agent_results
        
        return results
    
    def run(self, 
            training_tasks: List[Task],
            test_tasks: List[Task]) -> Dict[str, Any]:
        """Run meta-learning pipeline."""
        print("=== Meta-Learning Pattern Demo ===\n")
        
        # Add tasks
        print("1. Adding tasks...")
        self.add_training_tasks(training_tasks)
        self.add_test_tasks(test_tasks)
        
        # Meta-train
        print("\n2. Meta-training...")
        self.meta_train_all()
        
        # Test adaptation
        print("\n3. Testing rapid adaptation...")
        adaptation_results = self.test_adaptation()
        
        # Get insights
        print("\n4. Extracting meta-insights...")
        insights = {name: agent.get_meta_insights() 
                   for name, agent in self.agents.items()}
        
        # Get statistics
        stats = {name: agent.get_statistics() 
                for name, agent in self.agents.items()}
        
        return {
            "pattern": "MetaLearning",
            "agents": len(self.agents),
            "training_tasks": len(training_tasks),
            "test_tasks": len(test_tasks),
            "adaptation_results": adaptation_results,
            "insights": insights,
            "statistics": stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Meta-Learning Pattern Demo ===\n")
    
    # Create pattern
    pattern = MetaLearningPattern()
    
    # Create agents with different strategies
    print("1. Creating meta-learning agents...")
    maml_agent = MetaLearningAgent(
        name="MAML_Agent",
        strategy=MetaStrategy.MAML
    )
    pattern.register_agent(maml_agent)
    
    reptile_agent = MetaLearningAgent(
        name="Reptile_Agent",
        strategy=MetaStrategy.REPTILE
    )
    pattern.register_agent(reptile_agent)
    
    # Create training tasks (diverse domains)
    print("\n2. Creating training tasks...")
    training_tasks = [
        Task(id="train_1", name="Sentiment Classification", domain="text", 
             examples=[{"text": "good", "label": "pos"}]),
        Task(id="train_2", name="Spam Detection", domain="text",
             examples=[{"text": "buy now", "label": "spam"}]),
        Task(id="train_3", name="Image Classification", domain="vision",
             examples=[{"image": "cat.jpg", "label": "cat"}]),
        Task(id="train_4", name="Language Translation", domain="text",
             examples=[{"en": "hello", "fr": "bonjour"}]),
        Task(id="train_5", name="Recommendation", domain="recommendation",
             examples=[{"user": "1", "item": "A", "rating": 5}]),
        Task(id="train_6", name="Named Entity Recognition", domain="text",
             examples=[{"text": "John lives in NYC", "entities": ["John", "NYC"]}]),
        Task(id="train_7", name="Question Answering", domain="text",
             examples=[{"q": "What is AI?", "a": "Artificial Intelligence"}]),
        Task(id="train_8", name="Regression", domain="numeric",
             examples=[{"x": 1, "y": 2}]),
    ]
    
    # Create test tasks (new but similar domains)
    print("\n3. Creating test tasks...")
    test_tasks = [
        Task(id="test_1", name="Review Classification", domain="text",
             examples=[{"text": "amazing product", "label": "positive"}]),
        Task(id="test_2", name="Intent Detection", domain="text",
             examples=[{"text": "book a flight", "label": "booking"}]),
    ]
    
    # Run meta-learning
    print("\n4. Running meta-learning pipeline...")
    summary = pattern.run(training_tasks, test_tasks)
    
    print(f"\n5. Results:")
    print(f"   Agents: {summary['agents']}")
    print(f"   Training Tasks: {summary['training_tasks']}")
    print(f"   Test Tasks: {summary['test_tasks']}")
    
    print(f"\n6. Adaptation Results:")
    for agent_name, results in summary['adaptation_results'].items():
        print(f"\n   {agent_name}:")
        for result in results:
            print(f"     {result['task']}: {result['steps']} steps, "
                  f"improvement: {result['final_performance'] - result['initial_performance']:.2%}")
    
    print(f"\n7. Meta-Insights:")
    for agent_name, insight in summary['insights'].items():
        print(f"\n   {agent_name}:")
        for line in insight.split('\n'):
            print(f"   {line}")
    
    print(f"\n8. 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✓ Meta-learning enabled rapid adaptation to new tasks!")

    

Agentic Architecture Pattern: Meta Learning

Pattern 57: 57 Meta Learning – Line-by-Line Explanation

Overview

Pattern 57: Meta-Learning

Purpose: Learn how to learn – optimize learning algorithms and strategies themselves. “Learning to learn” by improving the learning process across multiple tasks.

Code Structure Analysis

File Information

• Pattern Number: 57
• Pattern Name: 57 Meta Learning
• File Path: patterns\learning_adaptation\57_meta_learning.py
• Total Lines: 504

Detailed Code Explanation

Module Docstring

"""
Pattern 57: Meta-Learning

Purpose: Learn how to learn - optimize learning algorithms and strategies themselves.
"Learning to learn" by improving the learning process across multiple tasks.
"""

Explanation: This module docstring describes the purpose and key features of the 57 Meta Learning 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

MetaStrategy (Line 17)

class MetaStrategy(Enum):
    MAML
    REPTILE
    LEARNING_TO_COMPARE
    OPTIMIZATION_AS_MODEL

Purpose: Meta-learning strategies.

Values: – MAML: Represents maml – REPTILE: Represents reptile – LEARNING_TO_COMPARE: Represents learning to compare – OPTIMIZATION_AS_MODEL: Represents optimization as model

Classes

MetaStrategy (Line 17)

Description: Meta-learning strategies.

Task (Line 26)

Description: A learning task.

LearningExperience (Line 37)

Description: Experience from learning a task.

MetaKnowledge (Line 50)

Description: Meta-level knowledge about learning.

MetaLearningAgent (Line 61)

Description: Meta-Learning Agent – Learn how to learn.

Optimizes the learning process itself by learning from experience across multiple tasks. Discovers optimal learning strategies, hyperparameters, and approaches.

Methods:

__init__ (Line 70)

def __init__(self, name, strategy):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – strategy: Parameter for strategy

meta_train (Line 98)

def meta_train(self, tasks):
    """Meta-train on multiple tasks to learn how to learn."""

Purpose: Meta-train on multiple tasks to learn how to learn.

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

_learn_task (Line 120)

def _learn_task(self, task):
    """Learn a single task and record the experience."""

Purpose: Learn a single task and record the experience.

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

_generate_learning_insights (Line 166)

def _generate_learning_insights(self, task, initial_perf, final_perf, steps):
    """Generate insights about the learning process."""

Purpose: Generate insights about the learning process.

Parameters: – self: Instance reference – task: Parameter for task – initial_perf: Parameter for initial_perf – final_perf: Parameter for final_perf – steps: Parameter for steps

_update_meta_knowledge (Line 201)

def _update_meta_knowledge(self, experience):
    """Update meta-knowledge based on learning experience."""

Purpose: Update meta-knowledge based on learning experience.

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

_extract_meta_insights (Line 235)

def _extract_meta_insights(self):
    """Extract high-level meta-learning insights."""

Purpose: Extract high-level meta-learning insights.

Parameters: – self: Instance reference

adapt_to_new_task (Line 255)

def adapt_to_new_task(self, task):
    """Rapidly adapt to a new task using meta-learned knowledge."""

Purpose: Rapidly adapt to a new task using meta-learned knowledge.

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

get_meta_insights (Line 300)

def get_meta_insights(self):
    """Get human-readable meta-learning insights."""

Purpose: Get human-readable meta-learning insights.

Parameters: – self: Instance reference

get_statistics (Line 319)

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

Purpose: Get comprehensive statistics.

Parameters: – self: Instance reference

MetaLearningPattern (Line 333)

Description: Meta-Learning Pattern – Learn how to learn.

Coordinates meta-learning agents to optimize learning strategies across multiple tasks.

Methods:

__init__ (Line 341)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 346)

def register_agent(self, agent):
    """Register meta-learning agent."""

Purpose: Register meta-learning agent.

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

add_training_tasks (Line 351)

def add_training_tasks(self, tasks):
    """Add tasks for meta-training."""

Purpose: Add tasks for meta-training.

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

add_test_tasks (Line 356)

def add_test_tasks(self, tasks):
    """Add tasks for testing adaptation."""

Purpose: Add tasks for testing adaptation.

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

meta_train_all (Line 361)

def meta_train_all(self):
    """Meta-train all agents."""

Purpose: Meta-train all agents.

Parameters: – self: Instance reference

test_adaptation (Line 367)

def test_adaptation(self):
    """Test rapid adaptation on new tasks."""

Purpose: Test rapid adaptation on new tasks.

Parameters: – self: Instance reference

run (Line 381)

def run(self, training_tasks, test_tasks):
    """Run meta-learning pipeline."""

Purpose: Run meta-learning pipeline.

Parameters: – self: Instance reference – training_tasks: Parameter for training_tasks – test_tasks: Parameter for test_tasks

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

    """
Pattern 56: Offline Learning

Purpose: Learn from complete datasets in batch mode before deployment.
Trains on static datasets and evaluates comprehensively before production use.
"""

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 BatchStrategy(Enum):
    """Strategies for batch learning."""
    FULL_BATCH = "full_batch"  # Train on entire dataset at once
    MINI_BATCH = "mini_batch"  # Train on small batches with multiple epochs
    STOCHASTIC = "stochastic"  # Train on random samples
    STRATIFIED = "stratified"  # Ensure balanced class distribution


@dataclass
class TrainingExample:
    """A training example."""
    id: str
    features: Dict[str, Any]
    label: str
    weight: float = 1.0
    fold: Optional[int] = None  # For cross-validation
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class EpochMetrics:
    """Metrics for a training epoch."""
    epoch: int
    train_loss: float
    train_accuracy: float
    val_loss: float
    val_accuracy: float
    learning_rate: float
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ModelCheckpoint:
    """Saved model checkpoint."""
    epoch: int
    accuracy: float
    loss: float
    model_state: Dict[str, Any]
    timestamp: datetime = field(default_factory=datetime.now)


class OfflineLearningAgent:
    """
    Offline Learning Agent - Learn from complete datasets in batch mode.
    
    Trains on static datasets with multiple epochs, proper train/val/test
    splits, and comprehensive evaluation before deployment.
    """
    
    def __init__(self, 
                 name: str,
                 task_description: str,
                 strategy: BatchStrategy = BatchStrategy.MINI_BATCH,
                 batch_size: int = 32,
                 epochs: int = 10,
                 learning_rate: float = 0.01):
        self.name = name
        self.task_description = task_description
        self.strategy = strategy
        self.batch_size = batch_size
        self.epochs = epochs
        self.learning_rate = learning_rate
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Datasets
        self.train_set: List[TrainingExample] = []
        self.val_set: List[TrainingExample] = []
        self.test_set: List[TrainingExample] = []
        
        # Training state
        self.model_patterns: Dict[str, Any] = {}
        self.trained: bool = False
        self.current_epoch: int = 0
        
        # Training history
        self.epoch_history: List[EpochMetrics] = []
        self.checkpoints: List[ModelCheckpoint] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_examples": 0,
            "training_time": 0.0,
            "best_val_accuracy": 0.0,
            "final_test_accuracy": 0.0
        }
    
    def prepare_data(self, 
                    examples: List[TrainingExample],
                    train_ratio: float = 0.7,
                    val_ratio: float = 0.15):
        """Split data into train/val/test sets."""
        # Shuffle data
        shuffled = examples.copy()
        random.shuffle(shuffled)
        
        # Calculate split points
        n = len(shuffled)
        train_end = int(n * train_ratio)
        val_end = int(n * (train_ratio + val_ratio))
        
        # Split data
        self.train_set = shuffled[:train_end]
        self.val_set = shuffled[train_end:val_end]
        self.test_set = shuffled[val_end:]
        
        self.statistics["total_examples"] = n
        
        print(f"   {self.name}: Data prepared - "
              f"Train: {len(self.train_set)}, Val: {len(self.val_set)}, Test: {len(self.test_set)}")
    
    def train(self):
        """Train model on complete dataset."""
        if not self.train_set:
            print(f"   {self.name}: No training data!")
            return
        
        print(f"\n   {self.name}: Training with {self.strategy.value} strategy...")
        start_time = datetime.now()
        
        for epoch in range(self.epochs):
            self.current_epoch = epoch + 1
            
            # Train epoch
            train_loss, train_acc = self._train_epoch()
            
            # Validate
            val_loss, val_acc = self._validate()
            
            # Record metrics
            metrics = EpochMetrics(
                epoch=self.current_epoch,
                train_loss=train_loss,
                train_accuracy=train_acc,
                val_loss=val_loss,
                val_accuracy=val_acc,
                learning_rate=self.learning_rate
            )
            self.epoch_history.append(metrics)
            
            # Save checkpoint if best so far
            if val_acc > self.statistics["best_val_accuracy"]:
                self.statistics["best_val_accuracy"] = val_acc
                self._save_checkpoint(val_acc, val_loss)
            
            # Progress update
            if (epoch + 1) % max(1, self.epochs // 5) == 0:
                print(f"   Epoch {self.current_epoch}/{self.epochs}: "
                      f"Train Acc: {train_acc:.2%}, Val Acc: {val_acc:.2%}")
        
        self.trained = True
        self.statistics["training_time"] = (datetime.now() - start_time).total_seconds()
        
        print(f"   Training complete! Best val accuracy: {self.statistics['best_val_accuracy']:.2%}")
    
    def _train_epoch(self) -> Tuple[float, float]:
        """Train for one epoch."""
        # Prepare batches based on strategy
        if self.strategy == BatchStrategy.FULL_BATCH:
            batches = [self.train_set]
        elif self.strategy == BatchStrategy.MINI_BATCH:
            batches = self._create_mini_batches(self.train_set, self.batch_size)
        elif self.strategy == BatchStrategy.STOCHASTIC:
            batches = [[ex] for ex in self.train_set]  # One example per batch
        else:  # STRATIFIED
            batches = self._create_stratified_batches(self.train_set, self.batch_size)
        
        # Train on batches
        total_loss = 0.0
        correct = 0
        total = 0
        
        for batch in batches:
            # Extract patterns from batch using LLM
            patterns = self._extract_batch_patterns(batch)
            
            # Update model patterns
            for pattern_id, pattern_data in patterns.items():
                if pattern_id in self.model_patterns:
                    # Average with existing pattern
                    self.model_patterns[pattern_id]["confidence"] = (
                        0.9 * self.model_patterns[pattern_id]["confidence"] + 
                        0.1 * pattern_data["confidence"]
                    )
                else:
                    self.model_patterns[pattern_id] = pattern_data
            
            # Simulate training metrics
            batch_loss = random.uniform(0.1, 0.5) * (1.0 - self.current_epoch / self.epochs)
            total_loss += batch_loss * len(batch)
            
            # Predict on batch
            for ex in batch:
                pred = self._predict_internal(ex)
                if pred == ex.label:
                    correct += 1
                total += 1
        
        avg_loss = total_loss / total if total > 0 else 0.0
        accuracy = correct / total if total > 0 else 0.0
        
        return avg_loss, accuracy
    
    def _validate(self) -> Tuple[float, float]:
        """Validate on validation set."""
        if not self.val_set:
            return 0.0, 0.0
        
        correct = 0
        total_loss = 0.0
        
        for ex in self.val_set:
            pred = self._predict_internal(ex)
            if pred == ex.label:
                correct += 1
            
            # Simulate loss
            total_loss += random.uniform(0.1, 0.4)
        
        accuracy = correct / len(self.val_set)
        avg_loss = total_loss / len(self.val_set)
        
        return avg_loss, accuracy
    
    def test(self) -> Dict[str, float]:
        """Test on test set."""
        if not self.test_set:
            print(f"   {self.name}: No test data!")
            return {"accuracy": 0.0}
        
        if not self.trained:
            print(f"   {self.name}: Model not trained yet!")
            return {"accuracy": 0.0}
        
        print(f"\n   {self.name}: Testing on {len(self.test_set)} examples...")
        
        correct = 0
        predictions = []
        
        for ex in self.test_set:
            pred = self.predict(ex)
            predictions.append(pred)
            if pred["predicted_label"] == ex.label:
                correct += 1
        
        accuracy = correct / len(self.test_set)
        self.statistics["final_test_accuracy"] = accuracy
        
        print(f"   Test Accuracy: {accuracy:.2%}")
        
        return {
            "accuracy": accuracy,
            "correct": correct,
            "total": len(self.test_set),
            "predictions": predictions
        }
    
    def _predict_internal(self, example: TrainingExample) -> str:
        """Internal prediction without full response."""
        if not self.model_patterns:
            return "unknown"
        
        # Simple prediction based on learned patterns
        # Use most confident pattern
        if self.model_patterns:
            best_pattern = max(self.model_patterns.values(), 
                             key=lambda p: p.get("confidence", 0))
            # Extract label from pattern description if possible
            if "label" in best_pattern:
                return best_pattern["label"]
        
        return "unknown"
    
    def predict(self, example: TrainingExample) -> Dict[str, Any]:
        """Make prediction on example."""
        if not self.trained:
            return {
                "example_id": example.id,
                "predicted_label": "untrained",
                "confidence": 0.0
            }
        
        # Create prompt with learned patterns
        patterns_text = ""
        for pattern_id, pattern_data in list(self.model_patterns.items())[:5]:
            patterns_text += f"  - {pattern_data.get('description', pattern_id)}\n"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are performing {self.task_description}. "
                      f"You have been trained on a dataset and learned these patterns:\n{patterns_text}"),
            ("human", f"Example: {example.features}\n\n"
                     f"Predict the label and confidence (0.0-1.0).\n"
                     f"Format: LABEL|CONFIDENCE")
        ])
        
        chain = prompt | self.llm
        
        try:
            response = chain.invoke({})
            content = response.content if isinstance(response.content, str) else str(response.content)
            parts = content.strip().split('|')
            
            predicted_label = parts[0].strip()
            confidence = float(parts[1].strip()) if len(parts) > 1 else 0.7
        except:
            predicted_label = self._predict_internal(example)
            confidence = 0.6
        
        return {
            "example_id": example.id,
            "predicted_label": predicted_label,
            "actual_label": example.label,
            "confidence": confidence,
            "correct": predicted_label == example.label
        }
    
    def _extract_batch_patterns(self, batch: List[TrainingExample]) -> Dict[str, Any]:
        """Extract patterns from a batch of examples."""
        if not batch:
            return {}
        
        # Group by label
        label_groups = {}
        for ex in batch:
            if ex.label not in label_groups:
                label_groups[ex.label] = []
            label_groups[ex.label].append(ex)
        
        patterns = {}
        for label, examples in label_groups.items():
            pattern_id = f"label_{label}"
            patterns[pattern_id] = {
                "description": f"Pattern for label {label}",
                "label": label,
                "confidence": len(examples) / len(batch),
                "support": len(examples)
            }
        
        return patterns
    
    def _create_mini_batches(self, 
                            examples: List[TrainingExample], 
                            batch_size: int) -> List[List[TrainingExample]]:
        """Create mini-batches."""
        batches = []
        for i in range(0, len(examples), batch_size):
            batches.append(examples[i:i + batch_size])
        return batches
    
    def _create_stratified_batches(self, 
                                  examples: List[TrainingExample],
                                  batch_size: int) -> List[List[TrainingExample]]:
        """Create stratified batches with balanced class distribution."""
        # Group by label
        label_groups = {}
        for ex in examples:
            if ex.label not in label_groups:
                label_groups[ex.label] = []
            label_groups[ex.label].append(ex)
        
        # Create balanced batches
        batches = []
        max_batches = len(examples) // batch_size
        
        for batch_idx in range(max_batches):
            batch = []
            for label, examples in label_groups.items():
                # Add examples from each class
                start_idx = (batch_idx * batch_size) // len(label_groups)
                end_idx = start_idx + batch_size // len(label_groups)
                batch.extend(examples[start_idx:end_idx])
            
            if batch:
                batches.append(batch[:batch_size])
        
        return batches
    
    def _save_checkpoint(self, accuracy: float, loss: float):
        """Save model checkpoint."""
        checkpoint = ModelCheckpoint(
            epoch=self.current_epoch,
            accuracy=accuracy,
            loss=loss,
            model_state={
                "patterns": self.model_patterns.copy(),
                "epoch": self.current_epoch
            }
        )
        self.checkpoints.append(checkpoint)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get comprehensive training statistics."""
        return {
            "name": self.name,
            "strategy": self.strategy.value,
            "trained": self.trained,
            "epochs_completed": self.current_epoch,
            "train_size": len(self.train_set),
            "val_size": len(self.val_set),
            "test_size": len(self.test_set),
            "patterns_learned": len(self.model_patterns),
            "checkpoints_saved": len(self.checkpoints),
            **self.statistics
        }


class OfflineLearningPattern:
    """
    Offline Learning Pattern - Batch learning from complete datasets.
    
    Coordinates multiple agents training on static datasets with
    different batch strategies.
    """
    
    def __init__(self):
        self.agents: Dict[str, OfflineLearningAgent] = {}
        self.dataset: List[TrainingExample] = []
    
    def register_agent(self, agent: OfflineLearningAgent):
        """Register offline learning agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name} ({agent.strategy.value})")
    
    def load_dataset(self, examples: List[TrainingExample]):
        """Load dataset for all agents."""
        self.dataset = examples
        print(f"   Loaded dataset: {len(examples)} examples")
        
        # Prepare data for each agent
        for agent in self.agents.values():
            agent.prepare_data(examples)
    
    def train_all(self):
        """Train all agents."""
        print("\n   Training all agents...")
        for agent in self.agents.values():
            agent.train()
    
    def test_all(self) -> Dict[str, Dict[str, float]]:
        """Test all agents."""
        print("\n   Testing all agents...")
        results = {}
        for name, agent in self.agents.items():
            results[name] = agent.test()
        return results
    
    def run(self, examples: List[TrainingExample]) -> Dict[str, Any]:
        """Run offline learning pipeline."""
        print("=== Offline Learning Pattern Demo ===\n")
        
        # Load dataset
        print("1. Loading dataset...")
        self.load_dataset(examples)
        
        # Train all agents
        print("\n2. Training agents...")
        self.train_all()
        
        # Test all agents
        print("\n3. Testing agents...")
        test_results = self.test_all()
        
        # Get statistics
        stats = {name: agent.get_statistics() 
                for name, agent in self.agents.items()}
        
        return {
            "pattern": "OfflineLearning",
            "agents": len(self.agents),
            "dataset_size": len(examples),
            "test_results": test_results,
            "statistics": stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Offline Learning Pattern Demo ===\n")
    
    # Create pattern
    pattern = OfflineLearningPattern()
    
    # Create agents with different strategies
    print("1. Creating agents...")
    full_batch_agent = OfflineLearningAgent(
        name="FullBatchAgent",
        task_description="classify sentiment as positive or negative",
        strategy=BatchStrategy.FULL_BATCH,
        epochs=5
    )
    pattern.register_agent(full_batch_agent)
    
    mini_batch_agent = OfflineLearningAgent(
        name="MiniBatchAgent",
        task_description="classify sentiment as positive or negative",
        strategy=BatchStrategy.MINI_BATCH,
        batch_size=16,
        epochs=10
    )
    pattern.register_agent(mini_batch_agent)
    
    # Generate dataset
    print("\n2. Generating dataset...")
    dataset = []
    
    # Positive examples
    for i in range(80):
        dataset.append(TrainingExample(
            id=f"pos_{i}",
            features={"text": f"positive_example_{i}", "score": random.uniform(0.6, 1.0)},
            label="positive"
        ))
    
    # Negative examples
    for i in range(80):
        dataset.append(TrainingExample(
            id=f"neg_{i}",
            features={"text": f"negative_example_{i}", "score": random.uniform(0.0, 0.4)},
            label="negative"
        ))
    
    # Run offline learning
    print("\n3. Running offline learning...")
    summary = pattern.run(dataset)
    
    print(f"\n4. Results:")
    print(f"   Agents: {summary['agents']}")
    print(f"   Dataset Size: {summary['dataset_size']}")
    
    print(f"\n5. Test Results:")
    for agent_name, results in summary['test_results'].items():
        print(f"\n   {agent_name}:")
        print(f"     Accuracy: {results['accuracy']:.2%}")
        print(f"     Correct: {results['correct']}/{results['total']}")
    
    print(f"\n6. Training 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✓ Offline learning completed comprehensive training and evaluation!")

    

Agentic Architecture Pattern: Offline Learning

Pattern 56: 56 Offline Learning – Line-by-Line Explanation

Overview

Pattern 56: Offline Learning

Purpose: Learn from complete datasets in batch mode before deployment. Trains on static datasets and evaluates comprehensively before production use.

Code Structure Analysis

File Information

• Pattern Number: 56
• Pattern Name: 56 Offline Learning
• File Path: patterns\learning_adaptation\56_offline_learning.py
• Total Lines: 562

Detailed Code Explanation

Module Docstring

"""
Pattern 56: Offline Learning

Purpose: Learn from complete datasets in batch mode before deployment.
Trains on static datasets and evaluates comprehensively before production use.
"""

Explanation: This module docstring describes the purpose and key features of the 56 Offline Learning 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

BatchStrategy (Line 17)

class BatchStrategy(Enum):
    FULL_BATCH
    MINI_BATCH
    STOCHASTIC
    STRATIFIED

Purpose: Strategies for batch learning.

Values: – FULL_BATCH: Represents full batch – MINI_BATCH: Represents mini batch – STOCHASTIC: Represents stochastic – STRATIFIED: Represents stratified

Classes

BatchStrategy (Line 17)

Description: Strategies for batch learning.

TrainingExample (Line 26)

Description: A training example.

EpochMetrics (Line 37)

Description: Metrics for a training epoch.

ModelCheckpoint (Line 49)

Description: Saved model checkpoint.

OfflineLearningAgent (Line 58)

Description: Offline Learning Agent – Learn from complete datasets in batch mode.

Trains on static datasets with multiple epochs, proper train/val/test splits, and comprehensive evaluation before deployment.

Methods:

__init__ (Line 66)

def __init__(self, name, task_description, strategy, batch_size, epochs, learning_rate):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – task_description: Parameter for task_description – strategy: Parameter for strategy – batch_size: Parameter for batch_size – epochs: Parameter for epochs – learning_rate: Parameter for learning_rate

prepare_data (Line 103)

def prepare_data(self, examples, train_ratio, val_ratio):
    """Split data into train/val/test sets."""

Purpose: Split data into train/val/test sets.

Parameters: – self: Instance reference – examples: Parameter for examples – train_ratio: Parameter for train_ratio – val_ratio: Parameter for val_ratio

train (Line 127)

def train(self):
    """Train model on complete dataset."""

Purpose: Train model on complete dataset.

Parameters: – self: Instance reference

_train_epoch (Line 171)

def _train_epoch(self):
    """Train for one epoch."""

Purpose: Train for one epoch.

Parameters: – self: Instance reference

_validate (Line 219)

def _validate(self):
    """Validate on validation set."""

Purpose: Validate on validation set.

Parameters: – self: Instance reference

test (Line 240)

def test(self):
    """Test on test set."""

Purpose: Test on test set.

Parameters: – self: Instance reference

_predict_internal (Line 273)

def _predict_internal(self, example):
    """Internal prediction without full response."""

Purpose: Internal prediction without full response.

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

predict (Line 289)

def predict(self, example):
    """Make prediction on example."""

Purpose: Make prediction on example.

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

_extract_batch_patterns (Line 332)

def _extract_batch_patterns(self, batch):
    """Extract patterns from a batch of examples."""

Purpose: Extract patterns from a batch of examples.

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

_create_mini_batches (Line 356)

def _create_mini_batches(self, examples, batch_size):
    """Create mini-batches."""

Purpose: Create mini-batches.

Parameters: – self: Instance reference – examples: Parameter for examples – batch_size: Parameter for batch_size

_create_stratified_batches (Line 365)

def _create_stratified_batches(self, examples, batch_size):
    """Create stratified batches with balanced class distribution."""

Purpose: Create stratified batches with balanced class distribution.

Parameters: – self: Instance reference – examples: Parameter for examples – batch_size: Parameter for batch_size

_save_checkpoint (Line 393)

def _save_checkpoint(self, accuracy, loss):
    """Save model checkpoint."""

Purpose: Save model checkpoint.

Parameters: – self: Instance reference – accuracy: Parameter for accuracy – loss: Parameter for loss

get_statistics (Line 406)

def get_statistics(self):
    """Get comprehensive training statistics."""

Purpose: Get comprehensive training statistics.

Parameters: – self: Instance reference

OfflineLearningPattern (Line 422)

Description: Offline Learning Pattern – Batch learning from complete datasets.

Coordinates multiple agents training on static datasets with different batch strategies.

Methods:

__init__ (Line 430)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 434)

def register_agent(self, agent):
    """Register offline learning agent."""

Purpose: Register offline learning agent.

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

load_dataset (Line 439)

def load_dataset(self, examples):
    """Load dataset for all agents."""

Purpose: Load dataset for all agents.

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

train_all (Line 448)

def train_all(self):
    """Train all agents."""

Purpose: Train all agents.

Parameters: – self: Instance reference

test_all (Line 454)

def test_all(self):
    """Test all agents."""

Purpose: Test all agents.

Parameters: – self: Instance reference

run (Line 462)

def run(self, examples):
    """Run offline learning pipeline."""

Purpose: Run offline learning pipeline.

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

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

    """
Pattern 55: Online Learning

Purpose: Learn continuously from streaming data in real-time.
Updates model incrementally as new data arrives, adapting to changing patterns.
"""

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


class UpdateStrategy(Enum):
    """Strategies for online model updates."""
    IMMEDIATE = "immediate"  # Update after each example
    MINI_BATCH = "mini_batch"  # Update after small batches
    SLIDING_WINDOW = "sliding_window"  # Maintain recent examples only
    ADAPTIVE = "adaptive"  # Adjust based on performance


@dataclass
class StreamingExample:
    """A single streaming data example."""
    id: str
    features: Dict[str, Any]
    label: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)
    weight: float = 1.0
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class ModelSnapshot:
    """Snapshot of model state at a point in time."""
    timestamp: datetime
    examples_seen: int
    accuracy: float
    parameters: Dict[str, Any]
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class PerformanceMetrics:
    """Real-time performance metrics."""
    accuracy: float
    recent_accuracy: float  # Last N examples
    examples_processed: int
    updates_performed: int
    avg_confidence: float
    timestamp: datetime = field(default_factory=datetime.now)


class OnlineLearningAgent:
    """
    Online Learning Agent - Learn continuously from streaming data.
    
    Adapts in real-time as new examples arrive, maintaining performance
    on non-stationary data streams. Uses LLM for pattern recognition.
    """
    
    def __init__(self, 
                 name: str,
                 task_description: str,
                 strategy: UpdateStrategy = UpdateStrategy.IMMEDIATE,
                 window_size: int = 100,
                 batch_size: int = 10):
        self.name = name
        self.task_description = task_description
        self.strategy = strategy
        self.window_size = window_size
        self.batch_size = batch_size
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Streaming data buffer
        self.data_buffer: Deque[StreamingExample] = deque(maxlen=window_size)
        self.batch_buffer: List[StreamingExample] = []
        
        # Model state
        self.learned_patterns: Dict[str, Any] = {}
        self.examples_seen: int = 0
        self.updates_performed: int = 0
        
        # Performance tracking
        self.predictions: List[Dict[str, Any]] = []
        self.snapshots: List[ModelSnapshot] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "correct_predictions": 0,
            "total_predictions": 0,
            "recent_correct": 0,
            "recent_total": 0
        }
    
    def process_stream(self, example: StreamingExample):
        """Process a single streaming example."""
        self.examples_seen += 1
        
        # Add to buffer
        if self.strategy == UpdateStrategy.SLIDING_WINDOW:
            self.data_buffer.append(example)
        else:
            self.data_buffer.append(example)
            self.batch_buffer.append(example)
        
        # Decide when to update
        should_update = False
        
        if self.strategy == UpdateStrategy.IMMEDIATE:
            should_update = True
        elif self.strategy == UpdateStrategy.MINI_BATCH:
            should_update = len(self.batch_buffer) >= self.batch_size
        elif self.strategy == UpdateStrategy.ADAPTIVE:
            # Update when performance drops
            recent_accuracy = self._calculate_recent_accuracy()
            should_update = recent_accuracy = self.batch_size
        
        # Perform update
        if should_update:
            self._update_model()
            if self.strategy != UpdateStrategy.SLIDING_WINDOW:
                self.batch_buffer.clear()
        
        print(f"   {self.name}: Processed example {example.id} (total: {self.examples_seen})")
    
    def _update_model(self):
        """Update model with current buffer."""
        if not self.data_buffer:
            return
        
        self.updates_performed += 1
        
        # Get recent examples for context
        recent_examples = list(self.data_buffer)[-min(20, len(self.data_buffer)):]
        
        # Extract patterns using LLM
        patterns = self._extract_patterns(recent_examples)
        
        # Update learned patterns
        for pattern_id, pattern_data in patterns.items():
            if pattern_id in self.learned_patterns:
                # Merge with existing pattern
                self.learned_patterns[pattern_id]["confidence"] = (
                    0.7 * self.learned_patterns[pattern_id]["confidence"] + 
                    0.3 * pattern_data["confidence"]
                )
                self.learned_patterns[pattern_id]["occurrences"] += 1
            else:
                self.learned_patterns[pattern_id] = pattern_data
                self.learned_patterns[pattern_id]["occurrences"] = 1
        
        # Take snapshot periodically
        if self.updates_performed % 10 == 0:
            self._take_snapshot()
        
        print(f"   {self.name}: Model updated (update #{self.updates_performed}, "
              f"{len(self.learned_patterns)} patterns)")
    
    def _extract_patterns(self, examples: List[StreamingExample]) -> Dict[str, Any]:
        """Extract patterns from examples using LLM."""
        if not examples:
            return {}
        
        # Prepare examples summary
        examples_text = ""
        for ex in examples[:5]:
            label_text = f"→ {ex.label}" if ex.label else "→ ?"
            examples_text += f"  - {ex.features} {label_text}\n"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are analyzing streaming data patterns. "
                      "Identify key patterns that could help with prediction."),
            ("human", f"Task: {self.task_description}\n\n"
                     f"Recent examples ({len(examples)} total):\n{examples_text}\n"
                     f"Identify 2-3 key patterns. Format each as: PATTERN_ID|DESCRIPTION|CONFIDENCE\n"
                     f"Example: pattern_1|High feature X correlates with label A|0.8")
        ])
        
        chain = prompt | self.llm
        patterns = {}
        
        try:
            response = chain.invoke({})
            content = response.content if isinstance(response.content, str) else str(response.content)
            lines = content.strip().split('\n')
            
            for line in lines[:3]:
                try:
                    parts = line.split('|')
                    if len(parts) >= 3:
                        pattern_id = parts[0].strip()
                        description = parts[1].strip()
                        confidence = float(parts[2].strip())
                        
                        patterns[pattern_id] = {
                            "description": description,
                            "confidence": confidence
                        }
                except:
                    continue
        except:
            pass
        
        # Add simple statistical patterns
        if examples and examples[0].label:
            # Label distribution
            label_counts = {}
            for ex in examples:
                if ex.label:
                    label_counts[ex.label] = label_counts.get(ex.label, 0) + 1
            
            if label_counts:
                most_common = max(label_counts.items(), key=lambda x: x[1])
                patterns["label_distribution"] = {
                    "description": f"Most common label: {most_common[0]}",
                    "confidence": most_common[1] / len(examples)
                }
        
        return patterns
    
    def predict(self, example: StreamingExample) -> Dict[str, Any]:
        """Make prediction on streaming example."""
        # Create prompt with learned patterns
        patterns_text = ""
        for pattern_id, pattern_data in list(self.learned_patterns.items())[:5]:
            patterns_text += f"  - {pattern_data['description']} (confidence: {pattern_data['confidence']:.2f})\n"
        
        if not patterns_text:
            patterns_text = "  - No patterns learned yet\n"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are performing {self.task_description}. "
                      f"Use the learned patterns to make predictions."),
            ("human", f"Learned patterns:\n{patterns_text}\n"
                     f"New example: {example.features}\n\n"
                     f"Predict the label and confidence (0.0-1.0).\n"
                     f"Format: LABEL|CONFIDENCE")
        ])
        
        chain = prompt | self.llm
        
        try:
            response = chain.invoke({})
            content = response.content if isinstance(response.content, str) else str(response.content)
            parts = content.strip().split('|')
            
            predicted_label = parts[0].strip()
            confidence = float(parts[1].strip()) if len(parts) > 1 else 0.5
        except:
            # Fallback: use most common pattern
            if self.learned_patterns and "label_distribution" in self.learned_patterns:
                predicted_label = self.learned_patterns["label_distribution"]["description"].split(": ")[1]
                confidence = 0.5
            else:
                predicted_label = "unknown"
                confidence = 0.3
        
        # Track prediction
        is_correct = example.label and predicted_label == example.label
        
        prediction = {
            "example_id": example.id,
            "predicted_label": predicted_label,
            "actual_label": example.label,
            "confidence": confidence,
            "correct": is_correct,
            "timestamp": datetime.now()
        }
        
        self.predictions.append(prediction)
        
        # Update statistics
        self.statistics["total_predictions"] += 1
        self.statistics["recent_total"] += 1
        
        if is_correct:
            self.statistics["correct_predictions"] += 1
            self.statistics["recent_correct"] += 1
        
        # Keep only recent stats (last 50)
        if self.statistics["recent_total"] > 50:
            self.statistics["recent_total"] = 50
            self.statistics["recent_correct"] = sum(
                1 for p in self.predictions[-50:] if p["correct"]
            )
        
        return prediction
    
    def _calculate_recent_accuracy(self) -> float:
        """Calculate accuracy on recent predictions."""
        if self.statistics["recent_total"] == 0:
            return 0.0
        return self.statistics["recent_correct"] / self.statistics["recent_total"]
    
    def _take_snapshot(self):
        """Take snapshot of current model state."""
        accuracy = (self.statistics["correct_predictions"] / 
                   self.statistics["total_predictions"] 
                   if self.statistics["total_predictions"] > 0 else 0.0)
        
        snapshot = ModelSnapshot(
            timestamp=datetime.now(),
            examples_seen=self.examples_seen,
            accuracy=accuracy,
            parameters={
                "num_patterns": len(self.learned_patterns),
                "updates": self.updates_performed
            }
        )
        self.snapshots.append(snapshot)
    
    def get_performance_metrics(self) -> PerformanceMetrics:
        """Get current performance metrics."""
        accuracy = (self.statistics["correct_predictions"] / 
                   self.statistics["total_predictions"] 
                   if self.statistics["total_predictions"] > 0 else 0.0)
        
        recent_accuracy = self._calculate_recent_accuracy()
        
        avg_confidence = (sum(p["confidence"] for p in self.predictions[-50:]) / 
                         min(50, len(self.predictions))
                         if self.predictions else 0.0)
        
        return PerformanceMetrics(
            accuracy=accuracy,
            recent_accuracy=recent_accuracy,
            examples_processed=self.examples_seen,
            updates_performed=self.updates_performed,
            avg_confidence=avg_confidence
        )
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get comprehensive statistics."""
        metrics = self.get_performance_metrics()
        
        return {
            "name": self.name,
            "strategy": self.strategy.value,
            "examples_seen": self.examples_seen,
            "updates_performed": self.updates_performed,
            "accuracy": metrics.accuracy,
            "recent_accuracy": metrics.recent_accuracy,
            "patterns_learned": len(self.learned_patterns),
            "snapshots": len(self.snapshots)
        }


class OnlineLearningPattern:
    """
    Online Learning Pattern - Continuous learning from data streams.
    
    Coordinates multiple agents learning from streaming data with
    different update strategies.
    """
    
    def __init__(self):
        self.agents: Dict[str, OnlineLearningAgent] = {}
        self.stream_history: List[StreamingExample] = []
    
    def register_agent(self, agent: OnlineLearningAgent):
        """Register online learning agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name} ({agent.strategy.value})")
    
    def simulate_stream(self, 
                       examples: List[StreamingExample],
                       test_interval: int = 10):
        """Simulate streaming data with online learning."""
        print("\n   Simulating data stream...")
        
        for i, example in enumerate(examples):
            self.stream_history.append(example)
            
            # Each agent processes the stream
            for agent in self.agents.values():
                agent.process_stream(example)
                
                # Test prediction every N examples
                if (i + 1) % test_interval == 0:
                    prediction = agent.predict(example)
            
            # Progress update
            if (i + 1) % 20 == 0:
                print(f"   Processed {i + 1}/{len(examples)} examples...")
    
    def run(self, examples: List[StreamingExample]) -> Dict[str, Any]:
        """Run online learning simulation."""
        print("=== Online Learning Pattern Demo ===\n")
        
        print("1. Simulating data stream...")
        self.simulate_stream(examples)
        
        print("\n2. Collecting performance metrics...")
        metrics = {name: agent.get_performance_metrics() 
                  for name, agent in self.agents.items()}
        
        stats = {name: agent.get_statistics() 
                for name, agent in self.agents.items()}
        
        return {
            "pattern": "OnlineLearning",
            "agents": len(self.agents),
            "stream_length": len(examples),
            "metrics": metrics,
            "statistics": stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Online Learning Pattern Demo ===\n")
    
    # Create pattern
    pattern = OnlineLearningPattern()
    
    # Create agents with different strategies
    print("1. Creating agents...")
    immediate_agent = OnlineLearningAgent(
        name="ImmediateAgent",
        task_description="classify sentiment as positive or negative",
        strategy=UpdateStrategy.IMMEDIATE
    )
    pattern.register_agent(immediate_agent)
    
    batch_agent = OnlineLearningAgent(
        name="BatchAgent",
        task_description="classify sentiment as positive or negative",
        strategy=UpdateStrategy.MINI_BATCH,
        batch_size=10
    )
    pattern.register_agent(batch_agent)
    
    adaptive_agent = OnlineLearningAgent(
        name="AdaptiveAgent",
        task_description="classify sentiment as positive or negative",
        strategy=UpdateStrategy.ADAPTIVE
    )
    pattern.register_agent(adaptive_agent)
    
    # Generate streaming data
    print("\n2. Generating streaming data...")
    stream = []
    
    # Initial positive examples
    for i in range(30):
        stream.append(StreamingExample(
            id=f"ex_{i}",
            features={"text": f"positive_text_{i}", "score": random.uniform(0.6, 1.0)},
            label="positive"
        ))
    
    # Shift to negative examples (concept drift)
    for i in range(30, 60):
        stream.append(StreamingExample(
            id=f"ex_{i}",
            features={"text": f"negative_text_{i}", "score": random.uniform(0.0, 0.4)},
            label="negative"
        ))
    
    # Mixed examples
    for i in range(60, 100):
        label = random.choice(["positive", "negative"])
        score = random.uniform(0.6, 1.0) if label == "positive" else random.uniform(0.0, 0.4)
        stream.append(StreamingExample(
            id=f"ex_{i}",
            features={"text": f"{label}_text_{i}", "score": score},
            label=label
        ))
    
    # Run online learning
    print("\n3. Running online learning...")
    summary = pattern.run(stream)
    
    print(f"\n4. Results:")
    print(f"   Agents: {summary['agents']}")
    print(f"   Stream Length: {summary['stream_length']}")
    
    print(f"\n5. Performance Metrics:")
    for agent_name, metrics in summary['metrics'].items():
        print(f"\n   {agent_name}:")
        print(f"     Accuracy: {metrics.accuracy:.2%}")
        print(f"     Recent Accuracy: {metrics.recent_accuracy:.2%}")
        print(f"     Examples Processed: {metrics.examples_processed}")
        print(f"     Updates Performed: {metrics.updates_performed}")
        print(f"     Avg Confidence: {metrics.avg_confidence:.2f}")
    
    print(f"\n6. Statistics:")
    for agent_name, stats in summary['statistics'].items():
        print(f"\n   {agent_name}:")
        for key, value in stats.items():
            if key != "name":
                print(f"     {key}: {value}")
    
    print("\n✓ Online learning adapted to streaming data in real-time!")

    

Agentic Architecture Pattern: Online Learning

Pattern 55: 55 Online Learning – Line-by-Line Explanation

Overview

Pattern 55: Online Learning

Purpose: Learn continuously from streaming data in real-time. Updates model incrementally as new data arrives, adapting to changing patterns.

Code Structure Analysis

File Information

• Pattern Number: 55
• Pattern Name: 55 Online Learning
• File Path: patterns\learning_adaptation\55_online_learning.py
• Total Lines: 500

Detailed Code Explanation

Module Docstring

"""
Pattern 55: Online Learning

Purpose: Learn continuously from streaming data in real-time.
Updates model incrementally as new data arrives, adapting to changing patterns.
"""

Explanation: This module docstring describes the purpose and key features of the 55 Online Learning pattern.

Imports Section

from typing import Dict, Any, List, Optional, Deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from collections import deque
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, Deque: 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

UpdateStrategy (Line 18)

class UpdateStrategy(Enum):
    IMMEDIATE
    MINI_BATCH
    SLIDING_WINDOW
    ADAPTIVE

Purpose: Strategies for online model updates.

Values: – IMMEDIATE: Represents immediate – MINI_BATCH: Represents mini batch – SLIDING_WINDOW: Represents sliding window – ADAPTIVE: Represents adaptive

Classes

UpdateStrategy (Line 18)

Description: Strategies for online model updates.

StreamingExample (Line 27)

Description: A single streaming data example.

ModelSnapshot (Line 38)

Description: Snapshot of model state at a point in time.

PerformanceMetrics (Line 48)

Description: Real-time performance metrics.

OnlineLearningAgent (Line 58)

Description: Online Learning Agent – Learn continuously from streaming data.

Adapts in real-time as new examples arrive, maintaining performance on non-stationary data streams. Uses LLM for pattern recognition.

Methods:

__init__ (Line 66)

def __init__(self, name, task_description, strategy, window_size, batch_size):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – task_description: Parameter for task_description – strategy: Parameter for strategy – window_size: Parameter for window_size – batch_size: Parameter for batch_size

process_stream (Line 100)

def process_stream(self, example):
    """Process a single streaming example."""

Purpose: Process a single streaming example.

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

_update_model (Line 131)

def _update_model(self):
    """Update model with current buffer."""

Purpose: Update model with current buffer.

Parameters: – self: Instance reference

_extract_patterns (Line 164)

def _extract_patterns(self, examples):
    """Extract patterns from examples using LLM."""

Purpose: Extract patterns from examples using LLM.

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

predict (Line 226)

def predict(self, example):
    """Make prediction on streaming example."""

Purpose: Make prediction on streaming example.

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

_calculate_recent_accuracy (Line 294)

def _calculate_recent_accuracy(self):
    """Calculate accuracy on recent predictions."""

Purpose: Calculate accuracy on recent predictions.

Parameters: – self: Instance reference

_take_snapshot (Line 300)

def _take_snapshot(self):
    """Take snapshot of current model state."""

Purpose: Take snapshot of current model state.

Parameters: – self: Instance reference

get_performance_metrics (Line 317)

def get_performance_metrics(self):
    """Get current performance metrics."""

Purpose: Get current performance metrics.

Parameters: – self: Instance reference

get_statistics (Line 337)

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

Purpose: Get comprehensive statistics.

Parameters: – self: Instance reference

OnlineLearningPattern (Line 353)

Description: Online Learning Pattern – Continuous learning from data streams.

Coordinates multiple agents learning from streaming data with different update strategies.

Methods:

__init__ (Line 361)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 365)

def register_agent(self, agent):
    """Register online learning agent."""

Purpose: Register online learning agent.

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

simulate_stream (Line 370)

def simulate_stream(self, examples, test_interval):
    """Simulate streaming data with online learning."""

Purpose: Simulate streaming data with online learning.

Parameters: – self: Instance reference – examples: Parameter for examples – test_interval: Parameter for test_interval

run (Line 391)

def run(self, examples):
    """Run online learning simulation."""

Purpose: Run online learning simulation.

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

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

    """
Pattern 54: Transfer Learning

Purpose: Transfer knowledge from source tasks/domains to target tasks/domains.
Enables learning in new domains with limited data by leveraging prior knowledge.
"""

from typing import Dict, Any, List, Optional, Set
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 TransferStrategy(Enum):
    """Strategies for transferring knowledge."""
    FINE_TUNING = "fine_tuning"
    FEATURE_EXTRACTION = "feature_extraction"
    DOMAIN_ADAPTATION = "domain_adaptation"
    MULTI_TASK = "multi_task"


@dataclass
class Task:
    """A learning task."""
    id: str
    name: str
    domain: str
    description: str
    examples: List[Dict[str, Any]] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Knowledge:
    """Transferable knowledge."""
    id: str
    source_task: str
    knowledge_type: str  # patterns, features, strategies
    content: Dict[str, Any]
    confidence: float
    applicability: Dict[str, float] = field(default_factory=dict)  # task_id -> score


@dataclass
class TransferResult:
    """Result of knowledge transfer."""
    source_task: str
    target_task: str
    strategy: TransferStrategy
    knowledge_transferred: List[str]
    performance_before: float
    performance_after: float
    improvement: float
    timestamp: datetime = field(default_factory=datetime.now)


class TransferLearningAgent:
    """
    Transfer Learning Agent - Transfer knowledge across tasks and domains.
    
    Learns from source tasks and applies that knowledge to new target tasks
    with limited training data. Uses LLM to identify transferable patterns.
    """
    
    def __init__(self, 
                 name: str,
                 strategy: TransferStrategy = TransferStrategy.FINE_TUNING):
        self.name = name
        self.strategy = strategy
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        self.source_tasks: Dict[str, Task] = {}
        self.target_tasks: Dict[str, Task] = {}
        self.knowledge_base: List[Knowledge] = []
        self.transfer_results: List[TransferResult] = []
        
        self.trained_tasks: Set[str] = set()
        self.statistics: Dict[str, Any] = {
            "tasks_learned": 0,
            "transfers_performed": 0,
            "avg_improvement": 0.0
        }
    
    def learn_source_task(self, task: Task):
        """Learn from a source task to build transferable knowledge."""
        print(f"\n   {self.name}: Learning source task '{task.name}'...")
        
        self.source_tasks[task.id] = task
        
        # Extract knowledge using LLM
        knowledge = self._extract_knowledge(task)
        self.knowledge_base.extend(knowledge)
        
        self.trained_tasks.add(task.id)
        self.statistics["tasks_learned"] += 1
        
        print(f"   Extracted {len(knowledge)} knowledge items from {task.name}")
    
    def transfer_to_target(self, target_task: Task) -> TransferResult:
        """Transfer knowledge from source tasks to target task."""
        print(f"\n   {self.name}: Transferring knowledge to '{target_task.name}'...")
        
        self.target_tasks[target_task.id] = target_task
        
        # Find relevant knowledge
        relevant_knowledge = self._find_relevant_knowledge(target_task)
        
        # Simulate performance before transfer
        performance_before = random.uniform(0.3, 0.5)
        
        # Apply transfer based on strategy
        if self.strategy == TransferStrategy.FINE_TUNING:
            performance_after = self._fine_tune_transfer(target_task, relevant_knowledge)
        elif self.strategy == TransferStrategy.FEATURE_EXTRACTION:
            performance_after = self._feature_extraction_transfer(target_task, relevant_knowledge)
        elif self.strategy == TransferStrategy.DOMAIN_ADAPTATION:
            performance_after = self._domain_adaptation_transfer(target_task, relevant_knowledge)
        else:  # MULTI_TASK
            performance_after = self._multi_task_transfer(target_task, relevant_knowledge)
        
        improvement = performance_after - performance_before
        
        result = TransferResult(
            source_task=",".join([k.source_task for k in relevant_knowledge]),
            target_task=target_task.id,
            strategy=self.strategy,
            knowledge_transferred=[k.id for k in relevant_knowledge],
            performance_before=performance_before,
            performance_after=performance_after,
            improvement=improvement
        )
        
        self.transfer_results.append(result)
        self.statistics["transfers_performed"] += 1
        
        # Update average improvement
        total_improvement = sum(r.improvement for r in self.transfer_results)
        self.statistics["avg_improvement"] = total_improvement / len(self.transfer_results)
        
        print(f"   Transfer complete: {performance_before:.2%} → {performance_after:.2%} (+{improvement:.2%})")
        
        return result
    
    def _extract_knowledge(self, task: Task) -> List[Knowledge]:
        """Extract transferable knowledge from task."""
        knowledge_items = []
        
        # Use LLM to identify patterns
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert at identifying transferable patterns in learning tasks. "
                      "Extract key patterns, features, and strategies that could apply to similar tasks."),
            ("human", f"Task: {task.name}\nDomain: {task.domain}\nDescription: {task.description}\n\n"
                     f"Identify 3 key transferable patterns. For each pattern, provide:\n"
                     f"1. Pattern name\n2. Why it's transferable\n3. Domains it could apply to\n\n"
                     f"Format: PATTERN|REASON|DOMAINS (comma-separated)")
        ])
        
        chain = prompt | self.llm
        try:
            response = chain.invoke({})
            content = response.content if isinstance(response.content, str) else str(response.content)
            lines = content.strip().split('\n')
            
            for i, line in enumerate(lines[:3]):
                try:
                    parts = line.split('|')
                    if len(parts) >= 3:
                        pattern_name = parts[0].strip()
                        reason = parts[1].strip()
                        domains = [d.strip() for d in parts[2].split(',')]
                        
                        knowledge = Knowledge(
                            id=f"k_{task.id}_{i}",
                            source_task=task.id,
                            knowledge_type="pattern",
                            content={
                                "pattern": pattern_name,
                                "reason": reason,
                                "domains": domains
                            },
                            confidence=0.8
                        )
                        knowledge_items.append(knowledge)
                except:
                    continue
        except:
            pass
        
        # Add generic knowledge from examples
        if task.examples:
            knowledge_items.append(Knowledge(
                id=f"k_{task.id}_examples",
                source_task=task.id,
                knowledge_type="examples",
                content={"examples": task.examples[:5]},
                confidence=0.6
            ))
        
        return knowledge_items
    
    def _find_relevant_knowledge(self, target_task: Task) -> List[Knowledge]:
        """Find knowledge relevant to target task."""
        relevant = []
        
        for knowledge in self.knowledge_base:
            # Check domain match
            if knowledge.knowledge_type == "pattern":
                domains = knowledge.content.get("domains", [])
                if target_task.domain in domains or "general" in domains:
                    relevant.append(knowledge)
            else:
                # Generic knowledge is always somewhat relevant
                relevant.append(knowledge)
        
        return relevant[:5]  # Top 5 most relevant
    
    def _fine_tune_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Fine-tuning transfer strategy."""
        base_performance = 0.5
        
        # Benefit from each knowledge item
        for k in knowledge:
            base_performance += k.confidence * 0.1
        
        # Benefit from similar domain
        if any(task.domain in k.content.get("domains", []) for k in knowledge):
            base_performance += 0.15
        
        return min(0.95, base_performance)
    
    def _feature_extraction_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Feature extraction transfer strategy."""
        base_performance = 0.45
        
        # Extract features from source
        feature_count = len(knowledge)
        base_performance += feature_count * 0.08
        
        return min(0.90, base_performance)
    
    def _domain_adaptation_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Domain adaptation transfer strategy."""
        base_performance = 0.40
        
        # Adapt knowledge to new domain
        adapted_count = sum(1 for k in knowledge 
                          if task.domain in k.content.get("domains", []))
        base_performance += adapted_count * 0.12
        
        return min(0.92, base_performance)
    
    def _multi_task_transfer(self, task: Task, knowledge: List[Knowledge]) -> float:
        """Multi-task transfer strategy."""
        base_performance = 0.55
        
        # Benefit from multiple source tasks
        unique_sources = len(set(k.source_task for k in knowledge))
        base_performance += unique_sources * 0.08
        
        return min(0.93, base_performance)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get transfer learning statistics."""
        return {
            "name": self.name,
            "strategy": self.strategy.value,
            "source_tasks": len(self.source_tasks),
            "target_tasks": len(self.target_tasks),
            "knowledge_items": len(self.knowledge_base),
            **self.statistics
        }


class TransferLearningPattern:
    """
    Transfer Learning Pattern - Enable knowledge transfer across tasks.
    
    Coordinates multiple agents to learn from source tasks and apply
    knowledge to new target tasks with limited data.
    """
    
    def __init__(self):
        self.agents: Dict[str, TransferLearningAgent] = {}
        self.tasks: Dict[str, Task] = {}
        self.transfer_history: List[TransferResult] = []
    
    def register_agent(self, agent: TransferLearningAgent):
        """Register transfer learning agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name} ({agent.strategy.value})")
    
    def add_task(self, task: Task):
        """Add a task to the pattern."""
        self.tasks[task.id] = task
    
    def train_source_tasks(self, task_ids: List[str]):
        """Train all agents on source tasks."""
        print("\n   Training on source tasks...")
        
        for task_id in task_ids:
            task = self.tasks.get(task_id)
            if task:
                for agent in self.agents.values():
                    agent.learn_source_task(task)
    
    def transfer_to_targets(self, target_task_ids: List[str]) -> Dict[str, List[TransferResult]]:
        """Transfer knowledge to target tasks."""
        print("\n   Transferring to target tasks...")
        
        results = {}
        for task_id in target_task_ids:
            task = self.tasks.get(task_id)
            if task:
                results[task_id] = []
                for agent in self.agents.values():
                    result = agent.transfer_to_target(task)
                    results[task_id].append(result)
                    self.transfer_history.append(result)
        
        return results
    
    def run(self, 
            source_task_ids: List[str],
            target_task_ids: List[str]) -> Dict[str, Any]:
        """Run transfer learning pipeline."""
        print("=== Transfer Learning Pattern Demo ===\n")
        
        # Train on source tasks
        print("1. Learning from source tasks...")
        self.train_source_tasks(source_task_ids)
        
        # Transfer to target tasks
        print("\n2. Transferring to target tasks...")
        transfer_results = self.transfer_to_targets(target_task_ids)
        
        # Get statistics
        stats = {name: agent.get_statistics() 
                for name, agent in self.agents.items()}
        
        return {
            "pattern": "TransferLearning",
            "agents": len(self.agents),
            "source_tasks": len(source_task_ids),
            "target_tasks": len(target_task_ids),
            "transfer_results": transfer_results,
            "statistics": stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Transfer Learning Pattern Demo ===\n")
    
    # Create pattern
    pattern = TransferLearningPattern()
    
    # Create agents with different strategies
    print("1. Creating agents...")
    fine_tune_agent = TransferLearningAgent(
        name="FineTuneAgent",
        strategy=TransferStrategy.FINE_TUNING
    )
    pattern.register_agent(fine_tune_agent)
    
    feature_agent = TransferLearningAgent(
        name="FeatureAgent",
        strategy=TransferStrategy.FEATURE_EXTRACTION
    )
    pattern.register_agent(feature_agent)
    
    # Create source tasks
    print("\n2. Creating source tasks...")
    sentiment_task = Task(
        id="sentiment",
        name="Sentiment Analysis",
        domain="text_classification",
        description="Classify text sentiment as positive/negative",
        examples=[
            {"text": "Great product!", "label": "positive"},
            {"text": "Terrible service", "label": "negative"},
        ]
    )
    pattern.add_task(sentiment_task)
    
    spam_task = Task(
        id="spam",
        name="Spam Detection",
        domain="text_classification",
        description="Classify emails as spam or not spam",
        examples=[
            {"text": "Win money now!", "label": "spam"},
            {"text": "Meeting tomorrow at 3pm", "label": "not_spam"},
        ]
    )
    pattern.add_task(spam_task)
    
    # Create target task
    print("\n3. Creating target task...")
    toxicity_task = Task(
        id="toxicity",
        name="Toxicity Detection",
        domain="text_classification",
        description="Classify text as toxic or not toxic",
        examples=[
            {"text": "You're an idiot", "label": "toxic"},
        ]
    )
    pattern.add_task(toxicity_task)
    
    # Run transfer learning
    print("\n4. Running transfer learning...")
    summary = pattern.run(
        source_task_ids=["sentiment", "spam"],
        target_task_ids=["toxicity"]
    )
    
    print(f"\n5. Results:")
    print(f"   Agents: {summary['agents']}")
    print(f"   Source Tasks: {summary['source_tasks']}")
    print(f"   Target Tasks: {summary['target_tasks']}")
    
    print(f"\n6. Transfer Results:")
    for task_id, results in summary['transfer_results'].items():
        print(f"\n   Target: {task_id}")
        for result in results:
            print(f"     {result.strategy.value}: {result.performance_before:.2%} → "
                  f"{result.performance_after:.2%} (+{result.improvement:.2%})")
    
    print(f"\n7. Agent Statistics:")
    for agent_name, stats in summary['statistics'].items():
        print(f"\n   {agent_name}:")
        for key, value in stats.items():
            if key != "name":
                print(f"     {key}: {value}")
    
    print("\n✓ Transfer learning enabled knowledge reuse across tasks!")

    

Agentic Architecture Pattern: Transfer Learning

Pattern 54: 54 Transfer Learning – Line-by-Line Explanation

Overview

Pattern 54: Transfer Learning

Purpose: Transfer knowledge from source tasks/domains to target tasks/domains. Enables learning in new domains with limited data by leveraging prior knowledge.

Code Structure Analysis

File Information

• Pattern Number: 54
• Pattern Name: 54 Transfer Learning
• File Path: patterns\learning_adaptation\54_transfer_learning.py
• Total Lines: 440

Detailed Code Explanation

Module Docstring

"""
Pattern 54: Transfer Learning

Purpose: Transfer knowledge from source tasks/domains to target tasks/domains.
Enables learning in new domains with limited data by leveraging prior knowledge.
"""

Explanation: This module docstring describes the purpose and key features of the 54 Transfer Learning pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
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, Set: 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

TransferStrategy (Line 17)

class TransferStrategy(Enum):
    FINE_TUNING
    FEATURE_EXTRACTION
    DOMAIN_ADAPTATION
    MULTI_TASK

Purpose: Strategies for transferring knowledge.

Values: – FINE_TUNING: Represents fine tuning – FEATURE_EXTRACTION: Represents feature extraction – DOMAIN_ADAPTATION: Represents domain adaptation – MULTI_TASK: Represents multi task

Classes

TransferStrategy (Line 17)

Description: Strategies for transferring knowledge.

Task (Line 26)

Description: A learning task.

Knowledge (Line 37)

Description: Transferable knowledge.

TransferResult (Line 48)

Description: Result of knowledge transfer.

TransferLearningAgent (Line 60)

Description: Transfer Learning Agent – Transfer knowledge across tasks and domains.

Learns from source tasks and applies that knowledge to new target tasks with limited training data. Uses LLM to identify transferable patterns.

Methods:

__init__ (Line 68)

def __init__(self, name, strategy):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – strategy: Parameter for strategy

learn_source_task (Line 87)

def learn_source_task(self, task):
    """Learn from a source task to build transferable knowledge."""

Purpose: Learn from a source task to build transferable knowledge.

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

transfer_to_target (Line 102)

def transfer_to_target(self, target_task):
    """Transfer knowledge from source tasks to target task."""

Purpose: Transfer knowledge from source tasks to target task.

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

_extract_knowledge (Line 147)

def _extract_knowledge(self, task):
    """Extract transferable knowledge from task."""

Purpose: Extract transferable knowledge from task.

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

_find_relevant_knowledge (Line 204)

def _find_relevant_knowledge(self, target_task):
    """Find knowledge relevant to target task."""

Purpose: Find knowledge relevant to target task.

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

_fine_tune_transfer (Line 220)

def _fine_tune_transfer(self, task, knowledge):
    """Fine-tuning transfer strategy."""

Purpose: Fine-tuning transfer strategy.

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

_feature_extraction_transfer (Line 234)

def _feature_extraction_transfer(self, task, knowledge):
    """Feature extraction transfer strategy."""

Purpose: Feature extraction transfer strategy.

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

_domain_adaptation_transfer (Line 244)

def _domain_adaptation_transfer(self, task, knowledge):
    """Domain adaptation transfer strategy."""

Purpose: Domain adaptation transfer strategy.

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

_multi_task_transfer (Line 255)

def _multi_task_transfer(self, task, knowledge):
    """Multi-task transfer strategy."""

Purpose: Multi-task transfer strategy.

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

get_statistics (Line 265)

def get_statistics(self):
    """Get transfer learning statistics."""

Purpose: Get transfer learning statistics.

Parameters: – self: Instance reference

TransferLearningPattern (Line 277)

Description: Transfer Learning Pattern – Enable knowledge transfer across tasks.

Coordinates multiple agents to learn from source tasks and apply knowledge to new target tasks with limited data.

Methods:

__init__ (Line 285)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 290)

def register_agent(self, agent):
    """Register transfer learning agent."""

Purpose: Register transfer learning agent.

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

add_task (Line 295)

def add_task(self, task):
    """Add a task to the pattern."""

Purpose: Add a task to the pattern.

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

train_source_tasks (Line 299)

def train_source_tasks(self, task_ids):
    """Train all agents on source tasks."""

Purpose: Train all agents on source tasks.

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

transfer_to_targets (Line 309)

def transfer_to_targets(self, target_task_ids):
    """Transfer knowledge to target tasks."""

Purpose: Transfer knowledge to target tasks.

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

run (Line 325)

def run(self, source_task_ids, target_task_ids):
    """Run transfer learning pipeline."""

Purpose: Run transfer learning pipeline.

Parameters: – self: Instance reference – source_task_ids: Parameter for source_task_ids – target_task_ids: Parameter for target_task_ids

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

    """
Pattern 53: Unsupervised Learning

Purpose: Learn patterns from unlabeled data through clustering, dimensionality reduction, and anomaly detection.
Uses LLM to interpret patterns and provide insights.
"""

from typing import Dict, Any, List, Optional, Tuple, Set
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 UnsupervisedTechnique(Enum):
    """Types of unsupervised learning techniques."""
    CLUSTERING = "clustering"
    ANOMALY_DETECTION = "anomaly_detection"
    DIMENSIONALITY_REDUCTION = "dimensionality_reduction"
    ASSOCIATION_RULES = "association_rules"


@dataclass
class DataPoint:
    """A data point for unsupervised learning."""
    id: str
    features: Dict[str, float]
    metadata: Dict[str, Any] = field(default_factory=dict)
    cluster_id: Optional[int] = None
    is_anomaly: bool = False
    anomaly_score: float = 0.0


@dataclass
class Cluster:
    """A cluster of data points."""
    id: int
    centroid: Dict[str, float]
    members: List[str] = field(default_factory=list)
    description: str = ""
    statistics: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Pattern:
    """A discovered pattern in data."""
    id: str
    technique: UnsupervisedTechnique
    description: str
    support: float
    confidence: float
    examples: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)


class UnsupervisedLearningAgent:
    """
    Unsupervised Learning Agent - Discover patterns in unlabeled data.
    
    Supports clustering, anomaly detection, and pattern discovery.
    Uses LLM for pattern interpretation and insight generation.
    """
    
    def __init__(self, 
                 name: str,
                 technique: UnsupervisedTechnique,
                 num_clusters: int = 3):
        self.name = name
        self.technique = technique
        self.num_clusters = num_clusters
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        self.data_points: List[DataPoint] = []
        self.clusters: List[Cluster] = []
        self.patterns: List[Pattern] = []
        self.anomalies: List[DataPoint] = []
        
        self.trained = False
        self.training_stats: Dict[str, Any] = {}
    
    def add_data(self, data_points: List[DataPoint]):
        """Add data points for analysis."""
        self.data_points.extend(data_points)
        print(f"   {self.name}: Added {len(data_points)} data points (total: {len(self.data_points)})")
    
    def train(self):
        """Discover patterns in data using selected technique."""
        print(f"\n   {self.name}: Training with {self.technique.value}...")
        
        if self.technique == UnsupervisedTechnique.CLUSTERING:
            self._perform_clustering()
        elif self.technique == UnsupervisedTechnique.ANOMALY_DETECTION:
            self._detect_anomalies()
        elif self.technique == UnsupervisedTechnique.DIMENSIONALITY_REDUCTION:
            self._reduce_dimensions()
        elif self.technique == UnsupervisedTechnique.ASSOCIATION_RULES:
            self._find_associations()
        
        self.trained = True
        print(f"   Training complete!")
    
    def _perform_clustering(self):
        """Perform k-means style clustering."""
        if not self.data_points:
            return
        
        # Initialize random centroids
        feature_keys = list(self.data_points[0].features.keys())
        centroids = []
        for i in range(self.num_clusters):
            centroid = {key: random.uniform(-1, 1) for key in feature_keys}
            centroids.append(centroid)
        
        # Iterate to convergence (simplified)
        for iteration in range(10):
            # Assign points to nearest centroid
            for point in self.data_points:
                distances = []
                for i, centroid in enumerate(centroids):
                    dist = self._euclidean_distance(point.features, centroid)
                    distances.append((dist, i))
                point.cluster_id = min(distances)[1]
            
            # Update centroids
            for i in range(self.num_clusters):
                cluster_points = [p for p in self.data_points if p.cluster_id == i]
                if cluster_points:
                    new_centroid = {}
                    for key in feature_keys:
                        new_centroid[key] = sum(p.features[key] for p in cluster_points) / len(cluster_points)
                    centroids[i] = new_centroid
        
        # Create cluster objects
        for i in range(self.num_clusters):
            cluster_points = [p for p in self.data_points if p.cluster_id == i]
            cluster = Cluster(
                id=i,
                centroid=centroids[i],
                members=[p.id for p in cluster_points]
            )
            
            # Get LLM interpretation
            cluster.description = self._interpret_cluster(cluster, cluster_points)
            self.clusters.append(cluster)
        
        self.training_stats["num_clusters"] = len(self.clusters)
        self.training_stats["cluster_sizes"] = [len(c.members) for c in self.clusters]
    
    def _detect_anomalies(self):
        """Detect anomalous data points."""
        if not self.data_points:
            return
        
        # Calculate mean and std for each feature
        feature_keys = list(self.data_points[0].features.keys())
        means = {}
        stds = {}
        
        for key in feature_keys:
            values = [p.features[key] for p in self.data_points]
            means[key] = sum(values) / len(values)
            variance = sum((v - means[key]) ** 2 for v in values) / len(values)
            stds[key] = variance ** 0.5
        
        # Calculate anomaly scores (z-score based)
        threshold = 2.0  # 2 standard deviations
        for point in self.data_points:
            scores = []
            for key in feature_keys:
                if stds[key] > 0:
                    z_score = abs((point.features[key] - means[key]) / stds[key])
                    scores.append(z_score)
            
            point.anomaly_score = max(scores) if scores else 0.0
            point.is_anomaly = point.anomaly_score > threshold
            
            if point.is_anomaly:
                self.anomalies.append(point)
        
        self.training_stats["num_anomalies"] = len(self.anomalies)
        self.training_stats["anomaly_rate"] = len(self.anomalies) / len(self.data_points)
    
    def _reduce_dimensions(self):
        """Perform dimensionality reduction (simplified PCA concept)."""
        if not self.data_points:
            return
        
        # Identify most important features (by variance)
        feature_keys = list(self.data_points[0].features.keys())
        variances = {}
        
        for key in feature_keys:
            values = [p.features[key] for p in self.data_points]
            mean = sum(values) / len(values)
            variance = sum((v - mean) ** 2 for v in values) / len(values)
            variances[key] = variance
        
        # Sort by variance
        sorted_features = sorted(variances.items(), key=lambda x: x[1], reverse=True)
        
        # Keep top features
        num_components = min(3, len(feature_keys))
        top_features = [f[0] for f in sorted_features[:num_components]]
        
        # Create reduced representation
        for point in self.data_points:
            reduced_features = {k: point.features[k] for k in top_features}
            point.metadata["reduced_features"] = reduced_features
        
        self.training_stats["original_dimensions"] = len(feature_keys)
        self.training_stats["reduced_dimensions"] = num_components
        self.training_stats["top_features"] = top_features
        self.training_stats["explained_variance"] = [variances[f] for f in top_features]
    
    def _find_associations(self):
        """Find association rules (frequent patterns)."""
        if not self.data_points:
            return
        
        # Find frequent feature combinations
        feature_keys = list(self.data_points[0].features.keys())
        
        # Look for correlated features
        for i, key1 in enumerate(feature_keys):
            for key2 in feature_keys[i+1:]:
                # Calculate correlation
                values1 = [p.features[key1] for p in self.data_points]
                values2 = [p.features[key2] for p in self.data_points]
                
                # Simple correlation
                mean1 = sum(values1) / len(values1)
                mean2 = sum(values2) / len(values2)
                
                covariance = sum((v1 - mean1) * (v2 - mean2) 
                               for v1, v2 in zip(values1, values2)) / len(values1)
                
                std1 = (sum((v - mean1) ** 2 for v in values1) / len(values1)) ** 0.5
                std2 = (sum((v - mean2) ** 2 for v in values2) / len(values2)) ** 0.5
                
                if std1 > 0 and std2 > 0:
                    correlation = covariance / (std1 * std2)
                    
                    if abs(correlation) > 0.7:  # Strong correlation
                        pattern = Pattern(
                            id=f"assoc_{len(self.patterns)}",
                            technique=UnsupervisedTechnique.ASSOCIATION_RULES,
                            description=f"{key1} and {key2} are {'positively' if correlation > 0 else 'negatively'} correlated",
                            support=abs(correlation),
                            confidence=abs(correlation),
                            examples=[f"{key1}↔{key2}: {correlation:.2f}"]
                        )
                        self.patterns.append(pattern)
        
        self.training_stats["num_patterns"] = len(self.patterns)
    
    def _euclidean_distance(self, features1: Dict[str, float], 
                          features2: Dict[str, float]) -> float:
        """Calculate Euclidean distance between feature vectors."""
        distance = 0.0
        for key in features1.keys():
            distance += (features1[key] - features2.get(key, 0)) ** 2
        return distance ** 0.5
    
    def _interpret_cluster(self, cluster: Cluster, points: List[DataPoint]) -> str:
        """Use LLM to interpret cluster meaning."""
        if not points:
            return "Empty cluster"
        
        # Summarize cluster characteristics
        feature_keys = list(cluster.centroid.keys())
        centroid_desc = ", ".join([f"{k}={v:.2f}" for k, v in cluster.centroid.items()])
        
        # Sample points
        sample_points = points[:3]
        samples_desc = "\n".join([
            f"  - {p.id}: " + ", ".join([f"{k}={v:.2f}" for k, v in p.features.items()])
            for p in sample_points
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a data analyst interpreting clustering results. "
                      "Provide a brief, insightful description of the cluster."),
            ("human", f"Cluster {cluster.id} has:\n"
                     f"- Centroid: {centroid_desc}\n"
                     f"- {len(points)} members\n"
                     f"- Sample members:\n{samples_desc}\n\n"
                     f"Describe this cluster in one concise sentence:")
        ])
        
        chain = prompt | self.llm
        try:
            response = chain.invoke({})
            content = response.content if isinstance(response.content, str) else str(response.content)
            return content.strip()
        except:
            return f"Cluster with {len(points)} members"
    
    def get_insights(self) -> str:
        """Get LLM-generated insights about discovered patterns."""
        if not self.trained:
            return "No insights available - train first"
        
        summary = f"Analyzed {len(self.data_points)} data points using {self.technique.value}.\n\n"
        
        if self.technique == UnsupervisedTechnique.CLUSTERING:
            summary += f"Found {len(self.clusters)} clusters:\n"
            for cluster in self.clusters:
                summary += f"  - Cluster {cluster.id}: {len(cluster.members)} members - {cluster.description}\n"
        
        elif self.technique == UnsupervisedTechnique.ANOMALY_DETECTION:
            summary += f"Detected {len(self.anomalies)} anomalies ({self.training_stats['anomaly_rate']:.1%}):\n"
            for anomaly in self.anomalies[:5]:
                summary += f"  - {anomaly.id}: score={anomaly.anomaly_score:.2f}\n"
        
        elif self.technique == UnsupervisedTechnique.DIMENSIONALITY_REDUCTION:
            summary += f"Reduced dimensions from {self.training_stats['original_dimensions']} to {self.training_stats['reduced_dimensions']}:\n"
            summary += f"  Top features: {', '.join(self.training_stats['top_features'])}\n"
        
        elif self.technique == UnsupervisedTechnique.ASSOCIATION_RULES:
            summary += f"Found {len(self.patterns)} association patterns:\n"
            for pattern in self.patterns:
                summary += f"  - {pattern.description} (confidence: {pattern.confidence:.2f})\n"
        
        return summary
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get training statistics."""
        return {
            "name": self.name,
            "technique": self.technique.value,
            "trained": self.trained,
            "data_points": len(self.data_points),
            **self.training_stats
        }


class UnsupervisedLearningPattern:
    """
    Unsupervised Learning Pattern - Discover patterns without labels.
    
    Coordinates multiple unsupervised learning agents with different techniques.
    """
    
    def __init__(self):
        self.agents: Dict[str, UnsupervisedLearningAgent] = {}
        self.data_points: List[DataPoint] = []
    
    def register_agent(self, agent: UnsupervisedLearningAgent):
        """Register unsupervised learning agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name} ({agent.technique.value})")
    
    def add_data(self, data_points: List[DataPoint]):
        """Add data for all agents to analyze."""
        self.data_points.extend(data_points)
        for agent in self.agents.values():
            agent.add_data(data_points)
    
    def train_all(self):
        """Train all agents."""
        print("\n   Training all agents...")
        for agent in self.agents.values():
            agent.train()
    
    def run(self, data_points: List[DataPoint]) -> Dict[str, Any]:
        """Run unsupervised learning pipeline."""
        print("=== Unsupervised Learning Pattern Demo ===\n")
        
        # Add data
        print("1. Adding data...")
        self.add_data(data_points)
        
        # Train all agents
        print("\n2. Training agents...")
        self.train_all()
        
        # Get insights
        print("\n3. Generating insights...")
        insights = {}
        for agent_name, agent in self.agents.items():
            insights[agent_name] = agent.get_insights()
        
        # Get statistics
        stats = {name: agent.get_statistics() 
                for name, agent in self.agents.items()}
        
        return {
            "pattern": "UnsupervisedLearning",
            "agents": len(self.agents),
            "data_points": len(self.data_points),
            "insights": insights,
            "statistics": stats
        }


# Example usage
if __name__ == "__main__":
    print("=== Unsupervised Learning Pattern Demo ===\n")
    
    # Create pattern
    pattern = UnsupervisedLearningPattern()
    
    # Create agents with different techniques
    print("1. Creating agents...")
    clustering_agent = UnsupervisedLearningAgent(
        name="ClusterAgent",
        technique=UnsupervisedTechnique.CLUSTERING,
        num_clusters=3
    )
    pattern.register_agent(clustering_agent)
    
    anomaly_agent = UnsupervisedLearningAgent(
        name="AnomalyDetector",
        technique=UnsupervisedTechnique.ANOMALY_DETECTION
    )
    pattern.register_agent(anomaly_agent)
    
    reduction_agent = UnsupervisedLearningAgent(
        name="DimensionReducer",
        technique=UnsupervisedTechnique.DIMENSIONALITY_REDUCTION
    )
    pattern.register_agent(reduction_agent)
    
    # Generate sample data
    print("\n2. Generating sample data...")
    data_points = []
    
    # Cluster 1: High x, low y
    for i in range(20):
        data_points.append(DataPoint(
            id=f"point_{i}",
            features={
                "x": random.uniform(0.7, 1.0),
                "y": random.uniform(0.0, 0.3),
                "z": random.uniform(0.4, 0.6)
            }
        ))
    
    # Cluster 2: Low x, high y
    for i in range(20, 40):
        data_points.append(DataPoint(
            id=f"point_{i}",
            features={
                "x": random.uniform(0.0, 0.3),
                "y": random.uniform(0.7, 1.0),
                "z": random.uniform(0.4, 0.6)
            }
        ))
    
    # Cluster 3: Mid x, mid y
    for i in range(40, 60):
        data_points.append(DataPoint(
            id=f"point_{i}",
            features={
                "x": random.uniform(0.4, 0.6),
                "y": random.uniform(0.4, 0.6),
                "z": random.uniform(0.4, 0.6)
            }
        ))
    
    # Add some anomalies
    data_points.append(DataPoint(
        id="anomaly_1",
        features={"x": 0.95, "y": 0.95, "z": 0.95}
    ))
    data_points.append(DataPoint(
        id="anomaly_2",
        features={"x": 0.05, "y": 0.05, "z": 0.05}
    ))
    
    # Run pattern
    print(f"\n3. Running unsupervised learning...")
    summary = pattern.run(data_points)
    
    print(f"\n4. Results:")
    print(f"   Data Points: {summary['data_points']}")
    print(f"   Agents: {summary['agents']}")
    
    print(f"\n5. Insights:")
    for agent_name, insight in summary['insights'].items():
        print(f"\n   {agent_name}:")
        print(f"   {insight}")
    
    print(f"\n6. Statistics:")
    for agent_name, stats in summary['statistics'].items():
        print(f"\n   {agent_name}:")
        for key, value in stats.items():
            if key != "name":
                print(f"     {key}: {value}")
    
    print("\n✓ Unsupervised learning discovered patterns without labels!")

    

Agentic Architecture Pattern: Unsupervised Learning

Pattern 53: 53 Unsupervised Learning – Line-by-Line Explanation

Overview

Pattern 53: Unsupervised Learning

Purpose: Learn patterns from unlabeled data through clustering, dimensionality reduction, and anomaly detection. Uses LLM to interpret patterns and provide insights.

Code Structure Analysis

File Information

• Pattern Number: 53
• Pattern Name: 53 Unsupervised Learning
• File Path: patterns\learning_adaptation\53_unsupervised_learning.py
• Total Lines: 494

Detailed Code Explanation

Module Docstring

"""
Pattern 53: Unsupervised Learning

Purpose: Learn patterns from unlabeled data through clustering, dimensionality reduction, and anomaly detection.
Uses LLM to interpret patterns and provide insights.
"""

Explanation: This module docstring describes the purpose and key features of the 53 Unsupervised Learning pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple, Set
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, Set: 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

UnsupervisedTechnique (Line 17)

class UnsupervisedTechnique(Enum):
    CLUSTERING
    ANOMALY_DETECTION
    DIMENSIONALITY_REDUCTION
    ASSOCIATION_RULES

Purpose: Types of unsupervised learning techniques.

Values: – CLUSTERING: Represents clustering – ANOMALY_DETECTION: Represents anomaly detection – DIMENSIONALITY_REDUCTION: Represents dimensionality reduction – ASSOCIATION_RULES: Represents association rules

Classes

UnsupervisedTechnique (Line 17)

Description: Types of unsupervised learning techniques.

DataPoint (Line 26)

Description: A data point for unsupervised learning.

Cluster (Line 37)

Description: A cluster of data points.

Pattern (Line 47)

Description: A discovered pattern in data.

UnsupervisedLearningAgent (Line 58)

Description: Unsupervised Learning Agent – Discover patterns in unlabeled data.

Supports clustering, anomaly detection, and pattern discovery. Uses LLM for pattern interpretation and insight generation.

Methods:

__init__ (Line 66)

def __init__(self, name, technique, num_clusters):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – technique: Parameter for technique – num_clusters: Parameter for num_clusters

add_data (Line 83)

def add_data(self, data_points):
    """Add data points for analysis."""

Purpose: Add data points for analysis.

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

train (Line 88)

def train(self):
    """Discover patterns in data using selected technique."""

Purpose: Discover patterns in data using selected technique.

Parameters: – self: Instance reference

_perform_clustering (Line 104)

def _perform_clustering(self):
    """Perform k-means style clustering."""

Purpose: Perform k-means style clustering.

Parameters: – self: Instance reference

_detect_anomalies (Line 151)

def _detect_anomalies(self):
    """Detect anomalous data points."""

Purpose: Detect anomalous data points.

Parameters: – self: Instance reference

_reduce_dimensions (Line 185)

def _reduce_dimensions(self):
    """Perform dimensionality reduction (simplified PCA concept)."""

Purpose: Perform dimensionality reduction (simplified PCA concept).

Parameters: – self: Instance reference

_find_associations (Line 217)

def _find_associations(self):
    """Find association rules (frequent patterns)."""

Purpose: Find association rules (frequent patterns).

Parameters: – self: Instance reference

_euclidean_distance (Line 258)

def _euclidean_distance(self, features1, features2):
    """Calculate Euclidean distance between feature vectors."""

Purpose: Calculate Euclidean distance between feature vectors.

Parameters: – self: Instance reference – features1: Parameter for features1 – features2: Parameter for features2

_interpret_cluster (Line 266)

def _interpret_cluster(self, cluster, points):
    """Use LLM to interpret cluster meaning."""

Purpose: Use LLM to interpret cluster meaning.

Parameters: – self: Instance reference – cluster: Parameter for cluster – points: Parameter for points

get_insights (Line 300)

def get_insights(self):
    """Get LLM-generated insights about discovered patterns."""

Purpose: Get LLM-generated insights about discovered patterns.

Parameters: – self: Instance reference

get_statistics (Line 328)

def get_statistics(self):
    """Get training statistics."""

Purpose: Get training statistics.

Parameters: – self: Instance reference

UnsupervisedLearningPattern (Line 339)

Description: Unsupervised Learning Pattern – Discover patterns without labels.

Coordinates multiple unsupervised learning agents with different techniques.

Methods:

__init__ (Line 346)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 350)

def register_agent(self, agent):
    """Register unsupervised learning agent."""

Purpose: Register unsupervised learning agent.

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

add_data (Line 355)

def add_data(self, data_points):
    """Add data for all agents to analyze."""

Purpose: Add data for all agents to analyze.

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

train_all (Line 361)

def train_all(self):
    """Train all agents."""

Purpose: Train all agents.

Parameters: – self: Instance reference

run (Line 367)

def run(self, data_points):
    """Run unsupervised learning pipeline."""

Purpose: Run unsupervised learning pipeline.

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

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

    """
Pattern 52: Supervised Learning Agent

Agent learns from labeled training data to make predictions on new inputs.
Implements training, validation, evaluation, and prediction using LLM-based learning.
"""

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_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


class LearningStrategy(Enum):
    """Strategies for learning from examples."""
    FEW_SHOT = "few_shot"  # Use examples in prompt
    FINE_TUNING = "fine_tuning"  # Simulate fine-tuning
    PATTERN_MATCHING = "pattern_matching"  # Match patterns


@dataclass
class TrainingExample:
    """Labeled training example."""
    input_data: str
    label: str
    weight: float = 1.0
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class ValidationResult:
    """Validation metrics."""
    accuracy: float
    precision: float
    recall: float
    f1_score: float
    confusion_matrix: Dict[str, int]


@dataclass
class Prediction:
    """Model prediction with confidence."""
    input_data: str
    predicted_label: str
    confidence: float
    reasoning: Optional[str] = None
    actual_label: Optional[str] = None
    correct: Optional[bool] = None
    timestamp: datetime = field(default_factory=datetime.now)


class SupervisedLearningAgent:
    """Agent that learns from labeled examples."""
    
    def __init__(self, name: str, task_description: str, 
                 model_name: str = "gpt-4", strategy: LearningStrategy = LearningStrategy.FEW_SHOT):
        self.name = name
        self.task_description = task_description
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.strategy = strategy
        self.training_examples: List[TrainingExample] = []
        self.validation_examples: List[TrainingExample] = []
        self.predictions: List[Prediction] = []
        self.is_trained = False
    
    def train(self, examples: List[TrainingExample], validation_split: float = 0.2):
        """Train on labeled examples."""
        print(f"\n📚 {self.name}: Training on {len(examples)} examples...")
        
        # Split into training and validation
        shuffled = examples.copy()
        random.shuffle(shuffled)
        
        split_idx = int(len(shuffled) * (1 - validation_split))
        self.training_examples = shuffled[:split_idx]
        self.validation_examples = shuffled[split_idx:]
        
        self.is_trained = True
        print(f"   Training examples: {len(self.training_examples)}")
        print(f"   Validation examples: {len(self.validation_examples)}")
    
    def _create_few_shot_prompt(self, input_data: str, num_examples: int = 5) -> str:
        """Create few-shot prompt with training examples."""
        prompt_parts = [f"Task: {self.task_description}\n"]
        prompt_parts.append("Learn from these labeled examples:\n")
        
        # Sample random training examples
        sample_examples = random.sample(
            self.training_examples, 
            min(num_examples, len(self.training_examples))
        )
        
        for i, ex in enumerate(sample_examples, 1):
            prompt_parts.append(f"\nExample {i}:")
            prompt_parts.append(f"Input: {ex.input_data}")
            prompt_parts.append(f"Label: {ex.label}")
        
        prompt_parts.append(f"\n\nNow predict the label for this input:")
        prompt_parts.append(f"Input: {input_data}")
        prompt_parts.append("Label:")
        
        return "\n".join(prompt_parts)
    
    def predict(self, input_data: str, return_reasoning: bool = False) -> Prediction:
        """Make prediction using learned patterns."""
        if not self.is_trained:
            raise ValueError(f"{self.name} has not been trained yet")
        
        # Create prompt based on strategy
        if self.strategy == LearningStrategy.FEW_SHOT:
            prompt_text = self._create_few_shot_prompt(input_data)
        else:
            prompt_text = f"Task: {self.task_description}\n\nInput: {input_data}\n\nLabel:"
        
        # Get prediction from LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a supervised learning model. "
                      f"Predict labels based on learned patterns. "
                      f"{'Provide brief reasoning.' if return_reasoning else 'Provide only the label.'}"),
            ("human", "{prompt}")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({"prompt": prompt_text})
        
        # Parse response
        lines = response.strip().split('\n')
        predicted_label = lines[0].strip()
        reasoning = '\n'.join(lines[1:]).strip() if len(lines) > 1 and return_reasoning else None
        
        # Estimate confidence (simplified)
        confidence = 0.85 if len(self.training_examples) > 10 else 0.70
        
        prediction = Prediction(
            input_data=input_data,
            predicted_label=predicted_label,
            confidence=confidence,
            reasoning=reasoning
        )
        
        self.predictions.append(prediction)
        return prediction
    
    def evaluate(self, test_examples: List[TrainingExample]) -> ValidationResult:
        """Evaluate model on test set."""
        print(f"\n🧪 Evaluating {self.name} on {len(test_examples)} test examples...")
        
        correct = 0
        predictions_dict: Dict[str, List[Prediction]] = {}
        confusion = {}
        
        for ex in test_examples:
            pred = self.predict(ex.input_data)
            pred.actual_label = ex.label
            pred.correct = (pred.predicted_label.lower().strip() == ex.label.lower().strip())
            
            if pred.correct:
                correct += 1
            
            # Track predictions by label
            if ex.label not in predictions_dict:
                predictions_dict[ex.label] = []
            predictions_dict[ex.label].append(pred)
            
            # Confusion matrix
            key = f"{ex.label}->{pred.predicted_label}"
            confusion[key] = confusion.get(key, 0) + 1
        
        accuracy = correct / len(test_examples) if test_examples else 0
        
        # Calculate precision, recall, F1 per class (simplified)
        precision = accuracy  # Simplified
        recall = accuracy
        f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        
        return ValidationResult(
            accuracy=accuracy,
            precision=precision,
            recall=recall,
            f1_score=f1_score,
            confusion_matrix=confusion
        )
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get learning statistics."""
        correct_predictions = sum(1 for p in self.predictions if p.correct)
        total_predictions = len([p for p in self.predictions if p.actual_label is not None])
        
        return {
            "name": self.name,
            "is_trained": self.is_trained,
            "training_examples": len(self.training_examples),
            "validation_examples": len(self.validation_examples),
            "total_predictions": len(self.predictions),
            "accuracy": correct_predictions / total_predictions if total_predictions > 0 else 0,
            "strategy": self.strategy.value
        }


class SupervisedLearningPattern:
    """
    Supervised Learning Agent Pattern - Learn from labeled data.
    
    Trains agents on labeled examples to make predictions on new inputs.
    Supports training, validation, evaluation, and prediction workflows.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.agents: Dict[str, SupervisedLearningAgent] = {}
    
    def create_agent(self, name: str, task_description: str, 
                    strategy: LearningStrategy = LearningStrategy.FEW_SHOT) -> SupervisedLearningAgent:
        """Create a supervised learning agent."""
        agent = SupervisedLearningAgent(name, task_description, self.model_name, strategy)
        self.agents[name] = agent
        return agent
    
    def train_test_split(self, examples: List[TrainingExample], 
                        test_size: float = 0.2) -> Tuple[List[TrainingExample], List[TrainingExample]]:
        """Split examples into training and test sets."""
        shuffled = examples.copy()
        random.shuffle(shuffled)
        
        split_idx = int(len(shuffled) * (1 - test_size))
        return shuffled[:split_idx], shuffled[split_idx:]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "total_agents": len(self.agents),
            "agents": {name: agent.get_statistics() for name, agent in self.agents.items()}
        }


# Example usage
if __name__ == "__main__":
    print("=== Supervised Learning Agent Demo ===\n")
    
    # Create pattern
    print("1. Creating supervised learning pattern...")
    pattern = SupervisedLearningPattern()
    
    # Task: Sentiment Analysis
    print(f"\n{'='*70}")
    print("📊 Task: Sentiment Analysis")
    print(f"{'='*70}")
    
    agent = pattern.create_agent(
        name="SentimentClassifier",
        task_description="Classify text sentiment as Positive, Negative, or Neutral"
    )
    
    # Create training data
    training_data = [
        TrainingExample("I love this product! Amazing quality!", "Positive"),
        TrainingExample("Best purchase ever! Highly recommend!", "Positive"),
        TrainingExample("Fantastic experience, will buy again", "Positive"),
        TrainingExample("Really impressed with the service", "Positive"),
        TrainingExample("Excellent quality and fast shipping", "Positive"),
        
        TrainingExample("Terrible experience, very disappointed", "Negative"),
        TrainingExample("Waste of money, doesn't work at all", "Negative"),
        TrainingExample("Poor quality, broke after one use", "Negative"),
        TrainingExample("Worst product I've ever bought", "Negative"),
        TrainingExample("Not worth it, total ripoff", "Negative"),
        
        TrainingExample("It's okay, nothing special", "Neutral"),
        TrainingExample("Average product, does the job", "Neutral"),
        TrainingExample("Not bad, not great either", "Neutral"),
        TrainingExample("Acceptable quality for the price", "Neutral"),
        TrainingExample("Standard product, meets expectations", "Neutral"),
    ]
    
    # Split data
    train_data, test_data = pattern.train_test_split(training_data, test_size=0.3)
    
    # Train agent
    print("\n2. Training agent...")
    agent.train(train_data, validation_split=0.2)
    
    # Make predictions
    print(f"\n{'='*70}")
    print("🔮 Making Predictions")
    print(f"{'='*70}\n")
    
    test_inputs = [
        "This is absolutely wonderful!",
        "Completely useless and broken",
        "It works as expected, nothing more"
    ]
    
    for test_input in test_inputs:
        pred = agent.predict(test_input, return_reasoning=False)
        print(f"Input: {pred.input_data}")
        print(f"Prediction: {pred.predicted_label}")
        print(f"Confidence: {pred.confidence:.2%}\n")
    
    # Evaluate on test set
    print(f"\n{'='*70}")
    print("📈 Model Evaluation")
    print(f"{'='*70}\n")
    
    results = agent.evaluate(test_data)
    
    print(f"Accuracy: {results.accuracy:.2%}")
    print(f"Precision: {results.precision:.2%}")
    print(f"Recall: {results.recall:.2%}")
    print(f"F1 Score: {results.f1_score:.2%}")
    
    print(f"\nConfusion Matrix:")
    for key, count in sorted(results.confusion_matrix.items()):
        print(f"  {key}: {count}")
    
    # Statistics
    print(f"\n{'='*70}")
    print("📊 Learning Statistics")
    print(f"{'='*70}")
    
    stats = pattern.get_statistics()
    
    for agent_name, agent_stats in stats['agents'].items():
        print(f"\nAgent: {agent_name}")
        print(f"  Trained: {agent_stats['is_trained']}")
        print(f"  Training examples: {agent_stats['training_examples']}")
        print(f"  Validation examples: {agent_stats['validation_examples']}")
        print(f"  Total predictions: {agent_stats['total_predictions']}")
        print(f"  Accuracy: {agent_stats['accuracy']:.2%}")
        print(f"  Strategy: {agent_stats['strategy']}")
    
    print("\n✓ Supervised learning enabled prediction from labeled data!")

    

Agentic Architecture Pattern: Supervised Learning

Pattern 52: 52 Supervised Learning – Line-by-Line Explanation

Overview

Pattern 52: Supervised Learning Agent

Agent learns from labeled training data to make predictions on new inputs. Implements training, validation, evaluation, and prediction using LLM-based learning.

Code Structure Analysis

File Information

• Pattern Number: 52
• Pattern Name: 52 Supervised Learning
• File Path: patterns\learning_adaptation\52_supervised_learning.py
• Total Lines: 336

Detailed Code Explanation

Module Docstring

"""
Pattern 52: Supervised Learning Agent

Agent learns from labeled training data to make predictions on new inputs.
Implements training, validation, evaluation, and prediction using LLM-based learning.
"""

Explanation: This module docstring describes the purpose and key features of the 52 Supervised Learning 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_core.output_parsers import StrOutputParser
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_core.output_parsers import StrOutputParser: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

LearningStrategy (Line 18)

class LearningStrategy(Enum):
    FEW_SHOT
    FINE_TUNING
    PATTERN_MATCHING

Purpose: Strategies for learning from examples.

Values: – FEW_SHOT: Represents few shot – FINE_TUNING: Represents fine tuning – PATTERN_MATCHING: Represents pattern matching

Classes

LearningStrategy (Line 18)

Description: Strategies for learning from examples.

TrainingExample (Line 26)

Description: Labeled training example.

ValidationResult (Line 35)

Description: Validation metrics.

Prediction (Line 45)

Description: Model prediction with confidence.

SupervisedLearningAgent (Line 56)

Description: Agent that learns from labeled examples.

Methods:

__init__ (Line 59)

def __init__(self, name, task_description, model_name, strategy):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – task_description: Parameter for task_description – model_name: LLM model to use (e.g., “gpt-4”) – strategy: Parameter for strategy

train (Line 70)

def train(self, examples, validation_split):
    """Train on labeled examples."""

Purpose: Train on labeled examples.

Parameters: – self: Instance reference – examples: Parameter for examples – validation_split: Parameter for validation_split

_create_few_shot_prompt (Line 86)

def _create_few_shot_prompt(self, input_data, num_examples):
    """Create few-shot prompt with training examples."""

Purpose: Create few-shot prompt with training examples.

Parameters: – self: Instance reference – input_data: Parameter for input_data – num_examples: Parameter for num_examples

predict (Line 108)

def predict(self, input_data, return_reasoning):
    """Make prediction using learned patterns."""

Purpose: Make prediction using learned patterns.

Parameters: – self: Instance reference – input_data: Parameter for input_data – return_reasoning: Parameter for return_reasoning

evaluate (Line 148)

def evaluate(self, test_examples):
    """Evaluate model on test set."""

Purpose: Evaluate model on test set.

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

get_statistics (Line 188)

def get_statistics(self):
    """Get learning statistics."""

Purpose: Get learning statistics.

Parameters: – self: Instance reference

SupervisedLearningPattern (Line 204)

Description: Supervised Learning Agent Pattern – Learn from labeled data.

Trains agents on labeled examples to make predictions on new inputs. Supports training, validation, evaluation, and prediction workflows.

Methods:

__init__ (Line 212)

def __init__(self, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

create_agent (Line 216)

def create_agent(self, name, task_description, strategy):
    """Create a supervised learning agent."""

Purpose: Create a supervised learning agent.

Parameters: – self: Instance reference – name: Identifier or label – task_description: Parameter for task_description – strategy: Parameter for strategy

train_test_split (Line 223)

def train_test_split(self, examples, test_size):
    """Split examples into training and test sets."""

Purpose: Split examples into training and test sets.

Parameters: – self: Instance reference – examples: Parameter for examples – test_size: Parameter for test_size

get_statistics (Line 232)

def get_statistics(self):
    """Get pattern statistics."""

Purpose: Get pattern 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 = LearningStrategy(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

    """
Pattern 51: Reinforcement Learning Agent

Agent learns optimal behavior through trial-and-error using rewards and penalties.
Implements Q-learning for state-action value estimation and policy improvement.
"""

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


class ActionType(Enum):
    """Types of actions in RL environment."""
    MOVE_UP = "move_up"
    MOVE_DOWN = "move_down"
    MOVE_LEFT = "move_left"
    MOVE_RIGHT = "move_right"
    INTERACT = "interact"
    WAIT = "wait"


@dataclass
class State:
    """Environment state."""
    position: Tuple[int, int]
    context: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)
    
    def __hash__(self):
        return hash((self.position, tuple(sorted(self.context.items()))))
    
    def __eq__(self, other):
        return self.position == other.position and self.context == other.context


@dataclass
class Experience:
    """Single experience (s, a, r, s')."""
    state: State
    action: ActionType
    reward: float
    next_state: State
    done: bool
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Episode:
    """Episode of experiences."""
    episode_number: int
    experiences: List[Experience] = field(default_factory=list)
    total_reward: float = 0.0
    steps: int = 0
    success: bool = False


class QLearningAgent:
    """Q-learning agent with exploration-exploitation."""
    
    def __init__(self, name: str, alpha: float = 0.1, gamma: float = 0.9, 
                 epsilon: float = 0.2, model_name: str = "gpt-4"):
        self.name = name
        self.alpha = alpha  # Learning rate
        self.gamma = gamma  # Discount factor
        self.epsilon = epsilon  # Exploration rate
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        # Q-table: state -> action -> value
        self.q_table: Dict[State, Dict[ActionType, float]] = defaultdict(lambda: defaultdict(float))
        self.experiences: List[Experience] = []
        self.episodes: List[Episode] = []
    
    def get_q_value(self, state: State, action: ActionType) -> float:
        """Get Q-value for state-action pair."""
        return self.q_table[state][action]
    
    def get_max_q_value(self, state: State, available_actions: List[ActionType]) -> float:
        """Get maximum Q-value for state."""
        if not available_actions:
            return 0.0
        return max(self.get_q_value(state, action) for action in available_actions)
    
    def select_action(self, state: State, available_actions: List[ActionType]) -> ActionType:
        """Select action using epsilon-greedy policy."""
        if not available_actions:
            return ActionType.WAIT
        
        # Exploration
        if random.random()  Dict[ActionType, float]:
        """Get action probability distribution for state."""
        q_values = {action: self.get_q_value(state, action) for action in available_actions}
        
        if not q_values:
            return {}
        
        # Softmax policy
        import math
        exp_q = {a: math.exp(q / 0.1) for a, q in q_values.items()}
        total = sum(exp_q.values())
        
        return {a: exp_q[a] / total for a in available_actions}
    
    def decay_epsilon(self, decay_rate: float = 0.99):
        """Decay exploration rate."""
        self.epsilon *= decay_rate
        self.epsilon = max(0.01, self.epsilon)  # Minimum epsilon


class GridEnvironment:
    """Simple grid world environment for RL."""
    
    def __init__(self, size: int = 5):
        self.size = size
        self.goal_position = (size - 1, size - 1)
        self.obstacles = [(1, 1), (2, 2), (3, 1)]
        self.current_position = (0, 0)
    
    def reset(self) -> State:
        """Reset to initial state."""
        self.current_position = (0, 0)
        return State(position=self.current_position, context={"steps": 0})
    
    def get_available_actions(self, position: Tuple[int, int]) -> List[ActionType]:
        """Get valid actions from position."""
        actions = []
        x, y = position
        
        if x > 0 and (x-1, y) not in self.obstacles:
            actions.append(ActionType.MOVE_LEFT)
        if x  0 and (x, y-1) not in self.obstacles:
            actions.append(ActionType.MOVE_UP)
        if y  Tuple[State, float, bool]:
        """Execute action and return (next_state, reward, done)."""
        x, y = self.current_position
        
        # Apply action
        if action == ActionType.MOVE_UP:
            y -= 1
        elif action == ActionType.MOVE_DOWN:
            y += 1
        elif action == ActionType.MOVE_LEFT:
            x -= 1
        elif action == ActionType.MOVE_RIGHT:
            x += 1
        
        # Check bounds and obstacles
        if (x = self.size or y = self.size or 
            (x, y) in self.obstacles):
            # Invalid move - stay in place with penalty
            reward = -1.0
            done = False
        elif (x, y) == self.goal_position:
            # Reached goal
            self.current_position = (x, y)
            reward = 10.0
            done = True
        else:
            # Valid move
            self.current_position = (x, y)
            reward = -0.1  # Small penalty for each step
            done = False
        
        next_state = State(position=self.current_position)
        return next_state, reward, done


class ReinforcementLearningPattern:
    """
    Reinforcement Learning Agent - Q-learning for optimal policy discovery.
    
    Agent learns through interaction with environment, updating Q-values
    based on rewards received for state-action pairs.
    """
    
    def __init__(self, alpha: float = 0.1, gamma: float = 0.9, epsilon: float = 0.2):
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon
        self.agent: Optional[QLearningAgent] = None
        self.environment: Optional[GridEnvironment] = None
    
    def create_agent(self, name: str) -> QLearningAgent:
        """Create RL agent."""
        agent = QLearningAgent(name, self.alpha, self.gamma, self.epsilon)
        self.agent = agent
        return agent
    
    def create_environment(self, size: int = 5) -> GridEnvironment:
        """Create environment."""
        env = GridEnvironment(size)
        self.environment = env
        return env
    
    def train_episode(self, max_steps: int = 100) -> Episode:
        """Train for one episode."""
        if not self.agent or not self.environment:
            raise ValueError("Agent and environment must be created first")
        
        state = self.environment.reset()
        episode = Episode(episode_number=len(self.agent.episodes))
        
        for step in range(max_steps):
            # Select action
            available_actions = self.environment.get_available_actions(state.position)
            action = self.agent.select_action(state, available_actions)
            
            # Take action
            next_state, reward, done = self.environment.step(action)
            
            # Learn
            available_next_actions = self.environment.get_available_actions(next_state.position)
            experience = Experience(state, action, reward, next_state, done)
            self.agent.learn_from_experience(experience, available_next_actions)
            
            # Record
            episode.experiences.append(experience)
            episode.total_reward += reward
            episode.steps += 1
            
            if done:
                episode.success = True
                break
            
            state = next_state
        
        self.agent.episodes.append(episode)
        return episode
    
    def train(self, num_episodes: int = 100, epsilon_decay: float = 0.99) -> List[Episode]:
        """Train agent for multiple episodes."""
        print(f"\n🎯 Training RL agent for {num_episodes} episodes...\n")
        
        episodes = []
        for i in range(num_episodes):
            episode = self.train_episode()
            episodes.append(episode)
            
            # Decay exploration
            self.agent.decay_epsilon(epsilon_decay)
            
            if (i + 1) % 20 == 0:
                recent_episodes = episodes[-20:]
                avg_reward = sum(e.total_reward for e in recent_episodes) / len(recent_episodes)
                avg_steps = sum(e.steps for e in recent_episodes) / len(recent_episodes)
                success_rate = sum(1 for e in recent_episodes if e.success) / len(recent_episodes)
                
                print(f"Episode {i+1}/{num_episodes}:")
                print(f"   Avg reward: {avg_reward:.2f}")
                print(f"   Avg steps: {avg_steps:.1f}")
                print(f"   Success rate: {success_rate:.1%}")
                print(f"   Epsilon: {self.agent.epsilon:.3f}\n")
        
        return episodes
    
    def get_training_metrics(self, episodes: List[Episode]) -> Dict[str, Any]:
        """Calculate training metrics."""
        total_episodes = len(episodes)
        successful = sum(1 for e in episodes if e.success)
        
        avg_reward = sum(e.total_reward for e in episodes) / total_episodes
        avg_steps = sum(e.steps for e in episodes) / total_episodes
        
        # Learning curve (reward over time)
        window = 10
        learning_curve = []
        for i in range(window, total_episodes, window):
            window_episodes = episodes[i-window:i]
            avg = sum(e.total_reward for e in window_episodes) / len(window_episodes)
            learning_curve.append(avg)
        
        return {
            "total_episodes": total_episodes,
            "successful_episodes": successful,
            "success_rate": successful / total_episodes,
            "avg_reward": avg_reward,
            "avg_steps": avg_steps,
            "q_table_size": len(self.agent.q_table),
            "learning_curve": learning_curve
        }


# Example usage
if __name__ == "__main__":
    print("=== Reinforcement Learning Agent Demo ===\n")
    
    # Create pattern
    print("1. Creating RL pattern with Q-learning...")
    pattern = ReinforcementLearningPattern(
        alpha=0.1,    # Learning rate
        gamma=0.9,    # Discount factor
        epsilon=0.3   # Initial exploration rate
    )
    
    # Create agent and environment
    print("2. Setting up agent and grid environment (5x5)...")
    agent = pattern.create_agent("Q-Learner")
    env = pattern.create_environment(size=5)
    
    print(f"   Goal: {env.goal_position}")
    print(f"   Obstacles: {env.obstacles}\n")
    
    # Train agent
    print("3. Training agent...")
    episodes = pattern.train(num_episodes=100, epsilon_decay=0.98)
    
    # Get metrics
    print(f"\n{'='*70}")
    print("📊 Training Results")
    print(f"{'='*70}")
    
    metrics = pattern.get_training_metrics(episodes)
    
    print(f"   Total episodes: {metrics['total_episodes']}")
    print(f"   Successful: {metrics['successful_episodes']}")
    print(f"   Success rate: {metrics['success_rate']:.1%}")
    print(f"   Avg reward: {metrics['avg_reward']:.2f}")
    print(f"   Avg steps: {metrics['avg_steps']:.1f}")
    print(f"   Q-table size: {metrics['q_table_size']} states learned")
    
    # Show learning curve
    if metrics['learning_curve']:
        print(f"\n   Learning curve (avg reward per 10 episodes):")
        for i, avg in enumerate(metrics['learning_curve'][:5]):
            print(f"   Episodes {i*10+1}-{(i+1)*10}: {avg:.2f}")
    
    # Test learned policy
    print(f"\n{'='*70}")
    print("🎮 Testing Learned Policy")
    print(f"{'='*70}\n")
    
    # Set epsilon to 0 for pure exploitation
    agent.epsilon = 0.0
    
    state = env.reset()
    print(f"Starting position: {state.position}")
    print(f"Goal: {env.goal_position}\n")
    
    path = [state.position]
    total_reward = 0.0
    
    for step in range(20):
        available_actions = env.get_available_actions(state.position)
        action = agent.select_action(state, available_actions)
        
        next_state, reward, done = env.step(action)
        path.append(next_state.position)
        total_reward += reward
        
        print(f"Step {step + 1}: {action.value} -> {next_state.position} (reward: {reward:+.1f})")
        
        if done:
            print(f"\n✅ Goal reached in {step + 1} steps!")
            print(f"   Total reward: {total_reward:.2f}")
            print(f"   Path: {' -> '.join(str(p) for p in path)}")
            break
        
        state = next_state
    
    print("\n✓ RL agent successfully learned optimal policy!")

    

Agentic Architecture Pattern: Reinforcement Learning

Pattern 51: 51 Reinforcement Learning – Line-by-Line Explanation

Overview

Pattern 51: Reinforcement Learning Agent

Agent learns optimal behavior through trial-and-error using rewards and penalties. Implements Q-learning for state-action value estimation and policy improvement.

Code Structure Analysis

File Information

• Pattern Number: 51
• Pattern Name: 51 Reinforcement Learning
• File Path: patterns\learning_adaptation\51_reinforcement_learning.py
• Total Lines: 412

Detailed Code Explanation

Module Docstring

"""
Pattern 51: Reinforcement Learning Agent

Agent learns optimal behavior through trial-and-error using rewards and penalties.
Implements Q-learning for state-action value estimation and policy improvement.
"""

Explanation: This module docstring describes the purpose and key features of the 51 Reinforcement Learning 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 collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
        import math  # 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

ActionType (Line 18)

class ActionType(Enum):
    MOVE_UP
    MOVE_DOWN
    MOVE_LEFT
    MOVE_RIGHT
    INTERACT
    WAIT

Purpose: Types of actions in RL environment.

Values: – MOVE_UP: Represents move up – MOVE_DOWN: Represents move down – MOVE_LEFT: Represents move left – MOVE_RIGHT: Represents move right – INTERACT: Represents interact – WAIT: Represents wait

Classes

ActionType (Line 18)

Description: Types of actions in RL environment.

State (Line 29)

Description: Environment state.

Methods:

__hash__ (Line 35)

def __hash__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

__eq__ (Line 38)

def __eq__(self, other):
    """No method docstring."""

Purpose: No method docstring.

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

Experience (Line 43)

Description: Single experience (s, a, r, s’).

Episode (Line 54)

Description: Episode of experiences.

QLearningAgent (Line 63)

Description: Q-learning agent with exploration-exploitation.

Methods:

__init__ (Line 66)

def __init__(self, name, alpha, gamma, epsilon, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – alpha: Parameter for alpha – gamma: Parameter for gamma – epsilon: Parameter for epsilon – model_name: LLM model to use (e.g., “gpt-4”)

get_q_value (Line 79)

def get_q_value(self, state, action):
    """Get Q-value for state-action pair."""

Purpose: Get Q-value for state-action pair.

Parameters: – self: Instance reference – state: Current workflow state – action: Parameter for action

get_max_q_value (Line 83)

def get_max_q_value(self, state, available_actions):
    """Get maximum Q-value for state."""

Purpose: Get maximum Q-value for state.

Parameters: – self: Instance reference – state: Current workflow state – available_actions: Parameter for available_actions

select_action (Line 89)

def select_action(self, state, available_actions):
    """Select action using epsilon-greedy policy."""

Purpose: Select action using epsilon-greedy policy.

Parameters: – self: Instance reference – state: Current workflow state – available_actions: Parameter for available_actions

update_q_value (Line 106)

def update_q_value(self, state, action, reward, next_state, available_next_actions, done):
    """Update Q-value using Q-learning update rule."""

Purpose: Update Q-value using Q-learning update rule.

Parameters: – self: Instance reference – state: Current workflow state – action: Parameter for action – reward: Parameter for reward – next_state: Parameter for next_state – available_next_actions: Parameter for available_next_actions – done: Parameter for done

learn_from_experience (Line 122)

def learn_from_experience(self, experience, available_next_actions):
    """Learn from single experience."""

Purpose: Learn from single experience.

Parameters: – self: Instance reference – experience: Parameter for experience – available_next_actions: Parameter for available_next_actions

get_policy (Line 134)

def get_policy(self, state, available_actions):
    """Get action probability distribution for state."""

Purpose: Get action probability distribution for state.

Parameters: – self: Instance reference – state: Current workflow state – available_actions: Parameter for available_actions

decay_epsilon (Line 148)

def decay_epsilon(self, decay_rate):
    """Decay exploration rate."""

Purpose: Decay exploration rate.

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

GridEnvironment (Line 154)

Description: Simple grid world environment for RL.

Methods:

__init__ (Line 157)

def __init__(self, size):
    """No method docstring."""

Purpose: No method docstring.

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

reset (Line 163)

def reset(self):
    """Reset to initial state."""

Purpose: Reset to initial state.

Parameters: – self: Instance reference

get_available_actions (Line 168)

def get_available_actions(self, position):
    """Get valid actions from position."""

Purpose: Get valid actions from position.

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

step (Line 184)

def step(self, action):
    """Execute action and return (next_state, reward, done)."""

Purpose: Execute action and return (next_state, reward, done).

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

ReinforcementLearningPattern (Line 219)

Description: Reinforcement Learning Agent – Q-learning for optimal policy discovery.

Agent learns through interaction with environment, updating Q-values based on rewards received for state-action pairs.

Methods:

__init__ (Line 227)

def __init__(self, alpha, gamma, epsilon):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – alpha: Parameter for alpha – gamma: Parameter for gamma – epsilon: Parameter for epsilon

create_agent (Line 234)

def create_agent(self, name):
    """Create RL agent."""

Purpose: Create RL agent.

Parameters: – self: Instance reference – name: Identifier or label

create_environment (Line 240)

def create_environment(self, size):
    """Create environment."""

Purpose: Create environment.

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

train_episode (Line 246)

def train_episode(self, max_steps):
    """Train for one episode."""

Purpose: Train for one episode.

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

train (Line 281)

def train(self, num_episodes, epsilon_decay):
    """Train agent for multiple episodes."""

Purpose: Train agent for multiple episodes.

Parameters: – self: Instance reference – num_episodes: Parameter for num_episodes – epsilon_decay: Parameter for epsilon_decay

get_training_metrics (Line 307)

def get_training_metrics(self, episodes):
    """Calculate training metrics."""

Purpose: Calculate training metrics.

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

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