Spaces:
Sleeping
Sleeping
| """ | |
| Custom Log Formatters for GEPA Optimizer. | |
| Provides formatters for: | |
| - Console output with colors and emoji | |
| - JSON structured logging for production | |
| - Plain text for file logging | |
| """ | |
| import json | |
| import logging | |
| from datetime import datetime | |
| from typing import Any, Dict, Optional | |
| # ANSI color codes for terminal output | |
| class Colors: | |
| """ANSI color codes for terminal coloring.""" | |
| RESET = "\033[0m" | |
| BOLD = "\033[1m" | |
| DIM = "\033[2m" | |
| # Log level colors | |
| DEBUG = "\033[36m" # Cyan | |
| INFO = "\033[32m" # Green | |
| WARNING = "\033[33m" # Yellow | |
| ERROR = "\033[31m" # Red | |
| CRITICAL = "\033[35m" # Magenta | |
| # Semantic colors | |
| TIMESTAMP = "\033[90m" # Gray | |
| MODULE = "\033[34m" # Blue | |
| MESSAGE = "\033[0m" # Default | |
| # Emoji prefixes for visual log scanning | |
| LEVEL_EMOJI = { | |
| logging.DEBUG: "🔍", | |
| logging.INFO: "ℹ️ ", | |
| logging.WARNING: "⚠️ ", | |
| logging.ERROR: "❌", | |
| logging.CRITICAL: "🚨", | |
| } | |
| # Level colors mapping | |
| LEVEL_COLORS = { | |
| logging.DEBUG: Colors.DEBUG, | |
| logging.INFO: Colors.INFO, | |
| logging.WARNING: Colors.WARNING, | |
| logging.ERROR: Colors.ERROR, | |
| logging.CRITICAL: Colors.CRITICAL, | |
| } | |
| class GepaFormatter(logging.Formatter): | |
| """ | |
| Custom formatter for GEPA Optimizer logs. | |
| Features: | |
| - Optional color output for console | |
| - Optional emoji prefixes for visual scanning | |
| - Structured extra fields support | |
| - Clean, readable format | |
| Example output: | |
| 2024-01-15 10:30:45 | INFO | ℹ️ gepa_optimizer.core.optimizer | Starting optimization iteration=5 | |
| """ | |
| def __init__( | |
| self, | |
| fmt: Optional[str] = None, | |
| datefmt: Optional[str] = None, | |
| use_colors: bool = True, | |
| include_emoji: bool = True, | |
| ): | |
| """ | |
| Initialize the formatter. | |
| Args: | |
| fmt: Format string (uses default if not provided) | |
| datefmt: Date format string | |
| use_colors: Whether to use ANSI colors | |
| include_emoji: Whether to include emoji prefixes | |
| """ | |
| super().__init__(fmt=fmt, datefmt=datefmt) | |
| self.use_colors = use_colors | |
| self.include_emoji = include_emoji | |
| def format(self, record: logging.LogRecord) -> str: | |
| """Format a log record with colors and emoji.""" | |
| # Store original values | |
| original_msg = record.msg | |
| original_levelname = record.levelname | |
| try: | |
| # Add emoji prefix if enabled | |
| if self.include_emoji: | |
| emoji = LEVEL_EMOJI.get(record.levelno, "") | |
| record.levelname = f"{emoji} {record.levelname}" | |
| # Add colors if enabled | |
| if self.use_colors: | |
| color = LEVEL_COLORS.get(record.levelno, Colors.RESET) | |
| record.levelname = f"{color}{record.levelname}{Colors.RESET}" | |
| record.name = f"{Colors.MODULE}{record.name}{Colors.RESET}" | |
| # Format extra fields if present | |
| extra_str = self._format_extra(record) | |
| if extra_str: | |
| record.msg = f"{record.msg} | {extra_str}" | |
| # Call parent formatter | |
| formatted = super().format(record) | |
| return formatted | |
| finally: | |
| # Restore original values | |
| record.msg = original_msg | |
| record.levelname = original_levelname | |
| def _format_extra(self, record: logging.LogRecord) -> str: | |
| """ | |
| Format extra fields from the log record. | |
| Extra fields are passed via the 'extra' parameter to logging calls: | |
| logger.info("Message", extra={"key": "value"}) | |
| """ | |
| # Standard LogRecord attributes to exclude | |
| standard_attrs = { | |
| 'name', 'msg', 'args', 'created', 'filename', 'funcName', | |
| 'levelname', 'levelno', 'lineno', 'module', 'msecs', | |
| 'pathname', 'process', 'processName', 'relativeCreated', | |
| 'stack_info', 'exc_info', 'exc_text', 'thread', 'threadName', | |
| 'taskName', 'message' | |
| } | |
| # Collect extra fields | |
| extra_fields = { | |
| k: v for k, v in record.__dict__.items() | |
| if k not in standard_attrs and not k.startswith('_') | |
| } | |
| if not extra_fields: | |
| return "" | |
| # Format as key=value pairs | |
| parts = [] | |
| for key, value in extra_fields.items(): | |
| if isinstance(value, str): | |
| parts.append(f"{key}={value}") | |
| elif isinstance(value, (int, float)): | |
| parts.append(f"{key}={value}") | |
| elif isinstance(value, bool): | |
| parts.append(f"{key}={str(value).lower()}") | |
| else: | |
| parts.append(f"{key}={repr(value)}") | |
| return " ".join(parts) | |
| class JsonFormatter(logging.Formatter): | |
| """ | |
| JSON formatter for structured logging. | |
| Outputs each log record as a single JSON line, suitable for: | |
| - Log aggregation systems (ELK, Splunk) | |
| - Cloud logging (CloudWatch, Stackdriver) | |
| - Log parsing and analysis | |
| Example output: | |
| {"timestamp": "2024-01-15T10:30:45.123Z", "level": "INFO", "logger": "gepa_optimizer.core", "message": "Starting optimization", "iteration": 5} | |
| """ | |
| def __init__( | |
| self, | |
| include_timestamp: bool = True, | |
| include_location: bool = False, | |
| ): | |
| """ | |
| Initialize JSON formatter. | |
| Args: | |
| include_timestamp: Include ISO timestamp | |
| include_location: Include file/line information | |
| """ | |
| super().__init__() | |
| self.include_timestamp = include_timestamp | |
| self.include_location = include_location | |
| def format(self, record: logging.LogRecord) -> str: | |
| """Format record as JSON string.""" | |
| log_dict: Dict[str, Any] = {} | |
| # Timestamp | |
| if self.include_timestamp: | |
| log_dict["timestamp"] = datetime.utcfromtimestamp( | |
| record.created | |
| ).isoformat() + "Z" | |
| # Core fields | |
| log_dict["level"] = record.levelname | |
| log_dict["logger"] = record.name | |
| log_dict["message"] = record.getMessage() | |
| # Location info | |
| if self.include_location: | |
| log_dict["file"] = record.filename | |
| log_dict["line"] = record.lineno | |
| log_dict["function"] = record.funcName | |
| # Exception info | |
| if record.exc_info: | |
| log_dict["exception"] = self.formatException(record.exc_info) | |
| # Extra fields | |
| standard_attrs = { | |
| 'name', 'msg', 'args', 'created', 'filename', 'funcName', | |
| 'levelname', 'levelno', 'lineno', 'module', 'msecs', | |
| 'pathname', 'process', 'processName', 'relativeCreated', | |
| 'stack_info', 'exc_info', 'exc_text', 'thread', 'threadName', | |
| 'taskName', 'message' | |
| } | |
| for key, value in record.__dict__.items(): | |
| if key not in standard_attrs and not key.startswith('_'): | |
| try: | |
| # Ensure value is JSON serializable | |
| json.dumps(value) | |
| log_dict[key] = value | |
| except (TypeError, ValueError): | |
| log_dict[key] = str(value) | |
| return json.dumps(log_dict, default=str) | |
| class CompactFormatter(logging.Formatter): | |
| """ | |
| Compact formatter for minimal log output. | |
| Useful for: | |
| - CI/CD pipelines | |
| - Reduced log verbosity | |
| - Quick debugging | |
| Example output: | |
| 10:30:45 INFO optimizer: Starting optimization | |
| """ | |
| def format(self, record: logging.LogRecord) -> str: | |
| """Format record in compact form.""" | |
| # Short timestamp (time only) | |
| time_str = datetime.fromtimestamp(record.created).strftime("%H:%M:%S") | |
| # Short module name (last part only) | |
| short_name = record.name.split(".")[-1] | |
| return f"{time_str} {record.levelname:5s} {short_name}: {record.getMessage()}" | |