""" Cache Service — Unified Caching & Semantic Cache ================================================= Unified caching layer with two backends: - Redis (local dev / Docker) — uses REDIS_URL env var - In-memory dict (HF Spaces) — used when REDIS_URL is not set Features: - Key-value caching with TTL - Rate-limit counter tracking - Semantic Caching (avoids redundant LLM generation for semantically matching queries) """ from __future__ import annotations import asyncio import json import time from typing import Any, Dict, Optional, Tuple, List import numpy as np import structlog logger = structlog.get_logger(__name__) class InMemoryCache: """Simple thread-safe TTL in-memory cache used when Redis is unavailable.""" def __init__(self): self._store: Dict[str, Tuple[str, float]] = {} self._lock = asyncio.Lock() async def get(self, key: str) -> Optional[str]: async with self._lock: entry = self._store.get(key) if entry is None: return None value, expiry = entry if expiry > 0 and time.time() > expiry: del self._store[key] return None return value async def set(self, key: str, value: str, ttl: int = 300) -> None: expiry = time.time() + ttl if ttl > 0 else -1 async with self._lock: self._store[key] = (value, expiry) async def delete(self, key: str) -> None: async with self._lock: self._store.pop(key, None) async def exists(self, key: str) -> bool: return await self.get(key) is not None async def incr(self, key: str, ttl: int = 60) -> int: async with self._lock: entry = self._store.get(key) if entry is None or (entry[1] > 0 and time.time() > entry[1]): count = 1 else: try: count = int(entry[0]) + 1 except ValueError: count = 1 expiry = time.time() + ttl self._store[key] = (str(count), expiry) return count class CacheService: """ Auto-selects Redis or InMemoryCache. Provides semantic caching via query vector similarity. """ def __init__(self): self._backend = None self._semantic_store: List[Dict[str, Any]] = [] def _init_backend(self): if self._backend is None: import os redis_url = os.environ.get("REDIS_URL") if redis_url: logger.info("Using Redis cache", url=redis_url) from services.cache import RedisCache self._backend = RedisCache(redis_url) else: logger.info("Using in-memory cache (no Redis configured)") self._backend = InMemoryCache() return self._backend async def get(self, key: str) -> Optional[str]: return await self._init_backend().get(key) async def set(self, key: str, value: Any, ttl: int = 300) -> None: if not isinstance(value, str): value = json.dumps(value) await self._init_backend().set(key, value, ttl) async def get_json(self, key: str) -> Optional[Any]: raw = await self.get(key) if raw is None: return None try: return json.loads(raw) except json.JSONDecodeError: return raw async def delete(self, key: str) -> None: await self._init_backend().delete(key) async def exists(self, key: str) -> bool: return await self._init_backend().exists(key) async def incr(self, key: str, ttl: int = 60) -> int: return await self._init_backend().incr(key, ttl) # ── Semantic Caching ─────────────────────────────────────── def get_semantic(self, query_vector: List[float], similarity_threshold: float = 0.93) -> Optional[Dict[str, Any]]: """Look up cached response for semantically matching query vector.""" if not self._semantic_store: return None q_vec = np.array(query_vector) for entry in self._semantic_store: cached_vec = np.array(entry["vector"]) similarity = float(np.dot(q_vec, cached_vec) / (np.linalg.norm(q_vec) * np.linalg.norm(cached_vec))) if similarity >= similarity_threshold: logger.info("Semantic cache hit!", similarity=round(similarity, 4)) return entry["response"] return None def set_semantic(self, query_vector: List[float], response_data: Dict[str, Any]) -> None: """Store query vector and response in semantic cache.""" self._semantic_store.append({ "vector": query_vector, "response": response_data, "timestamp": time.time(), }) # Keep store size bounded if len(self._semantic_store) > 200: self._semantic_store.pop(0) # ── Singleton instance ───────────────────────────────────────── cache_service = CacheService()