mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
feat(shadow_eval): measure both arms' cost so a job reports what the router would have saved
The attempt row now prices the real arm (the payload's response_cost plus its own routing classifier when it routed) beside the shadow arm (completion plus the classifier cost the routing decision writes back), and flags turns litellm's response cache served. A per-leg funnel table counts the eligible requests that produced no row (lost the sampling dice, unjudgeable shape, concurrency shed), so results can weigh judged rows against the traffic they stand for. Job results gain per-slice and overall arm spends plus the coverage counts, the budget gates charge the shadow arm's classifier spend against max_budget, and the dashboard shows the measured cost comparison beside the win rate Resolves LIT-6358 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Tin Chi Lo
parent
3300fc3a96
commit
2d5ffc56cc
16 changed files with 959 additions and 55 deletions
|
|
@ -0,0 +1,21 @@
|
|||
-- AlterTable
|
||||
ALTER TABLE "LiteLLM_ShadowEvalAttempt" ADD COLUMN IF NOT EXISTS "real_cost" DOUBLE PRECISION NOT NULL DEFAULT 0;
|
||||
|
||||
-- AlterTable
|
||||
ALTER TABLE "LiteLLM_ShadowEvalAttempt" ADD COLUMN IF NOT EXISTS "real_classifier_cost" DOUBLE PRECISION NOT NULL DEFAULT 0;
|
||||
|
||||
-- AlterTable
|
||||
ALTER TABLE "LiteLLM_ShadowEvalAttempt" ADD COLUMN IF NOT EXISTS "shadow_classifier_cost" DOUBLE PRECISION NOT NULL DEFAULT 0;
|
||||
|
||||
-- AlterTable
|
||||
ALTER TABLE "LiteLLM_ShadowEvalAttempt" ADD COLUMN IF NOT EXISTS "real_cache_hit" BOOLEAN NOT NULL DEFAULT false;
|
||||
|
||||
-- CreateTable
|
||||
CREATE TABLE IF NOT EXISTS "LiteLLM_ShadowEvalFunnel" (
|
||||
"job_id" TEXT NOT NULL,
|
||||
"not_sampled" INTEGER NOT NULL DEFAULT 0,
|
||||
"unjudgeable" INTEGER NOT NULL DEFAULT 0,
|
||||
"shed" INTEGER NOT NULL DEFAULT 0,
|
||||
|
||||
CONSTRAINT "LiteLLM_ShadowEvalFunnel_pkey" PRIMARY KEY ("job_id")
|
||||
);
|
||||
|
|
@ -1527,12 +1527,26 @@ model LiteLLM_ShadowEvalAttempt {
|
|||
confidence Float?
|
||||
judge_cost Float @default(0)
|
||||
shadow_cost Float @default(0)
|
||||
real_cost Float @default(0)
|
||||
real_classifier_cost Float @default(0)
|
||||
shadow_classifier_cost Float @default(0)
|
||||
real_cache_hit Boolean @default(false)
|
||||
error String?
|
||||
created_at DateTime @default(now())
|
||||
|
||||
@@index([job_id])
|
||||
}
|
||||
|
||||
// Per-leg sampling funnel counters the attempt rows cannot derive: requests an
|
||||
// admitting job saw but did not judge. attempted = the leg's attempt rows; the
|
||||
// leg's eligible traffic = not_sampled + unjudgeable + shed + attempted.
|
||||
model LiteLLM_ShadowEvalFunnel {
|
||||
job_id String @id
|
||||
not_sampled Int @default(0)
|
||||
unjudgeable Int @default(0)
|
||||
shed Int @default(0)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Workflow Run Tracking
|
||||
//
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ from litellm.types.management_endpoints.auto_router_endpoints import ShadowEvalD
|
|||
from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.proxy.db.shadow_eval_funnel import ShadowEvalFunnelStage
|
||||
from litellm.proxy.utils import PrismaClient
|
||||
from litellm.router import Router
|
||||
from litellm.types.utils import StandardLoggingPayload
|
||||
|
|
@ -386,6 +387,13 @@ def _judge_user_prompt(conversation: str, response_a: str, response_b: str) -> s
|
|||
)
|
||||
|
||||
|
||||
def _leg_eval_spend(sums: Mapping[str, object]) -> float:
|
||||
return sum(
|
||||
float(raw) if isinstance(raw := sums.get(column), (int, float)) else 0.0
|
||||
for column in ("judge_cost", "shadow_cost", "shadow_classifier_cost")
|
||||
)
|
||||
|
||||
|
||||
def _job_spend_counter_key(job_id: str) -> str:
|
||||
return f"spend:shadow_eval:{job_id}"
|
||||
|
||||
|
|
@ -412,6 +420,15 @@ async def _add_job_spend_to_counter(counter_key: str, cost: float) -> None:
|
|||
verbose_logger.warning("shadow_eval: spend counter increment failed for %s: %s", counter_key, e)
|
||||
|
||||
|
||||
def _record_funnel_event(job_id: str, stage: "ShadowEvalFunnelStage") -> None:
|
||||
try:
|
||||
from litellm.proxy.db.shadow_eval_funnel import record_shadow_eval_funnel_event
|
||||
|
||||
record_shadow_eval_funnel_event(job_id, stage)
|
||||
except Exception as e: # noqa: BLE001 # coverage stats are advisory; sampling must proceed
|
||||
verbose_logger.debug("shadow_eval: funnel increment failed for %s: %s", job_id, e)
|
||||
|
||||
|
||||
async def _key_or_team_is_over_budget(metadata: Mapping[str, object]) -> bool:
|
||||
"""Whether the shadowed key or its team is over budget, decided by the same owners
|
||||
the request path uses, so counter keys and thresholds can never drift from auth's.
|
||||
|
|
@ -474,6 +491,13 @@ def _routed_tier(metadata: Mapping[str, object]) -> str | None:
|
|||
return str(raw) if raw is not None else None
|
||||
|
||||
|
||||
def _decision_classifier_cost(metadata: Mapping[str, object]) -> float:
|
||||
"""What the arm's own routing decision says its classifier call billed: the money a
|
||||
completion cost alone omits, and 0 for a plain model that never classifies."""
|
||||
raw: Final = _routing_decision(metadata).get("classifier_cost")
|
||||
return float(raw) if isinstance(raw, (int, float)) else 0.0
|
||||
|
||||
|
||||
def _request_was_routed_by(request_metadata: Mapping[str, object], router_name: str) -> bool:
|
||||
"""Whether the router under evaluation served this request, which is what decides
|
||||
the direction it belongs to. A forward job skips its own router's traffic, since
|
||||
|
|
@ -489,6 +513,7 @@ class _CallFailure:
|
|||
|
||||
error: str
|
||||
cost: float = 0.0
|
||||
classifier_cost: float = 0.0
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -499,6 +524,7 @@ class _ShadowResponse:
|
|||
model: str
|
||||
tier: str | None
|
||||
cost: float
|
||||
classifier_cost: float
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -575,6 +601,7 @@ class ShadowEvalLogger(CustomLogger):
|
|||
jobs_cache: InMemoryCache | None = None,
|
||||
job_spend_reader: Callable[[str, float, float], Awaitable[float]] | None = None,
|
||||
job_spend_writer: Callable[[str, float], Awaitable[None]] | None = None,
|
||||
funnel_recorder: Callable[[str, "ShadowEvalFunnelStage"], None] | None = None,
|
||||
) -> None:
|
||||
"""Providers are callables so the proxy's lazily-initialized globals are resolved
|
||||
at call time, not at logger construction. The spend reader and writer wrap the
|
||||
|
|
@ -584,6 +611,7 @@ class ShadowEvalLogger(CustomLogger):
|
|||
self._jobs_cache = jobs_cache or _jobs_cache
|
||||
self._read_job_spend = job_spend_reader or _job_spend_from_counter
|
||||
self._write_job_spend = job_spend_writer or _add_job_spend_to_counter
|
||||
self._record_funnel = funnel_recorder or _record_funnel_event
|
||||
self._inflight_shadow_tasks: int = 0
|
||||
# Starts per job since the last cache fill, never decremented within a
|
||||
# generation; the refill absorbs written rows and resets.
|
||||
|
|
@ -610,7 +638,8 @@ class ShadowEvalLogger(CustomLogger):
|
|||
await prisma.db.litellm_shadowevalattempt.group_by(
|
||||
by=["job_id"],
|
||||
count=True,
|
||||
sum={"judge_cost": True, "shadow_cost": True}, # mutable-ok: Prisma aggregate spec
|
||||
# mutable-ok: Prisma aggregate spec
|
||||
sum={"judge_cost": True, "shadow_cost": True, "shadow_classifier_cost": True},
|
||||
where={"job_id": {"in": [str(record.id) for record in records]}}, # mutable-ok: Prisma filter
|
||||
)
|
||||
if records
|
||||
|
|
@ -619,8 +648,7 @@ class ShadowEvalLogger(CustomLogger):
|
|||
attempt_stats: Final = { # mutable-ok: frozen snapshot of the grouped read
|
||||
str(row["job_id"]): (
|
||||
int(row["_count"]["_all"]),
|
||||
float((row["_sum"] or {}).get("judge_cost") or 0.0)
|
||||
+ float((row["_sum"] or {}).get("shadow_cost") or 0.0),
|
||||
_leg_eval_spend(row["_sum"] or _EMPTY_METADATA),
|
||||
)
|
||||
for row in grouped or []
|
||||
}
|
||||
|
|
@ -646,6 +674,32 @@ class ShadowEvalLogger(CustomLogger):
|
|||
|
||||
#### hook ####
|
||||
|
||||
def _sampled_jobs(
|
||||
self,
|
||||
active_jobs: Sequence[ActiveShadowEvalJob],
|
||||
request_metadata: Mapping[str, object],
|
||||
request_id: str,
|
||||
) -> tuple[ActiveShadowEvalJob, ...]:
|
||||
"""The jobs that sample this request. A key can hold one job per direction, and a
|
||||
request routed by one job's router while bypassing the other's qualifies for both;
|
||||
each is separately budgeted, so both fire. An admitting job that loses the sampling
|
||||
dice is counted, so results can weigh judged rows against the traffic they stand for."""
|
||||
eligible: list[ActiveShadowEvalJob] = [] # mutable-ok: bucketed per-job admission
|
||||
now: Final = datetime.now(timezone.utc)
|
||||
for job in active_jobs:
|
||||
if (
|
||||
now >= job.ends_at
|
||||
or job.attempts + self._job_starts.get(job.id, 0) >= job.max_turns
|
||||
or (job.max_budget is not None and job.spend >= job.max_budget)
|
||||
or _request_was_routed_by(request_metadata, job.router_name) != (job.direction == "reverse")
|
||||
):
|
||||
continue
|
||||
if not _sample_hits(request_id, job.id, job.shadow_percentage):
|
||||
self._record_funnel(job.id, "not_sampled")
|
||||
continue
|
||||
eligible.append(job)
|
||||
return tuple(eligible)
|
||||
|
||||
async def async_log_success_event(
|
||||
self,
|
||||
kwargs: Mapping[str, object],
|
||||
|
|
@ -677,18 +731,8 @@ class ShadowEvalLogger(CustomLogger):
|
|||
return # only surfaces this table can normalize are comparable; unknown types fail closed
|
||||
if ops.wire_params and _request_mutating_guardrail_ran(request_metadata):
|
||||
return # the wire-body snapshot predates the rewrite; replaying it would resurrect stripped content
|
||||
# A key can hold one job per direction, and a request routed by one job's
|
||||
# router while bypassing the other's qualifies for both. Each is separately
|
||||
# budgeted, so both fire; the request is normalized once, and only when at
|
||||
# least one job sampled it.
|
||||
eligible: Final = tuple(
|
||||
job
|
||||
for job in (await self._active_jobs()).get(str(api_key_hash), ())
|
||||
if datetime.now(timezone.utc) < job.ends_at
|
||||
and job.attempts + self._job_starts.get(job.id, 0) < job.max_turns
|
||||
and (job.max_budget is None or job.spend < job.max_budget)
|
||||
and _sample_hits(request_id, job.id, job.shadow_percentage)
|
||||
and _request_was_routed_by(request_metadata, job.router_name) == (job.direction == "reverse")
|
||||
eligible: Final = self._sampled_jobs(
|
||||
(await self._active_jobs()).get(str(api_key_hash), ()), request_metadata, request_id
|
||||
)
|
||||
if not eligible:
|
||||
return
|
||||
|
|
@ -699,12 +743,18 @@ class ShadowEvalLogger(CustomLogger):
|
|||
response_obj,
|
||||
)
|
||||
if sample is None:
|
||||
for job in eligible:
|
||||
self._record_funnel(job.id, "unjudgeable")
|
||||
return
|
||||
messages, shadow_params, real_text = sample
|
||||
control_tier: Final = _routed_tier(request_metadata)
|
||||
real_cost: Final = float(payload.get("response_cost") or 0.0)
|
||||
real_cache_hit: Final = payload.get("cache_hit") is True
|
||||
real_classifier_cost: Final = _decision_classifier_cost(request_metadata)
|
||||
for job in eligible:
|
||||
if self._inflight_shadow_tasks >= _MAX_CONCURRENT_SHADOW_TASKS:
|
||||
return
|
||||
self._record_funnel(job.id, "shed")
|
||||
continue
|
||||
self._job_starts[job.id] = self._job_starts.get(job.id, 0) + 1
|
||||
self._inflight_shadow_tasks += 1
|
||||
asyncio.create_task(
|
||||
|
|
@ -714,6 +764,9 @@ class ShadowEvalLogger(CustomLogger):
|
|||
messages=messages,
|
||||
real_text=real_text,
|
||||
real_model=payload.get("model") or "",
|
||||
real_cost=real_cost,
|
||||
real_classifier_cost=real_classifier_cost,
|
||||
real_cache_hit=real_cache_hit,
|
||||
control_tier=control_tier,
|
||||
shadow_params=shadow_params,
|
||||
parent_metadata=MappingProxyType(dict(request_metadata)), # mutable-ok: frozen snapshot
|
||||
|
|
@ -734,6 +787,9 @@ class ShadowEvalLogger(CustomLogger):
|
|||
messages: Sequence[Mapping[str, object]],
|
||||
real_text: str,
|
||||
real_model: str,
|
||||
real_cost: float,
|
||||
real_classifier_cost: float,
|
||||
real_cache_hit: bool,
|
||||
control_tier: str | None,
|
||||
shadow_params: Mapping[str, object],
|
||||
parent_metadata: Mapping[str, object],
|
||||
|
|
@ -759,12 +815,30 @@ class ShadowEvalLogger(CustomLogger):
|
|||
except Exception as e: # noqa: BLE001 # detached task: nothing billed yet, record and never raise
|
||||
verbose_logger.debug("shadow_eval: pipeline failed for %s: %s", request_id, e)
|
||||
await self._record_attempt(
|
||||
prisma, job, request_id, control_tier, outcome="error", error=f"pipeline error: {e}"
|
||||
prisma,
|
||||
job,
|
||||
request_id,
|
||||
control_tier,
|
||||
outcome="error",
|
||||
error=f"pipeline error: {e}",
|
||||
real_cost=real_cost,
|
||||
real_classifier_cost=real_classifier_cost,
|
||||
real_cache_hit=real_cache_hit,
|
||||
)
|
||||
return
|
||||
if isinstance(shadow, _CallFailure):
|
||||
await self._record_attempt(
|
||||
prisma, job, request_id, control_tier, outcome="error", error=shadow.error, shadow_cost=shadow.cost
|
||||
prisma,
|
||||
job,
|
||||
request_id,
|
||||
control_tier,
|
||||
outcome="error",
|
||||
error=shadow.error,
|
||||
shadow_cost=shadow.cost,
|
||||
shadow_classifier_cost=shadow.classifier_cost,
|
||||
real_cost=real_cost,
|
||||
real_classifier_cost=real_classifier_cost,
|
||||
real_cache_hit=real_cache_hit,
|
||||
)
|
||||
return
|
||||
# From here the shadow call has billed, so every exit records its cost.
|
||||
|
|
@ -787,6 +861,10 @@ class ShadowEvalLogger(CustomLogger):
|
|||
shadow=shadow,
|
||||
judge_cost=verdict.cost,
|
||||
shadow_cost=shadow.cost,
|
||||
shadow_classifier_cost=shadow.classifier_cost,
|
||||
real_cost=real_cost,
|
||||
real_classifier_cost=real_classifier_cost,
|
||||
real_cache_hit=real_cache_hit,
|
||||
)
|
||||
return
|
||||
await self._record_attempt(
|
||||
|
|
@ -800,6 +878,10 @@ class ShadowEvalLogger(CustomLogger):
|
|||
confidence=verdict.confidence,
|
||||
judge_cost=verdict.cost,
|
||||
shadow_cost=shadow.cost,
|
||||
shadow_classifier_cost=shadow.classifier_cost,
|
||||
real_cost=real_cost,
|
||||
real_classifier_cost=real_classifier_cost,
|
||||
real_cache_hit=real_cache_hit,
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 # detached task: the shadow call billed, record its cost, never raise
|
||||
verbose_logger.debug("shadow_eval: pipeline failed for %s: %s", request_id, e)
|
||||
|
|
@ -812,6 +894,10 @@ class ShadowEvalLogger(CustomLogger):
|
|||
error=f"pipeline error: {e}",
|
||||
shadow=shadow,
|
||||
shadow_cost=shadow.cost,
|
||||
shadow_classifier_cost=shadow.classifier_cost,
|
||||
real_cost=real_cost,
|
||||
real_classifier_cost=real_classifier_cost,
|
||||
real_cache_hit=real_cache_hit,
|
||||
)
|
||||
|
||||
async def _record_attempt(
|
||||
|
|
@ -822,15 +908,20 @@ class ShadowEvalLogger(CustomLogger):
|
|||
control_tier: str | None,
|
||||
*,
|
||||
outcome: str,
|
||||
real_cost: float,
|
||||
real_classifier_cost: float,
|
||||
real_cache_hit: bool,
|
||||
shadow: _ShadowResponse | None = None,
|
||||
real_model: str = "",
|
||||
confidence: float | None = None,
|
||||
judge_cost: float = 0.0,
|
||||
shadow_cost: float = 0.0,
|
||||
shadow_classifier_cost: float = 0.0,
|
||||
error: str | None = None,
|
||||
) -> None:
|
||||
if judge_cost + shadow_cost > 0:
|
||||
await self._write_job_spend(_job_spend_counter_key(job.id), judge_cost + shadow_cost)
|
||||
eval_spend: Final = judge_cost + shadow_cost + shadow_classifier_cost
|
||||
if eval_spend > 0:
|
||||
await self._write_job_spend(_job_spend_counter_key(job.id), eval_spend)
|
||||
if prisma is None:
|
||||
return
|
||||
try:
|
||||
|
|
@ -845,6 +936,10 @@ class ShadowEvalLogger(CustomLogger):
|
|||
"confidence": confidence,
|
||||
"judge_cost": judge_cost,
|
||||
"shadow_cost": shadow_cost,
|
||||
"shadow_classifier_cost": shadow_classifier_cost,
|
||||
"real_cost": real_cost,
|
||||
"real_classifier_cost": real_classifier_cost,
|
||||
"real_cache_hit": real_cache_hit,
|
||||
"error": error[:_MAX_ERROR_CHARS] if error else None,
|
||||
}
|
||||
)
|
||||
|
|
@ -881,15 +976,23 @@ class ShadowEvalLogger(CustomLogger):
|
|||
)
|
||||
except Exception as e: # noqa: BLE001 # provider errors become error rows, not crashes
|
||||
verbose_logger.debug("shadow_eval: router call failed: %s", e)
|
||||
return _CallFailure(f"shadow router call failed: {_failure_detail(e)}")
|
||||
return _CallFailure(
|
||||
f"shadow router call failed: {_failure_detail(e)}",
|
||||
classifier_cost=_decision_classifier_cost(shadow_metadata),
|
||||
)
|
||||
text: Final = _chat_final_text(response)
|
||||
if not text:
|
||||
return _CallFailure("shadow router returned an empty response", cost=_call_cost(response))
|
||||
return _CallFailure(
|
||||
"shadow router returned an empty response",
|
||||
cost=_call_cost(response),
|
||||
classifier_cost=_decision_classifier_cost(shadow_metadata),
|
||||
)
|
||||
return _ShadowResponse(
|
||||
text=text,
|
||||
model=str(getattr(response, "model", None) or _routing_decision(shadow_metadata).get("routed_model") or ""),
|
||||
tier=_routed_tier(shadow_metadata),
|
||||
cost=_call_cost(response),
|
||||
classifier_cost=_decision_classifier_cost(shadow_metadata),
|
||||
)
|
||||
|
||||
async def _call_judge(
|
||||
|
|
|
|||
63
litellm/proxy/db/shadow_eval_funnel.py
Normal file
63
litellm/proxy/db/shadow_eval_funnel.py
Normal file
|
|
@ -0,0 +1,63 @@
|
|||
"""Pod-local queue of shadow-eval funnel increments, drained by the spend-update job.
|
||||
|
||||
The shadow-eval success hook counts the sampled-traffic outcomes that never produce an
|
||||
attempt row (a lost sampling dice roll, an unjudgeable request shape, a concurrency
|
||||
shed), so a job's results can state what share of its eligible traffic the judged rows
|
||||
represent. Counters are advisory coverage stats: a pod dying loses at most one flush
|
||||
interval, and a failed flush drops its batch because a repeated increment is worse
|
||||
than an undercount (same call as the auto-router session rollup flush).
|
||||
"""
|
||||
|
||||
from typing import TYPE_CHECKING, Final, Literal
|
||||
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.proxy.utils import PrismaClient
|
||||
|
||||
ShadowEvalFunnelStage = Literal["not_sampled", "unjudgeable", "shed"]
|
||||
|
||||
FUNNEL_STAGES: Final[tuple[ShadowEvalFunnelStage, ...]] = ("not_sampled", "unjudgeable", "shed")
|
||||
|
||||
_pending: dict[str, dict[ShadowEvalFunnelStage, int]] = {} # mutable-ok: module-level queue, single event loop
|
||||
|
||||
_UPSERT_FUNNEL_SQL: Final = f"""
|
||||
INSERT INTO "LiteLLM_ShadowEvalFunnel" (job_id, {", ".join(FUNNEL_STAGES)})
|
||||
VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (job_id) DO UPDATE SET
|
||||
{", ".join(f'{stage} = "LiteLLM_ShadowEvalFunnel".{stage} + EXCLUDED.{stage}' for stage in FUNNEL_STAGES)}
|
||||
"""
|
||||
|
||||
|
||||
def pending_shadow_eval_funnel_events() -> int:
|
||||
"""Queue census for the drain triggers: entries not yet flushed, so a funnel-only
|
||||
batch still wakes the spend job that would otherwise skip an empty-queue run."""
|
||||
return sum(sum(counters.values()) for counters in _pending.values())
|
||||
|
||||
|
||||
def record_shadow_eval_funnel_event(job_id: str, stage: ShadowEvalFunnelStage) -> None:
|
||||
"""Count one skipped request for one job leg; synchronous so the hook's read-modify-
|
||||
write cannot interleave with the flush's snapshot on the shared event loop."""
|
||||
counters: Final = _pending.setdefault(job_id, dict.fromkeys(FUNNEL_STAGES, 0)) # mutable-ok: queue entry
|
||||
counters[stage] += 1
|
||||
|
||||
|
||||
async def flush_shadow_eval_funnel(prisma_client: "PrismaClient") -> None:
|
||||
if not _pending:
|
||||
return
|
||||
batch: Final = dict(_pending) # mutable-ok: snapshot drained from the queue
|
||||
_pending.clear()
|
||||
for job_id, counters in batch.items():
|
||||
try:
|
||||
await prisma_client.db.execute_raw(
|
||||
_UPSERT_FUNNEL_SQL,
|
||||
job_id,
|
||||
*(counters[stage] for stage in FUNNEL_STAGES),
|
||||
)
|
||||
except Exception as flush_err: # noqa: BLE001 # drop this leg's batch: a repeated increment is worse than an undercount
|
||||
verbose_proxy_logger.error(
|
||||
"Spend tracking - shadow eval funnel flush failed for job %s, %s dropped: %s",
|
||||
job_id,
|
||||
counters,
|
||||
flush_err,
|
||||
)
|
||||
|
|
@ -119,6 +119,10 @@ class _ShadowEvalAttemptRow(Protocol):
|
|||
def error(self) -> str | None: ...
|
||||
|
||||
|
||||
class _ShadowEvalFunnelTable(Protocol):
|
||||
async def create_many(self, data: Sequence[Mapping[str, object]], skip_duplicates: bool) -> int: ...
|
||||
|
||||
|
||||
class _ShadowEvalAttemptTable(Protocol):
|
||||
async def find_first(
|
||||
self, *, where: Mapping[str, object], order: Mapping[str, str]
|
||||
|
|
@ -137,6 +141,10 @@ def _shadow_eval_jobs(prisma_client: "PrismaClient") -> _ShadowEvalJobTable:
|
|||
return prisma_client.db.litellm_shadowevaljob
|
||||
|
||||
|
||||
def _shadow_eval_funnel(prisma_client: "PrismaClient") -> _ShadowEvalFunnelTable:
|
||||
return prisma_client.db.litellm_shadowevalfunnel # pyright: ignore[reportAttributeAccessIssue] # generated client
|
||||
|
||||
|
||||
def _shadow_eval_attempts(prisma_client: "PrismaClient") -> _ShadowEvalAttemptTable:
|
||||
return prisma_client.db.litellm_shadowevalattempt
|
||||
|
||||
|
|
@ -820,6 +828,9 @@ class _AttemptAggRow(BaseModel):
|
|||
shadow_wins: int
|
||||
ties: int
|
||||
avg_confidence: float | None
|
||||
real_spend: float
|
||||
shadow_spend: float
|
||||
cache_hit_turns: int
|
||||
|
||||
|
||||
_ATTEMPT_AGG_ROWS: Final = TypeAdapter(list[_AttemptAggRow])
|
||||
|
|
@ -829,7 +840,10 @@ _ATTEMPT_AGG_SELECT: Final = """
|
|||
COUNT(*) FILTER (WHERE outcome = 'real')::int AS real_wins,
|
||||
COUNT(*) FILTER (WHERE outcome = 'shadow')::int AS shadow_wins,
|
||||
COUNT(*) FILTER (WHERE outcome = 'tie')::int AS ties,
|
||||
AVG(confidence)::float AS avg_confidence
|
||||
AVG(confidence)::float AS avg_confidence,
|
||||
COALESCE(SUM(real_cost + real_classifier_cost) FILTER (WHERE NOT real_cache_hit), 0)::float AS real_spend,
|
||||
COALESCE(SUM(shadow_cost + shadow_classifier_cost) FILTER (WHERE NOT real_cache_hit), 0)::float AS shadow_spend,
|
||||
COUNT(*) FILTER (WHERE real_cache_hit)::int AS cache_hit_turns
|
||||
FROM "LiteLLM_ShadowEvalAttempt"
|
||||
WHERE job_id = ANY($1::text[]) AND outcome != 'error'
|
||||
GROUP BY 1
|
||||
|
|
@ -850,7 +864,7 @@ WHERE j.api_key_id = ANY($1::text[]) AND j.stopped_at IS NULL
|
|||
OR (SELECT COUNT(*) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = j.id) >= j.max_turns
|
||||
OR (
|
||||
j.max_budget IS NOT NULL
|
||||
AND (SELECT COALESCE(SUM(a.judge_cost + a.shadow_cost), 0) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = j.id) >= j.max_budget
|
||||
AND (SELECT COALESCE(SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost), 0) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = j.id) >= j.max_budget
|
||||
)
|
||||
)
|
||||
"""
|
||||
|
|
@ -865,13 +879,23 @@ WHERE job_id = ANY($1::text[])
|
|||
"""
|
||||
|
||||
_ATTEMPT_COUNTS_SQL: Final = """
|
||||
SELECT a.job_id, COUNT(*)::int AS attempt_count, COALESCE(SUM(a.judge_cost + a.shadow_cost), 0)::float AS spend
|
||||
SELECT a.job_id, COUNT(*)::int AS attempt_count, COALESCE(SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost), 0)::float AS spend
|
||||
FROM "LiteLLM_ShadowEvalAttempt" a
|
||||
JOIN "LiteLLM_ShadowEvalJob" j ON j.id = a.job_id
|
||||
WHERE a.job_id = ANY($1::text[]) AND (j.stopped_at IS NULL OR a.created_at <= j.stopped_at)
|
||||
GROUP BY a.job_id
|
||||
"""
|
||||
|
||||
_FUNNEL_TOTALS_SQL: Final = """
|
||||
SELECT COUNT(*)::int AS legs_with_rows,
|
||||
COALESCE(SUM(not_sampled), 0)::int AS not_sampled,
|
||||
COALESCE(SUM(unjudgeable), 0)::int AS unjudgeable,
|
||||
COALESCE(SUM(shed), 0)::int AS shed
|
||||
FROM "LiteLLM_ShadowEvalFunnel"
|
||||
WHERE job_id = ANY($1::text[])
|
||||
"""
|
||||
|
||||
|
||||
_STOP_JOB_SQL: Final = """
|
||||
UPDATE "LiteLLM_ShadowEvalJob"
|
||||
SET stopped_by = $2, stopped_at = COALESCE(stopped_at, $3::timestamp)
|
||||
|
|
@ -883,12 +907,19 @@ WHERE group_id = $1 AND stopped_by IS NULL
|
|||
AND (SELECT COUNT(*) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = k.id) < k.max_turns
|
||||
AND (
|
||||
k.max_budget IS NULL
|
||||
OR (SELECT COALESCE(SUM(a.judge_cost + a.shadow_cost), 0) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = k.id) < k.max_budget
|
||||
OR (SELECT COALESCE(SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost), 0) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = k.id) < k.max_budget
|
||||
)
|
||||
)
|
||||
"""
|
||||
|
||||
|
||||
class _FunnelTotalsRow(BaseModel):
|
||||
legs_with_rows: int
|
||||
not_sampled: int
|
||||
unjudgeable: int
|
||||
shed: int
|
||||
|
||||
|
||||
class _AttemptCountRow(BaseModel):
|
||||
job_id: str
|
||||
attempt_count: int
|
||||
|
|
@ -937,6 +968,9 @@ def _slices(rows: Sequence[_AttemptAggRow]) -> tuple[ShadowEvalSlice, ...]:
|
|||
shadow_win_rate_pct=_pct_of(row.shadow_wins, row.turn_count),
|
||||
tie_rate_pct=_pct_of(row.ties, row.turn_count),
|
||||
avg_judge_confidence=round(row.avg_confidence or 0.0, 3),
|
||||
real_spend=row.real_spend,
|
||||
shadow_spend=row.shadow_spend,
|
||||
cache_hit_turns=row.cache_hit_turns,
|
||||
)
|
||||
for row in sorted(rows, key=lambda r: r.turn_count, reverse=True)
|
||||
)
|
||||
|
|
@ -1087,12 +1121,20 @@ async def _shadow_eval_results(prisma_client: "PrismaClient", legs: Sequence[_Le
|
|||
for row in by_leg
|
||||
)
|
||||
total_turns: Final = sum(r.turn_count for r in by_tier)
|
||||
funnel_rows: Final = await _query_raw(prisma_client, _FUNNEL_TOTALS_SQL, leg_ids)
|
||||
counted: Final = _FunnelTotalsRow.model_validate(funnel_rows[0]) if funnel_rows else None
|
||||
funnel: Final = counted if counted is not None and counted.legs_with_rows > 0 else None
|
||||
return ShadowEvalResult(
|
||||
by_tier=_slices(by_tier),
|
||||
by_current_model=_slices(by_model),
|
||||
by_key=_slices(by_key),
|
||||
overall_shadow_win_rate_pct=_pct_of(sum(r.shadow_wins for r in by_tier), total_turns),
|
||||
overall_tie_rate_pct=_pct_of(sum(r.ties for r in by_tier), total_turns),
|
||||
sampled_real_spend=sum(r.real_spend for r in by_tier),
|
||||
sampled_shadow_spend=sum(r.shadow_spend for r in by_tier),
|
||||
not_sampled_count=funnel.not_sampled if funnel is not None else None,
|
||||
unjudgeable_count=funnel.unjudgeable if funnel is not None else None,
|
||||
shed_count=funnel.shed if funnel is not None else None,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -1203,6 +1245,19 @@ async def start_shadow_eval(
|
|||
f"A requested key was claimed by another {data.direction} shadow eval job concurrently. Stop it first."
|
||||
),
|
||||
) from e
|
||||
# Seed a zero funnel row per leg NOW: a fully covered job never skips a request, so
|
||||
# waiting for the first skip would leave it indistinguishable from a pre-funnel job
|
||||
# (null coverage). A failed seed degrades this job to exactly that, nothing worse.
|
||||
try:
|
||||
created_legs: Final = await _shadow_eval_jobs(prisma_client).find_many(
|
||||
where={"group_id": group_id}
|
||||
) # mutable-ok: Prisma filter
|
||||
await _shadow_eval_funnel(prisma_client).create_many(
|
||||
data=[{"job_id": str(leg.id)} for leg in created_legs], # mutable-ok: Prisma payload
|
||||
skip_duplicates=True,
|
||||
)
|
||||
except Exception as seed_err: # noqa: BLE001 # coverage is advisory; the job must still start
|
||||
verbose_proxy_logger.error("shadow_eval: funnel seed failed for job %s: %s", group_id, seed_err)
|
||||
labels: Final = MappingProxyType({row.token: row for row in token_rows})
|
||||
return ShadowEvalJobResponse(
|
||||
job_id=group_id,
|
||||
|
|
|
|||
|
|
@ -1527,12 +1527,26 @@ model LiteLLM_ShadowEvalAttempt {
|
|||
confidence Float?
|
||||
judge_cost Float @default(0)
|
||||
shadow_cost Float @default(0)
|
||||
real_cost Float @default(0)
|
||||
real_classifier_cost Float @default(0)
|
||||
shadow_classifier_cost Float @default(0)
|
||||
real_cache_hit Boolean @default(false)
|
||||
error String?
|
||||
created_at DateTime @default(now())
|
||||
|
||||
@@index([job_id])
|
||||
}
|
||||
|
||||
// Per-leg sampling funnel counters the attempt rows cannot derive: requests an
|
||||
// admitting job saw but did not judge. attempted = the leg's attempt rows; the
|
||||
// leg's eligible traffic = not_sampled + unjudgeable + shed + attempted.
|
||||
model LiteLLM_ShadowEvalFunnel {
|
||||
job_id String @id
|
||||
not_sampled Int @default(0)
|
||||
unjudgeable Int @default(0)
|
||||
shed Int @default(0)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Workflow Run Tracking
|
||||
//
|
||||
|
|
|
|||
|
|
@ -6276,7 +6276,9 @@ async def _total_queued_spend_transactions(prisma_client: PrismaClient) -> int:
|
|||
tool_queue_size: Final = len(prisma_client.tool_usage_transactions)
|
||||
async with prisma_client._autorouter_turn_transactions_lock:
|
||||
autorouter_queue_size: Final = len(prisma_client.autorouter_turn_transactions)
|
||||
return spend_queue_size + tool_queue_size + autorouter_queue_size
|
||||
from litellm.proxy.db.shadow_eval_funnel import pending_shadow_eval_funnel_events
|
||||
|
||||
return spend_queue_size + tool_queue_size + autorouter_queue_size + pending_shadow_eval_funnel_events()
|
||||
|
||||
|
||||
async def update_daily_tag_spend(
|
||||
|
|
@ -6418,6 +6420,13 @@ async def update_spend_logs_job(
|
|||
autorouter_tracking_err,
|
||||
)
|
||||
|
||||
try:
|
||||
from litellm.proxy.db.shadow_eval_funnel import flush_shadow_eval_funnel
|
||||
|
||||
await flush_shadow_eval_funnel(prisma_client)
|
||||
except Exception as funnel_err: # noqa: BLE001 # a drain bug must not abort the spend job
|
||||
verbose_proxy_logger.error("Spend tracking - shadow eval funnel drain failed: %s", funnel_err)
|
||||
|
||||
|
||||
MAX_SPEND_LOG_DRAIN_ITERATIONS: Final = 20
|
||||
|
||||
|
|
|
|||
|
|
@ -362,6 +362,27 @@ class ShadowEvalSlice(BaseModel):
|
|||
)
|
||||
tie_rate_pct: float
|
||||
avg_judge_confidence: float
|
||||
real_spend: float = Field(
|
||||
default=0.0,
|
||||
description=(
|
||||
"USD the real arm billed on this slice's judged turns, completion plus its own routing "
|
||||
"classifier when it routed, excluding turns litellm's response cache served for free"
|
||||
),
|
||||
)
|
||||
shadow_spend: float = Field(
|
||||
default=0.0,
|
||||
description=(
|
||||
"USD the shadow arm billed on the same turns, completion plus its own routing classifier, "
|
||||
"excluding the judge and the same cache-served turns, so the two spends compare like for like"
|
||||
),
|
||||
)
|
||||
cache_hit_turns: int = Field(
|
||||
default=0,
|
||||
description=(
|
||||
"Judged turns litellm's response cache served, excluded from both spends: an adopted router "
|
||||
"would be served by the same cache, so those turns cost the same either way"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class ShadowEvalResult(BaseModel):
|
||||
|
|
@ -382,6 +403,29 @@ class ShadowEvalResult(BaseModel):
|
|||
)
|
||||
overall_shadow_win_rate_pct: float
|
||||
overall_tie_rate_pct: float
|
||||
sampled_real_spend: float = Field(
|
||||
default=0.0,
|
||||
description="USD the real arm billed across all judged turns, cache-served turns excluded",
|
||||
)
|
||||
sampled_shadow_spend: float = Field(
|
||||
default=0.0,
|
||||
description="USD the shadow arm billed across the same turns, judge excluded, like for like",
|
||||
)
|
||||
not_sampled_count: int | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Eligible requests the sampling dice skipped, summed over legs: the judged rows stand for "
|
||||
"judged + this many requests. None for jobs from before the funnel existed"
|
||||
),
|
||||
)
|
||||
unjudgeable_count: int | None = Field(
|
||||
default=None,
|
||||
description="Sampled requests whose shape could not be judged (tool-final turn, empty text)",
|
||||
)
|
||||
shed_count: int | None = Field(
|
||||
default=None,
|
||||
description="Sampled requests dropped by the per-pod concurrency cap, so quiet periods are overweighted",
|
||||
)
|
||||
|
||||
|
||||
class ShadowEvalJobKeyResponse(BaseModel):
|
||||
|
|
|
|||
|
|
@ -1527,12 +1527,26 @@ model LiteLLM_ShadowEvalAttempt {
|
|||
confidence Float?
|
||||
judge_cost Float @default(0)
|
||||
shadow_cost Float @default(0)
|
||||
real_cost Float @default(0)
|
||||
real_classifier_cost Float @default(0)
|
||||
shadow_classifier_cost Float @default(0)
|
||||
real_cache_hit Boolean @default(false)
|
||||
error String?
|
||||
created_at DateTime @default(now())
|
||||
|
||||
@@index([job_id])
|
||||
}
|
||||
|
||||
// Per-leg sampling funnel counters the attempt rows cannot derive: requests an
|
||||
// admitting job saw but did not judge. attempted = the leg's attempt rows; the
|
||||
// leg's eligible traffic = not_sampled + unjudgeable + shed + attempted.
|
||||
model LiteLLM_ShadowEvalFunnel {
|
||||
job_id String @id
|
||||
not_sampled Int @default(0)
|
||||
unjudgeable Int @default(0)
|
||||
shed Int @default(0)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Workflow Run Tracking
|
||||
//
|
||||
|
|
|
|||
|
|
@ -76,7 +76,11 @@ def _job_record(job: ActiveShadowEvalJob, api_key_id="key-hash") -> MagicMock:
|
|||
return record
|
||||
|
||||
|
||||
def _router(shadow_text="shadow answer", judge_json='{"preference": "A", "confidence": 0.9, "reasoning": "x"}'):
|
||||
def _router(
|
||||
shadow_text="shadow answer",
|
||||
judge_json='{"preference": "A", "confidence": 0.9, "reasoning": "x"}',
|
||||
classifier_cost=None,
|
||||
):
|
||||
"""One mock router serving the shadow call first, the judge call second, told apart by
|
||||
the internal-origin stamp rather than the model, since a reverse job's shadow arm names
|
||||
a plain model. Only the auto-router writes a routing decision back, and only a plain
|
||||
|
|
@ -90,7 +94,10 @@ def _router(shadow_text="shadow answer", judge_json='{"preference": "A", "confid
|
|||
if kwargs["metadata"].get(INTERNAL_CALL_ORIGIN_METADATA_KEY) != SHADOW_EVAL_ROUTER_CALL_ORIGIN:
|
||||
return {"choices": [{"message": {"content": judge_json}}]}
|
||||
if kwargs["model"] == "my-router":
|
||||
kwargs["metadata"]["routing_decision"] = {"tier_label": "SIMPLE", "routed_model": "cheap-model"}
|
||||
decision = {"tier_label": "SIMPLE", "routed_model": "cheap-model"}
|
||||
if classifier_cost is not None:
|
||||
decision["classifier_cost"] = classifier_cost
|
||||
kwargs["metadata"]["routing_decision"] = decision
|
||||
return {"choices": [{"message": {"content": shadow_text}}], "usage": {"completion_tokens": 5}}
|
||||
return ModelResponse(
|
||||
model=kwargs["model"],
|
||||
|
|
@ -119,14 +126,17 @@ def _spend_counter(store=None):
|
|||
def _logger(router=None, prisma=None, jobs=(), counter_store=None) -> ShadowEvalLogger:
|
||||
cache = InMemoryCache(max_size_in_memory=4, default_ttl=60)
|
||||
counter, read, write = _spend_counter(counter_store)
|
||||
funnel_events = []
|
||||
logger = ShadowEvalLogger(
|
||||
router_provider=lambda: router,
|
||||
prisma_provider=lambda: prisma,
|
||||
jobs_cache=cache,
|
||||
job_spend_reader=read,
|
||||
job_spend_writer=write,
|
||||
funnel_recorder=lambda job_id, stage: funnel_events.append((job_id, stage)),
|
||||
)
|
||||
logger._test_counter = counter
|
||||
logger._test_funnel = funnel_events
|
||||
if jobs:
|
||||
cache.set_cache("shadow_eval:active_jobs", {"key-hash": tuple(jobs)})
|
||||
return logger
|
||||
|
|
@ -138,7 +148,13 @@ def _routed_by(router_name="my-router", tier="COMPLEX"):
|
|||
|
||||
|
||||
def _success_kwargs(
|
||||
request_id="req-1", api_key_hash="key-hash", request_metadata=None, call_type="acompletion", model="claude-opus"
|
||||
request_id="req-1",
|
||||
api_key_hash="key-hash",
|
||||
request_metadata=None,
|
||||
call_type="acompletion",
|
||||
model="claude-opus",
|
||||
response_cost=None,
|
||||
cache_hit=None,
|
||||
):
|
||||
return {
|
||||
"standard_logging_object": {
|
||||
|
|
@ -147,6 +163,8 @@ def _success_kwargs(
|
|||
"model": model,
|
||||
"metadata": {"user_api_key_hash": api_key_hash},
|
||||
"model_parameters": {"temperature": 0.5, "stream": True},
|
||||
"response_cost": response_cost,
|
||||
"cache_hit": cache_hit,
|
||||
},
|
||||
"litellm_params": {"metadata": request_metadata or {}},
|
||||
"messages": [{"role": "user", "content": "what is 2+2"}],
|
||||
|
|
@ -896,6 +914,9 @@ class TestShadowPipeline:
|
|||
messages=({"role": "user", "content": "hi"},),
|
||||
real_text="real answer",
|
||||
real_model="claude-opus",
|
||||
real_cost=0.0,
|
||||
real_classifier_cost=0.0,
|
||||
real_cache_hit=False,
|
||||
control_tier=None,
|
||||
shadow_params={},
|
||||
parent_metadata={},
|
||||
|
|
@ -925,6 +946,9 @@ class TestShadowPipeline:
|
|||
messages=({"role": "user", "content": "hi"},),
|
||||
real_text="real answer",
|
||||
real_model="claude-opus",
|
||||
real_cost=0.0,
|
||||
real_classifier_cost=0.0,
|
||||
real_cache_hit=False,
|
||||
control_tier=None,
|
||||
shadow_params={},
|
||||
parent_metadata={"user_api_key_auth": UserAPIKeyAuth(api_key="sk-abc", max_budget=10.0)},
|
||||
|
|
@ -970,6 +994,9 @@ class TestShadowPipeline:
|
|||
messages=({"role": "user", "content": "hi"},),
|
||||
real_text="real answer",
|
||||
real_model="claude-opus",
|
||||
real_cost=0.0,
|
||||
real_classifier_cost=0.0,
|
||||
real_cache_hit=False,
|
||||
control_tier=None,
|
||||
shadow_params={},
|
||||
parent_metadata={},
|
||||
|
|
@ -997,6 +1024,9 @@ class TestShadowPipeline:
|
|||
messages=({"role": "user", "content": "hi"},),
|
||||
real_text="real answer",
|
||||
real_model="claude-opus",
|
||||
real_cost=0.0,
|
||||
real_classifier_cost=0.0,
|
||||
real_cache_hit=False,
|
||||
control_tier=None,
|
||||
shadow_params={},
|
||||
parent_metadata={},
|
||||
|
|
@ -1029,6 +1059,9 @@ class TestShadowPipeline:
|
|||
messages=({"role": "user", "content": "hi"},),
|
||||
real_text="real answer",
|
||||
real_model="claude-opus",
|
||||
real_cost=0.0,
|
||||
real_classifier_cost=0.0,
|
||||
real_cache_hit=False,
|
||||
control_tier=None,
|
||||
shadow_params={},
|
||||
parent_metadata={},
|
||||
|
|
@ -1058,6 +1091,9 @@ class TestShadowPipeline:
|
|||
messages=({"role": "user", "content": "hi"},),
|
||||
real_text="real answer",
|
||||
real_model="claude-opus",
|
||||
real_cost=0.0,
|
||||
real_classifier_cost=0.0,
|
||||
real_cache_hit=False,
|
||||
control_tier=None,
|
||||
shadow_params={"temperature": 0.2},
|
||||
parent_metadata=parent_metadata,
|
||||
|
|
@ -1271,3 +1307,167 @@ async def test_judge_call_resolves_its_arm_under_the_shadowed_keys_team(monkeypa
|
|||
|
||||
router.acompletion.assert_awaited_once()
|
||||
sdk.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCostComparison:
|
||||
"""The attempt row prices BOTH arms with what each actually billed: the real arm's
|
||||
payload cost plus its own classifier when it routed, the shadow arm's completion plus
|
||||
its write-back classifier cost, and the exact-cache flag that voids the comparison."""
|
||||
|
||||
async def test_success_row_records_both_arms_and_the_classifier(self, monkeypatch: pytest.MonkeyPatch):
|
||||
import litellm as litellm_module
|
||||
|
||||
monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005)
|
||||
router = _router(classifier_cost=0.0007)
|
||||
prisma = _prisma()
|
||||
logger = _logger(router=router, prisma=prisma, jobs=(_job(),))
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(response_cost=0.002), RESPONSE, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
row = prisma.db.litellm_shadowevalattempt.create.call_args.kwargs["data"]
|
||||
assert row["real_cost"] == 0.002
|
||||
assert row["real_classifier_cost"] == 0.0
|
||||
assert row["shadow_classifier_cost"] == 0.0007
|
||||
assert row["real_cache_hit"] is False
|
||||
assert logger._test_funnel == []
|
||||
|
||||
async def test_reverse_job_prices_the_real_arms_classifier(self, monkeypatch: pytest.MonkeyPatch):
|
||||
import litellm as litellm_module
|
||||
|
||||
monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005)
|
||||
router = _router()
|
||||
prisma = _prisma()
|
||||
job = _job(direction="reverse", baseline_model="gpt-4o-mini")
|
||||
logger = _logger(router=router, prisma=prisma, jobs=(job,))
|
||||
metadata = _routed_by()
|
||||
metadata["routing_decision"]["classifier_cost"] = 0.0004
|
||||
|
||||
await logger.async_log_success_event(
|
||||
_success_kwargs(request_metadata=metadata, response_cost=0.003), RESPONSE, None, None
|
||||
)
|
||||
await _drain(logger)
|
||||
|
||||
row = prisma.db.litellm_shadowevalattempt.create.call_args.kwargs["data"]
|
||||
assert row["real_cost"] == 0.003
|
||||
assert row["real_classifier_cost"] == 0.0004
|
||||
assert row["shadow_classifier_cost"] == 0.0
|
||||
|
||||
async def test_shadow_classifier_cost_charges_the_eval_budget_counter(self, monkeypatch: pytest.MonkeyPatch):
|
||||
import litellm as litellm_module
|
||||
|
||||
monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005)
|
||||
logger = _logger(router=_router(classifier_cost=0.0007), prisma=_prisma(), jobs=(_job(),))
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(response_cost=0.002), RESPONSE, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
assert logger._test_counter["spend:shadow_eval:job-1"] == pytest.approx(0.005 + 0.005 + 0.0007)
|
||||
|
||||
async def test_real_cost_never_charges_the_eval_budget_counter(self, monkeypatch: pytest.MonkeyPatch):
|
||||
import litellm as litellm_module
|
||||
|
||||
monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005)
|
||||
logger = _logger(router=_router(), prisma=_prisma(), jobs=(_job(),))
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(response_cost=99.0), RESPONSE, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
assert logger._test_counter["spend:shadow_eval:job-1"] == pytest.approx(0.005 + 0.005)
|
||||
|
||||
async def test_cache_served_turn_is_flagged_on_the_row(self, monkeypatch: pytest.MonkeyPatch):
|
||||
import litellm as litellm_module
|
||||
|
||||
monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005)
|
||||
prisma = _prisma()
|
||||
logger = _logger(router=_router(), prisma=prisma, jobs=(_job(),))
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(response_cost=0.0, cache_hit=True), RESPONSE, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
row = prisma.db.litellm_shadowevalattempt.create.call_args.kwargs["data"]
|
||||
assert row["real_cache_hit"] is True
|
||||
assert row["real_cost"] == 0.0
|
||||
|
||||
async def test_failed_shadow_call_still_records_its_classifier_cost(self, monkeypatch: pytest.MonkeyPatch):
|
||||
import litellm as litellm_module
|
||||
|
||||
monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005)
|
||||
router = _router(classifier_cost=0.0007)
|
||||
|
||||
async def failing_acompletion(**kwargs):
|
||||
if kwargs["metadata"].get(INTERNAL_CALL_ORIGIN_METADATA_KEY) == SHADOW_EVAL_ROUTER_CALL_ORIGIN:
|
||||
kwargs["metadata"]["routing_decision"] = {"tier_label": "SIMPLE", "classifier_cost": 0.0007}
|
||||
raise RuntimeError("provider down")
|
||||
return {"choices": [{"message": {"content": "unused"}}]}
|
||||
|
||||
router.acompletion = MagicMock(side_effect=failing_acompletion)
|
||||
prisma = _prisma()
|
||||
logger = _logger(router=router, prisma=prisma, jobs=(_job(),))
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(response_cost=0.002), RESPONSE, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
row = prisma.db.litellm_shadowevalattempt.create.call_args.kwargs["data"]
|
||||
assert row["outcome"] == "error"
|
||||
assert row["shadow_classifier_cost"] == 0.0007
|
||||
assert row["real_cost"] == 0.002
|
||||
assert logger._test_counter["spend:shadow_eval:job-1"] == pytest.approx(0.0007)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSamplingFunnel:
|
||||
"""Skips an admitting job cannot derive from attempt rows are counted per leg, so the
|
||||
judged rows can be weighed against the eligible traffic they stand for."""
|
||||
|
||||
async def test_a_lost_sampling_dice_roll_counts_not_sampled(self):
|
||||
from litellm.integrations.shadow_eval_logger import _sample_hits
|
||||
|
||||
job = _job(shadow_percentage=1.0)
|
||||
missing_id = next(
|
||||
f"req-miss-{n}" for n in range(10_000) if not _sample_hits(f"req-miss-{n}", job.id, job.shadow_percentage)
|
||||
)
|
||||
prisma = _prisma()
|
||||
logger = _logger(router=_router(), prisma=prisma, jobs=(job,))
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(request_id=missing_id), RESPONSE, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
assert logger._test_funnel == [("job-1", "not_sampled")]
|
||||
prisma.db.litellm_shadowevalattempt.create.assert_not_awaited()
|
||||
|
||||
async def test_an_unjudgeable_sampled_request_counts_unjudgeable(self):
|
||||
prisma = _prisma()
|
||||
logger = _logger(router=_router(), prisma=prisma, jobs=(_job(),))
|
||||
tool_final = {"choices": [{"message": {"content": None, "tool_calls": [{"type": "function", "function": {}}]}}]}
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(), tool_final, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
assert logger._test_funnel == [("job-1", "unjudgeable")]
|
||||
prisma.db.litellm_shadowevalattempt.create.assert_not_awaited()
|
||||
|
||||
async def test_a_concurrency_shed_counts_shed_and_starts_nothing(self):
|
||||
prisma = _prisma()
|
||||
logger = _logger(router=_router(), prisma=prisma, jobs=(_job(),))
|
||||
logger._inflight_shadow_tasks = 16
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None)
|
||||
|
||||
assert logger._test_funnel == [("job-1", "shed")]
|
||||
assert logger._job_starts == {}
|
||||
prisma.db.litellm_shadowevalattempt.create.assert_not_awaited()
|
||||
logger._inflight_shadow_tasks = 0
|
||||
|
||||
async def test_direction_mismatch_and_saturated_jobs_count_nothing(self):
|
||||
prisma = _prisma()
|
||||
saturated = _job(id="job-full", max_turns=1, attempts=1)
|
||||
wrong_direction = _job(id="job-rev", direction="reverse", baseline_model="gpt-4o-mini")
|
||||
logger = _logger(router=_router(), prisma=prisma, jobs=(saturated, wrong_direction))
|
||||
|
||||
await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None)
|
||||
await _drain(logger)
|
||||
|
||||
assert logger._test_funnel == []
|
||||
prisma.db.litellm_shadowevalattempt.create.assert_not_awaited()
|
||||
|
|
|
|||
|
|
@ -218,8 +218,20 @@ class TestFlush:
|
|||
sql, params = client.db.calls[0]
|
||||
assert sql == UPSERT_AUTOROUTER_SESSION_SQL
|
||||
assert params == (
|
||||
"k1", "s1", "live-auto", "complexity", "bedrock/haiku",
|
||||
"2026-08-01T12:00:00", 100, 0.01, 0.02, 1, 0, None, 0, "medium",
|
||||
"k1",
|
||||
"s1",
|
||||
"live-auto",
|
||||
"complexity",
|
||||
"bedrock/haiku",
|
||||
"2026-08-01T12:00:00",
|
||||
100,
|
||||
0.01,
|
||||
0.02,
|
||||
1,
|
||||
0,
|
||||
None,
|
||||
0,
|
||||
"medium",
|
||||
)
|
||||
|
||||
def test_a_connect_error_retries_the_same_statement(self):
|
||||
|
|
@ -275,7 +287,12 @@ def test_every_drain_trigger_reads_the_one_queue_census_owner():
|
|||
from litellm.proxy import utils as proxy_utils
|
||||
|
||||
owner_source = inspect.getsource(proxy_utils._total_queued_spend_transactions)
|
||||
for queue in ("spend_log_transactions", "tool_usage_transactions", "autorouter_turn_transactions"):
|
||||
for queue in (
|
||||
"spend_log_transactions",
|
||||
"tool_usage_transactions",
|
||||
"autorouter_turn_transactions",
|
||||
"pending_shadow_eval_funnel_events",
|
||||
):
|
||||
assert queue in owner_source, queue
|
||||
for site in (proxy_utils.update_spend, proxy_utils.update_spend_logs_job, proxy_utils._monitor_spend_logs_queue):
|
||||
assert "_total_queued_spend_transactions" in inspect.getsource(site), site.__name__
|
||||
|
|
|
|||
95
tests/test_litellm/proxy/db/test_shadow_eval_funnel.py
Normal file
95
tests/test_litellm/proxy/db/test_shadow_eval_funnel.py
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.proxy.db import shadow_eval_funnel
|
||||
from litellm.proxy.db.shadow_eval_funnel import (
|
||||
flush_shadow_eval_funnel,
|
||||
record_shadow_eval_funnel_event,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_queue():
|
||||
shadow_eval_funnel._pending.clear()
|
||||
yield
|
||||
shadow_eval_funnel._pending.clear()
|
||||
|
||||
|
||||
def _prisma() -> MagicMock:
|
||||
prisma = MagicMock()
|
||||
prisma.db.execute_raw = AsyncMock(return_value=1)
|
||||
return prisma
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_increments_aggregate_per_job_and_flush_upserts_and_clears():
|
||||
record_shadow_eval_funnel_event("leg-1", "not_sampled")
|
||||
record_shadow_eval_funnel_event("leg-1", "not_sampled")
|
||||
record_shadow_eval_funnel_event("leg-1", "shed")
|
||||
record_shadow_eval_funnel_event("leg-2", "unjudgeable")
|
||||
prisma = _prisma()
|
||||
|
||||
await flush_shadow_eval_funnel(prisma)
|
||||
|
||||
calls = {call.args[1]: call.args[2:] for call in prisma.db.execute_raw.await_args_list}
|
||||
assert calls == {"leg-1": (2, 0, 1), "leg-2": (0, 1, 0)}
|
||||
sql = prisma.db.execute_raw.await_args_list[0].args[0]
|
||||
assert "ON CONFLICT (job_id) DO UPDATE" in sql
|
||||
assert '"LiteLLM_ShadowEvalFunnel".not_sampled + EXCLUDED.not_sampled' in sql
|
||||
assert shadow_eval_funnel._pending == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_queue_touches_nothing():
|
||||
prisma = _prisma()
|
||||
|
||||
await flush_shadow_eval_funnel(prisma)
|
||||
|
||||
prisma.db.execute_raw.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_failed_upsert_drops_only_that_legs_batch():
|
||||
record_shadow_eval_funnel_event("leg-bad", "not_sampled")
|
||||
record_shadow_eval_funnel_event("leg-good", "shed")
|
||||
prisma = _prisma()
|
||||
|
||||
async def execute_raw(sql, job_id, *counts):
|
||||
if job_id == "leg-bad":
|
||||
raise RuntimeError("db down")
|
||||
return 1
|
||||
|
||||
prisma.db.execute_raw = AsyncMock(side_effect=execute_raw)
|
||||
|
||||
await flush_shadow_eval_funnel(prisma)
|
||||
|
||||
flushed = [call.args[1] for call in prisma.db.execute_raw.await_args_list]
|
||||
assert set(flushed) == {"leg-bad", "leg-good"}
|
||||
assert shadow_eval_funnel._pending == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_events_recorded_during_a_flush_survive_into_the_next_batch():
|
||||
record_shadow_eval_funnel_event("leg-1", "not_sampled")
|
||||
prisma = _prisma()
|
||||
|
||||
async def execute_raw(sql, job_id, *counts):
|
||||
record_shadow_eval_funnel_event("leg-2", "shed")
|
||||
return 1
|
||||
|
||||
prisma.db.execute_raw = AsyncMock(side_effect=execute_raw)
|
||||
|
||||
await flush_shadow_eval_funnel(prisma)
|
||||
|
||||
assert shadow_eval_funnel._pending == {"leg-2": {"not_sampled": 0, "unjudgeable": 0, "shed": 1}}
|
||||
|
||||
|
||||
def test_pending_count_feeds_the_drain_census():
|
||||
from litellm.proxy.db.shadow_eval_funnel import pending_shadow_eval_funnel_events
|
||||
|
||||
assert pending_shadow_eval_funnel_events() == 0
|
||||
record_shadow_eval_funnel_event("leg-1", "not_sampled")
|
||||
record_shadow_eval_funnel_event("leg-1", "shed")
|
||||
record_shadow_eval_funnel_event("leg-2", "unjudgeable")
|
||||
assert pending_shadow_eval_funnel_events() == 3
|
||||
|
|
@ -857,12 +857,8 @@ def _shadow_router() -> Router:
|
|||
_complexity_router_deployment(
|
||||
"my-router", {"SIMPLE": "cheap", "MEDIUM": "mid", "COMPLEX": "pricey"}, "mid"
|
||||
),
|
||||
_complexity_router_deployment(
|
||||
"sonnet-router", {"SIMPLE": "cheap", "MEDIUM": "house-sonnet"}, "cheap"
|
||||
),
|
||||
_complexity_router_deployment(
|
||||
"classifier-router", {"SIMPLE": "cheap"}, "cheap", classifier="pricey"
|
||||
),
|
||||
_complexity_router_deployment("sonnet-router", {"SIMPLE": "cheap", "MEDIUM": "house-sonnet"}, "cheap"),
|
||||
_complexity_router_deployment("classifier-router", {"SIMPLE": "cheap"}, "cheap", classifier="pricey"),
|
||||
_complexity_router_deployment("b-team-router", {"SIMPLE": "cheap", "MEDIUM": "b-tier"}, "cheap"),
|
||||
_complexity_router_deployment("prefixed-router", {"SIMPLE": "prefixed-tier"}, "prefixed-tier"),
|
||||
_complexity_router_deployment("bare-router", {"SIMPLE": "bare-tier"}, "bare-tier"),
|
||||
|
|
@ -1001,6 +997,7 @@ def _shadow_prisma(
|
|||
prisma.db.litellm_shadowevaljob.create_many = AsyncMock(return_value=1)
|
||||
prisma.db.litellm_shadowevaljob.update_many = AsyncMock(return_value=1)
|
||||
prisma.db.litellm_shadowevalattempt.find_first = AsyncMock(return_value=None)
|
||||
prisma.db.litellm_shadowevalfunnel.create_many = AsyncMock(return_value=1)
|
||||
prisma.attempt_rows = []
|
||||
|
||||
async def query_raw(sql: str, *params: object):
|
||||
|
|
@ -1014,8 +1011,11 @@ def _shadow_prisma(
|
|||
return [{"judged_count": 10, "error_count": 2, "judge_spend": 0.031}]
|
||||
if "SELECT job_id AS grp" in sql:
|
||||
return by_leg_rows if by_leg_rows is not None else []
|
||||
if 'FROM "LiteLLM_ShadowEvalFunnel"' in sql:
|
||||
return prisma.funnel_rows
|
||||
return agg_rows if agg_rows is not None else []
|
||||
|
||||
prisma.funnel_rows = []
|
||||
prisma.db.query_raw = AsyncMock(side_effect=query_raw)
|
||||
return prisma
|
||||
|
||||
|
|
@ -1053,7 +1053,7 @@ async def test_start_shadow_eval_writes_one_leg_per_key_in_one_statement(monkeyp
|
|||
assert ">= j.max_turns" in sweep_sql
|
||||
assert "j.max_budget IS NOT NULL" in sweep_sql
|
||||
assert ">= j.max_budget" in sweep_sql
|
||||
assert "SUM(a.judge_cost + a.shadow_cost)" in sweep_sql
|
||||
assert "SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost)" in sweep_sql
|
||||
assert "j.api_key_id = ANY($1::text[])" in sweep_sql
|
||||
assert sweep_keys == ["key-hash", "key-hash-2"]
|
||||
prisma.db.litellm_shadowevaljob.create_many.assert_awaited_once()
|
||||
|
|
@ -1330,12 +1330,52 @@ async def test_get_shadow_eval_job_pools_counts_and_slices_results_per_key(monke
|
|||
import litellm.proxy.proxy_server as proxy_server
|
||||
|
||||
tier_rows = [
|
||||
{"grp": "SIMPLE", "turn_count": 8, "real_wins": 2, "shadow_wins": 4, "ties": 2, "avg_confidence": 0.8},
|
||||
{"grp": "REASONING", "turn_count": 2, "real_wins": 2, "shadow_wins": 0, "ties": 0, "avg_confidence": 0.9},
|
||||
{
|
||||
"grp": "SIMPLE",
|
||||
"turn_count": 8,
|
||||
"real_wins": 2,
|
||||
"shadow_wins": 4,
|
||||
"ties": 2,
|
||||
"avg_confidence": 0.8,
|
||||
"real_spend": 0.08,
|
||||
"shadow_spend": 0.02,
|
||||
"cache_hit_turns": 1,
|
||||
},
|
||||
{
|
||||
"grp": "REASONING",
|
||||
"turn_count": 2,
|
||||
"real_wins": 2,
|
||||
"shadow_wins": 0,
|
||||
"ties": 0,
|
||||
"avg_confidence": 0.9,
|
||||
"real_spend": 0.04,
|
||||
"shadow_spend": 0.05,
|
||||
"cache_hit_turns": 0,
|
||||
},
|
||||
]
|
||||
leg_rows = [
|
||||
{"grp": "leg-1", "turn_count": 6, "real_wins": 1, "shadow_wins": 4, "ties": 1, "avg_confidence": 0.7},
|
||||
{"grp": "leg-2", "turn_count": 4, "real_wins": 3, "shadow_wins": 0, "ties": 1, "avg_confidence": 0.6},
|
||||
{
|
||||
"grp": "leg-1",
|
||||
"turn_count": 6,
|
||||
"real_wins": 1,
|
||||
"shadow_wins": 4,
|
||||
"ties": 1,
|
||||
"avg_confidence": 0.7,
|
||||
"real_spend": 0.07,
|
||||
"shadow_spend": 0.03,
|
||||
"cache_hit_turns": 0,
|
||||
},
|
||||
{
|
||||
"grp": "leg-2",
|
||||
"turn_count": 4,
|
||||
"real_wins": 3,
|
||||
"shadow_wins": 0,
|
||||
"ties": 1,
|
||||
"avg_confidence": 0.6,
|
||||
"real_spend": 0.05,
|
||||
"shadow_spend": 0.04,
|
||||
"cache_hit_turns": 1,
|
||||
},
|
||||
]
|
||||
prisma = _shadow_prisma(
|
||||
legs=[_leg_record(), _leg_record(id="leg-2", api_key_id="key-hash-2", max_turns=50)],
|
||||
|
|
@ -1359,6 +1399,14 @@ async def test_get_shadow_eval_job_pools_counts_and_slices_results_per_key(monke
|
|||
assert response.results.overall_tie_rate_pct == 20.0
|
||||
assert [(s.group, s.turn_count) for s in response.results.by_key] == [("key-hash", 6), ("key-hash-2", 4)]
|
||||
assert response.results.by_key[0].shadow_win_rate_pct == 66.7
|
||||
assert response.results.by_tier[0].real_spend == 0.08
|
||||
assert response.results.by_tier[0].shadow_spend == 0.02
|
||||
assert response.results.by_tier[0].cache_hit_turns == 1
|
||||
assert response.results.sampled_real_spend == pytest.approx(0.12)
|
||||
assert response.results.sampled_shadow_spend == pytest.approx(0.07)
|
||||
assert response.results.not_sampled_count is None
|
||||
assert response.results.unjudgeable_count is None
|
||||
assert response.results.shed_count is None
|
||||
assert [(key.api_key_id, key.max_turns) for key in response.keys] == [("key-hash", 200), ("key-hash-2", 50)]
|
||||
totals_args = [call.args for call in prisma.db.query_raw.await_args_list if "judged_count" in call.args[0]]
|
||||
assert totals_args == [(totals_args[0][0], ["leg-1", "leg-2"])]
|
||||
|
|
@ -1726,7 +1774,7 @@ async def test_stop_shadow_eval_stops_every_unstopped_leg_and_rejects_non_runnin
|
|||
assert ") < k.max_turns" in stop_sql
|
||||
assert "k.max_budget IS NULL" in stop_sql
|
||||
assert ") < k.max_budget" in stop_sql
|
||||
assert "SUM(a.judge_cost + a.shadow_cost)" in stop_sql
|
||||
assert "SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost)" in stop_sql
|
||||
assert (stop_group, stop_operator) == ("job-1", "admin")
|
||||
assert datetime.fromisoformat(stop_stamp).tzinfo is None
|
||||
assert prisma.db.execute_raw.await_count == 1
|
||||
|
|
@ -2008,9 +2056,7 @@ async def test_start_shadow_eval_sees_a_collision_hidden_behind_the_second_teams
|
|||
monkeypatch.setattr(proxy_server, "llm_router", _shadow_router())
|
||||
|
||||
monkeypatch.setattr(proxy_server, "prisma_client", _shadow_prisma(key_teams={"key-hash": "team-a"}))
|
||||
accepted = await start_shadow_eval(
|
||||
_start_request(router_name="b-team-router", judge_model="house-sonnet"), ADMIN
|
||||
)
|
||||
accepted = await start_shadow_eval(_start_request(router_name="b-team-router", judge_model="house-sonnet"), ADMIN)
|
||||
assert accepted.job_id
|
||||
|
||||
monkeypatch.setattr(
|
||||
|
|
@ -2076,3 +2122,65 @@ async def test_start_shadow_eval_matches_a_prefixed_judge_name_to_a_bare_tier_de
|
|||
assert exc.value.status_code == 400
|
||||
assert "bare-tier" in str(exc.value.detail)
|
||||
prisma.db.litellm_shadowevaljob.create_many.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_shadow_eval_job_sums_funnel_rows_across_legs(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Legs with funnel rows sum into job-level coverage counts; a job with no funnel
|
||||
rows at all reports None rather than a fabricated zero."""
|
||||
import litellm.proxy.proxy_server as proxy_server
|
||||
|
||||
tier_rows = [
|
||||
{
|
||||
"grp": "SIMPLE",
|
||||
"turn_count": 4,
|
||||
"real_wins": 1,
|
||||
"shadow_wins": 2,
|
||||
"ties": 1,
|
||||
"avg_confidence": 0.8,
|
||||
"real_spend": 0.05,
|
||||
"shadow_spend": 0.02,
|
||||
"cache_hit_turns": 0,
|
||||
},
|
||||
]
|
||||
prisma = _shadow_prisma(
|
||||
legs=[_leg_record(), _leg_record(id="leg-2", api_key_id="key-hash-2")],
|
||||
agg_rows=tier_rows,
|
||||
)
|
||||
prisma.funnel_rows = [{"legs_with_rows": 2, "not_sampled": 30, "unjudgeable": 5, "shed": 2}]
|
||||
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
|
||||
|
||||
response = await get_shadow_eval_job("job-1", VIEWER)
|
||||
|
||||
assert response.results.not_sampled_count == 30
|
||||
assert response.results.unjudgeable_count == 5
|
||||
assert response.results.shed_count == 2
|
||||
funnel_args = [call.args for call in prisma.db.query_raw.await_args_list if "ShadowEvalFunnel" in call.args[0]]
|
||||
assert funnel_args == [(funnel_args[0][0], ["leg-1", "leg-2"])]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_shadow_eval_seeds_a_zero_funnel_row_per_leg(monkeypatch: pytest.MonkeyPatch):
|
||||
"""A fully covered job never records a skip, so only a row seeded at creation
|
||||
separates 'nothing was skipped' from a job predating the funnel."""
|
||||
import litellm.proxy.proxy_server as proxy_server
|
||||
|
||||
stored_legs: list = []
|
||||
prisma = _shadow_prisma(legs=stored_legs)
|
||||
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
|
||||
monkeypatch.setattr(proxy_server, "llm_router", _shadow_router())
|
||||
|
||||
async def create_many(data):
|
||||
stored_legs.extend(
|
||||
_leg_record(id=f"leg-{row['api_key_id']}", api_key_id=row["api_key_id"], group_id=row["group_id"])
|
||||
for row in data
|
||||
)
|
||||
return len(data)
|
||||
|
||||
prisma.db.litellm_shadowevaljob.create_many = AsyncMock(side_effect=create_many)
|
||||
|
||||
await start_shadow_eval(_start_request(api_key_ids=("key-hash", "key-hash-2")), ADMIN)
|
||||
|
||||
seeded = prisma.db.litellm_shadowevalfunnel.create_many.call_args.kwargs
|
||||
assert sorted(row["job_id"] for row in seeded["data"]) == ["leg-key-hash", "leg-key-hash-2"]
|
||||
assert seeded["skip_duplicates"] is True
|
||||
|
|
|
|||
|
|
@ -100,6 +100,9 @@ const job = (overrides: Partial<ShadowEvalJob> = {}): ShadowEvalJob => ({
|
|||
shadow_win_rate_pct: 55.0,
|
||||
tie_rate_pct: 25.0,
|
||||
avg_judge_confidence: 0.81,
|
||||
real_spend: 0.4,
|
||||
shadow_spend: 0.1,
|
||||
cache_hit_turns: 2,
|
||||
},
|
||||
{
|
||||
group: "REASONING",
|
||||
|
|
@ -108,6 +111,9 @@ const job = (overrides: Partial<ShadowEvalJob> = {}): ShadowEvalJob => ({
|
|||
shadow_win_rate_pct: 33.3,
|
||||
tie_rate_pct: 16.7,
|
||||
avg_judge_confidence: 0.74,
|
||||
real_spend: 0.2,
|
||||
shadow_spend: 0.2,
|
||||
cache_hit_turns: 0,
|
||||
},
|
||||
],
|
||||
by_current_model: [
|
||||
|
|
@ -118,11 +124,19 @@ const job = (overrides: Partial<ShadowEvalJob> = {}): ShadowEvalJob => ({
|
|||
shadow_win_rate_pct: 45.0,
|
||||
tie_rate_pct: 25.0,
|
||||
avg_judge_confidence: 0.8,
|
||||
real_spend: 0.6,
|
||||
shadow_spend: 0.3,
|
||||
cache_hit_turns: 2,
|
||||
},
|
||||
],
|
||||
by_key: [],
|
||||
overall_shadow_win_rate_pct: 48.0,
|
||||
overall_tie_rate_pct: 22.0,
|
||||
sampled_real_spend: 0.6,
|
||||
sampled_shadow_spend: 0.3,
|
||||
not_sampled_count: 378,
|
||||
unjudgeable_count: 10,
|
||||
shed_count: 2,
|
||||
},
|
||||
created_at: "2026-08-07T00:00:00Z",
|
||||
ends_at: "2026-09-07T00:00:00Z",
|
||||
|
|
@ -496,10 +510,15 @@ describe("ShadowEvalSection", () => {
|
|||
shadow_win_rate_pct: 60.0,
|
||||
tie_rate_pct: 20.0,
|
||||
avg_judge_confidence: 0.9,
|
||||
real_spend: 0.9,
|
||||
shadow_spend: 0.5,
|
||||
cache_hit_turns: 0,
|
||||
},
|
||||
],
|
||||
overall_shadow_win_rate_pct: 60.0,
|
||||
overall_tie_rate_pct: 20.0,
|
||||
sampled_real_spend: 0.9,
|
||||
sampled_shadow_spend: 0.5,
|
||||
},
|
||||
}),
|
||||
],
|
||||
|
|
@ -590,6 +609,28 @@ describe("ShadowEvalSection", () => {
|
|||
expect(within(hungry).queryByText("running")).not.toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("shows the measured cost comparison with savings, both arm totals, and the sampling coverage", () => {
|
||||
const j = job({});
|
||||
mockHooks({ jobs: [j], detailsById: { "job-1": j } });
|
||||
render(<ShadowEvalSection />);
|
||||
expect(screen.getByText("Router cost vs your current model")).toBeInTheDocument();
|
||||
expect(screen.getByText("-50.0%")).toBeInTheDocument();
|
||||
expect(screen.getByText(/\$0\.3000 vs \$0\.6000 on the same judged turns/)).toBeInTheDocument();
|
||||
expect(screen.getByText(/2 cache-served turns excluded/)).toBeInTheDocument();
|
||||
expect(screen.getByText(/Measured on 42 of 433 eligible requests/)).toBeInTheDocument();
|
||||
expect(screen.getByText(/2 dropped under load/)).toBeInTheDocument();
|
||||
expect(screen.getAllByText("Router cost").length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("flips the cost comparison arms for a reverse job", () => {
|
||||
const reverse = job({ direction: "reverse", baseline_model: "gpt-4o-mini" });
|
||||
mockHooks({ jobs: [reverse], detailsById: { "job-1": reverse } });
|
||||
render(<ShadowEvalSection />);
|
||||
expect(screen.getByText("Router cost vs the baseline")).toBeInTheDocument();
|
||||
expect(screen.getByText(/\$0\.6000 vs \$0\.3000 on the same judged turns/)).toBeInTheDocument();
|
||||
expect(screen.getByText("+100.0%")).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("keeps an older job's verdicts reachable through the previous evaluations list", async () => {
|
||||
const user = userEvent.setup();
|
||||
const emptyOverrides: Partial<ShadowEvalJob> = {
|
||||
|
|
|
|||
|
|
@ -43,6 +43,18 @@ const routerWinRate = (direction: ShadowEvalDirection, slice: ShadowEvalSlice):
|
|||
const otherArmWinRate = (direction: ShadowEvalDirection, slice: ShadowEvalSlice): number =>
|
||||
direction === "reverse" ? slice.shadow_win_rate_pct : slice.real_win_rate_pct;
|
||||
|
||||
const routerArmSpend = (direction: ShadowEvalDirection, results: NonNullable<ShadowEvalJob["results"]>): number =>
|
||||
direction === "reverse" ? results.sampled_real_spend : results.sampled_shadow_spend;
|
||||
|
||||
const otherArmSpend = (direction: ShadowEvalDirection, results: NonNullable<ShadowEvalJob["results"]>): number =>
|
||||
direction === "reverse" ? results.sampled_shadow_spend : results.sampled_real_spend;
|
||||
|
||||
const routerSliceSpend = (direction: ShadowEvalDirection, slice: ShadowEvalSlice): number =>
|
||||
direction === "reverse" ? slice.real_spend : slice.shadow_spend;
|
||||
|
||||
const otherSliceSpend = (direction: ShadowEvalDirection, slice: ShadowEvalSlice): number =>
|
||||
direction === "reverse" ? slice.shadow_spend : slice.real_spend;
|
||||
|
||||
const routerMatchedOrBeatPct = (
|
||||
direction: ShadowEvalDirection,
|
||||
results: NonNullable<ShadowEvalJob["results"]>,
|
||||
|
|
@ -122,13 +134,19 @@ const SliceTable: React.FC<{
|
|||
<TableHeader>
|
||||
<TableRow>
|
||||
<TableHead>{groupHeader}</TableHead>
|
||||
{["Judged turns", "Router wins", `${otherArmLabel(direction)} wins`, "Ties", "Judge confidence"].map(
|
||||
(label) => (
|
||||
<TableHead key={label} className="text-right">
|
||||
{label}
|
||||
</TableHead>
|
||||
),
|
||||
)}
|
||||
{[
|
||||
"Judged turns",
|
||||
"Router wins",
|
||||
`${otherArmLabel(direction)} wins`,
|
||||
"Ties",
|
||||
"Judge confidence",
|
||||
"Router cost",
|
||||
`${otherArmLabel(direction)} cost`,
|
||||
].map((label) => (
|
||||
<TableHead key={label} className="text-right">
|
||||
{label}
|
||||
</TableHead>
|
||||
))}
|
||||
</TableRow>
|
||||
</TableHeader>
|
||||
<TableBody>
|
||||
|
|
@ -147,12 +165,54 @@ const SliceTable: React.FC<{
|
|||
<TableCell className="text-right tabular-nums">{pct(otherArmWinRate(direction, slice))}</TableCell>
|
||||
<TableCell className="text-right tabular-nums">{pct(slice.tie_rate_pct)}</TableCell>
|
||||
<TableCell className="text-right tabular-nums">{slice.avg_judge_confidence.toFixed(2)}</TableCell>
|
||||
<TableCell className="text-right tabular-nums">{usd(routerSliceSpend(direction, slice))}</TableCell>
|
||||
<TableCell className="text-right tabular-nums">{usd(otherSliceSpend(direction, slice))}</TableCell>
|
||||
</TableRow>
|
||||
))}
|
||||
</TableBody>
|
||||
</Table>
|
||||
);
|
||||
|
||||
const CostComparison: React.FC<{
|
||||
direction: ShadowEvalDirection;
|
||||
results: NonNullable<ShadowEvalJob["results"]>;
|
||||
errorCount: number;
|
||||
}> = ({ direction, results, errorCount }) => {
|
||||
const routerSpend = routerArmSpend(direction, results);
|
||||
const otherSpend = otherArmSpend(direction, results);
|
||||
if (routerSpend <= 0 && otherSpend <= 0) return null;
|
||||
const savingsPct = otherSpend > 0 ? ((otherSpend - routerSpend) / otherSpend) * 100 : null;
|
||||
const cacheHits = results.by_tier.reduce((sum, slice) => sum + slice.cache_hit_turns, 0);
|
||||
const judged = results.by_tier.reduce((sum, slice) => sum + slice.turn_count, 0);
|
||||
const eligible =
|
||||
results.not_sampled_count != null
|
||||
? judged + errorCount + results.not_sampled_count + (results.unjudgeable_count ?? 0) + (results.shed_count ?? 0)
|
||||
: null;
|
||||
return (
|
||||
<div className="flex flex-col gap-1 border-b px-6 py-4">
|
||||
<p className="text-[11px] uppercase tracking-wide text-muted-foreground">
|
||||
Router cost vs {direction === "reverse" ? "the baseline" : "your current model"}
|
||||
</p>
|
||||
<p
|
||||
className={`text-3xl font-semibold ${savingsPct != null && savingsPct > 0 ? "text-success" : "text-foreground"}`}
|
||||
>
|
||||
{savingsPct != null ? `${savingsPct > 0 ? "-" : "+"}${Math.abs(savingsPct).toFixed(1)}%` : "n/a"}
|
||||
</p>
|
||||
<p className="text-xs text-muted-foreground">
|
||||
{usd(routerSpend)} vs {usd(otherSpend)} on the same judged turns, each arm priced as completion plus its own
|
||||
routing classifier{cacheHits > 0 ? `; ${cacheHits.toLocaleString()} cache-served turns excluded` : ""}
|
||||
</p>
|
||||
{eligible != null && (
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Measured on {judged.toLocaleString()} of {eligible.toLocaleString()} eligible requests
|
||||
{(results.shed_count ?? 0) > 0 ? ` (${results.shed_count} dropped under load)` : ""}; projections beyond the
|
||||
sampled slice are extrapolation
|
||||
</p>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
};
|
||||
|
||||
const VerdictBar: React.FC<{ direction: ShadowEvalDirection; results: NonNullable<ShadowEvalJob["results"]> }> = ({
|
||||
direction,
|
||||
results,
|
||||
|
|
@ -276,6 +336,7 @@ const ResultsBody: React.FC<{ job: ShadowEvalJob; resultsError?: boolean }> = ({
|
|||
of {(job.judged_count ?? 0).toLocaleString()} judged responses
|
||||
</p>
|
||||
</div>
|
||||
<CostComparison direction={job.direction} results={results} errorCount={job.error_count ?? 0} />
|
||||
<VerdictBar direction={job.direction} results={results} />
|
||||
{results.by_current_model.length > 0 && (
|
||||
<SliceTable
|
||||
|
|
|
|||
45
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
45
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -34798,10 +34798,37 @@ export interface components {
|
|||
by_key: components["schemas"]["ShadowEvalSlice"][];
|
||||
/** By Tier */
|
||||
by_tier: components["schemas"]["ShadowEvalSlice"][];
|
||||
/**
|
||||
* Not Sampled Count
|
||||
* @description Eligible requests the sampling dice skipped, summed over legs: the judged rows stand for judged + this many requests. None for jobs from before the funnel existed
|
||||
*/
|
||||
not_sampled_count?: number | null;
|
||||
/** Overall Shadow Win Rate Pct */
|
||||
overall_shadow_win_rate_pct: number;
|
||||
/** Overall Tie Rate Pct */
|
||||
overall_tie_rate_pct: number;
|
||||
/**
|
||||
* Sampled Real Spend
|
||||
* @description USD the real arm billed across all judged turns, cache-served turns excluded
|
||||
* @default 0
|
||||
*/
|
||||
sampled_real_spend: number;
|
||||
/**
|
||||
* Sampled Shadow Spend
|
||||
* @description USD the shadow arm billed across the same turns, judge excluded, like for like
|
||||
* @default 0
|
||||
*/
|
||||
sampled_shadow_spend: number;
|
||||
/**
|
||||
* Shed Count
|
||||
* @description Sampled requests dropped by the per-pod concurrency cap, so quiet periods are overweighted
|
||||
*/
|
||||
shed_count?: number | null;
|
||||
/**
|
||||
* Unjudgeable Count
|
||||
* @description Sampled requests whose shape could not be judged (tool-final turn, empty text)
|
||||
*/
|
||||
unjudgeable_count?: number | null;
|
||||
};
|
||||
/**
|
||||
* ShadowEvalSlice
|
||||
|
|
@ -34811,13 +34838,31 @@ export interface components {
|
|||
ShadowEvalSlice: {
|
||||
/** Avg Judge Confidence */
|
||||
avg_judge_confidence: number;
|
||||
/**
|
||||
* Cache Hit Turns
|
||||
* @description Judged turns litellm's response cache served, excluded from both spends: an adopted router would be served by the same cache, so those turns cost the same either way
|
||||
* @default 0
|
||||
*/
|
||||
cache_hit_turns: number;
|
||||
/** Group */
|
||||
group: string;
|
||||
/**
|
||||
* Real Spend
|
||||
* @description USD the real arm billed on this slice's judged turns, completion plus its own routing classifier when it routed, excluding turns litellm's response cache served for free
|
||||
* @default 0
|
||||
*/
|
||||
real_spend: number;
|
||||
/**
|
||||
* Real Win Rate Pct
|
||||
* @description Share of judged turns the real arm won, meaning the response the caller actually received: the key's own model in forward mode, the router's pick in reverse
|
||||
*/
|
||||
real_win_rate_pct: number;
|
||||
/**
|
||||
* Shadow Spend
|
||||
* @description USD the shadow arm billed on the same turns, completion plus its own routing classifier, excluding the judge and the same cache-served turns, so the two spends compare like for like
|
||||
* @default 0
|
||||
*/
|
||||
shadow_spend: number;
|
||||
/**
|
||||
* Shadow Win Rate Pct
|
||||
* @description Share of judged turns the shadow arm won, meaning the duplicated response nobody was served: the router's pick in forward mode, baseline_model in reverse
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue