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