Understanding Semantic Temperature Collapse in AI Models

# Temperature scaling in neural networks
import torch
import torch.nn.functional as F

def apply_temperature(logits, temperature):
    """
    Apply temperature scaling to model logits.
    
    Detailed explanation:
    - Logits are raw scores that the model assigns to each possible next token
    - Temperature scales these scores before applying the softmax function
    - Low temperature (< 1.0): Increases differences between logits, making choices more deterministic
    - High temperature (> 1.0): Reduces differences, increasing randomness in choices
    
    Parameters:
    - logits: Tensor containing the model's raw scores [batch_size, vocab_size]
    - temperature: Temperature value (must be > 0)
    
    Returns:
    - probabilities: Normalized probability distribution after scaling
    """
    if temperature <= 0:
        raise ValueError("Temperature must be positive")
    
    # Step 1: Scale logits by dividing by temperature
    # This is the core of the temperature control mechanism
    scaled_logits = logits / temperature
    
    # Step 2: Apply softmax to convert to probabilities
    # Softmax ensures that the sum of probabilities is 1
    probabilities = F.softmax(scaled_logits, dim=-1)
    
    return probabilities

# Practical example to visualize the effect of temperature
# Let's simulate logits that a model might produce for 4 possible next words
logits = torch.tensor([2.0, 1.0, 0.5, -1.0])  # Raw model scores
temperatures = [0.1, 0.5, 1.0, 2.0]  # Different temperature settings

print("Effect of temperature on probability distribution:")
print("=" * 60)
print("Original logits:", logits.numpy())
print("Interpretation: Word1 most probable, Word4 least probable")
print()

for temp in temperatures:
    probs = apply_temperature(logits, temp)
    print(f"Temperature {temp}:")
    print(f"  Probabilities: {probs.numpy().round(3)}")
    print(f"  Most probable word: {torch.argmax(probs).item() + 1}")
    print(f"  Entropy: {-torch.sum(probs * torch.log(probs + 1e-8)):.3f}")
    print()

print("Observations:")
print("- T=0.1: Very focused on most probable word (deterministic)")
print("- T=1.0: Natural distribution based on original logits")
print("- T=2.0: More uniform, greater randomness in choices")

import matplotlib.pyplot as plt
import numpy as np

def visualize_temperature_effects():
    """
    Visualize how temperature affects probability distributions
    """
    # Simulated logits for next token prediction
    logits = np.array([3.0, 2.5, 2.0, 1.5, 1.0, 0.5, 0.0, -0.5])
    tokens = ['The', 'cat', 'sat', 'on', 'the', 'mat', '.', ',']
    
    temperatures = [0.1, 0.5, 1.0, 2.0]
    
    fig, axes = plt.subplots(2, 2, figsize=(15, 10))
    axes = axes.flatten()
    
    for i, temp in enumerate(temperatures):
        # Apply temperature scaling
        scaled_logits = logits / temp
        probs = np.exp(scaled_logits) / np.sum(np.exp(scaled_logits))
        
        # Plot distribution
        axes[i].bar(tokens, probs, color='steelblue', alpha=0.7)
        axes[i].set_title(f'Temperature = {temp}')
        axes[i].set_ylabel('Probability')
        axes[i].set_ylim(0, 1)
        axes[i].tick_params(axis='x', rotation=45)
        
        # Add entropy annotation
        entropy = -np.sum(probs * np.log(probs + 1e-8))
        axes[i].text(0.02, 0.98, f'Entropy: {entropy:.3f}', 
                    transform=axes[i].transAxes, va='top', 
                    bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5))
    
    plt.tight_layout()
    plt.show()

# Run visualization
visualize_temperature_effects()

import torch
import numpy as np
from collections import deque
from typing import Dict, List, Optional
import matplotlib.pyplot as plt

class TemperatureMonitor:
    """
    Comprehensive monitoring system for detecting semantic temperature collapse.
    
    Architectural Explanation:
    This system implements a multi-metric approach to monitor the semantic health
    of the model. Instead of relying on a single indicator, it combines different measures
    for a robust and reliable assessment of temperature collapse.
    
    Key Components:
    1. Continuous Tracking: Maintains a sliding window of recent metrics
    2. Multi-dimensional Analysis: evaluates entropy, diversity, repetition
    3. Intelligent Detection: Uses multiple thresholds to reduce false positives
    4. Complete Diagnostics: Provides detailed reports and recommendations
    """
    
    def __init__(self, window_size: int = 100, threshold: float = 0.3):
        """
        Initialize the monitoring system.
        
        Parameters:
        - window_size: Window size for historical metrics
        - threshold: Minimum diversity threshold considered healthy
        
        Technical Explanation:
        Deques with maxlen automatically implement a sliding window,
        keeping only the most recent data for efficient real-time analysis.
        """
        self.window_size = window_size
        self.threshold = threshold
        
        # Sliding windows for continuous tracking
        self.token_history = deque(maxlen=window_size)      # Generated token history
        self.entropy_history = deque(maxlen=window_size)     # Entropy history
        self.temperature_history = deque(maxlen=window_size) # Effective temperature history
        self.diversity_history = deque(maxlen=window_size)   # Diversity history
        self.repetition_scores = deque(maxlen=window_size)    # Repetition score history
        
    def update_metrics(self, logits: torch.Tensor, generated_tokens: List[int]) -> Dict[str, float]:
        """
        Update monitoring metrics with new generation data.
        
        Parameters:
        - logits: Tensor of raw model scores [batch_size, vocab_size]
        - generated_tokens: List of tokens generated by the model
        
        Returns:
        - Dictionary with all calculated metrics
        
        Detailed Explanation:
        This method is the heart of the monitoring system. It calculates multiple metrics
        that together provide a complete view of the model's semantic health.
        """
        # Step 1: Entropy calculation
        # Entropy measures uncertainty in the probability distribution
        # High entropy = greater randomness/creativity
        # Low entropy = greater determinism/predictability
        probs = torch.softmax(logits, dim=-1)
        mean_entropy = -torch.sum(probs * torch.log(probs + 1e-8), dim=-1).mean().item()
        
        # Step 2: Diversity metrics calculation
        # Analyze the last 20 tokens to evaluate local diversity
        recent_tokens = generated_tokens[-20:] if len(generated_tokens) >= 20 else generated_tokens
        unique_ratio = len(set(recent_tokens)) / len(recent_tokens)
        
        # Step 3: Effective temperature estimation
        # Convert observed entropy to an equivalent temperature
        # This allows us to understand what temperature the model is actually using
        vocab_size = logits.size(-1)
        max_entropy = np.log(vocab_size)  # Maximum possible entropy
        effective_temp = mean_entropy / max_entropy
        
        # Step 4: Update historical windows
        self.token_history.extend(generated_tokens)
        self.entropy_history.append(mean_entropy)
        self.temperature_history.append(effective_temp)
        self.diversity_history.append(unique_ratio)
        
        # Step 5: Repetition score calculation
        repetition_score = self._calculate_repetition_score(generated_tokens)
        self.repetition_scores.append(repetition_score)
        
        return {
            'entropy': mean_entropy,
            'effective_temperature': effective_temp,
            'diversity_ratio': unique_ratio,
            'repetition_score': repetition_score,
            'vocab_usage': len(set(self.token_history)) / len(self.token_history) if self.token_history else 0
        }
    
    def _calculate_repetition_score(self, tokens: List[int]) -> float:
        """
        Calculate repetition score based on n-gram analysis.
        
        Technical Explanation:
        N-gram analysis detects repetitive patterns at different scales:
        - Bigram (n=2): Consecutively repeated words
        - Trigram (n=3): Repeated short phrases
        - 4-gram: Longer repeated patterns
        
        A high score indicates problematic repetition.
        """
        if len(tokens) < 10:
            return 0.0  # Too few tokens for meaningful analysis
            
        repetition_score = 0.0
        
        # Multi-level analysis to capture different types of repetition
        for n in range(2, 5):  # Analysis from bigram to 4-gram
            if len(tokens) < n:
                continue
                
            # Extract all n-grams from the sequence
            ngrams = [tuple(tokens[i:i+n]) for i in range(len(tokens)-n+1)]
            unique_ngrams = len(set(ngrams))
            total_ngrams = len(ngrams)
            
            if total_ngrams > 0:
                # Calculate repetition ratio for this level
                repetition_ratio = 1 - (unique_ngrams / total_ngrams)
                repetition_score += repetition_ratio / 3  # Weighted average across levels
        
        return repetition_score
    
    def detect_collapse(self) -> bool:
        """
        Detect if semantic temperature collapse is occurring.
        
        Logical Explanation:
        Collapse is detected when MULTIPLE indicators are simultaneously
        below critical thresholds. This approach reduces false positives
        and provides more reliable detection.
        
        Collapse Indicators:
        1. Low diversity: Model uses limited vocabulary
        2. Low entropy: Choices are too predictable
        3. High repetition: Repetitive patterns in text
        """
        if len(self.diversity_history) < 10:
            return False  # Insufficient data for reliable evaluation
            
        # Calculate recent averages to smooth out fluctuations
        recent_diversity = np.mean(list(self.diversity_history)[-10:])
        recent_entropy = np.mean(list(self.entropy_history)[-10:])
        recent_repetition = np.mean(list(self.repetition_scores)[-10:])
        
        # Multi-criteria evaluation
        diversity_collapse = recent_diversity < self.threshold      # Critical diversity
        entropy_collapse = recent_entropy < 1.0                    # Entropy too low
        repetition_collapse = recent_repetition > 0.4              # Excessive repetition
        
        # Collapse detected if multiple indicators are present
        return (diversity_collapse and entropy_collapse) or repetition_collapse
    
    def get_diagnostic_report(self) -> Dict:
        """
        Generate a comprehensive diagnostic report of the model's state.
        
        Output Explanation:
        The report provides:
        - Overall status (HEALTHY/COLLAPSE_DETECTED)
        - Aggregate metrics for trend analysis
        - Trend directions (IMPROVING/DECLINING)
        - Specific actionable recommendations
        """
        if not self.entropy_history:
            return {"status": "INSUFFICIENT_DATA", "message": "Not enough data for diagnosis"}
        
        report = {
            'status': 'HEALTHY' if not self.detect_collapse() else 'COLLAPSE_DETECTED',
            'metrics': {
                'avg_entropy': np.mean(list(self.entropy_history)),
                'avg_temperature': np.mean(list(self.temperature_history)),
                'avg_diversity': np.mean(list(self.diversity_history)),
                'avg_repetition': np.mean(list(self.repetition_scores)),
                'vocab_utilization': len(set(self.token_history)) / len(self.token_history) if self.token_history else 0
            },
            'trends': self._calculate_trends(),
            'recommendations': self._generate_recommendations()
        }
        
        return report
    
    def _calculate_trends(self) -> Dict[str, str]:
        """
        Calculate trend directions for key metrics.
        
        Technical Explanation:
        Compares recent averages (last 5 values) with previous ones
        to determine if metrics are improving or declining.
        """
        trends = {}
        
        if len(self.entropy_history) >= 10:
            recent_entropy = list(self.entropy_history)[-5:]     # Last 5 values
            older_entropy = list(self.entropy_history)[-10:-5]  # Previous 5 values
            entropy_trend = np.mean(recent_entropy) - np.mean(older_entropy)
            trends['entropy'] = 'IMPROVING' if entropy_trend > 0 else 'DECLINING'
            
        if len(self.diversity_history) >= 10:
            recent_diversity = list(self.diversity_history)[-5:]
            older_diversity = list(self.diversity_history)[-10:-5]
            diversity_trend = np.mean(recent_diversity) - np.mean(older_diversity)
            trends['diversity'] = 'IMPROVING' if diversity_trend > 0 else 'DECLINING'
            
        return trends
    
    def _generate_recommendations(self) -> List[str]:
        """
        Generate actionable recommendations based on current state.
        
        Logical Explanation:
        Recommendations are specific and contextual, based on observed
        metrics and the model's health status.
        """
        recommendations = []
        
        if self.detect_collapse():
            # Recommendations for detected collapse
            recommendations.append("INCREASE_TEMPERATURE: Model shows signs of collapse")
            recommendations.append("DIVERSIFY_TRAINING_DATA: Consider augmenting training data")
            recommendations.append("IMPLEMENT_ADJUSTMENT: Use dynamic temperature adjustment")
        else:
            # Recommendations for optimization
            avg_temp = np.mean(list(self.temperature_history))
            if avg_temp < 0.5:
                recommendations.append("MODERATELY_INCREASE_TEMPERATURE: Current temperature too low")
            elif avg_temp > 1.5:
                recommendations.append("CONSIDER_REDUCING_TEMPERATURE: Current temperature too high")
            else:
                recommendations.append("MAINTAIN_CURRENT_SETTINGS: Temperature is well-balanced")
                
        return recommendations
    
    def visualize_metrics(self, save_path: Optional[str] = None):
        """
        Create visualizations of monitoring metrics.
        
        Dashboard Explanation:
        The 4-quadrant dashboard provides a complete view:
        1. Temperature trend: Monitors effective temperature over time
        2. Diversity: Tracks token uniqueness ratio
        3. Entropy distribution: Histogram of entropy values
        4. Repetition scores: Monitors repetition tendency
        """
        if len(self.entropy_history) < 2:
            print("Insufficient data for visualization")
            return
        
        # Create 2x2 dashboard
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        fig.suptitle('Semantic Temperature Monitoring Dashboard', fontsize=16)
        
        # Quadrant 1: Temperature trend
        axes[0, 0].plot(list(self.temperature_history), 'b-', linewidth=2)
        axes[0, 0].axhline(y=0.7, color='g', linestyle='--', label='Optimal')
        axes[0, 0].axhline(y=0.3, color='r', linestyle='--', label='Danger Zone')
        axes[0, 0].set_title('Effective Temperature Over Time')
        axes[0, 0].set_ylabel('Temperature')
        axes[0, 0].legend()
        axes[0, 0].grid(True, alpha=0.3)
        
        # Quadrant 2: Diversity trend
        axes[0, 1].plot(list(self.diversity_history), 'g-', linewidth=2)
        axes[0, 1].axhline(y=self.threshold, color='r', linestyle='--', label='Threshold')
        axes[0, 1].set_title('Token Diversity Ratio')
        axes[0, 1].set_ylabel('Diversity Ratio')
        axes[0, 1].legend()
        axes[0, 1].grid(True, alpha=0.3)
        
        # Quadrant 3: Entropy distribution
        axes[1, 0].hist(list(self.entropy_history), bins=20, alpha=0.7, color='purple')
        axes[1, 0].set_title('Entropy Distribution')
        axes[1, 0].set_xlabel('Entropy')
        axes[1, 0].set_ylabel('Frequency')
        axes[1, 0].grid(True, alpha=0.3)
        
        # Quadrant 4: Repetition scores
        axes[1, 1].plot(list(self.repetition_scores), 'r-', linewidth=2)
        axes[1, 1].axhline(y=0.4, color='orange', linestyle='--', label='Warning Level')
        axes[1, 1].set_title('Repetition Score Over Time')
        axes[1, 1].set_xlabel('Generation Step')
        axes[1, 1].set_ylabel('Repetition Score')
        axes[1, 1].legend()
        axes[1, 1].grid(True, alpha=0.3)
        
        plt.tight_layout()
        
        if save_path:
            plt.savefig(save_path, dpi=300, bbox_inches='tight')
        
        plt.show()

