""" RAG Multi-Lingual, Slice & Semantic Caching Engine Solves: 1. Multi-lingual requests (Arabic, Urdu, Hinglish, English) 2. Variable slice counts (1 to 6 answers) 3. Semantic variations ("benefits of praying salat" vs "benefits of praying") 4. Clean Sanad vs Matn separation for include_sanad toggle """ import os import hashlib import json import re import time from typing import List, Dict, Any, Optional, Tuple, Union from collections import OrderedDict import numpy as np from app.infrastructure.redis import redis_manager CACHE_VERSION = "v2" def normalize_query(query: str) -> str: """Normalize query text for deterministic cache hashing.""" q = query.strip().lower() q = re.sub(r'[^\w\s]', '', q) return re.sub(r'\s+', ' ', q).strip() def make_canonical_docs_key(query: str, books: Optional[List[str]] = None) -> str: """ Tier 1 Cache Key: Language-independent, stores canonical documents with all multilingual translations. Includes explicit book scope prefix to guarantee 100% scope isolation. """ clean_q = normalize_query(query) books_slug = ",".join(sorted([b.strip().lower() for b in (books or [])])) scope_tag = hashlib.sha256(books_slug.encode("utf-8")).hexdigest()[:8] if books_slug else "all" q_hash = hashlib.sha256(f"{clean_q}::{books_slug}".encode("utf-8")).hexdigest()[:16] return f"rag:{CACHE_VERSION}:canonical_docs:{scope_tag}:{q_hash}" def make_llm_answer_key(query: str, language: str, top_k: int, books: Optional[List[str]] = None, strict_mode: bool = False) -> str: """ Tier 2 Cache Key: Composite key strictly isolating normalized language, result count, books scope, strict mode, and query. Ensures language-dependent answers or explanations are never cross-contaminated. """ from app.rag.translations import normalize_language_code clean_q = normalize_query(query) lang = normalize_language_code(language) books_slug = ",".join(sorted([b.strip().lower() for b in (books or [])])) strict_flag = "strict" if strict_mode else "rag" q_hash = hashlib.sha256(f"{clean_q}::{books_slug}::{strict_flag}".encode("utf-8")).hexdigest()[:16] return f"rag:{CACHE_VERSION}:llm_answer:{lang}:{top_k}:{strict_flag}:{q_hash}" # ========================================================= # TIER 0: L1 PROCESS IN-MEMORY LRU CACHES (<0.05ms) # ========================================================= _L1_CANONICAL_CACHE: OrderedDict = OrderedDict() _MAX_L1_CANONICAL_ENTRIES = 5000 _L1_LLM_ANSWER_CACHE: OrderedDict = OrderedDict() _MAX_L1_LLM_ENTRIES = 2000 # ========================================================= # FAST IN-MEMORY SEMANTIC VECTOR CACHE (Ring Buffer) # Maps query vector -> canonical cache key (cosine similarity >= 0.92) # ========================================================= _SEMANTIC_CACHE_MAX_ENTRIES = 2000 _semantic_vectors: List[np.ndarray] = [] _semantic_keys: List[str] = [] _semantic_timestamps: List[float] = [] def register_semantic_vector(query_vec: List[float], canonical_key: str): """Register normalized query vector into the fast in-memory semantic ring buffer.""" global _semantic_vectors, _semantic_keys, _semantic_timestamps try: vec = np.array(query_vec, dtype=np.float32) norm = np.linalg.norm(vec) if norm > 0: vec = vec / norm else: return if len(_semantic_vectors) >= _SEMANTIC_CACHE_MAX_ENTRIES: _semantic_vectors.pop(0) _semantic_keys.pop(0) _semantic_timestamps.pop(0) _semantic_vectors.append(vec) _semantic_keys.append(canonical_key) _semantic_timestamps.append(time.time()) except Exception: pass _SEMANTIC_INDEX_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "semantic_vectors_index.npz") def save_semantic_index_to_disk(filepath: Optional[str] = None): """Save in-memory semantic vectors and keys to compressed disk archive.""" global _semantic_vectors, _semantic_keys p = filepath or _SEMANTIC_INDEX_PATH if _semantic_vectors and _semantic_keys: try: np.savez_compressed(p, vectors=np.array(_semantic_vectors), keys=np.array(_semantic_keys)) except Exception: pass def load_semantic_index_from_disk(filepath: Optional[str] = None) -> bool: """Load semantic vectors and keys into memory on startup (<5ms).""" global _semantic_vectors, _semantic_keys p = filepath or _SEMANTIC_INDEX_PATH if os.path.exists(p): try: data = np.load(p, allow_pickle=True) _semantic_vectors = [v for v in data["vectors"]] _semantic_keys = [str(k) for k in data["keys"]] return True except Exception: return False return False # Attempt auto-load on module import load_semantic_index_from_disk() def find_semantic_cache_key(query_vec: List[float], books: Optional[List[str]] = None, similarity_threshold: float = 0.90) -> Optional[str]: """ Finds if a semantically similar query was already answered under the SAME book scope. E.g. 'benefits of praying salat' vs 'benefits of praying' -> Similarity > 0.90 -> MATCH! Strictly prevents cross-book and cross-collection cache contamination. Runs in < 1 ms via NumPy dot product. """ if not _semantic_vectors: return None books_slug = ",".join(sorted([b.strip().lower() for b in (books or [])])) scope_tag = hashlib.sha256(books_slug.encode("utf-8")).hexdigest()[:8] if books_slug else "all" expected_prefix = f"rag:{CACHE_VERSION}:canonical_docs:{scope_tag}:" try: vec = np.array(query_vec, dtype=np.float32) norm = np.linalg.norm(vec) if norm > 0: vec = vec / norm else: return None # Filter vector candidates strictly matching this book scope matching_indices = [ i for i, k in enumerate(_semantic_keys) if k.startswith(expected_prefix) ] if not matching_indices: return None matrix = np.array([_semantic_vectors[i] for i in matching_indices], dtype=np.float32) scores = np.dot(matrix, vec) best_sub_idx = int(np.argmax(scores)) best_score = float(scores[best_sub_idx]) if best_score >= similarity_threshold: return _semantic_keys[matching_indices[best_sub_idx]] return None except Exception: return None # ========================================================= # TIER 1 CANONICAL RETRIEVAL WITH DYNAMIC SLICING & SANAD # ========================================================= async def get_cached_canonical_documents( query: str, top_k: int = 5, language: str = "english", books: Optional[List[str]] = None, query_vec: Optional[List[float]] = None, include_sanad: Optional[bool] = None, ) -> Optional[List[Dict[str, Any]]]: """ Retrieves cached canonical documents, slices them to top_k, and extracts localized texts. Supports: 1. Exact key match 2. Semantic vector match (if query_vec is provided) 3. Dynamic Sanad assembly """ key = make_canonical_docs_key(query, books) # ⚡ Check Tier 0 L1 In-Memory Process Cache (<0.05ms) cached_val = _L1_CANONICAL_CACHE.get(key) if cached_val: _L1_CANONICAL_CACHE.move_to_end(key) else: # ⚡ Check Tier 1 Remote Cloud Redis Cache cached_val = await redis_manager.get(key) if cached_val: if len(_L1_CANONICAL_CACHE) >= _MAX_L1_CANONICAL_ENTRIES: _L1_CANONICAL_CACHE.popitem(last=False) _L1_CANONICAL_CACHE[key] = cached_val # If exact key missed, test semantic vector cache strictly within the caller's book scope if not cached_val and query_vec is not None: semantic_key = find_semantic_cache_key(query_vec, books=books, similarity_threshold=0.92) if semantic_key: cached_val = _L1_CANONICAL_CACHE.get(semantic_key) if not cached_val: cached_val = await redis_manager.get(semantic_key) if cached_val: if len(_L1_CANONICAL_CACHE) >= _MAX_L1_CANONICAL_ENTRIES: _L1_CANONICAL_CACHE.popitem(last=False) _L1_CANONICAL_CACHE[semantic_key] = cached_val if not cached_val: return None try: docs = json.loads(cached_val) if not isinstance(docs, list) or not docs: return None # Slice to requested count (e.g. 1 to 6) sliced_docs = docs[:max(1, min(top_k, 6))] # Localize fields for caller's language using unified translation selector from app.rag.translations import select_hadith_translation localized = [] for d in sliced_docs: item = dict(d) sanad_chain = item.get("sanad", "") or item.get("narrator", "") # Select localized display text with non-silent fallback base_text, is_fallback, actual_lang = select_hadith_translation(item, language) item["_is_fallback"] = is_fallback item["_actual_lang"] = actual_lang # Dynamic Sanad formatting if include_sanad is True and sanad_chain and "[SANAD]:" not in base_text: item["display_text"] = f"[SANAD]: {sanad_chain}\n[MATN]: {base_text}" else: item["display_text"] = base_text localized.append(item) return localized except Exception: return None async def set_cached_canonical_documents( query: Union[str, List[str]], documents: List[Dict[str, Any]], books: Optional[List[str]] = None, query_vec: Optional[List[float]] = None, expire_seconds: int = 86400, # 24 hours TTL ) -> bool: """ Caches up to 6 canonical documents with all multilingual fields intact. Guarantees pure Matn and pure Sanad separation so neither is lost or permanently baked. Stores all available translations (English, Urdu, Hinglish, Arabic, Quran translations) together. Can cache under a single query string or a list of query aliases (e.g. [raw_query, rewritten_query]). """ if not documents: return False # Store up to 6 high-fidelity canonical documents with all available translations clean_docs = [] for d in documents[:6]: core_english = str(d.get("text_english", "") or "") extracted_sanad = str(d.get("sanad", "") or d.get("narrator", "") or "") # Isolate pure Matn from any pre-baked Sanad tags if "[SANAD]:" in core_english and "[MATN]:" in core_english: parts = core_english.split("[MATN]:") if not extracted_sanad: extracted_sanad = parts[0].replace("[SANAD]:", "").strip() core_english = parts[1].strip() elif "[MATN]:" in core_english: core_english = core_english.split("[MATN]:")[-1].strip() def _clean_field(val: Any) -> str: if val is None: return "" s = str(val).strip() if s.lower() in ("null", "none", "nil", "n/a", "undefined"): return "" return s # Build clean multilingual dictionary for all available translations translations_dict = {} if isinstance(d.get("translations"), dict): translations_dict = dict(d["translations"]) # Extract all translations directly from MongoDB record fields urdu_text = _clean_field(d.get("text_urdu") or d.get("text_ur") or translations_dict.get("ur.jalandhry")) arabic_text = _clean_field(d.get("text_arabic") or d.get("text_ar")) hinglish_text = _clean_field(d.get("text_hinglish")) clean_doc = { "hadith_id": d.get("hadith_id"), "book_name": d.get("book_name", ""), "book_no": d.get("book_no"), "chapter_name": d.get("chapter_name", ""), "chapter_no": d.get("chapter_no"), "volume": d.get("volume"), "surah_no": d.get("surah_no"), "ayah_no": d.get("ayah_no"), "reference": str(d.get("reference") or f"Hadith #{d.get('hadith_id', '')}"), "grade": d.get("grade", "N/A"), "narrator": str(d.get("narrator") or "Unknown"), "collection": str(d.get("collection", "")), "text_english": core_english, "text_arabic": arabic_text, "text_urdu": urdu_text, "text_hinglish": hinglish_text, "translations": translations_dict, "sanad": extracted_sanad, "audio_urls": d.get("audio_urls", []), } # Include any dynamic text_* language fields present in d for k, v in d.items(): if k.startswith("text_") and k not in clean_doc and v: cleaned_v = _clean_field(v) if cleaned_v: clean_doc[k] = cleaned_v clean_docs.append(clean_doc) payload = json.dumps(clean_docs, ensure_ascii=False) query_list = [query] if isinstance(query, str) else list(query) success_any = False for q_item in query_list: if not q_item or not str(q_item).strip(): continue key = make_canonical_docs_key(str(q_item).strip(), books) success = await redis_manager.set(key, payload, expire_seconds=expire_seconds) if success: success_any = True if len(_L1_CANONICAL_CACHE) >= _MAX_L1_CANONICAL_ENTRIES: _L1_CANONICAL_CACHE.popitem(last=False) _L1_CANONICAL_CACHE[key] = payload # Register into semantic ring buffer if vector is available if query_vec: register_semantic_vector(query_vec, key) return success_any def clear_rag_caches(): """Clear process L1 in-memory caches and semantic vectors.""" _L1_CANONICAL_CACHE.clear() _L1_LLM_ANSWER_CACHE.clear() _semantic_vectors.clear() _semantic_keys.clear() _semantic_timestamps.clear() # ========================================================= # TIER 2 COMPOSITE KEY CACHE FOR SYNTHESIZED LLM ANSWERS # ========================================================= async def get_cached_llm_answer( query: str, language: str, top_k: int, books: Optional[List[str]] = None, strict_mode: bool = False, ) -> Optional[Dict[str, Any]]: """Retrieves cached synthesized LLM answer from L1 memory or Redis in <1ms.""" key = make_llm_answer_key(query, language, top_k, books, strict_mode) # 1. Check in-memory L1 cache (<0.01ms) val = _L1_LLM_ANSWER_CACHE.get(key) if val: _L1_LLM_ANSWER_CACHE.move_to_end(key) print(f"🎯 [TIER 1 L1 CACHE HIT] Answer served from RAM (<0.05ms) for key={key}") return val # 2. Check Redis (<2ms) cached_val = await redis_manager.get(key) if not cached_val: return None try: parsed = json.loads(cached_val) if len(_L1_LLM_ANSWER_CACHE) >= _MAX_L1_LLM_ENTRIES: _L1_LLM_ANSWER_CACHE.popitem(last=False) _L1_LLM_ANSWER_CACHE[key] = parsed print(f"🎯 [TIER 1 REDIS CACHE HIT] Answer loaded from Redis (<2ms) for key={key}") return parsed except Exception: return None async def set_cached_llm_answer( query: str, language: str, top_k: int, answer_data: Dict[str, Any], books: Optional[List[str]] = None, strict_mode: bool = False, expire_seconds: int = 86400, ) -> bool: """Caches synthesized LLM answer in L1 memory and Redis under composite key.""" key = make_llm_answer_key(query, language, top_k, books, strict_mode) resp_text = answer_data.get("response", "") print(f"💾 [TIER 1 CACHE SAVED] Stored synthesized response in L1 RAM & Redis (len={len(resp_text)}, key={key})") if len(_L1_LLM_ANSWER_CACHE) >= _MAX_L1_LLM_ENTRIES: _L1_LLM_ANSWER_CACHE.popitem(last=False) _L1_LLM_ANSWER_CACHE[key] = answer_data payload = json.dumps(answer_data, ensure_ascii=False) return await redis_manager.set(key, payload, expire_seconds=expire_seconds)