Spaces:
Runtime error
Runtime error
| """ | |
| Question Effectiveness Tracker for monitoring and calculating question effectiveness metrics | |
| """ | |
| import asyncio | |
| import statistics | |
| from datetime import datetime, timedelta, timezone | |
| from typing import Dict, List, Optional | |
| from sqlalchemy.orm import Session | |
| from sqlalchemy import func | |
| from models.question_metrics import QuestionMetrics | |
| from models.conversation_entry import ConversationEntry | |
| from modules.monitoring import get_monitor | |
| class QuestionEffectivenessTracker: | |
| """ | |
| Tracks question effectiveness metrics and provides scoring for question selection. | |
| This class records question usage asynchronously, calculates effectiveness scores | |
| based on variance, completion rate, and time to answer, and provides caching | |
| for frequently accessed metrics. | |
| """ | |
| def __init__(self, db_session: Session, cache: Optional[Dict] = None, config: Optional[Dict] = None): | |
| """ | |
| Initialize the Question Effectiveness Tracker. | |
| Args: | |
| db_session: SQLAlchemy database session | |
| cache: Optional cache dictionary for storing metrics (defaults to empty dict) | |
| config: Optional configuration dictionary with keys: | |
| - min_samples: Minimum number of samples before calculating effectiveness (default: 10) | |
| - cache_ttl: Cache time-to-live in seconds (default: 3600 = 1 hour) | |
| - variance_weight: Weight for variance in effectiveness formula (default: 0.6) | |
| - completion_weight: Weight for completion rate (default: 0.3) | |
| - time_weight: Weight for time to answer (default: 0.1) | |
| - time_normalizer: Normalizer for time to answer in seconds (default: 300) | |
| """ | |
| self.db_session = db_session | |
| self.cache = cache if cache is not None else {} | |
| self.monitor = get_monitor() | |
| # Configuration with defaults | |
| config = config or {} | |
| self.min_samples = config.get('min_samples', 10) | |
| self.cache_ttl = config.get('cache_ttl', 3600) # 1 hour in seconds | |
| self.variance_weight = config.get('variance_weight', 0.6) | |
| self.completion_weight = config.get('completion_weight', 0.3) | |
| self.time_weight = config.get('time_weight', 0.1) | |
| self.time_normalizer = config.get('time_normalizer', 300) # 5 minutes in seconds | |
| async def record_usage( | |
| self, | |
| question_id: int, | |
| score: float, | |
| time_to_answer: int, | |
| completed: bool | |
| ) -> None: | |
| """ | |
| Record question usage asynchronously and update metrics. | |
| This method updates the question metrics in the database by: | |
| - Incrementing usage count | |
| - Updating score statistics for variance calculation | |
| - Updating completion rate | |
| - Updating average time to answer | |
| Args: | |
| question_id: ID of the question used | |
| score: Score received for the answer (0.0 to 1.0) | |
| time_to_answer: Time taken to answer in seconds | |
| completed: Whether the question was completed (not skipped) | |
| """ | |
| try: | |
| # Run database update in thread pool to avoid blocking | |
| await asyncio.get_event_loop().run_in_executor( | |
| None, | |
| self._update_metrics_sync, | |
| question_id, | |
| score, | |
| time_to_answer, | |
| completed | |
| ) | |
| # Invalidate cache for this question | |
| cache_key = f"effectiveness_{question_id}" | |
| if cache_key in self.cache: | |
| del self.cache[cache_key] | |
| except Exception as e: | |
| # Log error but don't raise - we don't want to block interview flow | |
| print(f"Error recording usage for question {question_id}: {e}") | |
| def _update_metrics_sync( | |
| self, | |
| question_id: int, | |
| score: float, | |
| time_to_answer: int, | |
| completed: bool | |
| ) -> None: | |
| """ | |
| Synchronous method to update metrics in database. | |
| This is called by record_usage in a thread pool executor. | |
| """ | |
| try: | |
| # Get or create metrics record | |
| metrics = self.db_session.query(QuestionMetrics).filter( | |
| QuestionMetrics.question_id == question_id | |
| ).first() | |
| old_effectiveness = metrics.effectiveness_score if metrics else 0.0 | |
| if not metrics: | |
| metrics = QuestionMetrics(question_id=question_id) | |
| self.db_session.add(metrics) | |
| # Get all scores for this question to calculate variance | |
| scores = self.db_session.query(ConversationEntry.score).filter( | |
| ConversationEntry.question_id == question_id, | |
| ConversationEntry.score.isnot(None) | |
| ).all() | |
| scores = [s[0] for s in scores] + [score] # Include current score | |
| # Calculate variance if we have enough samples | |
| if len(scores) >= 2: | |
| metrics.score_variance = statistics.variance(scores) | |
| else: | |
| metrics.score_variance = 0.0 | |
| # Update usage count | |
| metrics.usage_count += 1 | |
| # Update completion rate | |
| total_attempts = self.db_session.query(func.count(ConversationEntry.id)).filter( | |
| ConversationEntry.question_id == question_id | |
| ).scalar() or 0 | |
| completed_attempts = self.db_session.query(func.count(ConversationEntry.id)).filter( | |
| ConversationEntry.question_id == question_id, | |
| ConversationEntry.answer_text.isnot(None), | |
| ConversationEntry.answer_text != '' | |
| ).scalar() or 0 | |
| if completed: | |
| completed_attempts += 1 | |
| total_attempts += 1 | |
| metrics.completion_rate = completed_attempts / total_attempts if total_attempts > 0 else 1.0 | |
| # Calculate effectiveness score | |
| new_effectiveness = self.calculate_effectiveness(question_id) | |
| metrics.effectiveness_score = new_effectiveness | |
| # Commit changes | |
| self.db_session.commit() | |
| # Log effectiveness update to monitoring | |
| self.monitor.log_effectiveness_updated( | |
| question_id=question_id, | |
| old_score=old_effectiveness, | |
| new_score=new_effectiveness, | |
| usage_count=metrics.usage_count | |
| ) | |
| except Exception as e: | |
| self.db_session.rollback() | |
| print(f"Error in _update_metrics_sync for question {question_id}: {e}") | |
| raise | |
| def calculate_effectiveness(self, question_id: int) -> float: | |
| """ | |
| Calculate effectiveness metric for a question. | |
| Formula: | |
| effectiveness = (score_variance * 0.6) + (completion_rate * 0.3) + (avg_time / 300 * 0.1) | |
| Returns 0.0 if insufficient samples (< min_samples). | |
| Args: | |
| question_id: ID of the question | |
| Returns: | |
| Effectiveness score (0.0 to ~1.0, though can exceed 1.0 in edge cases) | |
| """ | |
| try: | |
| # Get metrics from database | |
| metrics = self.db_session.query(QuestionMetrics).filter( | |
| QuestionMetrics.question_id == question_id | |
| ).first() | |
| # Return 0.0 if no metrics or insufficient samples | |
| if not metrics or metrics.usage_count < self.min_samples: | |
| return 0.0 | |
| # Get average time to answer from conversation entries | |
| avg_time_result = self.db_session.query( | |
| func.avg(ConversationEntry.time_to_answer) | |
| ).filter( | |
| ConversationEntry.question_id == question_id, | |
| ConversationEntry.time_to_answer.isnot(None) | |
| ).scalar() | |
| avg_time = avg_time_result if avg_time_result is not None else 0 | |
| # Handle edge cases | |
| variance = metrics.score_variance if metrics.score_variance is not None else 0.0 | |
| completion = metrics.completion_rate if metrics.completion_rate is not None else 1.0 | |
| # Normalize time (cap at normalizer value to prevent excessive weight) | |
| time_component = min(avg_time / self.time_normalizer, 1.0) if self.time_normalizer > 0 else 0.0 | |
| # Calculate effectiveness using weighted formula | |
| effectiveness = ( | |
| variance * self.variance_weight + | |
| completion * self.completion_weight + | |
| time_component * self.time_weight | |
| ) | |
| return effectiveness | |
| except Exception as e: | |
| print(f"Error calculating effectiveness for question {question_id}: {e}") | |
| return 0.0 | |
| def get_effectiveness_scores(self, question_ids: List[int]) -> Dict[int, float]: | |
| """ | |
| Get effectiveness scores for multiple questions with caching. | |
| Uses cache when available (TTL 1 hour), queries database otherwise. | |
| Args: | |
| question_ids: List of question IDs | |
| Returns: | |
| Dictionary mapping question_id to effectiveness score | |
| """ | |
| results = {} | |
| questions_to_query = [] | |
| current_time = datetime.now(timezone.utc) | |
| # Check cache first | |
| for question_id in question_ids: | |
| cache_key = f"effectiveness_{question_id}" | |
| if cache_key in self.cache: | |
| cached_data = self.cache[cache_key] | |
| cache_time = cached_data.get('timestamp') | |
| # Check if cache is still valid (within TTL) | |
| if cache_time and (current_time - cache_time).total_seconds() < self.cache_ttl: | |
| results[question_id] = cached_data['score'] | |
| # Log cache hit | |
| self.monitor.log_cache_hit(cache_key) | |
| else: | |
| # Cache expired, need to refresh | |
| self.monitor.log_cache_miss(cache_key) | |
| questions_to_query.append(question_id) | |
| else: | |
| # Not in cache | |
| self.monitor.log_cache_miss(cache_key) | |
| questions_to_query.append(question_id) | |
| # Query database for non-cached questions | |
| if questions_to_query: | |
| for question_id in questions_to_query: | |
| try: | |
| score = self.calculate_effectiveness(question_id) | |
| results[question_id] = score | |
| # Update cache | |
| cache_key = f"effectiveness_{question_id}" | |
| self.cache[cache_key] = { | |
| 'score': score, | |
| 'timestamp': current_time | |
| } | |
| except Exception as e: | |
| print(f"Error getting effectiveness for question {question_id}: {e}") | |
| results[question_id] = 0.0 | |
| return results | |