# Practical usage example
def demonstrate_monitoring():
    """
    Demonstrate the monitoring system with simulated data.
    
    Demonstration Purpose:
    Shows how the system detects collapse by comparing normal generations
    with problematic generations that show signs of semantic collapse.
    """
    monitor = TemperatureMonitor()
    
    print("=== Monitoring System Demonstration ===")
    print("Phase 1: Normal generations (steps 0-19)")
    print("Phase 2: Problematic generations (steps 20-49)")
    print()
    
    # Simulate generation data
    for i in range(50):
        # Simulate logits (in practice, these come from the model)
        logits = torch.randn(1, 1000)  # Random logits for demonstration
        
        # Simulate generated tokens
        if i < 20:
            # Normal generation - good diversity
            tokens = torch.multinomial(torch.softmax(logits, dim=-1), 10).squeeze().tolist()
        else:
            # Simulate collapse - more repetitive patterns
            tokens = [1, 2, 3, 1, 2, 3, 1, 2, 3, 4]  # Repetitive pattern
        
        # Update metrics
        metrics = monitor.update_metrics(logits, tokens)
        
        if i % 10 == 0:
            print(f"Step {i:2d}: Diversity = {metrics['diversity_ratio']:.3f}, "
                  f"Entropy = {metrics['entropy']:.3f}, "
                  f"Repetition = {metrics['repetition_score']:.3f}")
    
    # Generate diagnostic report
    report = monitor.get_diagnostic_report()
    print(f"\n=== Diagnostic Report ===")
    print(f"Status: {report['status']}")
    print(f"Average Metrics:")
    for metric, value in report['metrics'].items():
        print(f"  {metric}: {value:.3f}")
    print(f"Trends: {report['trends']}")
    print(f"Recommendations: {', '.join(report['recommendations'])}")
    
    # Visualize metrics
    print(f"\n=== Dashboard Visualization ===")
    monitor.visualize_metrics()

if __name__ == "__main__":
    demonstrate_monitoring()

import time
from typing import List, Callable, Dict, Any

class TemperatureAlertSystem:
    """
    Real-time alert system for detecting temperature collapse.
    
    System Architecture:
    This system implements an Observer pattern for flexible notifications:
    1. Continuous Monitoring: Constantly checks collapse conditions
    2. Multiple Callbacks: Supports different notification channels
    3. Alert History: Maintains a log of all generated alerts
    4. Flexible Configuration: Allows customization of alert levels
    
    Use Cases:
    - Production system monitoring
    - Maintenance team notifications
    - Automatic intervention triggering
    - Real-time monitoring dashboards
    """
    
    def __init__(self, monitor: TemperatureMonitor):
        """
        Initialize the alert system.
        
        Parameters:
        - monitor: TemperatureMonitor instance for detection
        
        Architectural Explanation:
        The system uses an Observer pattern where callbacks are registered
        and called when alert conditions are detected. This allows
        flexible notification management without tight coupling.
        """
        self.monitor = monitor
        self.alert_callbacks: List[Callable] = []  # List of callback functions
        self.alert_history: List[Dict] = []       # Alert history
        self.alert_thresholds = {                   # Customizable thresholds
            'diversity': 0.3,
            'entropy': 1.0,
            'repetition': 0.4
        }
        
    def add_alert_callback(self, callback: Callable[[Dict], None]):
        """
        Add a callback function for alerts.
        
        Parameters:
        - callback: Function that accepts an alert dictionary as parameter
        
        Pattern Explanation:
        This method implements the Observer pattern by registering observers
        (callbacks) that will be notified when events occur.
        
        Callback example:
        ```python
        def my_alert_handler(alert):
            print(f"Alert: {alert['message']}")
            # Custom handling logic
        ```
        """
        self.alert_callbacks.append(callback)
        
    def check_and_alert(self) -> bool:
        """
        Check collapse conditions and send alert if necessary.
        
        Returns:
        - True if an alert was generated, False otherwise
        
        Logical Explanation:
        This method is the heart of the alert system. It performs checks
        and activates the notification chain when critical conditions
        are detected.
        """
        # Check if monitor detects collapse
        if not self.monitor.detect_collapse():
            return False
            
        # Get current diagnostic report
        diagnostic_report = self.monitor.get_diagnostic_report()
        
        # Generate complete alert
        alert = {
            'timestamp': time.time(),
            'level': self._determine_alert_level(diagnostic_report),
            'message': 'Semantic temperature collapse detected',
            'metrics': diagnostic_report['metrics'],
            'recommendations': diagnostic_report['recommendations'],
            'trends': diagnostic_report.get('trends', {}),
            'severity_score': self._calculate_severity_score(diagnostic_report)
        }
        
        # Add to history
        self.alert_history.append(alert)
        
        # Limit history size (keeps last 1000 alerts)
        if len(self.alert_history) > 1000:
            self.alert_history = self.alert_history[-1000:]
        
        # Notify all registered callbacks
        for callback in self.alert_callbacks:
            try:
                callback(alert)
            except Exception as e:
                print(f"Error in alert callback: {e}")
        
        return True
    
    def _determine_alert_level(self, report: Dict) -> str:
        """
        Determine alert level based on metrics.
        
        Classification Logic:
        - CRITICAL: Multiple critical metrics
        - WARNING: One critical metric or multiple borderline
        - INFO: Borderline metrics but not critical
        """
        metrics = report['metrics']
        
        critical_count = 0
        warning_count = 0
        
        if metrics['avg_diversity'] < self.alert_thresholds['diversity']:
            critical_count += 1
        elif metrics['avg_diversity'] < self.alert_thresholds['diversity'] * 1.2:
            warning_count += 1
            
        if metrics['avg_entropy'] < self.alert_thresholds['entropy']:
            critical_count += 1
        elif metrics['avg_entropy'] < self.alert_thresholds['entropy'] * 1.2:
            warning_count += 1
            
        if metrics['avg_repetition'] > self.alert_thresholds['repetition']:
            critical_count += 1
        elif metrics['avg_repetition'] > self.alert_thresholds['repetition'] * 0.8:
            warning_count += 1
        
        if critical_count >= 2:
            return 'CRITICAL'
        elif critical_count >= 1 or warning_count >= 2:
            return 'WARNING'
        elif warning_count >= 1:
            return 'INFO'
        else:
            return 'LOW'
    
    def _calculate_severity_score(self, report: Dict) -> float:
        """
        Calculate a normalized severity score (0-1).
        
        Calculation Explanation:
        Combines multiple metrics into a single score to prioritize
        interventions. Higher scores indicate more severe conditions.
        """
        metrics = report['metrics']
        
        # Normalize each metric (0-1 range)
        diversity_severity = max(0, 1 - (metrics['avg_diversity'] / 0.5))
        entropy_severity = max(0, 1 - (metrics['avg_entropy'] / 2.0))
        repetition_severity = min(1, metrics['avg_repetition'] / 0.6)
        
        # Metric weighting
        severity_score = (
            diversity_severity * 0.4 +      # 40% weight to diversity
            entropy_severity * 0.3 +         # 30% weight to entropy
            repetition_severity * 0.3        # 30% weight to repetition
        )
        
        return min(1.0, severity_score)
    
    def email_alert_callback(self, alert: Dict):
        """
        Example callback for email alerts.
        
        Implementation Explanation:
        In production, this would integrate with email services like
        SendGrid, AWS SES, or SMTP server. The example shows the basic
        structure of the email message.
        """
        print(f"📧 EMAIL ALERT: {alert['level']}")
        print(f"Timestamp: {time.ctime(alert['timestamp'])}")
        print(f"Message: {alert['message']}")
        print(f"Severity Score: {alert['severity_score']:.2f}")
        
        # Format metrics for readability
        metrics_text = "\n".join([
            f"  {metric}: {value:.3f}" 
            for metric, value in alert['metrics'].items()
        ])
        print(f"Metrics:\n{metrics_text}")
        
        # Format recommendations
        recommendations_text = "\n".join([
            f"  • {rec}" 
            for rec in alert['recommendations']
        ])
        print(f"Recommendations:\n{recommendations_text}")
        
        # In production: send actual email
        # self.email_service.send_alert_email(alert)
    
    def slack_alert_callback(self, alert: Dict):
        """
        Example callback for Slack alerts.
        
        Integration Explanation:
        Uses Slack Webhooks to send formatted messages
        to specific channels. The format is optimized for readability
        in the Slack interface.
        """
        print(f"💬 SLACK ALERT: {alert['level']}")
        
        # Format message for Slack
        slack_message = {
            "text": f"🚨 {alert['level']}: {alert['message']}",
            "attachments": [
                {
                    "color": self._get_slack_color(alert['level']),
                    "fields": [
                        {
                            "title": "Severity Score",
                            "value": f"{alert['severity_score']:.2f}",
                            "short": True
                        },
                        {
                            "title": "Diversity",
                            "value": f"{alert['metrics']['avg_diversity']:.3f}",
                            "short": True
                        },
                        {
                            "title": "Entropy",
                            "value": f"{alert['metrics']['avg_entropy']:.3f}",
                            "short": True
                        },
                        {
                            "title": "Repetition",
                            "value": f"{alert['metrics']['avg_repetition']:.3f}",
                            "short": True
                        }
                    ],
                    "footer": "Temperature Alert System",
                    "ts": alert['timestamp']
                }
            ]
        }
        
        print(f"Slack Message: {slack_message['text']}")
        
        # In production: send to Slack webhook
        # requests.post(slack_webhook_url, json=slack_message)
    
    def _get_slack_color(self, level: str) -> str:
        """
        Determine Slack message color based on level.
        """
        color_map = {
            'CRITICAL': 'danger',    # Red
            'WARNING': 'warning',     # Yellow
            'INFO': 'good',          # Green
            'LOW': '#36a64f'         # Light green
        }
        return color_map.get(level, 'good')
    
    def dashboard_alert_callback(self, alert: Dict):
        """
        Example callback for dashboard updates.
        
        Dashboard Explanation:
        This callback would update a real-time monitoring dashboard,
        showing current alerts and historical trends.
        """
        print(f"📊 DASHBOARD ALERT: {alert['level']} - {alert['message']}")
        
        # Data for dashboard visualization
        dashboard_data = {
            'alert_id': len(self.alert_history),
            'timestamp': alert['timestamp'],
            'level': alert['level'],
            'severity': alert['severity_score'],
            'metrics': alert['metrics'],
            'trends': alert.get('trends', {}),
            'active_alerts': len([a for a in self.alert_history 
                                 if time.time() - a['timestamp'] < 3600])  # Last hour
        }
        
        print(f"Dashboard Data: {dashboard_data}")
        
        # In production: WebSocket or API call to update dashboard
        # self.dashboard_service.update_alerts(dashboard_data)
    
    def get_alert_statistics(self) -> Dict:
        """
        Calculate statistics on historical alerts.
        
        Analytics Explanation:
        Provides insights on alert patterns to identify
        recurring problems and optimize thresholds.
        """
        if not self.alert_history:
            return {"total_alerts": 0}
        
        # Basic statistics
        total_alerts = len(self.alert_history)
        
        # Distribution by level
        level_counts = {}
        for alert in self.alert_history:
            level = alert['level']
            level_counts[level] = level_counts.get(level, 0) + 1
        
        # Recent alerts (last 24 hours)
        recent_time = time.time() - 86400  # 24 hours ago
        recent_alerts = len([a for a in self.alert_history if a['timestamp'] > recent_time])
        
        # Average severity
        avg_severity = sum(a['severity_score'] for a in self.alert_history) / total_alerts
        
        # Hourly trend
        hourly_distribution = {}
        for alert in self.alert_history:
            hour = time.localtime(alert['timestamp']).tm_hour
            hourly_distribution[hour] = hourly_distribution.get(hour, 0) + 1
        
        return {
            'total_alerts': total_alerts,
            'recent_alerts_24h': recent_alerts,
            'level_distribution': level_counts,
            'average_severity': avg_severity,
            'hourly_distribution': hourly_distribution,
            'most_common_hour': max(hourly_distribution.items(), key=lambda x: x[1])[0] if hourly_distribution else None
        }

