Personal Resume Agent

#!/usr/bin/env python3 """ Enhanced Interactive Task Sender for Resume Agent Features: Command history, conversation tracking, statistics, formatted output """ import requests import sys import json from datetime import datetime from collections import deque import os import time # ANSI color codes class Colors: HEADER = '\033[95m' BLUE = '\033[94m' CYAN = '\033[96m' GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' UNDERLINE = '\033[4m' # Statistics tracking class Stats: def __init__(self): self.total_tasks = 0 self.successful = 0 self.failed = 0 self.start_time = datetime.now() def record_success(self): self.total_tasks += 1 self.successful += 1 def record_failure(self): self.total_tasks += 1 self.failed += 1 def display(self): duration = datetime.now() - self.start_time print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f"{Colors.BOLD}📊 Session Statistics{Colors.ENDC}") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f" Total Tasks: {Colors.BOLD}{self.total_tasks}{Colors.ENDC}") print(f" Successful: {Colors.GREEN}{self.successful}{Colors.ENDC}") print(f" Failed: {Colors.RED}{self.failed}{Colors.ENDC}") print(f" Success Rate: {Colors.BOLD}{(self.successful/self.total_tasks*100) if self.total_tasks > 0 else 0:.1f}%{Colors.ENDC}") print(f" Session Duration: {duration.seconds}s") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n") class ConversationHistory: def __init__(self, max_size=10): self.history = deque(maxlen=max_size) def add(self, task, response, conv_id, msg_id): self.history.append({ 'task': task, 'response': response, 'conv_id': conv_id, 'msg_id': msg_id, 'timestamp': datetime.now() }) def display(self): if not self.history: print(f"{Colors.YELLOW}No conversation history yet{Colors.ENDC}\n") return print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f"{Colors.BOLD}💬 Recent Conversations ({len(self.history)}){Colors.ENDC}") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}") for i, item in enumerate(self.history, 1): timestamp = item['timestamp'].strftime('%H:%M:%S') print(f"\n{Colors.BOLD}[{i}] {timestamp}{Colors.ENDC}") print(f" {Colors.BLUE}Q:{Colors.ENDC} {item['task'][:60]}...") print(f" {Colors.GREEN}A:{Colors.ENDC} {item['response'][:60]}...") print(f" {Colors.CYAN}ID:{Colors.ENDC} {item['conv_id']}") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n") def clear_screen(): """Clear terminal screen""" os.system('clear' if os.name != 'nt' else 'cls') def format_response(text, max_width=70): """Format response text with word wrapping""" words = text.split() lines = [] current_line = [] current_length = 0 for word in words: if current_length + len(word) + 1 <= max_width: current_line.append(word) current_length += len(word) + 1 else: lines.append(' '.join(current_line)) current_line = [word] current_length = len(word) if current_line: lines.append(' '.join(current_line)) return '\n '.join(lines) def send_task(task_text, sender="user", stats=None, history=None): """Send task to resume agent with enhanced display""" url = "http://52.73.59.119:6050/a2a" payload = { "content": { "text": task_text, "type": "text" }, "role": "user" if sender == "user" else "agent", "conversation_id": f"task-{datetime.now().strftime('%Y%m%d-%H%M%S')}" } if sender != "user": payload["sender_agent_id"] = sender try: # Sending indicator print(f"\n{Colors.CYAN}{'─' * 70}{Colors.ENDC}") print(f"{Colors.BOLD}📤 Sending to resume-agent...{Colors.ENDC}") print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}") print(f"{Colors.BLUE}Q:{Colors.ENDC} {task_text}") # Add loading animation for i in range(3): print(f"\r {'.' * (i+1)}", end='', flush=True) time.sleep(0.3) print("\r ", end='') response = requests.post(url, json=payload, timeout=15) if response.status_code == 200: result = response.json() answer = result['parts'][0]['text'] conv_id = result['metadata']['conversation_id'] msg_id = result['metadata']['message_id'] # Success display print(f"\r{Colors.GREEN}✅ Response received!{Colors.ENDC}") print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}") print(f"{Colors.GREEN}{Colors.BOLD}A:{Colors.ENDC} {format_response(answer)}") print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}") print(f"{Colors.CYAN}📎 Conv: {conv_id[:20]}...{Colors.ENDC}") print(f"{Colors.CYAN}📎 Msg: {msg_id[:20]}...{Colors.ENDC}") # Update stats and history if stats: stats.record_success() if history: history.add(task_text, answer, conv_id, msg_id) return True else: print(f"\r{Colors.RED}❌ Failed: {response.status_code}{Colors.ENDC}") print(f" {response.text[:200]}") if stats: stats.record_failure() return False except requests.exceptions.Timeout: print(f"\r{Colors.RED}❌ Timeout: Agent took too long to respond{Colors.ENDC}") if stats: stats.record_failure() return False except Exception as e: print(f"\r{Colors.RED}❌ Error: {e}{Colors.ENDC}") if stats: stats.record_failure() return False def show_help(): """Display help information""" print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f"{Colors.BOLD}📚 Available Commands{Colors.ENDC}") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f" {Colors.BOLD}/help{Colors.ENDC} - Show this help message") print(f" {Colors.BOLD}/history{Colors.ENDC} - Show conversation history") print(f" {Colors.BOLD}/stats{Colors.ENDC} - Show session statistics") print(f" {Colors.BOLD}/clear{Colors.ENDC} - Clear screen") print(f" {Colors.BOLD}/info{Colors.ENDC} - Get agent information") print(f" {Colors.BOLD}/status{Colors.ENDC} - Check agent status") print(f" {Colors.BOLD}quit{Colors.ENDC} - Exit the dashboard") print(f"\n{Colors.YELLOW}💡 Example Questions:{Colors.ENDC}") print(f" • What is my work experience?") print(f" • What are my technical skills?") print(f" • Tell me about my education") print(f" • What are my key achievements?") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n") def main(): stats = Stats() history = ConversationHistory() # Welcome screen clear_screen() print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f"{Colors.BOLD}{Colors.HEADER}🤖 Resume Agent - Enhanced Interactive Dashboard{Colors.ENDC}") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f"{Colors.GREEN}✓{Colors.ENDC} Connected to: {Colors.BOLD}http://52.73.59.119:6050/a2a{Colors.ENDC}") print(f"{Colors.GREEN}✓{Colors.ENDC} Type {Colors.BOLD}/help{Colors.ENDC} for available commands") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n") if len(sys.argv) > 1: # Task from command line task = " ".join(sys.argv[1:]) send_task(task, stats=stats, history=history) stats.display() else: # Interactive mode while True: try: task = input(f"{Colors.BOLD}{Colors.BLUE}📝 >{Colors.ENDC} ").strip() if not task: continue # Handle special commands if task.lower() in ['quit', 'exit', 'q']: print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}") stats.display() print(f"{Colors.GREEN}👋 Thank you for using Resume Agent Dashboard!{Colors.ENDC}\n") break elif task.lower() == '/help': show_help() continue elif task.lower() == '/history': history.display() continue elif task.lower() == '/stats': stats.display() continue elif task.lower() == '/clear': clear_screen() print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}") print(f"{Colors.BOLD}{Colors.HEADER}🤖 Resume Agent Dashboard{Colors.ENDC}") print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n") continue # Send regular task send_task(task, stats=stats, history=history) except KeyboardInterrupt: print(f"\n\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}") stats.display() print(f"{Colors.GREEN}👋 Goodbye!{Colors.ENDC}\n") break except EOFError: break if __name__ == "__main__": main()