Spaces:
Sleeping
Sleeping
File size: 6,008 Bytes
b6ae869 | 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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """
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,
}
|