# Example usage of the alert system
def demonstrate_alert_system():
    """
    Demonstrate the complete alert system.
    """
    print("=== Alert System Demonstration ===")
    
    # Initialize components
    monitor = TemperatureMonitor()
    alert_system = TemperatureAlertSystem(monitor)
    
    # Register callbacks for different channels
    alert_system.add_alert_callback(alert_system.email_alert_callback)
    alert_system.add_alert_callback(alert_system.slack_alert_callback)
    alert_system.add_alert_callback(alert_system.dashboard_alert_callback)
    
    # Simulate data that causes collapse
    print("\n--- Simulating Normal Conditions ---")
    for i in range(5):
        # Simulate normal logits
        logits = torch.randn(1, 1000)
        tokens = torch.multinomial(torch.softmax(logits, dim=-1), 10).squeeze().tolist()
        monitor.update_metrics(logits, tokens)
        
        # Check alert (should be negative)
        alert_generated = alert_system.check_and_alert()
        print(f"Step {i}: Alert generated = {alert_generated}")
    
    print("\n--- Simulating Collapse Conditions ---")
    for i in range(5):
        # Simulate logits indicating collapse (low entropy)
        logits = torch.randn(1, 1000) * 0.1  # Low variance = low entropy
        tokens = [1, 2, 1, 2, 1, 2, 1, 2, 1, 2]  # Repetitive pattern
        monitor.update_metrics(logits, tokens)
        
        # Check alert (should be positive)
        alert_generated = alert_system.check_and_alert()
        print(f"Step {i}: Alert generated = {alert_generated}")
        
        if alert_generated:
            break  # Stop at first alert for demonstration
    
    # Show statistics
    print("\n--- Alert Statistics ---")
    stats = alert_system.get_alert_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")

if __name__ == "__main__":
    demonstrate_alert_system()

import torch
from transformers import GPT2LMHeadModel, GPT2Tokenizer
from transformers import LogitsProcessor, LogitsProcessorList
import numpy as np

class SemanticTemperatureController:
    """
    Comprehensive semantic temperature control system
    """
    
    def __init__(self, model_name='gpt2'):
        self.tokenizer = GPT2Tokenizer.from_pretrained(model_name)
        self.model = GPT2LMHeadModel.from_pretrained(model_name)
        self.monitor = TemperatureMonitor()
        
    def generate_with_temperature_control(self, prompt: str, max_length: int = 100, 
                                        base_temperature: float = 1.0, 
                                        adaptive: bool = True,
                                        monitor_collapse: bool = True) -> Dict:
        """
        Generate text with advanced temperature control and monitoring
        """
        inputs = self.tokenizer.encode(prompt, return_tensors='pt')
        
        # Choose processor based on settings
        processors = []
        if adaptive:
            processors.append(AdaptiveTemperatureProcessor(base_temperature))
        
        # Generate with monitoring
        with torch.no_grad():
            outputs = self.model.generate(
                inputs,
                max_length=max_length,
                temperature=base_temperature,
                do_sample=True,
                num_return_sequences=1,
                logits_processor=LogitsProcessorList(processors),
                return_dict_in_generate=True,
                output_scores=True
            )
        
        # Decode and analyze
        generated_text = self.tokenizer.decode(
            outputs.sequences[0], skip_special_tokens=True
        )
        
        # Update monitoring
        metrics = {}
        if monitor_collapse and outputs.scores:
            final_logits = outputs.scores[-1]
            generated_tokens = outputs.sequences[0].tolist()
            metrics = self.monitor.update_metrics(final_logits, generated_tokens)
        
        return {
            'text': generated_text,
            'metrics': metrics,
            'collapse_detected': self.monitor.detect_collapse(),
            'diagnostic_report': self.monitor.get_diagnostic_report()
        }
    
    def compare_temperature_settings(self, prompt: str, temperatures: List[float] = [0.3, 0.7, 1.0, 1.5]) -> Dict:
        """
        Compare outputs across different temperature settings
        """
        results = {}
        
        for temp in temperatures:
            result = self.generate_with_temperature_control(
                prompt, 
                max_length=50,
                base_temperature=temp,
                adaptive=False,
                monitor_collapse=True
            )
            results[f'temp_{temp}'] = result
            
        return results
    
    def demonstrate_collapse_detection(self):
        """
        Demonstrate collapse detection with controlled examples
        """
        print("=== Semantic Temperature Collapse Detection Demo ===")
        
        # Test with different scenarios
        test_cases = [
            ("The cat sat on the mat", "Simple repetitive context"),
            ("In quantum computing, qubits exist in superposition", "Complex technical context"),
            ("Once upon a time in a magical forest", "Creative narrative context")
        ]
        
        for prompt, description in test_cases:
            print(f"\nTesting: {description}")
            print(f"Prompt: '{prompt}'")
            print("-" * 50)
            
            # Test with low temperature (likely to cause collapse)
            result_low = self.generate_with_temperature_control(
                prompt, max_length=30, base_temperature=0.2, adaptive=False
            )
            
            # Test with optimal temperature
            result_optimal = self.generate_with_temperature_control(
                prompt, max_length=30, base_temperature=0.8, adaptive=False
            )
            
            print(f"Low Temp (0.2): {result_low['text']}")
            print(f"Collapse Detected: {result_low['collapse_detected']}")
            print(f"Optimal Temp (0.8): {result_optimal['text']}")
            print(f"Collapse Detected: {result_optimal['collapse_detected']}")

# Usage example
if __name__ == "__main__":
    controller = SemanticTemperatureController()
    
    # Demonstrate collapse detection
    controller.demonstrate_collapse_detection()
    
    # Compare temperature settings
    comparison = controller.compare_temperature_settings("The future of AI")
    
    print("\n=== Temperature Comparison ===")
    for temp_key, result in comparison.items():
        print(f"\n{temp_key}:")
        print(f"Text: {result['text']}")
        print(f"Metrics: {result['metrics']}")

from collections import deque
from typing import Dict, List, Tuple
import math

