Phase 14: Enhanced performance optimizations with connection pooling and lazy loading
This commit is contained in:
1 parent
979d19bdc1
commit
7103b218b9
2 files changed
+251
-1
No files matched your search
+215
-1
@@ -1,18 +1,209 @@
|
||||
"""
|
||||
Performance optimizations for MaxClaw Agent
|
||||
Enhanced with connection pooling, lazy loading, and advanced caching
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import functools
|
||||
from typing import Dict, Any, Optional, Callable
|
||||
import threading
|
||||
from typing import Dict, Any, Optional, Callable, List
|
||||
from pathlib import Path
|
||||
import json
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
# Add current directory to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
class ConnectionPool:
|
||||
"""Connection pool for managing API connections"""
|
||||
|
||||
def __init__(self, max_connections: int = 10):
|
||||
"""
|
||||
Initialize connection pool
|
||||
|
||||
Args:
|
||||
max_connections: Maximum number of connections in the pool
|
||||
"""
|
||||
self.max_connections = max_connections
|
||||
self.connections = {}
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def get_connection(self, provider: str, **kwargs):
|
||||
"""
|
||||
Get a connection from the pool
|
||||
|
||||
Args:
|
||||
provider: Provider name
|
||||
**kwargs: Connection parameters
|
||||
|
||||
Returns:
|
||||
Connection object
|
||||
"""
|
||||
with self.lock:
|
||||
key = f"{provider}:{str(sorted(kwargs.items()))}"
|
||||
|
||||
if key not in self.connections:
|
||||
# Create new connection
|
||||
if provider == "ollama":
|
||||
import requests
|
||||
base_url = kwargs.get("base_url", "http://localhost:11434")
|
||||
session = requests.Session()
|
||||
session.headers.update({"Content-Type": "application/json"})
|
||||
self.connections[key] = {
|
||||
"session": session,
|
||||
"base_url": base_url,
|
||||
"last_used": time.time()
|
||||
}
|
||||
elif provider in ["openai", "openrouter", "anthropic", "google"]:
|
||||
import requests
|
||||
session = requests.Session()
|
||||
api_key = kwargs.get("api_key", "")
|
||||
if api_key:
|
||||
session.headers.update({"Authorization": f"Bearer {api_key}"})
|
||||
self.connections[key] = {
|
||||
"session": session,
|
||||
"last_used": time.time()
|
||||
}
|
||||
|
||||
# Update last used time
|
||||
if key in self.connections:
|
||||
self.connections[key]["last_used"] = time.time()
|
||||
|
||||
return self.connections.get(key)
|
||||
|
||||
def cleanup_idle_connections(self, max_idle_time: int = 300):
|
||||
"""
|
||||
Clean up idle connections
|
||||
|
||||
Args:
|
||||
max_idle_time: Maximum idle time in seconds
|
||||
"""
|
||||
with self.lock:
|
||||
current_time = time.time()
|
||||
keys_to_remove = []
|
||||
|
||||
for key, conn in self.connections.items():
|
||||
if current_time - conn["last_used"] > max_idle_time:
|
||||
keys_to_remove.append(key)
|
||||
|
||||
for key in keys_to_remove:
|
||||
del self.connections[key]
|
||||
|
||||
def get_stats(self) -> Dict[str, Any]:
|
||||
"""Get connection pool statistics"""
|
||||
with self.lock:
|
||||
return {
|
||||
"total_connections": len(self.connections),
|
||||
"max_connections": self.max_connections,
|
||||
"connections": list(self.connections.keys())
|
||||
}
|
||||
|
||||
|
||||
class LazyLoader:
|
||||
"""Lazy loading for heavy resources"""
|
||||
|
||||
def __init__(self):
|
||||
self.loaded_modules = {}
|
||||
self.loaded_tools = {}
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def load_module(self, module_name: str, import_path: str):
|
||||
"""
|
||||
Lazily load a module
|
||||
|
||||
Args:
|
||||
module_name: Name to store the module under
|
||||
import_path: Import path for the module
|
||||
|
||||
Returns:
|
||||
Loaded module
|
||||
"""
|
||||
with self.lock:
|
||||
if module_name not in self.loaded_modules:
|
||||
try:
|
||||
module = __import__(import_path, fromlist=[''])
|
||||
self.loaded_modules[module_name] = module
|
||||
except ImportError as e:
|
||||
print(f"Warning: Could not load module {module_name}: {e}")
|
||||
self.loaded_modules[module_name] = None
|
||||
|
||||
return self.loaded_modules.get(module_name)
|
||||
|
||||
def load_tool(self, tool_name: str, loader_func: Callable):
|
||||
"""
|
||||
Lazily load a tool
|
||||
|
||||
Args:
|
||||
tool_name: Name to store the tool under
|
||||
loader_func: Function to load the tool
|
||||
|
||||
Returns:
|
||||
Loaded tool
|
||||
"""
|
||||
with self.lock:
|
||||
if tool_name not in self.loaded_tools:
|
||||
try:
|
||||
tool = loader_func()
|
||||
self.loaded_tools[tool_name] = tool
|
||||
except Exception as e:
|
||||
print(f"Warning: Could not load tool {tool_name}: {e}")
|
||||
self.loaded_tools[tool_name] = None
|
||||
|
||||
return self.loaded_tools.get(tool_name)
|
||||
|
||||
def clear(self):
|
||||
"""Clear all loaded resources"""
|
||||
with self.lock:
|
||||
self.loaded_modules.clear()
|
||||
self.loaded_tools.clear()
|
||||
|
||||
|
||||
class BatchProcessor:
|
||||
"""Batch processing for multiple operations"""
|
||||
|
||||
def __init__(self, max_workers: int = 5):
|
||||
"""
|
||||
Initialize batch processor
|
||||
|
||||
Args:
|
||||
max_workers: Maximum number of worker threads
|
||||
"""
|
||||
self.max_workers = max_workers
|
||||
self.executor = ThreadPoolExecutor(max_workers=max_workers)
|
||||
|
||||
def process_batch(self, tasks: List[Callable], timeout: int = 30) -> List[Any]:
|
||||
"""
|
||||
Process a batch of tasks in parallel
|
||||
|
||||
Args:
|
||||
tasks: List of callable tasks
|
||||
timeout: Timeout in seconds
|
||||
|
||||
Returns:
|
||||
List of results
|
||||
"""
|
||||
futures = [self.executor.submit(task) for task in tasks]
|
||||
results = []
|
||||
|
||||
try:
|
||||
for future in as_completed(futures, timeout=timeout):
|
||||
try:
|
||||
result = future.result()
|
||||
results.append(result)
|
||||
except Exception as e:
|
||||
results.append({"error": str(e)})
|
||||
except Exception as e:
|
||||
print(f"Batch processing error: {e}")
|
||||
|
||||
return results
|
||||
|
||||
def shutdown(self):
|
||||
"""Shutdown the executor"""
|
||||
self.executor.shutdown(wait=True)
|
||||
|
||||
|
||||
class PerformanceMonitor:
|
||||
"""Monitor and optimize performance"""
|
||||
|
||||
@@ -20,6 +211,9 @@ class PerformanceMonitor:
|
||||
self.metrics = {}
|
||||
self.cache = {}
|
||||
self.cache_ttl = 300 # 5 minutes default TTL
|
||||
self.connection_pool = ConnectionPool()
|
||||
self.lazy_loader = LazyLoader()
|
||||
self.batch_processor = BatchProcessor()
|
||||
|
||||
def timing(self, func: Callable) -> Callable:
|
||||
"""Decorator to measure function execution time"""
|
||||
@@ -90,6 +284,20 @@ class PerformanceMonitor:
|
||||
def clear_metrics(self):
|
||||
"""Clear performance metrics"""
|
||||
self.metrics.clear()
|
||||
|
||||
def get_connection_pool_stats(self) -> Dict[str, Any]:
|
||||
"""Get connection pool statistics"""
|
||||
return self.connection_pool.get_stats()
|
||||
|
||||
def cleanup_resources(self):
|
||||
"""Clean up resources"""
|
||||
self.connection_pool.cleanup_idle_connections()
|
||||
self.clear_cache()
|
||||
|
||||
def shutdown(self):
|
||||
"""Shutdown all resources"""
|
||||
self.batch_processor.shutdown()
|
||||
self.cleanup_resources()
|
||||
|
||||
|
||||
# Global performance monitor instance
|
||||
@@ -288,6 +496,12 @@ def get_performance_report() -> str:
|
||||
report += f"\nCache Statistics:\n"
|
||||
report += f" Cached Items: {cache_size}\n"
|
||||
|
||||
# Connection pool statistics
|
||||
pool_stats = performance_monitor.get_connection_pool_stats()
|
||||
report += f"\nConnection Pool:\n"
|
||||
report += f" Total Connections: {pool_stats['total_connections']}\n"
|
||||
report += f" Max Connections: {pool_stats['max_connections']}\n"
|
||||
|
||||
return report
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user