diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260804000000_add_auto_router_session_rollup/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260804000000_add_auto_router_session_rollup/migration.sql new file mode 100644 index 000000000000..2e3e26c295e2 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260804000000_add_auto_router_session_rollup/migration.sql @@ -0,0 +1,37 @@ +CREATE TABLE IF NOT EXISTS "LiteLLM_AutoRouterSession" ( + api_key TEXT NOT NULL, + session_id TEXT NOT NULL, + model_group TEXT NOT NULL, + router_kind TEXT NOT NULL, + baseline_model TEXT, + turns INTEGER NOT NULL DEFAULT 0, + turns_with_usage INTEGER NOT NULL DEFAULT 0, + total_tokens BIGINT NOT NULL DEFAULT 0, + spend DOUBLE PRECISION NOT NULL DEFAULT 0, + baseline_spend DOUBLE PRECISION NOT NULL DEFAULT 0, + first_visit_turns INTEGER NOT NULL DEFAULT 0, + first_visit_hits INTEGER NOT NULL DEFAULT 0, + warm_turns INTEGER NOT NULL DEFAULT 0, + warm_hits INTEGER NOT NULL DEFAULT 0, + expired_turns INTEGER NOT NULL DEFAULT 0, + expired_hits INTEGER NOT NULL DEFAULT 0, + unordered_turns INTEGER NOT NULL DEFAULT 0, + unordered_hits INTEGER NOT NULL DEFAULT 0, + unknown_ttl_turns INTEGER NOT NULL DEFAULT 0, + unknown_ttl_hits INTEGER NOT NULL DEFAULT 0, + cache_5m_turns INTEGER NOT NULL DEFAULT 0, + cache_1h_turns INTEGER NOT NULL DEFAULT 0, + cache_ttl_unknown_turns INTEGER NOT NULL DEFAULT 0, + tiers JSONB NOT NULL DEFAULT '{}', + first_turn_at TIMESTAMP(3) NOT NULL, + last_turn_at TIMESTAMP(3) NOT NULL, + updated_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "LiteLLM_AutoRouterSession_pkey" PRIMARY KEY (api_key, session_id, model_group) +); + +CREATE INDEX IF NOT EXISTS "idx_auto_router_session_started" + ON "LiteLLM_AutoRouterSession" (first_turn_at); + +CREATE INDEX IF NOT EXISTS "idx_auto_router_session_activity" + ON "LiteLLM_AutoRouterSession" (last_turn_at); diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index 17339541fd96..ae4bcdc9abe1 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -1393,6 +1393,45 @@ model LiteLLM_AdaptiveRouterSession { @@index([last_activity_at], map: "idx_adaptive_router_session_activity") } +model LiteLLM_AutoRouterSession { + api_key String + session_id String + model_group String + router_kind String + baseline_model String? + + turns Int @default(0) + turns_with_usage Int @default(0) + total_tokens BigInt @default(0) + spend Float @default(0) + baseline_spend Float @default(0) + + first_visit_turns Int @default(0) + first_visit_hits Int @default(0) + warm_turns Int @default(0) + warm_hits Int @default(0) + expired_turns Int @default(0) + expired_hits Int @default(0) + unordered_turns Int @default(0) + unordered_hits Int @default(0) + unknown_ttl_turns Int @default(0) + unknown_ttl_hits Int @default(0) + + cache_5m_turns Int @default(0) + cache_1h_turns Int @default(0) + cache_ttl_unknown_turns Int @default(0) + + tiers Json @default("{}") + + first_turn_at DateTime + last_turn_at DateTime + updated_at DateTime @default(now()) @updatedAt + + @@id([api_key, session_id, model_group]) + @@index([first_turn_at], map: "idx_auto_router_session_started") + @@index([last_turn_at], map: "idx_auto_router_session_activity") +} + // --------------------------------------------------------------------------- // Workflow Run Tracking // diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py index 4e5c5daaa325..0cda6fb46a58 100644 --- a/litellm/proxy/db/db_spend_update_writer.py +++ b/litellm/proxy/db/db_spend_update_writer.py @@ -51,6 +51,7 @@ ToolDiscoveryQueue, ) from litellm.proxy.route_llm_request import ROUTE_ENDPOINT_MAPPING +from litellm.proxy.spend_tracking.auto_router_sessions import AutoRouterSessionQueue, build_turn_facts from litellm.proxy.spend_tracking.compression_savings import ( extract_compression_saved_tokens, ) @@ -132,6 +133,46 @@ def __init__( self.daily_agent_spend_update_queue = DailySpendUpdateQueue() self.daily_org_spend_update_queue = DailySpendUpdateQueue() self.daily_tag_spend_update_queue = DailySpendUpdateQueue() + self.auto_router_session_queue = AutoRouterSessionQueue() + + def _enqueue_auto_router_turn(self, payload: SpendLogsPayload) -> None: + """Stage one auto-routed turn for the benchmarks rollup. + + Runs once per request rather than beside the daily transactions, which are built + per entity type and would count every turn six times over. Independent of + ``disable_spend_logs``, because the rollup is what the benchmarks dashboard reads + now. Never raises, and never blocks: a full queue drops the turn rather than + applying backpressure to spend tracking. + """ + try: + metadata: Final[SpendLogsMetadata] = json.loads(payload["metadata"]) + if not metadata.get("routing_decision"): + return + usage_raw: Final = metadata.get("usage_object") + usage_obj: Final = usage_raw if isinstance(usage_raw, dict) else None + cache_read_tokens: Final = _extract_cache_read_tokens(usage_obj) if usage_obj is not None else 0 + savings: Final = compute_savings_spend( + model=payload.get("model", None), + custom_llm_provider=payload.get("custom_llm_provider", None), + compression_saved_tokens=extract_compression_saved_tokens(metadata), + cache_read_input_tokens=cache_read_tokens, + routing_decision=metadata.get("routing_decision"), + model_id=payload.get("model_id"), + llm_router=_get_llm_router, + usage_object=usage_obj, + cost_breakdown=metadata.get("cost_breakdown"), + ) + turn: Final = build_turn_facts( + payload=payload, + metadata=metadata, + autorouter_savings=savings.autorouter, + cache_read_tokens=cache_read_tokens, + cache_creation_tokens=_extract_cache_creation_tokens(usage_obj) if usage_obj is not None else 0, + ) + if turn is not None: + self.auto_router_session_queue.update_queue.put_nowait(turn) + except Exception as e: # noqa: BLE001 # a dashboard rollup must never fail spend tracking + verbose_proxy_logger.debug("auto_router_sessions: could not stage turn (%s)", e) async def update_database( # LiteLLM management object fields @@ -209,6 +250,8 @@ async def update_database( "disable_spend_logs=True. Skipping writing spend logs to db. Other spend updates - Key/User/Team table will still occur." ) + self._enqueue_auto_router_turn(payload=payload) + # Single task replaces 11 create_task() calls asyncio.create_task( self._batch_database_updates( diff --git a/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py b/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py index 9e2a3acc42b5..b7729de42f77 100644 --- a/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py +++ b/litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py @@ -15,6 +15,9 @@ from litellm.proxy.db.db_transaction_queue.spend_logs_partition_manager import ( SpendLogsPartitionManager, ) +from litellm.proxy.spend_tracking.auto_router_benchmarks import ( + AUTO_ROUTER_SESSION_RETENTION_DAYS, +) from litellm.proxy.utils import PrismaClient @@ -186,22 +189,42 @@ async def _delete_old_tool_index_rows(self, prisma_client: PrismaClient, cutoff_ time_column="start_time", ) + async def _delete_old_auto_router_sessions(self, prisma_client: PrismaClient, cutoff_date: datetime) -> int: + """Expire auto-router rollups past the given cutoff. + + Keyed on last activity rather than session start, so a conversation still running + when the cutoff passes is not pruned out from under itself. + """ + return await self._delete_old_rows_batched( + prisma_client, + cutoff_date, + table_name="LiteLLM_AutoRouterSession", + key_columns=("api_key", "session_id", "model_group"), + time_column="last_turn_at", + ) + + def _auto_router_session_cutoff(self, retention_seconds: float | None) -> datetime: + """Rows the benchmarks endpoint can no longer read are collected at the read + horizon even with no retention configured; a shorter configured retention wins.""" + now: Final = datetime.now(timezone.utc) + horizon: Final = now - timedelta(days=AUTO_ROUTER_SESSION_RETENTION_DAYS) + return horizon if retention_seconds is None else max(horizon, now - timedelta(seconds=retention_seconds)) + async def cleanup_old_spend_logs(self, prisma_client: PrismaClient) -> None: """ Main cleanup function. Deletes old spend logs in batches. If pod_lock_manager is available, ensures only one pod runs cleanup. If no pod_lock_manager, runs cleanup without distributed locking. + + Spend logs and their tool index only expire when + ``maximum_spend_logs_retention_period`` is configured. The auto-router rollup is + collected on every run regardless; see ``_auto_router_session_cutoff``. """ lock_acquired = False try: verbose_proxy_logger.info("Cleanup job triggered at %s", datetime.now()) - if not self._should_delete_spend_logs(): - return - - if self.retention_seconds is None: - verbose_proxy_logger.error("Retention seconds is None, cannot proceed with cleanup") - return + retention_seconds: Final = self.retention_seconds if self._should_delete_spend_logs() else None # If we have a pod lock manager, try to acquire the lock if self.pod_lock_manager and self.pod_lock_manager.redis_cache: @@ -219,31 +242,39 @@ async def cleanup_old_spend_logs(self, prisma_client: PrismaClient) -> None: verbose_proxy_logger.info("Another pod is already running cleanup") return - cutoff_date: Final = datetime.now(timezone.utc) - timedelta(seconds=float(self.retention_seconds)) - verbose_proxy_logger.info("Removing logs older than %s", cutoff_date.isoformat()) + if retention_seconds is not None: + cutoff_date: Final = datetime.now(timezone.utc) - timedelta(seconds=float(retention_seconds)) + verbose_proxy_logger.info("Removing logs older than %s", cutoff_date.isoformat()) - if self.general_settings.get( - "use_spend_logs_partitioning", False - ) and await self.partition_manager.is_partitioned(prisma_client): - await self.partition_manager.ensure_partitions(prisma_client) - dropped: Final = await self.partition_manager.drop_partitions_older_than(prisma_client, cutoff_date) - verbose_proxy_logger.info( - "Dropped %d expired spend-log partitions: %s", - len(dropped), - dropped, - ) - # DROP only reclaims whole expired partitions. Expired rows can - # still sit in the DEFAULT partition (backfill, coverage gaps) - # or in a partition that spans the cutoff, so retention must - # also delete those stragglers row-wise. - total_deleted = await self._delete_old_logs(prisma_client, cutoff_date) - verbose_proxy_logger.info("Deleted %s expired logs not covered by dropped partitions", total_deleted) - else: - total_deleted = await self._delete_old_logs(prisma_client, cutoff_date) - verbose_proxy_logger.info("Deleted %s logs", total_deleted) + if self.general_settings.get( + "use_spend_logs_partitioning", False + ) and await self.partition_manager.is_partitioned(prisma_client): + await self.partition_manager.ensure_partitions(prisma_client) + dropped: Final = await self.partition_manager.drop_partitions_older_than(prisma_client, cutoff_date) + verbose_proxy_logger.info( + "Dropped %d expired spend-log partitions: %s", + len(dropped), + dropped, + ) + # DROP only reclaims whole expired partitions. Expired rows can + # still sit in the DEFAULT partition (backfill, coverage gaps) + # or in a partition that spans the cutoff, so retention must + # also delete those stragglers row-wise. + total_deleted = await self._delete_old_logs(prisma_client, cutoff_date) + verbose_proxy_logger.info( + "Deleted %s expired logs not covered by dropped partitions", total_deleted + ) + else: + total_deleted = await self._delete_old_logs(prisma_client, cutoff_date) + verbose_proxy_logger.info("Deleted %s logs", total_deleted) - index_deleted: Final = await self._delete_old_tool_index_rows(prisma_client, cutoff_date) - verbose_proxy_logger.info("Deleted %s expired tool index rows", index_deleted) + index_deleted: Final = await self._delete_old_tool_index_rows(prisma_client, cutoff_date) + verbose_proxy_logger.info("Deleted %s expired tool index rows", index_deleted) + + sessions_deleted: Final = await self._delete_old_auto_router_sessions( + prisma_client, self._auto_router_session_cutoff(retention_seconds) + ) + verbose_proxy_logger.info("Deleted %s expired auto-router session rollups", sessions_deleted) except Exception as e: # .exception() captures the traceback; str(e) alone on a Prisma/DB diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index e343d46f872a..86b6e6775e3f 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -16,10 +16,11 @@ import traceback import warnings from collections.abc import AsyncGenerator, Callable, Mapping -from datetime import datetime, timedelta, timezone +from datetime import date, datetime, timedelta, timezone from types import UnionType from typing import ( TYPE_CHECKING, + Annotated, Any, Final, Literal, @@ -544,6 +545,10 @@ def generate_feedback_box(): from litellm.proxy.route_llm_request import route_request from litellm.proxy.search_endpoints.endpoints import router as search_router from litellm.proxy.shutdown.graceful_shutdown_manager import GracefulShutdownManager +from litellm.proxy.spend_tracking.auto_router_benchmarks import ( + AutoRouterBenchmarksResponse, + fetch_benchmarks, +) from litellm.proxy.spend_tracking.budget_reservation import get_budget_window_start from litellm.proxy.spend_tracking.spend_management_endpoints import ( router as spend_management_router, @@ -5927,7 +5932,9 @@ async def _reschedule_spend_log_cleanup_job(self): """ Reschedule the spend log cleanup job based on current general_settings. This is called when maximum_spend_logs_retention_period is updated dynamically. - If the retention period is None, the job will be removed. + The job is always rescheduled: spend-log pruning inside it still requires the + retention setting, but the auto-router rollup is garbage collected past its + read horizon regardless. """ global scheduler, general_settings, prisma_client if scheduler is None: @@ -5940,53 +5947,50 @@ async def _reschedule_spend_log_cleanup_job(self): except Exception: pass # Job might not exist, which is fine - # Schedule new job if retention period is set (not None) - retention_period: Final = general_settings.get("maximum_spend_logs_retention_period") - if retention_period is not None: - from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import ( - SpendLogCleanup, - ) + from litellm.proxy.db.db_transaction_queue.spend_log_cleanup import ( + SpendLogCleanup, + ) - spend_log_cleanup: Final = SpendLogCleanup() - cleanup_cron: Final = general_settings.get("maximum_spend_logs_cleanup_cron") + spend_log_cleanup: Final = SpendLogCleanup() + cleanup_cron: Final = general_settings.get("maximum_spend_logs_cleanup_cron") - if cleanup_cron: - from apscheduler.triggers.cron import CronTrigger + if cleanup_cron: + from apscheduler.triggers.cron import CronTrigger - try: - cron_trigger: Final = CronTrigger.from_crontab(cleanup_cron) - scheduler.add_job( - spend_log_cleanup.cleanup_old_spend_logs, - cron_trigger, - args=[prisma_client], - id="spend_log_cleanup_job", - replace_existing=True, - misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, - ) - verbose_proxy_logger.info("Spend log cleanup rescheduled with cron: %s", cleanup_cron) - except ValueError: - verbose_proxy_logger.error("Invalid maximum_spend_logs_cleanup_cron value: %s", cleanup_cron) - else: - # Interval-based scheduling (existing behavior) - from litellm.litellm_core_utils.duration_parser import ( - duration_in_seconds, + try: + cron_trigger: Final = CronTrigger.from_crontab(cleanup_cron) + scheduler.add_job( + spend_log_cleanup.cleanup_old_spend_logs, + cron_trigger, + args=[prisma_client], + id="spend_log_cleanup_job", + replace_existing=True, + misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, ) + verbose_proxy_logger.info("Spend log cleanup rescheduled with cron: %s", cleanup_cron) + except ValueError: + verbose_proxy_logger.error("Invalid maximum_spend_logs_cleanup_cron value: %s", cleanup_cron) + else: + # Interval-based scheduling (existing behavior) + from litellm.litellm_core_utils.duration_parser import ( + duration_in_seconds, + ) - retention_interval: Final = general_settings.get("maximum_spend_logs_retention_interval", "1d") - try: - interval_seconds: Final = duration_in_seconds(retention_interval) - scheduler.add_job( - spend_log_cleanup.cleanup_old_spend_logs, - "interval", - seconds=interval_seconds + random.randint(0, 60), - args=[prisma_client], - id="spend_log_cleanup_job", - replace_existing=True, - misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, - ) - verbose_proxy_logger.info("Spend log cleanup rescheduled with interval: %s", retention_interval) - except ValueError: - verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value") + retention_interval: Final = general_settings.get("maximum_spend_logs_retention_interval", "1d") + try: + interval_seconds: Final = duration_in_seconds(retention_interval) + scheduler.add_job( + spend_log_cleanup.cleanup_old_spend_logs, + "interval", + seconds=interval_seconds + random.randint(0, 60), + args=[prisma_client], + id="spend_log_cleanup_job", + replace_existing=True, + misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, + ) + verbose_proxy_logger.info("Spend log cleanup rescheduled with interval: %s", retention_interval) + except ValueError: + verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value") async def _update_general_settings(self, db_general_settings: Json | None): """ @@ -8199,6 +8203,19 @@ async def initialize_scheduled_background_jobs( f"({tag_spend_update_interval / batch_writing_interval:.1f}x main job interval)" ) + ### AUTO-ROUTER BENCHMARKS ROLLUP (separate scheduler job) ### + from litellm.proxy.utils import update_auto_router_sessions + + scheduler.add_job( + update_auto_router_sessions, + "interval", + seconds=batch_writing_interval, + args=(prisma_client, proxy_logging_obj), + id="update_auto_router_sessions_job", + replace_existing=True, + misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, + ) + ### MONITOR SPEND LOGS QUEUE (queue-size-based job) ### if general_settings.get("disable_spend_logs", False) is False: from litellm.proxy.utils import _monitor_spend_logs_queue @@ -8315,42 +8332,43 @@ async def initialize_scheduled_background_jobs( await cls._initialize_spend_tracking_background_jobs(scheduler=scheduler) ### SPEND LOG CLEANUP ### - if general_settings.get("maximum_spend_logs_retention_period") is not None: - spend_log_cleanup: Final = SpendLogCleanup() - cleanup_cron: Final = general_settings.get("maximum_spend_logs_cleanup_cron") + ## Always scheduled: spend-log pruning inside it still requires the retention + ## setting, but the auto-router rollup is collected past its read horizon regardless + spend_log_cleanup: Final = SpendLogCleanup() + cleanup_cron: Final = general_settings.get("maximum_spend_logs_cleanup_cron") - if cleanup_cron: - from apscheduler.triggers.cron import CronTrigger + if cleanup_cron: + from apscheduler.triggers.cron import CronTrigger - try: - cron_trigger: Final = CronTrigger.from_crontab(cleanup_cron) - scheduler.add_job( - spend_log_cleanup.cleanup_old_spend_logs, - cron_trigger, - args=[prisma_client], - id="spend_log_cleanup_job", - replace_existing=True, - misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, - ) - verbose_proxy_logger.info("Spend log cleanup scheduled with cron: %s", cleanup_cron) - except ValueError: - verbose_proxy_logger.error("Invalid maximum_spend_logs_cleanup_cron value: %s", cleanup_cron) - else: - # Interval-based scheduling (existing behavior) - retention_interval: Final = general_settings.get("maximum_spend_logs_retention_interval", "1d") - try: - interval_seconds: Final = duration_in_seconds(retention_interval) - scheduler.add_job( - spend_log_cleanup.cleanup_old_spend_logs, - "interval", - seconds=interval_seconds + random.randint(0, 60), - args=[prisma_client], - id="spend_log_cleanup_job", - replace_existing=True, - misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, - ) - except ValueError: - verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value") + try: + cron_trigger: Final = CronTrigger.from_crontab(cleanup_cron) + scheduler.add_job( + spend_log_cleanup.cleanup_old_spend_logs, + cron_trigger, + args=[prisma_client], + id="spend_log_cleanup_job", + replace_existing=True, + misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, + ) + verbose_proxy_logger.info("Spend log cleanup scheduled with cron: %s", cleanup_cron) + except ValueError: + verbose_proxy_logger.error("Invalid maximum_spend_logs_cleanup_cron value: %s", cleanup_cron) + else: + # Interval-based scheduling (existing behavior) + retention_interval: Final = general_settings.get("maximum_spend_logs_retention_interval", "1d") + try: + interval_seconds: Final = duration_in_seconds(retention_interval) + scheduler.add_job( + spend_log_cleanup.cleanup_old_spend_logs, + "interval", + seconds=interval_seconds + random.randint(0, 60), + args=[prisma_client], + id="spend_log_cleanup_job", + replace_existing=True, + misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME, + ) + except ValueError: + verbose_proxy_logger.error("Invalid maximum_spend_logs_retention_interval value") ### CHECK BATCH COST ### if llm_router is not None and PROXY_BATCH_POLLING_ENABLED: try: @@ -16375,6 +16393,50 @@ async def get_adaptive_router_state( return {"routers": snapshots} +def _auto_router_error(message: str) -> dict[str, str]: + return {"error": message} # mutable-ok: HTTPException serializes its detail from a real dict + + +@router.get( + "/auto_router/benchmarks", + tags=["auto_router"], # mutable-ok: FastAPI types tags as List[str] + response_model=AutoRouterBenchmarksResponse, +) +async def get_auto_router_benchmarks( + start_date: date, + end_date: date, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], + model_group: str | None = None, +): + """Savings, session shape and prompt-cache behaviour for every auto-router. + + Admin-only. Reads the per-session rollup only; no per-request table is scanned. + `start_date` and `end_date` are inclusive calendar dates, clamped to the most recent + 30 days. Pass `model_group` to scope every figure to one auto-router. + """ + if not _user_has_admin_view(user_api_key_dict): + raise HTTPException( + status_code=403, + detail=_auto_router_error(CommonProxyErrors.not_allowed_access.value), + ) + if end_date < start_date: + raise HTTPException( + status_code=400, + detail=_auto_router_error("end_date must not be earlier than start_date."), + ) + if prisma_client is None: + raise HTTPException( + status_code=500, + detail=_auto_router_error(CommonProxyErrors.db_not_connected_error.value), + ) + return await fetch_benchmarks( + prisma_client=prisma_client, + start_date=start_date, + end_date=end_date, + model_group=model_group, + ) + + @router.get("/routes", dependencies=[Depends(user_api_key_auth)]) async def get_routes(): """ diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index 17339541fd96..ae4bcdc9abe1 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -1393,6 +1393,45 @@ model LiteLLM_AdaptiveRouterSession { @@index([last_activity_at], map: "idx_adaptive_router_session_activity") } +model LiteLLM_AutoRouterSession { + api_key String + session_id String + model_group String + router_kind String + baseline_model String? + + turns Int @default(0) + turns_with_usage Int @default(0) + total_tokens BigInt @default(0) + spend Float @default(0) + baseline_spend Float @default(0) + + first_visit_turns Int @default(0) + first_visit_hits Int @default(0) + warm_turns Int @default(0) + warm_hits Int @default(0) + expired_turns Int @default(0) + expired_hits Int @default(0) + unordered_turns Int @default(0) + unordered_hits Int @default(0) + unknown_ttl_turns Int @default(0) + unknown_ttl_hits Int @default(0) + + cache_5m_turns Int @default(0) + cache_1h_turns Int @default(0) + cache_ttl_unknown_turns Int @default(0) + + tiers Json @default("{}") + + first_turn_at DateTime + last_turn_at DateTime + updated_at DateTime @default(now()) @updatedAt + + @@id([api_key, session_id, model_group]) + @@index([first_turn_at], map: "idx_auto_router_session_started") + @@index([last_turn_at], map: "idx_auto_router_session_activity") +} + // --------------------------------------------------------------------------- // Workflow Run Tracking // diff --git a/litellm/proxy/spend_tracking/auto_router_benchmarks.py b/litellm/proxy/spend_tracking/auto_router_benchmarks.py new file mode 100644 index 000000000000..d8f478756e03 --- /dev/null +++ b/litellm/proxy/spend_tracking/auto_router_benchmarks.py @@ -0,0 +1,289 @@ +"""Read path for the auto-router benchmarks dashboard. + +Sums pre-folded rows from ``LiteLLM_AutoRouterSession`` and nothing else. ``summarize`` +produces both the per-router view and the totals, since averaging per-router percentages is +wrong. Every miss has one of four causes on one denominator: cold, prefix changed, aged out, or a turn whose cache state could not be established. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import astuple, dataclass, fields +from datetime import date, datetime, time, timedelta, timezone +from decimal import Decimal +from functools import reduce +from typing import TYPE_CHECKING, Final + +from pydantic import BaseModel + +if TYPE_CHECKING: + from litellm.proxy.utils import PrismaClient + +MAX_WINDOW_DAYS: Final = 30 +AUTO_ROUTER_SESSION_RETENTION_DAYS: Final = MAX_WINDOW_DAYS + 1 + + +@dataclass(frozen=True, slots=True) +class _Counters: + sessions: int = 0 + turns: int = 0 + turns_with_usage: int = 0 + total_tokens: int = 0 + total_session_seconds: float = 0.0 + spend: float = 0.0 + baseline_spend: float = 0.0 + first_visit_turns: int = 0 + first_visit_hits: int = 0 + warm_turns: int = 0 + warm_hits: int = 0 + expired_turns: int = 0 + expired_hits: int = 0 + unordered_turns: int = 0 + unordered_hits: int = 0 + unknown_ttl_turns: int = 0 + unknown_ttl_hits: int = 0 + cache_5m_turns: int = 0 + cache_1h_turns: int = 0 + cache_ttl_unknown_turns: int = 0 + + +class AutoRouterCacheBenchmark(BaseModel): + turns: int + hits: int + misses: int + hit_rate_pct: float + coverage_pct: float + first_visit_turns: int + first_visit_hits: int + first_visit_hit_rate_pct: float + warm_turns: int + warm_hits: int + warm_hit_rate_pct: float + expired_turns: int + expired_hits: int + expired_hit_rate_pct: float + unordered_turns: int + unordered_hits: int + unknown_ttl_turns: int + unknown_ttl_hits: int + five_minute_cache_turns: int + one_hour_cache_turns: int + unknown_cache_ttl_turns: int + cold_misses: int + prefix_change_misses: int + expired_misses: int + unattributed_misses: int + cold_miss_pct: float + prefix_change_miss_pct: float + expired_miss_pct: float + unattributed_miss_pct: float + + +class AutoRouterBenchmark(BaseModel): + sessions: int + turns: int + total_tokens: int + spend: float + baseline_spend: float + savings: float + savings_pct: float + saved_per_session: float + avg_turns_per_session: float + avg_session_seconds: float + avg_tokens_per_session: float + cache: AutoRouterCacheBenchmark | None + + +class AutoRouterGroupBenchmark(BaseModel): + model_group: str + router_kind: str + baseline_model: str | None + benchmark: AutoRouterBenchmark + + +class AutoRouterBenchmarksResponse(BaseModel): + start_date: date + end_date: date + routers_in_scope: int + totals: AutoRouterBenchmark + groups: tuple[AutoRouterGroupBenchmark, ...] + + +_DERIVED_COLUMNS: Final = ("sessions", "total_session_seconds") +_SUM_COLUMNS: Final = tuple(field.name for field in fields(_Counters) if field.name not in _DERIVED_COLUMNS) + +_AGGREGATE_SQL: Final = f""" +SELECT + model_group, + MAX(router_kind) AS router_kind, + MAX(baseline_model) AS baseline_model, + COUNT(*) AS sessions, + COALESCE(SUM(EXTRACT(EPOCH FROM (last_turn_at - first_turn_at))), 0) AS total_session_seconds, + {", ".join(f"COALESCE(SUM({column}), 0) AS {column}" for column in _SUM_COLUMNS)} +FROM "LiteLLM_AutoRouterSession" +WHERE first_turn_at >= $1::timestamptz AT TIME ZONE \'UTC\' + AND first_turn_at < $2::timestamptz AT TIME ZONE \'UTC\' + AND ($3::text IS NULL OR model_group = $3::text) +GROUP BY model_group +ORDER BY SUM(spend) DESC +""" + + +def _pct(numerator: float, denominator: float) -> float: + return 100.0 * numerator / denominator if denominator else 0.0 + + +def _per(numerator: float, denominator: float) -> float: + return numerator / denominator if denominator else 0.0 + + +def _number(value: object) -> float: + """Postgres aggregates surface through the driver as int, float, Decimal or a numeric + string depending on the column type (SUM over BIGINT and EXTRACT both yield NUMERIC).""" + if isinstance(value, (int, float, Decimal)): + return float(value) + if not isinstance(value, str): + return 0.0 + try: + return float(value) + except ValueError: + return 0.0 + + +def _counters_from_row(row: Mapping[str, object]) -> _Counters: + return _Counters(*(_number(row.get(field.name)) for field in fields(_Counters))) + + +def _combine(left: _Counters, right: _Counters) -> _Counters: + return _Counters(*(a + b for a, b in zip(astuple(left), astuple(right)))) + + +def _summarize_cache(counters: _Counters) -> AutoRouterCacheBenchmark | None: + """The cache view, or ``None`` when no turn in scope reported cache behaviour.""" + if counters.turns_with_usage == 0: + return None + covered: Final = counters.turns_with_usage + hits: Final = ( + counters.first_visit_hits + + counters.warm_hits + + counters.expired_hits + + counters.unordered_hits + + counters.unknown_ttl_hits + ) + misses: Final = covered - hits + cold: Final = counters.first_visit_turns - counters.first_visit_hits + prefix_changed: Final = counters.warm_turns - counters.warm_hits + savable: Final = counters.expired_turns - counters.expired_hits + unattributed: Final = ( + counters.unordered_turns - counters.unordered_hits + counters.unknown_ttl_turns - counters.unknown_ttl_hits + ) + return AutoRouterCacheBenchmark( + turns=covered, + hits=hits, + misses=misses, + hit_rate_pct=_pct(hits, covered), + coverage_pct=_pct(counters.turns_with_usage, counters.turns), + first_visit_turns=counters.first_visit_turns, + first_visit_hits=counters.first_visit_hits, + first_visit_hit_rate_pct=_pct(counters.first_visit_hits, counters.first_visit_turns), + warm_turns=counters.warm_turns, + warm_hits=counters.warm_hits, + warm_hit_rate_pct=_pct(counters.warm_hits, counters.warm_turns), + expired_turns=counters.expired_turns, + expired_hits=counters.expired_hits, + expired_hit_rate_pct=_pct(counters.expired_hits, counters.expired_turns), + unordered_turns=counters.unordered_turns, + unordered_hits=counters.unordered_hits, + unknown_ttl_turns=counters.unknown_ttl_turns, + unknown_ttl_hits=counters.unknown_ttl_hits, + five_minute_cache_turns=counters.cache_5m_turns, + one_hour_cache_turns=counters.cache_1h_turns, + unknown_cache_ttl_turns=counters.cache_ttl_unknown_turns, + cold_misses=cold, + prefix_change_misses=prefix_changed, + expired_misses=savable, + unattributed_misses=unattributed, + cold_miss_pct=_pct(cold, misses), + prefix_change_miss_pct=_pct(prefix_changed, misses), + expired_miss_pct=_pct(savable, misses), + unattributed_miss_pct=_pct(unattributed, misses), + ) + + +def summarize(counters: _Counters) -> AutoRouterBenchmark: + """One benchmark from raw counters, used for a single router and for the totals alike.""" + savings: Final = counters.baseline_spend - counters.spend + return AutoRouterBenchmark( + sessions=counters.sessions, + turns=counters.turns, + total_tokens=counters.total_tokens, + spend=counters.spend, + baseline_spend=counters.baseline_spend, + savings=savings, + savings_pct=_pct(savings, counters.baseline_spend), + saved_per_session=_per(savings, counters.sessions), + avg_turns_per_session=_per(counters.turns, counters.sessions), + avg_session_seconds=_per(counters.total_session_seconds, counters.sessions), + avg_tokens_per_session=_per(counters.total_tokens, counters.sessions), + cache=_summarize_cache(counters), + ) + + +def clamp_window(start_date: date, end_date: date, today: date) -> tuple[datetime, datetime]: + """The half-open UTC interval to read, clamped into the most recent ``MAX_WINDOW_DAYS`` + ending ``today``; ``end_date`` is inclusive to the caller, so the upper bound is the + start of the following day. This recency clamp is what makes garbage collecting rows + past ``AUTO_ROUTER_SESSION_RETENTION_DAYS`` safe: a pruned row is one no window can + read. A window entirely before the horizon degenerates to an empty interval.""" + span_end: Final = min(end_date, today) + span_start: Final = max(start_date, today - timedelta(days=MAX_WINDOW_DAYS - 1)) + return ( + datetime.combine(span_start, time.min, tzinfo=timezone.utc), + datetime.combine(span_end + timedelta(days=1), time.min, tzinfo=timezone.utc), + ) + + +def build_response( + rows: Sequence[Mapping[str, object]], + start_date: date, + end_date: date, +) -> AutoRouterBenchmarksResponse: + per_group: Final = tuple((row, _counters_from_row(row)) for row in rows) + totals: Final = reduce(_combine, (counters for _, counters in per_group), _Counters()) + return AutoRouterBenchmarksResponse( + start_date=start_date, + end_date=end_date, + routers_in_scope=len(per_group), + totals=summarize(totals), + groups=tuple( + AutoRouterGroupBenchmark( + model_group=str(row.get("model_group") or ""), + router_kind=str(row.get("router_kind") or ""), + baseline_model=row.get("baseline_model") if isinstance(row.get("baseline_model"), str) else None, + benchmark=summarize(counters), + ) + for row, counters in per_group + ), + ) + + +async def fetch_benchmarks( + prisma_client: PrismaClient, + start_date: date, + end_date: date, + model_group: str | None = None, +) -> AutoRouterBenchmarksResponse: + """Benchmarks for the window actually read; sessions are attributed to the window they + started in, and the response echoes the clamped dates rather than the requested ones.""" + window_start, window_end = clamp_window(start_date, end_date, today=datetime.now(timezone.utc).date()) + rows: Final = await prisma_client.db.query_raw( + _AGGREGATE_SQL, + window_start.isoformat(), + window_end.isoformat(), + model_group, + ) + return build_response( + rows=rows, + start_date=window_start.date(), + end_date=window_end.date() - timedelta(days=1), + ) diff --git a/litellm/proxy/spend_tracking/auto_router_sessions.py b/litellm/proxy/spend_tracking/auto_router_sessions.py new file mode 100644 index 000000000000..dd26767c695f --- /dev/null +++ b/litellm/proxy/spend_tracking/auto_router_sessions.py @@ -0,0 +1,285 @@ +"""Per-(api_key, session, auto-router) rollup behind the auto-router benchmarks dashboard. + +A turn enters the cache view only when it touched the cache, so a model with caching off +contributes nothing to the buckets or the hit rate. ``tiers`` holds +``{model: [refreshed_at, ttl, prefix_tokens]}``, so a turn's bucket is +a question about one model's own record and the upsert answers it against the row it is +already writing: absent or nothing live is a first visit, an earlier start is unordered, and +otherwise warm or expired on the idle gap against the TTL the entry was written with. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Final + +import litellm +from litellm._logging import verbose_proxy_logger +from litellm.proxy.db.db_transaction_queue.base_update_queue import BaseUpdateQueue + +if TYPE_CHECKING: + from litellm.proxy.utils import PrismaClient + +CACHE_TTL_5M_SECONDS: Final = 300.0 +CACHE_TTL_1H_SECONDS: Final = 3600.0 + + +@dataclass(frozen=True, slots=True) +class TurnFacts: + """One auto-routed turn, priced and reduced to what the rollup needs.""" + + api_key: str + session_id: str + model_group: str + router_kind: str + baseline_model: str | None + model: str + started_at: float + total_tokens: int + spend: float + baseline_spend: float + cache_hit: bool + cache_creation_tokens: int + cached_prefix_tokens: int + ttl_seconds: float | None + + +PARAM_NAMES: Final = ( + "api_key", + "session_id", + "model_group", + "router_kind", + "baseline_model", + "model", + "started_at", + "total_tokens", + "spend", + "baseline_spend", + "cache_hit", + "written_tokens", + "prefix_tokens", + "ttl", +) + +( + _API_KEY, + _SESSION_ID, + _MODEL_GROUP, + _ROUTER_KIND, + _BASELINE_MODEL, + _MODEL, + _STARTED_AT, + _TOTAL_TOKENS, + _SPEND, + _BASELINE_SPEND, + _CACHE_HIT, + _WRITTEN_TOKENS, + _PREFIX_TOKENS, + _TTL, +) = (f"${position}" for position in range(1, len(PARAM_NAMES) + 1)) + + +def bind(turn: TurnFacts) -> tuple[object, ...]: + """This turn's values, in ``PARAM_NAMES`` order.""" + return ( + turn.api_key, + turn.session_id, + turn.model_group, + turn.router_kind, + turn.baseline_model, + turn.model, + turn.started_at, + turn.total_tokens, + turn.spend, + turn.baseline_spend, + int(turn.cache_hit), + turn.cache_creation_tokens, + turn.cached_prefix_tokens, + turn.ttl_seconds, + ) + + +_SEEN: Final = f"t.tiers ? {_MODEL}" +_CACHED_AT: Final = f"(t.tiers -> {_MODEL} ->> 0)::float8" +_CACHED_TTL: Final = f"(t.tiers -> {_MODEL} ->> 1)::float8" +_CACHED_TOKENS: Final = f"(t.tiers -> {_MODEL} ->> 2)::float8" +_IDLE: Final = f"({_STARTED_AT}::float8 - {_CACHED_AT})" + +_COVERED: Final = f"{_PREFIX_TOKENS}::float8 > 0" +_LIVE: Final = f"{_SEEN} AND {_CACHED_TOKENS} > 0" +_UNORDERED: Final = f"{_COVERED} AND {_LIVE} AND {_IDLE} < 0" +_KNOWN_CACHED_TTL: Final = f"{_CACHED_TTL} IN ({CACHE_TTL_5M_SECONDS}, {CACHE_TTL_1H_SECONDS})" +_WARM: Final = f"{_COVERED} AND {_LIVE} AND {_KNOWN_CACHED_TTL} AND {_IDLE} >= 0 AND {_IDLE} <= {_CACHED_TTL}" +_EXPIRED: Final = f"{_COVERED} AND {_LIVE} AND {_KNOWN_CACHED_TTL} AND {_IDLE} > {_CACHED_TTL}" +_UNKNOWN_TTL: Final = f"{_COVERED} AND {_LIVE} AND {_IDLE} >= 0 AND NOT COALESCE({_KNOWN_CACHED_TTL}, FALSE)" +_REFRESHED: Final = f"NOT ({_SEEN}) OR ({_COVERED} AND (NOT ({_LIVE}) OR {_IDLE} >= 0))" +_REWROTE: Final = f"{_WRITTEN_TOKENS}::float8 > 0 AND (NOT ({_LIVE}) OR {_IDLE} >= 0 OR {_CACHED_TTL} IS NULL)" +_NEXT_TTL: Final = f"CASE WHEN {_REWROTE} THEN {_TTL}::float8 ELSE {_CACHED_TTL} END" +_NEXT_TTL_IS_5M: Final = f"{_NEXT_TTL} = {CACHE_TTL_5M_SECONDS}" +_NEXT_TTL_IS_1H: Final = f"{_NEXT_TTL} = {CACHE_TTL_1H_SECONDS}" + +_UPSERT_SQL: Final = f""" +INSERT INTO "LiteLLM_AutoRouterSession" AS t ( + api_key, session_id, model_group, router_kind, baseline_model, + turns, turns_with_usage, total_tokens, spend, baseline_spend, + first_visit_turns, first_visit_hits, + unknown_ttl_turns, unknown_ttl_hits, + cache_5m_turns, cache_1h_turns, cache_ttl_unknown_turns, + tiers, first_turn_at, last_turn_at, updated_at +) +VALUES ( + {_API_KEY}, {_SESSION_ID}, {_MODEL_GROUP}, {_ROUTER_KIND}, {_BASELINE_MODEL}, + 1, CASE WHEN {_COVERED} THEN 1 ELSE 0 END, {_TOTAL_TOKENS}::bigint, {_SPEND}, {_BASELINE_SPEND}, + CASE WHEN {_COVERED} THEN 1 ELSE 0 END, + CASE WHEN {_COVERED} THEN {_CACHE_HIT} ELSE 0 END, + 0, 0, + CASE WHEN {_COVERED} AND {_TTL}::float8 = {CACHE_TTL_5M_SECONDS} THEN 1 ELSE 0 END, + CASE WHEN {_COVERED} AND {_TTL}::float8 = {CACHE_TTL_1H_SECONDS} THEN 1 ELSE 0 END, + CASE WHEN {_COVERED} AND NOT COALESCE( + {_TTL}::float8 = {CACHE_TTL_5M_SECONDS} OR {_TTL}::float8 = {CACHE_TTL_1H_SECONDS}, FALSE + ) THEN 1 ELSE 0 END, + jsonb_build_object({_MODEL}, jsonb_build_array( + {_STARTED_AT}::float8, {_TTL}::float8, {_PREFIX_TOKENS}::float8 + )), + to_timestamp({_STARTED_AT}::float8) AT TIME ZONE 'UTC', + to_timestamp({_STARTED_AT}::float8) AT TIME ZONE 'UTC', + NOW() +) +ON CONFLICT (api_key, session_id, model_group) DO UPDATE SET + turns = t.turns + 1, + turns_with_usage = t.turns_with_usage + CASE WHEN {_COVERED} THEN 1 ELSE 0 END, + total_tokens = t.total_tokens + {_TOTAL_TOKENS}::bigint, + spend = t.spend + {_SPEND}, + baseline_spend = t.baseline_spend + {_BASELINE_SPEND}, + + first_visit_turns = t.first_visit_turns + CASE WHEN {_COVERED} AND NOT ({_LIVE}) THEN 1 ELSE 0 END, + first_visit_hits = t.first_visit_hits + CASE WHEN {_COVERED} AND NOT ({_LIVE}) THEN {_CACHE_HIT} ELSE 0 END, + unordered_turns = t.unordered_turns + CASE WHEN {_UNORDERED} THEN 1 ELSE 0 END, + unordered_hits = t.unordered_hits + CASE WHEN {_UNORDERED} THEN {_CACHE_HIT} ELSE 0 END, + warm_turns = t.warm_turns + CASE WHEN {_WARM} THEN 1 ELSE 0 END, + warm_hits = t.warm_hits + CASE WHEN {_WARM} THEN {_CACHE_HIT} ELSE 0 END, + expired_turns = t.expired_turns + CASE WHEN {_EXPIRED} THEN 1 ELSE 0 END, + expired_hits = t.expired_hits + CASE WHEN {_EXPIRED} THEN {_CACHE_HIT} ELSE 0 END, + unknown_ttl_turns = t.unknown_ttl_turns + CASE WHEN {_UNKNOWN_TTL} THEN 1 ELSE 0 END, + unknown_ttl_hits = t.unknown_ttl_hits + CASE WHEN {_UNKNOWN_TTL} THEN {_CACHE_HIT} ELSE 0 END, + + cache_5m_turns = t.cache_5m_turns + CASE WHEN {_COVERED} AND {_NEXT_TTL_IS_5M} THEN 1 ELSE 0 END, + cache_1h_turns = t.cache_1h_turns + CASE WHEN {_COVERED} AND {_NEXT_TTL_IS_1H} THEN 1 ELSE 0 END, + cache_ttl_unknown_turns = t.cache_ttl_unknown_turns + + CASE WHEN {_COVERED} AND NOT COALESCE({_NEXT_TTL_IS_5M} OR {_NEXT_TTL_IS_1H}, FALSE) THEN 1 ELSE 0 END, + baseline_model = COALESCE(t.baseline_model, {_BASELINE_MODEL}), + tiers = t.tiers || jsonb_build_object({_MODEL}, jsonb_build_array( + CASE WHEN {_REFRESHED} THEN {_STARTED_AT}::float8 ELSE {_CACHED_AT} END, + CASE WHEN {_REWROTE} THEN {_TTL}::float8 ELSE {_CACHED_TTL} END, + CASE WHEN {_REFRESHED} THEN {_PREFIX_TOKENS}::float8 ELSE {_CACHED_TOKENS} END + )), + first_turn_at = LEAST(t.first_turn_at, to_timestamp({_STARTED_AT}::float8) AT TIME ZONE 'UTC'), + last_turn_at = GREATEST(t.last_turn_at, to_timestamp({_STARTED_AT}::float8) AT TIME ZONE 'UTC'), + updated_at = NOW() +""" + + +class AutoRouterSessionQueue(BaseUpdateQueue): + """Stages turns in memory; a dedicated scheduler job writes an interval as one batch, + kept off the spend commit so a busy interval never delays budget enforcement. + + Sorted by session key so every pod locks rows in the same order (no cross-pod + deadlock), with a session's turns in the time order its classification depends on. + Never raises. + """ + + async def flush(self, prisma_client: PrismaClient) -> None: + staged: Final[Sequence[TurnFacts]] = await self.flush_all_updates_from_in_memory_queue() + if not staged: + return + try: + async with prisma_client.db.batch_() as batcher: + for turn in sorted(staged, key=lambda t: (t.api_key, t.session_id, t.model_group, t.started_at)): + batcher.execute_raw(_UPSERT_SQL, *bind(turn)) + except Exception as e: # noqa: BLE001 # a dashboard rollup must never fail spend tracking + verbose_proxy_logger.warning("auto_router_sessions: dropped %d turns (%s)", len(staged), e) + + +def ttl_seconds(usage_obj: Mapping[str, object] | None) -> float | None: + if usage_obj is None: + return None + details: Final = usage_obj.get("cache_creation_token_details") + if not isinstance(details, Mapping): + return None + five_minute: Final = int(details.get("ephemeral_5m_input_tokens") or 0) > 0 + one_hour: Final = int(details.get("ephemeral_1h_input_tokens") or 0) > 0 + if five_minute == one_hour: + return None + return CACHE_TTL_1H_SECONDS if one_hour else CACHE_TTL_5M_SECONDS + + +def _as_utc(moment: datetime) -> float: + return (moment.replace(tzinfo=timezone.utc) if moment.tzinfo is None else moment).timestamp() + + +def _started_at(start_time: object) -> float | None: + if isinstance(start_time, datetime): + return _as_utc(start_time) + if isinstance(start_time, str): + try: + return _as_utc(datetime.fromisoformat(start_time)) + except ValueError: + return None + return None + + +def build_turn_facts( + payload: Mapping[str, object], + metadata: Mapping[str, object], + autorouter_savings: float, + cache_read_tokens: int, + cache_creation_tokens: int, +) -> TurnFacts | None: + """One spend-log payload as a rollup turn, or ``None`` if it was not auto-routed. + + A recorded ``routing_decision`` is what says the request was auto-routed, and names the kind. + """ + decision: Final = metadata.get("routing_decision") + if not isinstance(decision, Mapping): + return None + router_kind: Final = decision.get("router_type") + api_key: Final = payload.get("api_key") + session_id: Final = payload.get("session_id") + model_group: Final = payload.get("model_group") + model: Final = payload.get("model") + if not ( + isinstance(router_kind, str) + and isinstance(api_key, str) + and isinstance(session_id, str) + and isinstance(model_group, str) + and isinstance(model, str) + and api_key + and session_id + and model_group + and model + ): + return None + started_at: Final = _started_at(payload.get("startTime")) + if started_at is None: + return None + usage_raw: Final = metadata.get("usage_object") + usage_obj: Final = usage_raw if isinstance(usage_raw, Mapping) else None + spend: Final = float(payload.get("spend") or 0.0) + return TurnFacts( + api_key=api_key, + session_id=session_id, + model_group=model_group, + router_kind=router_kind, + baseline_model=litellm.autorouter_savings_baseline_model, + model=model, + started_at=started_at, + total_tokens=int(payload.get("prompt_tokens") or 0) + int(payload.get("completion_tokens") or 0), + spend=spend, + baseline_spend=spend + autorouter_savings, + cache_hit=cache_read_tokens > 0, + cache_creation_tokens=cache_creation_tokens, + cached_prefix_tokens=max(cache_read_tokens, 0) + max(cache_creation_tokens, 0), + ttl_seconds=ttl_seconds(usage_obj), + ) diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 99e566c2da19..74a173d6b55e 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -5606,6 +5606,16 @@ async def update_daily_tag_spend( verbose_proxy_logger.error("Error updating daily tag spend: %s", e) +async def update_auto_router_sessions( + prisma_client: PrismaClient, + proxy_logging_obj: ProxyLogging, +): + """Separate scheduler job for the auto-router benchmarks rollup, kept off the + update_spend job so rollup upserts never extend the wall time of key, team and org + budget commits. The queue drops rather than blocks when full; the flush never raises.""" + await proxy_logging_obj.db_spend_update_writer.auto_router_session_queue.flush(prisma_client=prisma_client) + + async def update_spend_logs_job( prisma_client: PrismaClient, db_writer_client: AsyncHTTPHandler | None, diff --git a/schema.prisma b/schema.prisma index 17339541fd96..ae4bcdc9abe1 100644 --- a/schema.prisma +++ b/schema.prisma @@ -1393,6 +1393,45 @@ model LiteLLM_AdaptiveRouterSession { @@index([last_activity_at], map: "idx_adaptive_router_session_activity") } +model LiteLLM_AutoRouterSession { + api_key String + session_id String + model_group String + router_kind String + baseline_model String? + + turns Int @default(0) + turns_with_usage Int @default(0) + total_tokens BigInt @default(0) + spend Float @default(0) + baseline_spend Float @default(0) + + first_visit_turns Int @default(0) + first_visit_hits Int @default(0) + warm_turns Int @default(0) + warm_hits Int @default(0) + expired_turns Int @default(0) + expired_hits Int @default(0) + unordered_turns Int @default(0) + unordered_hits Int @default(0) + unknown_ttl_turns Int @default(0) + unknown_ttl_hits Int @default(0) + + cache_5m_turns Int @default(0) + cache_1h_turns Int @default(0) + cache_ttl_unknown_turns Int @default(0) + + tiers Json @default("{}") + + first_turn_at DateTime + last_turn_at DateTime + updated_at DateTime @default(now()) @updatedAt + + @@id([api_key, session_id, model_group]) + @@index([first_turn_at], map: "idx_auto_router_session_started") + @@index([last_turn_at], map: "idx_auto_router_session_activity") +} + // --------------------------------------------------------------------------- // Workflow Run Tracking // diff --git a/tests/proxy_behavior/spend/__init__.py b/tests/proxy_behavior/spend/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/proxy_behavior/spend/conftest.py b/tests/proxy_behavior/spend/conftest.py new file mode 100644 index 000000000000..60e3659794c2 --- /dev/null +++ b/tests/proxy_behavior/spend/conftest.py @@ -0,0 +1,36 @@ +"""A real Postgres connection for the rollup's SQL. + +The auto-router rollup classifies each turn inside its upsert, against the row's own +stored state, so the classification only exists when a real database evaluates it. +""" + +from dataclasses import dataclass + +import pytest_asyncio +from prisma import Prisma + + +@dataclass(frozen=True) +class PrismaClientShim: + """What the rollup writer needs from litellm's PrismaClient: a connected `db`.""" + + db: Prisma + + +@pytest_asyncio.fixture(scope="session", loop_scope="session") +async def prisma_db(): + db = Prisma() + await db.connect() + try: + yield db + finally: + await db.disconnect() + + +@pytest_asyncio.fixture(loop_scope="session") +async def rollup_client(prisma_db): + await prisma_db.execute_raw('DELETE FROM "LiteLLM_AutoRouterSession"') + try: + yield PrismaClientShim(db=prisma_db) + finally: + await prisma_db.execute_raw('DELETE FROM "LiteLLM_AutoRouterSession"') diff --git a/tests/proxy_behavior/spend/test_auto_router_session_rollup.py b/tests/proxy_behavior/spend/test_auto_router_session_rollup.py new file mode 100644 index 000000000000..d139a53494e2 --- /dev/null +++ b/tests/proxy_behavior/spend/test_auto_router_session_rollup.py @@ -0,0 +1,292 @@ +"""How the auto-router rollup buckets a turn, evaluated by a real Postgres. + +The upsert classifies each turn against the session's own cache record, so these assertions +are about SQL. They cover what the dashboard depends on: the three buckets partition every +turn, a tier that aged out is told from one that is still warm, the warming estimate is +priced on the prefix that was actually cached, and one caller cannot write into another's +rollup by reusing a session id. +""" + +import datetime as dt +from dataclasses import replace + +import pytest + +from litellm.proxy.spend_tracking.auto_router_sessions import AutoRouterSessionQueue, TurnFacts + +pytestmark = pytest.mark.asyncio(loop_scope="session") + +HAIKU = "anthropic/claude-haiku-4-5" +OPUS = "anthropic/claude-opus-4-8" +T0 = dt.datetime(2026, 8, 3, 12, 0, tzinfo=dt.timezone.utc).timestamp() + +TURN = TurnFacts( + api_key="key-a", + session_id="sess-1", + model_group="claude-auto", + router_kind="complexity", + baseline_model=OPUS, + model=HAIKU, + started_at=T0, + total_tokens=1000, + spend=0.01, + baseline_spend=0.05, + cache_hit=False, + cache_creation_tokens=2000, + cached_prefix_tokens=2000, + ttl_seconds=300.0, +) + + +async def _flush(client, turns) -> None: + queue = AutoRouterSessionQueue() + for turn in turns: + await queue.update_queue.put(turn) + await queue.flush(prisma_client=client) + + +async def _rows(db) -> list[dict]: + return await db.query_raw('SELECT * FROM "LiteLLM_AutoRouterSession" ORDER BY api_key, session_id') + + +def _epoch_of(stored: str) -> float: + parsed = dt.datetime.fromisoformat(stored) + return (parsed.replace(tzinfo=dt.timezone.utc) if parsed.tzinfo is None else parsed).timestamp() + + +async def test_a_sessions_opening_turn_on_a_tier_is_a_first_visit(rollup_client, prisma_db): + await _flush(rollup_client, [TURN]) + row = (await _rows(prisma_db))[0] + assert (row["turns"], row["first_visit_turns"], row["warm_turns"], row["expired_turns"]) == (1, 1, 0, 0) + assert row["tiers"] == {HAIKU: [T0, 300.0, 2000]} + + +async def test_a_second_tier_is_its_own_first_visit(rollup_client, prisma_db): + await _flush(rollup_client, [TURN, replace(TURN, model=OPUS, started_at=T0 + 10)]) + row = (await _rows(prisma_db))[0] + assert (row["turns"], row["first_visit_turns"]) == (2, 2) + assert set(row["tiers"]) == {HAIKU, OPUS} + + +async def test_a_tier_used_again_inside_its_ttl_is_warm(rollup_client, prisma_db): + await _flush(rollup_client, [TURN, replace(TURN, started_at=T0 + 60, cache_hit=True)]) + row = (await _rows(prisma_db))[0] + assert (row["warm_turns"], row["warm_hits"], row["expired_turns"]) == (1, 1, 0) + + +async def test_a_tier_used_again_past_its_ttl_has_expired(rollup_client, prisma_db): + await _flush(rollup_client, [TURN, replace(TURN, started_at=T0 + 900)]) + row = (await _rows(prisma_db))[0] + assert (row["warm_turns"], row["expired_turns"], row["expired_hits"]) == (0, 1, 0) + + +async def test_expiry_measures_against_the_ttl_the_cache_was_written_with(rollup_client, prisma_db): + """A one-hour entry is still warm at 30 minutes, even though the turn reading it + reports no one-hour evidence of its own and so carries the five minute default.""" + await _flush( + rollup_client, + [ + replace(TURN, ttl_seconds=3600.0, cache_creation_tokens=2000), + replace(TURN, started_at=T0 + 1800, ttl_seconds=300.0, cache_hit=True, cache_creation_tokens=0), + ], + ) + row = (await _rows(prisma_db))[0] + assert (row["warm_turns"], row["expired_turns"]) == (1, 0) + + +async def test_the_three_buckets_partition_every_turn(rollup_client, prisma_db): + await _flush( + rollup_client, + [ + TURN, + replace(TURN, started_at=T0 + 10), + replace(TURN, model=OPUS, started_at=T0 + 20), + replace(TURN, started_at=T0 + 5000), + ], + ) + row = (await _rows(prisma_db))[0] + assert ( + row["first_visit_turns"] + row["warm_turns"] + row["expired_turns"] + row["unknown_ttl_turns"] + == row["turns"] + == 4 + ) + + +async def test_a_turn_that_only_read_the_cache_leaves_the_written_terms_alone(rollup_client, prisma_db): + await _flush( + rollup_client, + [ + replace(TURN, ttl_seconds=3600.0, cache_creation_tokens=5000, cached_prefix_tokens=5000), + replace(TURN, started_at=T0 + 10, cache_hit=True, cache_creation_tokens=0, cached_prefix_tokens=5000), + ], + ) + row = (await _rows(prisma_db))[0] + assert row["tiers"][HAIKU] == [T0 + 10, 3600.0, 5000] + assert (row["cache_5m_turns"], row["cache_1h_turns"], row["cache_ttl_unknown_turns"]) == (0, 2, 0) + + +async def test_counters_accumulate_across_flushes(rollup_client, prisma_db): + await _flush(rollup_client, [TURN]) + await _flush(rollup_client, [replace(TURN, started_at=T0 + 10)]) + row = (await _rows(prisma_db))[0] + assert row["turns"] == 2 + assert row["spend"] == pytest.approx(0.02) + assert row["baseline_spend"] == pytest.approx(0.10) + assert row["total_tokens"] == 2000 + assert _epoch_of(row["first_turn_at"]) == pytest.approx(T0, abs=0.001) + assert _epoch_of(row["last_turn_at"]) == pytest.approx(T0 + 10, abs=0.001) + + +async def test_a_late_turn_cannot_rewind_the_session(rollup_client, prisma_db): + await _flush(rollup_client, [replace(TURN, started_at=T0 + 600)]) + await _flush(rollup_client, [replace(TURN, started_at=T0)]) + row = (await _rows(prisma_db))[0] + assert row["turns"] == 2 + assert row["tiers"][HAIKU][0] == T0 + 600 + assert _epoch_of(row["last_turn_at"]) == pytest.approx(T0 + 600, abs=0.001) + assert _epoch_of(row["first_turn_at"]) == pytest.approx(T0, abs=0.001) + + +async def test_two_callers_reusing_one_session_id_keep_separate_rollups(rollup_client, prisma_db): + await _flush( + rollup_client, + [ + replace(TURN, api_key="key-a", model=HAIKU), + replace(TURN, api_key="key-b", model=OPUS, started_at=T0 + 10), + ], + ) + rows = await _rows(prisma_db) + assert [row["api_key"] for row in rows] == ["key-a", "key-b"] + assert all(row["turns"] == 1 and row["first_visit_turns"] == 1 for row in rows) + + +async def test_a_turn_arriving_before_an_already_recorded_one_is_not_called_warm(rollup_client, prisma_db): + """Its cache state at its own time is unknowable, so it abstains rather than being + guessed at; a negative idle gap is not evidence of warmth.""" + await _flush(rollup_client, [replace(TURN, started_at=T0 + 600)]) + await _flush(rollup_client, [replace(TURN, started_at=T0)]) + row = (await _rows(prisma_db))[0] + assert row["turns"] == 2 + assert row["unordered_turns"] == 1 + assert row["warm_turns"] == 0 + assert row["expired_turns"] == 0 + assert row["first_visit_turns"] == 1 + assert row["tiers"][HAIKU][0] == T0 + 600 + + +async def test_every_turn_lands_in_exactly_one_bucket(rollup_client, prisma_db): + await _flush( + rollup_client, + [ + TURN, + replace(TURN, started_at=T0 + 10), + replace(TURN, model=OPUS, started_at=T0 + 20), + replace(TURN, started_at=T0 + 5000), + ], + ) + await _flush(rollup_client, [replace(TURN, started_at=T0 + 5)]) + row = (await _rows(prisma_db))[0] + buckets = ( + row["first_visit_turns"] + + row["warm_turns"] + + row["expired_turns"] + + row["unordered_turns"] + + row["unknown_ttl_turns"] + ) + assert buckets == row["turns"] == 5 + assert row["unordered_turns"] == 1 + + +async def test_a_first_visit_to_a_new_tier_does_not_invent_a_ttl(rollup_client, prisma_db): + await _flush( + rollup_client, + [ + TURN, + replace( + TURN, + model=OPUS, + started_at=T0 + 10, + ttl_seconds=None, + cache_creation_tokens=0, + cached_prefix_tokens=0, + ), + ], + ) + row = (await _rows(prisma_db))[0] + assert row["tiers"][OPUS] == [T0 + 10, None, 0] + + +async def test_a_cache_read_without_a_recorded_write_keeps_ttl_unknown(rollup_client, prisma_db): + await _flush( + rollup_client, + [ + replace(TURN, cache_creation_tokens=0, cached_prefix_tokens=2000, cache_hit=True, ttl_seconds=None), + replace( + TURN, + started_at=T0 + 60, + cache_creation_tokens=0, + cached_prefix_tokens=2000, + cache_hit=True, + ttl_seconds=None, + ), + ], + ) + row = (await _rows(prisma_db))[0] + assert row["tiers"][HAIKU] == [T0 + 60, None, 2000] + assert (row["first_visit_turns"], row["unknown_ttl_turns"], row["unknown_ttl_hits"]) == (1, 1, 1) + assert (row["cache_5m_turns"], row["cache_1h_turns"], row["cache_ttl_unknown_turns"]) == (0, 0, 2) + + +async def test_a_late_cache_write_revives_a_dead_tier(rollup_client, prisma_db): + await _flush( + rollup_client, + [replace(TURN, started_at=T0 + 600, cache_creation_tokens=0, cached_prefix_tokens=0, ttl_seconds=None)], + ) + await _flush(rollup_client, [replace(TURN, ttl_seconds=3600.0)]) + await _flush( + rollup_client, + [replace(TURN, started_at=T0 + 1200, cache_creation_tokens=0, cache_hit=True, ttl_seconds=None)], + ) + row = (await _rows(prisma_db))[0] + assert row["tiers"][HAIKU] == [T0 + 1200, 3600.0, 2000] + assert (row["first_visit_turns"], row["warm_turns"], row["expired_turns"]) == (1, 1, 0) + assert (row["cache_5m_turns"], row["cache_1h_turns"], row["cache_ttl_unknown_turns"]) == (0, 2, 0) + + +async def test_a_growing_conversation_records_the_whole_live_prefix(rollup_client, prisma_db): + """A warm turn on a growing prompt writes only the new segment, so recording + cache_creation_tokens alone would shrink the prefix and under-price later replays. + The live prefix is what was read plus what was written.""" + await _flush( + rollup_client, + [ + replace(TURN, cache_creation_tokens=2000, cached_prefix_tokens=2000), + replace(TURN, started_at=T0 + 60, cache_hit=True, cache_creation_tokens=500, cached_prefix_tokens=2500), + ], + ) + row = (await _rows(prisma_db))[0] + assert row["warm_turns"] == 1 + assert row["tiers"][HAIKU] == [T0 + 60, 300.0, 2500] + + +async def test_a_turn_that_touched_no_cache_is_left_out_of_the_cache_view(rollup_client, prisma_db): + """A model with caching off would otherwise read as a wall of first visits and drag the + hit rate down; it counts as a turn and nothing else.""" + await _flush( + rollup_client, + [ + replace(TURN, cache_creation_tokens=0, cached_prefix_tokens=0), + replace(TURN, started_at=T0 + 10, cache_creation_tokens=0, cached_prefix_tokens=0), + ], + ) + row = (await _rows(prisma_db))[0] + assert row["turns"] == 2 + assert row["turns_with_usage"] == 0 + assert ( + row["first_visit_turns"] + + row["warm_turns"] + + row["expired_turns"] + + row["unordered_turns"] + + row["unknown_ttl_turns"] + == 0 + ) diff --git a/tests/test_litellm/proxy/spend_tracking/test_auto_router_benchmarks.py b/tests/test_litellm/proxy/spend_tracking/test_auto_router_benchmarks.py new file mode 100644 index 000000000000..725b5483b266 --- /dev/null +++ b/tests/test_litellm/proxy/spend_tracking/test_auto_router_benchmarks.py @@ -0,0 +1,305 @@ +"""Read-path derivations for the auto-router benchmarks dashboard. + +Every figure the dashboard shows is a pure function of the rollup counters, so all of it +is exercised here without a database. +""" + +import datetime as dt +from dataclasses import fields +from decimal import Decimal + +import pytest + +from litellm.proxy.spend_tracking.auto_router_benchmarks import ( + MAX_WINDOW_DAYS, + _Counters, + build_response, + clamp_window, + summarize, +) + + +def _row(model_group: str, **overrides: object) -> dict[str, object]: + """Every counter at zero unless the case under test says otherwise.""" + return { + "model_group": model_group, + "router_kind": "complexity", + "baseline_model": "anthropic/claude-opus-4-8", + **{field.name: 0 for field in fields(_Counters)}, + "sessions": 1, + **overrides, + } + + +class TestSummarize: + def test_savings_are_the_difference_between_the_two_arms(self): + result = summarize(_Counters(sessions=4, turns=40, spend=364.59, baseline_spend=414.63)) + assert result.savings == pytest.approx(50.04) + assert result.savings_pct == pytest.approx(100 * 50.04 / 414.63) + assert result.saved_per_session == pytest.approx(50.04 / 4) + assert result.avg_turns_per_session == pytest.approx(10.0) + + def test_a_cache_thrashing_router_reports_a_signed_loss(self): + result = summarize(_Counters(sessions=1, turns=1, spend=5.0, baseline_spend=4.0)) + assert result.savings == pytest.approx(-1.0) + assert result.savings_pct < 0 + + def test_session_shape_averages_over_sessions_not_turns(self): + result = summarize(_Counters(sessions=2, turns=64, total_tokens=10_000, total_session_seconds=7200.0)) + assert result.avg_turns_per_session == pytest.approx(32.0) + assert result.avg_session_seconds == pytest.approx(3600.0) + assert result.avg_tokens_per_session == pytest.approx(5000.0) + + def test_an_empty_window_divides_by_nothing(self): + result = summarize(_Counters()) + assert result.sessions == 0 + assert result.savings_pct == 0.0 + assert result.saved_per_session == 0.0 + assert result.avg_turns_per_session == 0.0 + assert result.cache is None + + +class TestCacheView: + def test_hit_rate_is_weighted_by_turn_count_not_averaged_across_buckets(self): + cache = summarize( + _Counters( + sessions=1, + turns=3145, + turns_with_usage=3145, + warm_turns=2560, + warm_hits=2491, + first_visit_turns=146, + first_visit_hits=15, + expired_turns=439, + expired_hits=348, + ) + ).cache + assert cache is not None + hits = 2491 + 15 + 348 + assert cache.hit_rate_pct == pytest.approx(100 * hits / 3145) + mean_of_bucket_rates = ( + cache.warm_hit_rate_pct + cache.first_visit_hit_rate_pct + cache.expired_hit_rate_pct + ) / 3 + assert cache.hit_rate_pct != pytest.approx(mean_of_bucket_rates) + assert cache.hit_rate_pct > mean_of_bucket_rates + + def test_traffic_that_never_touched_the_cache_is_left_out_of_the_hit_rate(self): + """A model with caching off would otherwise read as a wall of misses. It is absent + from the buckets, and coverage says how much of the traffic that was.""" + cache = summarize(_Counters(sessions=1, turns=100, turns_with_usage=40, warm_turns=40, warm_hits=36)).cache + assert cache is not None + assert cache.turns == 40 + assert cache.coverage_pct == pytest.approx(40.0) + assert cache.hit_rate_pct == pytest.approx(90.0) + assert cache.misses == 4 + + def test_the_three_buckets_partition_every_turn(self): + counters = _Counters( + sessions=1, turns=10, turns_with_usage=10, warm_turns=6, first_visit_turns=2, expired_turns=2 + ) + cache = summarize(counters).cache + assert cache is not None + assert cache.warm_turns + cache.first_visit_turns + cache.expired_turns == cache.turns + + def test_every_miss_has_one_cause_and_they_stack_to_the_whole(self): + """A miss is cold by design, a changed prefix, an expiry, or a turn whose cache + state could not be established.""" + cache = summarize( + _Counters( + sessions=1, + turns=26, + turns_with_usage=26, + first_visit_turns=5, + first_visit_hits=1, + warm_turns=10, + warm_hits=8, + expired_turns=5, + expired_hits=2, + unordered_turns=4, + unordered_hits=3, + unknown_ttl_turns=2, + unknown_ttl_hits=1, + ) + ).cache + assert cache is not None + assert cache.hits == 15 + assert cache.misses == 11 + assert (cache.cold_misses, cache.prefix_change_misses, cache.expired_misses) == (4, 2, 3) + assert cache.unattributed_misses == 2 + assert ( + cache.cold_misses + cache.prefix_change_misses + cache.expired_misses + cache.unattributed_misses + == cache.misses + ) + assert ( + cache.cold_miss_pct + cache.prefix_change_miss_pct + cache.expired_miss_pct + cache.unattributed_miss_pct + == pytest.approx(100.0) + ) + + def test_an_unordered_turn_still_counts_toward_the_headline_hit_rate(self): + """Its cause is unknowable, but the provider still said whether it hit, so the + rate a reader looks at stays exact and only the attribution abstains.""" + cache = summarize( + _Counters( + sessions=1, + turns=10, + turns_with_usage=10, + warm_turns=6, + warm_hits=6, + unordered_turns=4, + unordered_hits=2, + ) + ).cache + assert cache is not None + assert cache.hits == 8 + assert cache.hit_rate_pct == pytest.approx(80.0) + + def test_a_router_with_no_cache_evidence_has_no_cache_view(self): + assert summarize(_Counters(sessions=1, turns=5, turns_with_usage=0)).cache is None + + def test_ttl_distribution_preserves_mixed_and_unknown_traffic(self): + cache = summarize( + _Counters( + sessions=1, + turns=10, + turns_with_usage=10, + first_visit_turns=10, + cache_5m_turns=3, + cache_1h_turns=6, + cache_ttl_unknown_turns=1, + ) + ).cache + assert cache is not None + assert cache.five_minute_cache_turns == 3 + assert cache.one_hour_cache_turns == 6 + assert cache.unknown_cache_ttl_turns == 1 + + +class TestTotals: + def test_totals_sum_the_counters_rather_than_averaging_group_rates(self): + """A big cheap router and a small expensive one must not be weighted equally.""" + response = build_response( + rows=[ + _row( + "big", + turns=1000, + turns_with_usage=1000, + warm_turns=1000, + warm_hits=900, + first_visit_turns=0, + first_visit_hits=0, + expired_turns=0, + expired_hits=0, + ), + _row( + "small", + turns=10, + turns_with_usage=10, + warm_turns=10, + warm_hits=1, + first_visit_turns=0, + first_visit_hits=0, + expired_turns=0, + expired_hits=0, + ), + ], + start_date=dt.date(2026, 7, 5), + end_date=dt.date(2026, 8, 3), + ) + assert response.routers_in_scope == 2 + assert response.totals.cache is not None + assert response.totals.cache.hit_rate_pct == pytest.approx(100 * 901 / 1010) + group_rates = [g.benchmark.cache.hit_rate_pct for g in response.groups if g.benchmark.cache] + assert response.totals.cache.hit_rate_pct != pytest.approx(sum(group_rates) / len(group_rates)) + + def test_totals_dollars_are_the_sum_of_every_router(self): + response = build_response( + rows=[_row("a", spend=3.0, baseline_spend=5.0), _row("b", spend=1.0, baseline_spend=9.0)], + start_date=dt.date(2026, 7, 5), + end_date=dt.date(2026, 8, 3), + ) + assert response.totals.spend == pytest.approx(4.0) + assert response.totals.baseline_spend == pytest.approx(14.0) + assert response.totals.savings == pytest.approx(10.0) + assert response.totals.sessions == 2 + + def test_the_response_echoes_the_window_actually_read(self): + """A caller asking for a year is told it got a month, not handed month-sized + numbers under year-sized dates.""" + response = build_response(rows=[], start_date=dt.date(2026, 7, 5), end_date=dt.date(2026, 8, 3)) + assert (response.start_date, response.end_date) == (dt.date(2026, 7, 5), dt.date(2026, 8, 3)) + + def test_an_empty_window_still_answers_with_zeroed_totals(self): + response = build_response(rows=[], start_date=dt.date(2026, 8, 1), end_date=dt.date(2026, 8, 3)) + assert response.routers_in_scope == 0 + assert response.groups == () + assert response.totals.turns == 0 + + def test_decimal_and_string_aggregates_survive_the_driver(self): + """SUM over BIGINT and EXTRACT(EPOCH) yield NUMERIC, which the driver may hand + back as Decimal or a numeric string; neither may read as zero traffic.""" + response = build_response( + rows=[ + _row( + "a", + sessions=Decimal("2"), + turns=Decimal("64"), + total_tokens=Decimal("10000"), + total_session_seconds="7200.5", + ) + ], + start_date=dt.date(2026, 8, 1), + end_date=dt.date(2026, 8, 3), + ) + assert response.totals.turns == 64 + assert response.totals.total_tokens == 10_000 + assert response.totals.avg_tokens_per_session == pytest.approx(5000.0) + assert response.totals.avg_session_seconds == pytest.approx(3600.25) + + def test_a_malformed_aggregate_reads_as_zero_rather_than_failing_the_page(self): + response = build_response( + rows=[_row("a", total_tokens="not-a-number")], + start_date=dt.date(2026, 8, 1), + end_date=dt.date(2026, 8, 3), + ) + assert response.totals.total_tokens == 0 + + def test_group_identity_is_carried_through(self): + response = build_response( + rows=[_row("claude-auto")], start_date=dt.date(2026, 8, 1), end_date=dt.date(2026, 8, 3) + ) + assert response.groups[0].model_group == "claude-auto" + assert response.groups[0].router_kind == "complexity" + assert response.groups[0].baseline_model == "anthropic/claude-opus-4-8" + + +class TestWindow: + TODAY = dt.date(2026, 8, 3) + + def test_end_date_is_inclusive(self): + start, end = clamp_window(dt.date(2026, 8, 3), dt.date(2026, 8, 3), today=self.TODAY) + assert start == dt.datetime(2026, 8, 3, tzinfo=dt.timezone.utc) + assert end == dt.datetime(2026, 8, 4, tzinfo=dt.timezone.utc) + + def test_a_wider_request_is_clamped_to_the_cap_measured_in_dates_spanned(self): + start, end = clamp_window(dt.date(2020, 1, 1), dt.date(2026, 8, 3), today=self.TODAY) + assert (end.date() - start.date()).days == MAX_WINDOW_DAYS + assert start == dt.datetime(2026, 7, 5, tzinfo=dt.timezone.utc) + + def test_a_window_inside_the_cap_is_left_alone(self): + start, end = clamp_window(dt.date(2026, 8, 1), dt.date(2026, 8, 3), today=self.TODAY) + assert start == dt.datetime(2026, 8, 1, tzinfo=dt.timezone.utc) + assert end == dt.datetime(2026, 8, 4, tzinfo=dt.timezone.utc) + + def test_the_window_cannot_start_before_the_recency_horizon(self): + """Retention leans on this: a row older than the horizon is one no window can + read, so pruning it is garbage collection rather than data loss.""" + start, _ = clamp_window(dt.date(2026, 6, 1), dt.date(2026, 7, 10), today=self.TODAY) + assert start == dt.datetime(2026, 7, 5, tzinfo=dt.timezone.utc) + + def test_a_future_end_date_is_clamped_to_today(self): + _, end = clamp_window(dt.date(2026, 8, 1), dt.date(2026, 9, 9), today=self.TODAY) + assert end == dt.datetime(2026, 8, 4, tzinfo=dt.timezone.utc) + + def test_a_window_entirely_before_the_horizon_reads_nothing(self): + start, end = clamp_window(dt.date(2026, 1, 1), dt.date(2026, 2, 1), today=self.TODAY) + assert end <= start diff --git a/tests/test_litellm/proxy/spend_tracking/test_auto_router_sessions.py b/tests/test_litellm/proxy/spend_tracking/test_auto_router_sessions.py new file mode 100644 index 000000000000..50c5d3102d98 --- /dev/null +++ b/tests/test_litellm/proxy/spend_tracking/test_auto_router_sessions.py @@ -0,0 +1,192 @@ +"""Reducing one spend-log payload to a rollup turn. + +Bucketing a turn is done by the upsert against the session's own cache record, so it is +covered in tests/proxy_behavior/spend against a real Postgres. What is pure, and covered +here, is deciding whether a request is an auto-routed turn at all and what it contributes. +""" + +import datetime as dt +from dataclasses import replace +from types import SimpleNamespace + +import pytest + +from litellm.proxy.spend_tracking.auto_router_sessions import ( + CACHE_TTL_1H_SECONDS, + CACHE_TTL_5M_SECONDS, + AutoRouterSessionQueue, + TurnFacts, + build_turn_facts, + ttl_seconds, +) + +T0 = dt.datetime(2026, 8, 3, 12, 0, tzinfo=dt.timezone.utc) + + +def _payload(**overrides: object) -> dict[str, object]: + base: dict[str, object] = { + "api_key": "hashed-key", + "session_id": "sess-1", + "model_group": "claude-auto", + "model": "claude-haiku-4-5", + "custom_llm_provider": "anthropic", + "startTime": T0.isoformat(), + "spend": 0.25, + "prompt_tokens": 900, + "completion_tokens": 100, + } + return {**base, **overrides} + + +def _metadata(**overrides: object) -> dict[str, object]: + base: dict[str, object] = { + "routing_decision": {"router_type": "complexity"}, + "usage_object": {"cache_read_input_tokens": 500, "cache_creation_input_tokens": 0}, + } + return {**base, **overrides} + + +def _build(payload=None, metadata=None, **kwargs): + return build_turn_facts( + payload=payload if payload is not None else _payload(), + metadata=metadata if metadata is not None else _metadata(), + autorouter_savings=kwargs.get("autorouter_savings", 0.75), + cache_read_tokens=kwargs.get("cache_read_tokens", 500), + cache_creation_tokens=kwargs.get("cache_creation_tokens", 0), + ) + + +class TestNotAutoRouted: + def test_a_request_with_no_routing_decision_is_not_a_turn(self): + assert _build(metadata={"usage_object": {}}) is None + + def test_a_routing_decision_without_a_kind_is_not_a_turn(self): + assert _build(metadata=_metadata(routing_decision={})) is None + + @pytest.mark.parametrize("field", ["api_key", "session_id", "model_group", "model"]) + def test_a_turn_missing_any_identity_field_is_dropped(self, field: str): + assert _build(payload=_payload(**{field: ""})) is None + assert _build(payload=_payload(**{field: None})) is None + + def test_an_unparseable_start_time_is_dropped(self): + assert _build(payload=_payload(startTime="not-a-timestamp")) is None + + +class TestTurnFacts: + def test_the_router_kind_is_read_from_the_decision_the_router_recorded(self): + built = _build() + assert built is not None and built.router_kind == "complexity" + + def test_the_baseline_arm_is_this_turn_plus_what_the_router_saved(self): + built = _build(autorouter_savings=0.75) + assert built is not None + assert (built.spend, built.baseline_spend) == (pytest.approx(0.25), pytest.approx(1.0)) + + def test_a_route_that_lost_money_carries_a_baseline_below_what_was_paid(self): + built = _build(autorouter_savings=-0.10) + assert built is not None and built.baseline_spend == pytest.approx(0.15) + + def test_tokens_are_the_whole_turn(self): + built = _build() + assert built is not None and built.total_tokens == 1000 + + def test_a_naive_start_time_is_read_as_utc(self): + naive = _build(payload=_payload(startTime=T0.replace(tzinfo=None))) + aware = _build(payload=_payload(startTime=T0)) + assert naive is not None and aware is not None + assert naive.started_at == aware.started_at == T0.timestamp() + + def test_a_cache_read_is_a_hit(self): + built = _build(cache_read_tokens=1) + assert built is not None and built.cache_hit is True + + def test_no_cache_read_is_a_miss(self): + built = _build(cache_read_tokens=0) + assert built is not None and built.cache_hit is False + + +class TestCacheEvidence: + def test_the_live_prefix_is_what_was_read_plus_what_was_written(self): + built = _build(cache_read_tokens=500, cache_creation_tokens=200) + assert built is not None and built.cached_prefix_tokens == 700 + + def test_a_turn_that_touched_no_cache_has_no_prefix(self): + """Coverage keys off this: a model with caching off is absent from the cache view + rather than counted as a miss.""" + built = _build(cache_read_tokens=0, cache_creation_tokens=0) + assert built is not None and built.cached_prefix_tokens == 0 + + def test_a_cache_read_does_not_guess_which_ttl_created_the_entry(self): + assert ttl_seconds({"cache_read_input_tokens": 10}) is None + assert ttl_seconds(None) is None + + def test_five_minute_cache_writes_are_scored_against_the_five_minute_tier(self): + usage = { + "cache_creation_input_tokens": 10, + "cache_creation_token_details": {"ephemeral_5m_input_tokens": 10}, + } + assert ttl_seconds(usage) == CACHE_TTL_5M_SECONDS + + def test_one_hour_cache_writes_are_scored_against_the_one_hour_tier(self): + usage = { + "cache_creation_input_tokens": 10, + "cache_creation_token_details": {"ephemeral_1h_input_tokens": 10}, + } + assert ttl_seconds(usage) == CACHE_TTL_1H_SECONDS + + def test_a_mixed_ttl_write_is_not_collapsed_to_one_ttl(self): + usage = { + "cache_creation_input_tokens": 20, + "cache_creation_token_details": { + "ephemeral_5m_input_tokens": 10, + "ephemeral_1h_input_tokens": 10, + }, + } + assert ttl_seconds(usage) is None + + +class TestFlushOrdering: + @pytest.mark.asyncio + async def test_turns_apply_in_session_key_order_and_in_time_order_within_a_session(self): + """Key order means every pod locks rollup rows in the same sequence (no cross-pod + deadlock); time order within a session is what classification depends on.""" + recorded: list[tuple[object, ...]] = [] + + class _DB: + def batch_(self): + return self + + async def __aenter__(self): + return SimpleNamespace(execute_raw=lambda sql, *params: recorded.append(params)) + + async def __aexit__(self, *exc: object) -> bool: + return False + + base = TurnFacts( + api_key="k", + session_id="a", + model_group="g", + router_kind="complexity", + baseline_model=None, + model="m", + started_at=0.0, + total_tokens=0, + spend=0.0, + baseline_spend=0.0, + cache_hit=False, + cache_creation_tokens=0, + cached_prefix_tokens=0, + ttl_seconds=None, + ) + queue = AutoRouterSessionQueue() + for turn in ( + replace(base, session_id="b", started_at=3.0), + replace(base, started_at=4.0), + replace(base, started_at=2.0), + replace(base, session_id="b", started_at=1.0), + ): + queue.update_queue.put_nowait(turn) + + await queue.flush(SimpleNamespace(db=_DB())) + + assert [(params[1], params[6]) for params in recorded] == [("a", 2.0), ("a", 4.0), ("b", 1.0), ("b", 3.0)] diff --git a/tests/test_litellm/proxy/test_spend_log_cleanup.py b/tests/test_litellm/proxy/test_spend_log_cleanup.py index 2ba9257e1da4..a360598930f8 100644 --- a/tests/test_litellm/proxy/test_spend_log_cleanup.py +++ b/tests/test_litellm/proxy/test_spend_log_cleanup.py @@ -158,8 +158,8 @@ async def test_cleanup_old_spend_logs_batch_deletion(): mock_db = MagicMock() # Mock execute_raw to return deleted counts (3 spend-log batches, then the - # tool-index cleanup's first batch returning 0) - mock_db.execute_raw = AsyncMock(side_effect=[1000, 500, 0, 0]) + # tool-index and auto-router-session cleanups each returning 0 on their first batch) + mock_db.execute_raw = AsyncMock(side_effect=[1000, 500, 0, 0, 0]) # Wire up mocks mock_prisma_client.db = mock_db @@ -179,7 +179,7 @@ async def test_cleanup_old_spend_logs_batch_deletion(): await cleaner.cleanup_old_spend_logs(mock_prisma_client) # Validate batching and deletion via raw SQL - assert mock_db.execute_raw.call_count == 4 + assert mock_db.execute_raw.call_count == 5 # Check the first call argument call_args_sql = mock_db.execute_raw.call_args_list[0][0][0] @@ -193,6 +193,17 @@ async def test_cleanup_old_spend_logs_batch_deletion(): tool_index_sql = mock_db.execute_raw.call_args_list[3][0][0] assert 'DELETE FROM "LiteLLM_SpendLogToolIndex"' in tool_index_sql + # The auto-router rollup expires on the same cutoff when retention is shorter than + # its read horizon, keyed on last activity so a conversation still running when the + # cutoff passes is not pruned mid-session + from datetime import datetime, timedelta, timezone + + session_sql, session_cutoff = mock_db.execute_raw.call_args_list[4][0][:2] + assert 'DELETE FROM "LiteLLM_AutoRouterSession"' in session_sql + assert '"last_turn_at" <' in session_sql + expected_session_cutoff = datetime.now(timezone.utc) - timedelta(days=7) + assert abs((session_cutoff - expected_session_cutoff).total_seconds()) < 60 + # The LiteLLM_DailyToolSpend rollup must outlive spend-log retention: it is # the only copy of tool spend history once its per-request sources expire, # so spend-log cleanup must never touch it. @@ -285,7 +296,7 @@ async def test_cleanup_uses_delete_when_partitioning_not_enabled(): from unittest.mock import AsyncMock, MagicMock mock_prisma_client = MagicMock() - mock_prisma_client.db.execute_raw = AsyncMock(side_effect=[10, 0, 0]) + mock_prisma_client.db.execute_raw = AsyncMock(side_effect=[10, 0, 0, 0]) partition_manager = MagicMock() partition_manager.is_partitioned = AsyncMock(return_value=True) @@ -316,7 +327,7 @@ async def test_cleanup_uses_delete_when_not_partitioned(): from unittest.mock import AsyncMock, MagicMock mock_prisma_client = MagicMock() - mock_prisma_client.db.execute_raw = AsyncMock(side_effect=[10, 0, 0]) + mock_prisma_client.db.execute_raw = AsyncMock(side_effect=[10, 0, 0, 0]) partition_manager = MagicMock() partition_manager.is_partitioned = AsyncMock(return_value=False) @@ -335,7 +346,7 @@ async def test_cleanup_uses_delete_when_not_partitioned(): await cleaner.cleanup_old_spend_logs(mock_prisma_client) partition_manager.drop_partitions_older_than.assert_not_awaited() - assert mock_prisma_client.db.execute_raw.await_count == 3 + assert mock_prisma_client.db.execute_raw.await_count == 4 delete_sql = mock_prisma_client.db.execute_raw.call_args_list[0][0][0] assert 'DELETE FROM "LiteLLM_SpendLogs"' in delete_sql @@ -343,22 +354,32 @@ async def test_cleanup_uses_delete_when_not_partitioned(): @pytest.mark.asyncio async def test_cleanup_old_spend_logs_no_retention_period(): """ - Test that no logs are deleted when no retention period is set + With no retention period set, spend logs are untouched but the auto-router rollup is + still collected at its read horizon: rows past it are unreadable by the benchmarks + endpoint, so the table stays bounded without any configuration. """ + from datetime import datetime, timedelta, timezone + + from litellm.proxy.spend_tracking.auto_router_benchmarks import AUTO_ROUTER_SESSION_RETENTION_DAYS + mock_prisma_client = MagicMock() - mock_prisma_client.db.execute_raw = AsyncMock() + mock_prisma_client.db.execute_raw = AsyncMock(return_value=0) cleaner = SpendLogCleanup(general_settings={}) # no retention await cleaner.cleanup_old_spend_logs(mock_prisma_client) - mock_prisma_client.db.execute_raw.assert_not_called() + assert mock_prisma_client.db.execute_raw.await_count == 1 + session_sql, cutoff = mock_prisma_client.db.execute_raw.call_args_list[0][0][:2] + assert 'DELETE FROM "LiteLLM_AutoRouterSession"' in session_sql + expected_cutoff = datetime.now(timezone.utc) - timedelta(days=AUTO_ROUTER_SESSION_RETENTION_DAYS) + assert abs((cutoff - expected_cutoff).total_seconds()) < 60 @pytest.mark.asyncio async def test_lock_not_released_when_not_acquired(): """ - Lock release should be skipped when _should_delete_spend_logs returns False - before the lock is ever acquired. + When another pod holds the cleanup lock, nothing is deleted (including the always-on + rollup collection) and the lock is not released by this pod. """ mock_prisma_client = MagicMock() mock_prisma_client.db.execute_raw = AsyncMock() @@ -366,17 +387,17 @@ async def test_lock_not_released_when_not_acquired(): mock_redis_cache = MagicMock() mock_pod_lock_manager = MagicMock() mock_pod_lock_manager.redis_cache = mock_redis_cache - mock_pod_lock_manager.acquire_lock = AsyncMock(return_value=True) + mock_pod_lock_manager.acquire_lock = AsyncMock(return_value=False) mock_pod_lock_manager.release_lock = AsyncMock() - # No retention setting → _should_delete_spend_logs() returns False before lock is acquired cleaner = SpendLogCleanup(general_settings={}) cleaner.pod_lock_manager = mock_pod_lock_manager await cleaner.cleanup_old_spend_logs(mock_prisma_client) - mock_pod_lock_manager.acquire_lock.assert_not_called() + mock_pod_lock_manager.acquire_lock.assert_called_once() mock_pod_lock_manager.release_lock.assert_not_called() + mock_prisma_client.db.execute_raw.assert_not_called() @pytest.mark.asyncio diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 864ef80d39de..fb26daf6f785 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -760,6 +760,30 @@ export interface paths { patch?: never; trace?: never; }; + "/auto_router/benchmarks": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Get Auto Router Benchmarks + * @description Savings, session shape and prompt-cache behaviour for every auto-router. + * + * Admin-only. Reads the per-session rollup only; no per-request table is scanned. + * `start_date` and `end_date` are inclusive calendar dates, clamped to the most recent + * 30 days. Pass `model_group` to scope every figure to one auto-router. + */ + get: operations["get_auto_router_benchmarks_auto_router_benchmarks_get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/auto_router/test_routing": { parameters: { query?: never; @@ -21252,6 +21276,121 @@ export interface components { [key: string]: unknown; } | null; }; + /** AutoRouterBenchmark */ + AutoRouterBenchmark: { + /** Avg Session Seconds */ + avg_session_seconds: number; + /** Avg Tokens Per Session */ + avg_tokens_per_session: number; + /** Avg Turns Per Session */ + avg_turns_per_session: number; + /** Baseline Spend */ + baseline_spend: number; + cache: components["schemas"]["AutoRouterCacheBenchmark"] | null; + /** Saved Per Session */ + saved_per_session: number; + /** Savings */ + savings: number; + /** Savings Pct */ + savings_pct: number; + /** Sessions */ + sessions: number; + /** Spend */ + spend: number; + /** Total Tokens */ + total_tokens: number; + /** Turns */ + turns: number; + }; + /** AutoRouterBenchmarksResponse */ + AutoRouterBenchmarksResponse: { + /** + * End Date + * Format: date + */ + end_date: string; + /** Groups */ + groups: components["schemas"]["AutoRouterGroupBenchmark"][]; + /** Routers In Scope */ + routers_in_scope: number; + /** + * Start Date + * Format: date + */ + start_date: string; + totals: components["schemas"]["AutoRouterBenchmark"]; + }; + /** AutoRouterCacheBenchmark */ + AutoRouterCacheBenchmark: { + /** Cold Miss Pct */ + cold_miss_pct: number; + /** Cold Misses */ + cold_misses: number; + /** Coverage Pct */ + coverage_pct: number; + /** Expired Hit Rate Pct */ + expired_hit_rate_pct: number; + /** Expired Hits */ + expired_hits: number; + /** Expired Miss Pct */ + expired_miss_pct: number; + /** Expired Misses */ + expired_misses: number; + /** Expired Turns */ + expired_turns: number; + /** First Visit Hit Rate Pct */ + first_visit_hit_rate_pct: number; + /** First Visit Hits */ + first_visit_hits: number; + /** First Visit Turns */ + first_visit_turns: number; + /** Five Minute Cache Turns */ + five_minute_cache_turns: number; + /** Hit Rate Pct */ + hit_rate_pct: number; + /** Hits */ + hits: number; + /** Misses */ + misses: number; + /** One Hour Cache Turns */ + one_hour_cache_turns: number; + /** Prefix Change Miss Pct */ + prefix_change_miss_pct: number; + /** Prefix Change Misses */ + prefix_change_misses: number; + /** Turns */ + turns: number; + /** Unattributed Miss Pct */ + unattributed_miss_pct: number; + /** Unattributed Misses */ + unattributed_misses: number; + /** Unknown Cache Ttl Turns */ + unknown_cache_ttl_turns: number; + /** Unknown Ttl Hits */ + unknown_ttl_hits: number; + /** Unknown Ttl Turns */ + unknown_ttl_turns: number; + /** Unordered Hits */ + unordered_hits: number; + /** Unordered Turns */ + unordered_turns: number; + /** Warm Hit Rate Pct */ + warm_hit_rate_pct: number; + /** Warm Hits */ + warm_hits: number; + /** Warm Turns */ + warm_turns: number; + }; + /** AutoRouterGroupBenchmark */ + AutoRouterGroupBenchmark: { + /** Baseline Model */ + baseline_model: string | null; + benchmark: components["schemas"]["AutoRouterBenchmark"]; + /** Model Group */ + model_group: string; + /** Router Kind */ + router_kind: string; + }; /** * AutoRouterRoutingTestRequest * @description A single prompt to classify against a complexity-router config that need not be saved yet. @@ -31329,6 +31468,14 @@ export interface components { * @description Keywords indicating reasoning-required content */ reasoning_keywords?: string[] | null; + /** + * Reminder Markers + * @description Override the (open, close) marker pair used to recognize and strip harness-injected reminder blocks before classification. Defaults to Claude Code's convention, ('', ''), when unset. Matching is case-insensitive. + */ + reminder_markers?: [ + string, + string + ] | null; /** * Return Raw Model Name * @description Return the resolved raw model name in the response model field instead of the client-requested complexity-router alias @@ -36304,6 +36451,39 @@ export interface operations { }; }; }; + get_auto_router_benchmarks_auto_router_benchmarks_get: { + parameters: { + query: { + start_date: string; + end_date: string; + model_group?: string | null; + }; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["AutoRouterBenchmarksResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; preview_auto_router_routing_auto_router_test_routing_post: { parameters: { query?: never;