class AdaptiveTemperatureProcessor(LogitsProcessor):
    """
    Processor that dynamically adjusts temperature based on generation patterns.
    
    System Architecture:
    1. Continuous Monitoring: Tracks generated tokens to detect patterns
    2. Statistical Analysis: Calculates diversity and repetition metrics
    3. Dynamic Adaptation: Modifies temperature in real-time
    4. Proactive Prevention: Anticipates problems before they manifest
    
    Use Cases:
    - Creative generation: prevents repetitive loops
    - Conversational dialogues: maintains naturalness
    - Technical writing: ensures coherence without repetition
    """
    
    def __init__(self, base_temperature: float = 1.0, min_temp: float = 0.1, max_temp: float = 2.0):
        """
        Initialize the adaptive processor.
        
        Parameters:
        - base_temperature: Starting temperature
        - min_temp: Minimum allowed temperature
        - max_temp: Maximum allowed temperature
        
        Architectural Explanation:
        The system uses different time windows to analyze patterns:
        - token_history: Complete history for long-term analysis
        - repetition_window: Short window to detect immediate repetitions
        - diversity_tracker: Tracks diversity over time
        """
        self.base_temperature = base_temperature
        self.min_temp = min_temp
        self.max_temp = max_temp
        
        # Temporal analysis windows
        self.token_history = deque(maxlen=100)      # Long term
        self.repetition_window = deque(maxlen=10)    # Short term
        self.diversity_tracker = deque(maxlen=50)    # Diversity tracking
        
        # Calculated metrics
        self.current_repetition_ratio = 0.0
        self.current_entropy = 0.0
        self.current_diversity = 0.0
        
        # Adjustment history
        self.adjustment_history = deque(maxlen=20)
        
    def __call__(self, input_ids: torch.Tensor, scores: torch.Tensor) -> torch.Tensor:
        """
        Process logits by applying adaptive temperature.
        
        Parameters:
        - input_ids: IDs of tokens generated so far
        - scores: Current model logits
        
        Returns:
        - Logits with temperature applied
        
        Logical Explanation:
        1. Update metrics with recent tokens
        2. Analyze problematic patterns
        3. Calculate optimal temperature
        4. Apply temperature to logits
        """
        # Update state with recent tokens
        self._update_token_history(input_ids)
        
        # Calculate current metrics
        self._calculate_metrics()
        
        # Determine adaptive temperature
        adjusted_temperature = self._calculate_adaptive_temperature()
        
        # Record adjustment
        self.adjustment_history.append({
            'timestamp': time.time(),
            'original_temp': self.base_temperature,
            'adjusted_temp': adjusted_temperature,
            'repetition_ratio': self.current_repetition_ratio,
            'entropy': self.current_entropy,
            'diversity': self.current_diversity
        })
        
        # Apply temperature to logits
        scores = scores / adjusted_temperature
        
        return scores
    
    def _update_token_history(self, input_ids: torch.Tensor):
        """
        Update token history for analysis.
        
        Implementation Explanation:
        Maintains different time windows for analysis at different scales:
        - Short window: detects immediate repetitions
        - Medium window: analyzes recent patterns
        - Long window: identifies long-term trends
        """
        if input_ids.size(1) > 0:
            # Get last generated token
            last_token = input_ids[0, -1].item()
            
            # Update different windows
            self.token_history.append(last_token)
            self.repetition_window.append(last_token)
            
            # Calculate and track diversity
            if len(self.token_history) >= 5:
                recent_tokens = list(self.token_history)[-10:]
                diversity = len(set(recent_tokens)) / len(recent_tokens)
                self.diversity_tracker.append(diversity)
    
    def _calculate_metrics(self):
        """
        Calculate metrics for quality assessment.
        
        Metrics Explanation:
        1. Repetition Ratio: Percentage of repeated tokens
        2. Entropy: Variety of token distribution
        3. Diversity: Uniqueness of recent tokens
        
        These metrics allow identification of different types of problems:
        - High repetition → low temperature
        - Low diversity → gradual temperature increase
        - Low entropy → possible semantic collapse
        """
        # Calculate repetition ratio
        if len(self.repetition_window) >= 3:
            unique_tokens = len(set(self.repetition_window))
            self.current_repetition_ratio = 1 - (unique_tokens / len(self.repetition_window))
        else:
            self.current_repetition_ratio = 0.0
        
        # Calculate current diversity
        if len(self.diversity_tracker) > 0:
            self.current_diversity = list(self.diversity_tracker)[-1]
        else:
            self.current_diversity = 1.0
        
        # Calculate entropy based on token distribution
        if len(self.token_history) >= 10:
            token_counts = {}
            for token in list(self.token_history)[-20:]:
                token_counts[token] = token_counts.get(token, 0) + 1
            
            # Calculate Shannon entropy
            total_tokens = sum(token_counts.values())
            probabilities = [count / total_tokens for count in token_counts.values()]
            self.current_entropy = -sum(p * math.log(p + 1e-8) for p in probabilities)
        else:
            self.current_entropy = 2.0  # Neutral value
    
    def _calculate_adaptive_temperature(self) -> float:
        """
        Calculate optimal temperature based on current metrics.
        
        Algorithm Explanation:
        Uses a combination of adjustment strategies:
        1. Repetition-based adjustment (high priority)
        2. Diversity-based adjustment (medium priority)
        3. Entropy-based adjustment (low priority)
        4. Safety limits to avoid extremes
        """
        # Start with base temperature
        adjusted_temp = self.base_temperature
        
        # 1. Adjustment for high repetition (high priority)
        if self.current_repetition_ratio > 0.6:
            # High repetition → significantly increase temperature
            adjustment_factor = 1.5 + (self.current_repetition_ratio - 0.6) * 2
            adjusted_temp = min(self.base_temperature * adjustment_factor, self.max_temp)
        elif self.current_repetition_ratio > 0.4:
            # Medium repetition → moderate increase
            adjusted_temp = self.base_temperature * 1.2
        elif self.current_repetition_ratio < 0.1:
            # Low repetition → slight reduction
            adjusted_temp = max(self.base_temperature * 0.9, self.min_temp)
        
        # 2. Adjustment for low diversity
        if self.current_diversity < 0.3:
            # Low diversity → increase temperature
            diversity_factor = 1.3 + (0.3 - self.current_diversity)
            adjusted_temp = min(adjusted_temp * diversity_factor, self.max_temp)
        elif self.current_diversity > 0.8:
            # High diversity → might slightly reduce
            adjusted_temp = max(adjusted_temp * 0.95, self.min_temp)
        
        # 3. Adjustment for low entropy
        if self.current_entropy < 1.0:
            # Low entropy → possible semantic collapse
            entropy_factor = 1.2 + (1.0 - self.current_entropy) * 0.5
            adjusted_temp = min(adjusted_temp * entropy_factor, self.max_temp)
        
        # 4. Apply safety limits
        adjusted_temp = max(self.min_temp, min(self.max_temp, adjusted_temp))
        
        # 5. Apply damping to avoid abrupt changes
        if len(self.adjustment_history) > 0:
            last_temp = self.adjustment_history[-1]['adjusted_temp']
            max_change = 0.3  # Maximum change per step
            adjusted_temp = max(last_temp - max_change, min(last_temp + max_change, adjusted_temp))
        
        return adjusted_temp
    
    def get_diagnostic_info(self) -> Dict:
        """
        Return diagnostic information about processor state.
        
        Output Explanation:
        Provides a complete view of system state for debugging
        and optimization. Includes current metrics, adjustment history
        and recommendations.
        """
        return {
            'current_metrics': {
                'repetition_ratio': self.current_repetition_ratio,
                'entropy': self.current_entropy,
                'diversity': self.current_diversity
            },
            'adjustment_history': list(self.adjustment_history)[-5:],  # Last 5 adjustments
            'recent_tokens': list(self.token_history)[-10:],           # Last 10 tokens
            'recommendations': self._generate_recommendations()
        }
    
    def _generate_recommendations(self) -> List[str]:
        """
        Generate recommendations based on current state.
        
        Logical Explanation:
        Analyzes current metrics and provides suggestions
        to improve generation quality.
        """
        recommendations = []
        
        if self.current_repetition_ratio > 0.6:
            recommendations.append("High repetition detected: consider significant temperature increase")
        elif self.current_repetition_ratio > 0.4:
            recommendations.append("Medium repetition: monitor emerging patterns")
        
        if self.current_diversity < 0.3:
            recommendations.append("Low diversity: increase temperature or reformat prompt")
        
        if self.current_entropy < 1.0:
            recommendations.append("Low entropy: possible semantic collapse in progress")
        
        if len(self.adjustment_history) >= 5:
            recent_adjustments = list(self.adjustment_history)[-5:]
            avg_adjustment = sum(a['adjusted_temp'] for a in recent_adjustments) / 5
            if avg_adjustment > self.base_temperature * 1.3:
                recommendations.append("Frequent upward adjustments: reconsider base temperature")
        
        return recommendations if recommendations else ["Metrics normal: stable generation"]

