src/utils/semantic_cache.py

"""
Two-layer semantic cache for LLM responses.

Layer 1: Exact string match (hash lookup, O(1), ~0ms)
Layer 2: Cosine similarity via sentence-transformers (~5ms after warmup)

Production note: Replace _exact_cache and _semantic_store with Redis
calls for distributed deployments. The interface stays identical.

Intents that skip cache entirely (too context-sensitive):
  - sentimental, hybrid, out_of_scope, escalation
"""

import time
import hashlib
import threading
import numpy as np
from typing import Optional

# ─── Config ─────────────────────────────────────────────────────────────────
SIMILARITY_THRESHOLD = 0.92   # High threshold — avoids "return policy" vs "exchange policy" false matches
MAX_CACHE_SIZE = 500          # Evict oldest entries beyond this
CACHE_TTL_SECONDS = 3600      # 1 hour — policy docs don't change frequently

# Intents that must NEVER be cached (personalized or emotional responses)
NON_CACHEABLE_INTENTS = {"sentimental", "hybrid", "out_of_scope", None}

# ─── Storage ─────────────────────────────────────────────────────────────────
_exact_cache: dict[str, dict] = {}         # hash → {response, timestamp, intent}
_semantic_store: list[dict] = []           # list of {embedding, response, query, timestamp}
_cache_lock = threading.Lock()

# ─── Embedding Model (lazy, shared with RAG/scope) ───────────────────────────
_embed_model = None
_embed_lock = threading.Lock()

def _get_embed_model():
    global _embed_model
    if _embed_model is None:
        with _embed_lock:
            if _embed_model is None:
                from sentence_transformers import SentenceTransformer
                # Reuses the same model already loaded by rag_retriever
                _embed_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
    return _embed_model

def _hash(text: str) -> str:
    return hashlib.sha256(text.strip().lower().encode()).hexdigest()

def _is_expired(entry: dict) -> bool:
    return (time.time() - entry["timestamp"]) > CACHE_TTL_SECONDS

def _evict_if_needed():
    """Remove expired entries. If still over limit, evict oldest."""
    global _semantic_store
    # Remove expired
    _semantic_store = [e for e in _semantic_store if not _is_expired(e)]
    # Trim oldest if still over limit
    if len(_semantic_store) > MAX_CACHE_SIZE:
        _semantic_store = _semantic_store[-MAX_CACHE_SIZE:]

# ─── Public Interface ────────────────────────────────────────────────────────
def get_cached_response(query: str, intent: Optional[str]) -> Optional[str]:
    """
    Check cache for a matching response.
    Returns cached response string, or None if no valid match found.
    Skips cache entirely for non-cacheable intents.
    """
    if intent in NON_CACHEABLE_INTENTS:
        return None

    query_clean = query.strip()

    # Layer 1: Exact match
    key = _hash(query_clean)
    with _cache_lock:
        if key in _exact_cache:
            entry = _exact_cache[key]
            if not _is_expired(entry):
                print(f"[semantic_cache] Exact cache HIT for: {query_clean[:50]}")
                return entry["response"]
            else:
                del _exact_cache[key]

    # Layer 2: Semantic similarity
    try:
        model = _get_embed_model()
        query_emb = model.encode([query_clean], normalize_embeddings=True)[0]

        with _cache_lock:
            best_score = 0.0
            best_response = None
            for entry in _semantic_store:
                if _is_expired(entry):
                    continue
                score = float(np.dot(entry["embedding"], query_emb))
                if score > best_score:
                    best_score = score
                    best_response = entry["response"]

        if best_score >= SIMILARITY_THRESHOLD and best_response:
            print(f"[semantic_cache] Semantic cache HIT (score={best_score:.3f}) for: {query_clean[:50]}")
            return best_response

    except Exception as e:
        print(f"[semantic_cache] Embedding error during lookup: {e}")

    return None


def store_in_cache(query: str, response: str, intent: Optional[str]):
    """
    Store a query-response pair in both cache layers.
    Skips non-cacheable intents.
    """
    if intent in NON_CACHEABLE_INTENTS:
        return

    if not query or not response:
        return

    query_clean = query.strip()
    now = time.time()

    # Layer 1: Exact store
    key = _hash(query_clean)
    with _cache_lock:
        _exact_cache[key] = {
            "response": response,
            "timestamp": now,
            "intent": intent,
        }

    # Layer 2: Semantic store
    try:
        model = _get_embed_model()
        emb = model.encode([query_clean], normalize_embeddings=True)[0]

        with _cache_lock:
            _evict_if_needed()
            _semantic_store.append({
                "embedding": emb,
                "response": response,
                "query": query_clean,
                "timestamp": now,
                "intent": intent,
            })
        print(f"[semantic_cache] Stored in cache (intent={intent}): {query_clean[:50]}")

    except Exception as e:
        print(f"[semantic_cache] Embedding error during store: {e}")


def get_cache_stats() -> dict:
    """Return cache statistics for monitoring."""
    with _cache_lock:
        return {
            "exact_cache_size": len(_exact_cache),
            "semantic_store_size": len(_semantic_store),
            "max_cache_size": MAX_CACHE_SIZE,
            "ttl_seconds": CACHE_TTL_SECONDS,
        }