Spaces:
Runtime error
Runtime error
| """ | |
| 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 | |