feat(proxy): T-X10 采纳 cortiq 语义缓存路由签名分桶——不同路由意图不互串答案
- normalizer.canonical_hash 增加可选 route_sig 段:键 = bucket|doc_version|sig|sha256 (旧三段格式向后兼容,旧条目随 TTL 自然淘汰) - ProxyConfig 新增 semcache.route_sig_scope:capabilities(默认,vision/tools 需求 签名)/ model(按模型隔离)/ none(旧行为);非法值回落 capabilities - semcache:签名升级为条目属性并分区 L2 语义扫描(仅键分桶不够——语义层仍会 跨签名命中);签名从缓存键第四段解析,重启重建零 schema 变更; 晋升别名键携带签名段;route_sig=None 的旧调用零过滤完全兼容 - routes:lookup/put 共用同一 norm_hash(消除 put 侧重复哈希),签名贯穿两层 - 新增 tests/test_route_sig.py 6 项(键格式/键空间分割/scope 三态/两层隔离/ 重建存活/旧调用兼容)
This commit is contained in:
@@ -57,6 +57,9 @@ class ProxyConfig:
|
||||
sim_threshold: float = 0.92
|
||||
max_entries: int = 300000
|
||||
promote_frequency: int = 5
|
||||
# T-X10(采纳 cortiq 路由签名分桶):语义缓存键的路由签名粒度
|
||||
# none | capabilities(默认:vision/tools 需求不同不互串) | model(按模型隔离)
|
||||
route_sig_scope: str = "capabilities"
|
||||
|
||||
def bucket(self, name: str) -> BucketCfg:
|
||||
"""取桶配置;未知名回落 default(D-P2 缺省桶)。"""
|
||||
@@ -161,4 +164,8 @@ def build_proxy_config(settings_dict: Dict[str, Any]) -> ProxyConfig:
|
||||
sim_threshold=float(sem.get("sim_threshold", 0.92) or 0.92),
|
||||
max_entries=int(sem.get("max_entries", 300000) or 300000),
|
||||
promote_frequency=int(sem.get("promote_frequency", 5) or 5),
|
||||
route_sig_scope=(str(sem.get("route_sig_scope"))
|
||||
if str(sem.get("route_sig_scope")) in
|
||||
("none", "capabilities", "model")
|
||||
else "capabilities"),
|
||||
)
|
||||
|
||||
@@ -83,13 +83,19 @@ def is_cacheable(body: dict) -> bool:
|
||||
return len(non_system) <= 1
|
||||
|
||||
|
||||
def canonical_hash(bucket: str, doc_version: int, body: dict) -> str:
|
||||
def canonical_hash(bucket: str, doc_version: int, body: dict,
|
||||
route_sig: str = "") -> str:
|
||||
"""规则 3:缓存键 = bucket + '|' + doc_version + '|' + sha256(norm)。
|
||||
|
||||
route_sig 非空时插入为第三段(T-X10,采纳 cortiq 路由签名分桶:
|
||||
不同路由意图——vision/tools 需求或模型——不互串答案);为空时保持
|
||||
旧三段格式(既有调用与旧缓存条目兼容,旧条目随 TTL 自然淘汰)。
|
||||
桶模板/资料前缀不参与哈希(由 bucket+doc_version 表达,资料更新 = 版本+1)。
|
||||
"""
|
||||
norm = normalize_messages(body)
|
||||
digest = hashlib.sha256(norm.encode("utf-8")).hexdigest()
|
||||
if route_sig:
|
||||
return f"{bucket}|{doc_version}|{route_sig}|{digest}"
|
||||
return f"{bucket}|{doc_version}|{digest}"
|
||||
|
||||
|
||||
|
||||
+35
-10
@@ -454,6 +454,23 @@ def reset_sense_runtimes() -> None:
|
||||
_sense_runtimes = {}
|
||||
|
||||
|
||||
def _route_sig(needs: Dict[str, Any], model: str, cfg: ProxyConfig) -> str:
|
||||
"""路由签名(T-X10,采纳 cortiq 语义缓存路由签名分桶)。
|
||||
|
||||
签名进入缓存键,使不同路由意图的请求不互串答案:
|
||||
- capabilities(默认):vision/tools 需求不同 -> 不同签名(多模态/工具请求
|
||||
不再命中纯文本缓存答案——正确性优先);
|
||||
- model:按模型名隔离(更保守,命中率换绝对隔离);
|
||||
- none:空签名(旧行为,最高命中率)。
|
||||
"""
|
||||
scope = getattr(cfg, "route_sig_scope", "capabilities")
|
||||
if scope == "model":
|
||||
return f"m:{model}"
|
||||
if scope == "capabilities":
|
||||
return f"v{int(bool(needs['vision']))}t{int(bool(needs['tools']))}"
|
||||
return ""
|
||||
|
||||
|
||||
def _cached_charge(body: dict, cfg: ProxyConfig, model: str) -> Dict[str, int]:
|
||||
"""缓存命中计费:成本 0,按未命中口径对入/出估 token 收售价(§7)。"""
|
||||
from gateway.proxy.pricing import compute
|
||||
@@ -510,9 +527,11 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
|
||||
body = {**body, "model": model}
|
||||
|
||||
# ---- 缓存分支(T-P6,§7 时序):仅缓存准入(stop+单轮)查询 ----
|
||||
# T-X10:缓存键带路由签名(scope 由 cfg.route_sig_scope 决定),且
|
||||
# lookup/put 共用同一 norm_hash(原先 put 侧重复计算一次)
|
||||
cacheable = False
|
||||
cache = None
|
||||
resolution = None
|
||||
norm_hash = ""
|
||||
try:
|
||||
if cfg.semcache_enabled:
|
||||
from gateway.proxy.normalizer import canonical_hash, is_cacheable
|
||||
@@ -520,11 +539,14 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
|
||||
bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket") or "default"))
|
||||
cacheable = is_cacheable(body)
|
||||
if cacheable:
|
||||
norm_hash = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
|
||||
sig = _route_sig(needs, model, cfg)
|
||||
norm_hash = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version,
|
||||
body, sig)
|
||||
cache = _get_semcache(cfg, ledger)
|
||||
norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
|
||||
sort_keys=True)
|
||||
hit = cache.lookup(norm_hash, norm_text, bucket_cfg.doc_version)
|
||||
hit = cache.lookup(norm_hash, norm_text, bucket_cfg.doc_version,
|
||||
route_sig=sig)
|
||||
if hit is not None:
|
||||
est = _estimate_hold_milli(body, cfg, model)
|
||||
if await asyncio.to_thread(
|
||||
@@ -585,7 +607,7 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
|
||||
ledger, model, est, t0,
|
||||
cache=cache, cacheable=cacheable,
|
||||
headers=headers, budget_mode=budget_mode,
|
||||
chain=chain)
|
||||
chain=chain, norm_hash=norm_hash)
|
||||
except UpstreamAborted as e:
|
||||
# 流中失败:按已收 usage 结算(无 usage 按字符估算),不缓存(D-P4)
|
||||
usage = sink.get("usage") or _estimate_usage_from_sink(sink)
|
||||
@@ -670,7 +692,8 @@ async def _stream_response(body, entry, sink, headers, client_wants_usage,
|
||||
async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
|
||||
model, est, t0, cache=None, cacheable=False,
|
||||
headers=None, budget_mode: str = "normal",
|
||||
chain: Optional[List[Dict[str, Any]]] = None):
|
||||
chain: Optional[List[Dict[str, Any]]] = None,
|
||||
norm_hash: str = ""):
|
||||
parts = []
|
||||
async for raw_bytes in upstream_stream(body, entry, sink,
|
||||
chain if chain is not None else [entry]):
|
||||
@@ -702,17 +725,19 @@ async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
|
||||
out_tok=usage.get("out", 0), upstream_cost_milli=br["upstream_cost_milli"],
|
||||
ttfb_ms=sink.get("ttfb_ms"), total_ms=total_ms, status="ok")
|
||||
# 缓存准入(D-P5):stop 且单轮且未命中来的 -> 写缓存
|
||||
if cache is not None and cacheable and sink.get("finish_reason", "stop") == "stop":
|
||||
# T-X10:键复用 lookup 时的 norm_hash(含路由签名),不再重复计算
|
||||
if cache is not None and cacheable and norm_hash \
|
||||
and sink.get("finish_reason", "stop") == "stop":
|
||||
try:
|
||||
from gateway.proxy.normalizer import canonical_hash
|
||||
from gateway.proxy.semcache import _sig_from_key
|
||||
bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket")
|
||||
or "default")) if headers else cfg.bucket("default")
|
||||
ckey = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
|
||||
norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
|
||||
sort_keys=True)
|
||||
cache.put(ckey, norm_text, "".join(parts), model,
|
||||
cache.put(norm_hash, norm_text, "".join(parts), model,
|
||||
doc_version=bucket_cfg.doc_version,
|
||||
ttl_hours=bucket_cfg.ttl_hours)
|
||||
ttl_hours=bucket_cfg.ttl_hours,
|
||||
route_sig=_sig_from_key(norm_hash))
|
||||
except Exception:
|
||||
pass
|
||||
fb = sink.get("upstream_fallback") or {}
|
||||
|
||||
+37
-12
@@ -62,10 +62,11 @@ def weighted_jaccard(ga: set, gb: set) -> float:
|
||||
|
||||
class CacheEntry:
|
||||
__slots__ = ("answer", "model", "q_norm", "g", "w", "created_ts", "ttl_ts",
|
||||
"doc_version", "hits")
|
||||
"doc_version", "hits", "sig")
|
||||
|
||||
def __init__(self, answer: str, model: str, q_norm: str,
|
||||
created_ts: float, ttl_ts: float, doc_version: int):
|
||||
created_ts: float, ttl_ts: float, doc_version: int,
|
||||
sig: str = ""):
|
||||
self.answer = answer
|
||||
self.model = model
|
||||
self.q_norm = q_norm
|
||||
@@ -75,6 +76,17 @@ class CacheEntry:
|
||||
self.ttl_ts = ttl_ts
|
||||
self.doc_version = doc_version
|
||||
self.hits = 0
|
||||
self.sig = sig # 路由签名(T-X10:L2 扫描按签名分区)
|
||||
|
||||
|
||||
def _sig_from_key(cache_key: str) -> str:
|
||||
"""从缓存键解析路由签名(T-X10)。
|
||||
|
||||
带签名键 = bucket|doc_version|sig|sha256(4 段);旧三段键无签名返回 ""。
|
||||
签名段约束不含 '|'(capabilities/model 两种 scope 均满足),故解析无歧义。
|
||||
"""
|
||||
parts = cache_key.split("|")
|
||||
return parts[2] if len(parts) >= 4 else ""
|
||||
|
||||
|
||||
class SemanticCache:
|
||||
@@ -106,7 +118,8 @@ class SemanticCache:
|
||||
return
|
||||
for r in rows:
|
||||
entry = CacheEntry(r["answer"], r["model"], r["q_norm"],
|
||||
r["created_ts"], r["ttl_ts"], r["doc_version"])
|
||||
r["created_ts"], r["ttl_ts"], r["doc_version"],
|
||||
sig=_sig_from_key(r["cache_key"]))
|
||||
entry.hits = r["hits"]
|
||||
self._index(r["cache_key"], entry, promote=False)
|
||||
|
||||
@@ -131,12 +144,18 @@ class SemanticCache:
|
||||
|
||||
# ---------- 查询 ----------
|
||||
def lookup(self, cache_key: str, norm_text: str,
|
||||
doc_version: int = 1) -> Optional[Dict[str, Any]]:
|
||||
"""L1 精确 -> L2 语义(§7 签名)。返回 {answer, model, level} 或 None。"""
|
||||
doc_version: int = 1,
|
||||
route_sig: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
||||
"""L1 精确 -> L2 语义(§7 签名)。返回 {answer, model, level} 或 None。
|
||||
|
||||
route_sig 非 None 时按签名分区(T-X10):L1 命中要求条目签名一致,
|
||||
L2 扫描跳过签名不一致的候选(不同路由意图不互串答案);
|
||||
None = 不校验(旧调用/旧测试完全兼容)。
|
||||
"""
|
||||
now = self._now()
|
||||
# L1
|
||||
entry = self._l1.get(cache_key)
|
||||
if entry is not None:
|
||||
if entry is not None and (route_sig is None or entry.sig == route_sig):
|
||||
if entry.ttl_ts < now:
|
||||
self._invalidate_key(cache_key)
|
||||
else:
|
||||
@@ -161,6 +180,8 @@ class SemanticCache:
|
||||
cand = self._l1.get(key)
|
||||
if cand is None or cand.ttl_ts < now or cand.doc_version != doc_version:
|
||||
continue
|
||||
if route_sig is not None and cand.sig != route_sig:
|
||||
continue # 签名分区:跨路由意图不参与计分
|
||||
# 规模上界预筛:w_inter <= lo 且 w_union >= hi,故 score <= lo/hi;
|
||||
# 严格小于阈值者不可能命中,跳过(不构建相交集合)。
|
||||
# 注意用严格不等式:lo/hi == 阈值的边界候选仍会进入精确计分,
|
||||
@@ -181,9 +202,12 @@ class SemanticCache:
|
||||
promoted = False
|
||||
if cand.hits >= self.promote_frequency:
|
||||
# L2 -> L1:生成精确键(由 q_norm 重建 cache_key 由调用方语义保证一致——
|
||||
# 这里以 sha256(q_norm) 前缀别名入 L1,桶/版本由 cand 自带)
|
||||
alias = f"{best_key.split('|')[0]}|{cand.doc_version}|promoted:" \
|
||||
f"{hashlib.sha256(cand.q_norm.encode()).hexdigest()[:16]}"
|
||||
# 这里以 sha256(q_norm) 前缀别名入 L1,桶/版本/签名由 cand 自带)
|
||||
h = hashlib.sha256(cand.q_norm.encode()).hexdigest()[:16]
|
||||
if cand.sig:
|
||||
alias = f"{best_key.split('|')[0]}|{cand.doc_version}|{cand.sig}|promoted:{h}"
|
||||
else:
|
||||
alias = f"{best_key.split('|')[0]}|{cand.doc_version}|promoted:{h}"
|
||||
self._index(alias, cand, promote=True)
|
||||
try:
|
||||
self.store.promote_semcache(best_key, alias, cand.hits)
|
||||
@@ -205,11 +229,12 @@ class SemanticCache:
|
||||
|
||||
# ---------- 写入 ----------
|
||||
def put(self, cache_key: str, q_norm: str, answer: str, model: str,
|
||||
doc_version: int = 1, ttl_hours: int = 72) -> None:
|
||||
"""写 L1 + 倒排 + sqlite 持久化(§7 签名)。"""
|
||||
doc_version: int = 1, ttl_hours: int = 72,
|
||||
route_sig: str = "") -> None:
|
||||
"""写 L1 + 倒排 + sqlite 持久化(§7 签名)。route_sig 随条目留存供 L2 分区。"""
|
||||
now = self._now()
|
||||
entry = CacheEntry(answer, model, q_norm, now, now + ttl_hours * 3600,
|
||||
doc_version)
|
||||
doc_version, sig=route_sig)
|
||||
self._index(cache_key, entry)
|
||||
try:
|
||||
self.store.put_semcache(cache_key, "default", q_norm, answer, model,
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
"""语义缓存路由签名分桶(T-X10,采纳 cortiq 路由签名设计)。"""
|
||||
from gateway.proxy.config import build_proxy_config
|
||||
from gateway.proxy.normalizer import canonical_hash
|
||||
from gateway.proxy.routes import _route_sig
|
||||
|
||||
_BODY = {"model": "m1", "messages": [{"role": "user", "content": "什么是递归"}]}
|
||||
|
||||
|
||||
def test_canonical_hash_legacy_format_without_sig():
|
||||
"""不带签名保持旧三段格式(旧调用/旧条目兼容)。"""
|
||||
h = canonical_hash("default", 1, _BODY)
|
||||
assert h.count("|") == 2
|
||||
h2 = canonical_hash("default", 1, _BODY, "")
|
||||
assert h == h2
|
||||
|
||||
|
||||
def test_canonical_hash_with_sig_divides_keyspace():
|
||||
"""签名不同 -> 键不同:vision 请求与纯文本请求不再共用缓存条目。"""
|
||||
plain = canonical_hash("default", 1, _BODY)
|
||||
vision = canonical_hash("default", 1, _BODY, "v1t0")
|
||||
tools = canonical_hash("default", 1, _BODY, "v0t1")
|
||||
assert len({plain, vision, tools}) == 3
|
||||
|
||||
|
||||
def test_route_sig_scopes():
|
||||
"""三种 scope 的签名形态:capabilities 默认 / model 精确 / none 旧格式。"""
|
||||
cfg = build_proxy_config({"proxy": {"enabled": True}})
|
||||
needs = {"vision": True, "tools": False, "min_context_tokens": 10}
|
||||
assert _route_sig(needs, "m1", cfg) == "v1t0"
|
||||
|
||||
cfg_model = build_proxy_config({"proxy": {"semcache": {"route_sig_scope": "model"}}})
|
||||
assert _route_sig(needs, "m1", cfg_model) == "m:m1"
|
||||
|
||||
cfg_none = build_proxy_config({"proxy": {"semcache": {"route_sig_scope": "none"}}})
|
||||
assert _route_sig(needs, "m1", cfg_none) == ""
|
||||
|
||||
# 非法 scope 回落 capabilities
|
||||
cfg_bad = build_proxy_config({"proxy": {"semcache": {"route_sig_scope": "bogus"}}})
|
||||
assert _route_sig(needs, "m1", cfg_bad) == "v1t0"
|
||||
|
||||
|
||||
def test_sig_isolation_end_to_end_via_semcache(tmp_path):
|
||||
"""经 SemanticCache 验证:签名分区 L1 精确层与 L2 语义扫描两层。"""
|
||||
from gateway.proxy.ledger import Ledger
|
||||
from gateway.proxy.semcache import SemanticCache
|
||||
|
||||
led = Ledger.init_db(tmp_path / "l.sqlite3")
|
||||
cache = SemanticCache(led, max_entries=100, sim_threshold=0.5,
|
||||
promote_frequency=5)
|
||||
k_text = canonical_hash("default", 1, _BODY, "v0t0")
|
||||
k_vision = canonical_hash("default", 1, _BODY, "v1t0")
|
||||
cache.put(k_text, "什么是递归", "纯文本答案", "m1", doc_version=1,
|
||||
route_sig="v0t0")
|
||||
# 相同问题、vision 签名:L1 键不同必不中;L2 语义扫描按签名分区也不中
|
||||
assert cache.lookup(k_vision, "什么是递归", doc_version=1,
|
||||
route_sig="v1t0") is None
|
||||
# 同签名:L1 精确命中
|
||||
hit = cache.lookup(k_text, "什么是递归", doc_version=1, route_sig="v0t0")
|
||||
assert hit is not None and hit["level"] == "exact"
|
||||
|
||||
|
||||
def test_sig_partition_survives_rebuild(tmp_path):
|
||||
"""重启重建(semcache 表无签名列):签名从缓存键解析恢复,分区仍有效。"""
|
||||
from gateway.proxy.ledger import Ledger
|
||||
from gateway.proxy.semcache import SemanticCache
|
||||
|
||||
led = Ledger.init_db(tmp_path / "l.sqlite3")
|
||||
c1 = SemanticCache(led, max_entries=100, sim_threshold=0.5, promote_frequency=5)
|
||||
k = canonical_hash("default", 1, _BODY, "v1t0")
|
||||
c1.put(k, "什么是递归", "多模态答案", "m1", doc_version=1, route_sig="v1t0")
|
||||
c2 = SemanticCache(led, max_entries=100, sim_threshold=0.5, promote_frequency=5)
|
||||
assert c2.lookup(k, "什么是递归", doc_version=1, route_sig="v1t0") is not None
|
||||
assert c2.lookup(canonical_hash("default", 1, _BODY, "v0t0"),
|
||||
"什么是递归", doc_version=1, route_sig="v0t0") is None
|
||||
|
||||
|
||||
def test_legacy_calls_without_sig_unchanged(tmp_path):
|
||||
"""不传签名(route_sig=None)的旧调用零过滤:既有测试行为不变。"""
|
||||
from gateway.proxy.ledger import Ledger
|
||||
from gateway.proxy.semcache import SemanticCache
|
||||
|
||||
led = Ledger.init_db(tmp_path / "l.sqlite3")
|
||||
cache = SemanticCache(led, max_entries=100, sim_threshold=0.5, promote_frequency=5)
|
||||
cache.put("k1", "请解释一下递归函数的概念", "答案", "m1", doc_version=1)
|
||||
assert cache.lookup("k1", "请解释一下递归函数的概念", doc_version=1) is not None
|
||||
# 语义路径同样不受影响(近似文本跨键命中)
|
||||
assert cache.lookup("k2", "请解释一下递归函数的概念呢", doc_version=1) is not None
|
||||
Reference in New Issue
Block a user