""" Context-Aware Selector for intelligent question selection based on interview context. This module provides the ContextAwareSelector class which integrates context analysis, diversity management, and effectiveness tracking to select optimal questions for each candidate based on their performance and knowledge gaps. """ import logging import time from typing import List, Dict, Optional from services.rag_service import RAGService from modules.question_diversity_manager import QuestionDiversityManager from modules.question_effectiveness_tracker import QuestionEffectivenessTracker from modules.interview_context import InterviewContext from modules.monitoring import get_monitor logger = logging.getLogger(__name__) class ContextAwareSelector: """ Selects interview questions based on context, diversity, and effectiveness. The ContextAwareSelector integrates multiple components to make intelligent question selection decisions: - Analyzes interview context to identify knowledge gaps and mastered topics - Adjusts difficulty based on candidate performance - Retrieves candidate questions from RAG service - Applies diversity filters to ensure comprehensive assessment - Ranks questions by effectiveness metrics - Selects the optimal question for the current interview state Implements Requirements: - 5.1: Maintain interview context with questions, answers, scores, gaps - 5.2: Consider candidate's performance on previous questions - 5.3: Identify knowledge gaps (score < 60%) - 5.4: Reduce probability for mastered topics (score > 85%) - 5.5: Adjust difficulty based on average score - 5.8: Complete selection within 500ms """ def __init__( self, rag_service: RAGService, diversity_manager: QuestionDiversityManager, effectiveness_tracker: QuestionEffectivenessTracker, knowledge_gap_threshold: float = 0.6, mastery_threshold: float = 0.85, mastery_reduction: float = 0.7 ): """ Initialize the Context-Aware Selector. Args: rag_service: RAG service for retrieving candidate questions diversity_manager: Manager for ensuring question diversity effectiveness_tracker: Tracker for question effectiveness metrics knowledge_gap_threshold: Score below which a topic is considered a gap (default: 0.6) mastery_threshold: Score above which a topic is considered mastered (default: 0.85) mastery_reduction: Probability reduction for mastered topics (default: 0.7 = 70% reduction) """ self.rag_service = rag_service self.diversity_manager = diversity_manager self.effectiveness_tracker = effectiveness_tracker self.knowledge_gap_threshold = knowledge_gap_threshold self.mastery_threshold = mastery_threshold self.mastery_reduction = mastery_reduction self.monitor = get_monitor() logger.info( f"ContextAwareSelector initialized " f"(gap_threshold={knowledge_gap_threshold}, " f"mastery_threshold={mastery_threshold})" ) def identify_knowledge_gaps( self, interview_context: InterviewContext ) -> List[str]: """ Identify topics where candidate scored below threshold. Analyzes the interview context to find topics where the candidate's average score is below the knowledge gap threshold (default 0.6). Returns topics prioritized by how far below the threshold they are. Args: interview_context: Current interview context with performance data Returns: List of topic names representing knowledge gaps, ordered by priority (worst performing topics first) """ # Get topic performance from context topic_performance = interview_context.get_topic_performance() if not topic_performance: logger.debug("No topic performance data available") return [] # Identify topics below threshold gaps = [] for topic, avg_score in topic_performance.items(): if avg_score < self.knowledge_gap_threshold: # Store topic with its gap size for prioritization gap_size = self.knowledge_gap_threshold - avg_score gaps.append((topic, gap_size, avg_score)) # Sort by gap size (largest gaps first) gaps.sort(key=lambda x: x[1], reverse=True) # Extract just the topic names gap_topics = [topic for topic, _, _ in gaps] if gap_topics: logger.info( f"Identified {len(gap_topics)} knowledge gaps: " f"{', '.join([f'{t} ({s:.2f})' for t, _, s in gaps])}" ) # Log knowledge gaps to monitoring avg_scores = {topic: score for topic, _, score in gaps} self.monitor.log_knowledge_gap_identified( interview_id=interview_context.interview_id, topics=gap_topics, avg_scores=avg_scores ) else: logger.debug("No knowledge gaps identified") return gap_topics def adjust_difficulty( self, interview_context: InterviewContext ) -> str: """ Determine appropriate difficulty level based on performance. Adjusts difficulty according to the rules: - If avg_score > 0.8: increase difficulty (easy→medium→hard) - If avg_score < 0.5: decrease difficulty (hard→medium→easy) - Otherwise: maintain current difficulty Args: interview_context: Current interview context with performance data Returns: Difficulty level string: 'easy', 'medium', or 'hard' """ avg_score = interview_context.get_average_score() current_difficulty = interview_context.current_difficulty # Define difficulty levels in order difficulty_levels = ['easy', 'medium', 'hard'] try: current_index = difficulty_levels.index(current_difficulty) except ValueError: # If current difficulty is invalid, default to medium logger.warning(f"Invalid difficulty '{current_difficulty}', defaulting to 'medium'") current_index = 1 current_difficulty = 'medium' # Apply adjustment rules if avg_score > 0.8: # Increase difficulty new_index = min(current_index + 1, len(difficulty_levels) - 1) new_difficulty = difficulty_levels[new_index] if new_difficulty != current_difficulty: logger.info( f"Increasing difficulty from '{current_difficulty}' to '{new_difficulty}' " f"(avg_score={avg_score:.2f})" ) # Log difficulty adjustment to monitoring self.monitor.log_difficulty_adjusted( interview_id=interview_context.interview_id, old_difficulty=current_difficulty, new_difficulty=new_difficulty, avg_score=avg_score, reason="high_performance" ) else: logger.debug(f"Already at maximum difficulty '{current_difficulty}'") return new_difficulty elif avg_score < 0.5: # Decrease difficulty new_index = max(current_index - 1, 0) new_difficulty = difficulty_levels[new_index] if new_difficulty != current_difficulty: logger.info( f"Decreasing difficulty from '{current_difficulty}' to '{new_difficulty}' " f"(avg_score={avg_score:.2f})" ) # Log difficulty adjustment to monitoring self.monitor.log_difficulty_adjusted( interview_id=interview_context.interview_id, old_difficulty=current_difficulty, new_difficulty=new_difficulty, avg_score=avg_score, reason="low_performance" ) else: logger.debug(f"Already at minimum difficulty '{current_difficulty}'") return new_difficulty else: # Maintain current difficulty logger.debug( f"Maintaining difficulty '{current_difficulty}' " f"(avg_score={avg_score:.2f})" ) return current_difficulty async def select_next_question( self, interview_context: InterviewContext, job_requirements: Optional[List[str]] = None ) -> Optional[Dict]: """ Select next question based on context, diversity, and effectiveness. Implements the selection algorithm: 1. Identify knowledge gaps and mastered topics 2. Adjust difficulty based on average performance 3. Get candidate questions from RAG service 4. Apply diversity filters 5. Rank by effectiveness metrics 6. Select top question Args: interview_context: Current interview context job_requirements: Optional list of job requirement topics to prioritize Returns: Dictionary containing selected question data, or None if no suitable question found """ start_time = time.time() logger.info(f"Selecting next question for interview {interview_context.interview_id}") try: # Step 1: Identify knowledge gaps and mastered topics knowledge_gaps = self.identify_knowledge_gaps(interview_context) topic_performance = interview_context.get_topic_performance() mastered_topics = [ topic for topic, score in topic_performance.items() if score >= self.mastery_threshold ] if mastered_topics: logger.info(f"Mastered topics: {', '.join(mastered_topics)}") # Step 2: Adjust difficulty target_difficulty = self.adjust_difficulty(interview_context) interview_context.current_difficulty = target_difficulty # Step 3: Get candidate questions from RAG # Prioritize knowledge gaps, then job requirements, then general topics query_topics = [] if knowledge_gaps: query_topics.extend(knowledge_gaps[:3]) # Top 3 gaps if job_requirements: query_topics.extend([req for req in job_requirements if req not in mastered_topics]) # If no specific topics, use a general query if not query_topics: query_topics = ["technical interview question"] # Query RAG for candidate questions candidate_questions = [] for topic in query_topics[:5]: # Limit to 5 queries for performance try: result = await self.rag_service.generate_question( job_description=topic, difficulty=target_difficulty ) if result.get('question'): # Build question dictionary question = { 'text': result['question'], 'topic': topic, 'difficulty': target_difficulty, 'question_type': result.get('category', 'technical'), 'context': result.get('context', ''), 'metadata': result.get('metadata', {}) } candidate_questions.append(question) except Exception as e: logger.warning(f"Error generating question for topic '{topic}': {e}") continue if not candidate_questions: logger.warning("No candidate questions generated from RAG") return None logger.info(f"Generated {len(candidate_questions)} candidate questions") # Step 4: Apply diversity filters filtered_questions = self.diversity_manager.filter_by_diversity( candidate_questions, interview_context ) if not filtered_questions: logger.warning("All candidate questions filtered out by diversity constraints") # Relax constraints and try again with original candidates filtered_questions = candidate_questions logger.info(f"After diversity filtering: {len(filtered_questions)} questions remain") # Step 5: Rank by effectiveness metrics # Note: Since these are newly generated questions, they may not have effectiveness scores # We'll use a simple scoring system based on context relevance scored_questions = [] for question in filtered_questions: score = 0.0 # Boost score for knowledge gap topics if question.get('topic') in knowledge_gaps: gap_index = knowledge_gaps.index(question['topic']) # Higher boost for higher priority gaps score += (len(knowledge_gaps) - gap_index) * 10 # Reduce score for mastered topics if question.get('topic') in mastered_topics: score *= (1 - self.mastery_reduction) # Boost score for job requirements if job_requirements and question.get('topic') in job_requirements: score += 5 # Add small random component to avoid always selecting same question import random score += random.uniform(0, 1) scored_questions.append((question, score)) # Sort by score (highest first) scored_questions.sort(key=lambda x: x[1], reverse=True) # Step 6: Select top question if scored_questions: selected_question, final_score = scored_questions[0] # Calculate latency latency_ms = (time.time() - start_time) * 1000 # Log question selection to monitoring self.monitor.log_question_selected( interview_id=interview_context.interview_id, question_id=selected_question.get('id', 0), # May be 0 for generated questions question_type=selected_question.get('question_type', 'technical'), difficulty=target_difficulty, topics=[selected_question.get('topic', '')], effectiveness_score=selected_question.get('effectiveness_score', 0.0), latency_ms=latency_ms ) logger.info( f"Selected question on topic '{selected_question.get('topic')}' " f"with score {final_score:.2f}" ) return selected_question logger.warning("No questions available after scoring") return None except Exception as e: logger.error(f"Error in select_next_question: {e}") # Log error to monitoring from modules.monitoring import ErrorType self.monitor.log_error( error_type=ErrorType.RAG_FAILURE, error_message=str(e), context={ 'interview_id': interview_context.interview_id, 'operation': 'question_selection' } ) return None