Spaces:
Sleeping
Sleeping
| """ | |
| Logging utility for agent actions - Policy Summarizer | |
| """ | |
| import logging | |
| import time | |
| from typing import Optional, List | |
| from functools import wraps | |
| # Configure logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s | %(levelname)s | %(name)s | %(message)s', | |
| datefmt='%Y-%m-%d %H:%M:%S' | |
| ) | |
| logger = logging.getLogger('PolicySummarizer') | |
| # Store logs for UI display | |
| _agent_logs: List[dict] = [] | |
| def get_logs() -> List[dict]: | |
| """Get all logged agent actions""" | |
| return _agent_logs.copy() | |
| def clear_logs(): | |
| """Clear all logs""" | |
| global _agent_logs | |
| _agent_logs = [] | |
| def log_agent_action( | |
| agent_name: str, | |
| action: str, | |
| input_summary: str, | |
| output_summary: str, | |
| duration_seconds: float, | |
| success: bool = True, | |
| error: Optional[str] = None | |
| ): | |
| """Log an agent action without sensitive data.""" | |
| log_entry = { | |
| "agent_name": agent_name, | |
| "action": action, | |
| "input_summary": input_summary[:200] + "..." if len(input_summary) > 200 else input_summary, | |
| "output_summary": output_summary[:200] + "..." if len(output_summary) > 200 else output_summary, | |
| "duration_seconds": round(duration_seconds, 2), | |
| "success": success, | |
| "error": error | |
| } | |
| _agent_logs.append(log_entry) | |
| status = "β" if success else "β" | |
| logger.info(f"{status} [{agent_name}] {action} ({duration_seconds:.2f}s)") | |
| if error: | |
| logger.error(f" Error: {error}") | |
| def format_logs_for_display() -> str: | |
| """Format logs for display in UI""" | |
| if not _agent_logs: | |
| return "No logs yet." | |
| lines = ["## Agent Activity Log\n"] | |
| for i, log in enumerate(_agent_logs, 1): | |
| status = "β " if log["success"] else "β" | |
| lines.append(f"### Step {i}: {log['agent_name']}") | |
| lines.append(f"- **Action:** {log['action']}") | |
| lines.append(f"- **Status:** {status}") | |
| lines.append(f"- **Duration:** {log['duration_seconds']}s") | |
| if log.get("error"): | |
| lines.append(f"- **Error:** {log['error']}") | |
| lines.append("") | |
| return "\n".join(lines) | |