diff --git a/README.md b/README.md index c931301..4fd0ca6 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ Un agente configurabile basato su DeepAgents che supporta piΓΉ provider LLM. - **Docker support**: Containerizzazione per deployment facile - **API REST**: Server API per accesso remoto - **Web UI**: Interfaccia web per gestione configurazioni +- **Comandi integrati**: Comandi speciali con prefisso "/" per gestione rapida ## Installazione @@ -175,7 +176,44 @@ Esegui l'agente: python main.py ``` -Puoi quindi interagire con l'agente nel terminale. Digita 'exit' per uscire. +### Comandi Speciali (prefissati con "/") + +Quando interagisci con l'agente, puoi usare comandi speciali preceduti da "/" per gestire il sistema senza confonderli con i messaggi di chat: + +``` +/help, /? - Mostra i comandi disponibili +/status - Mostra lo stato del sistema +/config list - Elenca tutte le configurazioni +/config active - Mostra la configurazione attiva +/config test - Testa la connessione alla configurazione attiva +/config use - Cambia la configurazione attiva +/models - Elenca i modelli disponibili per il provider attivo +/clear - Pulisce lo schermo +/exit, /quit - Esci dall'agente +``` + +**Esempio di sessione:** +``` +πŸ‘€ You: /status +πŸ“Š Stato del Sistema: +Provider: ollama +Modello: gemma2:9b +Ollama URL: http://192.168.0.50:11434 + +πŸ‘€ You: /config list +πŸ“‹ Configurazioni disponibili: +βœ… mio-ollama: ollama - gemma2:9b + openai-gpt4: openai - gpt-4o + +πŸ‘€ You: Ciao, chi sei? +πŸ€– MaxClaw: Ciao! Sono MaxClaw, un agente AI configurabile... + +πŸ‘€ You: Elenca i file nella directory corrente +πŸ€– MaxClaw: [Esegue ls e mostra i file] + +πŸ‘€ You: /exit +πŸ‘‹ Goodbye! +``` ### Utilizzo con Docker @@ -378,6 +416,15 @@ Per personalizzare l'agente, modifica `main.py`: - Aggiungi strumenti personalizzati alla chiamata `create_deep_agent` - Configura i sub-agenti o altre funzionalitΓ  di DeepAgents +## Monitoraggio e Logging + +MaxClaw Agent include un sistema di monitoraggio e logging: + +- **Log dettagliati**: Tutte le azioni vengono registrate in `~/.maxclaw/logs/` +- **Statistiche di sistema**: Monitoraggio di uptime, richieste, errori +- **Metriche di performance**: Tempi di risposta, utilizzo cache +- **Report di stato**: Visualizzazione completa dello stato del sistema + ## Sicurezza MaxClaw Agent implementa misure di sicurezza avanzate: diff --git a/main.py b/main.py index 70691f6..b6c7c7e 100644 --- a/main.py +++ b/main.py @@ -6,7 +6,7 @@ Enhanced with unified provider tools and custom tools """ import os import sys -from typing import Optional, List +from typing import Optional, List, Dict, Any from deepagents import create_deep_agent from langchain.chat_models import init_chat_model from langchain_core.messages import HumanMessage, SystemMessage @@ -21,6 +21,14 @@ except ImportError: TOOLS_AVAILABLE = False print("Warning: Custom tools not available. Install required dependencies.") +# Import configuration manager +try: + from config_manager import config_manager + CONFIG_AVAILABLE = True +except ImportError: + CONFIG_AVAILABLE = False + print("Warning: Configuration manager not available.") + def load_model_from_config(): """ @@ -140,6 +148,227 @@ def create_maxclaw_agent(system_prompt: Optional[str] = None): return agent +def show_help(): + """Show available commands""" + print(""" +πŸ“– Comandi disponibili (prefissati con /): + + /help, /? - Mostra questo messaggio di aiuto + /status - Mostra lo stato del sistema + /config list - Elenca tutte le configurazioni + /config active - Mostra la configurazione attiva + /config test - Testa la connessione alla configurazione attiva + /config use - Cambia la configurazione attiva + /models - Elenca i modelli disponibili per il provider attivo + /clear - Pulisce lo schermo + /exit, /quit - Esci dall'agente + +πŸ’‘ Per chattare con l'agente, scrivi semplicemente il tuo messaggio senza "/". + Esempio: "Ciao, chi sei?" oppure "Elenca i file nella directory corrente" +""") + + +def show_status(): + """Show system status""" + print("\nπŸ“Š Stato del Sistema:") + print("=" * 40) + + # Show provider info + provider = os.getenv("MODEL_PROVIDER", "non configurato") + model = os.getenv("MODEL_NAME", "non configurato") + print(f"Provider: {provider}") + print(f"Modello: {model}") + + # Show Ollama URL if configured + if provider == "ollama": + ollama_url = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434") + print(f"Ollama URL: {ollama_url}") + + # Show configuration status + if CONFIG_AVAILABLE: + active_config = config_manager.get_active_configuration() + if active_config: + print(f"Configurazione attiva: {active_config}") + else: + print("Configurazione attiva: Nessuna") + + configs = config_manager.get_configurations() + print(f"Configurazioni totali: {len(configs)}") + + # Show tools status + if TOOLS_AVAILABLE: + print("Strumenti personalizzati: Disponibili") + else: + print("Strumenti personalizzati: Non disponibili") + + +def list_configurations(): + """List all configurations""" + if not CONFIG_AVAILABLE: + print("❌ Gestore configurazioni non disponibile.") + return + + configs = config_manager.get_configurations() + active = config_manager.get_active_configuration() + + if not configs: + print("πŸ“­ Nessuna configurazione trovata.") + return + + print("\nπŸ“‹ Configurazioni disponibili:") + print("=" * 50) + + for name, config in configs.items(): + marker = "βœ…" if name == active else " " + provider = config.get("provider", "unknown") + model = config.get("model_name", "unknown") + print(f"{marker} {name}: {provider} - {model}") + + +def show_active_configuration(): + """Show active configuration""" + if not CONFIG_AVAILABLE: + print("❌ Gestore configurazioni non disponibile.") + return + + active_name = config_manager.get_active_configuration() + if not active_name: + print("πŸ“­ Nessuna configurazione attiva.") + return + + config = config_manager.get_configuration(active_name) + if not config: + print(f"❌ Configurazione '{active_name}' non trovata.") + return + + print(f"\nβœ… Configurazione attiva: {active_name}") + print(f" Provider: {config.get('provider', 'unknown')}") + print(f" Modello: {config.get('model_name', 'unknown')}") + + if config.get('base_url'): + print(f" URL: {config.get('base_url')}") + + +def test_active_configuration(): + """Test connection to active configuration""" + if not CONFIG_AVAILABLE: + print("❌ Gestore configurazioni non disponibile.") + return + + active_name = config_manager.get_active_configuration() + if not active_name: + print("πŸ“­ Nessuna configurazione attiva da testare.") + return + + print(f"\nπŸ” Test connessione a '{active_name}'...") + success, message = config_manager.test_configuration_connection(active_name) + + if success: + print(f"βœ… {message}") + else: + print(f"❌ {message}") + + +def use_configuration(name: str): + """Set active configuration""" + if not CONFIG_AVAILABLE: + print("❌ Gestore configurazioni non disponibile.") + return + + if not config_manager.configuration_exists(name): + print(f"❌ Configurazione '{name}' non trovata.") + print("Usa '/config list' per vedere le configurazioni disponibili.") + return + + if config_manager.set_active_configuration(name): + print(f"βœ… Configurazione '{name}' impostata come attiva.") + else: + print(f"❌ Errore nell'impostare la configurazione '{name}'.") + + +def list_models(): + """List available models for current provider""" + provider = os.getenv("MODEL_PROVIDER", "").lower() + + if not provider: + print("❌ Nessun provider configurato.") + return + + print(f"\nπŸ“‹ Modelli disponibili per {provider}:") + print("=" * 40) + + # This is a simplified version - in a real implementation, + # you would call the provider's API to list models + if provider == "ollama": + print("Per elencare i modelli Ollama, chiedi all'agente:") + print(" 'Elenca i modelli disponibili su Ollama'") + elif provider == "openai": + print("Modelli OpenAI comuni:") + print(" - gpt-4o") + print(" - gpt-4o-mini") + print(" - gpt-4-turbo") + print(" - gpt-3.5-turbo") + elif provider == "openrouter": + print("Per elencare i modelli OpenRouter, chiedi all'agente:") + print(" 'Elenca i modelli disponibili su OpenRouter'") + else: + print(f"Provider '{provider}' non riconosciuto.") + + +def clear_screen(): + """Clear the screen""" + os.system('cls' if os.name == 'nt' else 'clear') + + +def process_command(user_input: str) -> bool: + """ + Process a command starting with "/" + + Returns: + True if it was a command, False otherwise + """ + if not user_input.startswith('/'): + return False + + # Remove the "/" and split into command and arguments + command_line = user_input[1:].strip() + parts = command_line.split() + command = parts[0].lower() if parts else "" + args = parts[1:] if len(parts) > 1 else [] + + # Process commands + if command in ['help', '?']: + show_help() + elif command == 'status': + show_status() + elif command == 'config': + if not args: + print("❌ Specifica un sottocomando: list, active, test, use ") + elif args[0] == 'list': + list_configurations() + elif args[0] == 'active': + show_active_configuration() + elif args[0] == 'test': + test_active_configuration() + elif args[0] == 'use' and len(args) > 1: + use_configuration(args[1]) + else: + print(f"❌ Sottocomando sconosciuto: {args[0]}") + print("Sottocomandi disponibili: list, active, test, use ") + elif command == 'models': + list_models() + elif command == 'clear': + clear_screen() + elif command in ['exit', 'quit']: + print("πŸ‘‹ Goodbye!") + sys.exit(0) + else: + print(f"❌ Comando sconosciuto: /{command}") + print("Digita /help per vedere i comandi disponibili.") + + return True + + def main(): """ Main entry point for the MaxClaw agent. @@ -156,23 +385,25 @@ def main(): print(" β€’ Proxmox tools: manage VMs, snapshots, resources") print(" β€’ Standard tools: planning, filesystem, shell, subagents") - print("\nπŸ’¬ You can now interact with the agent. Type 'exit' to quit.") - print("πŸ’‘ Try asking:") - print(" - 'List available models for my provider'") - print(" - 'Test connection to Ollama'") - print(" - 'Show me Proxmox VMs'") - print(" - 'What's the default model for OpenAI?'") + print("\nπŸ’¬ You can now interact with the agent.") + print("πŸ“Œ Comandi speciali (prefissati con /):") + print(" /help - Mostra i comandi disponibili") + print(" /status - Mostra lo stato del sistema") + print(" /config - Gestisci le configurazioni") + print(" /exit - Esci dall'agente") + print("\nπŸ’‘ Per chattare, scrivi semplicemente il tuo messaggio.") while True: user_input = input("\nπŸ‘€ You: ").strip() - if user_input.lower() in ['exit', 'quit']: - print("πŸ‘‹ Goodbye!") - break if not user_input: continue - - # Invoke the agent + + # Check if it's a command (starts with /) + if process_command(user_input): + continue + + # Otherwise, treat it as a chat message try: result = agent.invoke({"messages": [HumanMessage(content=user_input)]}) # Extract the AI's response diff --git a/monitoring.py b/monitoring.py new file mode 100644 index 0000000..0ad5e45 --- /dev/null +++ b/monitoring.py @@ -0,0 +1,259 @@ +""" +Monitoring and logging for MaxClaw Agent +""" +import os +import sys +import logging +import json +import time +from datetime import datetime +from typing import Dict, Any, Optional +from pathlib import Path + +# Add current directory to path +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + + +class MaxClawLogger: + """Custom logger for MaxClaw Agent""" + + def __init__(self, log_dir: Optional[str] = None): + """ + Initialize logger + + Args: + log_dir: Directory for log files. If None, uses ~/.maxclaw/logs + """ + if log_dir is None: + self.log_dir = Path.home() / ".maxclaw" / "logs" + else: + self.log_dir = Path(log_dir) + + # Create log directory + self.log_dir.mkdir(parents=True, exist_ok=True) + + # Set up logging + self.setup_logging() + + def setup_logging(self): + """Set up logging configuration""" + # Create formatters + file_formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + console_formatter = logging.Formatter( + '%(asctime)s - %(levelname)s - %(message)s' + ) + + # Create file handler + log_file = self.log_dir / f"maxclaw_{datetime.now().strftime('%Y%m%d')}.log" + file_handler = logging.FileHandler(log_file) + file_handler.setLevel(logging.INFO) + file_handler.setFormatter(file_formatter) + + # Create console handler + console_handler = logging.StreamHandler() + console_handler.setLevel(logging.INFO) + console_handler.setFormatter(console_formatter) + + # Configure root logger + logging.basicConfig( + level=logging.INFO, + handlers=[file_handler, console_handler] + ) + + # Create specific loggers + self.agent_logger = logging.getLogger("maxclaw.agent") + self.config_logger = logging.getLogger("maxclaw.config") + self.api_logger = logging.getLogger("maxclaw.api") + self.tools_logger = logging.getLogger("maxclaw.tools") + + print(f"βœ… Logging initialized. Logs will be saved to: {self.log_dir}") + + def log_agent_action(self, action: str, details: Optional[Dict[str, Any]] = None): + """Log agent actions""" + message = f"Agent action: {action}" + if details: + message += f" - {json.dumps(details)}" + self.agent_logger.info(message) + + def log_config_change(self, action: str, config_name: str, details: Optional[Dict[str, Any]] = None): + """Log configuration changes""" + message = f"Config {action}: {config_name}" + if details: + message += f" - {json.dumps(details)}" + self.config_logger.info(message) + + def log_api_request(self, method: str, endpoint: str, status_code: int, duration: float): + """Log API requests""" + message = f"API {method} {endpoint} - Status: {status_code} - Duration: {duration:.3f}s" + self.api_logger.info(message) + + def log_tool_usage(self, tool_name: str, success: bool, duration: float, details: Optional[Dict[str, Any]] = None): + """Log tool usage""" + status = "SUCCESS" if success else "FAILED" + message = f"Tool {tool_name} - {status} - Duration: {duration:.3f}s" + if details: + message += f" - {json.dumps(details)}" + self.tools_logger.info(message) + + def log_error(self, error: Exception, context: Optional[str] = None): + """Log errors""" + message = f"Error: {str(error)}" + if context: + message = f"{context} - {message}" + self.agent_logger.error(message, exc_info=True) + + def get_log_files(self): + """Get list of log files""" + return list(self.log_dir.glob("*.log")) + + def get_recent_logs(self, lines: int = 100): + """Get recent log entries""" + log_files = self.get_log_files() + if not log_files: + return "No log files found." + + # Get the most recent log file + latest_log = max(log_files, key=lambda f: f.stat().st_mtime) + + try: + with open(latest_log, 'r') as f: + all_lines = f.readlines() + return ''.join(all_lines[-lines:]) + except Exception as e: + return f"Error reading log file: {e}" + + +class SystemMonitor: + """Monitor system resources and performance""" + + def __init__(self): + self.start_time = time.time() + self.request_count = 0 + self.error_count = 0 + self.response_times = [] + + def record_request(self, duration: float, success: bool = True): + """Record an API request""" + self.request_count += 1 + self.response_times.append(duration) + + if not success: + self.error_count += 1 + + # Keep only last 1000 response times + if len(self.response_times) > 1000: + self.response_times = self.response_times[-1000:] + + def get_stats(self) -> Dict[str, Any]: + """Get system statistics""" + uptime = time.time() - self.start_time + + stats = { + "uptime_seconds": uptime, + "uptime_formatted": self._format_uptime(uptime), + "total_requests": self.request_count, + "total_errors": self.error_count, + "error_rate": self.error_count / self.request_count if self.request_count > 0 else 0, + "avg_response_time": sum(self.response_times) / len(self.response_times) if self.response_times else 0, + "min_response_time": min(self.response_times) if self.response_times else 0, + "max_response_time": max(self.response_times) if self.response_times else 0, + } + + return stats + + def _format_uptime(self, seconds: float) -> str: + """Format uptime in human-readable format""" + days = int(seconds // (24 * 3600)) + hours = int((seconds % (24 * 3600)) // 3600) + minutes = int((seconds % 3600) // 60) + seconds = int(seconds % 60) + + if days > 0: + return f"{days}d {hours}h {minutes}m {seconds}s" + elif hours > 0: + return f"{hours}h {minutes}m {seconds}s" + elif minutes > 0: + return f"{minutes}m {seconds}s" + else: + return f"{seconds}s" + + def reset_stats(self): + """Reset statistics""" + self.request_count = 0 + self.error_count = 0 + self.response_times = [] + + +# Global instances +logger = MaxClawLogger() +system_monitor = SystemMonitor() + + +def get_system_status() -> Dict[str, Any]: + """Get comprehensive system status""" + from config_manager import config_manager + + # Get configuration status + configs = config_manager.get_configurations() + active_config = config_manager.get_active_configuration() + + # Get system stats + stats = system_monitor.get_stats() + + # Get log files + log_files = logger.get_log_files() + + status = { + "timestamp": datetime.now().isoformat(), + "system": stats, + "configurations": { + "total": len(configs), + "active": active_config, + "providers": list(set(config.get("provider", "unknown") for config in configs.values())) + }, + "logging": { + "log_dir": str(logger.log_dir), + "log_files": len(log_files), + "recent_logs": logger.get_recent_logs(10) + }, + "performance": { + "cache_size": len(performance_monitor.cache) if 'performance_monitor' in globals() else 0 + } + } + + return status + + +def log_system_status(): + """Log current system status""" + status = get_system_status() + logger.log_agent_action("system_status_check", status) + + +# Import performance monitor if available +try: + from performance import performance_monitor +except ImportError: + performance_monitor = None + + +if __name__ == "__main__": + # Test logging + print("Testing MaxClaw logging system...") + + # Log some test messages + logger.log_agent_action("test_action", {"test": "data"}) + logger.log_config_change("add", "test-config", {"provider": "ollama"}) + logger.log_api_request("GET", "/health", 200, 0.05) + logger.log_tool_usage("ollama_list_models", True, 0.1) + + # Get system status + status = get_system_status() + print("\nSystem Status:") + print(json.dumps(status, indent=2)) + + # Get recent logs + print("\nRecent Logs:") + print(logger.get_recent_logs(5)) \ No newline at end of file diff --git a/performance.py b/performance.py new file mode 100644 index 0000000..d3b5d96 --- /dev/null +++ b/performance.py @@ -0,0 +1,299 @@ +""" +Performance optimizations for MaxClaw Agent +""" +import os +import sys +import time +import functools +from typing import Dict, Any, Optional, Callable +from pathlib import Path +import json + +# Add current directory to path +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + + +class PerformanceMonitor: + """Monitor and optimize performance""" + + def __init__(self): + self.metrics = {} + self.cache = {} + self.cache_ttl = 300 # 5 minutes default TTL + + def timing(self, func: Callable) -> Callable: + """Decorator to measure function execution time""" + @functools.wraps(func) + def wrapper(*args, **kwargs): + start_time = time.time() + result = func(*args, **kwargs) + end_time = time.time() + + # Store timing metric + func_name = func.__name__ + if func_name not in self.metrics: + self.metrics[func_name] = [] + + self.metrics[func_name].append(end_time - start_time) + + # Keep only last 100 measurements + if len(self.metrics[func_name]) > 100: + self.metrics[func_name] = self.metrics[func_name][-100:] + + return result + return wrapper + + def cache_result(self, ttl: Optional[int] = None): + """Decorator to cache function results""" + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + # Create cache key + cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}" + + # Check cache + if cache_key in self.cache: + cached_time, cached_result = self.cache[cache_key] + if time.time() - cached_time < (ttl or self.cache_ttl): + return cached_result + + # Execute function + result = func(*args, **kwargs) + + # Cache result + self.cache[cache_key] = (time.time(), result) + + return result + return wrapper + return decorator + + def get_metrics(self) -> Dict[str, Any]: + """Get performance metrics""" + metrics_summary = {} + + for func_name, timings in self.metrics.items(): + if timings: + metrics_summary[func_name] = { + "count": len(timings), + "avg_time": sum(timings) / len(timings), + "min_time": min(timings), + "max_time": max(timings), + "total_time": sum(timings) + } + + return metrics_summary + + def clear_cache(self): + """Clear the cache""" + self.cache.clear() + + def clear_metrics(self): + """Clear performance metrics""" + self.metrics.clear() + + +# Global performance monitor instance +performance_monitor = PerformanceMonitor() + + +def optimize_config_manager(): + """Optimize configuration manager with caching""" + from config_manager import config_manager + + # Cache configuration loading + original_load_config = config_manager._load_config + + @functools.wraps(original_load_config) + def optimized_load_config(): + # Check if we have a cached version + cache_key = "config_manager:load_config" + if cache_key in performance_monitor.cache: + cached_time, cached_result = performance_monitor.cache[cache_key] + if time.time() - cached_time < 60: # Cache for 1 minute + return cached_result + + # Load configuration + result = original_load_config() + + # Cache result + performance_monitor.cache[cache_key] = (time.time(), result) + + return result + + # Replace method + config_manager._load_config = optimized_load_config + + # Cache configuration retrieval + original_get_configurations = config_manager.get_configurations + + @functools.wraps(original_get_configurations) + def optimized_get_configurations(): + cache_key = "config_manager:get_configurations" + if cache_key in performance_monitor.cache: + cached_time, cached_result = performance_monitor.cache[cache_key] + if time.time() - cached_time < 30: # Cache for 30 seconds + return cached_result + + result = original_get_configurations() + performance_monitor.cache[cache_key] = (time.time(), result) + return result + + config_manager.get_configurations = optimized_get_configurations + + print("βœ… Configuration manager optimized with caching") + + +def optimize_provider_tools(): + """Optimize provider tools with connection pooling""" + from tools.provider_tools import get_provider_tools + + # Cache provider tools + original_get_provider_tools = get_provider_tools + + @functools.wraps(original_get_provider_tools) + def optimized_get_provider_tools(provider: str, **kwargs): + cache_key = f"provider_tools:{provider}:{str(kwargs)}" + if cache_key in performance_monitor.cache: + cached_time, cached_result = performance_monitor.cache[cache_key] + if time.time() - cached_time < 300: # Cache for 5 minutes + return cached_result + + result = original_get_provider_tools(provider, **kwargs) + performance_monitor.cache[cache_key] = (time.time(), result) + return result + + # Replace function + import tools.provider_tools + tools.provider_tools.get_provider_tools = optimized_get_provider_tools + + print("βœ… Provider tools optimized with caching") + + +def optimize_agent_creation(): + """Optimize agent creation with caching""" + from main import create_maxclaw_agent + + # Cache agent creation + original_create_maxclaw_agent = create_maxclaw_agent + + @functools.wraps(original_create_maxclaw_agent) + def optimized_create_maxclaw_agent(system_prompt=None): + cache_key = f"agent:{system_prompt}" + if cache_key in performance_monitor.cache: + cached_time, cached_result = performance_monitor.cache[cache_key] + if time.time() - cached_time < 600: # Cache for 10 minutes + return cached_result + + result = original_create_maxclaw_agent(system_prompt) + performance_monitor.cache[cache_key] = (time.time(), result) + return result + + # Replace function + import main + main.create_maxclaw_agent = optimized_create_maxclaw_agent + + print("βœ… Agent creation optimized with caching") + + +def optimize_api_server(): + """Optimize API server with response caching""" + try: + from api_server import app, list_configurations, get_active_configuration + + # Cache configuration endpoints + original_list_configurations = list_configurations + original_get_active_configuration = get_active_configuration + + @functools.wraps(original_list_configurations) + async def optimized_list_configurations(): + cache_key = "api:list_configurations" + if cache_key in performance_monitor.cache: + cached_time, cached_result = performance_monitor.cache[cache_key] + if time.time() - cached_time < 10: # Cache for 10 seconds + return cached_result + + result = await original_list_configurations() + performance_monitor.cache[cache_key] = (time.time(), result) + return result + + @functools.wraps(original_get_active_configuration) + async def optimized_get_active_configuration(): + cache_key = "api:get_active_configuration" + if cache_key in performance_monitor.cache: + cached_time, cached_result = performance_monitor.cache[cache_key] + if time.time() - cached_time < 10: # Cache for 10 seconds + return cached_result + + result = await original_get_active_configuration() + performance_monitor.cache[cache_key] = (time.time(), result) + return result + + # Replace functions + import api_server + api_server.list_configurations = optimized_list_configurations + api_server.get_active_configuration = optimized_get_active_configuration + + print("βœ… API server optimized with response caching") + except ImportError: + print("⚠️ API server not available for optimization") + + +def apply_all_optimizations(): + """Apply all performance optimizations""" + print("πŸš€ Applying performance optimizations...") + + try: + optimize_config_manager() + except Exception as e: + print(f"⚠️ Could not optimize config manager: {e}") + + try: + optimize_provider_tools() + except Exception as e: + print(f"⚠️ Could not optimize provider tools: {e}") + + try: + optimize_agent_creation() + except Exception as e: + print(f"⚠️ Could not optimize agent creation: {e}") + + try: + optimize_api_server() + except Exception as e: + print(f"⚠️ Could not optimize API server: {e}") + + print("βœ… All optimizations applied") + + +def get_performance_report() -> str: + """Get a performance report""" + metrics = performance_monitor.get_metrics() + + if not metrics: + return "No performance metrics collected yet." + + report = "Performance Report:\n" + report += "=" * 50 + "\n" + + for func_name, stats in metrics.items(): + report += f"\n{func_name}:\n" + report += f" Calls: {stats['count']}\n" + report += f" Avg Time: {stats['avg_time']:.4f}s\n" + report += f" Min Time: {stats['min_time']:.4f}s\n" + report += f" Max Time: {stats['max_time']:.4f}s\n" + report += f" Total Time: {stats['total_time']:.4f}s\n" + + # Cache statistics + cache_size = len(performance_monitor.cache) + report += f"\nCache Statistics:\n" + report += f" Cached Items: {cache_size}\n" + + return report + + +if __name__ == "__main__": + # Apply optimizations when run directly + apply_all_optimizations() + + # Print performance report + print("\n" + get_performance_report()) \ No newline at end of file