class ContextAwareTemperatureProcessor(LogitsProcessor):
    """
    Processor that adjusts temperature based on semantic context.
    
    System Architecture:
    1. Contextual Analysis: Identifies the type of content generated
    2. Semantic Classification: Determines the nature of the text
    3. Contextual Adaptation: Modifies temperature based on context
    4. Continuous Learning: Improves decisions over time
    
    Use Cases:
    - Creative writing: higher temperature for originality
    - Technical documentation: lower temperature for precision
    - Dialogues: balanced temperature for naturalness
    """
    
    def __init__(self, base_temperature: float = 1.0):
        """
        Initialize the context-aware processor.
        
        Parameters:
        - base_temperature: Starting temperature
        
        Architectural Explanation:
        The system analyzes context at multiple levels:
        - Lexical: keywords and terminology
        - Syntactic: sentence structure
        - Semantic: meaning and thematic domain
        - Pragmatic: communicative intent
        """
        self.base_temperature = base_temperature
        
        # Dictionaries for contextual classification
        self.creative_keywords = [
            'story', 'imagine', 'creative', 'fiction', 'fantasy', 'dream', 
            'magical', 'adventure', 'poetry', 'art', 'invent', 'novel',
            'tale', 'legend', 'myth', 'whimsical', 'surreal', 'abstract'
        ]
        
        self.technical_keywords = [
            'algorithm', 'function', 'method', 'technical', 'scientific',
            'research', 'analysis', 'data', 'system', 'process', 'protocol',
            'specification', 'implementation', 'architecture', 'framework',
            'methodology', 'procedure', 'standard', 'documentation'
        ]
        
        self.conversational_keywords = [
            'hello', 'thank', 'please', 'sorry', 'feel', 'think', 'believe',
            'opinion', 'experience', 'conversation', 'dialogue', 'discuss',
            'chat', 'talk', 'communicate', 'interact', 'exchange'
        ]
        
        self.emotional_keywords = [
            'happy', 'sad', 'angry', 'excited', 'worried', 'confused',
            'frustrated', 'delighted', 'disappointed', 'surprised', 'proud',
            'emotional', 'feeling', 'sentiment', 'mood', 'atmosphere'
        ]
        
        # Decision history for learning
        self.decision_history = deque(maxlen=50)
        
    def __call__(self, input_ids: torch.Tensor, scores: torch.Tensor) -> torch.Tensor:
        """
        Process logits by applying context-aware temperature.
        
        Logical Explanation:
        1. Analyze recent context
        2. Classify content type
        3. Determine optimal temperature
        4. Apply temperature to logits
        5. Record decision for future improvement
        """
        # Analyze context
        context_analysis = self._analyze_context(input_ids)
        
        # Determine temperature based on context
        adjusted_temperature = self._calculate_context_temperature(context_analysis)
        
        # Record decision
        self.decision_history.append({
            'timestamp': time.time(),
            'context_analysis': context_analysis,
            'adjusted_temperature': adjusted_temperature,
            'base_temperature': self.base_temperature
        })
        
        # Apply temperature
        scores = scores / adjusted_temperature
        
        return scores
    
    def _analyze_context(self, input_ids: torch.Tensor) -> Dict:
        """
        Analyze the context of generated text.
        
        Analysis Explanation:
        Uses multiple techniques to understand context:
        1. Lexical analysis: counts keywords
        2. Structural analysis: examines sentence length
        3. Semantic analysis: identifies thematic patterns
        4. Pragmatic analysis: infers communicative intent
        """
        context_info = {}
        
        if input_ids.size(1) > 0:
            # Get recent context (last 50 tokens)
            recent_tokens = input_ids[0, -50:] if input_ids.size(1) >= 50 else input_ids[0]
            context_text = self._decode_tokens(recent_tokens).lower()
            
            # Lexical analysis
            context_info['creative_score'] = sum(1 for word in self.creative_keywords if word in context_text)
            context_info['technical_score'] = sum(1 for word in self.technical_keywords if word in context_text)
            context_info['conversational_score'] = sum(1 for word in self.conversational_keywords if word in context_text)
            context_info['emotional_score'] = sum(1 for word in self.emotional_keywords if word in context_text)
            
            # Structural analysis
            words = context_text.split()
            context_info['avg_word_length'] = sum(len(word) for word in words) / len(words) if words else 0
            context_info['sentence_count'] = len([s for s in context_text.split('.') if s.strip()])
            
            # Domain analysis
            context_info['domain'] = self._classify_domain(context_info)
            
            # Intent analysis
            context_info['intent'] = self._classify_intent(context_info)
        
        return context_info
    
    def _decode_tokens(self, tokens) -> str:
        """
        Decode tokens to text (placeholder method).
        
        In a real implementation, this would use the model's tokenizer
        to convert token IDs to readable text.
        """
        # Simplified implementation for demonstration
        return " ".join([f"token_{t.item()}" for t in tokens])
    
    def _classify_domain(self, context_info: Dict) -> str:
        """
        Classify the thematic domain of the context.
        
        Classification Explanation:
        Uses rule-based logic to determine the main domain
        of the text based on keyword scores.
        """
        scores = {
            'creative': context_info.get('creative_score', 0),
            'technical': context_info.get('technical_score', 0),
            'conversational': context_info.get('conversational_score', 0),
            'emotional': context_info.get('emotional_score', 0)
        }
        
        # Determine domain with highest score
        max_score = max(scores.values())
        if max_score == 0:
            return 'general'
        
        # Minimum threshold to consider a domain relevant
        threshold = 1
        relevant_domains = [domain for domain, score in scores.items() if score >= threshold]
        
        if not relevant_domains:
            return 'general'
        
        # Return domain with highest score
        return max(relevant_domains, key=lambda d: scores[d])
    
    def _classify_intent(self, context_info: Dict) -> str:
        """
        Classify the communicative intent.
        
        Intent Classification Explanation:
        Analyzes context to infer communicative purpose:
        - Informative: provide information
        - Creative: express creativity
        - Persuasive: convince the reader
        - Conversational: interact with the user
        """
        domain = context_info.get('domain', 'general')
        emotional_score = context_info.get('emotional_score', 0)
        conversational_score = context_info.get('conversational_score', 0)
        
        if domain == 'technical':
            return 'informative'
        elif domain == 'creative':
            return 'creative'
        elif conversational_score > 2:
            return 'conversational'
        elif emotional_score > 2:
            return 'emotional'
        else:
            return 'general'
    
    def _calculate_context_temperature(self, context_analysis: Dict) -> float:
        """
        Calculate optimal temperature based on contextual analysis.
        
        Logical Explanation:
        Uses a context-temperature mapping based on:
        1. Thematic domain: different domains require different temperatures
        2. Communicative intent: intent influences temperature
        3. Structural characteristics: word length, complexity
        4. Decision history: learning from past decisions
        """
        domain = context_analysis.get('domain', 'general')
        intent = context_analysis.get('intent', 'general')
        
        # Base domain-temperature mapping
        domain_temperature_map = {
            'creative': 1.3,      # High temperature for creativity
            'technical': 0.4,      # Low temperature for precision
            'conversational': 0.9, # Medium-high for naturalness
            'emotional': 0.7,      # Medium for controlled expression
            'general': 0.8         # Neutral for general cases
        }
        
        base_adjustment = domain_temperature_map.get(domain, 0.8)
        
        # Adjustments based on intent
        intent_adjustments = {
            'informative': -0.1,    # More precise
            'creative': +0.2,        # More original
            'conversational': +0.1,  # More natural
            'emotional': -0.05,      # Slightly more controlled
            'general': 0.0           # No adjustment
        }
        
        intent_adjustment = intent_adjustments.get(intent, 0.0)
        
        # Adjustments based on structural characteristics
        avg_word_length = context_analysis.get('avg_word_length', 5)
        structural_adjustment = 0.0
        
        if avg_word_length > 7:
            # Long words → technical/complex text → reduce temperature
            structural_adjustment = -0.1
        elif avg_word_length < 4:
            # Short words → simple/conversational text → increase temperature
            structural_adjustment = +0.1
        
        # Calculate final temperature
        adjusted_temperature = self.base_temperature + base_adjustment + intent_adjustment + structural_adjustment
        
        # Apply safety limits
        adjusted_temperature = max(0.2, min(1.8, adjusted_temperature))
        
        return adjusted_temperature
    
    def get_context_insights(self) -> Dict:
        """
        Return insights on analyzed context.
        
        Output Explanation:
        Provides a detailed view of decisions made
        and the reasoning behind temperature adjustments.
        """
        if not self.decision_history:
            return {"status": "No decisions made yet"}
        
        recent_decisions = list(self.decision_history)[-10:]
        
        # Analyze patterns in decisions
        domains = [d['context_analysis'].get('domain', 'general') for d in recent_decisions]
        domain_counts = {domain: domains.count(domain) for domain in set(domains)}
        
        avg_temperatures = [d['adjusted_temperature'] for d in recent_decisions]
        avg_temp = sum(avg_temperatures) / len(avg_temperatures)
        
        return {
            'recent_decisions': recent_decisions[-5:],
            'domain_distribution': domain_counts,
            'average_temperature': avg_temp,
            'temperature_variance': sum((t - avg_temp) ** 2 for t in avg_temperatures) / len(avg_temperatures),
            'recommendations': self._generate_context_recommendations()
        }
    
    def _generate_context_recommendations(self) -> List[str]:
        """
        Generate recommendations based on contextual analysis.
        
        Logical Explanation:
        Analyzes past decisions to provide suggestions
        for improving contextual temperature management.
        """
        recommendations = []
        
        if len(self.decision_history) < 5:
            return ["Insufficient data to generate recommendations"]
        
        recent_decisions = list(self.decision_history)[-10:]
        domains = [d['context_analysis'].get('domain', 'general') for d in recent_decisions]
        
        # Check if there's a dominant domain
        domain_counts = {domain: domains.count(domain) for domain in set(domains)}
        if domain_counts:
            dominant_domain = max(domain_counts, key=domain_counts.get)
            if domain_counts[dominant_domain] >= 7:
                recommendations.append(f"Predominantly '{dominant_domain}' domain: consider base temperature optimized for this domain")
        
        # Check temperature variability
        temps = [d['adjusted_temperature'] for d in recent_decisions]
        temp_variance = sum((t - sum(temps)/len(temps)) ** 2 for t in temps) / len(temps)
        
        if temp_variance > 0.1:
            recommendations.append("High temperature variability: consider more consistent approach")
        elif temp_variance < 0.01:
            recommendations.append("Low temperature variability: possible lack of contextual adaptation")
        
        return recommendations if recommendations else ["Optimal contextual management"]

# Example of combined processor usage
def demonstrate_adaptive_processing():
    """
    Demonstrate combined use of adaptive processors.
    """
    print("=== Adaptive Processing Demonstration ===")
    
    # Initialize processors
    adaptive_processor = AdaptiveTemperatureProcessor(base_temperature=0.8)
    context_processor = ContextAwareTemperatureProcessor(base_temperature=0.8)
    
    # Simulate generation with different contexts
    test_scenarios = [
        ("The quantum algorithm processes data", "technical"),
        ("Once upon a magical dream", "creative"),
        ("Hello, how are you feeling today?", "conversational"),
        ("I feel excited about this opportunity", "emotional")
    ]
    
    for prompt, expected_type in test_scenarios:
        print(f"\n--- Scenario: {expected_type.upper()} ---")
        print(f"Prompt: '{prompt}'")
        
        # Simulate input_ids (in practice would come from tokenizer)
        input_ids = torch.tensor([[1, 2, 3, 4, 5]])  # Placeholder
        
        # Process with adaptive processor
        adaptive_result = adaptive_processor(input_ids, torch.randn(1, 1000))
        adaptive_info = adaptive_processor.get_diagnostic_info()
        
        # Process with context processor
        context_result = context_processor(input_ids, torch.randn(1, 1000))
        context_insights = context_processor.get_context_insights()
        
        print(f"Adaptive Temperature: {adaptive_info['adjustment_history'][-1]['adjusted_temp']:.2f}")
        print(f"Context Temperature: {context_insights.get('average_temperature', 0.8):.2f}")
        print(f"Adaptive Metrics: Repetition={adaptive_info['current_metrics']['repetition_ratio']:.2f}, "
              f"Diversity={adaptive_info['current_metrics']['diversity']:.2f}")
        
        if context_insights.get('domain_distribution'):
            dominant_domain = max(context_insights['domain_distribution'], 
                               key=context_insights['domain_distribution'].get)
            print(f"Detected Domain: {dominant_domain}")

if __name__ == "__main__":
    demonstrate_adaptive_processing()

