aishastr.live

Agentic Architecture Pattern: Offline Learning

Written by

admin

in


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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts