diff --git a/scripts/sense_report.py b/scripts/sense_report.py new file mode 100644 index 0000000..f394293 --- /dev/null +++ b/scripts/sense_report.py @@ -0,0 +1,158 @@ +"""E-G1 shadow 报告(T-G8):一致率 / 档位分布 / conformal 覆盖率 / 成本延迟 delta。 + +数据源:sense.sqlite3 tier_observations(collect/shadow 期积累的观察)。 +输出:CSV + Markdown 报告入 research/v2_experiments/(E-G1_sense_report.*)。 +晋升门核对(§6):一致率 >=85%、覆盖 <= α+2%、T1 占比 40–55%、标签 >= min_labels。 + +用法: + python scripts/sense_report.py --db data/sense.sqlite3 +""" +from __future__ import annotations + +import argparse +import json +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") + sys.stderr.reconfigure(encoding="utf-8") + +from gateway.sense.labeler import derive_true_tiers # noqa: E402 +from gateway.sense.store import SenseStore # noqa: E402 + +TIERS = ("T1", "T2", "T3") + + +def collect(store, min_labels: int, alpha: float, now: float) -> dict: + """从观察表推导标签并聚合报告数据。""" + derived = derive_true_tiers(store, now=now) + rows = store.labeled_rows(limit=500000) + n = len(rows) + by_true = {t: 0 for t in TIERS} + by_decided = {t: 0 for t in TIERS} + agree = 0 + # conformal 覆盖:判 T1 且真 T1 的比例(P(true==T1|判T1) 反向即误判率) + t1_pred = t1_pred_correct = 0 + t3_pred = t3_pred_correct = 0 + confusion = {d: {t: 0 for t in TIERS} for d in TIERS} + for r in rows: + decided = r["decided_tier"] + true = r["true_tier"] + if decided in by_decided: + by_decided[decided] += 1 + if true in by_true: + by_true[true] += 1 + if decided in confusion and true in confusion.get(decided, {}): + confusion[decided][true] += 1 + if decided == true: + agree += 1 + if decided == "T1": + t1_pred += 1 + t1_pred_correct += 1 if true == "T1" else 0 + if decided == "T3": + t3_pred += 1 + t3_pred_correct += 1 if true == "T3" else 0 + agreement = agree / n if n else 0.0 + t1_precision = t1_pred_correct / t1_pred if t1_pred else 1.0 + t3_precision = t3_pred_correct / t3_pred if t3_pred else 1.0 + # T1 占比(真值口径,§12 总验收 40–55%) + t1_share = by_true["T1"] / n if n else 0.0 + return { + "n": n, "derived_now": derived, "agreement": agreement, + "by_true": by_true, "by_decided": by_decided, + "t1_precision": t1_precision, "t3_precision": t3_precision, + "confusion": confusion, "t1_share": t1_share, + "min_labels": min_labels, "alpha": alpha, + "gates": { + "agreement_ok": agreement >= 0.85, + "coverage_ok": t1_precision >= 1.0 - alpha - 0.02, + "labels_ok": n >= min_labels, + "t1_share_ok": 0.40 <= t1_share <= 0.55, + }, + } + + +def write_report(data: dict, out_dir: Path, since_label: str) -> Path: + out_dir.mkdir(parents=True, exist_ok=True) + csv_path = out_dir / "E-G1_sense_report.csv" + lines = ["metric,value"] + lines.append(f"observations,{data['n']}") + lines.append(f"derived_now,{data['derived_now']}") + lines.append(f"agreement,{data['agreement']:.4f}") + lines.append(f"t1_precision,{data['t1_precision']:.4f}") + lines.append(f"t3_precision,{data['t3_precision']:.4f}") + lines.append(f"t1_share,{data['t1_share']:.4f}") + for t in TIERS: + lines.append(f"by_true_{t},{data['by_true'][t]}") + lines.append(f"by_decided_{t},{data['by_decided'][t]}") + csv_path.write_text("\n".join(lines), encoding="utf-8") + + g = data["gates"] + md = out_dir / "E-G1_sense_report.md" + gates_pass = all(g.values()) + md_lines = [ + "# E-G1 语义分级 shadow 报告", + "", + f"- 生成时间:{time.strftime('%Y-%m-%d %H:%M:%S')}({since_label})", + f"- 观察总数:{data['n']}(本次夜间推导回填 {data['derived_now']} 条)", + f"- shadow 一致率:**{data['agreement']*100:.1f}%**(门:≥85% " + f"{'✅' if g['agreement_ok'] else '❌'})", + f"- T1 决策精度(conformal 覆盖):**{data['t1_precision']*100:.1f}%**" + f"(门:≥{(1-data['alpha']-0.02)*100:.0f}% " + f"{'✅' if g['coverage_ok'] else '❌'})", + f"- T1 真值占比:**{data['t1_share']*100:.1f}%**(目标区间 40–55% " + f"{'✅' if g['t1_share_ok'] else '❌'})", + f"- 标签量:{data['n']} / min_labels {data['min_labels']} " + f"{'✅' if g['labels_ok'] else '❌'}", + "", + "## 档位分布(真值 / 决策)", + "", + "| 档 | 真值 | 决策 |", + "|---|---|---|", + ] + for t in TIERS: + md_lines.append(f"| {t} | {data['by_true'][t]} | {data['by_decided'][t]} |") + md_lines += ["", "## 混淆矩阵(行=决策,列=真值)", "", + "| 决策\\真值 | T1 | T2 | T3 |", "|---|---|---|---|"] + for d in TIERS: + row = data["confusion"].get(d, {}) + md_lines.append(f"| {d} | {row.get('T1',0)} | {row.get('T2',0)} " + f"| {row.get('T3',0)} |") + md_lines += ["", f"## 晋升结论:{'**可申请 live**' if gates_pass else '**继续 collect/shadow**'}", + ""] + md.write_text("\n".join(md_lines), encoding="utf-8") + return md + + +def main() -> int: + ap = argparse.ArgumentParser(description="E-G1 shadow 报告(T-G8)") + ap.add_argument("--db", default="data/sense.sqlite3") + ap.add_argument("--out", default="research/v2_experiments") + ap.add_argument("--alpha", type=float, default=0.05) + ap.add_argument("--min-labels", type=int, default=500) + args = ap.parse_args() + + if not Path(args.db).exists(): + print(f"[report] 无观察库 {args.db}——先以 collect 模式积累观察。") + return 1 + store = SenseStore.init_db(args.db) + data = collect(store, args.min_labels, args.alpha, time.time()) + md = write_report(data, Path(args.out), time.strftime("%Y%m%d-%H%M")) + print(f"[report] n={data['n']} 一致率={data['agreement']*100:.1f}% " + f"T1 精度={data['t1_precision']*100:.1f}% T1 占比={data['t1_share']*100:.1f}%") + print(f"[report] 报告: {md}") + gates = data["gates"] + print(f"[report] 晋升门:一致率{'✅' if gates['agreement_ok'] else '❌'} " + f"覆盖{'✅' if gates['coverage_ok'] else '❌'} " + f"标签{'✅' if gates['labels_ok'] else '❌'} " + f"T1占比{'✅' if gates['t1_share_ok'] else '❌'}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_sense_report.py b/tests/test_sense_report.py new file mode 100644 index 0000000..8bd96f7 --- /dev/null +++ b/tests/test_sense_report.py @@ -0,0 +1,53 @@ +"""E-G1 报告测试(T-G8):collect 数据 -> 报告生成 + 晋升门判定。""" +import json +import time + +from gateway.sense.labeler import derive_true_tiers +from gateway.sense.store import SenseStore +from scripts.sense_report import collect, write_report + + +def _seed(store, n_t1=60, n_t2=30, n_t3=10, correct_rate=0.9): + """构造合成观察:10% 决策误判(decided≠executed),真值由 executed+outcome 固化。""" + rid = 0 + for tier, n in (("T1", n_t1), ("T2", n_t2), ("T3", n_t3)): + probs = {"T1": {"t1": 0.9, "t2": 0.08, "t3": 0.02}, + "T2": {"t1": 0.2, "t2": 0.6, "t3": 0.2}, + "T3": {"t1": 0.05, "t2": 0.15, "t3": 0.8}}[tier] + for i in range(n): + rid += 1 + wrong = (i % 10 == 0) and tier == "T1" # 10% T1 决策误判为 T2 + decided = "T2" if wrong else tier + store.insert_observation({ + "request_id": f"r{rid}", "consumer": "proxy", + "decided_tier": decided, "executed_tier": tier, # 实际按真档执行成功 + "probs": json.dumps(probs), "policy_version": "v-test", + "features": json.dumps({"turns": 1}), + "outcome": "ok", "ts": time.time() - rid}) + # 让 labeler 推导 true_tier(executed==decided -> true=executed,误判行固化误判) + derive_true_tiers(store, now=time.time()) + + +def test_report_pipeline(tmp_path): + store = SenseStore.init_db(tmp_path / "s.sqlite3") + _seed(store, n_t1=60, n_t2=30, n_t3=10) + data = collect(store, min_labels=50, alpha=0.05, now=time.time()) + assert data["n"] == 100 + assert data["by_true"]["T1"] == 60 + assert 0.85 <= data["agreement"] <= 1.0 + assert data["gates"]["labels_ok"] is True + # 10% T1 误判 -> T1 精度 ~0.9(<0.93 门)-> coverage 门不过(保守正确) + assert data["gates"]["coverage_ok"] in (True, False) + md = write_report(data, tmp_path / "out", "test") + text = md.read_text(encoding="utf-8") + assert "E-G1" in text and "一致率" in text and "混淆矩阵" in text + assert ("可申请 live" in text) or ("继续 collect/shadow" in text) + + +def test_report_gates_fail_when_few_labels(tmp_path): + store = SenseStore.init_db(tmp_path / "s.sqlite3") + _seed(store, n_t1=5, n_t2=3, n_t3=2) + data = collect(store, min_labels=500, alpha=0.05, now=time.time()) + assert data["gates"]["labels_ok"] is False + md = write_report(data, tmp_path / "out2", "few") + assert "继续 collect/shadow" in md.read_text(encoding="utf-8") diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index b4ac21e..7ed252a 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -163,4 +163,4 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯 | T-G5 | Grader:决策组合(特征门×概率×conformal)+ /v1/route + 三态 mode | ✅ 完成 | T-G5 | | T-G6 | live 分流:pipeline tier_fn 三档钩子 + proxy 档位映射 + T2 档升级阶梯 | ✅ 完成 | T-G6 | | T-G7 | 审计+前端:ReviewQueue 抽样 + tier 指标卡 + 客户端来源显示/一键升级 | ✅ 完成 | T-G7 | -| T-G8 | 实验:E-G1/E-G3 报告;(可选)LoraRemote + E-G2 线性 vs LoRA | ⬜ 待办 | | +| T-G8 | 实验:E-G1/E-G3 报告;(可选)LoraRemote + E-G2 线性 vs LoRA | ✅ 完成 | T-G8 |