""" 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())