diff --git a/gateway/sense/classifier.py b/gateway/sense/classifier.py new file mode 100644 index 0000000..0737afa --- /dev/null +++ b/gateway/sense/classifier.py @@ -0,0 +1,72 @@ +"""分级头(T-G4):LinearHead 纯 Python 推理 + LoraRemote 预留(T-G8)。 + +D-G5:serving 路径零新依赖——线性头推理是纯 Python 点积 + softmax(768 维 +≈0.1ms)。工件为 JSON(weights 3×dim / bias 3 / dim / version),由 +scripts/train_tier_head.py(numpy 离线训练)产出。 +缺失/损坏 -> ArtifactMissing(D-G4 降级信号)。 +""" +from __future__ import annotations + +import json +import math +from pathlib import Path +from typing import Any, Dict, List, Optional + +from gateway.sense.errors import ArtifactMissing + +TIERS = ("t1", "t2", "t3") + + +class LinearHead: + """线性有序三分类头(softmax;类别序 t1 "LinearHead": + p = Path(path) + if not p.exists(): + raise ArtifactMissing(f"线性头工件不存在: {p}") + try: + data = json.loads(p.read_text(encoding="utf-8")) + weights = data["weights"] + bias = data["bias"] + if not weights or not bias: + raise ValueError("空权重") + return cls(weights=weights, bias=bias, + version=str(data.get("version") or version or p.parent.name), + labels=data.get("labels")) + except ArtifactMissing: + raise + except Exception as e: # noqa: BLE001 + raise ArtifactMissing(f"线性头工件损坏: {type(e).__name__}: {e}") from e + + def predict(self, vec: List[float]) -> Dict[str, float]: + """点积 + softmax -> {t1, t2, t3} 概率(和为 1)。""" + logits = [] + for w, b in zip(self.weights, self.bias): + n = min(len(w), len(vec)) + logits.append(sum(wi * vi for wi, vi in zip(w[:n], vec[:n])) + b) + m = max(logits) + exps = [math.exp(z - m) for z in logits] + total = sum(exps) + return {label: e / total for label, e in zip(self.labels, exps)} + + +class LoraRemote: + """LoRA 远程分类(vLLM /v1/classify)——T-G8 可选实验,本版预留。""" + + def __init__(self, base_url: str, model: str, api_key: Optional[str] = None): + self.base_url = base_url.rstrip("/") + self.model = model + self.api_key = api_key + + def predict(self, vec: List[float]) -> Dict[str, float]: + raise NotImplementedError("T-G8 可选实验(LoRA/vLLM 分类服务)") diff --git a/requirements-ml.txt b/requirements-ml.txt index 1708756..6f037fe 100644 --- a/requirements-ml.txt +++ b/requirements-ml.txt @@ -1,11 +1,15 @@ -# 可选:接入真实开源小模型(RTX 4060 8GB 可跑 0.5B-4B 量化模型) -# 安装:pip install -r requirements-ml.txt -torch>=2.2 -transformers>=4.40 -accelerate>=0.30 -peft>=0.11 -sentencepiece -protobuf -bitsandbytes -# 轻量推理引擎(可选) -# llama-cpp-python +# 可选:接入真实开源小模型(RTX 4060 8GB 可跑 0.5B-4B 量化模型) +# 安装:pip install -r requirements-ml.txt +torch>=2.2 +transformers>=4.40 +accelerate>=0.30 +peft>=0.11 +sentencepiece +protobuf +bitsandbytes +# 轻量推理引擎(可选) +# llama-cpp-python + +# 语义分析器离线训练(T-G4):仅 scripts/train_tier_head.py 使用; +# serving 路径零新依赖(D-G5:线性头推理纯 Python 点积) +numpy>=1.26 diff --git a/scripts/train_tier_head.py b/scripts/train_tier_head.py new file mode 100644 index 0000000..20e9336 --- /dev/null +++ b/scripts/train_tier_head.py @@ -0,0 +1,185 @@ +"""离线训练三分类线性头(T-G4):softmax 回归(numpy),产出 JSON 工件。 + +用法: + python scripts/train_tier_head.py --db data/sense.sqlite3 --version v1 + python scripts/train_tier_head.py --synthetic # 合成数据自测(无需真实标签) + +输入:sense.sqlite3 tier_observations 中 true_tier 非空且 embedding(int8 BLOB) +可解析的行;按时间 70/15/15 切分 train/val/calib。 +输出:data/sense_models/{version}/head.json + metrics.json + 工件登记(active=0, +人工 /sense/admin/promote 切换)。 + +依赖:numpy(可选,requirements-ml.txt);serving 推理不依赖 numpy(D-G5)。 +""" +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") + +try: + import numpy as np +except ImportError: + print("缺少 numpy:pip install numpy(或 requirements-ml.txt);训练必需,serving 不需要。") + sys.exit(1) + +from gateway.sense.store import SenseStore # noqa: E402 + +TIERS = ["t1", "t2", "t3"] + + +def _deblob(blob) -> list[float]: + return [b - 256 if b > 127 else b for b in blob] + + +def load_rows(db_path: str): + """true_tier 非空且 embedding 可解析的行 -> (X, y, versions)。""" + store = SenseStore.init_db(db_path) + X, y, vers = [], [], [] + for row in store.labeled_rows(limit=200000): + blob = row["embedding"] + if not blob: + continue + X.append(_deblob(blob)) + y.append(TIERS.index(row["true_tier"].lower())) + vers.append(row.get("policy_version") or "") + return np.array(X, dtype=float), np.array(y), vers + + +def split(n: int, seed: int = 42): + idx = np.random.default_rng(seed).permutation(n) + t = int(n * 0.7) + v = int(n * 0.85) + return idx[:t], idx[t:v], idx[v:] + + +def softmax(z): + z = z - z.max(axis=1, keepdims=True) + e = np.exp(z) + return e / e.sum(axis=1, keepdims=True) + + +def train(X, y, epochs=300, lr=0.5, seed: int = 42): + """L2 正则 softmax 回归(全量梯度下降,类别权重均衡)。""" + n, d = X.shape + k = 3 + Y = np.zeros((n, k)) + for i, c in enumerate(y): + Y[i, c] = 1.0 + cls_weight = n / (k * np.array([(y == c).sum() or 1 for c in range(k)])) + W = np.zeros((d, k)) + b = np.zeros(k) + sw = np.array([cls_weight[c] for c in y]) + sw = sw / sw.sum() * n + rng = np.random.default_rng(seed) + for _ in range(epochs): + probs = softmax(X @ W + b) + grad = (probs - Y) * sw[:, None] + W -= lr * (X.T @ grad / n + 0.01 * W) + b -= lr * (grad.sum(axis=0) / n) + return W, b + + +def metrics(X, y, W, b) -> dict: + probs = softmax(X @ W + b) + pred = probs.argmax(axis=1) + macro_f1s = [] + for c in range(3): + tp = float(((pred == c) & (y == c)).sum()) + fp = float(((pred == c) & (y != c)).sum()) + fn = float(((pred != c) & (y == c)).sum()) + precision = tp / (tp + fp) if tp + fp else 0.0 + recall = tp / (tp + fn) if tp + fn else 0.0 + f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0 + macro_f1s.append(f1) + acc = float((pred == y).mean()) + # ordinal AUC 简化:P(判>=c 与 真>=c 一致) 均值 + return {"accuracy": round(acc, 4), + "macro_f1": round(sum(macro_f1s) / 3, 4), + "per_tier_f1": [round(f, 4) for f in macro_f1s]} + + +def run(db_path: str, version: str, models_dir: str, epochs: int) -> int: + store = SenseStore.init_db(db_path) + X, y, _ = load_rows(db_path) + if len(X) < 30: + print(f"[train] 标签样本不足({len(X)} < 30),跳过训练。" + "先以 collect 模式积累标签(夜间推导)。") + return 1 + tr, va, ca = split(len(X)) + W, b = train(X[tr], y[tr], epochs=epochs) + m_val = metrics(X[va], y[va], W, b) + probs_cal = (softmax(X[ca] @ W + b) * 1000).round().astype(int) / 1000 + calib_rows = [{"probs": json.dumps({TIERS[i]: float(probs_cal[j, i]) + for i in range(3)}), + "true_tier": TIERS[y[ca][j]]} for j in range(len(y[ca]))] + + out = Path(models_dir) / version + out.mkdir(parents=True, exist_ok=True) + (out / "head.json").write_text(json.dumps({ + "version": version, "dim": int(X.shape[1]), "labels": TIERS, + "weights": W.T.tolist(), "bias": b.tolist()}, ensure_ascii=False), encoding="utf-8") + metrics_data = {"version": version, "n": int(len(X)), "val": m_val, + "created": time.strftime("%Y-%m-%d %H:%M:%S")} + (out / "metrics.json").write_text(json.dumps(metrics_data, ensure_ascii=False, + indent=2), encoding="utf-8") + + # calib 阈值(用 T-G3 校准逻辑) + from gateway.sense.calibrate import compute_thresholds, save_thresholds + th = compute_thresholds(calib_rows, alpha=0.05, min_labels=5) + save_thresholds(store, models_dir, version, th, activate=False) + + store.register_artifact(version, "head", str(out / "head.json"), + metrics_data, active=False) + print(f"[train] version={version} n={len(X)} val={m_val}") + print(f"[train] 阈值={th}") + print(f"[train] 工件已登记(active=0),/sense/admin/promote 切换") + return 0 + + +def run_synthetic(models_dir: str) -> int: + """合成数据自测:三类可分高斯簇,验证训练-工件-LinearHead 推理链路。""" + rng = np.random.default_rng(7) + n, d = 300, 64 + centers = [rng.normal(0, 1, d), rng.normal(4, 1, d), rng.normal(-4, 1, d)] + X = np.vstack([c + rng.normal(0, 0.8, (n, d)) for c in centers]) + y = np.repeat([0, 1, 2], n) + W, b = train(X, y, epochs=200) + m = metrics(X, y, W, b) + assert m["macro_f1"] > 0.9, f"合成集 macro_F1 过低: {m}" + version = "synthetic" + out = Path(models_dir) / version + out.mkdir(parents=True, exist_ok=True) + (out / "head.json").write_text(json.dumps({ + "version": version, "dim": d, "labels": TIERS, + "weights": W.T.tolist(), "bias": b.tolist()}), encoding="utf-8") + print(f"[synthetic] macro_f1={m['macro_f1']} accuracy={m['accuracy']}" + f" -> {out/'head.json'}") + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description="训练三分类线性头(T-G4)") + ap.add_argument("--db", default="data/sense.sqlite3") + ap.add_argument("--version", default=time.strftime("v%Y%m%d")) + ap.add_argument("--models-dir", default="data/sense_models") + ap.add_argument("--epochs", type=int, default=300) + ap.add_argument("--synthetic", action="store_true", + help="合成数据自测训练链路(不需要真实标签)") + args = ap.parse_args() + if args.synthetic: + return run_synthetic(args.models_dir) + return run(args.db, args.version, args.models_dir, args.epochs) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_sense_classifier.py b/tests/test_sense_classifier.py new file mode 100644 index 0000000..9a19cd0 --- /dev/null +++ b/tests/test_sense_classifier.py @@ -0,0 +1,83 @@ +"""线性头测试(T-G4):纯 Python 推理与黄金向量一致 / 工件缺失 ArtifactMissing。""" +import json +import math +import tempfile +from pathlib import Path + +import pytest + +from gateway.sense.classifier import LinearHead +from gateway.sense.errors import ArtifactMissing + +W = [[0.5, -0.5, 0.0], [0.0, 0.5, -0.5], [-0.5, 0.0, 0.5]] +B = [0.1, 0.0, -0.1] + + +def _head(tmp_path): + p = tmp_path / "head.json" + p.write_text(json.dumps({"version": "test", "dim": 3, "labels": ["t1", "t2", "t3"], + "weights": W, "bias": B}), encoding="utf-8") + return p + + +def test_predict_matches_reference_softmax(tmp_path): + """纯 Python 点积 + softmax 与手工参考实现一致(黄金向量)。""" + head = LinearHead.load(_head(tmp_path)) + vec = [1.0, 2.0, 3.0] + logits = [sum(wi * vi for wi, vi in zip(w, vec)) + b for w, b in zip(W, B)] + m = max(logits) + exps = [math.exp(z - m) for z in logits] + total = sum(exps) + expected = {"t1": exps[0] / total, "t2": exps[1] / total, "t3": exps[2] / total} + probs = head.predict(vec) + assert sum(probs.values()) == pytest.approx(1.0, abs=1e-9) + for k, v in expected.items(): + assert probs[k] == pytest.approx(v, rel=1e-9) + + +def test_predict_argmax_consistent_with_logit_order(tmp_path): + head = LinearHead.load(_head(tmp_path)) + # [3,1,2] 点积最大维度是 0 -> t1 概率最大 + probs = head.predict([3.0, 1.0, 2.0]) + assert max(probs, key=probs.get) == "t1" + + +def test_load_missing_raises_artifact_missing(tmp_path): + with pytest.raises(ArtifactMissing): + LinearHead.load(tmp_path / "ghost.json") + + +def test_load_corrupted_raises_artifact_missing(tmp_path): + p = tmp_path / "bad.json" + p.write_text('{"weights": [], "bias": []}', encoding="utf-8") + with pytest.raises(ArtifactMissing): + LinearHead.load(p) + + +def test_train_synthetic_roundtrip(tmp_path): + """T-G4 验收:合成集训练 macro-F1 > 0.9,head.json 可被 LinearHead 推理。""" + import importlib.util + import numpy as np + + script = Path(__file__).resolve().parent.parent / "scripts" / "train_tier_head.py" + spec = importlib.util.spec_from_file_location("train_tier_head", script) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + rng = np.random.default_rng(7) + n, d = 150, 32 + centers = [rng.normal(0, 1, d), rng.normal(5, 1, d), rng.normal(-5, 1, d)] + X = np.vstack([c + rng.normal(0, 0.7, (n, d)) for c in centers]) + y = np.repeat([0, 1, 2], n) + W, b = mod.train(X, y, epochs=150) + m = mod.metrics(X, y, W, b) + assert m["macro_f1"] > 0.9, m + + out = tmp_path / "head.json" + out.write_text(json.dumps({"version": "synthetic", "dim": d, + "labels": ["t1", "t2", "t3"], + "weights": W.T.tolist(), "bias": b.tolist()}), + encoding="utf-8") + head = LinearHead.load(out) + probs = head.predict(X[0].tolist()) + assert sum(probs.values()) == pytest.approx(1.0, abs=1e-9) diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index c539ef6..337c2c3 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -159,7 +159,7 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯 | T-G2 | 观察埋点:observer + 三消费方埋点(pipeline/proxy/client) | ✅ 完成 | T-G2 | | T-G3 | 标签+校准:夜间 true_tier 推导 + split-conformal + 工件表 | ✅ 完成 | T-G3 | | T-G3b | KnnHead(架构变体 B):kNN 投票 + conformal-kNN + 按桶分区/封顶/压缩;hybrid fusion 预留(§14) | ⬜ 待办 | | -| T-G4 | 线性头:离线训练脚本 + LinearHead 纯 Python 推理 + 登记 | ⬜ 待办 | | +| T-G4 | 线性头:离线训练脚本 + LinearHead 纯 Python 推理 + 登记 | ✅ 完成 | T-G4 | | T-G5 | Grader:决策组合(特征门×概率×conformal)+ /v1/route + 三态 mode | ⬜ 待办 | | | T-G6 | live 分流:pipeline tier_fn 三档钩子 + proxy 档位映射 + T2 档升级阶梯 | ⬜ 待办 | | | T-G7 | 审计+前端:ReviewQueue 抽样 + tier 指标卡 + 客户端来源显示/一键升级 | ⬜ 待办 | |