TtlCache(全输入进键/惰性过期/容量逐出/命中零 DB)+ search 接线 (空结果不缓存,保住 gap 埋点)+ invalidate_cache 真实现(T-P2-10)。 实现: - packages/retrieval/cache.py(新建):CacheKey 全输入进键(query/body_shape/ occasion/entity_types 排序归一/rule_boost 排序序列化哈希/top_k/失效代数); TtlCache 惰性过期 + OrderedDict LRU(超限 popitem 最旧)+ hits/misses/ evictions 计数与 stats();clock 可注入(单测免 sleep,默认 monotonic); asyncio 单循环不加锁的前提写在模块 docstring;正确性红线(只缓存不可变 值、命中零 DB)同处。纯逻辑,不 import 任何项目内模块(值类型由 service 以 PEP 695 泛型注入,避免循环依赖)。 - packages/retrieval/service.py:__init__ 加 keyword-only cache=None(默认 禁用,行为与现状一致,既有调用点不受影响);from_settings 按 Settings 装配进程级共享缓存(retrieval_cache_enabled=False → None;TTL/容量变化 时重建);search() 在 embed 之前查缓存——命中直接返回(零 DB、零 embed), 写入与命中都做防御性拷贝(results 是 list,调用方拿不到缓存内引用), 仅有结果时写缓存(空结果不缓存,record_gap 每次都跑);invalidate_cache() 签名不变:自增 repo.INVALIDATION_VERSION(代数失效,旧代键永不命中, 与 T12 的契约保持)+ 清空共享实例(仅为立即回收内存,失效正确性只 依赖代数)。repo.py 未做任何改动。 测试: - 单测 tests/unit/test_retrieval_cache.py(21 例,fake repo/embedding + 注入时钟,不连库):同参二连查零 DB 且结果相等、键分量矩阵(query/ occasion/body_shape/entity_types/rule_boost/top_k 任一变化 → miss)、 TTL 过期边界(60.0 过期/59.9 命中)、LRU 逐出与命中率/计数、空结果 不缓存且 gap 埋点每次触发、invalidate_cache() 代数失效(含自增语义 保留断言)、cache=None 与现状一致、from_settings 装配开关与共享实例、 rule_boost 哈希确定性。全量单测 494 passed(基线 473 + 新增 21)。 - 集成 tests/integration/test_retrieval_cache.py(2 例,真实 PG + pgvector): 发布新实体 + invalidate_cache() 后同参检索立即可见(失效前缓存仍只吐 旧实体,证明缓存在工作——T-P2-10 验收语义);不失效、TTL=1s 到期 (asyncio.sleep 1.1s)后新实体自然可见(双路径验收)。集成全量 40 passed。 偏差说明:无(invalidate_cache() 在代数失效之外顺带清空共享缓存实例, 失效正确性只依赖代数,清空仅为立即释放内存)。
171 lines
6.8 KiB
Python
171 lines
6.8 KiB
Python
"""检索结果缓存的纯逻辑:全输入缓存键 + TTL/LRU 淘汰(T-P2-10).
|
||
|
||
刻意与 service 分离:键构造与淘汰策略必须能被单测逐项断言,混进
|
||
service 就会把缓存语义测试拖进 DB 打桩。本模块不 import 任何项目内
|
||
模块(``SearchOutcome`` 由 service 以泛型参数注入,避免 service ↔
|
||
cache 循环依赖),``INVALIDATION_VERSION`` 只是键里的一个普通整数。
|
||
|
||
正确性红线(改本模块前先读):
|
||
|
||
1. **全部影响结果的输入都必须进缓存键**(见 :class:`CacheKey`)——缺
|
||
一个分量就会把 A 条件的结果串给 B 条件。``rule_boost`` 是
|
||
``dict[int, float]``,按键排序序列化再哈希;
|
||
2. **只缓存不可变值**:值应是 frozen dataclass(``SearchOutcome``)。
|
||
它内部的 ``results`` 是 list(可变)——写入与命中时 service 都做
|
||
防御性拷贝,调用方永远拿不到缓存内 list 的引用,原地修改不可能
|
||
污染缓存。绝不缓存 ORM 对象或任何可变引用;
|
||
3. **命中路径零 DB**:缓存命中时不碰 session、不 embed、不重排,
|
||
直接返回缓存值。
|
||
|
||
线程安全:**不加锁,也别加**。检索跑在 asyncio 单事件循环里,
|
||
get/put 各自是一段无 await 的同步代码,事件循环只在 await 点切换
|
||
协程,单循环内不存在竞态窗口。加锁反而暗示它可以跨线程使用——那
|
||
是错误的使用方式;若未来把检索挪进线程池,必须重新评估本类。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import time
|
||
from collections import OrderedDict
|
||
from collections.abc import Callable, Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
|
||
|
||
def rule_boost_hash(rule_boost: Mapping[int, float]) -> str:
|
||
"""``rule_boost`` 的稳定哈希:按键排序序列化后 sha256.
|
||
|
||
dict 迭代顺序是插入序:``{1: 0.5, 2: 0.3}`` 与先放 2 再放 1 的同义
|
||
dict 逐项迭代结果不同,必须排序后再序列化,否则同义输入得到不同
|
||
键(白丢命中率)。float 用 ``repr`` 保留精确位,避免 ``str`` 截断
|
||
把不同的 boost 折叠成同一个键。
|
||
"""
|
||
payload = ",".join(f"{cid!r}:{boost!r}" for cid, boost in sorted(rule_boost.items()))
|
||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class CacheKey:
|
||
"""一次检索的全部输入指纹:任一分量不同就是另一次检索.
|
||
|
||
- ``version``:``repo.INVALIDATION_VERSION`` 的快照。发布/归档后
|
||
自增,新查找全部用新代构键,旧代条目(含检索途中被失效后写回的)
|
||
永不命中——这就是代数失效,不需要遍历删除;
|
||
- ``entity_types`` 排序归一:SQL 的 ``= ANY(...)`` 与顺序无关,
|
||
同集合不同顺序应共享同一条目;
|
||
- ``rule_boost_hash``:见 :func:`rule_boost_hash`。
|
||
"""
|
||
|
||
version: int
|
||
query_text: str
|
||
body_shape: str | None
|
||
occasion: str | None
|
||
entity_types: tuple[str, ...]
|
||
rule_boost_hash: str
|
||
top_k: int
|
||
|
||
@classmethod
|
||
def build(
|
||
cls,
|
||
*,
|
||
version: int,
|
||
query_text: str,
|
||
top_k: int,
|
||
body_shape: str | None = None,
|
||
occasion: str | None = None,
|
||
entity_types: Sequence[str] = (),
|
||
rule_boost: Mapping[int, float] | None = None,
|
||
) -> CacheKey:
|
||
"""从 search 的入参构造键(``rule_boost`` 在这里收敛成哈希)."""
|
||
return cls(
|
||
version=version,
|
||
query_text=query_text,
|
||
body_shape=body_shape,
|
||
occasion=occasion,
|
||
entity_types=tuple(sorted(entity_types)),
|
||
rule_boost_hash=rule_boost_hash(rule_boost or {}),
|
||
top_k=top_k,
|
||
)
|
||
|
||
|
||
class TtlCache[V]:
|
||
"""进程内 TTL + 容量 LRU 缓存:读时惰性过期,超限逐出最旧.
|
||
|
||
指标(``hits``/``misses``/``evictions``)先暴露成属性,Prometheus
|
||
接入时直接读,:meth:`stats` 给出汇总快照。
|
||
|
||
线程安全前提见模块 docstring:asyncio 单循环使用,不加锁。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
ttl_seconds: float,
|
||
max_entries: int,
|
||
clock: Callable[[], float] | None = None,
|
||
) -> None:
|
||
if ttl_seconds <= 0:
|
||
raise ValueError(f"ttl_seconds 必须为正, 得到 {ttl_seconds}")
|
||
if max_entries < 1:
|
||
raise ValueError(f"max_entries 必须至少为 1, 得到 {max_entries}")
|
||
self._ttl_seconds = ttl_seconds
|
||
self._max_entries = max_entries
|
||
# monotonic 而不是 wall clock:TTL 是时长语义,NTP 校时/回拨不该
|
||
# 让缓存集体"突然过期"或"永不过期"。clock 可注入是为了单测免 sleep。
|
||
self._clock = clock or time.monotonic
|
||
self._entries: OrderedDict[CacheKey, tuple[float, V]] = OrderedDict()
|
||
self.hits = 0
|
||
self.misses = 0
|
||
self.evictions = 0
|
||
|
||
@property
|
||
def ttl_seconds(self) -> float:
|
||
"""当前 TTL(秒)。from_settings 据此判断共享缓存是否需要重建."""
|
||
return self._ttl_seconds
|
||
|
||
@property
|
||
def max_entries(self) -> int:
|
||
"""容量上限。用途同上."""
|
||
return self._max_entries
|
||
|
||
def get(self, key: CacheKey) -> V | None:
|
||
"""命中返回值并移到 LRU 末尾;未命中/已过期返回 ``None``.
|
||
|
||
过期是惰性的:读时发现超龄就地删除并按 miss 计——evictions 只
|
||
描述容量压力(LRU 逐出),不掺入时间淘汰,两个数字语义分开。
|
||
"""
|
||
entry = self._entries.get(key)
|
||
if entry is None:
|
||
self.misses += 1
|
||
return None
|
||
stored_at, value = entry
|
||
if self._clock() - stored_at >= self._ttl_seconds:
|
||
del self._entries[key]
|
||
self.misses += 1
|
||
return None
|
||
self._entries.move_to_end(key) # 刚被访问过,是最新的
|
||
self.hits += 1
|
||
return value
|
||
|
||
def put(self, key: CacheKey, value: V) -> None:
|
||
"""写入/覆盖一个条目,并保证容量不超限(超出逐出最旧)."""
|
||
self._entries[key] = (self._clock(), value)
|
||
self._entries.move_to_end(key)
|
||
while len(self._entries) > self._max_entries:
|
||
self._entries.popitem(last=False)
|
||
self.evictions += 1
|
||
|
||
def clear(self) -> None:
|
||
"""整体清空。代数失效本身不需要它(旧代键永不命中),这里只为
|
||
``invalidate_cache()`` 立刻释放内存,不等 TTL/LRU 兜底。"""
|
||
self._entries.clear()
|
||
|
||
def stats(self) -> dict[str, int]:
|
||
"""命中/未命中/逐出/当前条目数(观测快照)."""
|
||
return {
|
||
"hits": self.hits,
|
||
"misses": self.misses,
|
||
"evictions": self.evictions,
|
||
"entries": len(self._entries),
|
||
}
|