File size: 2,946 Bytes
4d6c43c bb9c7e6 4d6c43c bb9c7e6 |
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 |
# cache_manager.py
from diskcache import Cache
from functools import wraps
import hashlib
import json
import os
from config import settings
# Handle Spaces environment - use /tmp if available
def get_cache_dir():
"""Get cache directory that works on Spaces"""
if os.path.exists("/tmp"):
return "/tmp/.cache"
return settings.CACHE_DIR
cache_dir = get_cache_dir()
if not os.path.exists(cache_dir):
os.makedirs(cache_dir, exist_ok=True)
# Initialize cache with error handling
try:
cache = Cache(cache_dir, size_limit=1000000000) # 1GB limit
except Exception as e:
print(f"Cache initialization failed: {e}. Using memory cache fallback.")
# Fallback to a simple dictionary-based cache
cache = {}
def cache_llm_response(expire=86400):
"""Cache decorator for LLM responses with fallback"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
if not settings.CACHE_ENABLED:
return func(*args, **kwargs)
# Create cache key
key_data = {
"func": func.__name__,
"args": str(args),
"kwargs": str(kwargs)
}
key = hashlib.md5(json.dumps(key_data, sort_keys=True).encode()).hexdigest()
# Check cache (with fallback for dictionary cache)
if isinstance(cache, dict):
if key in cache:
return cache[key]
else:
if key in cache:
return cache[key]
# Call function and cache result
result = func(*args, **kwargs)
# Store in cache (with fallback)
if isinstance(cache, dict):
cache[key] = result
else:
try:
cache.set(key, result, expire=expire)
except Exception:
# Fallback to dictionary if disk cache fails
cache[key] = result
return result
return wrapper
return decorator
def clear_cache():
"""Clear all cached responses"""
try:
if isinstance(cache, dict):
cache.clear()
else:
cache.clear()
except Exception as e:
print(f"Cache clearing failed: {e}")
def get_cache_stats():
"""Get cache statistics"""
try:
if isinstance(cache, dict):
return {
"size": sum(len(str(v)) for v in cache.values()),
"count": len(cache),
"enabled": settings.CACHE_ENABLED
}
else:
return {
"size": cache.volume(),
"count": len(cache),
"enabled": settings.CACHE_ENABLED
}
except Exception:
return {
"size": 0,
"count": 0,
"enabled": False
} |