Problem 114 · LLM Systems

Tenant-Scoped Semantic Cache with TTL

MediumAI engineering

01Problem

Build `SemanticCache(embed, threshold, ttl_seconds, max_per_tenant, clock)`. `get(tenant, query)` first tries an exact hit on the normalised query (NFKC, casefold, collapse whitespace, strip trailing `?!.`), then falls back to the most similar cached query **in the same tenant** with cosine >= `threshold`. Expired entries (older than TTL by the injected clock) never hit. `put(tenant, query, answer)` evicts the least recently used entry when a tenant exceeds `max_per_tenant`.

02Key insight

The tenant must be part of the key, not a filter applied afterwards: a cache that returns tenant A's answer to tenant B is a data breach, not a cache miss. Normalisation buys cheap exact hits for trivially different phrasings; the semantic fallback catches paraphrases but is where wrong answers come from, so the threshold is a precision knob you set from labelled pairs, not a guess. TTL bounds staleness when the underlying documents change, and an injected clock makes all of that testable without sleeping.

03Solution

solution.pypython60 lines

import math
import re
import unicodedata
from collections import OrderedDict
from typing import Callable

_WS = re.compile(r"\s+")


def normalise(q: str) -> str:
    q = unicodedata.normalize("NFKC", q).casefold()
    return _WS.sub(" ", q).strip().rstrip("?!.").strip()


def _cos(a: list[float], b: list[float]) -> float:
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(x * x for x in b))
    return 0.0 if not na or not nb else sum(x * y for x, y in zip(a, b)) / (na * nb)


class SemanticCache:
    def __init__(self, embed: Callable[[str], list[float]], threshold: float, ttl_seconds: float,
                 max_per_tenant: int, clock: Callable[[], float]) -> None:
        self.embed, self.threshold, self.ttl = embed, threshold, ttl_seconds
        self.max, self.clock = max_per_tenant, clock
        self._store: dict[str, OrderedDict[str, tuple[list[float], str, float]]] = {}

    def _live(self, tenant: str) -> OrderedDict[str, tuple[list[float], str, float]]:
        entries = self._store.setdefault(tenant, OrderedDict())
        now = self.clock()
        for key in [k for k, (_, _, t) in entries.items() if now - t >= self.ttl]:
            del entries[key]
        return entries

    def get(self, tenant: str, query: str) -> str | None:
        entries = self._live(tenant)
        key = normalise(query)
        if key in entries:
            entries.move_to_end(key)
            return entries[key][1]
        if not entries:
            return None
        qv = self.embed(key)
        best_key, best_sim = None, -1.0
        for k, (vec, _, _) in entries.items():
            s = _cos(qv, vec)
            if s > best_sim:
                best_key, best_sim = k, s
        if best_key is not None and best_sim >= self.threshold:
            entries.move_to_end(best_key)
            return entries[best_key][1]
        return None

    def put(self, tenant: str, query: str, answer: str) -> None:
        entries = self._live(tenant)
        key = normalise(query)
        entries[key] = (self.embed(key), answer, self.clock())
        entries.move_to_end(key)
        while len(entries) > self.max:
            entries.popitem(last=False)

Try it first.

Timeget O(m * d) for m live entries in the tenant (exact hit O(1)); put O(d) amortised

SpaceO(tenants * max_per_tenant * d)

tests.py · 9 passing

tests.pypython22 lines

VOCAB = ["refund", "policy", "return", "shipping", "time", "price"]
SYN = {"returns": "return", "refunds": "refund"}
def embed(text):
    words = [SYN.get(w, w) for w in re.findall(r"[a-z]+", text)]
    return [float(words.count(v)) for v in VOCAB]
now = [0.0]
c = SemanticCache(embed, threshold=0.9, ttl_seconds=60, max_per_tenant=2, clock=lambda: now[0])
c.put("acme", "What is the refund policy?", "30 days")
assert c.get("acme", "  what IS the   refund policy ") == "30 days"
assert c.get("globex", "What is the refund policy?") is None
assert c.get("acme", "refunds policy") == "30 days"
assert c.get("acme", "shipping time") is None
now[0] = 61.0
assert c.get("acme", "What is the refund policy?") is None
c.put("acme", "refund policy", "A")
c.put("acme", "shipping time", "B")
now[0] = 62.0
assert c.get("acme", "refund policy") == "A"
c.put("acme", "price", "C")
assert c.get("acme", "shipping time") is None
assert c.get("acme", "refund policy") == "A" and c.get("acme", "price") == "C"
assert normalise("Ｈello  World!?") == "hello world"

04Staff-level follow-up

At scale the linear scan becomes a per-tenant ANN index (or a vector DB namespace per tenant). Then cover invalidation: key entries by the corpus or prompt version so a document update or prompt change cannot serve stale answers, never cache answers that used per-user permissions unless the ACL set is in the key, and track the false-hit rate by sampling cache hits for judging.