"""Answer evaluation prediction service. Scores subjective answers and flags for teacher review using the answer_scorer model with rubric keyword coverage fallback when the model is unavailable. """ import logging import re import time import uuid from datetime import datetime, timezone from app.core.config import settings from app.core.exceptions import ModelNotLoadedError from app.data.loader import DatasetLoader from app.models.registry import ModelRegistry from app.monitoring.prediction_logger import PredictionLogger from app.schemas.answer_evaluation import AnswerEvaluationResponse from app.services.explanation_service import ExplanationService logger = logging.getLogger(__name__) class AnswerEvaluationService: """Scores subjective answers and flags for teacher review. Model: answer_scorer_v2_baseline_001 (TF-IDF + Ridge regression) Fallback: rubric keyword coverage + length heuristic """ def __init__( self, registry: ModelRegistry, loader: DatasetLoader, explainer: ExplanationService, pred_logger: PredictionLogger, ) -> None: self._registry = registry self._loader = loader self._explainer = explainer self._logger = pred_logger def predict( self, question_id: str, question_text: str, student_answer: str, model_answer: str, rubric: str, max_marks: int, grade: int, subject: str, lo_id: str, bloom_level: str, ) -> AnswerEvaluationResponse: """Score a student's subjective answer. Algorithm: 1. Check if answer_scorer model is loaded 2. If loaded: vectorize question+answer → predict marks 3. Else fallback: rubric keyword coverage scoring - Count rubric keywords present in student_answer - Scale by max_marks 4. Compute concepts_covered and missing_points from rubric 5. Generate feedback string 6. ALWAYS set teacher_review_required = True (V2 baseline) 7. Clamp predicted_marks to [0, max_marks] 8. Log prediction, return response """ start_time = time.perf_counter() prediction_id = str(uuid.uuid4()) timestamp = datetime.now(timezone.utc).isoformat() # Determine model version from registry metadata metadata = self._registry.get_metadata("answer_scorer") model_version = ( metadata.get("model_version", "answer_scorer_v2_baseline_001") if metadata else "answer_scorer_v2_baseline_001" ) # Parse rubric into concepts concepts = self._parse_rubric_concepts(rubric) # Compute concepts_covered and missing_points concepts_covered, missing_points = self._evaluate_concept_coverage( student_answer, concepts ) # Try model path source = "model" confidence = 0.0 predicted_marks = 0.0 try: predicted_marks, confidence = self._predict_with_model( question_text, student_answer, model_answer, max_marks ) if confidence < settings.low_confidence_threshold: logger.warning( "Answer scorer model confidence %.3f below threshold %.3f, using fallback", confidence, settings.low_confidence_threshold, ) predicted_marks, confidence = self._predict_fallback( concepts_covered, concepts, max_marks ) source = "fallback_rule_based" except (ModelNotLoadedError, Exception) as exc: logger.warning("Answer scorer model unavailable (%s), using fallback", exc) predicted_marks, confidence = self._predict_fallback( concepts_covered, concepts, max_marks ) source = "fallback_rule_based" # Clamp predicted_marks to [0, max_marks] predicted_marks = max(0.0, min(float(predicted_marks), float(max_marks))) # Generate feedback feedback = self._generate_feedback(concepts_covered, missing_points) # ALWAYS set teacher_review_required = True teacher_review_required = True response = AnswerEvaluationResponse( prediction_id=prediction_id, model_version=model_version, source=source, confidence=round(confidence, 4), timestamp=timestamp, predicted_marks=round(predicted_marks, 2), max_marks=max_marks, concepts_covered=concepts_covered, missing_points=missing_points, feedback=feedback, teacher_review_required=teacher_review_required, ) # Log prediction (do NOT log raw student_answer — only metadata) latency_ms = (time.perf_counter() - start_time) * 1000 self._logger.log( prediction_id=prediction_id, model_name="answer_scorer", model_version=model_version, endpoint="/ai/v2/evaluate-answer", input_summary={ "question_id": question_id, "student_answer_length": len(student_answer), "max_marks": max_marks, "grade": grade, "subject": subject, "lo_id": lo_id, "bloom_level": bloom_level, }, output={ "predicted_marks": round(predicted_marks, 2), "max_marks": max_marks, "confidence": round(confidence, 4), "source": source, "teacher_review_required": teacher_review_required, }, source=source, latency_ms=round(latency_ms, 2), ) return response def _predict_with_model( self, question_text: str, student_answer: str, model_answer: str, max_marks: int, ) -> tuple[float, float]: """Use the trained answer_scorer model to predict marks. Returns a tuple of (predicted_marks, confidence). """ model_data = self._registry.get_model("answer_scorer") model = model_data["model"] vectorizer = model_data.get("vectorizer") if vectorizer is None: raise ModelNotLoadedError("answer_scorer vectorizer not available") # Combine question and student answer for vectorization combined_text = f"{question_text} {student_answer}" text_vector = vectorizer.transform([combined_text]) # Predict marks raw_prediction = model.predict(text_vector)[0] predicted_marks = float(raw_prediction) # Scale prediction to max_marks range if needed # The model may predict on a normalized scale or raw marks predicted_marks = max(0.0, min(predicted_marks, float(max_marks))) # Compute confidence based on how close the prediction is to # the model answer similarity (heuristic for regression models) model_combined = f"{question_text} {model_answer}" model_vector = vectorizer.transform([model_combined]) # Use cosine-like similarity between student and model vectors # as a proxy for confidence from sklearn.metrics.pairwise import cosine_similarity similarity = cosine_similarity(text_vector, model_vector)[0][0] confidence = float(max(0.0, min(similarity, 1.0))) return predicted_marks, confidence def _predict_fallback( self, concepts_covered: list[str], all_concepts: list[str], max_marks: int, ) -> tuple[float, float]: """Fallback prediction using rubric keyword coverage scoring. Returns a tuple of (predicted_marks, confidence). """ total_concepts = len(all_concepts) if total_concepts == 0: # No rubric concepts to evaluate against return 0.0, 0.3 coverage_ratio = len(concepts_covered) / total_concepts predicted_marks = coverage_ratio * max_marks # Confidence for fallback is moderate and based on coverage clarity confidence = min(0.5 + (coverage_ratio * 0.2), 0.7) return predicted_marks, confidence def _parse_rubric_concepts(self, rubric: str) -> list[str]: """Split rubric into individual keywords/concepts. Splits by commas, periods, or semicolons and strips whitespace. Filters out empty strings. """ # Split by commas, periods, or semicolons parts = re.split(r"[,;.]", rubric) concepts = [part.strip() for part in parts if part.strip()] return concepts def _evaluate_concept_coverage( self, student_answer: str, concepts: list[str] ) -> tuple[list[str], list[str]]: """Check which rubric concepts appear in the student answer. Case-insensitive matching. Returns: A tuple of (concepts_covered, missing_points). """ answer_lower = student_answer.lower() concepts_covered: list[str] = [] missing_points: list[str] = [] for concept in concepts: if concept.lower() in answer_lower: concepts_covered.append(concept) else: missing_points.append(concept) return concepts_covered, missing_points def _generate_feedback( self, concepts_covered: list[str], missing_points: list[str] ) -> str: """Generate a feedback string summarizing coverage and gaps.""" parts: list[str] = [] if concepts_covered: covered_str = ", ".join(concepts_covered) parts.append(f"Student answer covers: {covered_str}.") if missing_points: missing_str = ", ".join(missing_points) parts.append(f"Missing key points: {missing_str}.") if not concepts_covered and not missing_points: parts.append("Unable to evaluate rubric coverage.") if not concepts_covered and missing_points: parts.append("Student answer does not address the key rubric concepts.") return " ".join(parts)