"""Unified evaluation report generator. Produces `evaluation_summary.md` from all model training results, comparing each model's primary metric against its acceptance target and documenting known limitations. """ import logging from datetime import datetime, timezone from pathlib import Path from app.core.config import settings from training.base_trainer import TrainingResult logger = logging.getLogger(__name__) # Human-readable display names for each model _MODEL_DISPLAY_NAMES: dict[str, str] = { "lo_tagger": "LO Tagger", "bloom_classifier": "Bloom Classifier", "difficulty_model": "Difficulty Model", "mastery_model": "Mastery Model", "risk_model": "Risk Model", "answer_scorer": "Answer Scorer", "recommender": "Recommender", } class EvaluationReportGenerator: """Generates unified evaluation_summary.md from all model metrics.""" ACCEPTANCE_TARGETS: dict[str, dict] = { "lo_tagger": { "metric": "top_3_accuracy", "target": 0.80, "label": "Top-3 Accuracy", }, "bloom_classifier": { "metric": "macro_f1", "target": 0.55, "label": "Macro F1", }, "difficulty_model": { "metric": "mae", "target": 0.15, "label": "MAE (lower=better)", "lower_is_better": True, }, "mastery_model": { "metric": "macro_f1", "target": 0.60, "label": "Macro F1", }, "risk_model": { "metric": "recall_positive", "target": 0.75, "label": "Recall (at-risk)", }, "answer_scorer": { "metric": "mae", "target": 0.80, "label": "MAE (lower=better)", "lower_is_better": True, }, "recommender": { "metric": "roc_auc_clicked", "target": 0.70, "label": "ROC-AUC (clicked)", }, } def __init__(self, artifact_base_dir: str | Path, reports_dir: str | Path) -> None: self._artifact_base_dir = Path(artifact_base_dir) self._reports_dir = Path(reports_dir) def generate(self, results: list[TrainingResult]) -> Path: """Generate evaluation_summary.md from training results. 1. Create reports_dir if it doesn't exist 2. Build header with timestamp, dataset version, seed 3. Build summary table comparing primary metrics against targets 4. Add per-model sections with full metrics 5. Add known limitations section 6. Write to evaluation_summary.md 7. Return path to generated report """ self._reports_dir.mkdir(parents=True, exist_ok=True) lines: list[str] = [] # Header self._build_header(lines) # Summary table self._build_summary_table(lines, results) # Per-model sections for result in results: self._build_model_section(lines, result) # Known limitations self._build_limitations_section(lines) # Write report report_path = self._reports_dir / "evaluation_summary.md" report_content = "\n".join(lines) report_path.write_text(report_content, encoding="utf-8") logger.info("Evaluation report written to %s", report_path) return report_path def _build_header(self, lines: list[str]) -> None: """Build report header with timestamp, dataset version, and seed.""" timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") lines.append("# Evaluation Summary Report") lines.append("") lines.append(f"**Generated:** {timestamp}") lines.append(f"**Dataset Version:** {settings.ai_service_version}") lines.append(f"**Seed:** {settings.seed}") lines.append("") def _build_summary_table( self, lines: list[str], results: list[TrainingResult] ) -> None: """Build summary table comparing each model's primary metric to target.""" lines.append("## Summary") lines.append("") lines.append( "| Model | Primary Metric | Actual | Target | Status |" ) lines.append( "|-------|---------------|--------|--------|--------|" ) for result in results: model_name = result.model_name display_name = _MODEL_DISPLAY_NAMES.get(model_name, model_name) target_info = self.ACCEPTANCE_TARGETS.get(model_name) if target_info is None: lines.append( f"| {display_name} | N/A | N/A | N/A | — |" ) continue metric_key = target_info["metric"] target_value = target_info["target"] label = target_info["label"] lower_is_better = target_info.get("lower_is_better", False) actual_value = self._extract_metric(result, metric_key) if actual_value is not None: actual_str = f"{actual_value:.4f}" if lower_is_better: passed = actual_value <= target_value else: passed = actual_value >= target_value status = "✓ PASS" if passed else "✗ FAIL" else: actual_str = "N/A" status = "— MISSING" target_str = f"{target_value:.2f}" lines.append( f"| {display_name} | {label} | {actual_str} | {target_str} | {status} |" ) lines.append("") def _build_model_section( self, lines: list[str], result: TrainingResult ) -> None: """Build per-model section with full metrics.""" display_name = _MODEL_DISPLAY_NAMES.get(result.model_name, result.model_name) lines.append(f"## {display_name}") lines.append("") lines.append(f"- **Model Version:** {result.model_version}") lines.append( f"- **Trained At:** {result.trained_at.strftime('%Y-%m-%dT%H:%M:%SZ')}" ) lines.append( f"- **Split Counts:** train={result.split_counts.get('train', 'N/A')}, " f"validation={result.split_counts.get('validation', 'N/A')}, " f"test={result.split_counts.get('test', 'N/A')}" ) lines.append("") # Extract test metrics (primary) and validation metrics (secondary) metrics_dict = result.metrics.get("metrics", {}) test_metrics = metrics_dict.get("test", {}) val_metrics = metrics_dict.get("validation", {}) # Display test metrics if test_metrics: lines.append("### Test Metrics") lines.append("") self._format_metrics_block(lines, test_metrics) lines.append("") # Display validation metrics if val_metrics: lines.append("### Validation Metrics") lines.append("") self._format_metrics_block(lines, val_metrics) lines.append("") # Limitations from the model's metrics.json limitations = result.metrics.get("limitations", []) if limitations: lines.append("### Model Limitations") lines.append("") for limitation in limitations: lines.append(f"- {limitation}") lines.append("") def _format_metrics_block(self, lines: list[str], metrics: dict) -> None: """Format a flat metrics dict as bullet points, skipping nested dicts.""" for key, value in metrics.items(): if key in ("per_class", "confusion_matrix"): # Skip large nested structures in the summary continue if isinstance(value, float): lines.append(f"- **{key}:** {value:.4f}") else: lines.append(f"- **{key}:** {value}") def _build_limitations_section(self, lines: list[str]) -> None: """Build the known limitations section.""" lines.append("## Known Limitations") lines.append("") lines.append("- All models trained on synthetic data only") lines.append( "- Class imbalance in Bloom (Create ~2%, Evaluate ~4%) " "and Risk (critical ~2%)" ) lines.append( "- LO Tagger has 194 classes with imbalanced distribution" ) lines.append( "- Mastery labels derived from synthetic score thresholds, " "not real teacher assessments" ) lines.append( "- Answer Scorer trained on synthetic rubric matches, " "teacher_review_required always True" ) lines.append( "- Recommender trained on synthetic click/completion signals" ) lines.append( "- Performance on real student data is unknown and may differ " "significantly" ) lines.append("") def _extract_metric( self, result: TrainingResult, metric_key: str ) -> float | None: """Extract a metric value from a TrainingResult. Looks in metrics["metrics"]["test"][metric_key] first, falls back to metrics["metrics"]["validation"][metric_key]. Returns None if not found. """ metrics_dict = result.metrics.get("metrics", {}) # Try test split first test_metrics = metrics_dict.get("test", {}) if metric_key in test_metrics: value = test_metrics[metric_key] if isinstance(value, (int, float)): return float(value) # Fall back to validation split val_metrics = metrics_dict.get("validation", {}) if metric_key in val_metrics: value = val_metrics[metric_key] if isinstance(value, (int, float)): return float(value) return None