cache.cost_model

Overview

Two classes, per the module docstring: CostModel estimates USD cost for each operation type, and DecisionEngine determines the optimal action (cache hit vs. LLM call) by maximizing a utility function balancing cost, latency, and quality.

OperationCostEstimate

@dataclass
class OperationCostEstimate:
    embedding_usd:   float = 0.0
    llm_input_usd:   float = 0.0
    llm_output_usd:  float = 0.0
    retrieval_usd:   float = 0.0

total_usd (property) sums all four fields. Returned by every CostModel.estimate_* method.

CostModel

class CostModel:
    def __init__(self, config: CostConfig) -> None

A stateless estimator (_CHARS_PER_TOKEN = 4.0 class constant — the standard rough English-text heuristic of ~4 characters per token).

estimate_embedding(self, text: str) -> float

Overview: approximates the USD cost of embedding one string.

Algorithm: tokens = len(text) / 4.0; cost = (tokens / 1000) * config.embedding_cost_per_1k_tokens.

Complexity: O(1) given len(text).

estimate_llm_call(self, input_text: str, expected_output_chars: int = 500, input_tokens: Optional[int] = None, output_tokens: Optional[int] = None) -> OperationCostEstimate

Overview: approximates the USD cost of a full LLM call (input + output tokens, plus a fixed internal "retrieval" compute cost).

Parameters:

Algorithm:

in_tok  = input_tokens  or len(input_text) / 4
out_tok = output_tokens or expected_output_chars / 4
llm_input_usd  = (in_tok / 1000)  * cost_per_1k_input
llm_output_usd = (out_tok / 1000) * cost_per_1k_output
retrieval_usd  = config.retrieval_cost_per_query   # fixed, not scaled by anything

Returns: OperationCostEstimate.

Example:

from zerda.cache.cost_model import CostModel
from zerda.cache.config_cache import CostConfig

model = CostModel(CostConfig())
est = model.estimate_llm_call("Summarize this document...", output_tokens=200)
print(est.total_usd)

estimate_cache_serve(self, entry: IntelligentCacheEntry) -> OperationCostEstimate

Returns an estimate with only retrieval_usd set (config.retrieval_cost_per_query) — the cost model treats serving from zerda.cache as having negligible embedding/LLM cost, only a small fixed internal compute cost.

cost_saved(self, entry: IntelligentCacheEntry, query_text: str) -> float

Overview: the headline "money saved" figure, used throughout the metrics layer.

Algorithm:

llm_est   = estimate_llm_call(query_text, output_tokens=entry.response_tokens)
cache_est = estimate_cache_serve(entry)
saved     = max(0.0, llm_est.total_usd - cache_est.total_usd)

Clamped at 0.0 so a cache hit is never reported as costing more than the equivalent LLM call.

RoutingResult

@dataclass
class RoutingResult:
    decision:          RoutingDecision
    confidence:        float
    utility_score:     float
    estimated_cost_usd: float
    reason:            str = ""
    lookup_latency_ms: float = 0.0

The return type of DecisionEngine.decide(). reason is a short, human-readable explanation string (e.g. "Cache utility 0.812 ≥ LLM utility 0.750"), useful for logging and debugging routing decisions.

DecisionEngine

class DecisionEngine:
    def __init__(self, config: CostConfig, cost_model: CostModel) -> None

Per its docstring: given a candidate cache entry (or None), decides whether to serve from cache or fall back to an LLM, using:

U = cost_weight * (1 - normalised_cost)
  + quality_weight * quality
  + latency_weight * (1 - normalised_latency)

Also maintains an adaptive similarity-threshold bonus that shifts based on observed hit quality, and cumulative cost-savings counters.

Class constants

decide(self, query: str, candidate: Optional[IntelligentCacheEntry], similarity: float, semantic_threshold: float, lookup_latency_ms: float = 0.0) -> RoutingResult

Overview

The core routing decision. Called once per CacheManager.get() invocation that reaches the semantic-search stage, with the top-ranked candidate (if any).

Internal Workflow

