File size: 6,569 Bytes
7c6ffa6 | 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 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """In-memory LRU cache for AI prompt results.
Prevents duplicate AI calls for identical prompts within the TTL window.
Uses an ``OrderedDict``-based LRU with a per-entry timestamp for TTL checks.
Default limits
--------------
- ``max_size``: 1000 entries
- ``ttl_seconds``: 3600 (1 hour)
Thread-safe via a single ``threading.Lock``.
Usage::
from app.core.prompt_cache import get_prompt_cache
cache = get_prompt_cache()
key = make_cache_key(task, model, prompt_hash)
cached = cache.get(key)
if cached is not None:
return cached
result = provider.generate_notes(...)
cache.set(key, result)
return result
"""
from __future__ import annotations
import hashlib
import json
import logging
import threading
import time
from collections import OrderedDict
from copy import deepcopy
from typing import Any
logger = logging.getLogger(__name__)
_REDIS_PREFIX = "pc:" # prompt-cache namespace in Redis
class PromptLRUCache:
"""Two-tier cache: in-process LRU (L1) + optional shared Redis (L2).
L1 keeps single-instance hot reads zero-latency. L2 (active only when
``REDIS_URL`` is set) shares results across every replica so a cache warmed
by one instance is reused by all — the difference between per-instance and
fleet-wide caching. Falls back cleanly to L1-only when Redis is absent.
"""
def __init__(self, max_size: int = 1000, ttl_seconds: float = 3600.0) -> None:
self.max_size = max_size
self.ttl_seconds = ttl_seconds
# OrderedDict value: (created_at: float, payload: dict)
self._store: OrderedDict[str, tuple[float, dict[str, Any]]] = OrderedDict()
self._lock = threading.Lock()
# Counters for observability
self._hits = 0
self._misses = 0
# ── L2 (Redis) helpers ─────────────────────────────────────────────────────
def _redis(self) -> Any | None:
try:
from app.core.redis_client import get_redis
return get_redis()
except Exception:
return None
# ── Public API ────────────────────────────────────────────────────────────
def get(self, key: str) -> dict[str, Any] | None:
"""Return a deep-copy of the cached value or ``None`` on miss / expiry."""
now = time.monotonic()
with self._lock:
entry = self._store.get(key)
if entry is not None:
created_at, value = entry
if now - created_at <= self.ttl_seconds:
self._store.move_to_end(key)
self._hits += 1
return deepcopy(value)
del self._store[key] # expired in L1
# L2: shared Redis lookup (outside lock — network call)
client = self._redis()
if client is not None:
try:
raw = client.get(_REDIS_PREFIX + key)
if raw:
value = json.loads(raw)
self._set_local(key, value) # promote into L1
with self._lock:
self._hits += 1
return deepcopy(value)
except Exception as exc:
logger.debug("Prompt cache L2 get failed: %s", type(exc).__name__)
with self._lock:
self._misses += 1
return None
def _set_local(self, key: str, value: dict[str, Any]) -> None:
now = time.monotonic()
with self._lock:
if key in self._store:
self._store.move_to_end(key)
self._store[key] = (now, value)
while len(self._store) > self.max_size:
self._store.popitem(last=False)
def set(self, key: str, value: dict[str, Any]) -> None:
"""Insert or refresh an entry in L1 and (when configured) shared L2."""
self._set_local(key, value)
client = self._redis()
if client is not None:
try:
client.setex(
_REDIS_PREFIX + key,
int(self.ttl_seconds),
json.dumps(value, ensure_ascii=False, separators=(",", ":")),
)
except Exception as exc:
logger.debug("Prompt cache L2 set failed: %s", type(exc).__name__)
def clear(self) -> None:
"""Empty the cache (used in tests)."""
with self._lock:
self._store.clear()
self._hits = 0
self._misses = 0
# ── Observability ─────────────────────────────────────────────────────────
@property
def size(self) -> int:
with self._lock:
return len(self._store)
def stats(self) -> dict[str, Any]:
with self._lock:
total = self._hits + self._misses
return {
"size": len(self._store),
"max_size": self.max_size,
"ttl_seconds": self.ttl_seconds,
"hits": self._hits,
"misses": self._misses,
"hit_rate": round(self._hits / total, 3) if total else 0.0,
}
# ── Singleton ─────────────────────────────────────────────────────────────────
_cache_instance: PromptLRUCache | None = None
_cache_lock = threading.Lock()
def get_prompt_cache() -> PromptLRUCache:
"""Return the process-level singleton PromptLRUCache."""
global _cache_instance
if _cache_instance is None:
with _cache_lock:
if _cache_instance is None:
_cache_instance = PromptLRUCache(max_size=1000, ttl_seconds=3600.0)
return _cache_instance
# ── Key helpers ───────────────────────────────────────────────────────────────
def make_prompt_cache_key(task_key: str, model_id: str, prompt: str) -> str:
"""Return a short SHA-256 hex key from task + model + prompt content."""
raw = f"{task_key}\x00{model_id}\x00{prompt}"
return hashlib.sha256(raw.encode("utf-8", errors="ignore")).hexdigest()
|