Spaces:
Running
Running
| #!/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" | |