Create monitoring_system.py

monitoring_system.py

@@ -0,0 +1,381 @@
+"""
+Monitoring System - Layer 4: System monitoring and analytics (Free Version)
+Lightweight monitoring for Hugging Face Spaces
+"""
+
+import time
+from typing import Dict, List, Any, Optional
+from datetime import datetime, timedelta
+from dataclasses import dataclass
+import json
+import pandas as pd
+import numpy as np
+
+
+@dataclass
+class Alert:
+    """Monitoring alert"""
+    id: str
+    severity: str  # info, warning, error
+    source: str
+    message: str
+    timestamp: datetime
+    acknowledged: bool = False
+    resolved: bool = False
+
+
+@dataclass
+class PerformanceSnapshot:
+    """Performance snapshot"""
+    timestamp: datetime
+    metrics: Dict[str, Any]
+    agent_status: Dict[str, str]
+
+
+class MonitoringSystem:
+    """Lightweight monitoring system for free Hugging Face Spaces"""
+    
+    def __init__(self, max_history: int = 1000):
+        self.alerts: List[Alert] = []
+        self.performance_history: List[PerformanceSnapshot] = []
+        self.max_history = max_history
+        
+        # System metrics
+        self.system_metrics = {
+            'start_time': datetime.now(),
+            'total_queries': 0,
+            'successful_queries': 0,
+            'failed_queries': 0,
+            'active_sessions': 0,
+            'peak_memory_mb': 0,
+            'average_response_time': 0.0
+        }
+        
+        # Agent metrics
+        self.agent_metrics = {
+            'diagnostician': {'calls': 0, 'avg_time': 0.0, 'errors': 0},
+            'socratic_tutor': {'calls': 0, 'avg_time': 0.0, 'errors': 0},
+            'calculator': {'calls': 0, 'avg_time': 0.0, 'errors': 0},
+            'knowledge_expert': {'calls': 0, 'avg_time': 0.0, 'errors': 0},
+            'practice_generator': {'calls': 0, 'avg_time': 0.0, 'errors': 0},
+            'assessment': {'calls': 0, 'avg_time': 0.0, 'errors': 0}
+        }
+        
+        # Alert rules
+        self.alert_rules = [
+            {
+                'condition': lambda metrics: metrics.get('failed_queries', 0) > 10,
+                'severity': 'warning',
+                'message': 'High number of failed queries detected'
+            },
+            {
+                'condition': lambda metrics: metrics.get('average_response_time', 0) > 5.0,
+                'severity': 'warning',
+                'message': 'Slow response times detected'
+            }
+        ]
+    
+    def update_metrics(self):
+        """Update system metrics (called periodically)"""
+        # Create snapshot
+        snapshot = PerformanceSnapshot(
+            timestamp=datetime.now(),
+            metrics=self.system_metrics.copy(),
+            agent_status={agent: 'active' for agent in self.agent_metrics.keys()}
+        )
+        
+        self.performance_history.append(snapshot)
+        
+        # Keep history within limits
+        if len(self.performance_history) > self.max_history:
+            self.performance_history = self.performance_history[-self.max_history:]
+        
+        # Check alerts
+        self._check_alerts()
+        
+        # Cleanup old alerts
+        self._cleanup_alerts()
+    
+    def record_query(self, success: bool, response_time: float = 0.0):
+        """Record a query"""
+        self.system_metrics['total_queries'] += 1
+        
+        if success:
+            self.system_metrics['successful_queries'] += 1
+        else:
+            self.system_metrics['failed_queries'] += 1
+        
+        # Update average response time
+        n = self.system_metrics['total_queries']
+        current_avg = self.system_metrics['average_response_time']
+        self.system_metrics['average_response_time'] = (
+            (current_avg * (n - 1) + response_time) / n if n > 1 else response_time
+        )
+    
+    def record_agent_call(self, agent_name: str, success: bool, execution_time: float):
+        """Record an agent call"""
+        if agent_name in self.agent_metrics:
+            metrics = self.agent_metrics[agent_name]
+            metrics['calls'] += 1
+            
+            # Update average time
+            n = metrics['calls']
+            current_avg = metrics['avg_time']
+            metrics['avg_time'] = (
+                (current_avg * (n - 1) + execution_time) / n if n > 1 else execution_time
+            )
+            
+            if not success:
+                metrics['errors'] += 1
+    
+    def get_system_metrics(self) -> Dict[str, Any]:
+        """Get current system metrics"""
+        # Calculate uptime
+        uptime = datetime.now() - self.system_metrics['start_time']
+        uptime_hours = uptime.total_seconds() / 3600
+        
+        # Calculate success rate
+        total = self.system_metrics['total_queries']
+        successful = self.system_metrics['successful_queries']
+        success_rate = successful / total if total > 0 else 1.0
+        
+        # Calculate active agents
+        active_agents = sum(1 for metrics in self.agent_metrics.values() if metrics['calls'] > 0)
+        
+        # Get recent performance
+        recent_snapshots = [s for s in self.performance_history 
+                          if s.timestamp > datetime.now() - timedelta(hours=1)]
+        
+        avg_response_time = self.system_metrics['average_response_time']
+        
+        if recent_snapshots:
+            recent_times = [s.metrics.get('average_response_time', 0) for s in recent_snapshots]
+            avg_response_time = sum(recent_times) / len(recent_times)
+        
+        return {
+            'uptime_hours': uptime_hours,
+            'total_queries': total,
+            'successful_queries': successful,
+            'failed_queries': self.system_metrics['failed_queries'],
+            'success_rate': success_rate,
+            'avg_response_time': avg_response_time,
+            'active_agents': active_agents,
+            'knowledge_coverage': self._estimate_knowledge_coverage(),
+            'system_health': self._calculate_system_health()
+        }
+    
+    def _estimate_knowledge_coverage(self) -> float:
+        """Estimate knowledge coverage percentage"""
+        # Simulated based on agent usage
+        total_calls = sum(metrics['calls'] for metrics in self.agent_metrics.values())
+        if total_calls == 0:
+            return 0.7  # Default coverage
+        
+        # More calls to knowledge expert indicates better coverage
+        knowledge_calls = self.agent_metrics.get('knowledge_expert', {}).get('calls', 0)
+        coverage = 0.7 + (knowledge_calls / max(total_calls, 1)) * 0.3
+        
+        return min(coverage, 0.95)
+    
+    def _calculate_system_health(self) -> float:
+        """Calculate overall system health score (0-100)"""
+        # Based on success rate and response time
+        success_rate = self.get_system_metrics()['success_rate']
+        avg_response_time = self.get_system_metrics()['avg_response_time']
+        
+        # Success component (0-70 points)
+        success_score = success_rate * 70
+        
+        # Response time component (0-30 points, faster = better)
+        if avg_response_time < 1.0:
+            response_score = 30
+        elif avg_response_time < 3.0:
+            response_score = 20
+        elif avg_response_time < 5.0:
+            response_score = 10
+        else:
+            response_score = 0
+        
+        health_score = success_score + response_score
+        return min(health_score, 100)
+    
+    def _check_alerts(self):
+        """Check for alert conditions"""
+        metrics = self.get_system_metrics()
+        
+        for rule in self.alert_rules:
+            if rule['condition'](metrics):
+                self._create_alert(
+                    severity=rule['severity'],
+                    source='system',
+                    message=rule['message']
+                )
+    
+    def _create_alert(self, severity: str, source: str, message: str):
+        """Create a new alert"""
+        alert_id = f"alert_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{len(self.alerts)}"
+        
+        alert = Alert(
+            id=alert_id,
+            severity=severity,
+            source=source,
+            message=message,
+            timestamp=datetime.now()
+        )
+        
+        self.alerts.append(alert)
+        print(f"🚨 Alert: {severity.upper()} - {message}")
+    
+    def _cleanup_alerts(self):
+        """Cleanup old alerts"""
+        # Keep only last 50 alerts
+        if len(self.alerts) > 50:
+            self.alerts = self.alerts[-50:]
+        
+        # Auto-resolve old info alerts (24 hours)
+        cutoff = datetime.now() - timedelta(hours=24)
+        for alert in self.alerts:
+            if alert.severity == 'info' and alert.timestamp < cutoff:
+                alert.resolved = True
+    
+    def get_agent_logs(self, limit: int = 20) -> pd.DataFrame:
+        """Get agent activity logs"""
+        logs = []
+        
+        for agent, metrics in self.agent_metrics.items():
+            if metrics['calls'] > 0:
+                logs.append({
+                    'agent': agent,
+                    'calls': metrics['calls'],
+                    'avg_time': metrics['avg_time'],
+                    'error_rate': metrics['errors'] / metrics['calls'] if metrics['calls'] > 0 else 0,
+                    'status': 'active' if metrics['calls'] > 0 else 'inactive'
+                })
+        
+        if logs:
+            df = pd.DataFrame(logs)
+            return df.sort_values('calls', ascending=False).head(limit)
+        
+        return pd.DataFrame(columns=['agent', 'calls', 'avg_time', 'error_rate', 'status'])
+    
+    def get_active_alerts(self) -> List[Alert]:
+        """Get active (unresolved) alerts"""
+        return [a for a in self.alerts if not a.resolved]
+    
+    def acknowledge_alert(self, alert_id: str) -> bool:
+        """Acknowledge an alert"""
+        for alert in self.alerts:
+            if alert.id == alert_id:
+                alert.acknowledged = True
+                return True
+        return False
+    
+    def resolve_alert(self, alert_id: str) -> bool:
+        """Resolve an alert"""
+        for alert in self.alerts:
+            if alert.id == alert_id:
+                alert.resolved = True
+                return True
+        return False
+    
+    def get_performance_report(self, period: str = '1h') -> Dict[str, Any]:
+        """Get performance report for a period"""
+        # Calculate time cutoff
+        if period == '1h':
+            cutoff = datetime.now() - timedelta(hours=1)
+        elif period == '24h':
+            cutoff = datetime.now() - timedelta(hours=24)
+        else:
+            cutoff = datetime.now() - timedelta(minutes=30)
+        
+        # Filter snapshots
+        period_snapshots = [s for s in self.performance_history if s.timestamp > cutoff]
+        
+        if not period_snapshots:
+            current_metrics = self.get_system_metrics()
+            return {
+                'period': period,
+                'data_points': 0,
+                'current_metrics': current_metrics,
+                'trend': 'insufficient_data'
+            }
+        
+        # Calculate period metrics
+        total_queries = sum(s.metrics.get('total_queries', 0) for s in period_snapshots)
+        successful_queries = sum(s.metrics.get('successful_queries', 0) for s in period_snapshots)
+        
+        response_times = [s.metrics.get('average_response_time', 0) for s in period_snapshots]
+        avg_response_time = sum(response_times) / len(response_times) if response_times else 0
+        
+        # Calculate trend
+        if len(period_snapshots) >= 2:
+            first_half = period_snapshots[:len(period_snapshots)//2]
+            second_half = period_snapshots[len(period_snapshots)//2:]
+            
+            first_rate = sum(s.metrics.get('successful_queries', 0) for s in first_half) / max(sum(s.metrics.get('total_queries', 0) for s in first_half), 1)
+            second_rate = sum(s.metrics.get('successful_queries', 0) for s in second_half) / max(sum(s.metrics.get('total_queries', 0) for s in second_half), 1)
+            
+            if second_rate > first_rate * 1.1:
+                trend = 'improving'
+            elif second_rate < first_rate * 0.9:
+                trend = 'declining'
+            else:
+                trend = 'stable'
+        else:
+            trend = 'stable'
+        
+        current_metrics = self.get_system_metrics()
+        
+        return {
+            'period': period,
+            'data_points': len(period_snapshots),
+            'queries_in_period': total_queries,
+            'success_rate_period': successful_queries / total_queries if total_queries > 0 else 0,
+            'avg_response_time_period': avg_response_time,
+            'current_metrics': current_metrics,
+            'trend': trend,
+            'health_score': current_metrics['system_health'],
+            'recommendations': self._generate_recommendations(current_metrics)
+        }
+    
+    def _generate_recommendations(self, metrics: Dict[str, Any]) -> List[str]:
+        """Generate recommendations based on metrics"""
+        recommendations = []
+        
+        if metrics['success_rate'] < 0.8:
+            recommendations.append('Improve query success rate by simplifying complex queries')
+        
+        if metrics['avg_response_time'] > 3.0:
+            recommendations.append('Optimize response time by caching frequent queries')
+        
+        if metrics['knowledge_coverage'] < 0.8:
+            recommendations.append('Expand knowledge base coverage for better answers')
+        
+        if not recommendations:
+            recommendations.append('System performance is good. Maintain current operations.')
+        
+        return recommendations[:3]
+    
+    def reset_metrics(self):
+        """Reset all metrics (for testing or restart)"""
+        self.system_metrics = {
+            'start_time': datetime.now(),
+            'total_queries': 0,
+            'successful_queries': 0,
+            'failed_queries': 0,
+            'active_sessions': 0,
+            'peak_memory_mb': 0,
+            'average_response_time': 0.0
+        }
+        
+        for agent in self.agent_metrics:
+            self.agent_metrics[agent] = {'calls': 0, 'avg_time': 0.0, 'errors': 0}
+        
+        self.performance_history = []
+        self.alerts = []
+        
+        print("🔄 Monitoring metrics reset")
+
+
+# Create global instance
+monitoring_system = MonitoringSystem()