""" Interview Context module for managing interview state and conversation history. This module provides the InterviewContext class which maintains the accumulated state of an interview including previous questions, answers, performance metrics, and knowledge gaps. It supports efficient querying and persistence to the database. """ import time from typing import List, Dict, Set, Optional from sqlalchemy.orm import Session from models.conversation_entry import ConversationEntry from models.interview import Interview from modules.monitoring import get_monitor, ErrorType class InterviewContext: """ Manages the state and conversation history of an interview. The InterviewContext maintains all relevant information about an ongoing or completed interview, including questions asked, answers given, scores, topics covered, and identified knowledge gaps. It provides methods for updating the context and querying performance metrics. Attributes: interview_id: Unique identifier for the interview db_session: SQLAlchemy database session for persistence questions: List of question dictionaries asked in the interview answers: List of answer dictionaries given by the candidate scores: List of scores for each question-answer pair topics_covered: Set of unique topics covered in the interview knowledge_gaps: List of topics where candidate showed weakness current_difficulty: Current difficulty level (easy, medium, hard) followup_depth: Current depth of consecutive follow-up questions """ def __init__(self, interview_id: int, db_session: Session): """ Initialize interview context with interview ID and database session. Args: interview_id: Unique identifier for the interview db_session: SQLAlchemy database session for persistence """ self.interview_id = interview_id self.db_session = db_session self.questions: List[Dict] = [] self.answers: List[Dict] = [] self.scores: List[float] = [] self.topics_covered: Set[str] = set() self.knowledge_gaps: List[str] = [] self.current_difficulty: str = "medium" self.followup_depth: int = 0 self.monitor = get_monitor() def add_qa_pair( self, question: Dict, answer: str, score: float, is_followup: bool = False ) -> None: """ Add a question-answer pair to the context. Updates the context with a new question-answer pair, including the score. Also updates topics_covered, scores list, and followup_depth based on whether this is a follow-up question. Args: question: Dictionary containing question data with keys: - id: Question ID (optional, may be None for follow-ups) - text: Question text - topic: Question topic (optional) - difficulty: Difficulty level (optional) - question_type: Type of question (optional) answer: The candidate's answer text score: Score for the answer (0.0 to 1.0) is_followup: Whether this is a follow-up question """ # Add question to questions list self.questions.append(question) # Add answer to answers list answer_dict = { 'text': answer, 'score': score, 'is_followup': is_followup } self.answers.append(answer_dict) # Add score to scores list self.scores.append(score) # Update topics_covered if topic is provided if 'topic' in question and question['topic']: self.topics_covered.add(question['topic']) # Update followup_depth if is_followup: self.followup_depth += 1 else: self.followup_depth = 0 def get_recent_questions(self, n: int = 5) -> List[Dict]: """ Get the last n questions asked in the interview. Args: n: Number of recent questions to retrieve (default: 5) Returns: List of question dictionaries, most recent last """ return self.questions[-n:] if len(self.questions) >= n else self.questions def get_average_score(self) -> float: """ Calculate the average score across all questions. Returns: Average score (0.0 to 1.0), or 0.0 if no scores recorded """ if not self.scores: return 0.0 return sum(self.scores) / len(self.scores) def get_topic_performance(self) -> Dict[str, float]: """ Get average score by topic. Calculates the average score for each topic that has been covered in the interview. Returns: Dictionary mapping topic names to average scores """ topic_scores: Dict[str, List[float]] = {} # Group scores by topic for i, question in enumerate(self.questions): if 'topic' in question and question['topic']: topic = question['topic'] if topic not in topic_scores: topic_scores[topic] = [] if i < len(self.scores): topic_scores[topic].append(self.scores[i]) # Calculate average for each topic topic_performance = {} for topic, scores in topic_scores.items(): if scores: topic_performance[topic] = sum(scores) / len(scores) return topic_performance def load_from_db(self) -> None: """ Load existing interview context from database. Restores the interview context by querying conversation entries from the database. Handles missing or corrupted data gracefully by logging warnings and continuing with partial data. Raises: No exceptions - handles errors gracefully """ start_time = time.time() try: # Query conversation entries for this interview, ordered by sequence entries = self.db_session.query(ConversationEntry).filter( ConversationEntry.interview_id == self.interview_id ).order_by(ConversationEntry.sequence_number).all() # Reset context state self.questions = [] self.answers = [] self.scores = [] self.topics_covered = set() self.followup_depth = 0 # Rebuild context from conversation entries for entry in entries: # Build question dictionary question = { 'id': entry.question_id, 'text': entry.question_text, 'topic': entry.topic, 'difficulty': entry.difficulty_level, 'question_type': entry.question_type } self.questions.append(question) # Build answer dictionary answer = { 'text': entry.answer_text or '', 'score': entry.score or 0.0, 'is_followup': entry.is_followup } self.answers.append(answer) # Add score if entry.score is not None: self.scores.append(entry.score) # Add topic to topics_covered if entry.topic: self.topics_covered.add(entry.topic) # Update followup_depth (track consecutive follow-ups) if entry.is_followup: self.followup_depth = entry.followup_depth else: self.followup_depth = 0 # Load interview metadata interview = self.db_session.query(Interview).filter( Interview.id == self.interview_id ).first() if interview: # Load knowledge gaps if available if interview.knowledge_gaps: self.knowledge_gaps = interview.knowledge_gaps # Infer current difficulty from last question or use default if entries and entries[-1].difficulty_level: self.current_difficulty = entries[-1].difficulty_level else: self.current_difficulty = "medium" # Calculate latency and log success latency_ms = (time.time() - start_time) * 1000 self.monitor.log_context_loaded( interview_id=self.interview_id, questions_count=len(self.questions), latency_ms=latency_ms ) except Exception as e: # Log error but don't raise - allow context to be used with empty state print(f"Warning: Error loading interview context for interview {self.interview_id}: {e}") # Log error to monitoring self.monitor.log_error( error_type=ErrorType.CONTEXT_LOADING_FAILURE, error_message=str(e), context={'interview_id': self.interview_id} ) # Initialize with empty state self.questions = [] self.answers = [] self.scores = [] self.topics_covered = set() self.knowledge_gaps = [] self.current_difficulty = "medium" self.followup_depth = 0 def save_to_db(self) -> None: """ Persist current context to database. Saves the current interview context by creating or updating conversation entries in the database. Also updates the interview record with topics covered and knowledge gaps. Handles database errors gracefully. Raises: No exceptions - handles errors gracefully """ start_time = time.time() try: # Get existing conversation entries count to determine sequence numbers existing_count = self.db_session.query(ConversationEntry).filter( ConversationEntry.interview_id == self.interview_id ).count() # Save only new entries (those not yet persisted) for i in range(existing_count, len(self.questions)): question = self.questions[i] answer = self.answers[i] if i < len(self.answers) else None score = self.scores[i] if i < len(self.scores) else None # Create conversation entry entry = ConversationEntry( interview_id=self.interview_id, sequence_number=i + 1, # 1-indexed question_id=question.get('id'), question_text=question.get('text', ''), answer_text=answer.get('text', '') if answer else '', score=score, is_followup=answer.get('is_followup', False) if answer else False, followup_depth=self.followup_depth if answer and answer.get('is_followup') else 0, difficulty_level=question.get('difficulty'), topic=question.get('topic'), question_type=question.get('question_type'), time_to_answer=None # TODO: Track time to answer ) self.db_session.add(entry) # Update interview record with context metadata interview = self.db_session.query(Interview).filter( Interview.id == self.interview_id ).first() if interview: # Update topics covered interview.topics_covered = list(self.topics_covered) # Update knowledge gaps interview.knowledge_gaps = self.knowledge_gaps # Update final score if interview is complete if self.scores: interview.final_score = self.get_average_score() # Commit changes self.db_session.commit() # Calculate latency and log success latency_ms = (time.time() - start_time) * 1000 self.monitor.log_context_saved( interview_id=self.interview_id, latency_ms=latency_ms ) except Exception as e: # Log error and rollback print(f"Error saving interview context for interview {self.interview_id}: {e}") self.db_session.rollback() # Log error to monitoring self.monitor.log_error( error_type=ErrorType.DATABASE_FAILURE, error_message=str(e), context={ 'interview_id': self.interview_id, 'operation': 'save_context' } ) # Don't raise - allow interview to continue even if persistence fails