299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""
|
|
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()) |