ZerdaZerda

cache.models

On this page

Core data models for the semantic-cache pipeline: thread-agnostic (no locks of their own — concurrency is managed by their owning components), serializable to/from plain dicts for storage backends, and self-describing.

BanditArm

Python · 4 lines
@dataclass
class BanditArm:
    alpha: float = 1.0
    beta: float = 1.0

Overview

A Beta-distribution arm for Thompson Sampling. alpha tracks successes (plus prior), beta tracks failures (plus prior). This is the statistical core that CachePolicyLearner uses to rank and evict IntelligentCacheEntry instances.

Methods

MethodReturnsBehavior
sample_reward() -> floatA draw from Beta(alpha, beta) via numpy.random.betaUsed for exploration during ranking
update_positive(reward: float = 1.0) -> None—alpha += reward
update_negative(penalty: float = 1.0) -> None—beta += penalty
expected_reward (property)floatalpha / (alpha + beta) — the distribution's mean
to_dict() -> Dict[str, float]{"alpha": ..., "beta": ...}Serialization
from_dict(d) -> BanditArm (classmethod)BanditArmDeserialization

Example

Python · 7 lines
from zerda.cache.models import BanditArm

arm = BanditArm()
arm.update_positive()          # a cache hit was judged good
arm.update_negative(0.5)       # a partial negative signal
print(arm.expected_reward)     # -> 2.0 / (2.0 + 1.5) = 0.571...
print(arm.sample_reward())     # stochastic draw for exploration

CostRecord

Python · 6 lines
@dataclass
class CostRecord:
    embedding_cost_usd:  float = 0.0
    llm_input_cost_usd:  float = 0.0
    llm_output_cost_usd: float = 0.0
    retrieval_cost_usd:  float = 0.0

A per-entry breakdown of estimated USD costs, attached to every IntelligentCacheEntry.cost_record. The total_usd property sums all four fields. to_dict() returns a dict with keys embedding_usd, llm_input_usd, llm_output_usd, retrieval_usd, and total_usd. Produced by CostModel and set on entries by CacheManager.put().

IntelligentCacheEntry

Python · 27 lines
@dataclass
class IntelligentCacheEntry:
    entry_id:   str
    tenant_id:  str = "default"
    namespace:  str = "global"
    raw_query:        str = ""
    normalized_query: str = ""
    response:         Any = None
    response_tokens:  int = 0
    query_tokens:     int = 0
    embedding:        Optional[np.ndarray] = None
    embedding_model:  str = ""
    confidence_score:    float = 1.0
    similarity_score:    float = 1.0
    quality_score:       float = 1.0
    bandit_arm:   BanditArm
    usage_count:  int = 0
    success_hits: int = 0
    failure_hits: int = 0
    cost_record:  CostRecord
    created_at:    float
    last_access:   float
    ttl_s:         Optional[float] = None
    size_bytes:    int = 0
    is_shared:     bool = False
    content_hash:  str = ""
    creation_route: RoutingDecision = RoutingDecision.LLM_FALLBACK

Overview

The extended cache entry used throughout the semantic-cache pipeline. Beyond a plain key/value pair, it carries the RL bandit state, cost accounting, multi-tenant metadata, and a content hash for integrity verification — everything CacheManager, CachePolicyLearner, SecurityGuard, TenantManager, and CostModel need to make decisions about a cached answer.

