Phase 7: Performance optimizations, monitoring, and command system with / prefix
This commit is contained in:
1 parent
6eace3abc0
commit
c44a4ae92a
4 files changed
+849
-13
No files matched your search
@@ -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 <name> - 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:
|
||||
|
||||
@@ -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 <name> - 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 <name>")
|
||||
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 <name>")
|
||||
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
|
||||
|
||||
+259
@@ -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))
|
||||
+299
@@ -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())
|
||||
Reference in new issue
Block a user