fix: gate cache writes and reads on a shared cacheability verdict
Cacheable is now a strict subset of deliverable: salvaged streams, content-less results and length/filter/unknown finishes are delivered but never written, and the read path applies the same verdict to the stored values before hit overrides (invalid stream_end or non-string finish read as a miss; 1.3.9 fidelity entries fall back to the completion's finish). Hits reset stream_end. The legacy _has_completion_output helper is folded into _is_cacheable. Red: 13 tests in tests/unit/test_cache.py failed before this change (uncacheable results were written and replayed; new fields not reset).
This commit is contained in:
@@ -24,7 +24,8 @@ from polygateway.types import (
|
|||||||
OpenAIChatCompletion,
|
OpenAIChatCompletion,
|
||||||
ResponseMode,
|
ResponseMode,
|
||||||
ThinkingObservation,
|
ThinkingObservation,
|
||||||
_has_completion_output,
|
_effective_finish,
|
||||||
|
_is_cacheable,
|
||||||
)
|
)
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
@@ -148,6 +149,54 @@ def build_cache_key(
|
|||||||
return _KEY_PREFIX + hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
return _KEY_PREFIX + hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||||||
|
|
||||||
|
|
||||||
|
def _response_cacheable(mode: ResponseMode, response: LLMResponse) -> bool:
|
||||||
|
"""写路径判据;与读路径共用 `_is_cacheable` 与 `_effective_finish`,口径不分叉。"""
|
||||||
|
return _is_cacheable(
|
||||||
|
mode=mode,
|
||||||
|
content=response.content,
|
||||||
|
finish_reason=_effective_finish(response.finish_reason, response.completion),
|
||||||
|
stream_end=response.stream_end,
|
||||||
|
completion=response.completion,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _restore_fields(raw: str, mode: ResponseMode) -> dict[str, Any]:
|
||||||
|
"""缓存原文 → LLMResponse 字段;不可缓存或形态坏抛 ValueError(调用方按 miss)。
|
||||||
|
|
||||||
|
可缓存判定对**存储原值**进行、先于命中覆盖: 覆盖后 `stream_end` 恒 None,
|
||||||
|
在那之后判会把 salvaged 条目放行(设计 §6)。
|
||||||
|
"""
|
||||||
|
data = json.loads(raw)
|
||||||
|
fields = {k: v for k, v in data.items() if k in _RESPONSE_FIELDS}
|
||||||
|
completion = fields.get("completion")
|
||||||
|
if completion is not None:
|
||||||
|
if not isinstance(completion, dict):
|
||||||
|
raise ValueError("缓存 completion 必须为对象")
|
||||||
|
fields["completion"] = OpenAIChatCompletion(**completion)
|
||||||
|
finish = _effective_finish(fields.get("finish_reason"), fields.get("completion"))
|
||||||
|
if finish is not None and not isinstance(finish, str):
|
||||||
|
raise ValueError("缓存 finish_reason 必须为字符串或 null")
|
||||||
|
if not _is_cacheable(
|
||||||
|
mode=mode,
|
||||||
|
content=fields.get("content", ""),
|
||||||
|
finish_reason=finish,
|
||||||
|
stream_end=fields.get("stream_end"),
|
||||||
|
completion=fields.get("completion"),
|
||||||
|
):
|
||||||
|
# 非法 stream_end 也落这里: 它关乎内容完整性,不像 thinking_observation
|
||||||
|
# 那样可以宽容降级
|
||||||
|
raise ValueError("缓存条目不可缓存(截断/空正文/非自然结束/非法 stream_end)")
|
||||||
|
fields["finish_reason"] = finish
|
||||||
|
# JSON 里存的是 StrEnum 的字符串值,不转就复活成裸 str,与字段注解分叉
|
||||||
|
# (下游 `is ThinkingObservation.OBSERVED` 会在命中路径上静默为 False);
|
||||||
|
# 键缺失即升级前写入的旧条目,交给 dataclass 默认值
|
||||||
|
if "thinking_observation" in fields:
|
||||||
|
fields["thinking_observation"] = _coerce_observation(fields["thinking_observation"])
|
||||||
|
if "applied_effort" in fields:
|
||||||
|
fields["applied_effort"] = _coerce_applied_effort(fields["applied_effort"])
|
||||||
|
return fields
|
||||||
|
|
||||||
|
|
||||||
class CacheMW:
|
class CacheMW:
|
||||||
"""洋葱第二层(遥测内、结构化外)。
|
"""洋葱第二层(遥测内、结构化外)。
|
||||||
|
|
||||||
@@ -195,6 +244,11 @@ class CacheMW:
|
|||||||
if hit is not None:
|
if hit is not None:
|
||||||
return hit
|
return hit
|
||||||
response = await call_next(request)
|
response = await call_next(request)
|
||||||
|
if not _response_cacheable(request.response_mode, response):
|
||||||
|
# 交付照常,只是不进 TTL: 截断/空正文/非自然结束固化后会把一次偶然
|
||||||
|
# 变成必然(设计 §6)。debug 级: 这是常规分流,不是故障
|
||||||
|
logger.debug("结果不可缓存(截断/空正文/非自然结束),跳过写入")
|
||||||
|
return response
|
||||||
try:
|
try:
|
||||||
serialized = self._serialize(response)
|
serialized = self._serialize(response)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
@@ -208,31 +262,15 @@ class CacheMW:
|
|||||||
def _rehydrate(self, raw: str, request: ChatRequest) -> LLMResponse | None:
|
def _rehydrate(self, raw: str, request: ChatRequest) -> LLMResponse | None:
|
||||||
"""反序列化 + 按本次调用的 structured 档零网络重建;失败按未命中。"""
|
"""反序列化 + 按本次调用的 structured 档零网络重建;失败按未命中。"""
|
||||||
try:
|
try:
|
||||||
data = json.loads(raw)
|
fields = _restore_fields(raw, request.response_mode)
|
||||||
fields = {k: v for k, v in data.items() if k in _RESPONSE_FIELDS}
|
|
||||||
completion = fields.get("completion")
|
|
||||||
if completion is not None:
|
|
||||||
if not isinstance(completion, dict):
|
|
||||||
raise ValueError("缓存 completion 必须为对象")
|
|
||||||
fields["completion"] = OpenAIChatCompletion(**completion)
|
|
||||||
if request.response_mode == "openai":
|
|
||||||
if fields.get("completion") is None:
|
|
||||||
raise ValueError("保真缓存缺 completion")
|
|
||||||
if not _has_completion_output(fields.get("content", ""), fields["completion"]):
|
|
||||||
raise ValueError("保真缓存无有效正文或工具产物")
|
|
||||||
structured_data = self._rebuild_structured(fields.get("content", ""), request)
|
structured_data = self._rebuild_structured(fields.get("content", ""), request)
|
||||||
# JSON 里存的是 StrEnum 的字符串值,不转就复活成裸 str,与字段注解分叉
|
|
||||||
# (下游 `is ThinkingObservation.OBSERVED` 会在命中路径上静默为 False);
|
|
||||||
# 键缺失即升级前写入的旧条目,交给 dataclass 默认值
|
|
||||||
if "thinking_observation" in fields:
|
|
||||||
fields["thinking_observation"] = _coerce_observation(fields["thinking_observation"])
|
|
||||||
if "applied_effort" in fields:
|
|
||||||
fields["applied_effort"] = _coerce_applied_effort(fields["applied_effort"])
|
|
||||||
fields.update(
|
fields.update(
|
||||||
cache_hit=True,
|
cache_hit=True,
|
||||||
latency_ms=0,
|
latency_ms=0,
|
||||||
ttft_ms=None,
|
ttft_ms=None,
|
||||||
max_inter_token_ms=None,
|
max_inter_token_ms=None,
|
||||||
|
# 结束方式描述的是"那次流",命中没有流可言(与 ttft_ms 同口径)
|
||||||
|
stream_end=None,
|
||||||
call_id=str(uuid.uuid4()),
|
call_id=str(uuid.uuid4()),
|
||||||
structured_data=structured_data,
|
structured_data=structured_data,
|
||||||
# 显式覆盖: 历史条目里的 `call_stats` 是个 dict,而 `_RESPONSE_FIELDS`
|
# 显式覆盖: 历史条目里的 `call_stats` 是个 dict,而 `_RESPONSE_FIELDS`
|
||||||
|
|||||||
@@ -663,23 +663,14 @@ def _is_cacheable(
|
|||||||
if _classify_finish(finish_reason) not in ("natural", "absent"):
|
if _classify_finish(finish_reason) not in ("natural", "absent"):
|
||||||
return False
|
return False
|
||||||
if mode == "openai":
|
if mode == "openai":
|
||||||
return completion is not None and _has_completion_output(content, completion)
|
# 判原文而非投影: 协议原文为空时不信任缓存里的旧投影(1.3.9 同口径)
|
||||||
return content.strip() != ""
|
if completion is None:
|
||||||
|
|
||||||
|
|
||||||
def _has_completion_output(content: str, completion: OpenAIChatCompletion | None) -> bool:
|
|
||||||
"""成功须有投影正文或完整工具;协议原文为空时不信任缓存旧投影。
|
|
||||||
|
|
||||||
只判定产物存在性,不替代协议结构校验,也不收紧快照类型的构造用途。
|
|
||||||
transport 与缓存共用此规则,避免空响应从缓存旁路成功。
|
|
||||||
"""
|
|
||||||
if completion is not None:
|
|
||||||
if completion.message.get("tool_calls"):
|
|
||||||
return True
|
|
||||||
raw_content = completion.message.get("content")
|
|
||||||
if not isinstance(raw_content, str) or not raw_content.strip():
|
|
||||||
return False
|
return False
|
||||||
return isinstance(content, str) and bool(content.strip())
|
raw = completion.message.get("content")
|
||||||
|
return _summarize_message(
|
||||||
|
completion.message, raw if isinstance(raw, str) else ""
|
||||||
|
).body_output
|
||||||
|
return content.strip() != ""
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
|
|||||||
@@ -30,6 +30,13 @@ def _resp(content="cached"):
|
|||||||
return LLMResponse(content, "", "m", "p", 1, 2, 30, None, None, False, "orig")
|
return LLMResponse(content, "", "m", "p", 1, 2, 30, None, None, False, "orig")
|
||||||
|
|
||||||
|
|
||||||
|
def _const(response):
|
||||||
|
async def terminal(request):
|
||||||
|
return response
|
||||||
|
|
||||||
|
return terminal
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
async def redis_cache():
|
async def redis_cache():
|
||||||
import redis.asyncio as aioredis
|
import redis.asyncio as aioredis
|
||||||
@@ -82,6 +89,23 @@ class TestRealRedisRoundtrip:
|
|||||||
assert first.cache_hit is False and second.cache_hit is True
|
assert first.cache_hit is False and second.cache_hit is True
|
||||||
assert calls["n"] == 1
|
assert calls["n"] == 1
|
||||||
|
|
||||||
|
async def test_completion_verdict_fields_roundtrip(self, redis_cache, namespace):
|
||||||
|
"""1.3.10: finish_reason/stream_end 经真实 Redis 往返;不可缓存结果不落库。"""
|
||||||
|
mw = CacheMW(
|
||||||
|
backend=redis_cache, model_fingerprint="m", default_namespace=namespace, ttl_s=60
|
||||||
|
)
|
||||||
|
from dataclasses import replace
|
||||||
|
|
||||||
|
ok = replace(_resp(), finish_reason="stop", stream_end="eof_after_finish")
|
||||||
|
first = await mw(ChatRequest(messages=_MSGS), _const(ok))
|
||||||
|
second = await mw(ChatRequest(messages=_MSGS), _const(ok))
|
||||||
|
assert first.cache_hit is False and second.cache_hit is True
|
||||||
|
assert second.finish_reason == "stop" and second.stream_end is None
|
||||||
|
truncated = replace(_resp(), stream_end="salvaged")
|
||||||
|
other = [{"role": "user", "content": "integration-salvaged"}]
|
||||||
|
await mw(ChatRequest(messages=other), _const(truncated))
|
||||||
|
assert await redis_cache.get(build_cache_key("m", other, namespace, None)) is None
|
||||||
|
|
||||||
async def test_namespace_isolation_on_shared_instance(self, redis_cache, namespace):
|
async def test_namespace_isolation_on_shared_instance(self, redis_cache, namespace):
|
||||||
"""多项目共用同一 Redis 时,namespace 不同绝不互相命中(防毒化铁律)。"""
|
"""多项目共用同一 Redis 时,namespace 不同绝不互相命中(防毒化铁律)。"""
|
||||||
other = f"pgw:test:{uuid.uuid4().hex}"
|
other = f"pgw:test:{uuid.uuid4().hex}"
|
||||||
|
|||||||
@@ -1002,3 +1002,117 @@ async def test_openai_tool_only_cache_hits_without_origin():
|
|||||||
hit = await mw(request, terminal)
|
hit = await mw(request, terminal)
|
||||||
assert hit.cache_hit and terminal.calls == 0
|
assert hit.cache_hit and terminal.calls == 0
|
||||||
assert hit.completion == completion
|
assert hit.completion == completion
|
||||||
|
|
||||||
|
|
||||||
|
# —— 1.3.10(#30/#31): 可缓存 ⊂ 可交付,读写共用同一判据(设计 §6)——
|
||||||
|
|
||||||
|
_OK_COMPLETION = OpenAIChatCompletion({"role": "assistant", "content": "ok"}, "stop", "up", None)
|
||||||
|
_THINK_LENGTH = OpenAIChatCompletion(
|
||||||
|
{"role": "assistant", "reasoning_content": "t"}, "length", "up", None
|
||||||
|
)
|
||||||
|
_THINK_STOP = OpenAIChatCompletion(
|
||||||
|
{"role": "assistant", "reasoning_content": "t"}, "stop", "up", None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("mode", "response"),
|
||||||
|
[
|
||||||
|
("text", _resp(stream_end="salvaged")),
|
||||||
|
("text", _resp(finish_reason="length")),
|
||||||
|
("text", _resp(finish_reason="content_filter")),
|
||||||
|
("text", _resp(finish_reason="end_turn")),
|
||||||
|
("openai", _resp(content="", finish_reason="length", completion=_THINK_LENGTH)),
|
||||||
|
("openai", _resp(content="", finish_reason="stop", completion=_THINK_STOP)),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_uncacheable_results_are_delivered_but_not_written(mode, response):
|
||||||
|
"""截断/空正文/非自然结束: 照常交回,但不写缓存,下次回源。"""
|
||||||
|
backend = InMemoryCache()
|
||||||
|
mw = _mw(backend)
|
||||||
|
terminal = _Terminal(response)
|
||||||
|
request = ChatRequest(messages=_MSGS, response_mode=mode)
|
||||||
|
assert (await mw(request, terminal)) is response
|
||||||
|
assert (await mw(request, terminal)).cache_hit is False
|
||||||
|
assert terminal.calls == 2 and not backend._store
|
||||||
|
|
||||||
|
|
||||||
|
def _stored(mode, **fields):
|
||||||
|
"""直接注入一条缓存原文(模拟旧版本或别的进程写入)。"""
|
||||||
|
backend = InMemoryCache()
|
||||||
|
data = dataclasses.asdict(_resp(**fields))
|
||||||
|
data.pop("structured_data", None)
|
||||||
|
data.pop("call_stats", None)
|
||||||
|
key = build_cache_key("m", _MSGS, "proj", None, response_mode=mode)
|
||||||
|
backend._store[key] = (float("inf"), json.dumps(data))
|
||||||
|
return backend, key
|
||||||
|
|
||||||
|
|
||||||
|
async def _hit(backend, mode):
|
||||||
|
terminal = _Terminal(_resp(content="fresh", completion=_OK_COMPLETION, finish_reason="stop"))
|
||||||
|
response = await _mw(backend)(ChatRequest(messages=_MSGS, response_mode=mode), terminal)
|
||||||
|
return response, terminal.calls
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("mode", "fields"),
|
||||||
|
[
|
||||||
|
("text", {"stream_end": "salvaged"}),
|
||||||
|
("text", {"stream_end": "bogus"}),
|
||||||
|
("text", {"finish_reason": "length"}),
|
||||||
|
("text", {"finish_reason": 5}),
|
||||||
|
("openai", {"content": "", "completion": _THINK_LENGTH, "finish_reason": "length"}),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_injected_uncacheable_entries_read_as_miss(mode, fields):
|
||||||
|
"""读路径对存储原值判定(覆盖之前): 不可缓存即 miss 回源。"""
|
||||||
|
backend, _ = _stored(mode, **fields)
|
||||||
|
response, calls = await _hit(backend, mode)
|
||||||
|
assert response.cache_hit is False and calls == 1
|
||||||
|
|
||||||
|
|
||||||
|
async def test_legacy_fidelity_length_entry_falls_back_to_completion_finish():
|
||||||
|
"""1.3.9 保真条目无顶层 finish: 回退 completion.finish_reason 识别为不可缓存。"""
|
||||||
|
completion = OpenAIChatCompletion({"role": "assistant", "content": "a"}, "length", None, None)
|
||||||
|
backend, key = _stored("openai", completion=completion)
|
||||||
|
data = json.loads(backend._store[key][1])
|
||||||
|
for name in ("finish_reason", "stream_end"):
|
||||||
|
data.pop(name)
|
||||||
|
backend._store[key] = (float("inf"), json.dumps(data))
|
||||||
|
response, calls = await _hit(backend, "openai")
|
||||||
|
assert response.cache_hit is False and calls == 1
|
||||||
|
|
||||||
|
|
||||||
|
async def test_legacy_text_entry_without_new_fields_still_hits():
|
||||||
|
"""1.3.9 文本条目无新键: 仍命中;命中 stream_end=None。"""
|
||||||
|
backend, key = _stored("text")
|
||||||
|
data = json.loads(backend._store[key][1])
|
||||||
|
for name in ("finish_reason", "stream_end"):
|
||||||
|
data.pop(name)
|
||||||
|
backend._store[key] = (float("inf"), json.dumps(data))
|
||||||
|
response, calls = await _hit(backend, "text")
|
||||||
|
assert response.cache_hit is True and calls == 0
|
||||||
|
assert response.stream_end is None and response.finish_reason is None
|
||||||
|
|
||||||
|
|
||||||
|
async def test_hit_resets_stream_end_and_keeps_effective_finish():
|
||||||
|
"""命中: stream_end 置 None;finish_reason 取判定所用值(顶层缺省时回退)。"""
|
||||||
|
backend, key = _stored("openai", completion=_OK_COMPLETION, stream_end="eof_after_finish")
|
||||||
|
data = json.loads(backend._store[key][1])
|
||||||
|
data.pop("finish_reason")
|
||||||
|
backend._store[key] = (float("inf"), json.dumps(data))
|
||||||
|
response, calls = await _hit(backend, "openai")
|
||||||
|
assert response.cache_hit is True and calls == 0
|
||||||
|
assert response.stream_end is None and response.finish_reason == "stop"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_new_fields_roundtrip_through_cache():
|
||||||
|
"""两新字段随序列化往返;命中只重置 stream_end。"""
|
||||||
|
backend = InMemoryCache()
|
||||||
|
mw = _mw(backend)
|
||||||
|
terminal = _Terminal(_resp(finish_reason="stop", stream_end="eof_after_finish"))
|
||||||
|
await mw(ChatRequest(messages=_MSGS), terminal)
|
||||||
|
stored = json.loads(next(iter(backend._store.values()))[1])
|
||||||
|
assert stored["finish_reason"] == "stop" and stored["stream_end"] == "eof_after_finish"
|
||||||
|
hit = await mw(ChatRequest(messages=_MSGS), terminal)
|
||||||
|
assert hit.cache_hit is True and hit.finish_reason == "stop" and hit.stream_end is None
|
||||||
|
|||||||
Reference in New Issue
Block a user