import requests
import json
import sys
import os
import re
from agent import CareerAgent
# ========== Configuration ==========
OLLAMA_API_URL = "http://localhost:11434/api/generate"
MODEL_NAME = "gemma3:12b"
# ========== Enhanced Helper Functions ==========
def calculate_job_profile_skill_priority_score(user_skills, job_profile_skills, course_skills):
"""
Calculate priority score based on job profile skills alignment.
Higher scores for courses that align with job profile requirements.
Skills required for the target role that the user lacks are given the highest priority.
"""
if not course_skills:
return 0
# Create skill maps for efficient lookup
user_skill_map = {
skill.get('skill_id'): {
'name': skill.get('skill_name'),
'current_level': int(skill.get('skill_type', 1))
} for skill in (user_skills or []) if skill.get('skill_id')
}
course_skill_map = {
skill.get('skill_id'): {
'name': skill.get('skill_name'),
'course_level': int(skill.get('skill_type', 1))
} for skill in course_skills if skill.get('skill_id')
}
# Also map by skill name for better matching
user_skill_name_map = {
skill.get('skill_name', '').lower(): {
'current_level': int(skill.get('skill_type', 1))
} for skill in (user_skills or []) if skill.get('skill_name')
}
course_skill_name_map = {
skill.get('skill_name', '').lower(): {
'course_level': int(skill.get('skill_type', 1))
} for skill in course_skills if skill.get('skill_name')
}
priority_score = 0
# Prioritize job profile skills
for target_skill in job_profile_skills:
skill_id = target_skill.get('skill_id')
skill_name = target_skill.get('skill_name', '').lower()
required_level = int(target_skill.get('skill_type', 1))
# Check if course covers this skill (by ID or name)
course_covers_skill = False
course_level = 0
if skill_id and skill_id in course_skill_map:
course_covers_skill = True
course_level = course_skill_map[skill_id]['course_level']
elif skill_name in course_skill_name_map:
course_covers_skill = True
course_level = course_skill_name_map[skill_name]['course_level']
if course_covers_skill:
# Get user's current level for this skill
user_current_level = 0
if skill_id and skill_id in user_skill_map:
user_current_level = user_skill_map[skill_id]['current_level']
elif skill_name in user_skill_name_map:
user_current_level = user_skill_name_map[skill_name]['current_level']
# CRITICAL: User doesn't have the required skill at all
if user_current_level == 0:
priority_score += 35 # Maximum critical priority
# HIGH: User has the skill but significantly below the required level
elif user_current_level < required_level:
gap = required_level - user_current_level
priority_score += 25 + (gap * 4) # Scale based on gap size
# MEDIUM: Course can advance the skill beyond the required level
elif course_level > user_current_level:
priority_score += 15
# LOW: Skill maintenance or reinforcement
else:
priority_score += 5
# Additional scoring for courses that teach new skills (not in target skills)
for course_skill in course_skills:
skill_id = course_skill.get('skill_id')
skill_name = course_skill.get('skill_name', '').lower()
# Check if this is a new skill not in target skills
is_target_skill = any(
(skill_id and target_skill.get('skill_id') == skill_id) or
(skill_name and target_skill.get('skill_name', '').lower() == skill_name)
for target_skill in job_profile_skills
)
if not is_target_skill:
# Check if user has this skill
user_current_level = 0
if skill_id and skill_id in user_skill_map:
user_current_level = user_skill_map[skill_id]['current_level']
elif skill_name in user_skill_name_map:
user_current_level = user_skill_name_map[skill_name]['current_level']
# Bonus for teaching new skills
if user_current_level == 0:
priority_score += 10 # Medium priority for new skills
elif int(course_skill.get('skill_type', 1)) > user_current_level:
priority_score += 8 # Low priority for skill advancement
return min(priority_score, 100) # Cap at 100
def enhanced_match_skills(user_skills, course_skills, job_profile_skills=None, designation=None):
"""Enhanced skill matching considering job profile requirements and career progression"""
# Base skill matching score
base_score = match_skills(user_skills, course_skills)
# Job profile alignment bonus (highest weight)
job_profile_bonus = 0
if job_profile_skills:
job_profile_bonus = calculate_job_profile_skill_priority_score(
user_skills, job_profile_skills, course_skills
)
# Career progression bonus based on designation
designation_bonus = calculate_designation_progression_bonus(
designation, course_skills, user_skills
)
# Skill gap filling bonus
skill_gap_bonus = calculate_skill_gap_bonus(user_skills, course_skills)
# Weighted combination of all factors
total_score = (
base_score * 0.25 + # Basic skill matching
job_profile_bonus * 0.45 + # Job profile alignment (highest priority)
designation_bonus * 0.20 + # Career progression
skill_gap_bonus * 0.10 # General skill development
)
return min(total_score, 100)
def calculate_skill_gap_bonus(user_skills, course_skills):
"""Calculate bonus for courses that fill skill gaps"""
if not course_skills:
return 0
user_skill_ids = {skill.get('skill_id') for skill in user_skills if skill.get('skill_id')}
course_skill_ids = {skill.get('skill_id') for skill in course_skills if skill.get('skill_id')}
# Bonus for teaching new skills
new_skills = course_skill_ids - user_skill_ids
new_skill_bonus = len(new_skills) * 5
# Bonus for advancing existing skills
advancement_bonus = 0
user_skill_levels = {
skill.get('skill_id'): int(skill.get('skill_type', 1))
for skill in user_skills if skill.get('skill_id')
}
for course_skill in course_skills:
skill_id = course_skill.get('skill_id')
course_level = int(course_skill.get('skill_type', 1))
user_level = user_skill_levels.get(skill_id, 0)
if user_level > 0 and course_level > user_level:
advancement_bonus += (course_level - user_level) * 3
return min(new_skill_bonus + advancement_bonus, 25)
def calculate_designation_progression_bonus(designation, course_skills, user_skills):
"""Calculate bonus based on career progression relevance for designation"""
if not designation or not course_skills:
return 0
designation_lower = designation.lower()
progression_bonus = 0
# Enhanced career progression mapping
progression_skills_map = {
'intern': {
'next_level': ['junior', 'associate'],
'skills': ['communication', 'time management', 'learning', 'basic technical'],
'bonus': 8
},
'junior': {
'next_level': ['engineer', 'developer', 'analyst'],
'skills': ['programming', 'problem solving', 'teamwork', 'project work'],
'bonus': 10
},
'engineer': {
'next_level': ['senior engineer', 'lead engineer', 'specialist'],
'skills': ['leadership', 'architecture', 'mentoring', 'advanced technical'],
'bonus': 12
},
'senior': {
'next_level': ['lead', 'principal', 'manager'],
'skills': ['leadership', 'strategy', 'system design', 'team management'],
'bonus': 15
},
'lead': {
'next_level': ['manager', 'director', 'principal'],
'skills': ['team leadership', 'project management', 'strategic planning', 'decision making'],
'bonus': 18
},
'manager': {
'next_level': ['senior manager', 'director'],
'skills': ['people management', 'budgeting', 'strategic thinking', 'stakeholder management'],
'bonus': 20
},
'analyst': {
'next_level': ['senior analyst', 'data scientist', 'consultant'],
'skills': ['data analysis', 'visualization', 'reporting', 'statistical analysis'],
'bonus': 12
},
'developer': {
'next_level': ['senior developer', 'tech lead', 'architect'],
'skills': ['advanced programming', 'code review', 'technical leadership', 'system design'],
'bonus': 12
}
}
# Find matching progression path
progression_info = None
for role_key, info in progression_skills_map.items():
if role_key in designation_lower:
progression_info = info
break
if not progression_info:
return 5 # Default small bonus
# Check if course covers progression-relevant skills
course_skill_names = [skill.get('skill_name', '').lower() for skill in course_skills]
relevant_skills = progression_info['skills']
base_bonus = progression_info['bonus']
skill_matches = 0
for relevant_skill in relevant_skills:
for course_skill_name in course_skill_names:
if any(keyword in course_skill_name for keyword in relevant_skill.split()):
skill_matches += 1
break
if skill_matches > 0:
progression_bonus = base_bonus + (skill_matches * 3)
return min(progression_bonus, 35)
def match_skills(user_skills, course_skills):
"""Original skill matching function (preserved for compatibility)"""
if not course_skills or not user_skills:
return 0
user_skill_levels = {
skill.get('skill_id'): int(skill.get('skill_type', 1))
for skill in user_skills if skill.get('skill_id')
}
course_skill_levels = {
skill.get('skill_id'): int(skill.get('skill_type', 1))
for skill in course_skills if skill.get('skill_id')
}
if not user_skill_levels or not course_skill_levels:
return 0
match_score = 0
total_possible_matches = 0
for skill_id, course_level in course_skill_levels.items():
total_possible_matches += 1
if skill_id in user_skill_levels:
user_level = user_skill_levels[skill_id]
# Perfect match
if course_level == user_level:
match_score += 3
# Course is one level higher (good for progression)
elif course_level == user_level + 1:
match_score += 2
# Course is higher level (challenging but beneficial)
elif course_level > user_level:
match_score += 1
else:
# New skill - moderate value
match_score += 1
if total_possible_matches > 0:
return min((match_score / (total_possible_matches * 3)) * 100, 100)
return 0
def analyze_job_profile_alignment(user_skills, job_profile_skills, job_profile_name):
"""Analyze how well user's current skills align with job profile requirements"""
if not job_profile_skills:
return {
"alignment_percentage": 0.0,
"missing_skills": [],
"skill_gaps": [],
"strengths": [],
"job_profile_name": job_profile_name or "Not specified"
}
user_skill_map = {
skill.get('skill_id'): {
'name': skill.get('skill_name'),
'level': int(skill.get('skill_type', 1))
} for skill in user_skills if skill.get('skill_id')
}
missing_skills = []
skill_gaps = []
strengths = []
total_required_skills = len(job_profile_skills)
aligned_skills = 0
for required_skill in job_profile_skills:
skill_id = required_skill.get('skill_id')
skill_name = required_skill.get('skill_name')
required_level = int(required_skill.get('skill_type', 1))
if skill_id in user_skill_map:
user_level = user_skill_map[skill_id]['level']
if user_level >= required_level:
strengths.append({
'skill_name': skill_name,
'user_level': user_level,
'required_level': required_level
})
aligned_skills += 1
else:
skill_gaps.append({
'skill_name': skill_name,
'user_level': user_level,
'required_level': required_level,
'gap': required_level - user_level
})
else:
missing_skills.append({
'skill_name': skill_name,
'required_level': required_level
})
alignment_percentage = (aligned_skills / total_required_skills) * 100 if total_required_skills > 0 else 0.0
return {
"alignment_percentage": round(alignment_percentage, 1),
"missing_skills": missing_skills,
"skill_gaps": skill_gaps,
"strengths": strengths,
"job_profile_name": job_profile_name or "Not specified"
}
def build_enhanced_prompt_section(job_profile_analysis):
"""Build the job profile analysis section for the prompt with proper formatting"""
alignment_pct = job_profile_analysis['alignment_percentage']
# Ensure the percentage is properly formatted as a string
alignment_text = f"{alignment_pct:.1f}%" if alignment_pct > 0 else "0.0%"
return f"""
PROFESSIONAL ANALYSIS CONTEXT:
Current Job Profile Alignment: {alignment_text}
Missing Critical Skills: {len(job_profile_analysis['missing_skills'])} skills
Skill Gaps to Address: {len(job_profile_analysis['skill_gaps'])} skills
Current Strengths: {len(job_profile_analysis['strengths'])} skills at required level
"""
def extract_course_recommendations(answer, course_mapping, user_skills, job_profile_skills, designation, user_query):
"""Enhanced course recommendation extraction with LLM-based target role analysis"""
recommended_courses = []
# Extract target role for better course scoring
target_role = extract_target_role_from_query(user_query)
# Extract course IDs from the answer
course_ids = re.findall(r'\b(\d+)\b', answer)
course_ids = list(set(course_ids)) # Remove duplicates
# Score and rank courses
course_scores = []
for course_id in course_ids:
if course_id in course_mapping:
course_data = course_mapping[course_id]
course_skills = course_data.get('skills', [])
# Use enhanced matching that considers target role
relevance_score = enhanced_match_skills_for_target_role(
user_skills=user_skills,
course_skills=course_skills,
job_profile_skills=job_profile_skills,
designation=designation,
target_role=target_role
)
course_scores.append({
'course_id': course_id,
'course_data': course_data,
'relevance_score': relevance_score
})
# Sort by relevance score
course_scores.sort(key=lambda x: x['relevance_score'], reverse=True)
# Build recommended courses list
for course_item in course_scores:
course_data = course_item['course_data']
recommended_courses.append({
"courseId": course_item['course_id'],
"courseName": course_data.get('name', 'Unknown Course'),
"shortDescription": course_data.get('short_description', ''),
"skills": course_data.get('skills', []),
"relevance_score": course_item['relevance_score']
})
return recommended_courses
def calculate_course_name_similarity(text1, text2):
"""Calculate similarity between two course names"""
text1_words = set(text1.lower().split())
text2_words = set(text2.lower().split())
if not text1_words or not text2_words:
return 0
intersection = text1_words.intersection(text2_words)
union = text1_words.union(text2_words)
return len(intersection) / len(union) if union else 0
def remove_course_ids_from_text(text, course_mapping):
"""Remove course IDs from the text response"""
# Remove various course ID patterns
patterns_to_remove = [
r'Course ID:?\s*[\w\d_\-\.]+',
r'\(Course ID:?\s*[\w\d_\-\.]+\)',
r'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9\.[^\.]+\.[^\.]+',
r'\(ID:?\s*[\w\d_\-\.]+\)',
r'ID:?\s*[\w\d_\-\.]+',
]
for pattern in patterns_to_remove:
text = re.sub(pattern, '', text, flags=re.IGNORECASE)
# Remove specific course IDs
for course_id in course_mapping:
text = text.replace(f"(Course ID: {course_id})", "")
text = text.replace(f"Course ID: {course_id}", "")
text = text.replace(f"(ID: {course_id})", "")
text = text.replace(f"ID: {course_id}", "")
text = text.replace(course_id, "")
# Clean up formatting
text = re.sub(r'\(\s*\)', '', text)
text = re.sub(r'\s{2,}', ' ', text)
text = re.sub(r':\s*-', ':', text)
text = text.strip()
return text
def query_ollama(prompt):
"""Query Ollama API with enhanced error handling"""
try:
headers = {'Content-Type': 'application/json'}
payload = {
"model": MODEL_NAME,
"prompt": f"You are a professional career advisor specializing in personalized career guidance. You analyze user skills, job profile requirements, and current designation to provide intelligent course recommendations that maximize career impact and professional growth.\n\n{prompt}\n\n",
"stream": False,
"options": {
"temperature": 0.7,
"top_p": 0.95,
"top_k": 40,
"stop": ["", ""]
}
}
response = requests.post(OLLAMA_API_URL, headers=headers, json=payload, timeout=30)
if response.status_code == 200:
response_json = response.json()
return response_json.get('response', '')
else:
return f"Error: API returned status code {response.status_code}"
except requests.exceptions.Timeout:
return "Error: Request timeout. Please try again."
except requests.exceptions.ConnectionError:
return "Error: Unable to connect to Ollama API. Please ensure Ollama is running."
except Exception as e:
return f"Error querying Ollama API: {str(e)}"
def get_skill_level_name(level):
"""Convert numeric skill level to readable name"""
level_map = {
1: "Beginner",
2: "Intermediate",
3: "Advanced"
}
return level_map.get(level, "None")
def analyze_skill_gaps(user_skills, job_profile_skills=None, job_profile_name=None):
"""Enhanced skill gap analysis with job profile consideration"""
if not user_skills and not job_profile_skills:
return {
"chart_data": {
"radar": {
"labels": ["No Skills Data Available"],
"datasets": [
{
"label": "Current Skills",
"data": [0],
"backgroundColor": "rgba(54, 162, 235, 0.2)",
"borderColor": "rgba(54, 162, 235, 1)",
"pointBackgroundColor": "rgba(54, 162, 235, 1)"
},
{
"label": "Target Skills",
"data": [3],
"backgroundColor": "rgba(255, 99, 132, 0.2)",
"borderColor": "rgba(255, 99, 132, 1)",
"pointBackgroundColor": "rgba(255, 99, 132, 1)"
}
]
},
"bar": {
"labels": ["No Skills Data Available"],
"datasets": [
{
"label": "Skill Gap",
"data": [3],
"backgroundColor": "rgba(255, 159, 64, 0.7)"
}
]
}
},
"summary": "No skill data available. Complete a skills assessment for personalized recommendations.",
"job_profile_gaps": [],
"other_gaps": [],
"strengths": [],
"total_skills_analyzed": 0
}
# Comprehensive skill analysis
all_skills = {}
# Process user skills
for skill in user_skills or []:
skill_id = skill.get('skill_id')
if skill_id:
all_skills[skill_id] = {
'name': skill.get('skill_name', 'Unknown Skill'),
'current_level': int(skill.get('skill_type', 1)),
'target_level': 3, # Default advanced target
'is_user_skill': True,
'is_job_profile_skill': False
}
# Process job profile skills
for skill in job_profile_skills or []:
skill_id = skill.get('skill_id')
if skill_id:
required_level = int(skill.get('skill_type', 1))
if skill_id in all_skills:
all_skills[skill_id]['target_level'] = max(required_level, all_skills[skill_id]['target_level'])
all_skills[skill_id]['is_job_profile_skill'] = True
else:
all_skills[skill_id] = {
'name': skill.get('skill_name', 'Unknown Skill'),
'current_level': 0,
'target_level': required_level,
'is_user_skill': False,
'is_job_profile_skill': True
}
# Calculate gaps and prepare analysis
skills_analysis = []
for skill_id, skill_data in all_skills.items():
gap = skill_data['target_level'] - skill_data['current_level']
skill_data['gap'] = max(gap, 0) # Ensure non-negative gaps
skills_analysis.append(skill_data)
# Prioritize: job profile skills first, then by gap size
skills_analysis.sort(key=lambda x: (not x['is_job_profile_skill'], -x['gap'], x['name']))
# Select top skills for visualization
top_skills = skills_analysis[:12] # Show more skills in analysis
# Prepare chart data
skill_names = [skill["name"] for skill in top_skills]
current_levels = [skill["current_level"] for skill in top_skills]
target_levels = [skill["target_level"] for skill in top_skills]
gap_values = [skill["gap"] for skill in top_skills]
chart_data = {
"radar": {
"labels": skill_names,
"datasets": [
{
"label": "Current Skills",
"data": current_levels,
"backgroundColor": "rgba(54, 162, 235, 0.2)",
"borderColor": "rgba(54, 162, 235, 1)",
"pointBackgroundColor": "rgba(54, 162, 235, 1)"
},
{
"label": f"Target Skills{' (' + job_profile_name + ')' if job_profile_name else ''}",
"data": target_levels,
"backgroundColor": "rgba(255, 99, 132, 0.2)",
"borderColor": "rgba(255, 99, 132, 1)",
"pointBackgroundColor": "rgba(255, 99, 132, 1)"
}
]
},
"bar": {
"labels": skill_names,
"datasets": [
{
"label": "Skill Gap",
"data": gap_values,
"backgroundColor": [
"rgba(255, 99, 132, 0.7)" if skill['is_job_profile_skill'] else "rgba(255, 159, 64, 0.7)"
for skill in top_skills
]
}
]
}
}
# Generate comprehensive summary
job_profile_gaps = [skill for skill in top_skills if skill['is_job_profile_skill'] and skill['gap'] > 0]
other_gaps = [skill for skill in top_skills if not skill['is_job_profile_skill'] and skill['gap'] > 0]
strengths = [skill for skill in top_skills if skill['gap'] == 0 and skill['current_level'] >= 2]
summary_parts = []
if job_profile_name:
if job_profile_gaps:
critical_gaps = [skill for skill in job_profile_gaps if skill['gap'] >= 2]
if critical_gaps:
summary_parts.append(f"CRITICAL: {len(critical_gaps)} essential {job_profile_name} skills need development")
else:
summary_parts.append(f"Minor gaps in {len(job_profile_gaps)} {job_profile_name} skills")
else:
summary_parts.append(f"✓ All {job_profile_name} skill requirements met")
if other_gaps:
summary_parts.append(f"{len(other_gaps)} additional skills for career advancement")
if strengths:
summary_parts.append(f"Strong in {len(strengths)} key areas")
# Generate the summary with the desired text for skill levels
if job_profile_gaps:
priority_skills = [f"{skill['name']} ({get_skill_level_name(skill['current_level'])} level)" for skill in job_profile_gaps]
summary_parts.append(f"For your {job_profile_name} role, priority skill gaps: {', '.join(priority_skills)}")
if other_gaps:
growth_opportunities = [f"{skill['name']} ({get_skill_level_name(skill['current_level'])} level)" for skill in other_gaps]
summary_parts.append(f"Additional growth opportunities: {', '.join(growth_opportunities)}")
summary = ". ".join(summary_parts) if summary_parts else "Complete skills assessment for detailed analysis."
return {
"chart_data": chart_data,
"summary": summary,
"job_profile_gaps": job_profile_gaps,
"other_gaps": other_gaps,
"strengths": strengths,
"total_skills_analyzed": len(skills_analysis)
}
def analyze_skill_gaps_with_target_role(user_skills, job_profile_skills=None, job_profile_name=None, user_query=""):
"""Enhanced skill gap analysis that prioritizes target role from query"""
# Extract target role from query
target_role = extract_target_role_from_query(user_query)
target_role_skills = get_target_role_required_skills(target_role) if target_role else []
# Create comprehensive skill analysis
all_skills = {}
# Process existing user skills
for skill in user_skills or []:
skill_id = skill.get('skill_id')
skill_name = skill.get('skill_name')
if skill_id and skill_name:
all_skills[skill_name.lower()] = {
'name': skill_name,
'current_level': int(skill.get('skill_type', 1)),
'target_level': int(skill.get('skill_type', 1)),
'is_user_skill': True,
'is_job_profile_skill': False,
'is_target_role_skill': False,
'priority': 'existing'
}
# Process target role skills (HIGHEST PRIORITY)
for skill_info in target_role_skills:
skill_name = skill_info['skill_name']
skill_key = skill_name.lower()
required_level = skill_info['skill_type']
priority = skill_info['priority']
if skill_key in all_skills:
# Update existing skill with target role requirements
all_skills[skill_key]['target_level'] = required_level
all_skills[skill_key]['is_target_role_skill'] = True
all_skills[skill_key]['priority'] = priority
else:
# NEW SKILL required for target role - CRITICAL PRIORITY
all_skills[skill_key] = {
'name': skill_name,
'current_level': 0,
'target_level': required_level,
'is_user_skill': False,
'is_job_profile_skill': False,
'is_target_role_skill': True,
'priority': 'critical' # Force critical for new target role skills
}
# Process current job profile skills
for skill in job_profile_skills or []:
skill_name = skill.get('skill_name')
if skill_name:
skill_key = skill_name.lower()
required_level = int(skill.get('skill_type', 1))
if skill_key in all_skills:
# Only update if not already a target role skill
if not all_skills[skill_key]['is_target_role_skill']:
all_skills[skill_key]['target_level'] = max(required_level, all_skills[skill_key]['target_level'])
all_skills[skill_key]['is_job_profile_skill'] = True
all_skills[skill_key]['priority'] = 'current_role'
else:
# NEW SKILL from job profile - HIGH PRIORITY
all_skills[skill_key] = {
'name': skill_name,
'current_level': 0,
'target_level': required_level,
'is_user_skill': False,
'is_job_profile_skill': True,
'is_target_role_skill': False,
'priority': 'high' # High priority for new job profile skills
}
# Calculate gaps and apply FIXED priority logic
skills_analysis = []
for skill_data in all_skills.values():
gap = skill_data['target_level'] - skill_data['current_level']
skill_data['gap'] = max(gap, 0)
# FIXED PRIORITY LOGIC: New skills (current_level = 0) get highest priority
if skill_data['current_level'] == 0 and skill_data['gap'] > 0:
if skill_data['is_target_role_skill']:
skill_data['priority'] = 'critical' # New target role skills = CRITICAL
skill_data['priority_score'] = 1 # Highest priority score
elif skill_data['is_job_profile_skill']:
skill_data['priority'] = 'high' # New job profile skills = HIGH
skill_data['priority_score'] = 2
else:
skill_data['priority'] = 'high' # Other new skills = HIGH
skill_data['priority_score'] = 2
else:
# Existing skills with gaps
if skill_data['is_target_role_skill']:
if skill_data.get('priority') == 'critical':
skill_data['priority_score'] = 3
else:
skill_data['priority_score'] = 4
elif skill_data['is_job_profile_skill']:
skill_data['priority_score'] = 5
else:
skill_data['priority_score'] = 6
skills_analysis.append(skill_data)
# Sort by priority: NEW SKILLS with lowest priority_score come first
skills_analysis.sort(key=lambda x: (
x.get('priority_score', 10), # Primary sort by priority score
-x['gap'], # Secondary sort by gap size (descending)
x['name'] # Tertiary sort by name for consistency
))
# Get top skills
top_skills = skills_analysis[:12]
# Prepare chart data
skill_names = [skill["name"] for skill in top_skills]
current_levels = [skill["current_level"] for skill in top_skills]
target_levels = [skill["target_level"] for skill in top_skills]
gap_values = [skill["gap"] for skill in top_skills]
# Create color coding based on priority - FIXED FOR NEW SKILLS
bar_colors = []
for skill in top_skills:
if skill['current_level'] == 0: # New skills get red coloring
if skill['is_target_role_skill']:
bar_colors.append("rgba(220, 53, 69, 0.9)") # Bright red for new target role skills
else:
bar_colors.append("rgba(255, 107, 107, 0.8)") # Lighter red for new job skills
elif skill['priority'] == 'critical':
bar_colors.append("rgba(220, 53, 69, 0.7)") # Red for critical existing skills
elif skill['is_target_role_skill']:
bar_colors.append("rgba(255, 193, 7, 0.8)") # Yellow for target role skills
elif skill['is_job_profile_skill']:
bar_colors.append("rgba(13, 202, 240, 0.7)") # Blue for current role skills
else:
bar_colors.append("rgba(108, 117, 125, 0.6)") # Gray for existing skills
chart_data = {
"radar": {
"labels": skill_names,
"datasets": [
{
"label": "Current Skills",
"data": current_levels,
"backgroundColor": "rgba(54, 162, 235, 0.2)",
"borderColor": "rgba(54, 162, 235, 1)",
"pointBackgroundColor": "rgba(54, 162, 235, 1)"
},
{
"label": f"Target Skills{' (' + target_role + ')' if target_role else (' (' + job_profile_name + ')' if job_profile_name else '')}",
"data": target_levels,
"backgroundColor": "rgba(255, 99, 132, 0.2)",
"borderColor": "rgba(255, 99, 132, 1)",
"pointBackgroundColor": "rgba(255, 99, 132, 1)"
}
]
},
"bar": {
"labels": skill_names,
"datasets": [
{
"label": "Skill Gap",
"data": gap_values,
"backgroundColor": bar_colors
}
]
}
}
# Generate categorized skill lists - FIXED CATEGORIZATION
# Separate new skills from existing skills with gaps
new_skills_needed = [skill for skill in top_skills if skill['current_level'] == 0 and skill['gap'] > 0]
critical_new_skills = [skill for skill in new_skills_needed if skill['is_target_role_skill']]
high_priority_new_skills = [skill for skill in new_skills_needed if not skill['is_target_role_skill']]
# Existing skills that need improvement
target_role_gaps = [skill for skill in top_skills if skill['is_target_role_skill'] and skill['gap'] > 0 and skill['current_level'] > 0]
current_role_gaps = [skill for skill in top_skills if skill['is_job_profile_skill'] and skill['gap'] > 0 and not skill['is_target_role_skill']]
strengths = [skill for skill in top_skills if skill['current_level'] >= skill['target_level'] and skill['current_level'] >= 2]
# CORRECTED: other_gaps should prioritize NEW SKILLS first
other_gaps = critical_new_skills + high_priority_new_skills + target_role_gaps
if not other_gaps: # Fallback if no target role skills
other_gaps = [skill for skill in top_skills if not skill['is_job_profile_skill'] and skill['gap'] > 0]
job_profile_gaps = current_role_gaps
# Generate summary with correct priority emphasis
summary_parts = []
# PRIORITY 1: New critical skills for target role
if critical_new_skills:
critical_skill_names = [skill['name'] for skill in critical_new_skills[:3]]
summary_parts.append(f"CRITICAL NEW SKILLS NEEDED: {', '.join(critical_skill_names)}")
# PRIORITY 2: New high priority skills
if high_priority_new_skills:
high_skill_names = [skill['name'] for skill in high_priority_new_skills[:3]]
summary_parts.append(f"HIGH PRIORITY NEW SKILLS: {', '.join(high_skill_names)}")
# PRIORITY 3: Target role existing skill improvements
if target_role and target_role_gaps:
existing_improvements = [f"{skill['name']} ({get_skill_level_name(skill['current_level'])} → {get_skill_level_name(skill['target_level'])})" for skill in target_role_gaps[:3]]
summary_parts.append(f"Enhance for {target_role}: {', '.join(existing_improvements)}")
if current_role_gaps and not target_role:
new_current_skills = [skill for skill in current_role_gaps if skill['current_level'] == 0]
if new_current_skills:
summary_parts.append(f"New skills for current role: {len(new_current_skills)} skills to acquire")
else:
summary_parts.append(f"Current role development: {len(current_role_gaps)} skills to enhance")
if strengths:
summary_parts.append(f"Strong foundation: {len(strengths)} advanced skills")
summary = ". ".join(summary_parts) if summary_parts else "Complete skills assessment for detailed analysis."
return {
"chart_data": chart_data,
"summary": summary,
"job_profile_gaps": job_profile_gaps,
"other_gaps": other_gaps, # Now correctly prioritizes new skills first
"strengths": strengths,
"total_skills_analyzed": len(skills_analysis)
}
def get_target_role_required_skills(target_role):
"""Get required skills for target role from query analysis - UPDATED"""
if not target_role:
return []
target_role_lower = target_role.lower()
# Enhanced skill mapping with more comprehensive roles
role_skill_mapping = {
'sales manager': [
{'skill_name': 'Sales Strategy', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Team Leadership', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Customer Relationship Management', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Negotiation', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Business Development', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Sales Forecasting', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Market Analysis', 'skill_type': 2, 'priority': 'medium'},
{'skill_name': 'Presentation Skills', 'skill_type': 2, 'priority': 'medium'},
],
'product manager': [
{'skill_name': 'Product Strategy', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Market Research', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'User Experience Design', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Data Analysis', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Agile Methodology', 'skill_type': 2, 'priority': 'medium'},
{'skill_name': 'Stakeholder Management', 'skill_type': 2, 'priority': 'medium'},
],
'marketing manager': [
{'skill_name': 'Digital Marketing', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Brand Management', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Content Strategy', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Social Media Marketing', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Marketing Analytics', 'skill_type': 2, 'priority': 'medium'},
{'skill_name': 'Campaign Management', 'skill_type': 2, 'priority': 'medium'},
],
'project manager': [
{'skill_name': 'Project Planning', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Risk Management', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Stakeholder Management', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Budget Management', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Quality Assurance', 'skill_type': 2, 'priority': 'medium'},
{'skill_name': 'Agile Methodology', 'skill_type': 2, 'priority': 'medium'},
],
'business analyst': [
{'skill_name': 'Business Process Analysis', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Requirements Gathering', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Data Modeling', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'SQL', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Process Improvement', 'skill_type': 2, 'priority': 'medium'},
{'skill_name': 'Stakeholder Management', 'skill_type': 2, 'priority': 'medium'},
],
'software development engineer': [
{'skill_name': 'Programming', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Software Development', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Algorithms', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Data Structures', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'System Design', 'skill_type': 2, 'priority': 'medium'},
{'skill_name': 'Debugging', 'skill_type': 2, 'priority': 'medium'},
]
}
# Find exact or partial matches
for role_key, skills in role_skill_mapping.items():
if role_key in target_role_lower or target_role_lower in role_key:
return skills
# Check for partial matches (e.g., "sales" in "sales manager")
role_words = role_key.split()
target_words = target_role_lower.split()
if any(word in target_words for word in role_words):
return skills
# Generic management skills if no specific role found but contains management keywords
if any(keyword in target_role_lower for keyword in ['manager', 'lead', 'director', 'head', 'supervisor']):
return [
{'skill_name': 'Team Leadership', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'Strategic Planning', 'skill_type': 3, 'priority': 'critical'},
{'skill_name': 'People Management', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Decision Making', 'skill_type': 2, 'priority': 'high'},
{'skill_name': 'Budget Management', 'skill_type': 2, 'priority': 'medium'},
{'skill_name': 'Performance Management', 'skill_type': 2, 'priority': 'medium'},
]
return []
def build_llm_skill_analysis_prompt(user_query, user_skills=None, job_profile_skills=None, job_profile_name=None, designation=None):
"""Build advanced LLM prompt for intelligent skill analysis based on user query"""
# User skills context
user_skills_text = ""
if user_skills:
user_skills_list = []
for skill in user_skills:
skill_name = skill.get('skill_name', 'Unknown')
skill_level = get_skill_level_name(int(skill.get('skill_type', 1)))
user_skills_list.append(f"{skill_name} ({skill_level})")
user_skills_text = f"Current user skills: {', '.join(user_skills_list)}"
# Job profile context
job_profile_text = ""
if job_profile_name and job_profile_skills:
job_skills_list = []
for skill in job_profile_skills:
skill_name = skill.get('skill_name', 'Unknown')
skill_level = get_skill_level_name(int(skill.get('skill_type', 1)))
job_skills_list.append(f"{skill_name} ({skill_level})")
job_profile_text = f"Current job profile '{job_profile_name}' requires: {', '.join(job_skills_list)}"
designation_text = f"Current designation: {designation}" if designation else ""
prompt = f"""
ADVANCED CAREER SKILL ANALYSIS REQUEST
User Query: "{user_query}"
{designation_text}
{job_profile_text}
{user_skills_text}
INSTRUCTIONS:
Analyze the user's query intelligently and provide skill analysis recommendations in EXACT JSON format below.
CRITICAL ANALYSIS RULES:
1. If user asks to transition to a NEW ROLE (e.g., "become sales manager"), focus ONLY on skills needed for that target role
2. Include user's existing skills ONLY if they are relevant to the target role from the query
3. If user asks general skill analysis, include both current role skills and growth opportunities
4. Always prioritize skills mentioned or implied in the user's specific query
5. Generate 8-12 relevant skills maximum for the target role/query
6. Skill levels: 1=Beginner, 2=Intermediate, 3=Advanced
EXACT OUTPUT FORMAT (JSON only, no explanations):
{{
"target_role_extracted": "extracted role from query or null",
"analysis_type": "role_transition|skill_enhancement|general_analysis",
"recommended_skills": [
{{
"skill_name": "Skill Name",
"current_user_level": 0-3,
"target_level": 1-3,
"priority": "critical|high|medium",
"relevance_reason": "brief reason why this skill is needed"
}}
],
"user_relevant_skills": [
{{
"skill_name": "User Skill Name",
"current_level": 1-3,
"target_enhancement_level": 1-3,
"relevance_to_query": "how this existing skill relates to the query"
}}
]
}}
"""
return prompt
def parse_llm_skill_analysis_response(llm_response):
"""Parse LLM response and extract structured skill analysis data"""
try:
# Clean the response - remove any text before/after JSON
llm_response = llm_response.strip()
# Find JSON content
start_idx = llm_response.find('{')
end_idx = llm_response.rfind('}') + 1
if start_idx == -1 or end_idx == 0:
return None
json_content = llm_response[start_idx:end_idx]
parsed_data = json.loads(json_content)
return parsed_data
except (json.JSONDecodeError, Exception) as e:
print(f"Error parsing LLM response: {e}")
return None
def generate_advanced_skill_analysis_with_llm(user_skills, job_profile_skills=None, job_profile_name=None, designation=None, user_query=""):
"""Generate advanced skill analysis using LLM intelligence"""
# Build LLM prompt for intelligent analysis
llm_prompt = build_llm_skill_analysis_prompt(
user_query=user_query,
user_skills=user_skills,
job_profile_skills=job_profile_skills,
job_profile_name=job_profile_name,
designation=designation
)
# Query LLM for intelligent skill analysis
llm_response = query_ollama(llm_prompt)
parsed_analysis = parse_llm_skill_analysis_response(llm_response)
if not parsed_analysis:
# Fallback to existing logic if LLM fails
return analyze_skill_gaps_with_target_role(user_skills, job_profile_skills, job_profile_name, user_query)
# Process LLM recommendations into chart format
all_skills = {}
# Process LLM recommended skills (highest priority)
for skill_rec in parsed_analysis.get('recommended_skills', []):
skill_name = skill_rec.get('skill_name', '')
if skill_name:
skill_key = skill_name.lower()
all_skills[skill_key] = {
'name': skill_name,
'current_level': int(skill_rec.get('current_user_level', 0)),
'target_level': int(skill_rec.get('target_level', 3)),
'is_user_skill': skill_rec.get('current_user_level', 0) > 0,
'is_llm_recommended': True,
'priority': skill_rec.get('priority', 'medium'),
'relevance_reason': skill_rec.get('relevance_reason', ''),
'gap': 0 # Will be calculated
}
# Process user's relevant existing skills
for user_skill_rec in parsed_analysis.get('user_relevant_skills', []):
skill_name = user_skill_rec.get('skill_name', '')
if skill_name:
skill_key = skill_name.lower()
if skill_key in all_skills:
# Update existing skill entry
all_skills[skill_key]['current_level'] = int(user_skill_rec.get('current_level', 1))
all_skills[skill_key]['target_level'] = max(
all_skills[skill_key]['target_level'],
int(user_skill_rec.get('target_enhancement_level', 2))
)
all_skills[skill_key]['is_user_skill'] = True
all_skills[skill_key]['relevance_reason'] = user_skill_rec.get('relevance_to_query', '')
else:
# Add new relevant user skill
all_skills[skill_key] = {
'name': skill_name,
'current_level': int(user_skill_rec.get('current_level', 1)),
'target_level': int(user_skill_rec.get('target_enhancement_level', 2)),
'is_user_skill': True,
'is_llm_recommended': False,
'priority': 'medium',
'relevance_reason': user_skill_rec.get('relevance_to_query', ''),
'gap': 0 # Will be calculated
}
# Calculate gaps and prepare final analysis
skills_analysis = []
for skill_data in all_skills.values():
skill_data['gap'] = max(skill_data['target_level'] - skill_data['current_level'], 0)
# Assign priority scores for sorting
if skill_data['priority'] == 'critical':
skill_data['priority_score'] = 1
elif skill_data['priority'] == 'high':
skill_data['priority_score'] = 2
else:
skill_data['priority_score'] = 3
# Boost priority for skills with large gaps
if skill_data['gap'] >= 2:
skill_data['priority_score'] -= 0.5
skills_analysis.append(skill_data)
# Sort by priority and gap size
skills_analysis.sort(key=lambda x: (x['priority_score'], -x['gap'], x['name']))
# Select top skills for visualization
top_skills = skills_analysis[:12]
# Prepare chart data
skill_names = [skill["name"] for skill in top_skills]
current_levels = [skill["current_level"] for skill in top_skills]
target_levels = [skill["target_level"] for skill in top_skills]
gap_values = [skill["gap"] for skill in top_skills]
# Create color coding based on priority and LLM recommendations
bar_colors = []
for skill in top_skills:
if skill['current_level'] == 0: # New skills needed
if skill['priority'] == 'critical':
bar_colors.append("rgba(220, 53, 69, 0.9)") # Bright red for critical new skills
else:
bar_colors.append("rgba(255, 107, 107, 0.8)") # Light red for new skills
elif skill['is_llm_recommended']:
if skill['priority'] == 'critical':
bar_colors.append("rgba(220, 53, 69, 0.7)") # Red for critical skills
elif skill['priority'] == 'high':
bar_colors.append("rgba(255, 193, 7, 0.8)") # Yellow for high priority
else:
bar_colors.append("rgba(13, 202, 240, 0.7)") # Blue for medium priority
else:
bar_colors.append("rgba(108, 117, 125, 0.6)") # Gray for existing skills
# Get target role from LLM analysis
target_role = parsed_analysis.get('target_role_extracted') or extract_target_role_from_query(user_query)
analysis_type = parsed_analysis.get('analysis_type', 'general_analysis')
chart_data = {
"radar": {
"labels": skill_names,
"datasets": [
{
"label": "Current Skills",
"data": current_levels,
"backgroundColor": "rgba(54, 162, 235, 0.2)",
"borderColor": "rgba(54, 162, 235, 1)",
"pointBackgroundColor": "rgba(54, 162, 235, 1)"
},
{
"label": f"Target Skills{' (' + target_role + ')' if target_role else ''}",
"data": target_levels,
"backgroundColor": "rgba(255, 99, 132, 0.2)",
"borderColor": "rgba(255, 99, 132, 1)",
"pointBackgroundColor": "rgba(255, 99, 132, 1)"
}
]
},
"bar": {
"labels": skill_names,
"datasets": [
{
"label": "Skill Gap",
"data": gap_values,
"backgroundColor": bar_colors
}
]
}
}
# Generate categorized skill lists
critical_gaps = [skill for skill in top_skills if skill['priority'] == 'critical' and skill['gap'] > 0]
high_priority_gaps = [skill for skill in top_skills if skill['priority'] == 'high' and skill['gap'] > 0]
medium_gaps = [skill for skill in top_skills if skill['priority'] == 'medium' and skill['gap'] > 0]
# Combine for other_gaps based on analysis type
if analysis_type == 'role_transition':
other_gaps = critical_gaps + high_priority_gaps + medium_gaps
job_profile_gaps = [] # Focus on target role, not current job
else:
# For general analysis, separate current job vs growth skills
job_profile_gaps = [skill for skill in top_skills if skill.get('is_job_profile_skill', False) and skill['gap'] > 0]
other_gaps = [skill for skill in top_skills if not skill.get('is_job_profile_skill', False) and skill['gap'] > 0]
strengths = [skill for skill in top_skills if skill['gap'] == 0 and skill['current_level'] >= 2]
# Generate summary based on LLM analysis
summary_parts = []
if critical_gaps:
critical_names = [skill['name'] for skill in critical_gaps[:3]]
summary_parts.append(f"CRITICAL SKILLS NEEDED: {', '.join(critical_names)}")
if high_priority_gaps:
high_names = [skill['name'] for skill in high_priority_gaps[:3]]
summary_parts.append(f"HIGH PRIORITY: {', '.join(high_names)}")
if target_role and analysis_type == 'role_transition':
new_skills_needed = len([s for s in critical_gaps + high_priority_gaps if s['current_level'] == 0])
if new_skills_needed > 0:
summary_parts.append(f"For {target_role} transition: {new_skills_needed} new skills to develop")
if strengths:
summary_parts.append(f"Strong foundation: {len(strengths)} advanced skills")
summary = ". ".join(summary_parts) if summary_parts else "Skill analysis completed based on your query requirements."
return {
"chart_data": chart_data,
"summary": summary,
"job_profile_gaps": job_profile_gaps,
"other_gaps": other_gaps,
"strengths": strengths,
"total_skills_analyzed": len(skills_analysis),
"llm_analysis_type": analysis_type,
"target_role_detected": target_role
}
def generate_scenario_appropriate_skill_analysis(user_context, user_query=""):
"""Updated function to use advanced LLM-based skill analysis"""
scenario = user_context['scenario']
current_job_profile_skills = user_context['current_job_profile_skills']
user_skills = user_context['user_skills']
current_job_profile_name = user_context['current_job_profile_name']
designation = user_context.get('designation', '')
# Always use advanced LLM analysis for better query understanding
if scenario == "scenario_1":
# Even with no data, LLM can analyze query and suggest skills
return generate_advanced_skill_analysis_with_llm(
user_skills=[],
job_profile_skills=[],
job_profile_name=None,
designation=designation,
user_query=user_query
)
elif scenario == "scenario_2":
# Only designation available
return generate_advanced_skill_analysis_with_llm(
user_skills=[],
job_profile_skills=[],
job_profile_name=None,
designation=designation,
user_query=user_query
)
else:
# Full data available - use comprehensive LLM analysis
return generate_advanced_skill_analysis_with_llm(
user_skills=user_skills,
job_profile_skills=current_job_profile_skills,
job_profile_name=current_job_profile_name,
designation=designation,
user_query=user_query
)
def extract_target_role_from_query(user_query):
"""Extract target role from user query with better parsing - UPDATED"""
if not user_query:
return None
query_lower = user_query.lower().strip()
# Enhanced role extraction patterns with more variations
role_patterns = [
r'(?:want to become|become a|become an|transition to|move to|switch to)\s+(?:a\s+|an\s+)?([a-zA-Z\s]+?)(?:\?|$|\.|,)',
r'(?:career in|job as|role as|position as|work as)\s+(?:a\s+|an\s+)?([a-zA-Z\s]+?)(?:\?|$|\.|,)',
r'(?:interested in|pursuing|targeting)\s+(?:a\s+|an\s+)?([a-zA-Z\s]+?)\s+(?:role|position|career|job)(?:\?|$|\.|,)',
r'(?:how to become|steps to become)\s+(?:a\s+|an\s+)?([a-zA-Z\s]+?)(?:\?|$|\.|,)',
r'(?:path to|roadmap to)\s+(?:a\s+|an\s+)?([a-zA-Z\s]+?)(?:\?|$|\.|,)',
]
for pattern in role_patterns:
match = re.search(pattern, query_lower)
if match:
extracted_role = match.group(1).strip()
# Clean and validate the extracted role
# Remove common filler words
filler_words = ['very', 'good', 'successful', 'better', 'great', 'excellent']
words = [word for word in extracted_role.split() if word not in filler_words]
extracted_role = ' '.join(words)
if len(extracted_role) > 2 and len(extracted_role.split()) <= 4:
return extracted_role.title()
return None
def build_scenario_appropriate_reasoning_with_target_role(user_context, course_details, course_name, user_query):
"""Build reasoning that focuses on target role from user query - UPDATED"""
target_role = extract_target_role_from_query(user_query)
designation = user_context['designation']
current_job_profile_name = user_context['current_job_profile_name']
if target_role:
# Check if course skills align with target role
course_skills = course_details.get('skills', [])
target_role_skills = get_target_role_required_skills(target_role)
# Check alignment
alignment_found = False
if course_skills and target_role_skills:
course_skill_names = [skill.get('skill_name', '').lower() for skill in course_skills]
target_skill_names = [skill['skill_name'].lower() for skill in target_role_skills]
for course_skill in course_skill_names:
for target_skill in target_skill_names:
if course_skill in target_skill or target_skill in course_skill:
alignment_found = True
break
if alignment_found:
break
if alignment_found:
return f"[CRITICAL FOR {target_role.upper()}] {course_name} develops essential skills required for {target_role} role and supports your career transition from {designation} background."
else:
return f"[CAREER TRANSITION] {course_name} builds foundational skills for {target_role} role and supports your career transition from {designation} background."
else:
return f"[PROFESSIONAL DEVELOPMENT] {course_name} enhances your capabilities as {designation} and aligns with your career development objectives."
# ========== Key Modifications for Career Assistant ==========
def determine_data_scenario(designation, job_profile_name, job_profile_skills, user_skills):
"""
Determine which scenario we're dealing with based on available data
"""
has_designation = bool(designation)
has_job_profile = bool(job_profile_name and job_profile_skills)
has_user_skills = bool(user_skills)
if not has_designation and not has_job_profile and not has_user_skills:
return "scenario_1" # Nothing available
elif has_designation and not has_job_profile and not has_user_skills:
return "scenario_2" # Only designation
elif has_designation and has_job_profile and not has_user_skills:
return "scenario_3" # Designation and job profile
elif has_designation and has_job_profile and has_user_skills:
return "scenario_4" # Everything available
else:
return "scenario_mixed" # Mixed scenarios
# ========== Modified Functions for Career Assistant Fix ==========
def handle_missing_data_scenarios(user_data):
user_info = list(user_data.values())[0] if user_data else {}
user_name = user_info.get('user_name', 'User')
designation = user_info.get('designation', '').strip()
job_profile = user_info.get('jobProfile', {})
job_profile_name = job_profile.get('job_profile_name', '').strip() if job_profile else ''
job_profile_skills = job_profile.get('job_profile_skills', []) if job_profile else []
user_skills = user_info.get('skills', [])
scenario = determine_data_scenario(designation, job_profile_name, job_profile_skills, user_skills)
return {
'user_name': user_name,
'designation': designation or 'Professional',
'current_job_profile_name': job_profile_name,
'current_job_profile_skills': job_profile_skills,
'user_skills': user_skills,
'scenario': scenario,
'completed_courses': user_info.get('completedCourses', []),
'assigned_courses': user_info.get('assignedCourses', [])
}
# ========== Updated Main Function Modifications ==========
def main(payload_file_path=None):
try:
if payload_file_path is None and len(sys.argv) > 1:
payload_file_path = sys.argv[1]
if payload_file_path is None:
raise ValueError("No payload file path provided")
with open(payload_file_path, 'r') as f:
payload = json.load(f)
user_data = payload.get("user_data", {})
client_all_courses_data = payload.get("client_all_courses_data", {})
client_id = payload.get("client_id", 3)
user_query = payload.get("user_query", "i want career guidance")
user_context = handle_missing_data_scenarios(user_data)
courses_data = client_all_courses_data.get(str(client_id), [])
course_mapping = {course['courseId']: course for course in courses_data}
# Analyze job profile alignment
job_profile_analysis = analyze_job_profile_alignment(
user_context['user_skills'],
user_context['current_job_profile_skills'],
user_context['current_job_profile_name']
)
# Build enhanced prompt with better LLM guidance
prompt = build_enhanced_prompt_for_scenario(user_context, user_query, course_mapping)
answer = query_ollama(prompt)
answer = remove_course_ids_from_text(answer, course_mapping)
# Extract course recommendations with enhanced target role analysis
recommended_courses = extract_course_recommendations(
answer,
course_mapping,
user_context['user_skills'],
user_context['current_job_profile_skills'],
user_context['designation'],
user_query
)
# Add enhanced reasoning for each course
for course in recommended_courses:
course_id = course["courseId"]
course_details = course_mapping.get(course_id, {})
reason = build_scenario_appropriate_reasoning_with_target_role(
user_context,
course_details,
course["courseName"],
user_query
)
course["reason"] = reason
course["scenario_relevance"] = user_context['scenario']
# CRITICAL UPDATE: Use advanced LLM-based skill analysis
skill_gap_analysis = generate_scenario_appropriate_skill_analysis(user_context, user_query)
# Ensure no empty job_profile_gaps or other_gaps
if not skill_gap_analysis.get('job_profile_gaps'):
skill_gap_analysis['job_profile_gaps'] = []
if not skill_gap_analysis.get('other_gaps'):
skill_gap_analysis['other_gaps'] = []
# Online course recommendations with target role awareness
online_course_recommendations = fetch_courses_with_agent_fixed(
user_skills=user_context['user_skills'],
current_job_profile_data={
'job_profile_name': user_context['current_job_profile_name'],
'job_profile_skills': user_context['current_job_profile_skills']
},
user_query=user_query,
platform_name="youtube"
)
# Enhanced answer with scenario-appropriate response
enhanced_answer = build_scenario_appropriate_response(user_context, answer, user_query)
# Detect target role for context
target_role_detected = extract_target_role_from_query(user_query)
final_output = {
"answer": enhanced_answer,
"recommended_courses": recommended_courses[:9],
"online_course_recommendations": online_course_recommendations,
"scenario": user_context['scenario'],
"skill_gap_chart": skill_gap_analysis,
"job_profile_analysis": job_profile_analysis,
"career_context": {
"user_name": user_context['user_name'],
"designation": user_context['designation'],
"current_job_profile_name": user_context['current_job_profile_name'],
"data_completeness": user_context['scenario'],
"target_role_from_query": target_role_detected,
"llm_analysis_type": skill_gap_analysis.get('llm_analysis_type', 'general_analysis')
}
}
print(json.dumps(final_output, indent=2))
except Exception as e:
import traceback
print(json.dumps({"error": str(e), "traceback": traceback.format_exc()}))
# ========== Modified Functions for Career Assistant Fix ==========
def build_scenario_appropriate_reasoning_fixed(user_context, course_details, course_name, user_query):
"""
Build appropriate reasoning based on available data scenario with correct context
"""
scenario = user_context['scenario']
designation = user_context['designation']
current_job_profile_name = user_context['current_job_profile_name']
if scenario == "scenario_1":
return f"[CAREER DEVELOPMENT] {course_name} supports your career objectives mentioned in your query about '{user_query}' and provides essential skills for professional growth."
elif scenario == "scenario_2":
return f"[PROFESSIONAL ADVANCEMENT] {course_name} enhances your {designation} capabilities and aligns with your career goals expressed in your query."
elif scenario == "scenario_3":
return f"[ROLE ENHANCEMENT] {course_name} builds upon your {designation} experience in {current_job_profile_name} and supports your career aspirations."
else:
return f"[STRATEGIC DEVELOPMENT] {course_name} addresses your career development needs as a {designation} while supporting the objectives in your query."
def build_enhanced_prompt_for_scenario(user_context, user_query, course_mapping):
"""Enhanced prompt building with better LLM guidance for skill analysis"""
scenario = user_context['scenario']
user_name = user_context['user_name']
designation = user_context['designation']
current_job_profile_name = user_context['current_job_profile_name']
user_skills = user_context['user_skills']
# Extract target role for context
target_role = extract_target_role_from_query(user_query)
# Build skills context
skills_context = ""
if user_skills:
skill_list = []
for skill in user_skills:
skill_name = skill.get('skill_name', 'Unknown')
skill_level = get_skill_level_name(int(skill.get('skill_type', 1)))
skill_list.append(f"{skill_name} ({skill_level})")
skills_context = f"User's current skills: {', '.join(skill_list)}"
# Build job profile context
job_context = ""
if current_job_profile_name:
job_context = f"Current job profile: {current_job_profile_name}"
# Build course context
courses_info = []
for course_id, course_data in list(course_mapping.items())[:20]: # Limit to prevent prompt overflow
course_name = course_data.get('name', 'Unknown')
course_skills = course_data.get('skills', [])
skill_names = [skill.get('skill_name', '') for skill in course_skills]
courses_info.append(f"Course {course_id}: {course_name} (Skills: {', '.join(skill_names) if skill_names else 'General'})")
courses_context = "Available courses:\n" + "\n".join(courses_info)
# Enhanced prompt based on scenario and target role
base_prompt = f"""
ADVANCED CAREER GUIDANCE REQUEST
User: {user_name}
Current Role: {designation}
{job_context}
{skills_context}
User Query: "{user_query}"
"""
if target_role:
base_prompt += f"""
DETECTED TARGET ROLE: {target_role}
CRITICAL INSTRUCTIONS FOR TARGET ROLE TRANSITION:
1. Focus recommendations on skills and courses needed for {target_role}
2. Prioritize courses that develop {target_role}-specific competencies
3. Consider the user's transition path from {designation} to {target_role}
4. Recommend courses that fill critical skill gaps for {target_role}
"""
scenario_specific_guidance = {
"scenario_1": "User has minimal career data. Provide foundational career guidance and recommend versatile courses that build core professional skills.",
"scenario_2": f"User has designation ({designation}) but limited other data. Focus on career advancement opportunities and skill development for their current role.",
"scenario_3": f"User has designation and job profile but no skills data. Recommend courses that develop job-specific skills and career progression.",
"scenario_4": "User has complete profile. Provide personalized recommendations based on skill gaps and career objectives."
}
base_prompt += f"""
SCENARIO GUIDANCE: {scenario_specific_guidance.get(scenario, 'Provide comprehensive career guidance.')}
{courses_context}
RESPONSE REQUIREMENTS:
1. Provide career guidance that directly addresses the user query
2. Recommend specific courses (mention course IDs) that align with the query
3. Focus on practical, actionable advice
4. Consider the user's current situation and target goals
5. Prioritize courses that have the highest impact for the user's specific query
Provide comprehensive career guidance response:
"""
return base_prompt
def get_query_focused_instructions(user_query, target_role_context, scenario):
"""
Get instructions that focus on the specific user query without repeating it
"""
instructions = f"""
CRITICAL RESPONSE REQUIREMENTS:
1. DO NOT repeat the user's question in your response
2. Focus entirely on providing actionable career guidance for their specific request
3. Address their career aspiration directly and professionally
4. Provide concrete steps and recommendations
5. Recommend 5-7 most relevant courses that support their career goal
RESPONSE STRUCTURE:
- Start directly with career guidance (no question repetition)
- Provide strategic analysis of their career transition/advancement
- Give specific recommendations for skill development
- Suggest actionable next steps
- Recommend courses that directly support their career objective
TARGET ROLE FOCUS:"""
if target_role_context:
instructions += f"""
The user wants to transition to: {target_role_context}
- Analyze requirements for {target_role_context} role
- Identify skill gaps and development needs
- Provide industry-specific guidance
- Recommend courses relevant to {target_role_context}
"""
else:
instructions += f"""
Focus on their career development needs mentioned in the query
- Understand their career objectives
- Provide relevant professional guidance
- Recommend appropriate skill development
"""
return instructions
def build_scenario_appropriate_response(user_context, base_answer, user_query):
"""Enhanced response building with better query-specific guidance"""
scenario = user_context['scenario']
user_name = user_context['user_name']
designation = user_context['designation']
target_role = extract_target_role_from_query(user_query)
# Build scenario-appropriate introduction
intro = f"Hello {user_name}! "
if target_role:
if scenario in ["scenario_1", "scenario_2"]:
intro += f"I understand you want to transition to {target_role}. Here's a comprehensive roadmap to help you achieve this career goal:"
else:
intro += f"Based on your background as {designation}, here's your personalized transition plan to become {target_role}:"
else:
scenario_intros = {
"scenario_1": "I'm here to help you with career guidance. Let me provide you with foundational recommendations:",
"scenario_2": f"As {designation}, here are personalized career development recommendations:",
"scenario_3": f"Based on your role as {designation}, here's targeted career guidance:",
"scenario_4": f"With your experience as {designation}, here are tailored recommendations:"
}
intro += scenario_intros.get(scenario, "Here's your personalized career guidance:")
# Enhance the base answer with scenario context
enhanced_answer = f"{intro}\n\n{base_answer}"
# Add scenario-specific closing
if target_role:
enhanced_answer += f"\n\nRemember, transitioning to {target_role} is a journey. Focus on building the critical skills first, then gradually develop the supporting competencies. Your success depends on consistent learning and practical application of these skills."
else:
enhanced_answer += "\n\nFocus on continuous learning and skill development. Each course you complete brings you closer to your career goals."
return enhanced_answer
def clean_answer_from_query_repetition(answer, user_query):
"""
Remove any repetition of user query from the AI response
"""
# Remove direct query repetitions
query_variations = [
user_query,
user_query.lower(),
user_query.capitalize(),
f'"{user_query}"',
f"'{user_query}'"
]
for variation in query_variations:
answer = answer.replace(variation, "")
# Remove common query introduction patterns
patterns_to_remove = [
r'.*(?:you asked|you mentioned|you want to|you said).*?\n',
r'.*(?:your question|your query|your request).*?\n',
r'.*(?:regarding|about|concerning).*?\n',
r'I understand that you.*?\n',
r'Based on your question.*?\n',
r'You mentioned that.*?\n'
]
for pattern in patterns_to_remove:
answer = re.sub(pattern, '', answer, flags=re.IGNORECASE)
# Clean up formatting
answer = re.sub(r'\n\s*\n', '\n\n', answer) # Remove extra blank lines
answer = answer.strip()
return answer
def enhanced_match_skills_for_target_role(user_skills, course_skills, job_profile_skills=None, designation=None, target_role=None):
"""Updated enhanced skill matching with LLM-aware target role analysis"""
# Use LLM to analyze if course is relevant to target role
if target_role and course_skills:
llm_relevance_prompt = f"""
COURSE RELEVANCE ANALYSIS
Target Role: {target_role}
Course Skills: {[skill.get('skill_name', '') for skill in course_skills]}
User Query Context: Transition to {target_role}
Rate relevance (0-100) of this course for the target role. Consider:
1. Direct skill alignment with target role requirements
2. Career transition value
3. Skill gap filling for the target role
Respond with just a number (0-100):
"""
llm_relevance_response = query_ollama(llm_relevance_prompt)
try:
llm_relevance_score = float(re.search(r'\d+', llm_relevance_response).group())
llm_relevance_score = min(max(llm_relevance_score, 0), 100)
except:
llm_relevance_score = 0
# Use LLM score as primary factor for target role scenarios
base_score = match_skills(user_skills, course_skills) * 0.2
llm_score = llm_relevance_score * 0.8
return min(base_score + llm_score, 100)
# Fallback to original logic
base_score = match_skills(user_skills, course_skills)
# Job profile alignment bonus
job_profile_bonus = 0
if job_profile_skills:
job_profile_bonus = calculate_job_profile_skill_priority_score(
user_skills, job_profile_skills, course_skills
)
# Career progression bonus
designation_bonus = calculate_designation_progression_bonus(
designation, course_skills, user_skills
)
# Skill gap filling bonus
skill_gap_bonus = calculate_skill_gap_bonus(user_skills, course_skills)
total_score = (
base_score * 0.25 +
job_profile_bonus * 0.45 +
designation_bonus * 0.20 +
skill_gap_bonus * 0.10
)
return min(total_score, 100)
def calculate_target_role_alignment_score(target_role, course_skills, user_skills):
"""
Calculate alignment score based on target role requirements
"""
if not target_role or not course_skills:
return 0
target_role_lower = target_role.lower()
# Define skill requirements for common target roles
role_skill_requirements = {
'sales manager': ['leadership', 'communication', 'sales', 'negotiation', 'team management', 'customer relationship'],
'product manager': ['product management', 'strategy', 'analytics', 'communication', 'leadership', 'market research'],
'data scientist': ['machine learning', 'python', 'statistics', 'data analysis', 'sql', 'deep learning'],
'software engineer': ['programming', 'software development', 'algorithms', 'system design', 'debugging'],
'project manager': ['project management', 'leadership', 'planning', 'risk management', 'communication'],
'marketing manager': ['marketing', 'digital marketing', 'brand management', 'analytics', 'communication'],
'business analyst': ['business analysis', 'data analysis', 'requirements gathering', 'sql', 'process improvement'],
'tech lead': ['technical leadership', 'architecture', 'mentoring', 'programming', 'system design'],
'consultant': ['consulting', 'problem solving', 'communication', 'analytical thinking', 'presentation skills']
}
# Find matching role requirements
required_skills = []
for role_key, skills in role_skill_requirements.items():
if role_key in target_role_lower:
required_skills = skills
break
if not required_skills:
# Generic management/leadership skills for unknown roles
required_skills = ['leadership', 'communication', 'management', 'problem solving', 'strategic thinking']
# Calculate alignment score
course_skill_names = [skill.get('skill_name', '').lower() for skill in course_skills]
alignment_score = 0
for required_skill in required_skills:
for course_skill_name in course_skill_names:
if required_skill in course_skill_name or any(word in course_skill_name for word in required_skill.split()):
alignment_score += 15 # High score for target role relevant skills
break
return min(alignment_score, 75) # Cap the bonus
def calculate_target_role_priority_score(target_role, course_skills, user_skills):
"""
Calculate priority score specifically for target role requirements
CRITICAL priority for missing target role skills
"""
if not target_role or not course_skills:
return 0
target_role_skills = get_target_role_required_skills(target_role)
if not target_role_skills:
return 0
# Create user skill maps
user_skill_name_map = {
skill.get('skill_name', '').lower(): {
'current_level': int(skill.get('skill_type', 1))
} for skill in (user_skills or []) if skill.get('skill_name')
}
course_skill_name_map = {
skill.get('skill_name', '').lower(): {
'course_level': int(skill.get('skill_type', 1))
} for skill in course_skills if skill.get('skill_name')
}
priority_score = 0
for target_skill_info in target_role_skills:
skill_name = target_skill_info['skill_name'].lower()
required_level = target_skill_info['skill_type']
skill_priority = target_skill_info['priority']
# Check if course teaches this target role skill
if skill_name in course_skill_name_map or any(
skill_name in course_skill_name.lower() or course_skill_name.lower() in skill_name
for course_skill_name in course_skill_name_map.keys()
):
# Get user's current level
user_current_level = user_skill_name_map.get(skill_name, {}).get('current_level', 0)
# Priority scoring based on skill importance and user's current level
if skill_priority == 'critical':
if user_current_level == 0:
priority_score += 30 # CRITICAL: Missing critical skill
elif user_current_level < required_level:
priority_score += 25 # HIGH: Below required level for critical skill
else:
priority_score += 10 # MEDIUM: Enhancement of critical skill
elif skill_priority == 'high':
if user_current_level == 0:
priority_score += 20 # HIGH: Missing high-priority skill
elif user_current_level < required_level:
priority_score += 15 # MEDIUM-HIGH: Below required level
else:
priority_score += 8 # MEDIUM: Enhancement
else: # medium priority
if user_current_level == 0:
priority_score += 12 # MEDIUM: Missing medium-priority skill
elif user_current_level < required_level:
priority_score += 10 # MEDIUM: Below required level
else:
priority_score += 5 # LOW: Enhancement
return min(priority_score, 100)
def extract_course_recommendations(answer, course_mapping, user_skills, job_profile_skills=None, designation=None, user_query=""):
"""
Enhanced course recommendation extraction with target role consideration
"""
target_role = extract_target_role_from_query(user_query)
recommended_courses = []
courses_found = set()
# Direct course name matching
for course_id, course_details in course_mapping.items():
course_name = course_details['name']
if course_name.lower() in answer.lower() and course_id not in courses_found:
match_score = enhanced_match_skills_for_target_role(
user_skills,
course_details.get('skills', []),
job_profile_skills,
designation,
target_role
)
recommended_courses.append({
"courseId": course_id,
"courseName": course_name,
"matchScore": f"{match_score:.1f}%"
})
courses_found.add(course_id)
# If we have fewer than 9 courses, add more based on target role relevance
if len(recommended_courses) < 9:
additional_courses_needed = 9 - len(recommended_courses)
# Score all remaining courses for target role relevance
remaining_courses = []
for course_id, course_details in course_mapping.items():
if course_id not in courses_found:
match_score = enhanced_match_skills_for_target_role(
user_skills,
course_details.get('skills', []),
job_profile_skills,
designation,
target_role
)
remaining_courses.append({
"courseId": course_id,
"courseName": course_details['name'],
"matchScore": f"{match_score:.1f}%",
"score_value": match_score
})
# Sort by relevance to target role and add top courses
remaining_courses.sort(key=lambda x: x['score_value'], reverse=True)
for course in remaining_courses[:additional_courses_needed]:
del course['score_value'] # Remove internal scoring field
recommended_courses.append(course)
# Sort final list by match score
recommended_courses.sort(key=lambda x: float(x['matchScore'].replace('%', '')), reverse=True)
return recommended_courses[:9]
def fetch_courses_with_agent_fixed(user_skills, current_job_profile_data=None, user_query="", platform_name="youtube"):
"""
Fetch courses using CareerAgent with target role focus
"""
try:
# Initialize CareerAgent with user skills, query, and job profile data
agent = CareerAgent(
user_skills=user_skills,
user_query=user_query,
platform_name=platform_name,
job_profile_data=current_job_profile_data
)
# Fetch and recommend courses
courses = agent.fetch_courses()
recommended_courses = agent.recommend_courses(courses)
return recommended_courses
except Exception as e:
print(f"Error in fetch_courses_with_agent: {str(e)}")
return []
# ------------------------------------------------------------------------------------------------------------------------------------------------------------
if __name__ == "__main__":
default_payload = {
"client_id": 3,
"user_query": "i want to become sales manager ?",
"user_data": {
"4": {
"user_name": "Kunal Jain",
"managerId": "13962",
"designation": "Engineer",
"jobProfile": {
"job_profile_name": "data scientist",
"job_profile_skills": [
{ "skill_id": "20", "skill_type": "1", "skill_name": "Machine Learning" },
{ "skill_id": "15", "skill_type": "1", "skill_name": "Python" },
{ "skill_id": "17", "skill_type": "1", "skill_name": "Deep Learning" }
]
},
"assignedCourses": ["6", "5", "62", "98", "82", "291", "455", "433", "353", "580"],
"completedCourses": ["111", "114"],
"skills": [
{ "skill_id": "6", "skill_type": "3", "skill_name": "Leadership" },
{ "skill_id": "7", "skill_type": "1", "skill_name": "Management" },
{ "skill_id": "9", "skill_type": "2", "skill_name": "Communication" },
{ "skill_id": "242", "skill_type": "3", "skill_name": "Project Management" },
{ "skill_id": "26", "skill_type": "2", "skill_name": "Teamwork" },
{ "skill_id": "8", "skill_type": "1", "skill_name": "Problem Solving" },
{ "skill_id": "5", "skill_type": "2", "skill_name": "Time Management" },
{ "skill_id": "190", "skill_type": "2", "skill_name": "Analytical Skills" },
{ "skill_id": "70", "skill_type": "1", "skill_name": "Creativity" },
{ "skill_id": "224", "skill_type": "1", "skill_name": "Adaptability" },
{ "skill_id": "252", "skill_type": "1", "skill_name": "Innovation" },
{ "skill_id": "234", "skill_type": "2", "skill_name": "Strategic Thinking" },
{ "skill_id": "241", "skill_type": "1", "skill_name": "Decision Making" }
]
}
},
"client_all_courses_data": {
"3": [
{
"courseId": "43",
"name": "Python Programming for Complete Beginners",
"short_description": "Python Programming for Complete Beginners",
"description": "",
"skills": []
},
{
"courseId": "58",
"name": "Laravel Beginner",
"short_description": "Laravel Beginner description",
"description": "",
"skills": []
},
{
"courseId": "60",
"name": "New Course",
"short_description": "New Course short desc",
"description": "",
"skills": [
{
"skill_id": "7",
"skill_type": "3",
"skill_name": "Management"
},
{
"skill_id": "14",
"skill_type": "2",
"skill_name": "JavaScript"
}
]
},
{
"courseId": "71",
"name": "Demo Course",
"short_description": "Demo Course",
"description": "Demo Course
",
"skills": [
{
"skill_id": "15",
"skill_type": "3",
"skill_name": "Python"
}
]
},
{
"courseId": "72",
"name": "Course By Author",
"short_description": "Course By Author",
"description": "Course By Author
",
"skills": []
},
]
}
}
if len(sys.argv) > 1:
main(sys.argv[1])
else:
import tempfile
# Save default_payload to a temporary file
with tempfile.NamedTemporaryFile(mode='w+', suffix='.json', delete=False) as temp_file:
json.dump(default_payload, temp_file)
temp_file_path = temp_file.name
# Call main with the temporary file path
main(temp_file_path)