""" Monitoring and logging for LangCode 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 LangCodeLogger: """Custom logger for LangCode Agent""" def __init__(self, log_dir: Optional[str] = None): """ Initialize logger Args: log_dir: Directory for log files. If None, uses ~/.langcode/logs """ if log_dir is None: self.log_dir = Path.home() / ".langcode" / "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"langcode_{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("langcode.agent") self.config_logger = logging.getLogger("langcode.config") self.api_logger = logging.getLogger("langcode.api") self.tools_logger = logging.getLogger("langcode.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 = LangCodeLogger() 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 LangCode 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))