Spaces:
Sleeping
Sleeping
File size: 4,945 Bytes
a30ab12 d816f3a a30ab12 d816f3a a30ab12 d816f3a a30ab12 d816f3a a30ab12 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | 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})")
|