diff --git a/src/polygateway/middleware/cache.py b/src/polygateway/middleware/cache.py index e199f62..e46440d 100644 --- a/src/polygateway/middleware/cache.py +++ b/src/polygateway/middleware/cache.py @@ -24,7 +24,8 @@ from polygateway.types import ( OpenAIChatCompletion, ResponseMode, ThinkingObservation, - _has_completion_output, + _effective_finish, + _is_cacheable, ) if TYPE_CHECKING: @@ -148,6 +149,54 @@ def build_cache_key( 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: """洋葱第二层(遥测内、结构化外)。 @@ -195,6 +244,11 @@ class CacheMW: if hit is not None: return hit response = await call_next(request) + if not _response_cacheable(request.response_mode, response): + # 交付照常,只是不进 TTL: 截断/空正文/非自然结束固化后会把一次偶然 + # 变成必然(设计 §6)。debug 级: 这是常规分流,不是故障 + logger.debug("结果不可缓存(截断/空正文/非自然结束),跳过写入") + return response try: serialized = self._serialize(response) except Exception as exc: @@ -208,31 +262,15 @@ class CacheMW: def _rehydrate(self, raw: str, request: ChatRequest) -> LLMResponse | None: """反序列化 + 按本次调用的 structured 档零网络重建;失败按未命中。""" try: - 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) - 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("保真缓存无有效正文或工具产物") + fields = _restore_fields(raw, request.response_mode) 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( cache_hit=True, latency_ms=0, ttft_ms=None, max_inter_token_ms=None, + # 结束方式描述的是"那次流",命中没有流可言(与 ttft_ms 同口径) + stream_end=None, call_id=str(uuid.uuid4()), structured_data=structured_data, # 显式覆盖: 历史条目里的 `call_stats` 是个 dict,而 `_RESPONSE_FIELDS` diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 6268a84..94aa0ec 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -663,23 +663,14 @@ def _is_cacheable( if _classify_finish(finish_reason) not in ("natural", "absent"): return False if mode == "openai": - return completion is not None and _has_completion_output(content, completion) - return content.strip() != "" - - -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(): + # 判原文而非投影: 协议原文为空时不信任缓存里的旧投影(1.3.9 同口径) + if completion is None: 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) diff --git a/tests/integration/test_redis_cache.py b/tests/integration/test_redis_cache.py index 341c69b..ce9f32f 100644 --- a/tests/integration/test_redis_cache.py +++ b/tests/integration/test_redis_cache.py @@ -30,6 +30,13 @@ def _resp(content="cached"): 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 async def redis_cache(): import redis.asyncio as aioredis @@ -82,6 +89,23 @@ class TestRealRedisRoundtrip: assert first.cache_hit is False and second.cache_hit is True 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): """多项目共用同一 Redis 时,namespace 不同绝不互相命中(防毒化铁律)。""" other = f"pgw:test:{uuid.uuid4().hex}" diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py index 959474a..6afabd0 100644 --- a/tests/unit/test_cache.py +++ b/tests/unit/test_cache.py @@ -1002,3 +1002,117 @@ async def test_openai_tool_only_cache_hits_without_origin(): hit = await mw(request, terminal) assert hit.cache_hit and terminal.calls == 0 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