""" Performance monitoring dashboard for LangCode Agent """ import os import sys import time import json from datetime import datetime, timedelta from typing import Dict, List, Any, Optional from pathlib import Path import threading # Add current directory to path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) class PerformanceDashboard: """Performance monitoring dashboard""" def __init__(self, data_dir: Optional[str] = None): """ Initialize performance dashboard Args: data_dir: Directory for storing performance data """ if data_dir is None: self.data_dir = Path.home() / ".langcode" / "performance" else: self.data_dir = Path(data_dir) # Create data directory self.data_dir.mkdir(parents=True, exist_ok=True) # Performance metrics self.metrics = { "requests": [], "errors": [], "response_times": [], "tool_usage": {}, "provider_usage": {}, "system_stats": [] } # Load existing data self._load_data() # Start background monitoring self._start_monitoring() def _load_data(self): """Load existing performance data""" data_file = self.data_dir / "performance_data.json" if data_file.exists(): try: with open(data_file, 'r') as f: data = json.load(f) self.metrics.update(data) except Exception as e: print(f"Warning: Could not load performance data: {e}") def _save_data(self): """Save performance data to file""" data_file = self.data_dir / "performance_data.json" try: with open(data_file, 'w') as f: json.dump(self.metrics, f, indent=2) except Exception as e: print(f"Warning: Could not save performance data: {e}") def _start_monitoring(self): """Start background monitoring thread""" def monitor(): while True: time.sleep(60) # Collect data every minute self._collect_system_stats() self._save_data() thread = threading.Thread(target=monitor, daemon=True) thread.start() def _collect_system_stats(self): """Collect system statistics""" try: import psutil stats = { "timestamp": datetime.now().isoformat(), "cpu_percent": psutil.cpu_percent(interval=1), "memory_percent": psutil.virtual_memory().percent, "disk_usage": psutil.disk_usage('/').percent, "network_connections": len(psutil.net_connections()) } self.metrics["system_stats"].append(stats) # Keep only last 1000 entries if len(self.metrics["system_stats"]) > 1000: self.metrics["system_stats"] = self.metrics["system_stats"][-1000:] except ImportError: # psutil not available, skip system stats pass except Exception as e: print(f"Warning: Could not collect system stats: {e}") def record_request(self, endpoint: str, method: str, status_code: int, duration: float): """Record an API request""" request = { "timestamp": datetime.now().isoformat(), "endpoint": endpoint, "method": method, "status_code": status_code, "duration": duration } self.metrics["requests"].append(request) # Keep only last 10000 requests if len(self.metrics["requests"]) > 10000: self.metrics["requests"] = self.metrics["requests"][-10000:] def record_error(self, error_type: str, error_message: str, context: Optional[str] = None): """Record an error""" error = { "timestamp": datetime.now().isoformat(), "type": error_type, "message": error_message, "context": context } self.metrics["errors"].append(error) # Keep only last 1000 errors if len(self.metrics["errors"]) > 1000: self.metrics["errors"] = self.metrics["errors"][-1000:] def record_response_time(self, operation: str, duration: float): """Record response time for an operation""" response_time = { "timestamp": datetime.now().isoformat(), "operation": operation, "duration": duration } self.metrics["response_times"].append(response_time) # Keep only last 5000 response times if len(self.metrics["response_times"]) > 5000: self.metrics["response_times"] = self.metrics["response_times"][-5000:] def record_tool_usage(self, tool_name: str, success: bool, duration: float): """Record tool usage""" if tool_name not in self.metrics["tool_usage"]: self.metrics["tool_usage"][tool_name] = { "total_calls": 0, "successful_calls": 0, "failed_calls": 0, "total_duration": 0, "avg_duration": 0 } tool_stats = self.metrics["tool_usage"][tool_name] tool_stats["total_calls"] += 1 tool_stats["total_duration"] += duration if success: tool_stats["successful_calls"] += 1 else: tool_stats["failed_calls"] += 1 tool_stats["avg_duration"] = tool_stats["total_duration"] / tool_stats["total_calls"] def record_provider_usage(self, provider: str, model: str, success: bool): """Record provider usage""" key = f"{provider}:{model}" if key not in self.metrics["provider_usage"]: self.metrics["provider_usage"][key] = { "total_calls": 0, "successful_calls": 0, "failed_calls": 0 } provider_stats = self.metrics["provider_usage"][key] provider_stats["total_calls"] += 1 if success: provider_stats["successful_calls"] += 1 else: provider_stats["failed_calls"] += 1 def get_summary(self, hours: int = 24) -> Dict[str, Any]: """ Get performance summary for the last N hours Args: hours: Number of hours to include in summary Returns: Dictionary with performance summary """ cutoff_time = datetime.now() - timedelta(hours=hours) # Filter recent data recent_requests = [ r for r in self.metrics["requests"] if datetime.fromisoformat(r["timestamp"]) > cutoff_time ] recent_errors = [ e for e in self.metrics["errors"] if datetime.fromisoformat(e["timestamp"]) > cutoff_time ] recent_response_times = [ rt for rt in self.metrics["response_times"] if datetime.fromisoformat(rt["timestamp"]) > cutoff_time ] # Calculate statistics total_requests = len(recent_requests) total_errors = len(recent_errors) if total_requests > 0: error_rate = total_errors / total_requests avg_response_time = sum(r["duration"] for r in recent_requests) / total_requests else: error_rate = 0 avg_response_time = 0 # Status code distribution status_codes = {} for request in recent_requests: code = request["status_code"] status_codes[code] = status_codes.get(code, 0) + 1 # Endpoint usage endpoints = {} for request in recent_requests: endpoint = request["endpoint"] endpoints[endpoint] = endpoints.get(endpoint, 0) + 1 # Tool usage summary tool_summary = {} for tool_name, stats in self.metrics["tool_usage"].items(): tool_summary[tool_name] = { "total_calls": stats["total_calls"], "success_rate": stats["successful_calls"] / stats["total_calls"] if stats["total_calls"] > 0 else 0, "avg_duration": stats["avg_duration"] } # Provider usage summary provider_summary = {} for provider_key, stats in self.metrics["provider_usage"].items(): provider_summary[provider_key] = { "total_calls": stats["total_calls"], "success_rate": stats["successful_calls"] / stats["total_calls"] if stats["total_calls"] > 0 else 0 } # System stats (if available) system_stats = {} if self.metrics["system_stats"]: latest_stats = self.metrics["system_stats"][-1] system_stats = { "cpu_percent": latest_stats.get("cpu_percent", 0), "memory_percent": latest_stats.get("memory_percent", 0), "disk_usage": latest_stats.get("disk_usage", 0) } return { "period_hours": hours, "total_requests": total_requests, "total_errors": total_errors, "error_rate": error_rate, "avg_response_time": avg_response_time, "status_codes": status_codes, "endpoints": endpoints, "tool_usage": tool_summary, "provider_usage": provider_summary, "system_stats": system_stats, "generated_at": datetime.now().isoformat() } def get_detailed_report(self, hours: int = 24) -> str: """ Get detailed performance report Args: hours: Number of hours to include in report Returns: Formatted report string """ summary = self.get_summary(hours) report = f"📊 Performance Report (Last {hours} hours)\n" report += "=" * 50 + "\n\n" # Request statistics report += f"📈 Request Statistics:\n" report += f" Total Requests: {summary['total_requests']}\n" report += f" Total Errors: {summary['total_errors']}\n" report += f" Error Rate: {summary['error_rate']:.2%}\n" report += f" Avg Response Time: {summary['avg_response_time']:.3f}s\n\n" # Status codes if summary['status_codes']: report += f"📋 Status Code Distribution:\n" for code, count in sorted(summary['status_codes'].items()): report += f" {code}: {count}\n" report += "\n" # Top endpoints if summary['endpoints']: report += f"🔗 Top Endpoints:\n" sorted_endpoints = sorted(summary['endpoints'].items(), key=lambda x: x[1], reverse=True)[:10] for endpoint, count in sorted_endpoints: report += f" {endpoint}: {count}\n" report += "\n" # Tool usage if summary['tool_usage']: report += f"🔧 Tool Usage:\n" for tool_name, stats in summary['tool_usage'].items(): report += f" {tool_name}:\n" report += f" Calls: {stats['total_calls']}\n" report += f" Success Rate: {stats['success_rate']:.2%}\n" report += f" Avg Duration: {stats['avg_duration']:.3f}s\n" report += "\n" # Provider usage if summary['provider_usage']: report += f"🤖 Provider Usage:\n" for provider_key, stats in summary['provider_usage'].items(): report += f" {provider_key}:\n" report += f" Calls: {stats['total_calls']}\n" report += f" Success Rate: {stats['success_rate']:.2%}\n" report += "\n" # System stats if summary['system_stats']: report += f"💻 System Statistics:\n" report += f" CPU Usage: {summary['system_stats']['cpu_percent']}%\n" report += f" Memory Usage: {summary['system_stats']['memory_percent']}%\n" report += f" Disk Usage: {summary['system_stats']['disk_usage']}%\n" return report def clear_old_data(self, days: int = 30): """ Clear data older than specified days Args: days: Number of days to keep data """ cutoff_time = datetime.now() - timedelta(days=days) # Filter out old data self.metrics["requests"] = [ r for r in self.metrics["requests"] if datetime.fromisoformat(r["timestamp"]) > cutoff_time ] self.metrics["errors"] = [ e for e in self.metrics["errors"] if datetime.fromisoformat(e["timestamp"]) > cutoff_time ] self.metrics["response_times"] = [ rt for rt in self.metrics["response_times"] if datetime.fromisoformat(rt["timestamp"]) > cutoff_time ] self.metrics["system_stats"] = [ s for s in self.metrics["system_stats"] if datetime.fromisoformat(s["timestamp"]) > cutoff_time ] # Save cleaned data self._save_data() print(f"✅ Cleared data older than {days} days") # Global performance dashboard instance performance_dashboard = PerformanceDashboard() def get_performance_summary(hours: int = 24) -> Dict[str, Any]: """ Get performance summary Args: hours: Number of hours to include Returns: Performance summary dictionary """ return performance_dashboard.get_summary(hours) def get_performance_report(hours: int = 24) -> str: """ Get formatted performance report Args: hours: Number of hours to include Returns: Formatted report string """ return performance_dashboard.get_detailed_report(hours) if __name__ == "__main__": # Print performance report when run directly print(get_performance_report())