class ProductionTemperatureIntegration:
    """
    Integration system for production environments.
    
    Integration Architecture:
    1. API Gateway: Unified interface for services
    2. Monitoring Dashboard: Real-time visualization
    3. Alert System: Automatic notifications
    4. Configuration Manager: Centralized settings management
    5. Analytics Engine: Performance analysis
    """
    
    def __init__(self, model_service, monitoring_service):
        """
        Initialize the integration system.
        
        Parameters:
        - model_service: Main language model service
        - monitoring_service: Existing monitoring service
        
        Architectural Explanation:
        The system integrates with existing infrastructure through:
        - Wrapper pattern to avoid modifying existing code
        - Event-driven architecture for asynchronous communication
        - Plugin system for extensibility
        - Configuration as Code for settings management
        """
        self.model_service = model_service
        self.monitoring_service = monitoring_service
        
        # Temperature management components
        self.adaptive_processor = AdaptiveTemperatureProcessor()
        self.context_processor = ContextAwareTemperatureProcessor()
        self.temperature_monitor = TemperatureMonitor()
        
        # Integration services
        self.api_gateway = TemperatureAPIGateway()
        self.dashboard = TemperatureDashboard()
        self.alert_manager = TemperatureAlertManager()
        self.config_manager = TemperatureConfigManager()
        
    def generate_with_temperature_control(self, request):
        """
        Generate text with complete temperature control.
        
        Flow Explanation:
        1. Request validation
        2. Context analysis
        3. Temperature strategy selection
        4. Generation with monitoring
        5. Post-processing and validation
        6. Logging and analytics
        """
        try:
            # 1. Request validation
            validated_request = self._validate_request(request)
            
            # 2. Context analysis
            context_analysis = self._analyze_request_context(validated_request)
            
            # 3. Temperature strategy selection
            temperature_strategy = self._select_temperature_strategy(context_analysis)
            
            # 4. Generation with monitoring
            generation_result = self._generate_with_monitoring(
                validated_request, temperature_strategy
            )
            
            # 5. Post-processing
            processed_result = self._post_process_result(generation_result, context_analysis)
            
            # 6. Logging and analytics
            self._log_generation_event(validated_request, processed_result, context_analysis)
            
            return processed_result
            
        except Exception as e:
            self._handle_generation_error(e, request)
            raise
    
    def _validate_request(self, request):
        """
        Validate generation request.
        
        Validation Explanation:
        Verifies that the request contains all required fields
        and that values are within acceptable ranges.
        """
        required_fields = ['prompt', 'max_length']
        for field in required_fields:
            if field not in request:
                raise ValueError(f"Missing required field: {field}")
        
        # Validazione dei valori
        if request['max_length'] < 1 or request['max_length'] > 5000:
            raise ValueError("max_length must be between 1 and 5000")
        
        if 'temperature' in request:
            temp = request['temperature']
            if temp < 0.1 or temp > 2.0:
                raise ValueError("temperature must be between 0.1 and 2.0")
        
        return request
    
    def _analyze_request_context(self, request):
        """
        Analyze request context.
        
        Analysis Explanation:
        Extracts contextual information from the request to
        determine the optimal temperature strategy.
        """
        context = {
            'prompt_length': len(request['prompt']),
            'prompt_complexity': self._calculate_prompt_complexity(request['prompt']),
            'user_preferences': request.get('user_preferences', {}),
            'application_context': request.get('application_context', 'general'),
            'quality_requirements': request.get('quality_requirements', {})
        }
        
        # Semantic analysis of prompt
        context['semantic_analysis'] = self._analyze_prompt_semantics(request['prompt'])
        
        return context
    
    def _calculate_prompt_complexity(self, prompt):
        """
        Calculate prompt complexity.
        
        Metric Explanation:
        Uses multiple metrics to assess complexity:
        - Word length
        - Vocabulary diversity
        - Syntactic structure
        - Semantic complexity
        """
        words = prompt.split()
        
        # Lexical metrics
        avg_word_length = sum(len(word) for word in words) / len(words) if words else 0
        vocabulary_diversity = len(set(words)) / len(words) if words else 1
        
        # Structural metrics
        sentence_count = len([s for s in prompt.split('.') if s.strip()])
        avg_sentence_length = len(words) / sentence_count if sentence_count > 0 else len(words)
        
        # Complexity score calculation (0-1)
        complexity_score = (
            min(avg_word_length / 10, 1) * 0.3 +           # Word length
            (1 - vocabulary_diversity) * 0.2 +              # Vocabulary diversity
            min(avg_sentence_length / 30, 1) * 0.3 +        # Sentence length
            len(prompt.split()) / 100 * 0.2                  # Total length
        )
        
        return min(complexity_score, 1.0)
    
    def _analyze_prompt_semantics(self, prompt):
        """
        Analyze semantic aspects of the prompt.
        
        Analysis Explanation:
        Identifies semantic characteristics that influence
        the choice of optimal temperature.
        """
        prompt_lower = prompt.lower()
        
        semantic_features = {
            'creative_indicators': sum(1 for word in 
                ['create', 'imagine', 'story', 'invent', 'design'] 
                if word in prompt_lower),
            'technical_indicators': sum(1 for word in 
                ['analyze', 'calculate', 'implement', 'algorithm', 'technical'] 
                if word in prompt_lower),
            'question_indicators': sum(1 for word in 
                ['what', 'how', 'why', 'when', 'where', 'explain'] 
                if word in prompt_lower),
            'emotional_indicators': sum(1 for word in 
                ['feel', 'emotion', 'happy', 'sad', 'excited'] 
                if word in prompt_lower)
        }
        
        # Determine primary type
        max_score = max(semantic_features.values())
        if max_score == 0:
            primary_type = 'general'
        else:
            primary_type = max(semantic_features, key=semantic_features.get)
        
        semantic_features['primary_type'] = primary_type
        semantic_features['confidence'] = max_score / max(sum(semantic_features.values()), 1)
        
        return semantic_features
    
    def _select_temperature_strategy(self, context):
        """
        Select temperature strategy based on context.
        
        Selection Explanation:
        Uses rule-based logic to determine
        which temperature strategy to use.
        """
        semantic_analysis = context['semantic_analysis']
        primary_type = semantic_analysis['primary_type']
        complexity = context['prompt_complexity']
        
        # Available strategies
        strategies = {
            'adaptive_only': {
                'processor': self.adaptive_processor,
                'base_temperature': 0.8,
                'description': 'Adaptive processing only'
            },
            'context_only': {
                'processor': self.context_processor,
                'base_temperature': 0.8,
                'description': 'Context-aware processing only'
            },
            'hybrid': {
                'processors': [self.context_processor, self.adaptive_processor],
                'base_temperature': 0.8,
                'description': 'Hybrid approach with both processors'
            },
            'fixed': {
                'processor': None,
                'base_temperature': self._get_fixed_temperature(primary_type),
                'description': 'Fixed temperature based on context'
            }
        }
        
        # Selection logic
        if complexity > 0.7:
            # High complexity → hybrid approach
            return strategies['hybrid']
        elif semantic_analysis['confidence'] > 0.7:
            # High confidence in semantic type → context-aware
            return strategies['context_only']
        elif primary_type in ['creative', 'technical']:
            # Specific types → targeted approach
            return strategies['context_only']
        else:
            # Default → adaptive
            return strategies['adaptive_only']
    
    def _get_fixed_temperature(self, semantic_type):
        """
        Return fixed temperature based on semantic type.
        
        Mapping Explanation:
        Optimized temperatures for different content types.
        """
        temperature_map = {
            'creative': 1.2,
            'technical': 0.4,
            'question': 0.7,
            'emotional': 0.8,
            'general': 0.8
        }
        
        return temperature_map.get(semantic_type, 0.8)
    
    def _generate_with_monitoring(self, request, strategy):
        """
        Generate text with active monitoring.
        
        Process Explanation:
        Executes generation by applying the selected strategy
        and constantly monitoring quality metrics.
        """
        # Prepare generation parameters
        generation_params = {
            'prompt': request['prompt'],
            'max_length': request['max_length'],
            'temperature': strategy['base_temperature'],
            'processors': strategy.get('processors', [strategy.get('processor')]),
            'monitoring': True
        }
        
        # Execute generation
        result = self.model_service.generate(**generation_params)
        
        # Analyze result
        if result.get('scores'):
            # Update monitoring
            self.temperature_monitor.update_metrics(
                result['scores'][-1], 
                result['tokens']
            )
            
            # Check collapse
            collapse_detected = self.temperature_monitor.detect_collapse()
            
            if collapse_detected:
                # Generate alert
                alert_data = {
                    'timestamp': time.time(),
                    'type': 'temperature_collapse',
                    'request_id': request.get('id'),
                    'metrics': self.temperature_monitor.get_diagnostic_report()
                }
                self.alert_manager.send_alert(alert_data)
                
                # Attempt recovery
                result = self._attempt_recovery(request, strategy)
        
        return result
    
    def _attempt_recovery(self, original_request, failed_strategy):
        """
        Attempt to recover from temperature collapse.
        
        Recovery Explanation:
        Implements different recovery strategies:
        1. Temperature increase
        2. Strategy change
        3. Regeneration with modified prompt
        """
        recovery_attempts = [
            # Attempt 1: Increase temperature
            {
                'strategy': 'adaptive_only',
                'base_temperature': failed_strategy['base_temperature'] * 1.5,
                'description': 'Increased temperature'
            },
            # Attempt 2: Change strategy
            {
                'strategy': 'hybrid',
                'base_temperature': 1.0,
                'description': 'Switched to hybrid strategy'
            },
            # Attempt 3: Modified prompt
            {
                'strategy': 'adaptive_only',
                'base_temperature': 1.2,
                'modified_prompt': original_request['prompt'] + "\nBe creative and diverse.",
                'description': 'Modified prompt with diversity instruction'
            }
        ]
        
        for attempt in recovery_attempts:
            try:
                modified_request = original_request.copy()
                modified_request['temperature'] = attempt['base_temperature']
                if 'modified_prompt' in attempt:
                    modified_request['prompt'] = attempt['modified_prompt']
                
                result = self.model_service.generate(**modified_request)
                
                # Verify if recovery worked
                if result.get('scores'):
                    self.temperature_monitor.update_metrics(
                        result['scores'][-1], 
                        result['tokens']
                    )
                    
                    if not self.temperature_monitor.detect_collapse():
                        # Recovery successful
                        result['recovery_info'] = {
                            'successful': True,
                            'strategy_used': attempt['description'],
                            'original_strategy': failed_strategy['description']
                        }
                        return result
                
            except Exception as e:
                # Log error and continue with next attempt
                print(f"Recovery attempt failed: {attempt['description']} - {e}")
                continue
        
        # All attempts failed
        return {
            'text': "Unable to generate diverse content. Please try rephrasing your request.",
            'recovery_info': {
                'successful': False,
                'attempts_made': len(recovery_attempts)
            }
        }
    
    def _post_process_result(self, result, context):
        """
        Apply post-processing to result.
        
        Post-Processing Explanation:
        Improves result quality through:
        1. Filtering inappropriate content
        2. Grammatical correction
        3. Readability optimization
        4. Metadata addition
        """
        processed_result = result.copy()
        
        # Add metadata
        processed_result['metadata'] = {
            'generation_context': context,
            'temperature_used': result.get('temperature_used', 0.8),
            'quality_metrics': self._calculate_quality_metrics(result.get('text', '')),
            'processing_timestamp': time.time()
        }
        
        # Apply safety filters
        processed_result['text'] = self._apply_safety_filters(result.get('text', ''))
        
        # Optimize readability
        processed_result['text'] = self._optimize_readability(processed_result['text'])
        
        return processed_result
    
    def _calculate_quality_metrics(self, text):
        """
        Calculate text quality metrics.
        
        Metrics Explanation:
        Evaluates different aspects of generated text quality.
        """
        words = text.split()
        
        metrics = {
            'word_count': len(words),
            'sentence_count': len([s for s in text.split('.') if s.strip()]),
            'avg_word_length': sum(len(word) for word in words) / len(words) if words else 0,
            'vocabulary_diversity': len(set(words)) / len(words) if words else 1,
            'readability_score': self._calculate_readability(text)
        }
        
        return metrics
    
    def _calculate_readability(self, text):
        """
        Calculate a simplified readability score.
        
        Calculation Explanation:
        Uses a simplified formula based on:
        - Average word length
        - Average sentence length
        """
        words = text.split()
        sentences = [s for s in text.split('.') if s.strip()]
        
        if not words or not sentences:
            return 0.5
        
        avg_word_length = sum(len(word) for word in words) / len(words)
        avg_sentence_length = len(words) / len(sentences)
        
        # Simplified formula (higher = more readable)
        readability = max(0, min(1, 1 - (avg_word_length / 10 + avg_sentence_length / 30) / 2))
        
        return readability
    
    def _apply_safety_filters(self, text):
        """
        Apply safety filters to text.
        
        Filters Explanation:
        Implements basic filters for inappropriate content.
        """
        # List of words to filter (simplified)
        filter_words = ['inappropriate', 'offensive', 'harmful']  # Example
        
        filtered_text = text
        for word in filter_words:
            filtered_text = filtered_text.replace(word, '[FILTERED]')
        
        return filtered_text
    
    def _optimize_readability(self, text):
        """
        Optimize text readability.
        
        Optimization Explanation:
        Applies simple improvements to readability.
        """
        # Add spaces after punctuation if missing
        text = text.replace('.,', '. ,').replace('.,', '. ,')
        
        # Ensure there are spaces after periods
        text = text.replace('.', '. ').replace('  ', ' ')
        
        return text.strip()
    
    def _log_generation_event(self, request, result, context):
        """
        Log generation event for analytics.
        
        Logging Explanation:
        Records detailed information for future analysis.
        """
        log_entry = {
            'timestamp': time.time(),
            'request_id': request.get('id'),
            'prompt_length': len(request['prompt']),
            'context_type': context['semantic_analysis']['primary_type'],
            'strategy_used': result.get('recovery_info', {}).get('strategy_used', 'initial'),
            'temperature_used': result.get('metadata', {}).get('temperature_used', 0.8),
            'quality_score': result.get('metadata', {}).get('quality_metrics', {}).get('readability_score', 0.5),
            'recovery_attempted': 'recovery_info' in result,
            'success': result.get('recovery_info', {}).get('successful', True)
        }
        
        # Send to monitoring service
        self.monitoring_service.log_event(log_entry)
    
    def _handle_generation_error(self, error, request):
        """
        Handle generation errors.
        
        Error Handling Explanation:
        Implements robust error handling.
        """
        error_log = {
            'timestamp': time.time(),
            'error_type': type(error).__name__,
            'error_message': str(error),
            'request_id': request.get('id'),
            'prompt_preview': request.get('prompt', '')[:100] + '...' if len(request.get('prompt', '')) > 100 else request.get('prompt', '')
        }
        
        # Log error
        self.monitoring_service.log_error(error_log)
        
        # Send critical alert
        self.alert_manager.send_critical_alert({
            'type': 'generation_error',
            'error': error_log
        })

