app_cache.py

import hashlib
import re
import numpy as np
from collections import Counter


def normalize(text: str) -> str:
    return re.sub(r"\s+", " ", text.strip().lower())


class ExactCache:
    """Correctness-safe: a hit means the request is identical after normalization."""

    def __init__(self):
        self.store = {}

    def _key(self, prompt):
        return hashlib.sha256(normalize(prompt).encode()).hexdigest()

    def get(self, prompt):
        return self.store.get(self._key(prompt))

    def put(self, prompt, response):
        self.store[self._key(prompt)] = response


class SemanticCache:
    """Returns a stored response when a new request is similar enough. Here
    similarity is bag-of-words cosine -- a transparent stand-in for a real
    sentence-embedding model, chosen so the failure modes are visible."""

    def __init__(self, threshold):
        self.entries = []          # list of (vector, response)
        self.threshold = threshold

    @staticmethod
    def _cosine(a, b):
        keys = set(a) | set(b)
        va = np.array([a.get(k, 0) for k in keys], float)
        vb = np.array([b.get(k, 0) for k in keys], float)
        na, nb = np.linalg.norm(va), np.linalg.norm(vb)
        return float(va @ vb / (na * nb)) if na and nb else 0.0

    def _vec(self, prompt):
        return Counter(normalize(prompt).split())

    def get(self, prompt):
        q = self._vec(prompt)
        best_sim, best_resp = -1.0, None
        for vec, resp in self.entries:
            sim = self._cosine(q, vec)
            if sim > best_sim:
                best_sim, best_resp = sim, resp
        return (best_resp, best_sim) if best_sim >= self.threshold else (None, best_sim)

    def put(self, prompt, response):
        self.entries.append((self._vec(prompt), response))


if __name__ == "__main__":
    ex = ExactCache()
    ex.put("What is a KV cache?", "CACHED-ANSWER")
    print("EXACT CACHE")
    print(f"  whitespace/case variant -> {ex.get('what is a  KV cache? ')}")
    print(f"  paraphrase              -> {ex.get('Explain the KV cache')}")

    sc = SemanticCache(threshold=0.50)
    sc.put("How do I reduce KV cache memory?", "CACHED-ANSWER")
    print("\nSEMANTIC CACHE (threshold 0.50)")
    for q in ["How do I reduce KV cache memory?",           # identical
              "How to shrink the KV cache memory usage?",   # true paraphrase
              "How do I reduce GPU memory?",                # overlap, DIFFERENT
              "What is grouped query attention?"]:          # unrelated
        resp, sim = sc.get(q)
        print(f"  sim={sim:4.2f}  {'HIT ' if resp else 'miss'}  {q}")