from typing import Dict, List, Any, Optional import re from models.scenario import RoleplayScenario from models.assessment import SkillAssessment, SkillScore, ConversationTurn from services.groq_service import GroqService from services.ollama_service import OllamaService from utils.json_handler import JSONHandler class SkillAnalyzer: def __init__(self, groq_service=None, ollama_service=None): # Use provided services or create new ones (for backward compatibility) self.groq_service = groq_service or GroqService() self.ollama_service = ollama_service or OllamaService() self.json_handler = JSONHandler() def _get_ai_service(self, use_groq: bool): """Get the appropriate AI service based on groqRoleplay flag""" return self.groq_service if use_groq else self.ollama_service def _parse_score(self, score_val: Any) -> int: """Robustly parse a score value from AI response.""" if score_val is None: return 0 if isinstance(score_val, int): return score_val if isinstance(score_val, float): return int(score_val) if isinstance(score_val, str): # Clean string s = score_val.strip().lower() if not s: return 0 # Handle "X/10" or "X / 10" format if '/' in s: try: return int(float(s.split('/')[0].strip())) except (ValueError, IndexError): pass # Handle "X out of 10" format if ' out of ' in s: try: return int(float(s.lower().split(' out of ')[0].strip())) except (ValueError, IndexError): pass # Try to extract the first number found in the string try: match = re.search(r'(\d+\.?\d*)', s) if match: return int(float(match.group(1))) except (ValueError, TypeError): pass return 0 # ────────────────────────────────────────────────────────── # CONVERSATION DEPTH ANALYSIS — Core scoring guard # ────────────────────────────────────────────────────────── def _analyze_conversation_depth(self, conversation_turns: List[ConversationTurn], scenario: RoleplayScenario) -> Dict[str, Any]: """ Analyze the depth and quality of the learner's participation. Returns engagement metrics and a hard score cap. """ learner_turns = [t for t in conversation_turns if t.speaker == 'learner'] learner_turn_count = len(learner_turns) # Aggregate all learner text all_learner_text = " ".join(t.message.strip() for t in learner_turns) total_learner_words = len(all_learner_text.split()) if all_learner_text.strip() else 0 avg_words_per_turn = (total_learner_words / learner_turn_count) if learner_turn_count > 0 else 0 # Check if learner messages are just greetings / introductions is_only_greeting = self._is_greeting_only(all_learner_text) if learner_turn_count > 0 else True # Check objective coverage (rough keyword match) objective_coverage = self._estimate_objective_coverage(all_learner_text, scenario) # ── Classify engagement level & set hard score cap ── if learner_turn_count == 0: engagement_level = "none" max_allowed_score = 0 elif learner_turn_count == 1 and is_only_greeting: engagement_level = "minimal" max_allowed_score = 2 elif learner_turn_count == 1 and total_learner_words < 15 and objective_coverage < 0.1: engagement_level = "minimal" max_allowed_score = 2 elif learner_turn_count <= 2 and total_learner_words < 25 and objective_coverage < 0.2: engagement_level = "low" max_allowed_score = 3 elif learner_turn_count <= 3 and objective_coverage < 0.4: engagement_level = "moderate" max_allowed_score = 5 elif learner_turn_count >= 4 and objective_coverage >= 0.5: engagement_level = "excellent" max_allowed_score = 10 # no cap else: engagement_level = "good" max_allowed_score = 8 depth_analysis = { "learner_turn_count": learner_turn_count, "total_learner_words": total_learner_words, "avg_words_per_turn": round(avg_words_per_turn, 1), "is_only_greeting": is_only_greeting, "objective_coverage": round(objective_coverage, 2), "engagement_level": engagement_level, "max_allowed_score": max_allowed_score, } print(f"DEBUG: Conversation depth analysis: {depth_analysis}") return depth_analysis def _is_greeting_only(self, text: str) -> bool: """ Determine whether the learner's combined text is essentially just a greeting / self-introduction with no substantive content. """ cleaned = text.lower().strip() # Remove punctuation for matching cleaned = re.sub(r'[^\w\s]', '', cleaned) # Common greeting-only patterns greeting_patterns = [ r'^(hi|hello|hey|good\s+(morning|afternoon|evening)|greetings)\b', r'\b(my name is|i am|i\'m)\s+\w+', r'\b(here to help|here to assist|how can i help|how may i help)\b', r'\b(nice to meet|pleased to meet|welcome)\b', ] # Check if the text is short enough to be a greeting word_count = len(cleaned.split()) if word_count > 30: return False # Too long to be just a greeting # Score how many greeting patterns match greeting_matches = sum(1 for p in greeting_patterns if re.search(p, cleaned)) # If most of the content matches greeting patterns, it's a greeting # Also check for absence of substantive action words action_indicators = [ r'\b(check|verify|confirm|look into|investigate|resolve|arrange|schedule|offer|provide|ensure)\b', r'\b(room|reservation|booking|account|order|issue|problem|request|complaint)\b', r'\b(available|status|update|information|details|options)\b', ] action_matches = sum(1 for p in action_indicators if re.search(p, cleaned)) # It's a greeting if: has greeting patterns AND no action content AND is short if greeting_matches >= 1 and action_matches == 0 and word_count <= 20: return True # Very short messages (< 8 words) with greetings are always just greetings if word_count < 8 and greeting_matches >= 1: return True return False def _estimate_objective_coverage(self, learner_text: str, scenario: RoleplayScenario) -> float: """ Estimate what fraction of the scenario objectives the learner's text addresses. Returns a float between 0.0 and 1.0. """ if not learner_text.strip(): return 0.0 learner_lower = learner_text.lower() # Extract meaningful keywords from the objective objective_text = scenario.objective.lower() # Split objective into sub-objectives (usually comma or & separated) sub_objectives = re.split(r'[,&\n]+', objective_text) sub_objectives = [s.strip() for s in sub_objectives if len(s.strip()) > 3] if not sub_objectives: return 0.0 # For each sub-objective, extract keywords and check if learner addressed them addressed_count = 0 for sub_obj in sub_objectives: # Get significant words (skip common stop words) stop_words = {'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could', 'should', 'may', 'might', 'can', 'shall'} keywords = [w for w in sub_obj.split() if w not in stop_words and len(w) > 2] if keywords: # Check if any keyword from this sub-objective appears in learner text matches = sum(1 for kw in keywords if kw in learner_lower) if matches >= max(1, len(keywords) * 0.3): # At least 30% keyword match addressed_count += 1 coverage = addressed_count / len(sub_objectives) if sub_objectives else 0.0 return min(coverage, 1.0) def _build_engagement_metadata_text(self, depth_analysis: Dict[str, Any]) -> str: """ Build a human-readable engagement metadata string for injection into the LLM prompt. """ return ( f"--- CONVERSATION ENGAGEMENT SUMMARY ---\n" f"Learner Turn Count: {depth_analysis['learner_turn_count']}\n" f"Total Learner Words: {depth_analysis['total_learner_words']}\n" f"Average Words Per Turn: {depth_analysis['avg_words_per_turn']}\n" f"Learner Response Type: {'Generic greeting/introduction ONLY' if depth_analysis['is_only_greeting'] else 'Contains substantive content'}\n" f"Objective Coverage: {int(depth_analysis['objective_coverage'] * 100)}%\n" f"Engagement Level: {depth_analysis['engagement_level'].upper()}\n" f"Maximum Allowed Score: {depth_analysis['max_allowed_score']}/10\n" f"--- END ENGAGEMENT SUMMARY ---" ) def _apply_score_cap(self, skill_scores: List[SkillScore], max_allowed_score: int) -> List[SkillScore]: """ Clamp every skill score to the engagement-based maximum. Returns a new list of SkillScore objects with capped scores. """ capped_scores = [] for ss in skill_scores: original_score = ss.score capped_score = min(original_score, max_allowed_score) if capped_score != original_score: print(f"DEBUG: Score capped for '{ss.skill_name}': {original_score} → {capped_score} (max allowed: {max_allowed_score})") capped_scores.append(SkillScore( skill_name=ss.skill_name, score=capped_score, evidence=ss.evidence, strengths=ss.strengths, improvement_areas=ss.improvement_areas, )) return capped_scores # ────────────────────────────────────────────────────────── # MAIN ANALYSIS METHOD # ────────────────────────────────────────────────────────── async def analyze_session(self, session_id: str, scenario: RoleplayScenario, conversation_turns: List[ConversationTurn], is_admin: int, use_groq: bool = True) -> Optional[SkillAssessment]: """Perform comprehensive skill analysis on a completed session""" # Check if learner actually participated in the conversation learner_turns = [turn for turn in conversation_turns if turn.speaker == 'learner'] if not learner_turns: # No learner participation - return zero assessment return self._create_zero_assessment(session_id, scenario, conversation_turns, is_admin) # ── Step 1: Analyze conversation depth ── depth_analysis = self._analyze_conversation_depth(conversation_turns, scenario) # If engagement is "none", return zero assessment directly if depth_analysis["engagement_level"] == "none": return self._create_zero_assessment(session_id, scenario, conversation_turns, is_admin) # ── Step 2: Build engagement metadata for LLM ── engagement_metadata = self._build_engagement_metadata_text(depth_analysis) # ── Step 3: Get LLM analysis with engagement context ── conversation_data = [ { 'speaker': turn.speaker, 'message': turn.message } for turn in conversation_turns ] ai_service = self._get_ai_service(use_groq) analysis_result = await ai_service.analyze_skills( scenario.to_dict(), conversation_data, engagement_metadata ) if not analysis_result: return None try: # ── Step 4: Extract skill scores from LLM response ── skill_scores = [] for skill_name in scenario.skills_to_assess: if skill_name in analysis_result['skill_analysis']: skill_data = analysis_result['skill_analysis'][skill_name] skill_score = SkillScore( skill_name=skill_name, score=self._parse_score(skill_data.get('score', 0)), evidence=skill_data.get('evidence', []), strengths=skill_data.get('strengths', []), improvement_areas=skill_data.get('improvement_areas', []) ) skill_scores.append(skill_score) # ── Step 5: Apply engagement-based score cap ── max_allowed = depth_analysis["max_allowed_score"] skill_scores = self._apply_score_cap(skill_scores, max_allowed) # ── Step 6: Calculate overall score (from capped scores) ── if skill_scores: overall_score = sum(score.score for score in skill_scores) / len(skill_scores) else: overall_score = 0.0 # Also cap the overall score overall_score = min(overall_score, float(max_allowed)) # Determine performance level performance_level = self._determine_performance_level(overall_score) print(f"DEBUG: Final overall score: {overall_score} (engagement: {depth_analysis['engagement_level']}, cap: {max_allowed})") # Create assessment assessment = SkillAssessment.create_new( session_id=session_id, scenario_id=scenario.session_id, overall_score=round(overall_score, 1), performance_level=performance_level, skill_scores=skill_scores, conversation_turns=conversation_turns, conversation_analysis=analysis_result.get('conversation_analysis', {}), recommendations=analysis_result.get('recommendations', {}) ) # Save assessment print(f"DEBUG: Attempting to save assessment for session {session_id}") if self.json_handler.save_assessment(assessment.to_dict(), is_admin): print(f"DEBUG: Assessment saved successfully") return assessment else: print("DEBUG: Failed to save assessment") return None except Exception as e: print(f"Error processing analysis result: {e}") import traceback; traceback.print_exc() return None def _create_zero_assessment(self, session_id: str, scenario: RoleplayScenario, conversation_turns: List[ConversationTurn], is_admin: int) -> Optional[SkillAssessment]: """Create assessment with zero scores when learner didn't participate""" # Create zero skill scores for all skills skill_scores = [] for skill_name in scenario.skills_to_assess: skill_score = SkillScore( skill_name=skill_name, score=0, evidence=["No learner responses detected in the conversation"], strengths=[], improvement_areas=["Participate actively in the roleplay conversation"] ) skill_scores.append(skill_score) # Create assessment with zero scores assessment = SkillAssessment.create_new( session_id=session_id, scenario_id=scenario.session_id, overall_score=0.0, performance_level="No Participation", skill_scores=skill_scores, conversation_turns=conversation_turns, conversation_analysis={ "strengths": [], "critical_moments": [], "missed_opportunities": ["Complete lack of participation in the roleplay scenario"] }, recommendations={ "immediate_focus": ["Start participating in roleplay conversations"], "practice_suggestions": ["Begin with simple greetings and basic responses"], "advanced_skills": [] } ) # Save assessment print(f"DEBUG: Creating zero assessment for non-participating learner in session {session_id}") if self.json_handler.save_assessment(assessment.to_dict(), is_admin): print(f"DEBUG: Zero assessment saved successfully") return assessment else: print("DEBUG: Failed to save zero assessment") return None def _determine_performance_level(self, overall_score: float) -> str: """Determine performance level based on overall score""" if overall_score >= 9.0: return "Expert" elif overall_score >= 7.0: return "Advanced" elif overall_score >= 5.0: return "Intermediate" elif overall_score >= 3.0: return "Beginner" else: return "Needs Improvement" def load_assessment(self, session_id: str, is_admin: int) -> Optional[SkillAssessment]: """Load an existing assessment""" assessment_data = self.json_handler.load_assessment(session_id, is_admin) if assessment_data: return SkillAssessment.from_dict(assessment_data) return None def list_assessments(self, is_admin: int) -> List[Dict[str, Any]]: """List all assessments""" return self.json_handler.list_assessments(is_admin) def get_skill_summary(self, assessments: List[SkillAssessment]) -> Dict[str, Any]: """Generate summary statistics across multiple assessments""" if not assessments: return {} # Collect all skill scores skill_data = {} overall_scores = [] for assessment in assessments: overall_scores.append(assessment.overall_score) for skill_score in assessment.skill_scores: skill_name = skill_score.skill_name if skill_name not in skill_data: skill_data[skill_name] = [] skill_data[skill_name].append(skill_score.score) # Calculate averages skill_averages = {} for skill_name, scores in skill_data.items(): skill_averages[skill_name] = { 'average_score': round(sum(scores) / len(scores), 1), 'best_score': max(scores), 'latest_score': scores[-1], 'improvement': scores[-1] - scores[0] if len(scores) > 1 else 0 } return { 'total_sessions': len(assessments), 'overall_average': round(sum(overall_scores) / len(overall_scores), 1), 'best_overall': max(overall_scores), 'latest_overall': overall_scores[-1], 'skill_breakdown': skill_averages, 'performance_trend': self._calculate_trend(overall_scores) } def _calculate_trend(self, scores: List[float]) -> str: """Calculate performance trend""" if len(scores) < 2: return "Insufficient data" recent_avg = sum(scores[-3:]) / len(scores[-3:]) # Last 3 scores early_avg = sum(scores[:3]) / len(scores[:3]) # First 3 scores improvement = recent_avg - early_avg if improvement > 1.0: return "Improving" elif improvement < -1.0: return "Declining" else: return "Stable"