aaa / app /monitoring /prediction_logger.py
work-sejal
Deploy AI service with FastAPI
70ea7be
Raw
History Blame Contribute Delete
4.84 kB
"""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,
}