Personal Resume Agent

#!/usr/bin/env python3 """ NEST-Compatible Resume Agent Integrates PersonalResumeAgent with NANDA framework for A2A communication Optimized for CPU-only, t3.small deployment """ import os import sys import asyncio import logging # Add agent source to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) # Import NEST framework sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'nest')) from nanda_core.core.adapter import NANDA # Import resume agent components from personal_resume_agent import PersonalResumeAgent # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class ResumeAgentLogic: """ Wrapper for PersonalResumeAgent to work with NANDA Handles async-to-sync conversion and message routing """ def __init__(self, data_directory: str): self.agent = PersonalResumeAgent(data_directory=data_directory) self.initialized = False self._loop = None async def initialize(self): """Initialize the resume agent and RAG system""" try: self.initialized = await self.agent.initialize() if self.initialized: logger.info("✅ Resume agent initialized successfully") else: logger.warning("⚠️ Resume agent initialized but no resume files found") return self.initialized except Exception as e: logger.error(f"❌ Failed to initialize resume agent: {e}") return False def _run_async(self, coro): """Run async coroutine in sync context""" try: # Use existing event loop or create new one try: loop = asyncio.get_running_loop() except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(coro) except Exception as e: logger.error(f"Error running async operation: {e}") raise def handle_message(self, message: str, conversation_id: str) -> str: """ Handle incoming messages from NANDA framework Args: message: The incoming message text conversation_id: Unique conversation identifier Returns: Response string """ try: # Check if agent is initialized if not self.initialized: return "❌ Resume agent not initialized. Please add resume files to the data directory." # Handle special commands if message.lower().strip() in ['/info', 'info', '/help', 'help']: return self._get_agent_info() elif message.lower().startswith('/skills') or 'skill match' in message.lower(): return self._handle_skill_match(message) elif message.lower().strip() in ['/status', 'status']: return self._get_status() # Handle regular queries else: return self._process_query(message) except Exception as e: logger.error(f"Error processing message: {e}") return f"❌ Error processing query: {str(e)}" def _get_agent_info(self) -> str: """Get agent information and capabilities""" try: info = self.agent.get_agent_info() response = f"""📋 Personal Resume Agent Status: {'✅ Ready' if info['initialized'] else '⚠️ Not Initialized'} Embedding Model: all-MiniLM-L6-v2 (CPU-optimized) Capabilities: """ for cap in info['capabilities']: response += f" • {cap}\n" if info.get('resume_summary'): response += f"\n{info['resume_summary']}" return response.strip() except Exception as e: logger.error(f"Error getting agent info: {e}") return f"❌ Error retrieving agent info: {str(e)}" def _handle_skill_match(self, message: str) -> str: """Handle skill matching requests""" try: # Extract job requirements from message if ':' in message: job_req = message.split(':', 1)[1].strip() else: # Remove command keywords and use rest as job requirements job_req = message.replace('skill match', '').replace('/skills', '').strip() if not job_req: return "Please provide job requirements. Example: '/skills Python, AWS, Docker'" # Get skill match result = self._run_async(self.agent.get_skill_match(job_req)) response = f"""🎯 Skill Match Analysis Job Requirements: {job_req} Match Score: {result.get('match_percentage', 0):.1f}% Confidence: {result.get('confidence', 0):.2f} Matching Skills: """ matching = result.get('matching_skills', []) if matching: for skill in matching[:10]: # Show top 10 response += f" ✓ {skill}\n" else: response += " No direct matches found\n" missing = result.get('missing_skills', []) if missing: response += f"\nGap Areas:\n" for skill in missing[:5]: # Show top 5 gaps response += f" • {skill}\n" return response.strip() except Exception as e: logger.error(f"Error matching skills: {e}") return f"❌ Error analyzing skill match: {str(e)}" def _get_status(self) -> str: """Get agent status""" return f"""🔍 Agent Status Initialized: {'✅ Yes' if self.initialized else '❌ No'} Type: Resume RAG Agent (NEST-enabled) Backend: ChromaDB 1.0.15 (CPU-optimized) Communication: A2A Protocol """ def _process_query(self, query: str) -> str: """Process regular resume queries""" try: result = self._run_async(self.agent.process_query(query)) response = result.get('response', 'No response available') confidence = result.get('confidence', 0) # Add confidence indicator if confidence < 0.3: response += "\n\n⚠️ Low confidence answer - information may be limited." return response except Exception as e: logger.error(f"Error processing query: {e}") return f"❌ Error: {str(e)}" def register_agent_with_metadata(agent_id, public_url, registry_url): """Register agent with full metadata in NANDA registry""" import requests # Full agent metadata for visibility agent_data = { "agent_id": agent_id, "agent_url": public_url, "endpoint": public_url, "a2a_endpoint": f"{public_url}/a2a", "name": f"Personal Resume Agent - {agent_id}", "description": "AI agent providing professional background, skills, and experience information using RAG (Retrieval-Augmented Generation)", "expertise": ["resume analysis", "career information", "skill matching", "professional background", "work experience"], "specialization": "personal-resume-assistant", "status": "running", "protocol_version": "A2A-1.0", "deployment_method": "nanda_adapter", "capabilities": ["query_resume", "skill_matching", "experience_summary", "education_info"] } try: response = requests.post(f"{registry_url}/register", json=agent_data, timeout=10) if response.status_code == 200: print(f"✅ Agent '{agent_id}' registered with full metadata") print(f" Name: {agent_data['name']}") print(f" Expertise: {', '.join(agent_data['expertise'][:3])}...") return True else: print(f"⚠️ Registration response: {response.status_code}") print(f" {response.text}") return False except Exception as e: print(f"⚠️ Registration error: {e}") return False def main(): """Main function to start the NEST-compatible resume agent""" # Configuration from environment AGENT_ID = os.getenv("AGENT_ID", "vikram-resume-agent") PORT = int(os.getenv("PORT", "6050")) REGISTRY_URL = os.getenv("REGISTRY_URL", None) PUBLIC_URL = os.getenv("PUBLIC_URL", None) DATA_DIR = os.getenv("DATA_DIR", "./data") print("=" * 60) print("🤖 NEST Resume Agent - CPU Optimized") print("=" * 60) print(f"Agent ID: {AGENT_ID}") print(f"Port: {PORT}") print(f"Data Directory: {DATA_DIR}") print(f"Registry: {REGISTRY_URL or 'Not configured (local mode)'}") print(f"Public URL: {PUBLIC_URL or 'Not configured'}") print("=" * 60) # Create resume agent logic resume_logic = ResumeAgentLogic(data_directory=DATA_DIR) # Initialize resume agent print("\n📦 Initializing resume agent...") loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) initialized = loop.run_until_complete(resume_logic.initialize()) if not initialized: print("\n⚠️ WARNING: Resume agent initialization incomplete") print(" Make sure resume files (PDF, DOCX, TXT, MD) are in the data directory") print(f" Data directory: {os.path.abspath(DATA_DIR)}") print("\n The agent will start but won't be able to answer questions until") print(" you add resume files and restart.") # Create NANDA agent print("\n🔧 Creating NANDA adapter...") nanda = NANDA( agent_id=AGENT_ID, agent_logic=resume_logic.handle_message, port=PORT, registry_url=REGISTRY_URL, public_url=PUBLIC_URL, enable_telemetry=False ) print(f""" {'=' * 60} 📝 Resume Agent Commands: {'=' * 60} Regular Queries: • "What is my work experience?" • "What are my technical skills?" • "Tell me about my education" Special Commands: • /info or info - Show agent capabilities • /status - Show agent status • /skills Python, AWS, Docker - Match skills to job requirements • skill match: Python, AWS - Alternative skill match syntax Agent-to-Agent: • @other-agent check my resume for Python skills • @recruiter-bot analyze my background {'=' * 60} 🚀 Starting on http://0.0.0.0:{PORT}/a2a 🛑 Press Ctrl+C to stop {'=' * 60} """) # Register with full metadata if registry configured if REGISTRY_URL and PUBLIC_URL: print("\n📝 Registering agent with full metadata...") register_agent_with_metadata(AGENT_ID, PUBLIC_URL, REGISTRY_URL) try: # Start NANDA server (don't use built-in registration, we already registered) nanda.start(register=False) except KeyboardInterrupt: print("\n\n🛑 Resume agent stopped gracefully") except Exception as e: logger.error(f"Error starting agent: {e}") print(f"\n❌ Failed to start agent: {e}") sys.exit(1) if __name__ == "__main__": main()