feat(#97): add SHAP narrator task type to training pipeline

Add narrator as a third task alongside explainer and summarizer.
The narrator takes per-student SHAP values + profile and generates
advisor-facing narratives grounded in ML feature attribution.

- prompts.py: NARRATOR_SCHEMA, NARRATOR_STUDENT_SYSTEM, build_narrator_prompt()
- seed.py: generate_synthetic_student_profiles() with SHAP data
- distill.py: narrator in _TASK_CONFIG, included in main() distillation loop
- eval.py: _NARRATOR_REQUIRED_KEYS, shap_grounding ship criterion (>= 80%),
  check_shap_grounding() metric (counts feature name mentions in narrative)
- prepare.py: narrator added to task iteration

Author: William-Hill

training/distill.py

@@ -21,15 +21,18 @@
 from training.config import get_training_data_dir, load_school_config, write_jsonl
 from training.prompts import (
     EXPLAINER_STUDENT_SYSTEM,
+    NARRATOR_STUDENT_SYSTEM,
     SUMMARIZER_STUDENT_SYSTEM,
     build_explainer_prompt,
+    build_narrator_prompt,
     build_summarizer_prompt,
     build_system_prompt,
 )
 from training.seed import (
     format_as_chatml,
     generate_synthetic_course_pairings,
     generate_synthetic_query_results,
+    generate_synthetic_student_profiles,
 )
 
 # Cost tracking
@@ -136,6 +139,11 @@ def call_teacher(system: str, user: str, backend: str, model: str) -> str:
 _FLUSH_INTERVAL = 25
 
 _TASK_CONFIG = {
+    "narrator": {
+        "prompt_builder": build_narrator_prompt,
+        "student_system": NARRATOR_STUDENT_SYSTEM,
+        "format_user": lambda config, data: json.dumps(data, ensure_ascii=False, default=str),
+    },
     "explainer": {
         "prompt_builder": build_explainer_prompt,
         "student_system": EXPLAINER_STUDENT_SYSTEM,
@@ -245,28 +253,43 @@ def main(school: str, local: bool = False) -> None:
     data_dir = get_training_data_dir(school)
     pairs_dir = data_dir / "pairs"
 
-    synthetic_pairings = generate_synthetic_course_pairings(config, count=pairs_per_task)
-    synthetic_results = generate_synthetic_query_results(config, count=pairs_per_task)
-
     system_prompt = build_system_prompt(config)
 
+    all_counts: dict[str, int] = {}
+
+    # Narrator
+    print(f"\n{'='*60}\nNARRATOR — generating {pairs_per_task} pairs\n{'='*60}")
+    synthetic_profiles = generate_synthetic_student_profiles(config, count=pairs_per_task)
+    narrator_pairs = generate_pairs(
+        config=config, seed_data=synthetic_profiles,
+        count=pairs_per_task, task="narrator", outfile=pairs_dir / "narrator.jsonl",
+        system_prompt=system_prompt,
+    )
+    all_counts["narrator"] = len(narrator_pairs)
+
+    # Explainer
     print(f"\n{'='*60}\nEXPLAINER — generating {pairs_per_task} pairs\n{'='*60}")
-    explainer_pairs = generate_explainer_pairs(
+    synthetic_pairings = generate_synthetic_course_pairings(config, count=pairs_per_task)
+    explainer_pairs = generate_pairs(
         config=config, seed_data=synthetic_pairings,
-        count=pairs_per_task, outfile=pairs_dir / "explainer.jsonl",
+        count=pairs_per_task, task="explainer", outfile=pairs_dir / "explainer.jsonl",
         system_prompt=system_prompt,
     )
+    all_counts["explainer"] = len(explainer_pairs)
 
+    # Summarizer
     print(f"\n{'='*60}\nSUMMARIZER — generating {pairs_per_task} pairs\n{'='*60}")
-    summarizer_pairs = generate_summarizer_pairs(
+    synthetic_results = generate_synthetic_query_results(config, count=pairs_per_task)
+    summarizer_pairs = generate_pairs(
         config=config, seed_data=synthetic_results,
-        count=pairs_per_task, outfile=pairs_dir / "summarizer.jsonl",
+        count=pairs_per_task, task="summarizer", outfile=pairs_dir / "summarizer.jsonl",
         system_prompt=system_prompt,
     )
+    all_counts["summarizer"] = len(summarizer_pairs)
 
     print(f"\n{'='*60}\nDISTILLATION COMPLETE\n{'='*60}")
-    print(f"  Explainer: {len(explainer_pairs)} pairs")
-    print(f"  Summarizer: {len(summarizer_pairs)} pairs")
+    for task_name, count in all_counts.items():
+        print(f"  {task_name.capitalize()}: {count} pairs")
     _print_cost_summary()
 
 

training/eval.py

@@ -31,6 +31,13 @@
     "related_intervention",
 }
 
+_NARRATOR_REQUIRED_KEYS: set[str] = {
+    "narrative",
+    "key_drivers",
+    "recommended_actions",
+    "data_limitations",
+}
+
 _SUMMARIZER_REQUIRED_KEYS: set[str] = {
     "summary",
     "key_insights",
@@ -44,6 +51,12 @@
 # ---------------------------------------------------------------------------
 
 SHIP_CRITERIA: dict[str, dict[str, float]] = {
+    "narrator": {
+        "json_validity": 0.95,
+        "schema_adherence": 0.90,
+        "shap_grounding": 0.80,
+        "caveat_inclusion": 0.85,
+    },
     "explainer": {
         "json_validity": 0.95,
         "schema_adherence": 0.90,
@@ -120,9 +133,11 @@ def check_schema_adherence(outputs: list[str], task: str) -> float:
     """Fraction of valid JSON outputs that contain all required keys."""
     if not outputs:
         return 0.0
-    required = (
-        _EXPLAINER_REQUIRED_KEYS if task == "explainer" else _SUMMARIZER_REQUIRED_KEYS
-    )
+    required = {
+        "narrator": _NARRATOR_REQUIRED_KEYS,
+        "explainer": _EXPLAINER_REQUIRED_KEYS,
+        "summarizer": _SUMMARIZER_REQUIRED_KEYS,
+    }.get(task, _SUMMARIZER_REQUIRED_KEYS)
     passing = 0
     total = 0
     for text in outputs:
@@ -147,7 +162,7 @@ def check_caveat_inclusion(outputs: list[str], task: str) -> float:
     """
     if not outputs:
         return 0.0
-    caveat_key = "data_limitations" if task == "explainer" else "caveats"
+    caveat_key = "caveats" if task == "summarizer" else "data_limitations"
     passing = 0
     total = 0
     for text in outputs:
@@ -169,6 +184,51 @@ def check_caveat_inclusion(outputs: list[str], task: str) -> float:
     return passing / total if total else 0.0
 
 
+def check_shap_grounding(outputs: list[str], inputs: list[dict[str, Any]], min_features: int = 2) -> float:
+    """Fraction of narrator outputs that mention at least `min_features` of the top-3 SHAP features.
+
+    Extracts feature names from the input's SHAP data and checks whether the
+    narrative text references them (case-insensitive, underscore-tolerant).
+    """
+    if not outputs:
+        return 0.0
+    passing = 0
+    total = 0
+    for output_text, input_data in zip(outputs, inputs):
+        total += 1
+        # Collect top SHAP feature names from all models in the input
+        shap_data = input_data.get("shap", {})
+        top_features: list[str] = []
+        for model_attrs in shap_data.values():
+            for entry in model_attrs.get("top_positive", [])[:3]:
+                top_features.append(entry["feature"])
+            for entry in model_attrs.get("top_negative", [])[:3]:
+                top_features.append(entry["feature"])
+        # Deduplicate while preserving order
+        seen = set()
+        unique_features = []
+        for f in top_features:
+            if f not in seen:
+                seen.add(f)
+                unique_features.append(f)
+        top_features = unique_features[:6]  # top 3 per direction, deduplicated
+
+        if not top_features:
+            passing += 1  # no SHAP data to ground against
+            continue
+
+        # Check how many features appear in the output (case-insensitive, underscores → spaces)
+        output_lower = output_text.lower().replace("_", " ")
+        mentioned = sum(
+            1 for f in top_features
+            if f.lower().replace("_", " ") in output_lower
+        )
+        if mentioned >= min_features:
+            passing += 1
+
+    return passing / total if total else 0.0
+
+
 def check_factual_grounding(outputs: list[str], inputs: list[dict[str, Any]]) -> float:
     """Fraction of outputs that contain numeric values referenced in their input.
 
@@ -314,8 +374,11 @@ def run_eval(school: str, task: str) -> ShipDecision:
         "json_validity": check_json_validity(outputs),
         "schema_adherence": check_schema_adherence(outputs, task),
         "caveat_inclusion": check_caveat_inclusion(outputs, task),
-        "factual_grounding": check_factual_grounding(outputs, inputs),
     }
+    if task == "narrator":
+        metrics["shap_grounding"] = check_shap_grounding(outputs, inputs)
+    else:
+        metrics["factual_grounding"] = check_factual_grounding(outputs, inputs)
 
     print(f"\n[eval] Results for {school}/{task}:")
     for k, v in metrics.items():
@@ -337,13 +400,13 @@ def main() -> None:
     parser.add_argument("--school", required=True, help="School directory name (e.g. bishop-state)")
     parser.add_argument(
         "--task",
-        choices=["explainer", "summarizer"],
+        choices=["narrator", "explainer", "summarizer"],
         default=None,
         help="Task to evaluate (default: both)",
     )
     args = parser.parse_args()
 
-    tasks = [args.task] if args.task else ["explainer", "summarizer"]
+    tasks = [args.task] if args.task else ["narrator", "explainer", "summarizer"]
     results: dict[str, ShipDecision] = {}
     for task in tasks:
         print(f"\n{'='*60}\nEVAL: {task.upper()}\n{'='*60}")

training/prepare.py

@@ -133,7 +133,7 @@ def process_task(school: str, task: str) -> dict[str, int]:
 
 def main(school: str) -> None:
     """Run preparation for all tasks."""
-    for task in ("explainer", "summarizer"):
+    for task in ("narrator", "explainer", "summarizer"):
         try:
             process_task(school, task)
         except FileNotFoundError as e:

training/prompts.py

@@ -26,13 +26,28 @@
     "caveats": ["data limitations relevant to this specific query"],
 }
 
+NARRATOR_SCHEMA = {
+    "narrative": "2-3 sentence explanation grounded in SHAP feature attribution",
+    "key_drivers": ["ranked list of factors with direction and magnitude"],
+    "recommended_actions": ["3-5 specific, actionable interventions"],
+    "data_limitations": ["caveats about the prediction"],
+}
+
 EXPLAINER_STUDENT_SYSTEM = (
     "You are a student success analyst. Given course pairing data, generate a "
     "structured JSON explanation. Include: explanation, structural_factors, "
     "student_impact, advisor_recommendation, data_limitations, and "
     "related_intervention. Respond with ONLY valid JSON."
 )
 
+NARRATOR_STUDENT_SYSTEM = (
+    "You are a student success analyst. Given a student profile with ML prediction "
+    "attribution (SHAP values), generate a structured JSON explanation. Include: "
+    "narrative, key_drivers, recommended_actions, and data_limitations. "
+    "Ground your narrative in the SHAP values — cite specific features by name "
+    "and magnitude. Respond with ONLY valid JSON."
+)
+
 SUMMARIZER_STUDENT_SYSTEM = (
     "You are a student success analyst. Given a query and its results, generate "
     "a structured JSON summary. Include: summary, key_insights, context, "
@@ -195,6 +210,62 @@ def build_system_prompt(config: dict[str, Any]) -> str:
     return "\n\n".join(sections)
 
 
+def build_narrator_prompt(
+    config: dict[str, Any],
+    student_data: dict[str, Any],
+) -> str:
+    """Build the teacher prompt for generating a SHAP-grounded student narrative."""
+    schema_str = json.dumps(NARRATOR_SCHEMA, indent=2)
+    profile = student_data.get("student_profile", {})
+    shap_data = student_data.get("shap", {})
+    risk_factors = student_data.get("risk_factors", [])
+    readiness_score = student_data.get("readiness_score", "N/A")
+    readiness_level = student_data.get("readiness_level", "unknown")
+
+    # Format SHAP attribution section
+    shap_lines = []
+    for model_name, attrs in shap_data.items():
+        shap_lines.append(f"\n  {model_name} model (base prediction: {attrs.get('base_value', 'N/A')}):")
+        for f in attrs.get("top_positive", []):
+            shap_lines.append(f"    + {f['feature']} = {f['value']} (pushes prediction UP by {f['shap_value']})")
+        for f in attrs.get("top_negative", []):
+            shap_lines.append(f"    - {f['feature']} = {f['value']} (pushes prediction DOWN by {abs(f['shap_value'])})")
+
+    profile_str = json.dumps(profile, indent=2, default=str)
+    risk_str = "\n".join(f"- {r}" for r in risk_factors) if risk_factors else "None identified"
+
+    interventions = config.get("school", {}).get("interventions", {}).get("active", [])
+    intervention_lines = []
+    for i in interventions:
+        intervention_lines.append(f"- {i['name']} ({i['type']}): {i.get('effectiveness', 'unknown')}")
+    interventions_str = "\n".join(intervention_lines) if intervention_lines else "None listed"
+
+    return f"""A student at this institution has a readiness score of {readiness_score} ({readiness_level}).
+Analyze their ML prediction factors and write an advisor-facing explanation.
+
+STUDENT PROFILE:
+{profile_str}
+
+RISK FACTORS (rule-engine identified):
+{risk_str}
+
+ML MODEL FEATURE ATTRIBUTION (SHAP values — what drives each prediction):
+{''.join(shap_lines) if shap_lines else 'No SHAP data available'}
+
+AVAILABLE INTERVENTIONS:
+{interventions_str}
+
+Generate a JSON response with this exact schema:
+{schema_str}
+
+Guidelines:
+- Ground the narrative in SHAP values. Cite at least 2 of the top contributing features by name and magnitude.
+- Explain in plain language what each factor means for this student's likelihood of success.
+- Make recommended actions specific to this institution — reference active interventions by name when relevant.
+- Include at least one data limitation or caveat about the prediction.
+- Do NOT speculate beyond what the SHAP values and profile data show."""
+
+
 def build_explainer_prompt(
     config: dict[str, Any],
     course_data: dict[str, Any],