# Complete integration example
def demonstrate_production_integration():
    """
    Demonstrate complete integration in production environment.
    """
    print("=== Production Integration Demonstration ===")
    
    # Simulate external services
    class MockModelService:
        def generate(self, **kwargs):
            return {
                'text': f"Generated text with temp {kwargs.get('temperature', 0.8)}",
                'tokens': [1, 2, 3, 4, 5],
                'scores': [torch.randn(1, 1000) for _ in range(5)]
            }
    
    class MockMonitoringService:
        def log_event(self, event):
            print(f"Logged event: {event['timestamp']} - {event['context_type']}")
        
        def log_error(self, error):
            print(f"Logged error: {error['error_type']}")
    
    # Initialize integration system
    integration = ProductionTemperatureIntegration(
        MockModelService(),
        MockMonitoringService()
    )
    
    # Generation tests
    test_requests = [
        {
            'id': 'req_001',
            'prompt': 'Write a creative story about a magical forest',
            'max_length': 200,
            'application_context': 'creative_writing'
        },
        {
            'id': 'req_002', 
            'prompt': 'Explain the quantum computing algorithm',
            'max_length': 150,
            'application_context': 'technical_documentation'
        },
        {
            'id': 'req_003',
            'prompt': 'How are you feeling today?',
            'max_length': 100,
            'application_context': 'conversation'
        }
    ]
    
    for request in test_requests:
        print(f"\n--- Processing Request: {request['id']} ---")
        print(f"Prompt: {request['prompt']}")
        
        try:
            result = integration.generate_with_temperature_control(request)
            print(f"Generated: {result['text'][:100]}...")
            print(f"Quality Score: {result['metadata']['quality_metrics']['readability_score']:.2f}")
            
            if 'recovery_info' in result:
                print(f"Recovery: {result['recovery_info']}")
                
        except Exception as e:
            print(f"Error: {e}")

if __name__ == "__main__":
    demonstrate_production_integration()

class TemperatureCurriculumTrainer:
    """
    Implements curriculum learning for temperature robustness
    """
    
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.curriculum_stages = [
            {'temperature': 0.3, 'duration': 1000, 'description': 'Deterministic phase'},
            {'temperature': 0.7, 'duration': 1000, 'description': 'Balanced phase'},
            {'temperature': 1.0, 'duration': 1000, 'description': 'Creative phase'},
            {'temperature': 1.5, 'duration': 500, 'description': 'High-creativity phase'},
            {'temperature': 0.5, 'duration': 500, 'description': 'Refinement phase'}
        ]
    
    def train_with_curriculum(self, dataset, epochs_per_stage=1):
        """
        Train model following temperature curriculum
        """
        optimizer = torch.optim.AdamW(self.model.parameters(), lr=5e-5)
        
        for stage_idx, stage in enumerate(self.curriculum_stages):
            print(f"Stage {stage_idx + 1}: {stage['description']} (τ={stage['temperature']})")
            
            for epoch in range(epochs_per_stage):
                total_loss = 0
                num_batches = 0
                
                for batch in dataset:
                    # Prepare batch data
                    inputs = self.tokenizer(batch['text'], return_tensors='pt', padding=True, truncation=True)
                    
                    # Forward pass with stage-specific temperature
                    outputs = self.model(**inputs, temperature=stage['temperature'])
                    loss = outputs.loss
                    
                    # Backward pass
                    loss.backward()
                    optimizer.step()
                    optimizer.zero_grad()
                    
                    total_loss += loss.item()
                    num_batches += 1
                
                avg_loss = total_loss / num_batches
                print(f"  Epoch {epoch + 1}: Average Loss = {avg_loss:.4f}")
                
                # Evaluate temperature robustness
                robustness_score = self.evaluate_temperature_robustness(stage['temperature'])
                print(f"  Temperature Robustness: {robustness_score:.3f}")
    
    def evaluate_temperature_robustness(self, test_temperature):
        """
        Evaluate model robustness at specific temperature
        """
        test_prompts = [
            "The future of technology",
            "Once upon a time",
            "Scientific research shows",
            "In conclusion"
        ]
        
        diversity_scores = []
        
        for prompt in test_prompts:
            inputs = self.tokenizer.encode(prompt, return_tensors='pt')
            
            with torch.no_grad():
                outputs = self.model.generate(
                    inputs,
                    max_length=50,
                    temperature=test_temperature,
                    do_sample=True,
                    return_dict_in_generate=True,
                    output_scores=True
                )
            
            generated_text = self.tokenizer.decode(outputs.sequences[0], skip_special_tokens=True)
            diversity = len(set(generated_text.split())) / len(generated_text.split())
            diversity_scores.append(diversity)
        
        return np.mean(diversity_scores)

class TemperatureAwareDistillation:
    """
    Knowledge distillation that accounts for temperature effects
    """
    
    def __init__(self, teacher_model, student_model, tokenizer):
        self.teacher_model = teacher_model
        self.student_model = student_model
        self.tokenizer = tokenizer
        
    def distill_with_temperature_curriculum(self, dataset, temperatures=[0.5, 0.8, 1.2]):
        """
        Distill knowledge using temperature curriculum
        """
        optimizer = torch.optim.AdamW(self.student_model.parameters(), lr=5e-5)
        
        for temp_idx, current_temp in enumerate(temperatures):
            print(f"Distillation Stage {temp_idx + 1}: Temperature = {current_temp}")
            
            total_loss = 0
            num_batches = 0
            
            for batch in dataset:
                inputs = self.tokenizer(batch['text'], return_tensors='pt', padding=True, truncation=True)
                
                # Teacher forward pass
                with torch.no_grad():
                    teacher_outputs = self.teacher_model(**inputs, temperature=current_temp)
                    teacher_logits = teacher_outputs.logits
                
                # Student forward pass
                student_outputs = self.student_model(**inputs, temperature=current_temp)
                student_logits = student_outputs.logits
                
                # Temperature-aware knowledge distillation loss
                loss = self.temperature_aware_kd_loss(
                    student_logits, teacher_logits, current_temp
                )
                
                # Backward pass
                loss.backward()
                optimizer.step()
                optimizer.zero_grad()
                
                total_loss += loss.item()
                num_batches += 1
            
            avg_loss = total_loss / num_batches
            print(f"  Average Loss: {avg_loss:.4f}")
    
    def temperature_aware_kd_loss(self, student_logits, teacher_logits, temperature):
        """
        Knowledge distillation loss that accounts for temperature
        """
        # Apply temperature scaling
        soft_student = torch.softmax(student_logits / temperature, dim=-1)
        soft_teacher = torch.softmax(teacher_logits / temperature, dim=-1)
        
        # KL divergence loss
        kd_loss = torch.nn.functional.kl_div(
            torch.log(soft_student), soft_teacher, reduction='batchmean'
        )
        
        # Temperature weighting (higher temperature = more emphasis on diversity)
        temp_weight = min(temperature / 1.0, 2.0)  # Cap at 2.0
        
        return kd_loss * temp_weight

class TemperatureOptimizationRL:
    """
    Reinforcement learning for optimal temperature selection
    """
    
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.action_space = np.linspace(0.1, 2.0, 20)  # 20 temperature values
        self.q_table = np.zeros((100, len(self.action_space)))  # State-action Q-values
        self.learning_rate = 0.1
        self.discount_factor = 0.95
        self.epsilon = 0.1  # Exploration rate
        
    def get_state_index(self, metrics):
        """
        Convert metrics to state index
        """
        # Discretize continuous metrics
        entropy_bucket = min(int(metrics['entropy'] * 10), 49)
        diversity_bucket = min(int(metrics['diversity_ratio'] * 10), 49)
        
        return entropy_bucket * 2 + diversity_bucket // 25
    
    def select_temperature(self, state_index, training=True):
        """
        Select temperature using epsilon-greedy policy
        """
        if training and np.random.random() < self.epsilon:
            # Explore: random action
            action_idx = np.random.randint(len(self.action_space))
        else:
            # Exploit: best action
            action_idx = np.argmax(self.q_table[state_index])
        
        return self.action_space[action_idx], action_idx
    
    def calculate_reward(self, metrics, temperature):
        """
        Calculate reward based on generation quality
        """
        # Base reward from diversity
        diversity_reward = metrics['diversity_ratio'] * 10
        
        # Entropy reward
        entropy_reward = metrics['entropy'] * 2
        
        # Temperature appropriateness penalty
        if temperature < 0.3:
            temp_penalty = -5  # Too low
        elif temperature > 1.5:
            temp_penalty = -3  # Too high
        else:
            temp_penalty = 0
        
        # Repetition penalty
        if metrics['diversity_ratio'] < 0.3:
            repetition_penalty = -10
        else:
            repetition_penalty = 0
        
        total_reward = diversity_reward + entropy_reward + temp_penalty + repetition_penalty
        return total_reward
    
    def train_episode(self, prompt, max_steps=10):
        """
        Train one episode of temperature optimization
        """
        state = None
        total_reward = 0
        
        for step in range(max_steps):
            # Generate with current temperature
            if state is None:
                # Initial state: use default temperature
                temperature = 0.8
                action_idx = np.argmin(np.abs(self.action_space - temperature))
            else:
                temperature, action_idx = self.select_temperature(state, training=True)
            
            # Generate text and get metrics
            inputs = self.tokenizer.encode(prompt, return_tensors='pt')
            outputs = self.model.generate(
                inputs,
                max_length=20,
                temperature=temperature,
                do_sample=True,
                return_dict_in_generate=True,
                output_scores=True
            )
            
            # Calculate metrics
            if outputs.scores:
                final_logits = outputs.scores[-1]
                probs = torch.softmax(final_logits, dim=-1)
                entropy = -torch.sum(probs * torch.log(probs + 1e-8), dim=-1).mean().item()
                
                generated_tokens = outputs.sequences[0].tolist()
                unique_ratio = len(set(generated_tokens[-10:])) / min(len(generated_tokens), 10)
                
                metrics = {
                    'entropy': entropy,
                    'diversity_ratio': unique_ratio
                }
            else:
                metrics = {'entropy': 1.0, 'diversity_ratio': 1.0}
            
            # Calculate reward
            reward = self.calculate_reward(metrics, temperature)
            total_reward += reward
            
            # Update Q-table
            new_state = self.get_state_index(metrics)
            
            if state is not None:
                old_q = self.q_table[state, action_idx]
                next_max_q = np.max(self.q_table[new_state])
                new_q = old_q + self.learning_rate * (reward + self.discount_factor * next_max_q - old_q)
                self.q_table[state, action_idx] = new_q
            
            state = new_state
        
        return total_reward
    
    def optimize_temperature(self, prompt, num_episodes=100):
        """
        Optimize temperature for a specific prompt
        """
        episode_rewards = []
        
        for episode in range(num_episodes):
            reward = self.train_episode(prompt)
            episode_rewards.append(reward)
            
            if episode % 10 == 0:
                avg_reward = np.mean(episode_rewards[-10:])
                print(f"Episode {episode}: Average Reward = {avg_reward:.2f}")
        
        # Return best temperature found
        final_state = self.get_state_index({'entropy': 2.0, 'diversity_ratio': 0.8})
        best_action_idx = np.argmax(self.q_table[final_state])
        best_temperature = self.action_space[best_action_idx]
        
        return best_temperature, episode_rewards

