Spaces:
Runtime error
Runtime error
| """ | |
| 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 | |
| class Alert: | |
| """Monitoring alert""" | |
| id: str | |
| severity: str # info, warning, error | |
| source: str | |
| message: str | |
| timestamp: datetime | |
| acknowledged: bool = False | |
| resolved: bool = False | |
| 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() |