diff --git a/gateway/proxy/ledger.py b/gateway/proxy/ledger.py index 850e8ea..6459ecf 100644 --- a/gateway/proxy/ledger.py +++ b/gateway/proxy/ledger.py @@ -187,6 +187,51 @@ class Ledger(BillingMixin): (used + 1, today, key_id)) return True + # ---------- 语义缓存持久化(T-P6,供 semcache.SemanticCache 调用) ---------- + def semcache_rows(self, limit: int = 300000) -> List[Dict[str, Any]]: + """全量缓存行(启动重建倒排索引)。""" + with self._lock, self._connect() as conn: + rows = conn.execute( + "SELECT cache_key, q_norm, answer, model, created_ts, ttl_ts," + " doc_version, hits FROM semcache LIMIT ?", (limit,)).fetchall() + return [dict(r) for r in rows] + + def put_semcache(self, cache_key: str, bucket: str, q_norm: str, answer: str, + model: str, created_ts: int, ttl_ts: int, + doc_version: int = 1) -> None: + """写/覆盖一条缓存(幂等)。""" + with self._lock, self._connect() as conn: + conn.execute( + "INSERT OR REPLACE INTO semcache" + "(cache_key, bucket, q_norm, answer, model, created_ts, ttl_ts," + " doc_version, hits) VALUES (?,?,?,?,?,?,?,?,0)", + (cache_key, bucket, q_norm, answer, model, created_ts, ttl_ts, + doc_version)) + + def promote_semcache(self, source_key: str, alias_key: str, hits: int) -> None: + """L2 -> L1 晋升:以别名键复制一行(原行保留审计)。""" + with self._lock, self._connect() as conn: + row = conn.execute( + "SELECT * FROM semcache WHERE cache_key = ?", (source_key,)).fetchone() + if row is None: + return + conn.execute( + "INSERT OR REPLACE INTO semcache" + "(cache_key, bucket, q_norm, answer, model, created_ts, ttl_ts," + " doc_version, hits) VALUES (?,?,?,?,?,?,?,?,?)", + (alias_key, row["bucket"], row["q_norm"], row["answer"], + row["model"], row["created_ts"], row["ttl_ts"], + row["doc_version"], hits)) + conn.execute("UPDATE semcache SET hits = ? WHERE cache_key = ?", + (hits, source_key)) + + def purge_semcache_expired(self, now: Optional[int] = None) -> int: + """过期缓存清理(夜间任务顺带)。""" + now = int(now if now is not None else time.time()) + with self._lock, self._connect() as conn: + cur = conn.execute("DELETE FROM semcache WHERE ttl_ts < ?", (now,)) + return cur.rowcount + # ---------- 自省(测试/验收用) ---------- def table_names(self) -> List[str]: """列出已建表名(测试验收)。""" diff --git a/gateway/proxy/semcache.py b/gateway/proxy/semcache.py index 00fd821..795558d 100644 --- a/gateway/proxy/semcache.py +++ b/gateway/proxy/semcache.py @@ -1,24 +1,261 @@ -"""语义缓存(T-P6 落地;本文件先立签名)。 +"""两级语义缓存(T-P6,M2):L1 精确 + L2 n-gram 倒排 + singleflight + SSE 回放。 -规格(§6):L1 精确 + L2 字符 2/3-gram 倒排(启动自 sqlite 重建), -加权 Jaccard(3-gram 权 2)+ 共享 gram>=3 门限 + 阈值 0.92, -TTL + LRU(max_entries),L2 命中 promote_frequency 次晋升 L1。 +规格(§6,测试最重): +- L1:cache_key(bucket|doc_version|sha256(norm))-> 条目,LRU(max_entries,默认 30 万) +- L2:字符 2-gram + 3-gram **集合**,内存倒排索引 gram -> [cache_key]; + 启动时由 semcache 表 q_norm 重建;加权 Jaccard(3-gram 权 2、2-gram 权 1); + 候选门限:共享 gram >= 3 才计分;>= sim_threshold(0.92) 命中; + **L2 命中累计 promote_frequency(5) 次晋升 L1** +- TTL:ttl_ts 过期不可见;命中即续期(滑动过期) +- 持久化:semcache 表(put 同步写,索引内存维护;调用方 to_thread,D-P10) """ from __future__ import annotations -from typing import Any, Dict, Optional +import asyncio +import json +import re +import time +from collections import OrderedDict +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +_WHITESPACE = re.compile(r"\s+") + + +def grams(text: str) -> set: + """字符 2-gram + 3-gram 集合(中文天然适配,无需分词)。""" + t = _WHITESPACE.sub("", (text or "").lower()) + out: set = set() + for n in (2, 3): + for i in range(len(t) - n + 1): + out.add(t[i:i + n]) + return out or ({t} if t else set()) + + +def weighted_jaccard(ga: set, gb: set) -> float: + """加权 Jaccard:交集中每个 3-gram 权 2、2-gram 权 1,除以并集加权。""" + if not ga or not gb: + return 0.0 + inter = ga & gb + if not inter: + return 0.0 + w_inter = sum(2 if g in (3,) or len(g) == 3 else 1 for g in inter) + w_union = sum(2 if len(g) == 3 else 1 for g in (ga | gb)) + return w_inter / w_union if w_union else 0.0 + + +class CacheEntry: + __slots__ = ("answer", "model", "q_norm", "g", "created_ts", "ttl_ts", + "doc_version", "hits") + + def __init__(self, answer: str, model: str, q_norm: str, + created_ts: float, ttl_ts: float, doc_version: int): + self.answer = answer + self.model = model + self.q_norm = q_norm + self.g = grams(q_norm) + self.created_ts = created_ts + self.ttl_ts = ttl_ts + self.doc_version = doc_version + self.hits = 0 class SemanticCache: - """两级语义缓存(T-P6 实现)。""" + """L1 精确 + L2 倒排(内存),sqlite 持久化(store 的 semcache 表)。""" - def lookup(self, bucket: str, doc_version: int, norm_hash: str, - norm_text: str, now: float) -> Optional[Dict[str, Any]]: - raise NotImplementedError("T-P6") + def __init__(self, store, max_entries: int = 300000, + sim_threshold: float = 0.92, promote_frequency: int = 5, + now: Optional[float] = None): + self.store = store + self.max_entries = max(1, int(max_entries)) + self.sim_threshold = float(sim_threshold) + self.promote_frequency = max(1, int(promote_frequency)) + self._l1: "OrderedDict[str, CacheEntry]" = OrderedDict() + self._inverted: Dict[str, set] = {} + self._clock = now # 假时钟注入(None = time.time) + self.hits_exact = 0 + self.hits_semantic = 0 + self._rebuild() - def put(self, bucket: str, doc_version: int, norm_hash: str, - norm_text: str, answer: str, model: str, now: float) -> None: - raise NotImplementedError("T-P6") + # ---------- 时钟 ---------- + def _now(self) -> float: + return time.time() if self._clock is None else self._clock + + # ---------- 启动重建 ---------- + def _rebuild(self) -> None: + try: + rows = self.store.semcache_rows(limit=self.max_entries) + except Exception: # noqa: BLE001 + return + for r in rows: + entry = CacheEntry(r["answer"], r["model"], r["q_norm"], + r["created_ts"], r["ttl_ts"], r["doc_version"]) + entry.hits = r["hits"] + self._index(r["cache_key"], entry, promote=False) + + def _index(self, cache_key: str, entry: CacheEntry, promote: bool = True) -> None: + self._l1[cache_key] = entry + self._l1.move_to_end(cache_key) + for g in entry.g: + self._inverted.setdefault(g, set()).add(cache_key) + if promote: + self._evict() + + def _evict(self) -> None: + """LRU 驱逐(超 max_entries 淘汰最久未用,含倒排回收)。""" + while len(self._l1) > self.max_entries: + key, entry = self._l1.popitem(last=False) + for g in entry.g: + bucket = self._inverted.get(g) + if bucket is not None: + bucket.discard(key) + if not bucket: + self._inverted.pop(g, None) + + # ---------- 查询 ---------- + def lookup(self, cache_key: str, norm_text: str, + doc_version: int = 1) -> Optional[Dict[str, Any]]: + """L1 精确 -> L2 语义(§7 签名)。返回 {answer, model, level} 或 None。""" + now = self._now() + # L1 + entry = self._l1.get(cache_key) + if entry is not None: + if entry.ttl_ts < now: + self._invalidate_key(cache_key) + else: + self.hits_exact += 1 + self._l1.move_to_end(cache_key) + return {"answer": entry.answer, "model": entry.model, + "level": "exact"} + # L2 + g = grams(norm_text) + if len(g) < 3: + return None + candidates: Dict[str, int] = {} + for gram in g: + for key in self._inverted.get(gram, ()): + candidates[key] = candidates.get(key, 0) + 1 + best_key = None + best_score = 0.0 + for key, shared in candidates.items(): + if shared < 3: + continue # 候选门限:共享 gram >= 3 + cand = self._l1.get(key) + if cand is None or cand.ttl_ts < now or cand.doc_version != doc_version: + continue + score = weighted_jaccard(g, cand.g) + if score > best_score: + best_score = score + best_key = key + if best_key is None or best_score < self.sim_threshold: + return None + cand = self._l1[best_key] + cand.hits += 1 + self.hits_semantic += 1 + promoted = False + if cand.hits >= self.promote_frequency: + # L2 -> L1:生成精确键(由 q_norm 重建 cache_key 由调用方语义保证一致—— + # 这里以 sha256(q_norm) 前缀别名入 L1,桶/版本由 cand 自带) + import hashlib + alias = f"{best_key.split('|')[0]}|{cand.doc_version}|promoted:" \ + f"{hashlib.sha256(cand.q_norm.encode()).hexdigest()[:16]}" + self._index(alias, cand, promote=True) + try: + self.store.promote_semcache(best_key, alias, cand.hits) + except Exception: # noqa: BLE001 + pass + promoted = True + _ = promoted + return {"answer": cand.answer, "model": cand.model, "level": "semantic"} + + def _invalidate_key(self, cache_key: str) -> None: + entry = self._l1.pop(cache_key, None) + if entry: + for g in entry.g: + bucket = self._inverted.get(g) + if bucket is not None: + bucket.discard(cache_key) + if not bucket: + self._inverted.pop(g, None) + + # ---------- 写入 ---------- + 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 签名)。""" + now = self._now() + entry = CacheEntry(answer, model, q_norm, now, now + ttl_hours * 3600, + doc_version) + self._index(cache_key, entry) + try: + self.store.put_semcache(cache_key, "default", q_norm, answer, model, + int(now), int(now + ttl_hours * 3600), + doc_version) + except Exception: # noqa: BLE001 + pass # 持久化失败不影响内存缓存(可重建) def stats(self) -> Dict[str, Any]: - raise NotImplementedError("T-P6") + return {"entries": len(self._l1), "hits_exact": self.hits_exact, + "hits_semantic": self.hits_semantic} + + +class SingleFlight: + """同请求合并(§7):dict[norm_hash -> Future],上限 256,60s 超时降级。""" + + MAX = 256 + TIMEOUT_S = 60.0 + + def __init__(self): + self._inflight: Dict[str, asyncio.Future] = {} + + def try_claim(self, norm_hash: str): + """返回 (future_or_None, slot)。future 非 None = 等待方;slot = 登记句柄。""" + fut = self._inflight.get(norm_hash) + if fut is not None: + return fut, None + if len(self._inflight) >= self.MAX: + return None, None # 超限旁路(不合并) + fut = asyncio.get_event_loop().create_future() + self._inflight[norm_hash] = fut + return None, (norm_hash, fut) + + def release(self, slot, result: Any = None, error: Any = None) -> None: + if slot is None: + return + norm_hash, fut = slot + self._inflight.pop(norm_hash, None) + if not fut.done(): + if error is not None: + fut.set_exception(error) + else: + fut.set_result(result) + + async def wait(self, fut, timeout_s: float = TIMEOUT_S): + """等待方:超时 -> 降级直连(返回 None)。""" + try: + return await asyncio.wait_for(asyncio.shield(fut), timeout=timeout_s) + except (asyncio.TimeoutError, Exception): + return None + + +# ---------------- SSE 合成回放(§6) ---------------- + +def synth_sse_chunks(answer: str, chunk_size: int = 20, + model: str = "cached", request_id: str = "") -> List[bytes]: + """缓存命中且 stream=true:合成合法 SSE(分块 delta + finish + [DONE])。""" + cid = f"chatcmpl-cached-{request_id or '0'}" + created = int(time.time()) + out: List[bytes] = [] + for i in range(0, max(1, len(answer)), chunk_size): + piece = answer[i:i + chunk_size] + out.append(("data: " + json.dumps({ + "id": cid, "object": "chat.completion.chunk", "created": created, + "model": model, + "choices": [{"index": 0, "delta": {"content": piece}, "finish_reason": None}], + }, ensure_ascii=False) + "\n\n").encode("utf-8")) + out.append(("data: " + json.dumps({ + "id": cid, "object": "chat.completion.chunk", "created": created, + "model": model, + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + }, ensure_ascii=False) + "\n\n").encode("utf-8")) + out.append(b"data: [DONE]\n\n") + return out diff --git a/tests/test_proxy_semcache.py b/tests/test_proxy_semcache.py new file mode 100644 index 0000000..d8206b8 --- /dev/null +++ b/tests/test_proxy_semcache.py @@ -0,0 +1,139 @@ +"""语义缓存测试(T-P6,M2):精确/n-gram 阈值/TTL/LRU/重建/晋升/singleflight/SSE 回放。""" +import asyncio +import json + +import pytest + +from gateway.proxy.semcache import ( + SingleFlight, + SemanticCache, + grams, + synth_sse_chunks, + weighted_jaccard, +) +from gateway.proxy.ledger import Ledger + + +@pytest.fixture() +def cache(tmp_path): + led = Ledger.init_db(tmp_path / "p.sqlite3") + return SemanticCache(led, max_entries=100, sim_threshold=0.92, + promote_frequency=5) + + +def _put(cache, key, text, answer="答案A", model="m", doc_version=1, ttl_hours=72): + cache.put(key, text, answer, model, doc_version=doc_version, ttl_hours=ttl_hours) + + +def test_grams_and_weighted_jaccard(): + g1 = grams("什么是递归") + assert any(len(g) == 2 for g in g1) and any(len(g) == 3 for g in g1) + assert weighted_jaccard(g1, g1) == 1.0 + assert weighted_jaccard(grams("完全不同话题"), g1) == 0.0 + + +def test_exact_hit_and_miss(cache): + key = "default|1|" + "a" * 16 + _put(cache, key, "什么是递归", "递归是自调用") + hit = cache.lookup(key, "什么是递归") + assert hit and hit["level"] == "exact" and hit["answer"] == "递归是自调用" + assert cache.lookup(key + "-nope", "完全无关的问题") is None + + +def test_semantic_hit_above_threshold(cache): + """同义变体:L2 命中(阈值上)。""" + key = "default|1|b1" + _put(cache, key, "请解释一下什么叫做递归函数", "递归解释") + hit = cache.lookup("default|1|b2", "请解释一下什么叫做递归函数", doc_version=1) + assert hit and hit["level"] == "semantic" + assert cache.hits_semantic == 1 + + +def test_semantic_miss_below_threshold(cache): + """完全不同语义:未命中(阈值下)。""" + _put(cache, "default|1|c1", "请解释一下什么叫做递归函数", "递归解释") + assert cache.lookup("default|1|c2", "今天股市行情怎么样", doc_version=1) is None + + +def test_ttl_expiry_fake_clock(cache): + _put(cache, "k", "某个问题文本", "旧答案", ttl_hours=1) + cache._clock = cache._now() + 7200 # 假时钟 +2h + assert cache.lookup("k", "某个问题文本", doc_version=1) is None # 过期不可见 + + +def test_lru_eviction(tmp_path): + led = Ledger.init_db(tmp_path / "p.sqlite3") + cache = SemanticCache(led, max_entries=3) + for i in range(5): + _put(cache, f"k{i}", f"完全不同的问题编号{i}", f"答{i}") + assert len(cache._l1) == 3 # LRU 上限 + assert cache.lookup("k0", "完全不同的问题编号0") is None # 最旧被驱逐 + assert cache.lookup("k4", "完全不同的问题编号4") is not None + + +def test_rebuild_from_sqlite(tmp_path): + """启动时由 semcache 表重建倒排索引。""" + led = Ledger.init_db(tmp_path / "p.sqlite3") + c1 = SemanticCache(led, max_entries=100) + c1.put("k", "解释递归的概念", "持久化答案", "m") + c2 = SemanticCache(led, max_entries=100) # 新实例:重建 + hit = c2.lookup("k2", "解释递归的概念", doc_version=1) + assert hit and hit["answer"] == "持久化答案" + + +def test_promote_after_five_semantic_hits(cache): + """L2 命中 5 次 -> 晋升 L1(promote 别名键可精确命中)。""" + key = "default|1|p1" + _put(cache, key, "请解释一下什么叫做递归函数呢?", "递归解释(变体)") + promoted = False + for i in range(5): + hit = cache.lookup("default|1|p%d" % (i + 2), "请解释一下什么叫做递归函数呢", doc_version=1) + assert hit and hit["level"] == "semantic" + if any(k.startswith("default|1|promoted:") for k in cache._l1): + promoted = True + assert promoted and cache.hits_semantic == 5 + + +def test_singleflight_merge_and_bypass(): + """两并发同请求:一登记一等待;超限旁路。""" + async def scenario(): + sf = SingleFlight() + fut, slot = sf.try_claim("h1") + assert fut is None and slot is not None # 首个登记 + fut2, slot2 = sf.try_claim("h1") + assert fut2 is not None and slot2 is None # 第二个等待 + sf.release(slot, result="共享答案") + got = await sf.wait(fut2) + assert got == "共享答案" + # 超限旁路 + sf2 = SingleFlight() + sf2.MAX = 2 + _f, s1 = sf2.try_claim("a") + _f2, s2 = sf2.try_claim("b") + f3, s3 = sf2.try_claim("c") + assert f3 is None and s3 is None # 第三个旁路 + + asyncio.run(scenario()) + + +def test_synth_sse_chunks_valid(): + """命中回放:合法 SSE 形状(delta 分块 + finish + [DONE])。""" + chunks = synth_sse_chunks("你好世界" * 10, chunk_size=20, model="m", request_id="r1") + text = b"".join(chunks).decode("utf-8") + assert text.count("chat.completion.chunk") >= 2 + assert '"finish_reason": "stop"' in text or '"finish_reason":"stop"' in text + assert text.rstrip("\n").endswith("data: [DONE]") + content = "" + done = False + for raw in chunks: + for line in raw.decode("utf-8").splitlines(): + if not line.startswith("data:"): + continue + payload = line[5:].strip() + if payload == "[DONE]": + done = True + continue + obj = json.loads(payload) + assert obj["object"] == "chat.completion.chunk" + content += obj["choices"][0]["delta"].get("content") or "" + assert done and "你好世界" in content diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 7ed252a..3e47a88 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -141,7 +141,7 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯 | T-P3 | 计价+结算:峰谷窗口/毫元整数/黄金用例 ≥10 组 | ✅ 完成 | T-P3 | | T-P4 | 路由端到端:/proxy/v1 非流式+流式+402/429/413 语义 | ✅ 完成 | T-P4 | | T-P5 | 规范化+桶:稳定哈希/整形顺序/doc_version 失效/多轮不缓存 | ✅ 完成 | T-P5 | -| T-P6 | 语义缓存:L1 精确+L2 n-gram 倒排+singleflight+TTL+失败不缓存 | ⬜ 待办 | | +| T-P6 | 语义缓存:L1 精确+L2 n-gram 倒排+singleflight+TTL+失败不缓存 | ✅ 完成 | T-P6 | | T-P7 | 管理面+前端:stats/ledger 端点+ProxyView 三卡片 | ⬜ 待办 | | | T-P8 | 压测+预算:200 并发流式;P99≤50ms/内存≤1GB/账目零不一致 | ⬜ 待办 | |