diff --git a/gateway/proxy/config.py b/gateway/proxy/config.py index 9868490..c5c2f6b 100644 --- a/gateway/proxy/config.py +++ b/gateway/proxy/config.py @@ -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"), ) diff --git a/gateway/proxy/normalizer.py b/gateway/proxy/normalizer.py index f4917c8..64ff6b7 100644 --- a/gateway/proxy/normalizer.py +++ b/gateway/proxy/normalizer.py @@ -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}" diff --git a/gateway/proxy/routes.py b/gateway/proxy/routes.py index a944225..932cd9f 100644 --- a/gateway/proxy/routes.py +++ b/gateway/proxy/routes.py @@ -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 {} diff --git a/gateway/proxy/semcache.py b/gateway/proxy/semcache.py index 67a7b10..d1688ee 100644 --- a/gateway/proxy/semcache.py +++ b/gateway/proxy/semcache.py @@ -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, diff --git a/tests/test_route_sig.py b/tests/test_route_sig.py new file mode 100644 index 0000000..da550bf --- /dev/null +++ b/tests/test_route_sig.py @@ -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