Spaces:
Running
Running
| """Cache TTL en mémoire (LLM + RAG) pour réduire les appels API.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import threading | |
| import time | |
| from collections import OrderedDict | |
| from typing import Any | |
| from app.config import settings | |
| class TTLCache: | |
| """Cache LRU à durée de vie limitée, thread-safe.""" | |
| def __init__( | |
| self, | |
| *, | |
| max_entries: int | None = None, | |
| ttl_seconds: int | None = None, | |
| ) -> None: | |
| self.max_entries = max_entries or settings.cache_max_entries | |
| self.ttl_seconds = ttl_seconds or settings.cache_ttl_seconds | |
| self._data: OrderedDict[str, tuple[float, Any]] = OrderedDict() | |
| self._lock = threading.Lock() | |
| def _is_fresh(self, expires_at: float) -> bool: | |
| return time.time() < expires_at | |
| def get(self, key: str) -> Any | None: | |
| if not settings.cache_enabled: | |
| return None | |
| with self._lock: | |
| item = self._data.get(key) | |
| if item is None: | |
| return None | |
| expires_at, value = item | |
| if not self._is_fresh(expires_at): | |
| del self._data[key] | |
| return None | |
| self._data.move_to_end(key) | |
| return value | |
| def set(self, key: str, value: Any) -> None: | |
| if not settings.cache_enabled: | |
| return | |
| expires_at = time.time() + self.ttl_seconds | |
| with self._lock: | |
| self._data[key] = (expires_at, value) | |
| self._data.move_to_end(key) | |
| while len(self._data) > self.max_entries: | |
| self._data.popitem(last=False) | |
| def clear(self) -> None: | |
| with self._lock: | |
| self._data.clear() | |
| def make_cache_key(*parts: Any) -> str: | |
| """Construit une clé stable à partir d'objets sérialisables.""" | |
| payload = json.dumps(parts, ensure_ascii=False, sort_keys=True, default=str) | |
| return hashlib.sha256(payload.encode("utf-8")).hexdigest() | |
| llm_cache = TTLCache() | |
| rag_cache = TTLCache() | |
| web_cache = TTLCache() | |