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