From 9dae7dda982d271d7f17d22fe44c84043842012f Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 6 Jul 2026 22:39:42 -0400 Subject: [PATCH] =?UTF-8?q?feat(adapters):=20=E5=AE=9E=E7=8E=B0=20RedisRes?= =?UTF-8?q?ponseCache?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit content-addressed sha256 缓存键，Redis 不可用时静默降级。 Co-Authored-By: Claude Opus 4.6 (1M context) --- adapters/redis_cache.py | 95 +++++++++++++++++++++++ tests/unit/test_redis_cache.py | 136 +++++++++++++++++++++++++++++++++ 2 files changed, 231 insertions(+) create mode 100644 adapters/redis_cache.py create mode 100644 tests/unit/test_redis_cache.py diff --git a/adapters/redis_cache.py b/adapters/redis_cache.py new file mode 100644 index 0000000..09358d0 --- /dev/null +++ b/adapters/redis_cache.py @@ -0,0 +1,95 @@ +"""Redis 响应缓存 — content-addressed SHA256 键，Redis 不可用时静默降级。""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +from typing import Any + +from loguru import logger + +from core.types import LLMResponse + + +class RedisResponseCache: + """基于 Redis 的 LLM 响应缓存。 + + 使用 content-addressed 策略：key = sha256(model + json(messages))， + 值为 JSON 序列化的 LLMResponse。 + + 当 Redis 不可用时静默降级：get 返回 None，set 吞异常并记录 warning。 + + Args: + redis: 异步 Redis 客户端实例（duck-typed，需支持 get/set 方法）。 + ttl_s: 缓存过期时间（秒）。 + """ + + def __init__(self, redis: Any, ttl_s: int) -> None: + self._redis = redis + self._ttl_s = ttl_s + + def _build_key(self, model: str, messages: list[dict[str, str]]) -> str: + """构造 content-addressed 缓存键。 + + Args: + model: 模型名称。 + messages: 消息列表。 + + Returns: + sha256 哈希字符串作为 Redis 键。 + """ + payload = json.dumps( + {"model": model, "messages": messages}, + sort_keys=True, + ensure_ascii=False, + ) + digest = hashlib.sha256(payload.encode("utf-8")).hexdigest() + return f"llm_cache:{digest}" + + async def get( + self, model: str, messages: list[dict[str, str]] + ) -> LLMResponse | None: + """从缓存读取 LLM 响应。 + + Args: + model: 模型名称。 + messages: 消息列表。 + + Returns: + 缓存命中时返回 LLMResponse，未命中或 Redis 异常时返回 None。 + """ + try: + key = self._build_key(model, messages) + raw = await self._redis.get(key) + except Exception: + logger.warning("Redis 缓存读取失败，降级为未命中") + return None + + if raw is None: + return None + + data = json.loads(raw) + return LLMResponse(**data) + + async def set( + self, + model: str, + messages: list[dict[str, str]], + response: LLMResponse, + ) -> None: + """将 LLM 响应写入缓存。 + + Args: + model: 模型名称。 + messages: 消息列表。 + response: 待缓存的 LLMResponse。 + """ + try: + key = self._build_key(model, messages) + value = json.dumps( + dataclasses.asdict(response), ensure_ascii=False + ) + await self._redis.set(key, value, ex=self._ttl_s) + except Exception: + logger.warning("Redis 缓存写入失败，跳过缓存") diff --git a/tests/unit/test_redis_cache.py b/tests/unit/test_redis_cache.py new file mode 100644 index 0000000..e4e6cfd --- /dev/null +++ b/tests/unit/test_redis_cache.py @@ -0,0 +1,136 @@ +"""RedisResponseCache 单元测试。""" + +from __future__ import annotations + +import pytest + +from adapters.redis_cache import RedisResponseCache +from core.types import LLMResponse + + +@pytest.fixture() +def fake_redis(): + """创建 fakeredis 异步实例。""" + try: + import fakeredis.aioredis + except ImportError: + pytest.skip("fakeredis not installed") + return fakeredis.aioredis.FakeRedis(decode_responses=True) + + +@pytest.fixture() +def sample_response() -> LLMResponse: + """构造一个标准测试用 LLMResponse。""" + return LLMResponse( + content="你好世界", + thinking="思考过程", + model="gpt-4o", + provider="openai", + prompt_tokens=10, + completion_tokens=5, + latency_ms=120, + ttft_ms=45.0, + max_inter_token_ms=12.3, + cache_hit=False, + call_id="test-call-001", + ) + + +MESSAGES = [{"role": "user", "content": "你好"}] + + +class _BrokenRedis: + """模拟 Redis 连接异常的假实现。""" + + async def get(self, _key: str) -> None: + raise ConnectionError("redis down") + + async def set(self, _key: str, _value: str, **_kw: object) -> None: + raise ConnectionError("redis down") + + +# ── 测试用例 ────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_cache_miss_returns_none(fake_redis: object) -> None: + """缓存未命中时返回 None。""" + cache = RedisResponseCache(redis=fake_redis, ttl_s=60) + result = await cache.get("gpt-4o", MESSAGES) + assert result is None + + +@pytest.mark.asyncio +async def test_cache_roundtrip( + fake_redis: object, sample_response: LLMResponse +) -> None: + """set 后 get 应返回相同内容。""" + cache = RedisResponseCache(redis=fake_redis, ttl_s=300) + await cache.set("gpt-4o", MESSAGES, sample_response) + result = await cache.get("gpt-4o", MESSAGES) + + assert result is not None + assert result.content == sample_response.content + assert result.thinking == sample_response.thinking + assert result.model == sample_response.model + assert result.provider == sample_response.provider + assert result.prompt_tokens == sample_response.prompt_tokens + assert result.completion_tokens == sample_response.completion_tokens + assert result.latency_ms == sample_response.latency_ms + assert result.ttft_ms == sample_response.ttft_ms + assert result.max_inter_token_ms == sample_response.max_inter_token_ms + assert result.cache_hit == sample_response.cache_hit + assert result.call_id == sample_response.call_id + + +@pytest.mark.asyncio +async def test_different_messages_different_keys( + fake_redis: object, sample_response: LLMResponse +) -> None: + """不同 messages 应产生不同缓存键。""" + cache = RedisResponseCache(redis=fake_redis, ttl_s=300) + messages_a = [{"role": "user", "content": "问题A"}] + messages_b = [{"role": "user", "content": "问题B"}] + + await cache.set("gpt-4o", messages_a, sample_response) + + assert await cache.get("gpt-4o", messages_a) is not None + assert await cache.get("gpt-4o", messages_b) is None + + +@pytest.mark.asyncio +async def test_different_models_different_keys( + fake_redis: object, sample_response: LLMResponse +) -> None: + """不同 model 应产生不同缓存键。""" + cache = RedisResponseCache(redis=fake_redis, ttl_s=300) + + await cache.set("gpt-4o", MESSAGES, sample_response) + + assert await cache.get("gpt-4o", MESSAGES) is not None + assert await cache.get("claude-3-opus", MESSAGES) is None + + +@pytest.mark.asyncio +async def test_graceful_degradation_on_error() -> None: + """Redis 不可用时静默降级：get→None，set→不报错。""" + cache = RedisResponseCache(redis=_BrokenRedis(), ttl_s=60) + response = LLMResponse( + content="x", + thinking="", + model="m", + provider="p", + prompt_tokens=0, + completion_tokens=0, + latency_ms=0, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + call_id="broken-test", + ) + + # get 应返回 None 而非抛异常 + assert await cache.get("m", MESSAGES) is None + + # set 应静默吞掉异常 + await cache.set("m", MESSAGES, response)