Spaces:

sahihislam
/

superadmin

Running on Zero

feat: resolve nearby masjids 500 error, add popular sanctuaries, and optimize search architecture

ae9655d 24 days ago

16.2 kB

"""
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)

	"""
	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)