CacheManager

The route-decision cache: a thread-safe LRU+TTL cache mapping a normalized query hash to (route_name, group_name), letting repeated queries skip embedding and scoring entirely.

_LRUTTLCache (internal)

class _LRUTTLCache:
    def __init__(self, max_size: int, ttl: int): ...

Generic O(1) LRU cache with per-entry TTL, backing CacheManager. Documented here because its exact eviction semantics directly determine CacheManager's observable behavior.

Constructor

Internally backed by collections.OrderedDict (insertion/access order tracks recency) and a threading.Lock guarding every operation.

Methods

get(self, key: str) -> Optional[Any]

Under the lock: if key is absent, increments misses and returns None. If present but time.time() - timestamp > ttl, deletes the expired entry, increments misses, returns None. Otherwise moves the entry to the end (most-recently-used position via move_to_end), increments hits, and returns the stored value.

Complexity: O(1) amortized (dict lookup + OrderedDict.move_to_end).

set(self, key: str, value: Any)

Under the lock: if the key already exists, moves it to the end first; then stores (value, time.time()) — this always refreshes the timestamp on overwrite, so re-setting an existing key resets its TTL clock. If the cache now exceeds max_size, calls _evict().

delete(self, key: str) -> bool

Removes a key if present; returns whether it was found.

clear(self)

Empties the entire cache and resets hits/misses counters to 0.

[!NOTE] clear() resets hit/miss statistics Unlike delete()/get() misses, clear() doesn't just empty the data — it also zeroes self.hits and self.misses. Any historical hit-rate reporting via CacheManager.stats() reflects data since the last clear() call, not since process start, if clear() has ever been invoked.

size (property)

len(self._data).

hit_rate (property)

hits / (hits + misses) * 100 if any requests have been made, else 0.0. Returned as a percentage (0–100), not a 0.0–1.0 fraction.

_evict(self) (private)

Two-phase eviction, called automatically from set() when over capacity:

1. Expire first: scan all entries, remove any where now - timestamp > ttl.
2. Then LRU-trim: if still over max_size after expiration, repeatedly popitem(last=False) (removes the oldest / least-recently-used entry, since OrderedDict.popitem(last=False) pops from the front) until at or under the limit.

Thread safety: Every public method acquires self._lock — safe for concurrent use across threads. _evict() itself is only ever called from within set()'s already-held lock, so it doesn't re-acquire it.

CacheManager

class CacheManager:
    def __init__(self, cfg: CacheConfig): ...

Public caching interface used by HierarchicalRouter. Despite the module docstring describing a conceptual "two-tier" cache (route decisions + embeddings), only the route-decision tier is actually implemented in this class — the embedding cache described in the docstring lives entirely inside EmbeddingProvider (see EmbeddingProvider) and is not surfaced, wrapped, or coordinated by CacheManager in the analyzed source.

Constructor

Reads cfg.enabled into self._enabled, and constructs a single _LRUTTLCache(max_size=cfg.max_size, ttl=cfg.ttl_seconds) as self._route_cache.

Methods

get_decision(self, query: str) -> Optional[Tuple[str, Optional[str]]]

Returns None immediately if caching is disabled (cfg.enabled = False). Otherwise looks up self._route_cache.get(self._query_key(query)), returning a (route_name, group_name) tuple or None.

set_decision(self, query: str, route_name: str, group_name: Optional[str] = None)

No-op if disabled. Otherwise stores (route_name, group_name) under the query's normalized hash key.

invalidate(self, query: str) -> bool

Removes a single cached query's decision (e.g. useful after updating that specific route). Returns whether an entry was found and removed. Not gated by self._enabled — this method runs regardless of the enabled flag (unlike get_decision/set_decision), though if caching was never enabled there would be nothing to invalidate in practice.

clear(self)

Flushes the entire route decision cache (and resets its hit/miss counters, per _LRUTTLCache.clear()). Called automatically by HierarchicalRouter whenever routes/groups are registered, updated, unregistered, or re-encoded — see HierarchicalRouter.

stats(self) -> Dict[str, Any]

Returns:

{
    "enabled": bool,
    "ttl_seconds": int,
    "route_cache": {
        "size": int,
        "max_size": int,
        "hits": int,
        "misses": int,
        "hit_rate": float,  # percentage, rounded to 2 decimals
    },
}

Query normalization

CacheManager._query_key(query: str) -> str (static method)

normalised = " ".join(query.lower().split())
return hashlib.sha256(normalised.encode()).hexdigest()

Lowercases the query and collapses all runs of whitespace (including leading/trailing) into single spaces before hashing with SHA-256. This means "What's the WEATHER today?" and "what's the weather today?" hash identically, but punctuation/wording differences ("weather today" vs "today's weather") produce completely different keys — there is no semantic/fuzzy matching in the cache key, only exact-normalized-string matching.

Magic methods

__repr__(self) -> str

f"CacheManager(enabled={self._enabled}, route_cache_size={self._route_cache.size})".

Examples

1. Basic:

from zerda.router import CacheManager 
from zerda.router import CacheConfig

cache = CacheManager(CacheConfig(ttl_seconds=60, max_size=500))
cache.set_decision("what's the weather?", route_name="weather", group_name="tools")
hit = cache.get_decision("What's the WEATHER?")  # normalized — still a hit
print(hit)  # ("weather", "tools")

2. Invalidating a stale decision after a route change:

router.get_route("weather").add_example("what's it like outside")
router.re_encode_route("weather")  # this already calls cache.clear() internally
# Or, to invalidate just one query without a full flush:
cache.invalidate("what's the weather?")

3. Monitoring cache effectiveness:

stats = cache.stats()
print(f"Hit rate: {stats['route_cache']['hit_rate']}%")
if stats['route_cache']['size'] >= stats['route_cache']['max_size'] * 0.9:
    print("Cache nearing capacity — consider raising max_size")

4. Disabling caching entirely (e.g. for testing deterministic routing):

cache = CacheManager(CacheConfig(enabled=False))
cache.set_decision("test query", route_name="x")  # no-op
print(cache.get_decision("test query"))  # None

Best practices

• Don't rely on CacheManager for embedding-level caching — that's EmbeddingProvider's job, and it has different (unbounded, no-TTL) characteristics. See the caveat under Configuration → CacheConfig.
• Set ttl_seconds based on how often your route configuration changes in practice — since most mutation methods on HierarchicalRouter already call cache.clear() for you, TTL mainly protects against routing decisions becoming stale due to feedback bias drift (a route's score_bias shifting enough that the previously-cached decision is no longer the best one) rather than route/group structural changes.

Related APIs: HierarchicalRouter, CacheConfig, EmbeddingProvider