feat(sense): T-X3 路由决策缓存——live 模式 embed+线性头去重(采纳 cortiq 决策哈希缓存)

- gateway/sense/decision_cache.py:DecisionCache(sha256(scope+文本) 键、
  LRU + TTL 60s、4096 条上限、时钟可注入),对齐 auth._AuthCache 进程内模式
- Grader:仅 live 模式缓存纯决策负载(probs/head_version/vec);
  命中跳过 embed + 线性头预测;tier 仍按当前 conformal 阈值即时重算;
  观察落盘(observer.log)不因缓存命中跳过——校准数据完整性不受影响
- collect/shadow 模式不走缓存(校准必须全量);invalidate() 同步清空缓存
  (阈值/工件切换即时生效,TTL 仅兜底陈旧)

pytest 455 passed(T-X2 后 447 + 8)
This commit is contained in:
tzt
2026-09-18 22:41:15 +08:00
parent a043493548
commit 27a092a3f8
4 changed files with 267 additions and 17 deletions
+74
View File
@@ -0,0 +1,74 @@
"""路由决策缓存(T-X3,采纳 cortiq router_client 的决策哈希缓存思路)。
- key = sha256(scope + 0x00 + 归一文本)LRU + TTL(默认 60s / 4096 条,
对齐 auth._AuthCache 的进程内模式,D-P9 单进程前提)。
- 只缓存**纯决策负载**(如 probs/head_version/vec);调用方的观察与审计
落盘职责不因缓存命中而跳过。
- 时钟可注入(now(monotonic)),供测试确定性。
"""
from __future__ import annotations
import hashlib
import threading
import time
from collections import OrderedDict
from typing import Any, Callable, Optional
_DEFAULT_TTL_S = 60.0
_DEFAULT_MAX = 4096
class DecisionCache:
"""进程内决策缓存:LRU + TTL,键为 (scope, 归一文本) 的 sha256。"""
def __init__(self, maxsize: int = _DEFAULT_MAX, ttl_s: float = _DEFAULT_TTL_S,
now: Optional[Callable[[], float]] = None):
self._max = max(1, int(maxsize))
self._ttl = float(ttl_s)
self._now = now or time.monotonic
self._data: "OrderedDict[str, tuple[float, Any]]" = OrderedDict()
self._lock = threading.Lock()
self.hits = 0
self.misses = 0
@staticmethod
def key(scope: str, text: str) -> str:
"""缓存键:scope 隔离消费方/模式,文本归一由调用方完成。"""
return hashlib.sha256(f"{scope}\x00{text}".encode("utf-8")).hexdigest()
def get(self, scope: str, text: str) -> Optional[Any]:
now = self._now()
k = self.key(scope, text)
with self._lock:
item = self._data.get(k)
if item is None:
self.misses += 1
return None
ts, payload = item
if now - ts > self._ttl:
self._data.pop(k, None)
self.misses += 1
return None
self._data.move_to_end(k)
self.hits += 1
return payload
def put(self, scope: str, text: str, payload: Any) -> None:
k = self.key(scope, text)
with self._lock:
self._data[k] = (self._now(), payload)
self._data.move_to_end(k)
while len(self._data) > self._max:
self._data.popitem(last=False)
def clear(self) -> None:
"""清空条目(阈值/工件版本切换时由 Grader.invalidate 调用)。"""
with self._lock:
self._data.clear()
def stats(self) -> dict:
with self._lock:
return {"entries": len(self._data), "hits": self.hits,
"misses": self.misses,
"hit_rate": round(self.hits / (self.hits + self.misses), 4)
if (self.hits + self.misses) else 0.0}
+26 -8
View File
@@ -17,6 +17,7 @@ from typing import Any, Dict, List, Optional
from gateway.sense.calibrate import load_active from gateway.sense.calibrate import load_active
from gateway.sense.classifier import LinearHead from gateway.sense.classifier import LinearHead
from gateway.sense.config import SenseConfig from gateway.sense.config import SenseConfig
from gateway.sense.decision_cache import DecisionCache
from gateway.sense.embedder import embed from gateway.sense.embedder import embed
from gateway.sense.errors import EmbedderDown from gateway.sense.errors import EmbedderDown
from gateway.sense.features import gate from gateway.sense.features import gate
@@ -53,6 +54,7 @@ class Grader:
self._head: Optional[LinearHead] = None self._head: Optional[LinearHead] = None
self._head_loaded = False self._head_loaded = False
self._thresholds: Optional[Dict[str, Any]] = None self._thresholds: Optional[Dict[str, Any]] = None
self._dcache = DecisionCache() # T-X3live 模式决策缓存
def _load_head(self): def _load_head(self):
if not self._head_loaded: if not self._head_loaded:
@@ -66,10 +68,11 @@ class Grader:
return self._head return self._head
def invalidate(self) -> None: def invalidate(self) -> None:
"""工件 promote 后调用(重载 active 工件与阈值)。""" """工件 promote 后调用(重载 active 工件与阈值;同步清空决策缓存)。"""
self._head = None self._head = None
self._head_loaded = False self._head_loaded = False
self._thresholds = None self._thresholds = None
self._dcache.clear()
def _thresholds_cached(self) -> Dict[str, Any]: def _thresholds_cached(self) -> Dict[str, Any]:
if self._thresholds is None: if self._thresholds is None:
@@ -79,22 +82,32 @@ class Grader:
async def decide(self, query_or_messages, consumer: str, async def decide(self, query_or_messages, consumer: str,
domain: str = "", request_id: str = "", domain: str = "", request_id: str = "",
executed_tier: str = "") -> TierDecision: executed_tier: str = "") -> TierDecision:
"""分级决策(§8 时序;全模式写观察)。""" """分级决策(§8 时序;全模式写观察)。
T-X3live 模式下对 (consumer, 文本) 的纯决策负载(probs/head_version/vec
做 60s LRU 缓存,命中时跳过 embed + 线性头;观察/审计照常落盘。
collect/shadow 模式不走缓存(校准数据必须全量产出)。
"""
ts = time.time() ts = time.time()
rid = request_id or ("rt" + uuid.uuid4().hex[:10]) 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) feats = gate(query_or_messages, consumer, self.cfg)
probs: Dict[str, float] = {} probs: Dict[str, float] = {}
head_version = "" head_version = ""
fallback = False fallback = False
vec: Optional[List[int]] = None 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: try:
vec = await embed(feats and (query_or_messages vec = await embed(text, self.cfg.embedder)
if isinstance(query_or_messages, str)
else "\n".join(
str(m.get("content") or "")
for m in query_or_messages)),
self.cfg.embedder)
except EmbedderDown: except EmbedderDown:
fallback = True fallback = True
@@ -105,6 +118,11 @@ class Grader:
if not fallback and vec is not None: if not fallback and vec is not None:
probs = head.predict([float(v) for v in vec]) probs = head.predict([float(v) for v in vec])
head_version = head.version 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 = self._thresholds_cached()
th_version = str(th.get("version") or "") th_version = str(th.get("version") or "")
+157
View File
@@ -0,0 +1,157 @@
"""决策缓存测试(T-X3):LRU+TTL 单元契约 + Grader live 模式接线(embed 去重)。"""
import asyncio
import pytest
from gateway.sense.config import build_sense_config
from gateway.sense.decision_cache import DecisionCache
from gateway.sense.grader import Grader
from gateway.sense.store import SenseStore
# ---------------- DecisionCache 单元 ----------------
class _Clock:
def __init__(self):
self.t = 100.0
def __call__(self):
return self.t
def test_key_scope_and_text_sensitive():
assert DecisionCache.key("proxy", "abc") == DecisionCache.key("proxy", "abc")
assert DecisionCache.key("proxy", "abc") != DecisionCache.key("pipeline", "abc")
assert DecisionCache.key("proxy", "abc") != DecisionCache.key("proxy", "abd")
def test_put_get_hit_miss_counters():
c = DecisionCache()
assert c.get("s", "q") is None
c.put("s", "q", {"tier": "T1"})
assert c.get("s", "q") == {"tier": "T1"}
st = c.stats()
assert st["hits"] == 1 and st["misses"] == 1 and st["entries"] == 1
def test_ttl_expiry_with_fake_clock():
clock = _Clock()
c = DecisionCache(ttl_s=60.0, now=clock)
c.put("s", "q", {"v": 1})
assert c.get("s", "q") == {"v": 1}
clock.t += 61.0 # 超时
assert c.get("s", "q") is None
assert c.stats()["entries"] == 0 # 过期条目被清除
def test_lru_eviction_at_maxsize():
c = DecisionCache(maxsize=2)
c.put("s", "a", 1)
c.put("s", "b", 2)
c.get("s", "a") # a 变为最近使用
c.put("s", "c", 3) # 淘汰 b
assert c.get("s", "a") == 1
assert c.get("s", "b") is None
assert c.get("s", "c") == 3
def test_clear_keeps_counters_but_drops_entries():
c = DecisionCache()
c.put("s", "q", 1)
c.get("s", "q")
c.clear()
assert c.get("s", "q") is None
assert c.stats()["hits"] == 1 # 计数保留(审计口径不回退)
# ---------------- Grader live 模式接线 ----------------
class _CountingObserver:
def __init__(self):
self.logs = []
def log(self, obs):
self.logs.append(obs)
class _FakeHead:
version = "test-head"
def predict(self, vec):
return {"t1": 0.95, "t2": 0.03, "t3": 0.02}
def _make_grader(tmp_path, mode, counter, observer=None):
"""构造测试 Graderembed 打点计数 + FakeHead 注入。"""
import gateway.sense.grader as gr
cfg = build_sense_config({"sense": {
"enabled": True, "mode": mode,
"db_path": str(tmp_path / "s.sqlite3"),
"models_dir": str(tmp_path / "models")}})
store = SenseStore.init_db(cfg.db_path)
g = Grader(cfg, store, observer or _CountingObserver())
g._load_head = lambda: _FakeHead()
async def fake_embed(text, c):
counter["n"] += 1
return [1, 2, 3]
orig = gr.embed
gr.embed = fake_embed
return g, orig
def test_grader_live_mode_dedupes_embed(tmp_path):
"""live 模式:同文本第二次 decide 命中缓存,embed 只跑一次;观察仍写两条。"""
import gateway.sense.grader as gr
counter = {"n": 0}
observer = _CountingObserver()
g, orig = _make_grader(tmp_path, "live", counter, observer)
try:
async def run():
d1 = await g.decide("什么是递归", "proxy")
d2 = await g.decide("什么是递归", "proxy")
d3 = await g.decide("换一个问题", "proxy")
return d1, d2, d3
d1, d2, d3 = asyncio.run(run())
finally:
gr.embed = orig
assert d1.tier == "T1" and d2.tier == "T1" and d3.tier == "T1"
assert counter["n"] == 2 # 同文本去重:2 个不同文本各 1 次
assert d2.head_version == d1.head_version
assert len(observer.logs) == 3 # 观察不因缓存命中而跳过
def test_grader_collect_mode_no_cache(tmp_path):
"""collect/shadow:校准数据必须全量产出,不走缓存。"""
import gateway.sense.grader as gr
counter = {"n": 0}
g, orig = _make_grader(tmp_path, "collect", counter)
try:
async def run():
await g.decide("同一个问题", "proxy")
await g.decide("同一个问题", "proxy")
asyncio.run(run())
finally:
gr.embed = orig
assert counter["n"] == 2
def test_grader_invalidate_clears_cache(tmp_path):
"""invalidate(工件切换)后缓存清空:同文本重新 embed。"""
import gateway.sense.grader as gr
counter = {"n": 0}
g, orig = _make_grader(tmp_path, "live", counter)
try:
async def run():
await g.decide("什么是递归", "proxy")
g.invalidate()
await g.decide("什么是递归", "proxy")
asyncio.run(run())
finally:
gr.embed = orig
assert counter["n"] == 2
+1
View File
@@ -167,3 +167,4 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯
| OPT-1 | 分支推进:语义缓存 L2 查找 3.39x(免并集计分+预筛)+ 安全加固(15 高危清零:SSRF/路径穿越/假凭据) | ✅ 完成 | ad3bf41 | | OPT-1 | 分支推进:语义缓存 L2 查找 3.39x(免并集计分+预筛)+ 安全加固(15 高危清零:SSRF/路径穿越/假凭据) | ✅ 完成 | ad3bf41 |
| T-X1 | 预算四档渐进干预(外部采纳 ai-model-router):budget_mode 整数基点判定(80/95/100%+ optimize/cheap 自动降档 + X-Budget-Mode 上报;黄金用例锁边界 | ✅ 完成 | T-X1 | | T-X1 | 预算四档渐进干预(外部采纳 ai-model-router):budget_mode 整数基点判定(80/95/100%+ optimize/cheap 自动降档 + X-Budget-Mode 上报;黄金用例锁边界 | ✅ 完成 | T-X1 |
| T-X2 | 上游有序降级链(外部采纳 cortiq tier 链):池内候选按档位/单价排序、首 token 前 failover、X-Upstream-Fallback 三元组响应头 + admin/stats failover 聚合 | ✅ 完成 | T-X2 | | T-X2 | 上游有序降级链(外部采纳 cortiq tier 链):池内候选按档位/单价排序、首 token 前 failover、X-Upstream-Fallback 三元组响应头 + admin/stats failover 聚合 | ✅ 完成 | T-X2 |
| T-X3 | 路由决策缓存(外部采纳 cortiq 决策哈希缓存):DecisionCachesha256+LRU+TTL60s4096 条)前置 grader live 模式,embed+线性头去重;观察/审计不跳过、invalidate 同步清空、collect/shadow 不缓存 | ✅ 完成 | T-X3 |