""" Updated agent.py to prioritize user query over job profile data """ from typing import List, Dict, Any from datetime import datetime from youtube_mcp import YouTubeMCP from google_courses_mcp import GoogleCoursesMCP from coursera_mcp import CourseraMCP from khan_academy_mcp import KhanAcademyMCP import re class CareerAgent: def __init__(self, user_skills: List[Dict], user_query: str, platform_name: str = "youtube", job_profile_data: Dict = None): self.user_skills = user_skills self.user_query = user_query self.job_profile_data = job_profile_data self.platform_name = platform_name self.mcp = self.get_mcp_instance() self.skill_gaps = self.identify_skill_gaps() # NEW: Extract target role from user query self.target_role = self.extract_target_role_from_query() def extract_target_role_from_query(self) -> str: """Extract target role from user query for career transition scenarios.""" if not self.user_query: return "" query_lower = self.user_query.lower() # Common patterns for career transition queries transition_patterns = [ r"transition to\s+([^,.!?]+)", r"become\s+a?\s*([^,.!?]+)", r"move to\s+([^,.!?]+)", r"switch to\s+([^,.!?]+)", r"career change to\s+([^,.!?]+)", r"want to be\s+a?\s*([^,.!?]+)" ] for pattern in transition_patterns: match = re.search(pattern, query_lower) if match: target_role = match.group(1).strip() # Clean up the target role target_role = re.sub(r'\s+', ' ', target_role) return target_role return "" def fetch_courses(self) -> List[Dict[str, Any]]: """Fetch courses from the appropriate MCP with user query priority.""" return self.mcp.fetch_courses( self.user_skills, self.skill_gaps, self.job_profile_data, self.user_query, self.target_role ) def identify_skill_gaps(self) -> List[Dict]: """Identify skill gaps prioritizing user query intent over current job profile.""" gaps = [] # PRIORITY 1: If user query indicates career transition, focus on target role skills target_role = self.extract_target_role_from_query() if target_role: # For career transition, identify skills needed for target role target_skills = self.get_target_role_skills(target_role) # Create skill gaps based on target role requirements user_skill_map = {skill.get('skill_name', '').lower(): skill for skill in self.user_skills} for target_skill in target_skills: skill_name_lower = target_skill['skill_name'].lower() if skill_name_lower not in user_skill_map: # User doesn't have this target skill gaps.append({ 'skill_id': target_skill.get('skill_id', f'target_{len(gaps)}'), 'skill_name': target_skill['skill_name'], 'current_level': 0, 'target_level': target_skill.get('required_level', 3), 'gap_type': 'transition_skill', 'priority': 'critical', # Highest priority for transition 'source': 'user_query_target' }) else: # Check if current level is sufficient current_skill = user_skill_map[skill_name_lower] current_level = int(current_skill.get('skill_type', 1)) required_level = target_skill.get('required_level', 3) if current_level < required_level: gaps.append({ 'skill_id': current_skill.get('skill_id'), 'skill_name': target_skill['skill_name'], 'current_level': current_level, 'target_level': required_level, 'gap_type': 'transition_upgrade', 'priority': 'high', 'source': 'user_query_target' }) # PRIORITY 2: Consider current job profile skills (secondary priority) if self.job_profile_data and isinstance(self.job_profile_data, dict): job_profile_skills = self.job_profile_data.get('job_profile_skills', []) if job_profile_skills and not target_role: # Only if no career transition detected user_skill_map = {skill.get('skill_name', '').lower(): skill for skill in self.user_skills} for job_skill in job_profile_skills: job_skill_name = job_skill.get('skill_name', '').lower() required_level = int(job_skill.get('skill_type', 3)) user_skill = user_skill_map.get(job_skill_name) if not user_skill: gaps.append({ 'skill_id': job_skill.get('skill_id', 'missing'), 'skill_name': job_skill.get('skill_name', ''), 'current_level': 0, 'target_level': required_level, 'gap_type': 'current_role_skill', 'priority': 'medium', # Lower priority than transition skills 'source': 'current_job_profile' }) else: current_level = int(user_skill.get('skill_type', 1)) if current_level < required_level: gaps.append({ 'skill_id': job_skill.get('skill_id'), 'skill_name': job_skill.get('skill_name', ''), 'current_level': current_level, 'target_level': required_level, 'gap_type': 'current_role_upgrade', 'priority': 'medium', 'source': 'current_job_profile' }) # PRIORITY 3: General skill improvements (lowest priority) if not gaps: # Fallback if no specific gaps identified for skill in self.user_skills: skill_level = int(skill.get('skill_type', 1)) if skill_level < 3: # Room for improvement gaps.append({ 'skill_id': skill.get('skill_id'), 'skill_name': skill.get('skill_name'), 'current_level': skill_level, 'target_level': 3, 'gap_type': 'general_improvement', 'priority': 'low', 'source': 'user_skills' }) # Sort gaps by priority: critical > high > medium > low priority_order = {'critical': 1, 'high': 2, 'medium': 3, 'low': 4} gaps.sort(key=lambda x: priority_order.get(x.get('priority', 'medium'), 3)) return gaps def get_target_role_skills(self, target_role: str) -> List[Dict]: """Get required skills for target role based on common role requirements.""" target_role_lower = target_role.lower() # Define skill requirements for common roles role_skills_map = { 'sales manager': [ {'skill_name': 'Sales Management', 'required_level': 3}, {'skill_name': 'Team Leadership', 'required_level': 3}, {'skill_name': 'Customer Relationship Management', 'required_level': 3}, {'skill_name': 'Communication', 'required_level': 3}, {'skill_name': 'Negotiation', 'required_level': 3}, {'skill_name': 'Strategic Planning', 'required_level': 2}, {'skill_name': 'Performance Management', 'required_level': 2} ], 'project manager': [ {'skill_name': 'Project Management', 'required_level': 3}, {'skill_name': 'Agile Methodology', 'required_level': 3}, {'skill_name': 'Risk Management', 'required_level': 2}, {'skill_name': 'Leadership', 'required_level': 3}, {'skill_name': 'Communication', 'required_level': 3} ], 'marketing manager': [ {'skill_name': 'Digital Marketing', 'required_level': 3}, {'skill_name': 'Brand Management', 'required_level': 3}, {'skill_name': 'Marketing Strategy', 'required_level': 3}, {'skill_name': 'Analytics', 'required_level': 2}, {'skill_name': 'Communication', 'required_level': 3} ], 'product manager': [ {'skill_name': 'Product Management', 'required_level': 3}, {'skill_name': 'User Experience Design', 'required_level': 2}, {'skill_name': 'Data Analysis', 'required_level': 2}, {'skill_name': 'Strategic Planning', 'required_level': 3}, {'skill_name': 'Communication', 'required_level': 3} ] } # Find best match for target role for role, skills in role_skills_map.items(): if role in target_role_lower or any(word in target_role_lower for word in role.split()): return skills # Fallback: generic management/leadership skills for unrecognized roles return [ {'skill_name': 'Leadership', 'required_level': 3}, {'skill_name': 'Management', 'required_level': 3}, {'skill_name': 'Communication', 'required_level': 3}, {'skill_name': 'Strategic Thinking', 'required_level': 2} ] def get_mcp_instance(self): """Get the appropriate MCP instance based on the platform name.""" if self.platform_name.lower() == "youtube": return YouTubeMCP() elif self.platform_name.lower() == "google": return GoogleCoursesMCP() elif self.platform_name.lower() == "khan_academy": return KhanAcademyMCP() else: raise ValueError(f"Unsupported platform: {self.platform_name}") def recommend_courses(self, courses): """Recommend courses with user query priority and meaningful reasons.""" recommended_courses = [] # Determine primary context (user query vs job profile) target_role = self.extract_target_role_from_query() primary_context = target_role if target_role else ( self.job_profile_data.get('job_profile_name', '') if self.job_profile_data else '' ) current_year = datetime.now().year for course in courses: snippet = course.get('snippet', {}) title = snippet.get('title', '') description = snippet.get('description', '') # Extract thumbnail URL thumbnails = snippet.get('thumbnails', {}) thumbnail_url = None if 'high' in thumbnails: thumbnail_url = thumbnails['high'].get('url') elif 'medium' in thumbnails: thumbnail_url = thumbnails['medium'].get('url') elif 'default' in thumbnails: thumbnail_url = thumbnails['default'].get('url') # Find which skill gaps this course addresses (prioritize critical/high gaps) addressed_gaps = [] gap_priorities = [] critical_gaps = [gap for gap in self.skill_gaps if gap.get('priority') == 'critical'] high_gaps = [gap for gap in self.skill_gaps if gap.get('priority') == 'high'] # Check critical gaps first for gap in critical_gaps: gap_name = gap['skill_name'].lower() if gap_name in title.lower() or gap_name in description.lower(): addressed_gaps.append(gap_name) gap_priorities.append('critical') # Then check high priority gaps for gap in high_gaps: gap_name = gap['skill_name'].lower() if (gap_name in title.lower() or gap_name in description.lower()) and gap_name not in addressed_gaps: addressed_gaps.append(gap_name) gap_priorities.append('high') # If no specific gaps found, use top priority gaps as fallback if not addressed_gaps and self.skill_gaps: top_gaps = [gap for gap in self.skill_gaps if gap.get('priority') in ['critical', 'high']][:2] if not top_gaps: top_gaps = self.skill_gaps[:2] addressed_gaps = [gap['skill_name'].lower() for gap in top_gaps] gap_priorities = [gap.get('priority', 'medium') for gap in top_gaps] # Create enhanced reason based on user query priority reason_parts = [] if target_role: reason_parts.append(f"Essential for your career transition to {target_role}") elif primary_context: reason_parts.append(f"Recommended for your target role as {primary_context}") else: reason_parts.append("Recommended to enhance your professional skills") if addressed_gaps: if 'critical' in gap_priorities: reason_parts.append(f"Addresses critical transition skills: {', '.join(addressed_gaps)}") elif 'high' in gap_priorities: reason_parts.append(f"Covers essential skills: {', '.join(addressed_gaps)}") else: reason_parts.append(f"Helps improve skills in: {', '.join(addressed_gaps)}") reason_parts.append(f"Course content: {description[:150]}...") reason = ". ".join(reason_parts) # Calculate match score with user query priority match_score = 60 # Base score # Higher boost for career transition courses if target_role and any(word in title.lower() for word in target_role.split()): match_score += 30 elif 'critical' in gap_priorities: match_score += 25 elif 'high' in gap_priorities: match_score += 15 elif 'medium' in gap_priorities: match_score += 10 # Additional boost for direct query match if self.user_query: query_words = self.user_query.lower().split() title_matches = sum(1 for word in query_words if word in title.lower()) match_score += min(title_matches * 5, 15) # Get video ID video_id = course.get('id', {}).get('videoId', '') # Create course recommendation object course_recommendation = { 'courseId': "auto-generated", 'courseName': title, 'platform': self.platform_name, 'instructor': snippet.get('channelTitle', ''), 'duration': "Self-paced", 'matchScore': min(match_score, 100), 'reason': reason, 'score': round(min(match_score, 100) / 10, 1), 'isOnlineCourse': True, 'url': f"https://www.youtube.com/watch?v={video_id}", 'type': "video", 'lastUpdated': current_year, 'addressed_gaps': addressed_gaps, 'gap_priorities': gap_priorities, 'target_role_alignment': target_role if target_role else None, 'transition_focused': bool(target_role) } if thumbnail_url: course_recommendation['thumbnail'] = thumbnail_url recommended_courses.append(course_recommendation) # Sort by match score (highest first) recommended_courses.sort(key=lambda x: x['matchScore'], reverse=True) return recommended_courses