File size: 5,234 Bytes
939c0c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """
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()
|