From 76fd636fae32345c96b744c68ded14353eb89792 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E5=AE=97=E5=B9=B3?= Date: Tue, 30 Jun 2026 07:24:55 +0000 Subject: [PATCH] feat(m4/m5): T4.2 instruct-check + QA judge table in eval report MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - eval/instruct_check.py (T4.2): per-category (6 types) parse_rate + rule_acc + stub-judge mean + usability; pass bars anomaly-JSON>80%, overall usability>70%. - eval/report.py: wire eval_qa_task across all categories so the report now covers design §6.5 问答均分 (QA judge 0-5 per category per predictor), not just anomaly-detection metrics. Fix latent bug: describe rule judge expects a dict ref, so parse the JSON ref_answer before passing. - Verified trivial-only smoke renders both tables; all_report VUS-PR=0.26 (not inflated) vs PA-F1~1.0 confirms PA is control-only. --- src/tsmm/eval/instruct_check.py | 213 ++++++++++++++++++++++++++++++++ src/tsmm/eval/report.py | 52 +++++++- 2 files changed, 261 insertions(+), 4 deletions(-) create mode 100644 src/tsmm/eval/instruct_check.py diff --git a/src/tsmm/eval/instruct_check.py b/src/tsmm/eval/instruct_check.py new file mode 100644 index 0000000..708f0b4 --- /dev/null +++ b/src/tsmm/eval/instruct_check.py @@ -0,0 +1,213 @@ +"""Instruction-following check (T4.2). + +Samples N held-out items per task category from ``eval_synth.jsonl`` and runs the +trained model, then reports per-category: + +- **parse_rate**: for the structured categories (anomaly / describe / forecast) + the fraction of answers whose JSON parses; for free-text categories + (root_cause / event / compare) the fraction of non-empty answers. +- **rule_acc**: rule-judge score where the category is rule-checkable. +- **judge_mean**: stub LLM-judge 0-5 mean (token-overlap heuristic, offline). + +Design T4.2 pass bars (§4.3 M4 出口验证): +- anomaly JSON parse-success rate > 80% +- 6-category overall usability (parse or non-empty + judge>0) > 70% + +Heavy (GPU, model.generate) — imports the wrapper lazily. +""" +from __future__ import annotations + +import argparse +import json +import os +import random +from collections import defaultdict +from typing import Dict, List + +import numpy as np + + +CATEGORIES = ["anomaly", "describe", "forecast", "root_cause", "event", "compare"] + + +def load_by_category(path: str, per_cat: int, seed: int = 0) -> Dict[str, List[dict]]: + buckets: Dict[str, List[dict]] = defaultdict(list) + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + s = json.loads(line) + buckets[s.get("category", "?")].append(s) + rng = random.Random(seed) + out = {} + for cat in CATEGORIES: + pool = buckets.get(cat, []) + rng.shuffle(pool) + out[cat] = pool[:per_cat] + return out + + +def _build_model(stage1_ckpt=None, stage2_ckpt=None): + import torch + from ..model.ts_encoder import TSEncoder + from ..model.projector import Projector + from ..model.wrapper import MultimodalTSModel + from ..data.collator import Collator + + llm = os.environ.get("TSMM_LLM_PATH", "/home/zhangzp/models/Qwen2.5-0.5B-Instruct") + m = MultimodalTSModel(llm, TSEncoder(), Projector(), dtype=torch.bfloat16).cuda() + ckpt = stage2_ckpt or stage1_ckpt + if ckpt and os.path.exists(ckpt): + st = torch.load(ckpt, map_location="cpu") + m.encoder.load_state_dict(st["encoder"]) + m.projector.load_state_dict(st["projector"]) + print(f"[instruct_check] loaded {ckpt}") + lora_dir = os.path.join(os.path.dirname(ckpt), "lora_adapter") + if stage2_ckpt and os.path.exists(lora_dir): + from peft import PeftModel + m.llm = PeftModel.from_pretrained(m.llm, lora_dir) + print(f"[instruct_check] loaded lora {lora_dir}") + m.eval() + return m, Collator(max_T=512) + + +def _generate(m, col, sample) -> str: + import torch + batch = col([sample]) + with torch.no_grad(): + txt = m.generate(batch, max_new_tokens=96) + return txt[0] if txt else "" + + +def evaluate(samples_by_cat, model, col, judge_backend="stub") -> Dict[str, Dict]: + from .parse_answer import parse_anomaly_segments + from .qa_judge import RuleJudge, LLMJudge, _extract_json + + rule_j = RuleJudge() + llm_j = LLMJudge(backend=judge_backend) + report: Dict[str, Dict] = {} + + for cat, samples in samples_by_cat.items(): + if not samples: + continue + n = len(samples) + parsed = 0 # structured-parse success OR non-empty prose + nonempty = 0 + rule_scores: List[float] = [] + judge_scores: List[float] = [] + examples: List[Dict] = [] + + for s in samples: + pred = _generate(model, col, s) + ref = s.get("answer", "") + instr = s.get("instruction") or s.get("question") or "" + + ok = False + if cat == "anomaly": + _, ok = parse_anomaly_segments(pred) + elif cat in ("describe", "forecast"): + ok = _extract_json(pred) is not None + else: # free-text + ok = bool(pred.strip()) + if ok: + parsed += 1 + if pred.strip(): + nonempty += 1 + + r = rule_j.judge( + cat, pred, + ref_segs=[(seg.get("start", 0), seg.get("end", 0)) + for seg in s.get("segments", [])], + T=len(s.get("labels", [])) or 1, + ref_answer=ref if cat == "describe" else "", + ref_forecast=_forecast_list(ref) if cat == "forecast" else [], + ) + if r.get("score") is not None: + rule_scores.append(r["score"]) + + j = llm_j.score(instr, ref, pred) + judge_scores.append(j["score"]) + + if len(examples) < 2: + examples.append({ + "pred": (pred or "")[:160], + "ref": (ref or "")[:160], + "parsed": ok, + }) + + # usability = parsed (structured-or-nonempty) AND judge thinks there's + # some content (judge>0). Conservative per-design ">70% usable". + usable = sum(1 for sc in judge_scores if sc > 0) + report[cat] = { + "n": n, + "parse_rate": parsed / n, + "nonempty_rate": nonempty / n, + "rule_acc": float(np.mean(rule_scores)) if rule_scores else None, + "judge_mean": float(np.mean(judge_scores)), + "usability": usable / n, + "examples": examples, + } + return report + + +def _forecast_list(ref_answer): + """Extract the forecast value list from a forecast ref answer JSON.""" + try: + obj = json.loads(ref_answer) if isinstance(ref_answer, str) else ref_answer + fc = obj.get("forecast") + if isinstance(fc, list) and fc: + inner = fc[0] + if isinstance(inner, list): + return [float(x) for x in inner] + return [float(x) for x in fc] + except (ValueError, TypeError): + pass + return [] + + +def run(args: argparse.Namespace) -> Dict: + samples_by_cat = load_by_category(args.data, args.per_cat, args.seed) + model, col = _build_model(args.stage1_ckpt, args.stage2_ckpt) + rep = evaluate(samples_by_cat, model, col, args.judge_backend) + + # headline summary + anomaly_parse = rep.get("anomaly", {}).get("parse_rate", 0.0) + overall_usability = float(np.mean([r["usability"] for r in rep.values()])) if rep else 0.0 + summary = { + "anomaly_json_parse_rate": anomaly_parse, + "anomaly_json_pass": anomaly_parse > 0.80, + "overall_usability": overall_usability, + "overall_usability_pass": overall_usability > 0.70, + } + print("\n=== T4.2 instruction-following summary ===") + print(json.dumps(summary, indent=2, ensure_ascii=False)) + print("\n=== per-category ===") + for cat, r in rep.items(): + print(f"[{cat}] n={r['n']} parse={r['parse_rate']:.2f} " + f"rule={r['rule_acc']} judge={r['judge_mean']:.2f} " + f"usable={r['usability']:.2f}") + for ex in r["examples"]: + print(f" pred={ex['pred']!r}") + return {"summary": summary, "per_category": rep} + + +def main() -> None: + p = argparse.ArgumentParser(description="Instruction-following check (T4.2)") + p.add_argument("--data", default="data/eval_synth.jsonl") + p.add_argument("--per_cat", type=int, default=10) + p.add_argument("--seed", type=int, default=0) + p.add_argument("--stage1_ckpt", default="checkpoints/stage1/stage1_final.pt") + p.add_argument("--stage2_ckpt", default="checkpoints/stage2/stage2_final.pt") + p.add_argument("--judge_backend", default="stub") + p.add_argument("--out", default="reports/instruct_check.json") + args = p.parse_args() + result = run(args) + os.makedirs(os.path.dirname(args.out) or ".", exist_ok=True) + with open(args.out, "w", encoding="utf-8") as f: + json.dump(result, f, indent=2, ensure_ascii=False) + print(f"\n[instruct_check] wrote {args.out}") + + +if __name__ == "__main__": + main() diff --git a/src/tsmm/eval/report.py b/src/tsmm/eval/report.py index 7436b70..80d4e20 100644 --- a/src/tsmm/eval/report.py +++ b/src/tsmm/eval/report.py @@ -100,11 +100,20 @@ def eval_qa_task(samples, predictor, judge_backend="stub") -> Dict[str, Any]: cat = s.get("category", "root_cause") pred = predictor(s) ref = s.get("answer", "") + # describe expects a dict ref; parse JSON if possible, else {} + ref_for_describe = ref + if cat == "describe": + try: + ref_for_describe = json.loads(ref) if isinstance(ref, str) else (ref or {}) + if not isinstance(ref_for_describe, dict): + ref_for_describe = {} + except (ValueError, TypeError): + ref_for_describe = {} r = rule_judge.judge(cat, pred, ref_segs=[(seg.get("start", 0), seg.get("end", 0)) for seg in s.get("segments", [])], T=len(s.get("labels", [])) or 1, - ref_answer=ref) + ref_answer=ref_for_describe) if r.get("parsed"): parse_ok += 1 if r.get("score") is not None: @@ -189,17 +198,25 @@ def run_eval(args: argparse.Namespace) -> str: # run anomaly eval per dataset × predictor results: Dict[str, Dict[str, Any]] = defaultdict(dict) + qa_results: Dict[str, Dict[str, Dict[str, Any]]] = defaultdict(lambda: defaultdict(dict)) for ds_name, path in datasets: samples = list(stream_jsonl(path)) - anom = [s for s in samples if s.get("category") == "anomaly"] + by_cat: Dict[str, List[dict]] = defaultdict(list) + for s in samples: + by_cat[s.get("category", "?")].append(s) + anom = by_cat.get("anomaly", []) for pname, pred in predictors.items(): results[ds_name][pname] = eval_anomaly_task(anom, pred) + # QA judge across all categories (covers design §6.5 问答均分) + for cat, cat_samples in by_cat.items(): + qa_results[ds_name][pname][cat] = eval_qa_task( + cat_samples, pred, judge_backend=args.judge_backend) # render markdown - return _render_report(results, datasets, predictors.keys(), args) + return _render_report(results, qa_results, datasets, predictors.keys(), args) -def _render_report(results, datasets, predictor_names, args) -> str: +def _render_report(results, qa_results, datasets, predictor_names, args) -> str: today = date.today().isoformat() lines = [ f"# Eval Report — {today}", @@ -237,6 +254,33 @@ def _render_report(results, datasets, predictor_names, args) -> str: lines.append("_PA = point-adjust; shown as control only, not headline._") lines.append("") + # QA judge table (covers design §6.5 success criterion 问答均分) + qa_ds = qa_results.get(ds_name, {}) + if qa_ds: + cats = sorted({c for pm in qa_ds.values() for c in pm}) + lines.append(f"### QA judge (stub 0-5) — {ds_name}") + lines.append("") + header = ["predictor"] + cats + ["mean"] + lines.append("| " + " | ".join(header) + " |") + lines.append("|" + "|".join(["---"] * len(header)) + "|") + for pname in predictor_names: + pm = qa_ds.get(pname, {}) + if not pm: + continue + row_vals = [] + for c in cats: + jm = pm.get(c, {}).get("judge_mean") + row_vals.append(f"{jm:.2f}" if jm is not None else "-") + mean_jm = float(np.mean([pm[c]["judge_mean"] + for c in cats + if pm.get(c, {}).get("judge_mean") is not None])) if any(pm.get(c) for c in cats) else 0.0 + row = [pname] + row_vals + [f"{mean_jm:.2f}"] + lines.append("| " + " | ".join(row) + " |") + lines.append("") + lines.append("_QA judge = offline stub token-overlap heuristic; " + "swap `--judge_backend openai` for GPT-4o scoring._") + lines.append("") + out_path = os.path.join(args.report_dir, f"eval-{today}.md") with open(out_path, "w", encoding="utf-8") as f: f.write("\n".join(lines))