Spaces:
Sleeping
Sleeping
| """Weekly digest assembly and Resend delivery.""" | |
| from __future__ import annotations | |
| import json | |
| import urllib.error | |
| import urllib.request | |
| from collections import Counter, defaultdict | |
| from pathlib import Path | |
| from typing import Any | |
| from scripts.lib.chat_eval import analytics | |
| from scripts.lib.chat_eval.report import build_report | |
| from scripts.lib.chat_eval.turn_log import iter_turns | |
| _RESEND_EMAILS_URL = "https://api.resend.com/emails" | |
| # A chat is classified as probe traffic (regression-suite) if its opener | |
| # appears this many times or more across the window. Deterministic test | |
| # batteries fire the same opener repeatedly; real students don't. | |
| # This is more robust than a turn-count threshold — a real student whose | |
| # bot crashed on turn 1 still gets counted as a real chat. | |
| _PROBE_OPENER_REPEATS = 3 | |
| # Topics that are catalog scaffolding, not real economics concepts. | |
| # Skip them from "what students struggled with" since their inclusion | |
| # would be operationally meaningful but pedagogically meaningless. | |
| _META_TOPICS = {"m_g_r_e_c_o_n780__fall1", "not_topic", "t_e_m_p_l_a_t_e", ""} | |
| # Dimension labels for human-facing output. snake_case is fine for tables, | |
| # but the TL;DR sentence needs friendlier copy. | |
| _DIMENSION_LABELS = { | |
| "substance": "substance", | |
| "voice": "faculty voice", | |
| "rigor": "rigor", | |
| "pedagogy_fit": "pedagogy fit", | |
| "citation_discipline": "citation discipline", | |
| } | |
| _EMAIL_CSS = """ | |
| body { font-family: -apple-system, "Segoe UI", Helvetica, Arial, sans-serif; | |
| color: #1d1d1f; line-height: 1.55; max-width: 720px; margin: 0 auto; | |
| padding: 1.5rem; } | |
| h1 { font-size: 1.6rem; margin-bottom: 0.5rem; } | |
| h2 { font-size: 1.25rem; margin-top: 2rem; padding-bottom: 0.3rem; | |
| border-bottom: 1px solid #e1e1e1; } | |
| h3 { font-size: 1.05rem; margin-top: 1.3rem; } | |
| table { border-collapse: collapse; margin: 0.75rem 0; width: 100%; } | |
| th, td { text-align: left; border-bottom: 1px solid #e1e1e1; | |
| padding: 0.4rem 0.6rem; font-size: 0.92em; vertical-align: top; } | |
| th { color: #555; font-weight: 600; } | |
| code { background: #f4f4f6; padding: 0.1em 0.35em; border-radius: 3px; | |
| font-family: ui-monospace, Menlo, Consolas, monospace; font-size: 0.92em; } | |
| blockquote { border-left: 3px solid #00539b; color: #555; margin: 0.75rem 0; | |
| padding: 0.1rem 0 0.1rem 1rem; } | |
| """ | |
| def render_html(markdown_str: str) -> str: | |
| """Render a markdown document to a self-contained HTML email body.""" | |
| import markdown as _markdown | |
| body = _markdown.markdown(markdown_str, extensions=["tables"]) | |
| return ( | |
| "<!DOCTYPE html>\n" | |
| "<html><head><meta charset=\"utf-8\">" | |
| f"<style>{_EMAIL_CSS}</style></head>" | |
| f"<body>{body}</body></html>" | |
| ) | |
| def send_via_resend( | |
| *, | |
| api_key: str, | |
| sender: str, | |
| recipients: list[str], | |
| subject: str, | |
| html_body: str, | |
| ) -> None: | |
| """POST one email through the Resend API. Raises on non-2xx.""" | |
| payload = json.dumps({ | |
| "from": sender, | |
| "to": recipients, | |
| "subject": subject, | |
| "html": html_body, | |
| }).encode("utf-8") | |
| req = urllib.request.Request( | |
| _RESEND_EMAILS_URL, | |
| data=payload, | |
| method="POST", | |
| headers={ | |
| "Authorization": f"Bearer {api_key}", | |
| "Content-Type": "application/json", | |
| "User-Agent": "me-hub-weekly-digest/1.0", | |
| }, | |
| ) | |
| try: | |
| with urllib.request.urlopen(req, timeout=30) as resp: | |
| if resp.status >= 300: | |
| body = resp.read().decode("utf-8", "replace") | |
| raise RuntimeError(f"Resend returned {resp.status}: {body}") | |
| except urllib.error.HTTPError as e: | |
| try: | |
| body = e.read().decode("utf-8", "replace") | |
| except Exception: | |
| body = "" | |
| raise RuntimeError(f"Resend returned {e.code}: {body}") from e | |
| def faculty_digest( | |
| turns_log_path: Path, | |
| tuning_log_path: Path, | |
| *, | |
| since: str, | |
| today: str, | |
| window_days: int = 7, | |
| ) -> str: | |
| """Build the full faculty-forward weekly digest markdown body. | |
| Sections in priority order: TL;DR, what students struggled with, did the | |
| bot help, one thing to change, cost, operational. Filters probe traffic | |
| out of student-facing sections; surfaces it under operational signals. | |
| """ | |
| all_turns = list(iter_turns(turns_log_path, since=since)) | |
| chats = _split_into_chats(all_turns) | |
| real_chats, probe_chats = _classify_chats(chats) | |
| real_turns = [t for c in real_chats for t in c] | |
| real_judged = [t for t in real_turns if (t.get("judge") or {}).get("scores")] | |
| parts: list[str] = [] | |
| parts.append(f"# Weekly briefing — MGRECON780 chat tutor") | |
| parts.append(f"_{today} · last {window_days} days_") | |
| parts.append("") | |
| parts.extend(_section_tldr(real_chats, real_turns, real_judged, probe_chats)) | |
| parts.extend(_section_struggles(real_turns, real_judged)) | |
| parts.extend(_section_bot_value(real_judged, real_turns)) | |
| parts.extend(_section_one_change(real_judged, real_turns, turns_log_path, since)) | |
| parts.extend(_section_cost(turns_log_path, since)) | |
| parts.extend(_section_operational( | |
| all_turns, real_chats, probe_chats, tuning_log_path, since, | |
| )) | |
| parts.extend(_section_full_detail( | |
| turns_log_path, tuning_log_path, since=since, window_days=window_days, | |
| )) | |
| return "\n".join(parts) | |
| def _section_full_detail( | |
| turns_log_path: Path, | |
| tuning_log_path: Path, | |
| *, | |
| since: str, | |
| window_days: int, | |
| ) -> list[str]: | |
| """Append the full v1 report below the faculty briefing. Same content as | |
| the pre-redesign email — tuning loop, all analytics tables, chat sessions — | |
| for operators or faculty who want to drill into the underlying data. | |
| """ | |
| parts = ["", "---", "", "# Full report", ""] | |
| parts.append(_tuning_loop_summary(tuning_log_path, since=since)) | |
| parts.append(build_report( | |
| turns_log_path, | |
| since=since[:10] if since else None, | |
| title=f"Chat feedback detail (last {window_days} days)", | |
| heading_level=2, | |
| )) | |
| return parts | |
| def _classify_chats( | |
| chats: list[list[dict]], | |
| ) -> tuple[list[list[dict]], list[list[dict]]]: | |
| """Split chats into (real, probe). | |
| A chat is probe traffic if its opener (first user_message) appears | |
| `_PROBE_OPENER_REPEATS` or more times across all chats. Real students | |
| type unique openers; regression suites fire the same battery repeatedly. | |
| """ | |
| opener_freq: Counter[str] = Counter() | |
| for chat in chats: | |
| if chat: | |
| opener_freq[chat[0].get("user_message", "")] += 1 | |
| real, probe = [], [] | |
| for chat in chats: | |
| if not chat: | |
| continue | |
| opener = chat[0].get("user_message", "") | |
| if opener_freq[opener] >= _PROBE_OPENER_REPEATS: | |
| probe.append(chat) | |
| else: | |
| real.append(chat) | |
| return real, probe | |
| def _split_into_chats(turns: list[dict]) -> list[list[dict]]: | |
| """Group turns into chats. New chat when session_id changes or | |
| turn_index_in_session resets to 0.""" | |
| chats: list[list[dict]] = [] | |
| current: list[dict] = [] | |
| last_sid = None | |
| for turn in sorted(turns, key=lambda t: t.get("ts", "")): | |
| sid = turn.get("session_id", "") | |
| idx = turn.get("turn_index_in_session", 0) | |
| if not current or sid != last_sid or idx == 0: | |
| if current: | |
| chats.append(current) | |
| current = [turn] | |
| else: | |
| current.append(turn) | |
| last_sid = sid | |
| if current: | |
| chats.append(current) | |
| return chats | |
| def _section_tldr( | |
| real_chats: list[list[dict]], | |
| real_turns: list[dict], | |
| real_judged: list[dict], | |
| probe_chats: list[list[dict]], | |
| ) -> list[str]: | |
| parts = ["## TL;DR", ""] | |
| if not real_turns: | |
| parts.append( | |
| "No real student traffic this week. " | |
| f"{len(probe_chats)} regression-suite probe(s) ran against the bot." | |
| ) | |
| parts.append("") | |
| return parts | |
| score_map = _mean_scores(real_judged) | |
| lowest = ( | |
| min(score_map.items(), key=lambda kv: kv[1]) | |
| if score_map else (None, None) | |
| ) | |
| highest = ( | |
| max(score_map.items(), key=lambda kv: kv[1]) | |
| if score_map else (None, None) | |
| ) | |
| missed_concepts = _top_misconceptions_missed(real_judged, top_n=1) | |
| miss_phrase = ( | |
| f"`{missed_concepts[0][0]}` slipped past the bot {missed_concepts[0][1]['missed']}x" | |
| if missed_concepts else "no misconceptions slipped past the bot" | |
| ) | |
| no_response = _no_response_turns(real_turns) | |
| no_response_note = ( | |
| f" ⚠ **{len(no_response)} student turn(s) got no response from the bot.**" | |
| if no_response else "" | |
| ) | |
| parts.append( | |
| f"**{len(real_chats)} student chats** ({len(real_turns)} turns, " | |
| f"{len(real_judged)} graded). " | |
| f"Bot was strongest on **{_DIMENSION_LABELS.get(highest[0], highest[0])}** " | |
| f"({highest[1]:.1f}/5) and weakest on " | |
| f"**{_DIMENSION_LABELS.get(lowest[0], lowest[0])}** ({lowest[1]:.1f}/5); " | |
| f"{miss_phrase}.{no_response_note}" | |
| ) | |
| parts.append("") | |
| return parts | |
| def _section_struggles( | |
| real_turns: list[dict], | |
| real_judged: list[dict], | |
| ) -> list[str]: | |
| parts = ["## 1. Where students struggled", ""] | |
| if not real_turns: | |
| parts.append("_No real student traffic this week._") | |
| parts.append("") | |
| return parts | |
| # Top stuck concepts: syntopicon topics that appeared in a turn whose | |
| # learning_trajectory was "clarification" (judge flagged that the | |
| # student needed the bot to re-explain). Each (topic, turn) pair | |
| # counts once — so a single chat doesn't inflate the count via 10 | |
| # retrievals to the same topic. | |
| stuck_topic_counts: Counter[str] = Counter() | |
| for turn in real_judged: | |
| judge = turn.get("judge") or {} | |
| if judge.get("learning_trajectory") != "clarification": | |
| continue | |
| seen_in_turn: set[str] = set() | |
| for r in turn.get("retrieved", []): | |
| topic = r.get("topic") or "" | |
| if not topic or topic in _META_TOPICS or topic.isdigit(): | |
| continue | |
| if topic in seen_in_turn: | |
| continue | |
| stuck_topic_counts[topic] += 1 | |
| seen_in_turn.add(topic) | |
| if stuck_topic_counts: | |
| parts.append("**Concepts where students needed re-explanation** " | |
| "(turns flagged as `clarification` by the judge):") | |
| parts.append("") | |
| parts.append("| Topic | Re-explained in turns |") | |
| parts.append("|---|---|") | |
| for topic, n in stuck_topic_counts.most_common(5): | |
| parts.append(f"| {topic.replace('_', ' ')} | {n} |") | |
| parts.append("") | |
| # Misconceptions slipping through (bot didn't correct what student showed). | |
| missed = _top_misconceptions_missed(real_judged, top_n=5) | |
| if missed: | |
| parts.append("**Misconceptions students exhibited that the bot didn't address:**") | |
| parts.append("") | |
| parts.append("| Misconception | Times missed | Times handled |") | |
| parts.append("|---|---|---|") | |
| for mid, stats in missed: | |
| parts.append(f"| `{mid}` | {stats['missed']} | {stats['handled']} |") | |
| parts.append("") | |
| # Representative confused-student questions: real-student turns with | |
| # is_follow_up=True or rigor<=2. Show 3. | |
| confused = _representative_confused_questions(real_turns, real_judged, top_n=3) | |
| if confused: | |
| parts.append("**Where students got stuck (in their own words):**") | |
| parts.append("") | |
| for q in confused: | |
| parts.append(f"> {q}") | |
| parts.append("") | |
| if not stuck_topic_counts and not missed and not confused: | |
| parts.append("_No clear struggle patterns this week — students mostly " | |
| "got what they asked for on the first try._") | |
| parts.append("") | |
| return parts | |
| def _section_bot_value( | |
| real_judged: list[dict], | |
| real_turns: list[dict], | |
| ) -> list[str]: | |
| parts = ["## 2. Did the bot help?", ""] | |
| if not real_judged: | |
| parts.append("_No judged turns yet for real student chats._") | |
| parts.append("") | |
| return parts | |
| # Mean scores overall + bar visualization. | |
| score_map = _mean_scores(real_judged) | |
| parts.append("**Quality scores** (1–5, averaged over judged turns):") | |
| parts.append("") | |
| parts.append("| Dimension | Score | |") | |
| parts.append("|---|---|---|") | |
| for dim in ["substance", "voice", "rigor", "pedagogy_fit", "citation_discipline"]: | |
| if dim not in score_map: | |
| continue | |
| v = score_map[dim] | |
| bar = _score_bar(v) | |
| label = _DIMENSION_LABELS.get(dim, dim) | |
| parts.append(f"| {label} | {v:.2f} | {bar} |") | |
| parts.append("") | |
| # Score by intent — does the bot answer "explain" well but flunk "debug"? | |
| # Drop `<error>` (planner failure) since it's a system event, not an intent. | |
| by_intent = { | |
| i: s for i, s in _scores_by_intent(real_judged).items() | |
| if i != "<error>" | |
| } | |
| if by_intent: | |
| parts.append("**By student intent** (mean across all 5 dimensions, n=count of judged turns):") | |
| parts.append("") | |
| parts.append("| Intent | n | Mean |") | |
| parts.append("|---|---|---|") | |
| for intent, stats in sorted(by_intent.items(), key=lambda kv: -kv[1]["n"]): | |
| parts.append(f"| {intent} | {stats['n']} | {stats['mean']:.2f} |") | |
| parts.append("") | |
| # Pedagogical opportunities taken. | |
| taken = sum(1 for t in real_judged | |
| if (t.get("judge") or {}).get("pedagogical_opportunity_taken") is True) | |
| missed_opps = sum(1 for t in real_judged | |
| if (t.get("judge") or {}).get("pedagogical_opportunity_taken") is False) | |
| total_opps = taken + missed_opps | |
| if total_opps: | |
| pct = 100.0 * taken / total_opps | |
| parts.append(f"**Teachable moments:** the bot acted on **{taken}/{total_opps}** " | |
| f"({pct:.0f}%) — i.e., {missed_opps} times the judge flagged a " | |
| "moment where the bot could have pushed the student further but didn't.") | |
| parts.append("") | |
| # Bot-side production bug: empty response. Surface separately from voice | |
| # failures since this isn't a quality issue but a "the bot crashed" issue. | |
| no_response = _no_response_turns(real_turns) | |
| if no_response: | |
| parts.append( | |
| f"**⚠ No response generated** for {len(no_response)} student turn(s) — " | |
| "the chat handler returned an empty reply. Investigate the chat " | |
| "orchestrator logs for these turn IDs:" | |
| ) | |
| parts.append("") | |
| for t in no_response[:5]: | |
| opener = _flatten(t.get("user_message") or "", 80) | |
| parts.append(f"- `{t.get('turn_id', '?')}` — \"{opener}\"") | |
| parts.append("") | |
| # Voice failures — show up to 3 with full diagnosis. Newlines in | |
| # response/diagnosis are flattened so the markdown blockquote stays | |
| # intact (otherwise embedded bullets break out of the quote). | |
| vf = _voice_failures_from_turns(real_turns) | |
| if vf: | |
| parts.append(f"**Voice failures** ({len(vf)} turns scored ≤2 on voice — " | |
| "these read as generic AI, not as your faculty):") | |
| parts.append("") | |
| parts.append("| Student asked | Bot replied | Judge diagnosis |") | |
| parts.append("|---|---|---|") | |
| for v in vf[:3]: | |
| q = _flatten(v["user_message"], 80) | |
| r = _flatten(v["response"], 140) or "_(no response generated)_" | |
| d = _flatten(v["diagnosis"], 200) | |
| parts.append(f"| {q} | {r} | {d} |") | |
| parts.append("") | |
| return parts | |
| def _section_one_change( | |
| real_judged: list[dict], | |
| real_turns: list[dict], | |
| turns_log_path: Path, | |
| since: str, | |
| ) -> list[str]: | |
| """Rank-ordered, opinionated recommendations. Top 3 max.""" | |
| parts = ["## 3. If you change one thing this week", ""] | |
| recs: list[tuple[int, str]] = [] # (priority, markdown line) | |
| # No-response bug always wins — students typed something and got nothing. | |
| no_response = _no_response_turns(real_turns) | |
| if no_response: | |
| recs.append(( | |
| 200, | |
| f"**Fix the no-response bug.** {len(no_response)} student turn(s) got " | |
| "an empty reply this week — students walked away with nothing. " | |
| "This is a production bug, not a quality issue; check the chat " | |
| "orchestrator's error path.", | |
| )) | |
| # Misconceptions slipping through — highest priority if missed >= 2 | |
| missed = _top_misconceptions_missed(real_judged, top_n=3) | |
| for mid, stats in missed: | |
| if stats["missed"] >= 2: | |
| recs.append(( | |
| 100 + stats["missed"], | |
| f"**Tighten the prompt's `{mid}` correction.** Students " | |
| f"showed this misconception {stats['missed']} time(s) and the " | |
| f"bot let it pass. Add an explicit displacement rule in the " | |
| "prompt's misconception section.", | |
| )) | |
| # Intent-classifier mismatches: turns where the judge flagged | |
| # debug-intent answers being treated as Socratic explanation. Use the | |
| # judge's own diagnosis text rather than guessing from message length. | |
| debug_voice_misses = [ | |
| t for t in real_judged | |
| if t.get("intent") == "debug" | |
| and ((t.get("judge") or {}).get("scores") or {}).get("voice", 5) <= 2 | |
| ] | |
| if len(debug_voice_misses) >= 2: | |
| recs.append(( | |
| 90, | |
| f"**Fix the debug-mode behavior.** {len(debug_voice_misses)} debug-intent " | |
| "turn(s) drew voice ≤2 — students typed an answer (e.g., `C,M`, `68`) " | |
| "and the bot ran Socratic exploration instead of confirming or " | |
| "correcting. Tighten the debug-mode section of the chat prompt to " | |
| "default to direct confirmation when the student is submitting an answer.", | |
| )) | |
| # Doc-pruning opportunity — high retrieval, zero util_pct on real-student turns | |
| util = analytics.doc_utilization(real_turns) if real_turns else {} | |
| dead_docs = [ | |
| (doc_id, s) for doc_id, s in util.items() | |
| if s["retrieved_count"] >= 20 and s["util_pct"] == 0.0 | |
| ] | |
| if dead_docs: | |
| top_dead = sorted(dead_docs, key=lambda kv: -kv[1]["retrieved_count"])[:3] | |
| names = ", ".join(f"`{d[0]}`" for d in top_dead) | |
| recs.append(( | |
| 70, | |
| f"**Prune dead retrievals.** {len(dead_docs)} doc(s) were retrieved " | |
| "20+ times in real student chats but the bot used 0% of them. " | |
| f"Top offenders: {names}. These dilute the prompt and burn tokens.", | |
| )) | |
| # Faculty framing fidelity if mostly 1-2 | |
| fid = [ | |
| v for v in ( | |
| (t.get("judge") or {}).get("faculty_framing_fidelity") | |
| for t in real_judged | |
| ) | |
| if isinstance(v, int) | |
| ] | |
| if fid and sum(1 for v in fid if v <= 2) / len(fid) >= 0.30: | |
| recs.append(( | |
| 60, | |
| f"**Lopomo/Vaidya voice isn't carrying.** {sum(1 for v in fid if v <= 2)}/" | |
| f"{len(fid)} graded turns scored faculty-framing fidelity ≤2/5. " | |
| "Faculty may want to add 1-2 signature phrasings to the voice " | |
| "section of the chat-tutor prompt.", | |
| )) | |
| if not recs: | |
| parts.append( | |
| "_No high-confidence recommendation this week — sample is small " | |
| "or quality is steady. Re-check next Monday._" | |
| ) | |
| parts.append("") | |
| return parts | |
| for i, (_, line) in enumerate(sorted(recs, key=lambda r: -r[0])[:3], 1): | |
| parts.append(f"{i}. {line}") | |
| parts.append("") | |
| return parts | |
| def _section_cost(turns_log_path: Path, since: str) -> list[str]: | |
| """Compact cost section. Aggregate spend + the cache-prefix savings | |
| opportunity if it's non-trivial.""" | |
| parts = ["## 4. Cost", ""] | |
| cost = analytics.cost_summary(turns_log_path, since=since) | |
| if cost["n_turns"] == 0: | |
| parts.append("_No turns this window._") | |
| parts.append("") | |
| return parts | |
| parts.append( | |
| f"**${cost['total_cost_usd']:.2f}** across **{cost['n_turns']} turns** " | |
| f"(mean **${cost['mean_cost_usd']:.3f}**/turn)." | |
| ) | |
| parts.append("") | |
| cache = analytics.cache_summary(turns_log_path, since=since) | |
| if cache.get("n_turns_with_data", 0) > 0: | |
| hit = cache["cache_hit_rate"] * 100 | |
| savings = cache["estimated_savings_usd"] | |
| if savings < 0: | |
| parts.append( | |
| f"_Caching is currently a **net loss of ${-savings:.2f}** " | |
| f"at {hit:.1f}% hit rate — the cache writes (1.25x premium) " | |
| "are outweighing the reads. Configure a permanent prefix " | |
| "to flip this number._" | |
| ) | |
| parts.append("") | |
| else: | |
| parts.append(f"_Cache savings vs no-cache: **${savings:.2f}** " | |
| f"at {hit:.1f}% hit rate._") | |
| parts.append("") | |
| rec = analytics.cache_recommendation(turns_log_path, since=since, ttl="1h", top_n=15) | |
| if rec.get("recommended"): | |
| parts.append( | |
| f"**Caching opportunity:** pinning the top {len(rec['recommended'])} " | |
| f"retrieved docs into a 1-hour prefix would save approximately " | |
| f"**${rec['total_recommended_savings_usd']:.2f}** at current usage. " | |
| "(One-line operator task.)" | |
| ) | |
| parts.append("") | |
| return parts | |
| def _section_operational( | |
| all_turns: list[dict], | |
| real_chats: list[list[dict]], | |
| probe_chats: list[list[dict]], | |
| tuning_log_path: Path, | |
| since: str, | |
| ) -> list[str]: | |
| parts = ["## 5. Operational", ""] | |
| parts.append( | |
| f"- **Traffic:** {len(real_chats)} student chats " | |
| f"({sum(len(c) for c in real_chats)} turns) + " | |
| f"{len(probe_chats)} probe / regression-test chats " | |
| f"({sum(len(c) for c in probe_chats)} turns). " | |
| "Student-facing sections above exclude probe traffic." | |
| ) | |
| tuning_entries = list(iter_turns(tuning_log_path, since=since)) | |
| if tuning_entries: | |
| accepted = sum(1 for e in tuning_entries if e.get("outcome") == "accepted") | |
| frozen = sum(1 for e in tuning_entries if e.get("outcome") == "rolled_back") | |
| parts.append( | |
| f"- **Tuning loop:** {len(tuning_entries)} daily run(s), " | |
| f"{accepted} penalty applied, {frozen} freeze(s)." | |
| ) | |
| else: | |
| parts.append( | |
| "- **Tuning loop:** no daily-tuning entries in this window. " | |
| "If digest is running from the cron (which calls `fetch-state` first), " | |
| "investigate the `tune-retrieval` workflow." | |
| ) | |
| error_intents = sum(1 for t in all_turns if t.get("intent") == "<error>") | |
| if error_intents: | |
| parts.append( | |
| f"- **Errors:** {error_intents} turn(s) recorded `<error>` intent " | |
| "(planner failure — investigate the planner logs)." | |
| ) | |
| parts.append("") | |
| return parts | |
| def _mean_scores(judged: list[dict]) -> dict[str, float]: | |
| sums: dict[str, float] = defaultdict(float) | |
| counts: dict[str, int] = defaultdict(int) | |
| for t in judged: | |
| for k, v in ((t.get("judge") or {}).get("scores") or {}).items(): | |
| if isinstance(v, (int, float)): | |
| sums[k] += v | |
| counts[k] += 1 | |
| return {k: sums[k] / counts[k] for k in sums if counts[k]} | |
| def _scores_by_intent(judged: list[dict]) -> dict[str, dict[str, Any]]: | |
| sums: dict[str, float] = defaultdict(float) | |
| counts: dict[str, int] = defaultdict(int) | |
| for t in judged: | |
| intent = t.get("intent") or "unknown" | |
| scores = ((t.get("judge") or {}).get("scores") or {}).values() | |
| nums = [v for v in scores if isinstance(v, (int, float))] | |
| if not nums: | |
| continue | |
| sums[intent] += sum(nums) / len(nums) | |
| counts[intent] += 1 | |
| return { | |
| i: {"n": counts[i], "mean": sums[i] / counts[i]} | |
| for i in counts | |
| } | |
| def _top_misconceptions_missed( | |
| judged: list[dict], *, top_n: int = 5, | |
| ) -> list[tuple[str, dict[str, int]]]: | |
| counts: dict[str, dict[str, int]] = defaultdict( | |
| lambda: {"missed": 0, "handled": 0} | |
| ) | |
| for t in judged: | |
| for flag in ((t.get("judge") or {}).get("misconception_flags") or []): | |
| mid = flag.get("id", "") | |
| if not mid: | |
| continue | |
| in_q = bool(flag.get("in_query")) | |
| handled = bool(flag.get("handled_in_response")) | |
| if in_q and not handled: | |
| counts[mid]["missed"] += 1 | |
| if handled: | |
| counts[mid]["handled"] += 1 | |
| ranked = sorted( | |
| counts.items(), | |
| key=lambda kv: (-kv[1]["missed"], -kv[1]["handled"]), | |
| ) | |
| return [(mid, s) for mid, s in ranked if s["missed"] > 0][:top_n] | |
| def _representative_confused_questions( | |
| real_turns: list[dict], | |
| real_judged: list[dict], | |
| *, | |
| top_n: int = 3, | |
| ) -> list[str]: | |
| """Real student questions where the bot's answer scored low on rigor | |
| (≤3) — these are the moments the judge flagged the response as not | |
| rigorous enough for the question being asked. Distinct messages only.""" | |
| candidates = [ | |
| t for t in real_judged | |
| if ((t.get("judge") or {}).get("scores") or {}).get("rigor", 5) <= 3 | |
| ] | |
| candidates.sort( | |
| key=lambda t: ((t.get("judge") or {}).get("scores") or {}).get("rigor", 5), | |
| ) | |
| seen_msgs: set[str] = set() | |
| out: list[str] = [] | |
| for t in candidates: | |
| msg = (t.get("user_message") or "").strip() | |
| if not msg or len(msg) <= 4 or msg in seen_msgs: | |
| continue | |
| seen_msgs.add(msg) | |
| out.append(msg[:160] + ("…" if len(msg) > 160 else "")) | |
| if len(out) >= top_n: | |
| break | |
| return out | |
| def _no_response_turns(turns: list[dict]) -> list[dict]: | |
| """Turns where the bot generated no response at all — production bug. | |
| Detected by empty response field on a turn whose user_message was non-empty.""" | |
| return [ | |
| t for t in turns | |
| if (t.get("user_message") or "").strip() | |
| and not (t.get("response") or "").strip() | |
| ] | |
| def _flatten(text: str, max_len: int) -> str: | |
| """Collapse whitespace, escape pipes (markdown table cells), truncate. | |
| Used to safely embed user/bot text inside markdown table cells.""" | |
| if not text: | |
| return "" | |
| flat = " ".join(text.split()).replace("|", "\\|") | |
| if len(flat) > max_len: | |
| return flat[:max_len].rstrip() + "…" | |
| return flat | |
| def _voice_failures_from_turns( | |
| real_turns: list[dict], *, threshold: int = 2, | |
| ) -> list[dict[str, Any]]: | |
| out = [] | |
| for t in real_turns: | |
| judge = t.get("judge") or {} | |
| if not judge.get("scores"): | |
| continue | |
| if judge["scores"].get("voice", 5) > threshold: | |
| continue | |
| out.append({ | |
| "turn_id": t.get("turn_id"), | |
| "user_message": t.get("user_message", ""), | |
| "response": t.get("response", ""), | |
| "voice_score": judge["scores"]["voice"], | |
| "diagnosis": judge.get("diagnosis", ""), | |
| }) | |
| return out | |
| def _score_bar(value: float, *, width: int = 10) -> str: | |
| """Inline visual bar: filled blocks for the score, empty for the rest. | |
| Renders fine in HTML email — Unicode block chars survive most clients.""" | |
| filled = max(0, min(width, round(value / 5.0 * width))) | |
| return "█" * filled + "░" * (width - filled) | |
| # --------------------------------------------------------------------------- | |
| # Tuning-loop section for the appended full-detail report. Mirrors the v1 | |
| # layout — daily-run headline, freeze banner, penalty changes table, | |
| # catalog drift footnote, identity footer. | |
| def _tuning_loop_summary(tuning_log_path: Path, *, since: str) -> str: | |
| entries = list(iter_turns(tuning_log_path, since=since)) | |
| if not entries: | |
| return _tuning_empty_section() | |
| return _tuning_render_section(entries) | |
| def _tuning_empty_section() -> str: | |
| return ( | |
| "## Tuning loop this week\n" | |
| "\n" | |
| "Tuning loop did not run this week - investigate the tune-retrieval workflow.\n" | |
| "\n" | |
| ) | |
| def _tuning_render_section(entries: list[dict]) -> str: | |
| entries = sorted(entries, key=lambda e: e.get("ts", "")) | |
| runs = len(entries) | |
| accepted = [e for e in entries if e.get("outcome") == "accepted"] | |
| freezes = [e for e in entries if e.get("outcome") == "rolled_back"] | |
| quiet = [ | |
| e for e in entries | |
| if e.get("outcome") in ("skipped_no_eligible_doc", "skipped_insufficient_turns") | |
| ] | |
| latest = entries[-1] | |
| parts: list[str] = ["## Tuning loop this week", ""] | |
| headline_bits = [_pluralize(runs, "daily run", "daily runs")] | |
| if accepted: | |
| headline_bits.append(_pluralize(len(accepted), "penalty applied", "penalties applied")) | |
| if freezes: | |
| headline_bits.append(_pluralize(len(freezes), "freeze", "freezes")) | |
| if quiet: | |
| headline_bits.append(_pluralize(len(quiet), "quiet day", "quiet days")) | |
| parts.append(" | ".join(headline_bits)) | |
| parts.append("") | |
| if latest.get("outcome") in ("skipped_frozen", "rolled_back"): | |
| reason = latest.get("frozen_reason") or _freeze_reason_from_rollback(latest) | |
| parts.append( | |
| f"> **Tuning loop is currently frozen** ({reason}). " | |
| "To unfreeze: `python -m scripts.lib.chat_eval.cli tune --unfreeze`." | |
| ) | |
| parts.append("") | |
| if accepted: | |
| parts.extend(_tuning_accepted_table(accepted)) | |
| if freezes: | |
| parts.append("### Freezes") | |
| parts.append("") | |
| for entry in freezes: | |
| parts.append(f"- {_freeze_line(entry)}") | |
| parts.append("") | |
| cleanup_union = sorted({ | |
| doc_id | |
| for entry in entries | |
| for doc_id in (entry.get("cleanup_dropped_doc_ids") or []) | |
| }) | |
| if cleanup_union: | |
| ids_str = ", ".join(f"`{doc_id}`" for doc_id in cleanup_union) | |
| parts.append(f"_Catalog drift: dropped stale penalties for {ids_str}._") | |
| parts.append("") | |
| parts.append("---") | |
| parts.append("") | |
| parts.append( | |
| f"_rubric_version: {latest.get('rubric_version', '?')} | " | |
| f"judge: {latest.get('judge_model', '?')} | " | |
| f"chat: {latest.get('chat_model', '?')} | " | |
| f"embed: {latest.get('embedding_model', '?')} | " | |
| f"catalog: {latest.get('catalog_artifact_version_id', '?')}_" | |
| ) | |
| parts.append("") | |
| return "\n".join(parts) | |
| def _tuning_accepted_table(accepted: list[dict]) -> list[str]: | |
| parts = [ | |
| "### Penalty changes", | |
| "", | |
| "| Date | doc_id | kind | penalty before -> after | retrievals | util % | recent |", | |
| "|---|---|---|---|---|---|---|", | |
| ] | |
| for entry in accepted: | |
| action = entry.get("step_action") or {} | |
| date = (entry.get("ts") or "")[:10] | |
| doc_id = action.get("doc_id", "?") | |
| kind = action.get("kind", "?") | |
| before = action.get("penalty_before", "?") | |
| after = action.get("penalty_after", "?") | |
| retrieved = action.get("retrieved_count", "?") | |
| util_pct = action.get("util_pct", "?") | |
| history = action.get("recent_util_history") or [] | |
| sparkline = _tuning_sparkline([h.get("util_pct", 0.0) for h in history]) | |
| parts.append( | |
| f"| {date} | `{doc_id}` | {kind} | {before} -> {after} | " | |
| f"{retrieved} | {util_pct} | {sparkline} |" | |
| ) | |
| parts.append("") | |
| return parts | |
| def _pluralize(n: int, singular: str, plural: str) -> str: | |
| return f"{n} {singular if n == 1 else plural}" | |
| _TUNING_SPARKLINE_BLOCKS = "▁▂▃▄▅▆▇█" | |
| def _tuning_sparkline(values: list[float]) -> str: | |
| if not values: | |
| return "" | |
| out = [] | |
| for value in values: | |
| idx = min(int((float(value) / 100.0) * len(_TUNING_SPARKLINE_BLOCKS)), | |
| len(_TUNING_SPARKLINE_BLOCKS) - 1) | |
| out.append(_TUNING_SPARKLINE_BLOCKS[max(idx, 0)]) | |
| return "".join(out) | |
| def _drop_summary(entry: dict) -> tuple[str, object]: | |
| rc = entry.get("regression_check") or {} | |
| return rc.get("max_drop_dimension") or "?", rc.get("max_drop_value", "?") | |
| def _freeze_reason_from_rollback(entry: dict) -> str: | |
| dim, drop = _drop_summary(entry) | |
| return f"{dim} regressed delta={drop}" | |
| def _freeze_line(entry: dict) -> str: | |
| ts = (entry.get("ts") or "")[:10] | |
| dim, drop = _drop_summary(entry) | |
| confidence = (entry.get("regression_check") or {}).get("confidence") or {} | |
| n = confidence.get("n_after", "?") | |
| t_stat = confidence.get("max_drop_t_stat") | |
| extra = f", t={t_stat:.2f}" if isinstance(t_stat, (int, float)) else "" | |
| return f"{ts} - frozen on rolled_back: {dim} delta={drop} (n={n}{extra})" | |