Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
CREATE TABLE IF NOT EXISTS "LiteLLM_AutoRouterDailySpend" (
"date" TEXT NOT NULL,
"api_key" TEXT NOT NULL,
"user_id" TEXT NOT NULL,
"router_name" TEXT NOT NULL,
"router_type" TEXT NOT NULL,
"turns" INTEGER NOT NULL DEFAULT 0,
"spend" DOUBLE PRECISION NOT NULL DEFAULT 0,
"saved_spend" DOUBLE PRECISION NOT NULL DEFAULT 0,
"savings_estimated_turns" INTEGER NOT NULL DEFAULT 0,
"savings_estimated_actual_spend" DOUBLE PRECISION NOT NULL DEFAULT 0,
"savings_estimated_saved_spend" DOUBLE PRECISION NOT NULL DEFAULT 0,
"classifier_cost" DOUBLE PRECISION NOT NULL DEFAULT 0,
"classifier_cost_recorded_turns" INTEGER NOT NULL DEFAULT 0,

CONSTRAINT "LiteLLM_AutoRouterDailySpend_pkey" PRIMARY KEY ("date", "api_key", "user_id", "router_name", "router_type")
);
21 changes: 21 additions & 0 deletions litellm-proxy-extras/litellm_proxy_extras/schema.prisma
Original file line number Diff line number Diff line change
Expand Up @@ -1744,6 +1744,27 @@ model LiteLLM_AutoRouterUserSession {
@@index([user_id, last_turn_at], map: "idx_autorouter_user_session_user_last_turn")
}

// Auto-routed requests per UTC request day and router: the selected-day money behind the
// auto-router usage view. Written in the same statement as the session rollup, so a day row
// and its session row never disagree; corrected in the same transaction as late baselines.
model LiteLLM_AutoRouterDailySpend {
date String
api_key String
user_id String
router_name String
router_type String
turns Int @default(0)
spend Float @default(0)
saved_spend Float @default(0)
savings_estimated_turns Int @default(0)
savings_estimated_actual_spend Float @default(0)
savings_estimated_saved_spend Float @default(0)
classifier_cost Float @default(0)
classifier_cost_recorded_turns Int @default(0)

@@id([date, api_key, user_id, router_name, router_type])
}