entry_id has no default (it's the one required positional field), but in practice CacheManager.put() always sets it equal to the entry's exact cache key for deduplication purposes.

__post_init__

Two side effects run automatically after construction:

  1. If content_hash is falsy, compute it via _compute_hash().
  2. If size_bytes == 0, compute it via _estimate_size().

Private helpers

MethodBehavior
_compute_hash() -> strSHA-256 hex digest of f"{normalized_query}:{response}" (UTF-8, replacing undecodable bytes)
_estimate_size() -> intlen(normalized_query.encode()) + len(str(response).encode()), plus embedding.nbytes if an embedding is present

Access & lifecycle methods

MethodEffect
touch() -> Nonelast_access = now; usage_count += 1
record_hit(positive: bool, reward: float = 1.0) -> NoneOn positive: success_hits += 1 and bandit_arm.update_positive(reward). Otherwise: failure_hits += 1 and bandit_arm.update_negative(reward)
is_expired() -> boolFalse if ttl_s is None; otherwise now > created_at + ttl_s
age_s() -> floatnow - created_at
idle_s() -> floatnow - last_access

eviction_score(reward_w=0.50, usage_w=0.30, recency_w=0.20, decay=0.995) -> float

Composite score where lower means higher eviction priority. Combines:

  • reward_score — bandit_arm.expected_reward
  • usage_norm — min(1.0, usage_count / 100.0)
  • recency_norm — max(0.0, 1.0 - idle_s() / 86400.0) (a 24-hour recency window)
Text · 2 lines
score = (reward_w*reward_score + usage_w*usage_norm + recency_w*recency_norm)
        * decay ** max(0, usage_count - 1)

The multiplicative decay ** (usage_count - 1) term means, counter-intuitively, that more-used entries have their composite score shrunk more by this formula (since the exponent grows with usage_count), all else being equal — an aspect worth understanding before tuning the weights; see Known Issues for discussion.

Called by CachePolicyLearner.eviction_ranking() with decay=cfg.usage_decay, leaving reward_w/usage_w/recency_w at their defaults (the analogous EvictionConfig.reward_weight/usage_weight/recency_weight fields are not threaded through to this call in the analyzed source).

Serialization

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

Returns a JSON-serializable dict with every field, converting embedding to a plain list (.tolist()) or None, bandit_arm/cost_record to their own to_dict() forms, and creation_route to its .value string.

from_storage_dict(cls, d: Dict[str, Any]) -> IntelligentCacheEntry (classmethod)

The inverse operation, reconstructing numpy arrays, BanditArm, CostRecord, and RoutingDecision from the plain dict. Used by every storage backend (SQLiteStorage, RedisStorage) on read.

[!WARNING] from_storage_dict uses direct dict indexing, not .get(), for most fields Nearly every field access in from_storage_dict (e.g. d["entry_id"], d["tenant_id"], d["usage_count"]) uses [] rather than .get(...) with a fallback. A payload written by a different (e.g. older or newer) version of this dataclass that is missing any of these keys will raise KeyError rather than degrading gracefully. The exceptions are the nested cost_record sub-fields, which do use .get(..., 0.0).

Examples

Python · 20 lines
from zerda.cache.models import IntelligentCacheEntry
import numpy as np

entry = IntelligentCacheEntry(
    entry_id="abc123",
    tenant_id="acme",
    raw_query="What is RAG?",
    normalized_query="what is rag",
    response="Retrieval-Augmented Generation combines retrieval with generation.",
    embedding=np.random.rand(128).astype(np.float32),
    ttl_s=3600,
)

entry.touch()
entry.record_hit(positive=True)
print(entry.eviction_score())

payload = entry.to_storage_dict()
restored = IntelligentCacheEntry.from_storage_dict(payload)
assert restored.content_hash == entry.content_hash

CacheLookupResult

Python · 8 lines
@dataclass
class CacheLookupResult:
    hit:          bool             = False
    decision:     RoutingDecision  = RoutingDecision.LLM_FALLBACK
    entry:        Optional[IntelligentCacheEntry] = None
    similarity:   float            = 0.0
    latency_ms:   float            = 0.0
    cost_saved_usd: float          = 0.0

The return type of CacheManager.get() / get_sync(). Exposes a convenience response property:

Python · 3 lines
@property
def response(self) -> Any:
    return self.entry.response if self.entry else None

so callers can write result.response instead of result.entry.response if result.entry else None.

Example

Python · 5 lines
result = await manager.get("hello")
if result.hit:
    print(result.decision, result.similarity, result.response)
else:
    print("miss — call your LLM")

TenantQuotaState

Python · 10 lines
@dataclass
class TenantQuotaState:
    tenant_id:        str
    memory_used_mb:   float = 0.0
    entry_count:      int   = 0
    requests_this_min: int  = 0
    last_rpm_reset:   float
    memory_quota_mb:  float = 512.0
    max_entries:      int   = 10_000
    request_quota_rpm: int  = 1000

Mutable, in-memory quota-tracking state for one tenant, owned and updated by TenantManager.

Method / propertyBehavior
reset_rpm_if_needed() -> NoneIf ≥60s have elapsed since last_rpm_reset, resets requests_this_min to 0 and bumps last_rpm_reset
memory_ok (property)memory_used_mb < memory_quota_mb
entries_ok (property)entry_count < max_entries
rpm_ok (property)Calls reset_rpm_if_needed() then checks requests_this_min < request_quota_rpm
charge_request() -> boolCalls reset_rpm_if_needed(); if not rpm_ok, returns False without charging; otherwise increments requests_this_min and returns True

Related pages