class CustomerServiceTemperatureManager:
    """
    Specialized temperature management for customer service chatbots
    """
    
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.conversation_history = []
        
    def generate_response(self, user_input, conversation_context=None):
        """
        Generate contextually appropriate customer service response
        """
        # Detect user intent
        intent = self.detect_intent(user_input)
        
        # Determine appropriate temperature based on intent
        temperature = self._get_intent_based_temperature(intent)
        
        # Generate response with temperature control
        prompt = self._build_prompt(user_input, conversation_context)
        
        inputs = self.tokenizer.encode(prompt, return_tensors='pt')
        outputs = self.model.generate(
            inputs,
            max_length=150,
            temperature=temperature,
            do_sample=True,
            num_return_sequences=3,  # Generate multiple candidates
            return_dict_in_generate=True,
            output_scores=True
        )
        
        # Select best response
        best_response = self._select_best_response(outputs, intent)
        
        # Update conversation history
        self.conversation_history.append({
            'user_input': user_input,
            'bot_response': best_response,
            'intent': intent,
            'temperature': temperature
        })
        
        return best_response
    
    def detect_intent(self, user_input):
        """
        Detect user intent for temperature adjustment
        """
        # Simple keyword-based intent detection
        intent_keywords = {
            'complaint': ['complaint', 'problem', 'issue', 'wrong', 'broken'],
            'question': ['what', 'how', 'why', 'when', 'where'],
            'greeting': ['hello', 'hi', 'hey', 'good morning'],
            'technical': ['technical', 'specification', 'feature', 'function'],
            'emotional': ['frustrated', 'angry', 'confused', 'worried']
        }
        
        user_input_lower = user_input.lower()
        
        for intent, keywords in intent_keywords.items():
            if any(keyword in user_input_lower for keyword in keywords):
                return intent
        
        return 'general'
    
    def _get_intent_based_temperature(self, intent):
        """
        Determine temperature based on detected intent
        """
        temperature_map = {
            'complaint': 0.3,      # Low temp: empathetic, consistent
            'question': 0.7,       # Medium temp: informative, clear
            'greeting': 0.9,       # Higher temp: friendly, varied
            'technical': 0.4,      # Low temp: precise, accurate
            'emotional': 0.5,      # Medium-low temp: supportive, careful
            'general': 0.8         # Medium-high temp: helpful, natural
        }
        
        return temperature_map.get(intent, 0.8)
    
    def _build_prompt(self, user_input, conversation_context):
        """
        Build context-aware prompt for generation
        """
        if conversation_context and len(conversation_context) > 0:
            context_str = "\n".join([
                f"User: {turn['user_input']}\nBot: {turn['bot_response']}"
                for turn in conversation_context[-3:]  # Last 3 turns
            ])
            return f"{context_str}\nUser: {user_input}\nBot:"
        else:
            return f"User: {user_input}\nBot:"
    
    def _select_best_response(self, outputs, intent):
        """
        Select the best response from multiple candidates
        """
        candidates = []
        for output in outputs.sequences:
            response = self.tokenizer.decode(output, skip_special_tokens=True)
            candidates.append(response)
        
        # Score candidates based on intent-appropriate criteria
        best_score = -float('inf')
        best_response = candidates[0]
        
        for candidate in candidates:
            score = self._score_response(candidate, intent)
            if score > best_score:
                best_score = score
                best_response = candidate
        
        return best_response
    
    def _score_response(self, response, intent):
        """
        Score response based on intent-specific criteria
        """
        score = 0
        
        # Length appropriateness
        if intent == 'complaint':
            # Complaints need thorough responses
            if len(response.split()) > 20:
                score += 2
        elif intent == 'question':
            # Questions should be answered concisely
            if 10 <= len(response.split()) <= 30:
                score += 2
        
        # Sentiment appropriateness
        if intent == 'emotional':
            # Check for empathetic language
            empathetic_words = ['understand', 'sorry', 'help', 'assist']
            if any(word in response.lower() for word in empathetic_words):
                score += 3
        
        # Avoid repetition
        words = response.lower().split()
        unique_ratio = len(set(words)) / len(words) if words else 0
        score += unique_ratio * 5
        
        return score

class EducationalTemperatureManager:
    """
    Temperature management for educational AI tutors
    """
    
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.student_proficiency = {}  # Track student proficiency levels
        self.topic_difficulty = {}      # Track topic difficulty levels
        
    def generate_explanation(self, student_id, topic, question, proficiency_level=None):
        """
        Generate explanation with temperature adjusted for educational context
        """
        # Determine or retrieve student proficiency
        if proficiency_level is None:
            proficiency_level = self.student_proficiency.get(student_id, 'intermediate')
        
        # Get topic difficulty
        topic_difficulty = self.topic_difficulty.get(topic, 'medium')
        
        # Calculate optimal temperature
        temperature = self._calculate_educational_temperature(proficiency_level, topic_difficulty)
        
        # Build educational prompt
        prompt = self._build_educational_prompt(topic, question, proficiency_level)
        
        # Generate explanation
        inputs = self.tokenizer.encode(prompt, return_tensors='pt')
        outputs = self.model.generate(
            inputs,
            max_length=200,
            temperature=temperature,
            do_sample=True,
            num_return_sequences=1,
            return_dict_in_generate=True,
            output_scores=True
        )
        
        explanation = self.tokenizer.decode(outputs.sequences[0], skip_special_tokens=True)
        
        # Update student proficiency based on interaction
        self._update_student_proficiency(student_id, topic, explanation)
        
        return explanation
    
    def _calculate_educational_temperature(self, proficiency, difficulty):
        """
        Calculate temperature based on educational factors
        """
        # Base temperature matrix
        temp_matrix = {
            'beginner': {'easy': 0.6, 'medium': 0.5, 'hard': 0.4},
            'intermediate': {'easy': 0.8, 'medium': 0.7, 'hard': 0.6},
            'advanced': {'easy': 1.0, 'medium': 0.9, 'hard': 0.8}
        }
        
        base_temp = temp_matrix[proficiency][difficulty]
        
        # Adjust for learning objectives
        if difficulty == 'hard' and proficiency == 'beginner':
            # Simplify complex topics for beginners
            base_temp *= 0.8
        elif difficulty == 'easy' and proficiency == 'advanced':
            # Add depth for advanced students
            base_temp *= 1.2
        
        return max(0.3, min(1.5, base_temp))  # Clamp to reasonable range
    
    def _build_educational_prompt(self, topic, question, proficiency):
        """
        Build context-appropriate educational prompt
        """
        proficiency_instructions = {
            'beginner': "Explain in simple terms with clear examples. Avoid jargon.",
            'intermediate': "Provide a balanced explanation with some technical details.",
            'advanced': "Give a comprehensive explanation with technical depth and nuance."
        }
        
        instruction = proficiency_instructions[proficiency]
        
        return f"Topic: {topic}\nQuestion: {question}\nInstructions: {instruction}\nExplanation:"
    
    def _update_student_proficiency(self, student_id, topic, explanation):
        """
        Update student proficiency model based on interaction
        """
        # This would typically involve more sophisticated tracking
        # For now, we'll use a simple heuristic approach
        if student_id not in self.student_proficiency:
            self.student_proficiency[student_id] = 'beginner'
        
        # Simple progression logic (would be more sophisticated in practice)
        current_level = self.student_proficiency[student_id]
        progression_map = {'beginner': 'intermediate', 'intermediate': 'advanced'}
        
        # In practice, this would be based on student performance metrics
        # For demonstration, we'll randomly progress occasionally
        import random
        if random.random() < 0.1:  # 10% chance of progression
            if current_level in progression_map:
                self.student_proficiency[student_id] = progression_map[current_level]

# Real-world implementation example from a content platform
class ContentPlatformTemperatureManager:
    """
    Production-ready temperature management for content creation
    """
    
    def __init__(self):
        self.model = load_model("gpt-4")  # Hypothetical model loading
        self.monitor = TemperatureMonitor()
        self.content_analyzer = ContentAnalyzer()
        
    def generate_content(self, content_request):
        """
        Generate content with sophisticated temperature control
        """
        # Analyze content requirements
        content_type = content_request.get('type', 'article')
        tone = content_request.get('tone', 'neutral')
        audience = content_request.get('audience', 'general')
        
        # Determine base temperature
        base_temp = self._get_content_temperature(content_type, tone, audience)
        
        # Generate with monitoring
        result = self.generate_with_monitoring(
            prompt=content_request['prompt'],
            max_length=content_request.get('max_length', 500),
            base_temperature=base_temp,
            adaptive=True
        )
        
        # Quality assessment
        quality_score = self.content_analyzer.assess_quality(result['text'])
        
        # Adjust and regenerate if necessary
        if quality_score < 0.7:
            adjusted_temp = base_temp * 1.2  # Increase temperature for more creativity
            result = self.generate_with_monitoring(
                prompt=content_request['prompt'],
                max_length=content_request.get('max_length', 500),
                base_temperature=adjusted_temp,
                adaptive=True
            )
        
        return {
            'content': result['text'],
            'metrics': result['metrics'],
            'quality_score': quality_score,
            'temperature_used': base_temp
        }
    
    def _get_content_temperature(self, content_type, tone, audience):
        """
        Determine optimal temperature for content generation
        """
        # Content type temperature mapping
        type_temps = {
            'technical_article': 0.4,
            'blog_post': 0.8,
            'creative_story': 1.2,
            'marketing_copy': 0.9,
            'social_media': 1.1,
            'tutorial': 0.5
        }
        
        # Tone adjustments
        tone_adjustments = {
            'formal': -0.2,
            'casual': 0.1,
            'professional': -0.1,
            'creative': 0.3,
            'humorous': 0.4
        }
        
        # Audience adjustments
        audience_adjustments = {
            'technical': -0.2,
            'general': 0.0,
            'creative': 0.2,
            'business': -0.1
        }
        
        base_temp = type_temps.get(content_type, 0.8)
        base_temp += tone_adjustments.get(tone, 0.0)
        base_temp += audience_adjustments.get(audience, 0.0)
        
        return max(0.2, min(1.5, base_temp))

class ProductionBestPractices:
    """
    Implementation of production-ready best practices
    """
    
    def __init__(self):
        self.monitor = TemperatureMonitor()
        self.quality_assessor = QualityAssessor()
        self.alert_manager = AlertManager()
        
    def implement_comprehensive_monitoring(self):
        """
        Implement comprehensive monitoring system
        """
        monitoring_config = {
            'metrics_to_track': [
                'diversity_ratio',
                'entropy',
                'repetition_score',
                'vocabulary_usage',
                'semantic_coherence'
            ],
            'alert_thresholds': {
                'diversity_ratio': 0.3,
                'entropy': 1.0,
                'repetition_score': 0.4
            },
            'monitoring_frequency': 'continuous',
            'reporting_schedule': 'hourly'
        }
        
        return self.monitor.setup_monitoring(monitoring_config)
    
    def quality_assessment_pipeline(self, generated_text, context):
        """
        Comprehensive quality assessment pipeline
        """
        quality_metrics = {
            'coherence': self.quality_assessor.assess_coherence(generated_text),
            'relevance': self.quality_assessor.assess_relevance(generated_text, context),
            'creativity': self.quality_assessor.assess_creativity(generated_text),
            'readability': self.quality_assessor.assess_readability(generated_text),
            'temperature_appropriateness': self.assess_temperature_appropriateness(generated_text, context)
        }
        
        overall_quality = np.mean(list(quality_metrics.values()))
        
        return {
            'overall_score': overall_quality,
            'detailed_metrics': quality_metrics,
            'recommendations': self.generate_quality_recommendations(quality_metrics)
        }
    
    def generate_quality_recommendations(self, metrics):
        """
        Generate actionable recommendations based on quality metrics
        """
        recommendations = []
        
        if metrics['coherence'] < 0.7:
            recommendations.append("Reduce temperature to improve coherence")
        if metrics['creativity'] < 0.5:
            recommendations.append("Increase temperature to enhance creativity")
        if metrics['relevance'] < 0.6:
            recommendations.append("Improve prompt engineering and context understanding")
        if metrics['readability'] < 0.7:
            recommendations.append("Adjust sentence structure and vocabulary complexity")
        
        return recommendations