#!/usr/bin/env python3 """ Utility functions for agent operations """ import time from typing import Tuple import logging # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class PerformanceMonitor: """ Monitor agent performance and timing """ def __init__(self): self.metrics = {} def start_timer(self, operation: str) -> None: """ Start timing an operation """ self.metrics[f"{operation}_start"] = time.time() def end_timer(self, operation: str) -> float: """ End timing an operation and return duration """ start_time = self.metrics.get(f"{operation}_start") if start_time: duration = time.time() - start_time self.metrics[f"{operation}_duration"] = duration return duration return 0.0 def get_metrics(self) -> dict: """ Get all collected metrics """ return self.metrics.copy() def reset(self) -> None: """ Reset all metrics """ self.metrics.clear() def validate_query(query: str) -> Tuple[bool, str]: """ Validate user query """ if not query or not query.strip(): return False, "Query cannot be empty." if len(query) > 2500: return False, "Query is too long. Please keep it under 1000 characters." return True, None def format_error_message(error: str) -> str: """ Format error messages for user display """ error_map = { "Server unreachable": "❌ The legal database is currently unavailable. Please try again in a moment.", "timeout": "❌ The request timed out. Please try again.", "invalid json": "❌ There was an issue processing the response. Please try again.", "health check failed": "❌ The system is initializing. Please wait a moment and try again." } for key, message in error_map.items(): if key.lower() in error.lower(): return message return f"❌ An error occurred: {error}" def create_safe_filename(query: str, timestamp: str) -> str: """ Create a safe filename for logging purposes """ # Remove problematic characters safe_query = "".join(c for c in query if c.isalnum() or c in (' ', '-', '_')).strip() safe_query = safe_query[:50] # Limit length return f"{timestamp}_{safe_query}.log"