Spaces:
Running
Running
| """Caching layer. | |
| Two distinct mechanisms, per the reference docs: | |
| 1. Prompt caching — handled inside each provider (`cache_control` for Anthropic, | |
| implicit prefixes for Gemini/OpenAI). The job here is just to *structure* the | |
| prompt so the static prefix is cacheable and the dynamic suffix is not. | |
| 2. Semantic caching — an application-level cache that short-circuits the LLM | |
| entirely when a near-identical request was already answered. Uses | |
| sentence-transformers embeddings if installed, else a normalized-text hash | |
| bucket (exact-ish match) so the feature works with zero extra deps. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import re | |
| import time | |
| from dataclasses import dataclass | |
| from typing import Optional | |
| def normalize_for_cache(text: str) -> str: | |
| """Lowercase + collapse whitespace — gives the hash fallback a fighting chance | |
| at matching semantically-identical-but-formatted-differently prompts.""" | |
| return re.sub(r"\s+", " ", (text or "").strip().lower()) | |
| class CacheEntry: | |
| key: str | |
| value: str | |
| ts: float | |
| embedding: Optional[list] = None | |
| class SemanticCache: | |
| def __init__(self, threshold: float = 0.92, ttl_seconds: int = 3600, | |
| enabled: bool = True) -> None: | |
| self.threshold = threshold | |
| self.ttl = ttl_seconds | |
| self.enabled = enabled | |
| self._entries: list[CacheEntry] = [] | |
| self._embedder = None | |
| self._tried_embedder = False | |
| self.hits = 0 | |
| self.misses = 0 | |
| def _get_embedder(self): | |
| if self._tried_embedder: | |
| return self._embedder | |
| self._tried_embedder = True | |
| try: | |
| from sentence_transformers import SentenceTransformer | |
| self._embedder = SentenceTransformer("all-MiniLM-L6-v2") | |
| except Exception: | |
| self._embedder = None | |
| return self._embedder | |
| def _embed(self, text: str): | |
| emb = self._get_embedder() | |
| if emb is None: | |
| return None | |
| return emb.encode(text, normalize_embeddings=True).tolist() | |
| def _cosine(a: list, b: list) -> float: | |
| dot = sum(x * y for x, y in zip(a, b)) | |
| return dot # vectors are normalized | |
| def _purge_expired(self) -> None: | |
| now = time.time() | |
| self._entries = [e for e in self._entries if now - e.ts < self.ttl] | |
| def get(self, prompt: str) -> Optional[str]: | |
| if not self.enabled: | |
| return None | |
| self._purge_expired() | |
| norm = normalize_for_cache(prompt) | |
| emb = self._embed(norm) | |
| if emb is None: | |
| # hash fallback: exact normalized match | |
| key = hashlib.sha256(norm.encode()).hexdigest() | |
| for e in self._entries: | |
| if e.key == key: | |
| self.hits += 1 | |
| return e.value | |
| self.misses += 1 | |
| return None | |
| # semantic match | |
| best, best_sim = None, 0.0 | |
| for e in self._entries: | |
| if e.embedding is None: | |
| continue | |
| sim = self._cosine(emb, e.embedding) | |
| if sim > best_sim: | |
| best, best_sim = e, sim | |
| if best is not None and best_sim >= self.threshold: | |
| self.hits += 1 | |
| return best.value | |
| self.misses += 1 | |
| return None | |
| def put(self, prompt: str, value: str) -> None: | |
| if not self.enabled: | |
| return | |
| norm = normalize_for_cache(prompt) | |
| emb = self._embed(norm) | |
| key = hashlib.sha256(norm.encode()).hexdigest() | |
| self._entries.append(CacheEntry(key=key, value=value, ts=time.time(), embedding=emb)) | |
| def stats(self) -> dict: | |
| total = self.hits + self.misses | |
| return { | |
| "enabled": self.enabled, | |
| "entries": len(self._entries), | |
| "hits": self.hits, | |
| "misses": self.misses, | |
| "hit_rate": round(self.hits / total, 3) if total else 0.0, | |
| "backend": "embeddings" if self._embedder else "hash-fallback", | |
| } | |