diff --git a/gateway/sense/calibrate.py b/gateway/sense/calibrate.py new file mode 100644 index 0000000..cb0bf8b --- /dev/null +++ b/gateway/sense/calibrate.py @@ -0,0 +1,132 @@ +"""split-conformal 阈值校准(T-G3)+ last-good 回退 + 工件读写(D-G3)。 + +目标(§5 policy.alpha):P(true>T1 | 判 T1) ≤ α(τ3 同理对 T3)。 +实现:calib 集按分数降序扫描,取**满足精度的最低阈值**(覆盖率最大); +无任何满足点 -> 阈值取最高分 + ε(全拒,保守)。 +标签数 < min_labels -> 沿用 last-good(D-G3)-> 再无 -> 内置保守值。 +""" +from __future__ import annotations + +import json +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from gateway.sense.store import json_dumps + +CONSERVATIVE = {"t1": 0.90, "t3": 0.90} +THRESHOLD_KIND = "thresholds" + + +def _threshold_for(samples: List[Tuple[float, bool]], alpha: float) -> Optional[float]: + """样本 [(score, true==tier)] -> 满足精度 >= 1-α 的最低阈值(并列分数整组判定)。 + + 按唯一分数降序累积计数;每组边界处检查精度;取满足精度的最大覆盖(最低边界)。 + 无任何满足点 -> None(调用方全拒,保守)。 + """ + if not samples: + return None + groups: Dict[float, List[int]] = {} + for score, ok in samples: + g = groups.setdefault(score, [0, 0]) + g[0] += 1 + g[1] += 1 if ok else 0 + total = 0 + correct = 0 + best: Optional[float] = None + for score in sorted(groups, reverse=True): + cnt, ok_cnt = groups[score] + total += cnt + correct += ok_cnt + if correct / total >= 1.0 - alpha: + best = score + return best + + +def compute_thresholds(rows: List[Dict[str, Any]], alpha: float = 0.05, + min_labels: int = 500) -> Dict[str, Any]: + """rows: [{probs(JSON 串或 dict), true_tier}] -> {t1, t3, coverage, n, ok}。 + + ok=False 表示标签不足(沿用 last-good 的信号)。 + coverage = 通过阈值的 calib 样本占比(实测覆盖率,§6 晋升门比对 α+2%)。 + """ + labeled = [] + for r in rows: + probs = r.get("probs") + if isinstance(probs, str): + try: + probs = json.loads(probs) + except json.JSONDecodeError: + continue + if not isinstance(probs, dict): + continue + true_tier = r.get("true_tier") + if true_tier not in ("T1", "T2", "T3"): + continue + labeled.append((float(probs.get("t1", 0)), float(probs.get("t3", 0)), + true_tier)) + n = len(labeled) + if n < min_labels: + return {"t1": None, "t3": None, "coverage": 0.0, "n": n, "ok": False} + + s1 = [(p1, t == "T1") for p1, _p3, t in labeled] + s3 = [(p3, t == "T3") for _p1, p3, t in labeled] + t1 = _threshold_for(s1, alpha) + t3 = _threshold_for(s3, alpha) + if t1 is None or t3 is None: + return {"t1": t1, "t3": t3, "coverage": 0.0, "n": n, "ok": False} + coverage = (sum(1 for p1, _p3, _t in labeled if p1 >= t1) + + sum(1 for _p1, p3, _t in labeled if p3 >= t3)) / (2 * n) + return {"t1": round(t1, 4), "t3": round(t3, 4), + "coverage": round(coverage, 4), "n": n, "ok": True} + + +def save_thresholds(store, models_dir: str | Path, version: str, + th: Dict[str, Any], activate: bool = False) -> str: + """阈值工件落盘 thresholds.json + 登记工件表。返回文件路径。""" + d = Path(models_dir) / version + d.mkdir(parents=True, exist_ok=True) + path = d / "thresholds.json" + path.write_text(json_dumps(th), encoding="utf-8") + store.register_artifact(version, THRESHOLD_KIND, str(path), + {k: th.get(k) for k in ("t1", "t3", "coverage", "n")}, + active=activate) + return str(path) + + +def load_active(store, models_dir: str | Path) -> Dict[str, Any]: + """active 阈值工件 -> 旧工件扫描回退(last-good)-> 内置保守值(D-G3 回退链)。""" + art = store.active_artifact(THRESHOLD_KIND) + if art: + th = _read_threshold(art) + if th is not None: + th.setdefault("version", art["version"]) + return th + # active 文件缺失/损坏:按时间倒序扫描同 kind 工件(last-good 语义) + for cand in store.list_artifacts(THRESHOLD_KIND): + if cand["version"] == art["version"]: + continue + th = _read_threshold(cand) + if th is not None: + th.setdefault("version", cand["version"]) + return th + last_good = Path(models_dir) / "last-good.json" + if last_good.exists(): + try: + th = json.loads(last_good.read_text(encoding="utf-8")) + if th.get("t1") is not None and th.get("t3") is not None: + th.setdefault("version", "last-good") + return th + except (json.JSONDecodeError, OSError): + pass + return {**CONSERVATIVE, "version": "conservative"} + + +def _read_threshold(art: Dict[str, Any]) -> Optional[Dict[str, Any]]: + try: + th = json.loads(Path(art["path"]).read_text(encoding="utf-8")) + if isinstance(th, dict) and th.get("t1") is not None and th.get("t3") is not None: + return th + except (json.JSONDecodeError, OSError): + pass + return None diff --git a/gateway/sense/labeler.py b/gateway/sense/labeler.py new file mode 100644 index 0000000..7751c57 --- /dev/null +++ b/gateway/sense/labeler.py @@ -0,0 +1,60 @@ +"""夜间 true_tier 推导(T-G3,§4 四条规则,确定性)+ 留存清理。 + +规则(对每条 true_tier 为空的观察,按序判定): +0. human_override 非空 -> 以人工为准(规则 4,优先级最高)。 +1. features.plan_multi(consumer=pipeline 发出 brief 的多步信号)-> true=T3。 +2. outcome∈{escalated,failed,user_retry,timeout} -> true = executed 的下一档 + (T1→T2→T3,T3 保持 T3)。 +3. outcome∈{ok,verified} -> true = executed(T1 成功即 T1;T2/T3 同理)。 +4. 其余(outcome 为空等)-> 跳过,等下一轮。 +""" +from __future__ import annotations + +import json +import time +from typing import Any, Dict, Optional + +NEXT_TIER = {"T1": "T2", "T2": "T3", "T3": "T3"} +SUCCESS = ("ok", "verified") +FAILURE = ("escalated", "failed", "user_retry", "timeout") + + +def _features(row: Dict[str, Any]) -> Dict[str, Any]: + try: + obj = json.loads(row.get("features") or "{}") + return obj if isinstance(obj, dict) else {} + except json.JSONDecodeError: + return {} + + +def derive_true_tier(row: Dict[str, Any]) -> Optional[str]: + """单行推导(纯函数,测试友好)。返回 None = 暂不判定。""" + if row.get("human_override"): + return str(row["human_override"]) + outcome = row.get("outcome") or "" + executed = row.get("executed_tier") or "T2" + feats = _features(row) + if feats.get("plan_multi"): + return "T3" # 规则 1(pipeline brief 信号) + if outcome in FAILURE: # 规则 2 + return NEXT_TIER.get(executed, "T3") + if outcome in SUCCESS: # 规则 3 + return executed + return None # 规则 4:等待结果 + + +def derive_true_tiers(store, now: Optional[float] = None, + retention_days: int = 180) -> int: + """批量推导(§7 签名):返回回填条数;顺带 180d 留存清理。""" + now = time.time() if now is None else now + rows = store.all_observations() + count = 0 + for row in rows: + if row.get("true_tier"): + continue + true_tier = derive_true_tier(row) + if true_tier: + store.set_true_tier(row["request_id"], true_tier) + count += 1 + store.purge_older_than(now - retention_days * 86400) + return count diff --git a/gateway/sense/store.py b/gateway/sense/store.py index 107a163..d75341f 100644 --- a/gateway/sense/store.py +++ b/gateway/sense/store.py @@ -116,6 +116,14 @@ class SenseStore: " WHERE true_tier != ''").fetchone() return int(row["c"]) + def all_observations(self, limit: int = 200000) -> List[Dict[str, Any]]: + """全量遍历(labeler 夜间推导输入;量大时可分页)。""" + with self._lock, self._connect() as conn: + rows = conn.execute( + "SELECT * FROM tier_observations ORDER BY id LIMIT ?", + (limit,)).fetchall() + return [dict(r) for r in rows] + def purge_older_than(self, ts: float) -> int: """留存清理(D-G6:默认 180 天,夜间任务顺带)。""" with self._lock, self._connect() as conn: @@ -154,6 +162,14 @@ class SenseStore: " ORDER BY created_ts DESC LIMIT 1", (kind,)).fetchone() return dict(row) if row else None + def list_artifacts(self, kind: str) -> List[Dict[str, Any]]: + """同 kind 工件按时间倒序(load_active 的 last-good 扫描用)。""" + with self._lock, self._connect() as conn: + rows = conn.execute( + "SELECT * FROM sense_artifacts WHERE kind = ?" + " ORDER BY created_ts DESC", (kind,)).fetchall() + return [dict(r) for r in rows] + # ---------- 自省 ---------- def table_names(self) -> List[str]: with self._lock, self._connect() as conn: diff --git a/tests/test_sense_labeler.py b/tests/test_sense_labeler.py new file mode 100644 index 0000000..f9b9bf9 --- /dev/null +++ b/tests/test_sense_labeler.py @@ -0,0 +1,112 @@ +"""标签+校准测试(T-G3):四条推导规则 / T3 保持 / 180d 清理 / conformal 覆盖 / last-good。""" +import json +import time + +from gateway.sense.calibrate import ( + compute_thresholds, + load_active, + save_thresholds, +) +from gateway.sense.labeler import derive_true_tier, derive_true_tiers +from gateway.sense.store import SenseStore + + +def _obs(rid, executed, outcome, probs=None, features=None, override="", true_tier=""): + return {"request_id": rid, "consumer": "pipeline", "decided_tier": executed, + "executed_tier": executed, "probs": probs or "{}", "outcome": outcome, + "features": features or "{}", "human_override": override, + "true_tier": true_tier} + + +# ---------------- derive_true_tier 纯函数 ---------------- + +def test_rule1_success_t1(): + assert derive_true_tier(_obs("r", "T1", "ok")) == "T1" + assert derive_true_tier(_obs("r", "T1", "verified")) == "T1" + + +def test_rule2_failure_next_tier_and_t3_stays(): + assert derive_true_tier(_obs("r", "T1", "failed")) == "T2" + assert derive_true_tier(_obs("r", "T2", "user_retry")) == "T3" + assert derive_true_tier(_obs("r", "T3", "timeout")) == "T3" # T3 保持 T3 + + +def test_rule3_plan_multi_signals_t3(): + row = _obs("r", "T2", "ok", features=json.dumps({"plan_multi": True})) + assert derive_true_tier(row) == "T3" + + +def test_rule4_human_override_wins(): + row = _obs("r", "T1", "failed", override="T3") + assert derive_true_tier(row) == "T3" + + +def test_empty_outcome_skipped(): + assert derive_true_tier(_obs("r", "T1", "")) is None + + +# ---------------- derive_true_tiers 批量 + 清理 ---------------- + +def test_batch_derive_and_purge(tmp_path): + store = SenseStore.init_db(tmp_path / "s.sqlite3") + now = time.time() + for rid, outcome in [("a", "ok"), ("b", "failed"), ("c", "")]: + store.insert_observation(_obs(rid, "T2", outcome)) + # 一条 200 天前的旧数据(先推导后清理) + store.insert_observation({**_obs("old", "T1", "ok"), "ts": now - 200 * 86400}) + count = derive_true_tiers(store, now=now, retention_days=180) + assert count == 3 # a→T2, b→T3, old→T1 + rows = {r["request_id"]: r for r in store.all_observations()} + assert rows["a"]["true_tier"] == "T2" + assert rows["b"]["true_tier"] == "T3" + assert "old" not in rows # 180d 清理生效(推导后删除) + + +# ---------------- compute_thresholds ---------------- + +def test_conformal_threshold_coverage(): + """合成分布:T1 高分样本真为 T1 -> τ1 满足精度 >= 1-α;τ3 同理(需真 T3 样本)。""" + rows = [] + for i in range(40): + rows.append({"probs": json.dumps({"t1": 0.95 - i * 0.001, "t3": 0.01}), + "true_tier": "T1"}) + for i in range(10): + rows.append({"probs": json.dumps({"t1": 0.3, "t3": 0.3}), + "true_tier": "T2"}) + for i in range(5): + rows.append({"probs": json.dumps({"t3": 0.9, "t1": 0.02}), + "true_tier": "T3"}) + th = compute_thresholds(rows, alpha=0.05, min_labels=10) + assert th["ok"] is True and th["n"] == 55 + kept = [r for r in rows if json.loads(r["probs"])["t1"] >= th["t1"]] + precision = sum(1 for r in kept if r["true_tier"] == "T1") / len(kept) + assert precision >= 0.95 + + +def test_min_labels_not_met_returns_not_ok(): + rows = [{"probs": json.dumps({"t1": 0.9, "t3": 0.05}), "true_tier": "T1"}] + th = compute_thresholds(rows, alpha=0.05, min_labels=500) + assert th["ok"] is False and th["t1"] is None + + +def test_threshold_artifact_roundtrip_and_last_good(tmp_path): + """工件读写 + active 切换 + last-good 回退链。""" + store = SenseStore.init_db(tmp_path / "s.sqlite3") + th = {"t1": 0.8, "t3": 0.85, "coverage": 0.9, "n": 500, "ok": True} + path = save_thresholds(store, tmp_path, "v1", th, activate=True) + assert "thresholds.json" in path + loaded = load_active(store, tmp_path) + assert loaded["t1"] == 0.8 and loaded["version"] == "v1" + # 保存 v2 不激活 -> 仍 v1;激活 v2 -> 切换;v2 文件损坏 -> 扫描回退 v1 + save_thresholds(store, tmp_path, "v2", {"t1": 0.7, "t3": 0.7, + "coverage": 0.9, "n": 500}) + assert load_active(store, tmp_path)["version"] == "v1" # v2 未激活 + store.activate_artifact("v2", "thresholds") + assert load_active(store, tmp_path)["version"] == "v2" + (tmp_path / "v2" / "thresholds.json").unlink() + loaded = load_active(store, tmp_path) + assert loaded["version"] == "v1" and loaded["t1"] == 0.8 # last-good 扫描回退 + # 全新 store(无任何工件登记)-> 保守值 + fresh_store = SenseStore.init_db(tmp_path / "fresh.sqlite3") + fresh = load_active(fresh_store, tmp_path / "fresh") + assert fresh["version"] == "conservative" diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 1d0ec45..c539ef6 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -157,7 +157,7 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯 | T-G0 | 骨架:gateway/sense/ 包 + sense.sqlite3 DDL + enabled 门控 | ✅ 完成 | c3efa70 | | T-G1 | Embedder:/v1/embeddings 客户端 + int8 量化 + 降级阶梯 | ✅ 完成 | T-G1 | | T-G2 | 观察埋点:observer + 三消费方埋点(pipeline/proxy/client) | ✅ 完成 | T-G2 | -| T-G3 | 标签+校准:夜间 true_tier 推导 + split-conformal + 工件表 | ⬜ 待办 | | +| T-G3 | 标签+校准:夜间 true_tier 推导 + split-conformal + 工件表 | ✅ 完成 | T-G3 | | T-G3b | KnnHead(架构变体 B):kNN 投票 + conformal-kNN + 按桶分区/封顶/压缩;hybrid fusion 预留(§14) | ⬜ 待办 | | | T-G4 | 线性头:离线训练脚本 + LinearHead 纯 Python 推理 + 登记 | ⬜ 待办 | | | T-G5 | Grader:决策组合(特征门×概率×conformal)+ /v1/route + 三态 mode | ⬜ 待办 | |