1. No candidate: immediately return LLM_FALLBACK with confidence=1.0, utility_score=0.0, and the estimated LLM call cost as estimated_cost_usd.
2. Effective similarity threshold: max(config.cost_weight, semantic_threshold + self._current_threshold_bonus). Note this floors the effective threshold at config.cost_weight (default 0.40) regardless of how low semantic_threshold or the adaptive bonus would otherwise push it — see Known Issues for why this specific floor value looks like it may be an unintended reuse of cost_weight rather than a deliberate minimum-similarity constant.
3. Below effective threshold: return LLM_FALLBACK, incrementing _recent_cache_misses, with reason naming the actual similarity and threshold values.
4. At/above threshold: compute cache_utility (via _cache_utility) and llm_utility (via _llm_utility).
   • If cache_utility >= llm_utility: accumulate cost_saved into _total_saved_usd, increment _recent_cache_hits, and return EXACT_HIT if similarity >= 0.999 else SEMANTIC_HIT.
   • Otherwise: accumulate the estimated LLM cost into _total_llm_cost_usd, increment _recent_cache_misses, and return LLM_FALLBACK.

Algorithm — utility functions (private)

_cache_utility(query, entry, similarity, latency_ms) -> float:

cost_norm  = 1 - min(1, cache_serve_cost / 0.05)
quality    = entry.confidence_score * similarity
lat_norm   = 1 - min(1, latency_ms / 2000)
utility    = cost_weight*cost_norm + quality_weight*quality + latency_weight*lat_norm

_llm_utility(query) -> float:

cost_norm = 1 - min(1, llm_call_cost / 0.05)
lat_norm  = 1 - min(1, 1000 / 2000)      # LLM latency is hard-coded at 1000ms for this estimate
utility   = cost_weight*cost_norm + quality_weight*1.0 + latency_weight*lat_norm
           # LLM quality is always assumed to be the maximum, 1.0

[!NOTE] LLM latency in _llm_utility is a hard-coded constant, not measured _llm_utility always uses a fixed 1000.0 ms figure for the LLM's latency component — it has no way to know the actual latency of a downstream LLM call the caller hasn't made yet. This is a deliberate simplifying assumption baked into the formula, not a bug, but worth knowing when interpreting RoutingResult.utility_score for the LLM path.

Complexity

O(1) — a small, fixed number of arithmetic operations per call.

Thread Safety

Not thread-safe against concurrent calls. _recent_cache_hits, _recent_cache_misses, _total_saved_usd, and _total_llm_cost_usd are plain instance attributes mutated without any lock. DecisionEngine is instantiated once per CacheManager and its decide() method is awaited from CacheManager.get()'s async control flow, so concurrent coroutines sharing one event loop won't race at the Python-bytecode level for these specific increments (no await occurs between reading and incrementing within decide()), but concurrent access from multiple threads (e.g. via get_sync() calls on different threads) is not synchronized. See Known Issues.

update_threshold(self, was_positive: bool) -> None

Overview: nudges _current_threshold_bonus based on feedback quality — called from CacheManager.feedback().

Algorithm: step = 0.01; if was_positive, subtract step from the bonus (relaxing the effective threshold, clamped at a floor of -0.10); otherwise add step (tightening, clamped at a ceiling of +0.10).

get_stats(self) -> Dict[str, float]

Returns cache_decision_rate (hits / (hits+misses)), total_saved_usd, total_llm_cost_usd, and threshold_bonus.

Examples

from zerda.cache.cost_model import CostModel, DecisionEngine
from zerda.cache.config_cache import CostConfig

cfg = CostConfig()
model = CostModel(cfg)
engine = DecisionEngine(cfg, model)

result = engine.decide(
    query="what is rag",
    candidate=my_entry,
    similarity=0.93,
    semantic_threshold=0.85,
    lookup_latency_ms=12.4,
)
print(result.decision, result.reason)

engine.update_threshold(was_positive=True)   # after positive feedback
print(engine.get_stats())

Related APIs

• CostConfig — all weight/cost tunables
• CacheManager.get() / feedback() — the sole callers
• IntelligentCacheEntry — the candidate type