"""Extraction-accuracy + token-efficiency eval on Modal (A10G). Runs MiniCPM-V 4.6 over the labeled synthetic dataset (scripts/make_dataset.py) with TWO prompt variants — the production prompt and a compact one — and scores, per variant: - worry-level accuracy (through guardian's own triage rules + heuristics) - scam recall (scam letters that reached caution/warning) - false-alarm rate (clean letters pushed above "low") - document-type accuracy, amount/deadline hit rates - JSON parse failures, mean latency, output size, prompt size (token economy) Build/test-time only; the deployed runtime stays 100% local. Prepare data, then run: .venv\\Scripts\\python.exe scripts\\make_dataset.py .venv\\Scripts\\modal.exe run modal_eval.py Report lands in modal_artifacts/eval_report.json. """ from __future__ import annotations import json import os import sys import modal sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src")) app = modal.App("mystery-mail-guardian-eval") image = ( modal.Image.debian_slim(python_version="3.12") .pip_install( "transformers[torch]>=5.7.0", "torchvision", "av", "accelerate", "soundfile", "numpy", "Pillow", "json-repair", ) .add_local_dir("src", "/root/src") .add_local_dir(os.path.join("dataset", "eval"), "/root/dataset") ) # The token-economy challenger: same JSON contract, ~60% fewer prompt tokens. COMPACT_PROMPT = """You help an elderly person understand a photographed letter. \ Read it and answer with ONE JSON object only: {"is_document": bool, "readable": bool, "document_type": one of ["utility_bill","bank","government_tax","medical","insurance","subscription","charity","marketing","suspected_scam","personal","other"], "sender": str|null, "what_they_want": str|null, "amount": str|null, "deadline": str|null, "requested_action": str|null, "scam_signals": [{"id": one of [%SIGNALS%], "evidence": short quote}], "explanation": {"what_this_is": "1-2 plain English sentences", "key_facts": [up to 4 short facts with real values from the letter], "worry_reasons": [one short sentence per signal], "what_to_do": [2-4 safe steps]}} Rules: copy real values from the letter; never repeat this schema's words; \ never declare it definitely a scam or definitely safe; no phone numbers, \ links, or emails from the letter in what_to_do; null/[] when absent.""" @app.function(gpu="A10G", image=image, timeout=3600, secrets=[modal.Secret.from_name("huggingface")]) def evaluate(only: str = "") -> dict: import sys import time sys.path.insert(0, "/root/src") from guardian import extract, prompts, triage extract.load() from PIL import Image with open("/root/dataset/labels.jsonl", encoding="utf-8") as f: rows = [json.loads(line) for line in f if line.strip()] variants = { "production": prompts.analysis_prompt("en"), "compact": COMPACT_PROMPT.replace( "%SIGNALS%", ",".join(f'"{s}"' for s in triage.SIGNAL_IDS)), } if only: variants = {only: variants[only]} report: dict = {"n_letters": len(rows), "variants": {}} for vname, prompt in variants.items(): m = {"prompt_chars": len(prompt), "parse_failures": 0, "level_hits": 0, "doc_type_hits": 0, "amount_hits": 0, "amount_total": 0, "deadline_hits": 0, "deadline_total": 0, "scam_total": 0, "scam_caught": 0, "clean_total": 0, "false_alarms": 0, "latency_s": [], "raw_chars": [], "per_case": []} for row in rows: img = Image.open(f"/root/dataset/{row['file']}").convert("RGB") t0 = time.time() raw = extract.run_vision(img, prompt) dt = time.time() - t0 m["latency_s"].append(dt) m["raw_chars"].append(len(raw)) data = triage.parse_model_json(raw) if data is None: m["parse_failures"] += 1 m["per_case"].append({"file": row["file"], "name": row["name"], "level": "PARSE_FAIL"}) continue ex = triage.validate_extraction(data) heuristic_text = " | ".join(filter(None, [ ex.sender, ex.what_they_want, ex.requested_action, ex.what_this_is, *ex.key_facts, *ex.worry_reasons, *[s.evidence for s in ex.scam_signals]])) signals = triage.merge_signals( ex.scam_signals, triage.run_heuristics(heuristic_text)) level = triage.worry_level(ex.document_type, signals) is_scam = bool(row["signal_ids"]) if level in row["acceptable_levels"]: m["level_hits"] += 1 if ex.document_type == row["document_type"]: m["doc_type_hits"] += 1 if is_scam: m["scam_total"] += 1 if level in ("caution", "warning"): m["scam_caught"] += 1 else: m["clean_total"] += 1 if level != "low" and row["name"] != "lookalike_ad": m["false_alarms"] += 1 if row["amount"]: m["amount_total"] += 1 blob = " ".join(filter(None, [ex.amount, *ex.key_facts])) if row["amount"].lstrip("$") in blob: m["amount_hits"] += 1 if row["deadline"]: m["deadline_total"] += 1 blob = " ".join(filter(None, [ex.deadline, *ex.key_facts])) day = row["deadline"].split(",")[0] # "June 28" if day in blob: m["deadline_hits"] += 1 m["per_case"].append({"file": row["file"], "name": row["name"], "gold": row["expected_level"], "level": level, "doc_type": ex.document_type, "seconds": round(dt, 1)}) n = len(rows) report["variants"][vname] = { "prompt_chars": m["prompt_chars"], "parse_failures": m["parse_failures"], "level_accuracy": round(m["level_hits"] / n, 3), "doc_type_accuracy": round(m["doc_type_hits"] / n, 3), "scam_recall": round(m["scam_caught"] / max(m["scam_total"], 1), 3), "false_alarm_rate": round(m["false_alarms"] / max(m["clean_total"], 1), 3), "amount_hit_rate": round(m["amount_hits"] / max(m["amount_total"], 1), 3), "deadline_hit_rate": round(m["deadline_hits"] / max(m["deadline_total"], 1), 3), "mean_latency_s": round(sum(m["latency_s"]) / n, 2), "mean_output_chars": int(sum(m["raw_chars"]) / n), "per_case": m["per_case"], } return report @app.local_entrypoint() def main(only: str = ""): report = evaluate.remote(only) os.makedirs("modal_artifacts", exist_ok=True) with open(os.path.join("modal_artifacts", "eval_report.json"), "w", encoding="utf-8") as f: json.dump(report, f, indent=2, ensure_ascii=False) cols = ["prompt_chars", "parse_failures", "level_accuracy", "doc_type_accuracy", "scam_recall", "false_alarm_rate", "amount_hit_rate", "deadline_hit_rate", "mean_latency_s", "mean_output_chars"] print(f"\n{report['n_letters']} letters per variant\n") header = f"{'metric':<22}" + "".join(f"{v:>14}" for v in report["variants"]) print(header) for c in cols: print(f"{c:<22}" + "".join( f"{report['variants'][v][c]:>14}" for v in report["variants"])) print("\nsaved: modal_artifacts/eval_report.json")