Class ModelExperienceStore

uncommon_route/model_experience.py:102–465

class ModelExperienceStore:
    def __init__(
        self,
        storage: ModelExperienceStorage | None = None,
        *,
        alpha: float = 0.25,
        now_fn: object = None,
    ) -> None:
        self._storage = storage or FileModelExperienceStorage()
        self._alpha = max(0.05, min(0.9, alpha))
        self._now = now_fn if callable(now_fn) else time.time
        self._records: dict[str, ModelExperienceRecord] = {}
        self._load()

    def observe(
        self,
        model: str,
        mode: RoutingMode | str,
        tier: Tier | str,
        *,
        success: bool,
        ttft_ms: float | None = None,
        tps: float | None = None,
        total_input_tokens: int | None = None,
        uncached_input_tokens: int | None = None,
        cache_read_tokens: int = 0,
        cache_write_tokens: int = 0,
        input_cost_multiplier: float | None = None,
    ) -> None:
        record = self._get_or_create(model, mode, tier)
        record.requests += 1
        if success:
            record.successes += 1
        else:
            record.failures += 1
        target = 1.0 if success else 0.0
        record.success_ewma = self._blend(record.success_ewma, target)
        if ttft_ms is not None and ttft_ms > 0:
            record.ttft_ms_ewma = self._blend_metric(record.ttft_ms_ewma, ttft_ms)
        if tps is not None and tps > 0:
            record.tps_ewma = self._blend_metric(record.tps_ewma, tps)
        normalized_total_input = max(
            0,
            int(total_input_tokens or 0),
            int(uncached_input_tokens or 0) + max(0, int(cache_read_tokens)) + max(0, int(cache_write_tokens)),
        )
        if normalized_total_input > 0:
            hit_ratio = max(0.0, min(1.0, float(cache_read_tokens) / float(normalized_total_input)))
            write_ratio = max(0.0, min(1.0, float(cache_write_tokens) / float(normalized_total_input)))
            record.cache_hit_ratio_ewma = self._blend_metric(record.cache_hit_ratio_ewma, hit_ratio)
            record.cache_write_ratio_ewma = self._blend_metric(record.cache_write_ratio_ewma, write_ratio)
        if input_cost_multiplier is not None and input_cost_multiplier > 0:
            record.input_cost_multiplier_ewma = self._blend_metric(
                record.input_cost_multiplier_ewma,
                input_cost_multiplier,
            )
        # Only update reward for FAILURES (real signal).
        # HTTP 200 success is an availability signal, not a quality signal.