// Shadow eval: evaluation of an auto-router against one or more keys' live traffic, in
// either direction. forward duplicates the requests the keys did not route through the
// router through it, answering whether they should adopt it; reverse duplicates the
Expand Down
144 changes: 105 additions & 39 deletions litellm/proxy/db/autorouter_session_rollup.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,12 @@
At request time the spend writer builds one AutoRouterTurnTransaction per successful
auto-routed request (a request whose metadata carries a routing_decision) and queues it
on the prisma client. The spend-log flush job drains the queue into
key and user session rollups with one atomic statement per turn: each upsert classifies
key and user session rollups, plus the per-day router rollup, with one atomic statement
per turn: each upsert classifies
the turn (same model, first visit, return to a model the session already used, out of
order) against the row's own columns, so nothing is read before the write and concurrent
pods compose. The benchmarks endpoint aggregates these rows and can recover matching historical
costs from retained spend logs when estimate coverage predates these columns.
pods compose. The benchmarks endpoint reads session shape from the session rows and money from the
day rows, so spend and savings count only requests on the selected UTC days.
"""

from __future__ import annotations
Expand Down Expand Up @@ -71,45 +72,82 @@
GROUP BY router_name, router_type, kv.key
) per_tier
GROUP BY router_name, router_type
),
sessions AS (
SELECT
router_name,
router_type,
COUNT(*)::int AS sessions,
SUM(turns)::int AS session_turns,
SUM(unordered_turns)::int AS unordered_turns,
SUM(covered_turns)::int AS covered_turns,
SUM(cache_hits)::int AS cache_hits,
SUM(same_model_turns)::int AS same_model_turns,
SUM(same_model_hits)::int AS same_model_hits,
SUM(first_visit_turns)::int AS first_visit_turns,
SUM(first_visit_hits)::int AS first_visit_hits,
SUM(return_turns)::int AS return_turns,
SUM(return_hits)::int AS return_hits,
SUM(return_expired_misses)::int AS return_expired_misses,
SUM(return_within_ttl_misses)::int AS return_within_ttl_misses,
SUM(ttl_5m_turns)::int AS ttl_5m_turns,
SUM(ttl_1h_turns)::int AS ttl_1h_turns,
SUM(total_tokens)::bigint AS total_tokens,
SUM(EXTRACT(EPOCH FROM (last_turn_at - first_turn_at)))::float8 AS session_seconds
FROM windowed
GROUP BY router_name, router_type
),
days AS (
SELECT
router_name,
router_type,
SUM(turns)::int AS turns,
SUM(spend)::float8 AS spend,
SUM(saved_spend)::float8 AS saved_spend,
SUM(savings_estimated_turns)::int AS savings_estimated_turns,
SUM(savings_estimated_actual_spend)::float8 AS savings_estimated_actual_spend,
SUM(savings_estimated_saved_spend)::float8 AS savings_estimated_saved_spend,
SUM(classifier_cost)::float8 AS classifier_cost,
SUM(classifier_cost_recorded_turns)::int AS classifier_cost_recorded_turns
FROM "LiteLLM_AutoRouterDailySpend"
WHERE date >= $5 AND date <= $6
AND ($3::text IS NULL OR api_key = $3::text)
AND ($4::text IS NULL OR user_id = $4::text)
GROUP BY router_name, router_type
)
SELECT
agg.*,
COALESCE(tier_maps.tier_turns, '{{}}'::jsonb) AS tier_turns
FROM (
SELECT
router_name,
router_type,
COUNT(*)::int AS sessions,
COALESCE(SUM(turns), 0)::int AS turns,
COALESCE(SUM(unordered_turns), 0)::int AS unordered_turns,
COALESCE(SUM(covered_turns), 0)::int AS covered_turns,
COALESCE(SUM(cache_hits), 0)::int AS cache_hits,
COALESCE(SUM(same_model_turns), 0)::int AS same_model_turns,
COALESCE(SUM(same_model_hits), 0)::int AS same_model_hits,
COALESCE(SUM(first_visit_turns), 0)::int AS first_visit_turns,
COALESCE(SUM(first_visit_hits), 0)::int AS first_visit_hits,
COALESCE(SUM(return_turns), 0)::int AS return_turns,
COALESCE(SUM(return_hits), 0)::int AS return_hits,
COALESCE(SUM(return_expired_misses), 0)::int AS return_expired_misses,
COALESCE(SUM(return_within_ttl_misses), 0)::int AS return_within_ttl_misses,
COALESCE(SUM(ttl_5m_turns), 0)::int AS ttl_5m_turns,
COALESCE(SUM(ttl_1h_turns), 0)::int AS ttl_1h_turns,
COALESCE(SUM(total_tokens), 0)::bigint AS total_tokens,
COALESCE(SUM(spend), 0)::float8 AS spend,
COALESCE(SUM(saved_spend), 0)::float8 AS saved_spend,
COALESCE(SUM(savings_estimated_turns), 0)::int AS savings_estimated_turns,
COALESCE(SUM(savings_estimated_actual_spend), 0)::float8 AS savings_estimated_actual_spend,
CASE WHEN BOOL_AND(savings_estimated_turns = turns AND classifier_cost_recorded_turns = turns)
THEN SUM(classifier_cost)::float8 END AS savings_estimated_classifier_cost,
COALESCE(SUM(savings_estimated_saved_spend), 0)::float8 AS savings_estimated_saved_spend,
COALESCE(SUM(classifier_cost), 0)::float8 AS classifier_cost,
COALESCE(SUM(classifier_cost_recorded_turns), 0)::int AS classifier_cost_recorded_turns,
COALESCE(SUM(EXTRACT(EPOCH FROM (last_turn_at - first_turn_at))), 0)::float8 AS session_seconds
FROM windowed
GROUP BY router_name, router_type
) agg
COALESCE(tier_maps.tier_turns, '{{}}'::jsonb) AS tier_turns,
COALESCE(sessions.sessions, 0) AS sessions,
COALESCE(sessions.session_turns, 0) AS session_turns,
COALESCE(sessions.unordered_turns, 0) AS unordered_turns,
COALESCE(sessions.covered_turns, 0) AS covered_turns,
COALESCE(sessions.cache_hits, 0) AS cache_hits,
COALESCE(sessions.same_model_turns, 0) AS same_model_turns,
COALESCE(sessions.same_model_hits, 0) AS same_model_hits,
COALESCE(sessions.first_visit_turns, 0) AS first_visit_turns,
COALESCE(sessions.first_visit_hits, 0) AS first_visit_hits,
COALESCE(sessions.return_turns, 0) AS return_turns,
COALESCE(sessions.return_hits, 0) AS return_hits,
COALESCE(sessions.return_expired_misses, 0) AS return_expired_misses,
COALESCE(sessions.return_within_ttl_misses, 0) AS return_within_ttl_misses,
COALESCE(sessions.ttl_5m_turns, 0) AS ttl_5m_turns,
COALESCE(sessions.ttl_1h_turns, 0) AS ttl_1h_turns,
COALESCE(sessions.total_tokens, 0) AS total_tokens,
COALESCE(sessions.session_seconds, 0) AS session_seconds,
COALESCE(days.turns, 0) AS turns,
COALESCE(days.spend, 0) AS spend,
COALESCE(days.saved_spend, 0) AS saved_spend,
COALESCE(days.savings_estimated_turns, 0) AS savings_estimated_turns,
COALESCE(days.savings_estimated_actual_spend, 0) AS savings_estimated_actual_spend,
COALESCE(days.savings_estimated_saved_spend, 0) AS savings_estimated_saved_spend,
COALESCE(days.classifier_cost, 0) AS classifier_cost,
COALESCE(days.classifier_cost_recorded_turns, 0) AS classifier_cost_recorded_turns
FROM sessions
FULL OUTER JOIN days USING (router_name, router_type)
LEFT JOIN tier_maps USING (router_name, router_type)
ORDER BY agg.spend DESC
ORDER BY spend DESC, router_name, router_type
"""


