""" Structured Logger for Voice Agent Provides JSON-formatted logging for better observability """ import logging import json from datetime import datetime import sys class JsonFormatter(logging.Formatter): """Format logs as JSON""" def format(self, record): log_data = { "timestamp": datetime.now().isoformat(), "level": record.levelname, "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno } # Add extra fields if present if hasattr(record, 'user_id'): log_data['user_id'] = record.user_id if hasattr(record, 'session_id'): log_data['session_id'] = record.session_id if hasattr(record, 'duration'): log_data['duration'] = record.duration if hasattr(record, 'error'): log_data['error'] = record.error return json.dumps(log_data) class StructuredLogger: """Structured logger with JSON output""" def __init__(self, name="voice-agent"): self.logger = logging.getLogger(name) self.logger.setLevel(logging.INFO) # Remove existing handlers self.logger.handlers = [] # Console handler with JSON format handler = logging.StreamHandler(sys.stdout) handler.setFormatter(JsonFormatter()) self.logger.addHandler(handler) def info(self, message, **kwargs): """Log info message with optional context""" extra = {k: v for k, v in kwargs.items()} self.logger.info(message, extra=extra) def error(self, message, **kwargs): """Log error message with optional context""" extra = {k: v for k, v in kwargs.items()} self.logger.error(message, extra=extra) def warning(self, message, **kwargs): """Log warning message with optional context""" extra = {k: v for k, v in kwargs.items()} self.logger.warning(message, extra=extra) def debug(self, message, **kwargs): """Log debug message with optional context""" extra = {k: v for k, v in kwargs.items()} self.logger.debug(message, extra=extra) # Global logger instance logger = StructuredLogger()