app/engine/semantic_cache.py

import json
import logging
import math
from typing import Any, Dict, List, Optional
from app.core.queue import get_redis_client
from app.engine.indexer import dense_embeddings

logger = logging.getLogger(__name__)

def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
    if not vec1 or not vec2 or len(vec1) != len(vec2):
        return 0.0
    dot_product = sum(a * b for a, b in zip(vec1, vec2))
    norm_a = math.sqrt(sum(a * a for a in vec1))
    norm_b = math.sqrt(sum(b * b for b in vec2))
    if norm_a == 0.0 or norm_b == 0.0:
        return 0.0
    return dot_product / (norm_a * norm_b)

async def get_cached_answer(query: str, similarity_threshold: float = 0.92) -> Optional[Dict[str, Any]]:
    try:
        redis = await get_redis_client()
        keys = await redis.keys("semantic_cache:*")
        if not keys:
            return None
            
        embedder = dense_embeddings()
        query_vec = embedder.embed_query(query)
        
        best_sim = 0.0
        best_match = None
        
        for k in keys[:100]:  # Check up to 100 recent cached items
            raw = await redis.get(k)
            if not raw:
                continue
            data = json.loads(raw)
            cached_vec = data.get("embedding")
            if not cached_vec:
                continue
                
            sim = cosine_similarity(query_vec, cached_vec)
            if sim > best_sim and sim >= similarity_threshold:
                best_sim = sim
                best_match = data
                
        if best_match:
            logger.info(f"Semantic cache HIT (similarity: {best_sim:.4f} >= {similarity_threshold}) for query: '{query}'")
            formatted_sources = []
            for s in best_match.get("sources", []):
                src_dict = dict(s) if isinstance(s, dict) else {"source": str(s)}
                if "snippet" not in src_dict:
                    src_dict["snippet"] = src_dict.get("page_content", src_dict.get("source", best_match["answer"]))[:250]
                if "source" not in src_dict:
                    src_dict["source"] = src_dict.get("doc_id", "cached_doc")
                formatted_sources.append(src_dict)
            return {
                "answer": best_match["answer"],
                "sources": formatted_sources,
                "confidence": best_match.get("confidence", 0.99),
                "cached": True,
                "similarity": best_sim
            }
        logger.info(f"Semantic cache MISS (highest similarity: {best_sim:.4f} < {similarity_threshold}) for query: '{query}'")
        return None
    except Exception as e:
        logger.warning(f"Semantic cache lookup failed ({e})")
        return None

async def set_cached_answer(query: str, answer: str, sources: List[Any], confidence: float, ttl_seconds: int = 86400) -> None:
    try:
        redis = await get_redis_client()
        embedder = dense_embeddings()
        query_vec = embedder.embed_query(query)
        
        formatted_sources = []
        for s in sources:
            if isinstance(s, dict):
                src_dict = dict(s)
                if "snippet" not in src_dict:
                    src_dict["snippet"] = src_dict.get("page_content", src_dict.get("source", str(s)))[:250]
                if "source" not in src_dict:
                    src_dict["source"] = src_dict.get("doc_id", "cached_doc")
                formatted_sources.append(src_dict)
            elif hasattr(s, "dict") or hasattr(s, "model_dump"):
                src_dict = s.model_dump() if hasattr(s, "model_dump") else s.dict()
                if "snippet" not in src_dict:
                    src_dict["snippet"] = getattr(s, "page_content", str(s))[:250]
                if "source" not in src_dict:
                    src_dict["source"] = getattr(s, "metadata", {}).get("source", getattr(s, "metadata", {}).get("doc_id", "cached_doc"))
                formatted_sources.append(src_dict)
            elif hasattr(s, "page_content"):
                formatted_sources.append({
                    "source": getattr(s, "metadata", {}).get("source", getattr(s, "metadata", {}).get("doc_id", "cached_doc")),
                    "doc_id": getattr(s, "metadata", {}).get("doc_id", "cached_doc"),
                    "snippet": s.page_content[:250]
                })
            else:
                formatted_sources.append({"source": "cached_doc", "snippet": str(s)[:250]})
                
        key = f"semantic_cache:{abs(hash(query))}"
        payload = {
            "query": query,
            "embedding": query_vec,
            "answer": answer,
            "sources": formatted_sources,
            "confidence": confidence
        }
        await redis.setex(key, ttl_seconds, json.dumps(payload))
        logger.info(f"Cached semantic answer for query: '{query}'")
    except Exception as e:
        logger.warning(f"Failed to set semantic cache ({e})")