""" Skill analysis for TNA system using LLM Follows department.py pattern - simple, direct, effective """ import json import logging from typing import Dict, List from langchain_ollama import ChatOllama logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def analyze_skills_and_courses(user_query: str, admin_name: str, department_name: str, filtered_skills: List[Dict], total_users_in_department: int, all_skills_data: List[Dict], offline_courses: List[Dict]) -> Dict: """ Analyze user query and recommend skills and courses. Simple. Direct. No overthinking. """ try: # Initialize LLM (same as department.py) llm = ChatOllama(model="gemma3:12b", temperature=0.2) # Prepare skill data dept_skills_text = "" if department_name != "Null" and filtered_skills: # Extract skills from filtered_skills structure dept_skill_names = [] if filtered_skills and len(filtered_skills) > 0: if 'skills' in filtered_skills[0]: dept_skill_names = filtered_skills[0]['skills'] if dept_skill_names: dept_skills_text = f"\nDepartment ({department_name}) Skills ({len(dept_skill_names)} skills from {total_users_in_department} users):\n" dept_skills_text += "\n".join([f"- {skill}" for skill in dept_skill_names[:10]]) # Limit to 10 for brevity # Prepare org skills org_skills_text = "" if all_skills_data: org_skills_text = f"\nAll Organization Skills ({len(all_skills_data)} total):\n" org_skills_text += "\n".join([f"- {skill['skill_name']}" for skill in all_skills_data[:20]]) # Limit to 20 # Prepare courses courses_text = "" if offline_courses: courses_text = f"\nAvailable Offline Courses ({len(offline_courses)} total):\n" for course in offline_courses[:10]: # Limit to 10 skills_list = course.get('skills', []) skills_str = ", ".join(skills_list) if skills_list else "No skills listed" courses_text += f"- {course['course_name']} (ID: {course['course_id']}) - Skills: {skills_str}\n" # Create focused skill analysis prompt prompt = f"""You are analyzing skills for a Training Needs Analysis request. USER GOAL: {user_query} {dept_skills_text} {org_skills_text} TASK: Recommend 10-15 skills highly relevant to the user's goal. ANALYSIS STEPS: 1. Evaluate department skills for relevance to user goal (≥70% relevant) 2. Evaluate organization skills for relevance to user goal (≥70% relevant) 3. If total < 10 skills, suggest external skills relevant to the goal (≥70% relevant) 4. Stop when you reach 10-15 total skills OUTPUT FORMAT (JSON): {{ "department_skills": [ {{ "skill_name": "skill name", "relevance_score": 85, "reasoning": "why this skill helps achieve the goal" }} ], "organization_skills": [ {{ "skill_name": "skill name", "relevance_score": 75, "reasoning": "why this skill helps achieve the goal" }} ], "external_skills": [ {{ "skill_name": "skill name", "relevance_score": 78, "reasoning": "why this skill helps achieve the goal" }} ] }} IMPORTANT: - Only include skills with ≥70% relevance to the user goal - Aim for 10-15 total skills across all categories - Focus on skills that directly help achieve the stated goal Provide ONLY the JSON response, no additional text.""" # Get LLM response response = llm.invoke(prompt).content.strip() logger.info(f"LLM skill analysis response length: {len(response)}") # Parse JSON response try: if '{' in response and '}' in response: json_start = response.index('{') json_end = response.rindex('}') + 1 json_str = response[json_start:json_end] parsed_result = json.loads(json_str) # Process skills and create strategic presentation dept_skills = parsed_result.get('department_skills', []) org_skills = parsed_result.get('organization_skills', []) ext_skills = parsed_result.get('external_skills', []) # Filter skills by relevance score ≥70% dept_skills = [s for s in dept_skills if s.get('relevance_score', 0) >= 70] org_skills = [s for s in org_skills if s.get('relevance_score', 0) >= 70] ext_skills = [s for s in ext_skills if s.get('relevance_score', 0) >= 70] total_skills = len(dept_skills) + len(org_skills) + len(ext_skills) # Build strategic presentation with skills only presentation_parts = [] presentation_parts.append(f"Based on your goal '{user_query}', here are {total_skills} recommended skills:\n") if dept_skills and department_name != "Null": presentation_parts.append(f"DEPARTMENT SKILLS ({department_name}):") for skill in dept_skills: presentation_parts.append(f"• {skill['skill_name']} - {skill['reasoning']}") presentation_parts.append("") if org_skills: presentation_parts.append("ORGANIZATION SKILLS:") for skill in org_skills: presentation_parts.append(f"• {skill['skill_name']} - {skill['reasoning']}") presentation_parts.append("") if ext_skills: presentation_parts.append("EXTERNAL SKILLS:") for skill in ext_skills: presentation_parts.append(f"• {skill['skill_name']} - {skill['reasoning']}") presentation_parts.append("") presentation_parts.append(f"Total: {total_skills} skills recommended") # Combine all skills for recommended_skills field all_skills = [] for skill in dept_skills: all_skills.append({**skill, "source": "department"}) for skill in org_skills: all_skills.append({**skill, "source": "organization"}) for skill in ext_skills: all_skills.append({**skill, "source": "external"}) result = { "strategic_presentation": "\n".join(presentation_parts), "recommended_skills": all_skills, "courses_offline": [], # Empty for now "online_recommended_courses": [] # Empty for now } logger.info(f"Successfully parsed skill analysis with {total_skills} skills (dept: {len(dept_skills)}, org: {len(org_skills)}, ext: {len(ext_skills)})") return result except (json.JSONDecodeError, ValueError) as e: logger.warning(f"JSON parse failed: {e}") # Fallback response return { "strategic_presentation": f"Based on your goal '{user_query}', we recommend focusing on relevant skill development to achieve your objectives. Please try again for detailed skill recommendations.", "recommended_skills": [], "courses_offline": [], "online_recommended_courses": [] } except Exception as e: logger.error(f"Skill analysis failed: {e}") return { "strategic_presentation": "Unable to complete analysis at this time. Please try again.", "courses_offline": [], "online_recommended_courses": [], "error": str(e) } # Test function with mock LLM response def test_skill_analysis_logic(): """Test the data processing logic without LLM dependency""" # Test data similar to your PHP example test_filtered_skills = [ { 'user_id': 0, 'skills': ['Python', 'JavaScript', 'Communication'] } ] test_all_skills = [ {'skill_id': 1375, 'skill_name': 'communication skill'}, {'skill_id': 1376, 'skill_name': 'technical skills'}, {'skill_id': 121, 'skill_name': 'JavaScript'}, {'skill_id': 122, 'skill_name': 'Python'} ] test_courses = [ { 'course_id': '43', 'course_name': 'Python Programming for Complete Beginners', 'short_description': 'Python Programming for Complete Beginners', 'description': 'Learn Python from scratch', 'skills': ['python', 'programming'] } ] # Mock successful response mock_response = { "strategic_presentation": "Based on your goal to improve technical ability, we recommend focusing on Python and JavaScript skills from your department, plus additional technical skills from the organization. The Python course directly aligns with your needs.", "recommended_skills": [ { "skill_name": "Python", "source": "department", "relevance_score": 9.0, "reasoning": "Direct match for technical improvement goals" }, { "skill_name": "JavaScript", "source": "department", "relevance_score": 8.5, "reasoning": "Essential for web development skills" } ], "courses_offline": [ { "course_id": "43", "name": "Python Programming for Complete Beginners", "reason": "Perfect match for Python skill development", "percentage": 90.0 } ], "online_recommended_courses": [ { "title": "Advanced JavaScript Concepts", "platform": "Udemy", "instructor": "JavaScript Expert", "duration": "6 weeks", "skill_level": "Intermediate", "description": "Deep dive into advanced JavaScript concepts", "url": "https://udemy.com/advanced-js", "price": "Paid", "rating": "4.7/5", "skills_covered": ["JavaScript", "ES6", "Async Programming"], "relevance_score": 85, "reason": "Complements department JavaScript skills" } ] } print("=== SKILL ANALYSIS TEST ===") print(f"User Query: Want to improve my technical ability") print(f"Department: Developer") print(f"Department Skills: {test_filtered_skills[0]['skills']}") print(f"Total Org Skills: {len(test_all_skills)}") print(f"Available Courses: {len(test_courses)}") print("\n=== MOCK RESPONSE ===") print(json.dumps(mock_response, indent=2)) return mock_response if __name__ == "__main__": # Run logic test instead of LLM test when Ollama isn't available try: # Try actual LLM call first result = analyze_skills_and_courses( user_query="Want to improve my technical ability", admin_name="Ascent", department_name="Developer", filtered_skills=[{'user_id': 0, 'skills': ['Python', 'JavaScript']}], total_users_in_department=5, all_skills_data=[{'skill_id': 122, 'skill_name': 'Python'}], offline_courses=[{ 'course_id': '43', 'course_name': 'Python Programming for Complete Beginners', 'skills': ['python'] }] ) print("=== ACTUAL LLM RESPONSE ===") print(json.dumps(result, indent=2)) except Exception as e: print(f"LLM not available ({str(e)}), running logic test:") test_skill_analysis_logic()