tzt
1aaa05cacc
- gateway/sense/promotion.py：PromotionTable 状态机（candidate/promoted/demoted）
* 晋升门：n_total>=n_min 且 通过率>=promote_lb 且 soak 浸泡期满足
* 退化：promoted 期间通过率<demote_lb 自动降级；降级后可凭数据恢复
* 一票否决：tier=T3 观察不进通过率统计，已晋升者立即降级
* 持久化 sense_promotion 表（CREATE IF NOT EXISTS 幂等）
- grader：live 模式对已晋升标签的 T2 决策（无 T3 信号/非 fallback）升级 T1，
hard_gates.promotion_applied 如实标注；T1/T3/fallback 路径不受影响
- 回填链路：T1 审计抽样带 promo:<label> 标签 -> 人工 verdict approve=ok
经 /review/{id} 提交时回填晋升表（失败不影响审核主流程）
- 双闸门语义：conformal 阈值保证单条决策风险率，晋升表保证标签级接管节奏；
collect 数据不足（n_min 未满）不晋升，全部迁移可经 T-X4 留痕审计
pytest 474 passed（T-X4 后 466 + 8）
257 lines
10 KiB
Python
257 lines
10 KiB
Python
"""分级决策组合（T-G5，§8 主时序）：
|
||
|
||
embedder.embed（挂 -> fallback=T2+规则门，D-G4）
|
||
-> features.gate（t1_hard_ok）
|
||
-> LinearHead.predict -> {p1,p2,p3}
|
||
-> conformal 阈值：p1>=τ1 且 t1_hard_ok -> T1；p3>=τ3 或 repo_signals -> T3；其余 T2
|
||
-> mode 裁剪：collect/shadow 只写观察（decided≠executed），live 返回决策
|
||
-> observer.log（全模式必写）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
import uuid
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from gateway.sense.calibrate import load_active
|
||
from gateway.sense.classifier import LinearHead
|
||
from gateway.sense.config import SenseConfig
|
||
from gateway.sense.decision_cache import DecisionCache
|
||
from gateway.sense.embedder import embed
|
||
from gateway.sense.errors import EmbedderDown
|
||
from gateway.sense.features import gate
|
||
from gateway.sense.promotion import promotion_label
|
||
from gateway.sense.store import json_dumps
|
||
|
||
|
||
@dataclass
|
||
class TierDecision:
|
||
tier: str # 决策档（live 时即执行档的候选）
|
||
probs: Dict[str, float]
|
||
hard_gates: Dict[str, Any]
|
||
thresholds_version: str = ""
|
||
head_version: str = ""
|
||
mode: str = "collect"
|
||
fallback: bool = False # True = embedder/工件故障，规则门退化
|
||
executed_tier: str = "" # 现行为执行的档位（collect/shadow 记录用）
|
||
|
||
|
||
def _rule_tier(feats) -> str:
|
||
"""规则门兜底（无模型时的决策）：仓库级 -> T3；t1 硬门过 -> T1；否则 T2。"""
|
||
if feats.repo_signals:
|
||
return "T3"
|
||
if feats.t1_hard_ok:
|
||
return "T1"
|
||
return "T2"
|
||
|
||
|
||
def _decision_reasons(fallback: bool, probs: Dict[str, float],
|
||
th: Dict[str, Any], feats) -> List[str]:
|
||
"""决策理由（T-X4）：按 §8 判定顺序产出，可解释路由的最小完备集。"""
|
||
if fallback:
|
||
r = ["fallback:rule_gate"]
|
||
if feats.repo_signals:
|
||
r.append("repo_signals")
|
||
if feats.t1_hard_ok:
|
||
r.append("t1_hard_ok")
|
||
return r
|
||
p1 = probs.get("t1", 0.0) >= float(th.get("t1", 0.9))
|
||
p3 = probs.get("t3", 0.0) >= float(th.get("t3", 0.9))
|
||
t1_ok = p1 and feats.t1_hard_ok
|
||
t3_ok = p3 or feats.repo_signals
|
||
r: List[str] = []
|
||
if t3_ok and not t1_ok:
|
||
r.append("t3_ok")
|
||
elif t1_ok:
|
||
r.append("t1_ok")
|
||
else:
|
||
r.append("default_t2")
|
||
if p1:
|
||
r.append("p1>=threshold")
|
||
if not feats.t1_hard_ok:
|
||
r.append("gate:t1_hard_ok=0")
|
||
if p3:
|
||
r.append("p3>=threshold")
|
||
if feats.repo_signals:
|
||
r.append("repo_signals")
|
||
return r
|
||
|
||
|
||
def _rejected_tiers(fallback: bool, probs: Dict[str, float],
|
||
th: Dict[str, Any], feats) -> List[Dict[str, str]]:
|
||
"""落选档位及原因（T2 为缺省档不记录；fallback 时两条均给 fallback 原因）。"""
|
||
if fallback:
|
||
return [{"tier": "T1", "reason": "fallback"},
|
||
{"tier": "T3", "reason": "fallback"}]
|
||
p1 = probs.get("t1", 0.0) >= float(th.get("t1", 0.9))
|
||
p3 = probs.get("t3", 0.0) >= float(th.get("t3", 0.9))
|
||
t1_ok = p1 and feats.t1_hard_ok
|
||
t3_ok = p3 or feats.repo_signals
|
||
out: List[Dict[str, str]] = []
|
||
if not t3_ok:
|
||
out.append({"tier": "T3", "reason": "p3<threshold 且无 repo_signals"})
|
||
if not t1_ok:
|
||
out.append({"tier": "T1",
|
||
"reason": "p1<threshold" if not p1 else "t1_hard_ok=0"})
|
||
return out
|
||
|
||
|
||
class Grader:
|
||
"""决策器（持有 active 工件缓存；工件/阈值切换后调 invalidate）。"""
|
||
|
||
def __init__(self, cfg: SenseConfig, store, observer=None, promotion=None):
|
||
self.cfg = cfg
|
||
self.store = store
|
||
self.observer = observer
|
||
self._promotion = promotion # T-X5：晋升表（可 None）
|
||
self._head: Optional[LinearHead] = None
|
||
self._head_loaded = False
|
||
self._thresholds: Optional[Dict[str, Any]] = None
|
||
self._dcache = DecisionCache() # T-X3：live 模式决策缓存
|
||
|
||
def _load_head(self):
|
||
if not self._head_loaded:
|
||
art = self.store.active_artifact("head")
|
||
if art:
|
||
try:
|
||
self._head = LinearHead.load(art["path"], version=art["version"])
|
||
except Exception: # noqa: BLE001
|
||
self._head = None
|
||
self._head_loaded = True
|
||
return self._head
|
||
|
||
def invalidate(self) -> None:
|
||
"""工件 promote 后调用（重载 active 工件与阈值；同步清空决策缓存）。"""
|
||
self._head = None
|
||
self._head_loaded = False
|
||
self._thresholds = None
|
||
self._dcache.clear()
|
||
|
||
def _thresholds_cached(self) -> Dict[str, Any]:
|
||
if self._thresholds is None:
|
||
self._thresholds = load_active(self.store, self.cfg.models_dir)
|
||
return self._thresholds
|
||
|
||
async def decide(self, query_or_messages, consumer: str,
|
||
domain: str = "", request_id: str = "",
|
||
executed_tier: str = "") -> TierDecision:
|
||
"""分级决策（§8 时序；全模式写观察）。
|
||
|
||
T-X3：live 模式下对 (consumer, 文本) 的纯决策负载（probs/head_version/vec）
|
||
做 60s LRU 缓存，命中时跳过 embed + 线性头；观察/审计照常落盘。
|
||
collect/shadow 模式不走缓存（校准数据必须全量产出）。
|
||
"""
|
||
ts = time.time()
|
||
rid = request_id or ("rt" + uuid.uuid4().hex[:10])
|
||
text = (query_or_messages if isinstance(query_or_messages, str)
|
||
else "\n".join(str(m.get("content") or "")
|
||
for m in query_or_messages))
|
||
feats = gate(query_or_messages, consumer, self.cfg)
|
||
probs: Dict[str, float] = {}
|
||
head_version = ""
|
||
fallback = False
|
||
|
||
vec: Optional[List[int]] = None
|
||
cached = self._dcache.get(consumer, text) \
|
||
if self.cfg.mode == "live" else None
|
||
if cached is not None:
|
||
probs = dict(cached["probs"])
|
||
head_version = str(cached["head_version"])
|
||
vec = cached["vec"]
|
||
else:
|
||
try:
|
||
vec = await embed(text, self.cfg.embedder)
|
||
except EmbedderDown:
|
||
fallback = True
|
||
|
||
head = None if fallback else self._load_head()
|
||
if head is None:
|
||
fallback = True
|
||
|
||
if not fallback and vec is not None:
|
||
probs = head.predict([float(v) for v in vec])
|
||
head_version = head.version
|
||
if self.cfg.mode == "live":
|
||
self._dcache.put(consumer, text,
|
||
{"probs": dict(probs),
|
||
"head_version": head_version,
|
||
"vec": vec})
|
||
|
||
th = self._thresholds_cached()
|
||
th_version = str(th.get("version") or "")
|
||
|
||
# ---- 决策 ----
|
||
if fallback:
|
||
tier = _rule_tier(feats) # D-G4 规则门退化
|
||
else:
|
||
t1_ok = (probs.get("t1", 0.0) >= float(th.get("t1", 0.9))
|
||
and feats.t1_hard_ok)
|
||
t3_ok = (probs.get("t3", 0.0) >= float(th.get("t3", 0.9))
|
||
or feats.repo_signals)
|
||
if t3_ok and not t1_ok:
|
||
tier = "T3"
|
||
elif t1_ok:
|
||
tier = "T1"
|
||
else:
|
||
tier = "T2"
|
||
|
||
# ---- 晋升表应用（T-X5：conformal 之外的统计闸门；仅 live）----
|
||
# 仅对 T2 决策且无任何 T3 信号时，允许已晋升标签升级 T1；
|
||
# T1/T3 决策与 fallback 路径不受晋升表影响。
|
||
promotion_applied = False
|
||
if (self._promotion is not None and self.cfg.mode == "live"
|
||
and not fallback and tier == "T2"
|
||
and not feats.repo_signals
|
||
and probs.get("t3", 0.0) < float(th.get("t3", 0.9))):
|
||
if self._promotion.is_promoted(promotion_label(consumer, domain)):
|
||
tier = "T1"
|
||
promotion_applied = True
|
||
|
||
# ---- mode 裁剪（D-G7）----
|
||
if self.cfg.mode == "live":
|
||
executed = tier # live：决策即执行
|
||
else:
|
||
# collect/shadow：现行为——规则门等价（T1 门/T3 信号）近似 v2 现状
|
||
executed = executed_tier or _rule_tier(feats)
|
||
|
||
decision = TierDecision(
|
||
tier=tier, probs=probs,
|
||
hard_gates={"turns": feats.turns, "est_tokens": feats.est_tokens,
|
||
"single_turn": feats.single_turn,
|
||
"intent_blocked": feats.intent_blocked,
|
||
"repo_signals": feats.repo_signals,
|
||
"over_length": feats.over_length,
|
||
"t1_hard_ok": feats.t1_hard_ok,
|
||
"promotion_applied": promotion_applied},
|
||
thresholds_version=th_version, head_version=head_version,
|
||
mode=self.cfg.mode, fallback=fallback,
|
||
executed_tier=executed)
|
||
|
||
# ---- 决策留痕（T-X4：reasons / candidate_scores / rejected 三元结构）----
|
||
try:
|
||
self.store.insert_decision({
|
||
"ts": ts, "request_id": rid, "consumer": consumer,
|
||
"decided_tier": tier, "executed_tier": executed,
|
||
"mode": self.cfg.mode, "fallback": fallback,
|
||
"policy_version": f"head:{head_version}|th:{th_version}",
|
||
"q_hash": __import__("hashlib").sha256(
|
||
text.encode("utf-8")).hexdigest()[:32],
|
||
"reasons": json_dumps(_decision_reasons(fallback, probs, th, feats)),
|
||
"candidate_scores": json_dumps(probs),
|
||
"rejected": json_dumps(_rejected_tiers(fallback, probs, th, feats)),
|
||
})
|
||
except Exception: # noqa: BLE001
|
||
pass # 留痕失败不影响决策（观察与代理主链路优先）
|
||
|
||
# ---- 观察必写（全模式）----
|
||
if self.observer is not None:
|
||
self.observer.log(__import__("gateway.sense.observer",
|
||
fromlist=["Observation"]).Observation(
|
||
request_id=rid, consumer=consumer, decided_tier=tier,
|
||
executed_tier=executed, probs=probs,
|
||
policy_version=f"head:{head_version}|th:{th_version}",
|
||
features=decision.hard_gates, embedding=vec,
|
||
bucket="default", domain=domain, ts=ts))
|
||
return decision
|