Expand Down Expand Up @@ -391,15 +429,43 @@ def _session_upsert_sql(*, user_scoped: bool) -> str:
"""


_DAY_UPSERT_SQL: Final = f"""
day_rollup AS (
INSERT INTO "LiteLLM_AutoRouterDailySpend" AS d (
date, api_key, user_id, router_name, router_type, turns, spend, saved_spend, savings_estimated_turns,
savings_estimated_actual_spend, savings_estimated_saved_spend, classifier_cost, classifier_cost_recorded_turns
)
VALUES (
({_TURN_AT}::timestamp)::date::text, {_p("api_key")}::text, {_p("user_id")}::text, {_p("router_name")},
{_p("router_type")}, 1, {_p("spend")}::float8, {_p("saved_spend")}::float8, {_p("savings_estimated_turns")}::int,
{_p("savings_estimated_actual_spend")}::float8, {_p("savings_estimated_saved_spend")}::float8,
{_p("classifier_cost")}::float8, 1
)
ON CONFLICT (date, api_key, user_id, router_name, router_type) DO UPDATE SET
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved
turns = d.turns + 1,
spend = d.spend + EXCLUDED.spend,
saved_spend = d.saved_spend + EXCLUDED.saved_spend,
savings_estimated_turns = d.savings_estimated_turns + EXCLUDED.savings_estimated_turns,
savings_estimated_actual_spend = d.savings_estimated_actual_spend + EXCLUDED.savings_estimated_actual_spend,
savings_estimated_saved_spend = d.savings_estimated_saved_spend + EXCLUDED.savings_estimated_saved_spend,
classifier_cost = d.classifier_cost + EXCLUDED.classifier_cost,
classifier_cost_recorded_turns = d.classifier_cost_recorded_turns + 1
RETURNING 1
)
"""

UPSERT_AUTOROUTER_SESSION_SQL: Final = f"""
WITH key_rollup AS (
{_session_upsert_sql(user_scoped=False)}
RETURNING 1
)
), {_DAY_UPSERT_SQL}
{_session_upsert_sql(user_scoped=True)}
"""

UPSERT_AUTOROUTER_USER_SESSION_SQL: Final = _session_upsert_sql(user_scoped=True)
UPSERT_AUTOROUTER_USER_SESSION_SQL: Final = f"""
WITH {_DAY_UPSERT_SQL}
{_session_upsert_sql(user_scoped=True)}
"""


def _as_sql_param(value: str | float | bool | datetime | None) -> str | float | None:
Expand Down
25 changes: 25 additions & 0 deletions litellm/proxy/db/baseline_accounting.py
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,8 @@ class _Change(BaseModel):
actual_delta: float
savings_delta: float
daily: DailyBaselineAttribution | None
date: str | None = None
router_type: str | None = None


class _TransactionManager(Protocol):
Expand Down Expand Up @@ -303,6 +305,26 @@ def _session_correction_sql(*, user_scoped: bool) -> str:

_UPDATE_SESSIONS: Final = _session_correction_sql(user_scoped=False)
_UPDATE_USER_SESSIONS: Final = _session_correction_sql(user_scoped=True)
_UPDATE_DAYS: Final = """
WITH totals AS (
SELECT date, api_key, user_id, router_name, router_type, SUM(covered_delta)::int AS covered_delta,
SUM(actual_delta) AS actual_delta, SUM(savings_delta) AS savings_delta
FROM jsonb_to_recordset($1::jsonb) AS x(
date text, api_key text, user_id text, router_name text, router_type text,
covered_delta int, actual_delta float8, savings_delta float8
)
WHERE date IS NOT NULL
GROUP BY date, api_key, user_id, router_name, router_type
)
UPDATE "LiteLLM_AutoRouterDailySpend" AS day
SET saved_spend = day.saved_spend + totals.savings_delta,
savings_estimated_turns = day.savings_estimated_turns + totals.covered_delta,
savings_estimated_actual_spend = day.savings_estimated_actual_spend + totals.actual_delta,
savings_estimated_saved_spend = day.savings_estimated_saved_spend + totals.savings_delta
FROM totals
WHERE day.date = totals.date AND day.api_key = totals.api_key AND day.user_id = totals.user_id
AND day.router_name = totals.router_name AND day.router_type = totals.router_type
"""


def _primary_transaction(client: PrismaClient) -> _TransactionManager:
Expand Down Expand Up @@ -331,6 +353,8 @@ def _change(record: BaselineAccountingRecord, old: BaselinePublication | None, n
savings_delta=(current.savings if current is not None else 0.0)
- (previous.savings if previous is not None else 0.0),
daily=record.daily,
date=record.turn.turn_at.date().isoformat() if record.turn is not None else None,
router_type=record.turn.router_type if record.turn is not None else None,
)


Expand Down Expand Up @@ -373,6 +397,7 @@ async def _publish(db: SupportsRawQueries, changes: Sequence[_Change]) -> None:
await db.execute_raw(_UPDATE_SESSIONS, serialized)
if any(change.user_id for change in changes):
await db.execute_raw(_UPDATE_USER_SESSIONS, serialized)
await db.execute_raw(_UPDATE_DAYS, serialized)
for entity, table in DAILY_SPEND_TABLES.items():
if adjustments := tuple(
change.daily.adjustment(target, change.savings_delta, change.request_id)
Expand Down
22 changes: 19 additions & 3 deletions litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py
Original file line number Diff line number Diff line change
Expand Up @@ -540,6 +540,18 @@ async def _delete_old_autorouter_user_session_rows(
deadline=deadline,
)

async def _delete_old_autorouter_daily_rows(
self, prisma_client: PrismaClient, cutoff_day: str, deadline: float
) -> TableCleanupResult:
return await self._delete_old_rows_batched(
prisma_client,
cutoff_day,
table_name="LiteLLM_AutoRouterDailySpend",
key_columns=("date", "api_key", "user_id", "router_name", "router_type"),
time_column="date",
deadline=deadline,
)

async def _delete_old_health_check_rows(
self, prisma_client: PrismaClient, cutoff_date: datetime, deadline: float
) -> TableCleanupResult:
Expand Down Expand Up @@ -623,16 +635,20 @@ async def _clean_session_rollup(
except Exception: # noqa: BLE001 # retained observations are retried by the next cleanup job
verbose_proxy_logger.warning("Auto-router baseline retention remains pending")
sessions_result: Final = await self._delete_old_autorouter_session_rows(
prisma_client, session_cutoff, self._group_deadline(deadline, 2)
prisma_client, session_cutoff, self._group_deadline(deadline, 3)
)
verbose_proxy_logger.info("Deleted %s expired auto-router session rollup rows", sessions_result.rows_deleted)
user_sessions_result: Final = await self._delete_old_autorouter_user_session_rows(
prisma_client, session_cutoff, deadline
prisma_client, session_cutoff, self._group_deadline(deadline, 2)
)
verbose_proxy_logger.info(
"Deleted %s expired auto-router user session rollup rows", user_sessions_result.rows_deleted
)
return (sessions_result, user_sessions_result)
days_result: Final = await self._delete_old_autorouter_daily_rows(
prisma_client, session_cutoff.date().isoformat(), deadline
)
verbose_proxy_logger.info("Deleted %s expired auto-router daily rollup rows", days_result.rows_deleted)
return (sessions_result, user_sessions_result, days_result)

async def _clean_health_checks(
self, prisma_client: PrismaClient, retention_seconds: int, deadline: float
Expand Down
Loading
Loading