Spaces:
Sleeping
Sleeping
| """Prediction logger for monitoring and drift detection. | |
| Logs every prediction to a JSONL file with anonymized student IDs. | |
| Logging is synchronous but lightweight (append to file). | |
| """ | |
| import hashlib | |
| import json | |
| import logging | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| logger = logging.getLogger(__name__) | |
| # Fields that must never appear in prediction logs (PII protection) | |
| _PII_FIELDS = {"student_name", "student_answer", "parent_contact"} | |
| class PredictionLogger: | |
| """Logs predictions to JSONL for monitoring and drift detection. | |
| Logging is synchronous but lightweight (append to file). | |
| Student IDs are anonymized via stable hash before logging. | |
| """ | |
| def __init__(self, log_dir: Path, salt: str) -> None: | |
| self._log_dir = Path(log_dir) | |
| self._log_dir.mkdir(parents=True, exist_ok=True) | |
| self._log_path = self._log_dir / "prediction_logs.jsonl" | |
| self._salt = salt | |
| def log( | |
| self, | |
| prediction_id: str, | |
| model_name: str, | |
| model_version: str, | |
| endpoint: str, | |
| input_summary: dict, | |
| output: dict, | |
| source: str, | |
| latency_ms: float, | |
| ) -> None: | |
| """Append a prediction log entry. Non-blocking, fire-and-forget.""" | |
| try: | |
| # Anonymize student_id in input_summary if present | |
| safe_input = self._sanitize_input_summary(input_summary) | |
| entry = { | |
| "prediction_id": prediction_id, | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "model_name": model_name, | |
| "model_version": model_version, | |
| "endpoint": endpoint, | |
| "input_summary": safe_input, | |
| "output": output, | |
| "source": source, | |
| "latency_ms": latency_ms, | |
| } | |
| with open(self._log_path, "a", encoding="utf-8") as f: | |
| f.write(json.dumps(entry) + "\n") | |
| except Exception: | |
| # Fire-and-forget: never let logging failures propagate | |
| logger.exception("Failed to write prediction log entry") | |
| def _anonymize_student_id(self, student_id: str) -> str: | |
| """SHA-256 hash with salt, truncated to 16 chars.""" | |
| return hashlib.sha256( | |
| f"{self._salt}:{student_id}".encode() | |
| ).hexdigest()[:16] | |
| def _sanitize_input_summary(self, input_summary: dict) -> dict: | |
| """Remove PII fields and anonymize student_id in input summary.""" | |
| safe = {} | |
| for key, value in input_summary.items(): | |
| if key in _PII_FIELDS: | |
| continue | |
| if key == "student_id" and isinstance(value, str): | |
| safe["student_id"] = self._anonymize_student_id(value) | |
| else: | |
| safe[key] = value | |
| return safe | |
| def get_recent_stats(self, model_name: str, hours: int = 24) -> dict: | |
| """Return prediction count and avg confidence for last N hours.""" | |
| count = 0 | |
| total_confidence = 0.0 | |
| if not self._log_path.exists(): | |
| return {"prediction_count": 0, "avg_confidence": None} | |
| cutoff = datetime.now(timezone.utc).timestamp() - (hours * 3600) | |
| try: | |
| with open(self._log_path, "r", encoding="utf-8") as f: | |
| for line in f: | |
| line = line.strip() | |
| if not line: | |
| continue | |
| try: | |
| entry = json.loads(line) | |
| except json.JSONDecodeError: | |
| continue | |
| if entry.get("model_name") != model_name: | |
| continue | |
| # Parse timestamp and check if within window | |
| ts_str = entry.get("timestamp", "") | |
| try: | |
| ts = datetime.fromisoformat(ts_str).timestamp() | |
| except (ValueError, TypeError): | |
| continue | |
| if ts < cutoff: | |
| continue | |
| count += 1 | |
| # Extract confidence from output | |
| output = entry.get("output", {}) | |
| confidence = output.get("confidence") | |
| if confidence is not None: | |
| try: | |
| total_confidence += float(confidence) | |
| except (ValueError, TypeError): | |
| pass | |
| except Exception: | |
| logger.exception("Failed to read prediction log for stats") | |
| return {"prediction_count": 0, "avg_confidence": None} | |
| avg_confidence = (total_confidence / count) if count > 0 else None | |
| return { | |
| "prediction_count": count, | |
| "avg_confidence": avg_confidence, | |
| } | |