Spaces:
Build error
Build error
File size: 3,717 Bytes
9ff91b1 | 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 | """SQLite-backed AI response cache with TTL and LRU retention."""
import hashlib
import json
import threading
from pathlib import Path
from ..settings import settings
from ..storage.database import database
class CacheManager:
"""Manage cached AI responses with TTL expiration and LRU eviction."""
def __init__(self, cache_dir="output/cache", max_entries=None, ttl_days=None):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.cache_file = self.cache_dir / "ai_responses.json"
self.max_entries = max_entries or settings.retention.max_cache_entries
self.ttl_seconds = (ttl_days or settings.retention.cache_days) * 86400
self._lock = threading.RLock()
self._migrate_legacy_cache()
def _migrate_legacy_cache(self):
"""Import the old JSON cache once; SQLite is canonical afterwards."""
if database.migration_applied("legacy_ai_cache_json_v1"):
return
legacy = self._load_cache()
for key, entry in legacy.items():
if not isinstance(entry, dict) or "response" not in entry:
continue
if database.cache_get(key, self.ttl_seconds) is None:
database.cache_set(key, str(entry["response"]), int(entry.get("text_length") or 0))
database.mark_migration("legacy_ai_cache_json_v1")
def _load_cache(self):
"""Load cache from disk."""
if self.cache_file.exists():
try:
with open(self.cache_file, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"Warning: Could not load cache: {e}")
return {}
return {}
def _generate_key(self, text):
"""Generate cache key from input text.
``text`` here is treated as the full composite key (model | prompt
| user text) — see ``ai_client.generate_response``.
"""
return hashlib.sha256(text.encode('utf-8')).hexdigest()
def get(self, text):
"""Get cached response if available and not expired."""
key = self._generate_key(text)
with self._lock:
entry = database.cache_get(key, self.ttl_seconds)
if entry is None:
return None
print(f"Cache hit (key: {key[:8]}...)")
return entry['response']
def set(self, text, response):
"""Cache a response with timestamp tracking."""
key = self._generate_key(text)
with self._lock:
database.cache_set(key, response, len(text))
database.apply_retention()
print(f"Response cached (key: {key[:8]}...)")
def flush(self):
"""Compatibility no-op: SQLite commits each cache transaction."""
def clear(self):
"""Clear all cached responses."""
with self._lock:
database.cache_clear()
print("Cache cleared")
def get_stats(self):
"""Get cache statistics (snapshot — safe to call from any thread)."""
with self._lock:
stats = database.cache_stats(self.ttl_seconds)
total = stats["total"]
expired_count = stats["expired"]
cache_size = settings.database_path.stat().st_size if settings.database_path.exists() else 0
return {
'total_entries': total,
'expired_entries': expired_count,
'active_entries': total - expired_count,
'max_entries': self.max_entries,
'ttl_days': self.ttl_seconds / 86400,
'cache_file': str(settings.database_path),
'cache_size_kb': round(cache_size / 1024, 2),
}
|