Mokhtar
fix(ux-performance): format suggestions as 1), 2), 3), ensure broad program/faculty coverage, enable real-time token streaming, and optimize provider fallbacks
3479de3 | # cache/manager.py — Multi-Level Response Cache Manager (L1 RAM, L2 Normalized, L3 Redis) | |
| import os | |
| import json | |
| import time | |
| from pathlib import Path | |
| from typing import Dict, Optional, Any | |
| from cache.keys import KeyManager, normalize_arabic, HNU_KNOWLEDGE_VERSION | |
| # Optional Redis integration | |
| REDIS_URL = os.getenv("REDIS_URL") | |
| REDIS_API_KEY = os.getenv("REDIS_API_KEY") or os.getenv("REDIS_PASSWORD") | |
| REDIS_HOST = os.getenv("REDIS_HOST", "localhost") | |
| REDIS_PORT = int(os.getenv("REDIS_PORT", "6379")) | |
| redis_client = None | |
| if REDIS_URL: | |
| try: | |
| import redis | |
| client = redis.from_url(REDIS_URL, decode_responses=True, socket_timeout=2) | |
| client.ping() | |
| redis_client = client | |
| print("✅ Redis Cache Connected!") | |
| except Exception as e: | |
| print(f"⚠️ Redis connection disabled: {e}") | |
| elif REDIS_API_KEY and os.getenv("REDIS_HOST"): | |
| try: | |
| import redis | |
| client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, password=REDIS_API_KEY, decode_responses=True, socket_timeout=2) | |
| client.ping() | |
| redis_client = client | |
| print("✅ Redis Cache Connected via API Key!") | |
| except Exception as e: | |
| print(f"⚠️ Redis host '{REDIS_HOST}' unreachable. Operating in high-speed RAM cache mode.") | |
| class CacheManager: | |
| """Multi-tier cache manager (L1 RAM, L2 Normalized, L3 Redis).""" | |
| def __init__(self, cache_file_path: Path, max_size: int = 1000): | |
| self.cache_file_path = Path(cache_file_path) | |
| self.max_size = max_size | |
| self.cache: Dict[str, Dict[str, Any]] = {} | |
| self.norm_map: Dict[str, str] = {} | |
| self._load_cache() | |
| def _load_cache(self): | |
| """Load cache from disk JSON and build normalized map.""" | |
| if self.cache_file_path.exists(): | |
| try: | |
| with open(self.cache_file_path, "r", encoding="utf-8") as f: | |
| raw_data = json.load(f) | |
| # Filter for entries matching current knowledge version | |
| for k, v in raw_data.items(): | |
| if v.get("knowledge_version") == HNU_KNOWLEDGE_VERSION or k.startswith(f"hnu:{HNU_KNOWLEDGE_VERSION}"): | |
| self.cache[k] = v | |
| self._rebuild_norm_map() | |
| print(f"📦 Cache Loaded: {len(self.cache)} entries matching version {HNU_KNOWLEDGE_VERSION}.") | |
| except Exception as e: | |
| print(f"⚠️ Error loading disk cache: {e}") | |
| def _rebuild_norm_map(self): | |
| """Rebuild normalized query string to cache key mapping.""" | |
| self.norm_map = {} | |
| for key, entry in self.cache.items(): | |
| query = entry.get("query", "") | |
| if query: | |
| norm_q = normalize_arabic(query) | |
| self.norm_map[norm_q] = key | |
| def _save_cache(self): | |
| """Save active cache entries to disk.""" | |
| self.cache_file_path.parent.mkdir(parents=True, exist_ok=True) | |
| try: | |
| with open(self.cache_file_path, "w", encoding="utf-8") as f: | |
| json.dump(self.cache, f, ensure_ascii=False, indent=2) | |
| except Exception as e: | |
| print(f"⚠️ Disk cache write error: {e}") | |
| def get(self, query: str, language: str = "ar") -> Optional[str]: | |
| """Lookup cached answer via L1 RAM key, L2 normalized string, or L3 Redis.""" | |
| key = KeyManager.generate_answer_key(query, language) | |
| # L1: Exact RAM Hash Match (<0.1 ms) | |
| entry = self.cache.get(key) | |
| if entry is not None: | |
| entry["hits"] = entry.get("hits", 0) + 1 | |
| self._save_cache() | |
| return entry["response"] | |
| # L2: Normalized String Match in RAM | |
| norm_q = normalize_arabic(query) | |
| matched_key = self.norm_map.get(norm_q) | |
| if matched_key and matched_key in self.cache: | |
| entry = self.cache[matched_key] | |
| entry["hits"] = entry.get("hits", 0) + 1 | |
| self._save_cache() | |
| return entry["response"] | |
| # L3: Redis Lookup | |
| if redis_client: | |
| try: | |
| redis_data = redis_client.get(key) | |
| if redis_data: | |
| parsed = json.loads(redis_data) | |
| resp = parsed.get("response") | |
| if resp: | |
| # Store back into L1/L2 | |
| self.put(query, resp, language=language) | |
| return resp | |
| except Exception as e: | |
| print(f"⚠️ Redis read error: {e}") | |
| return None | |
| def put(self, query: str, response: str, language: str = "ar"): | |
| """Store query response into L1, L2, Disk, and L3 Redis.""" | |
| key = KeyManager.generate_answer_key(query, language) | |
| if len(self.cache) >= self.max_size: | |
| oldest_key = min(self.cache, key=lambda k: self.cache[k].get("timestamp", 0)) | |
| del self.cache[oldest_key] | |
| data = { | |
| "query": query, | |
| "response": response, | |
| "knowledge_version": HNU_KNOWLEDGE_VERSION, | |
| "language": language, | |
| "timestamp": time.time(), | |
| "hits": 0 | |
| } | |
| self.cache[key] = data | |
| norm_q = normalize_arabic(query) | |
| self.norm_map[norm_q] = key | |
| self._save_cache() | |
| if redis_client: | |
| try: | |
| redis_client.set(key, json.dumps(data, ensure_ascii=False)) | |
| except Exception as e: | |
| print(f"⚠️ Redis write error: {e}") | |
| def clear(self): | |
| """Clear all cache entries across RAM, Disk, and Redis.""" | |
| self.cache = {} | |
| self.norm_map = {} | |
| self._save_cache() | |
| if redis_client: | |
| try: | |
| keys = redis_client.keys(f"hnu:{HNU_KNOWLEDGE_VERSION}:*") | |
| if keys: | |
| redis_client.delete(*keys) | |
| except Exception as e: | |
| print(f"⚠️ Redis clear error: {e}") | |
| def stats(self) -> Dict[str, Any]: | |
| total_hits = sum(e.get("hits", 0) for e in self.cache.values()) | |
| return { | |
| "knowledge_version": HNU_KNOWLEDGE_VERSION, | |
| "total_entries": len(self.cache), | |
| "max_size": self.max_size, | |
| "total_hits": total_hits, | |
| "redis_connected": redis_client is not None | |
| } | |