diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260811125249_add_shadow_eval/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260811125249_add_shadow_eval/migration.sql new file mode 100644 index 000000000000..63e8a0d9c5a8 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260811125249_add_shadow_eval/migration.sql @@ -0,0 +1,55 @@ +-- CreateTable +CREATE TABLE "LiteLLM_ShadowEvalJob" ( + "id" TEXT NOT NULL, + "api_key_id" TEXT NOT NULL, + "router_name" TEXT NOT NULL, + "shadow_percentage" DOUBLE PRECISION NOT NULL, + "judge_model" TEXT NOT NULL, + "status" TEXT NOT NULL DEFAULT 'pending', + "request_count" INTEGER NOT NULL DEFAULT 0, + "completed_count" INTEGER NOT NULL DEFAULT 0, + "failed_count" INTEGER NOT NULL DEFAULT 0, + "last_error" TEXT, + "cost_estimate" DOUBLE PRECISION, + "cost_actual" DOUBLE PRECISION NOT NULL DEFAULT 0, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "created_by" TEXT, + "ends_at" TIMESTAMP(3), + "completed_at" TIMESTAMP(3), + + CONSTRAINT "LiteLLM_ShadowEvalJob_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "LiteLLM_ShadowEvalVerdict" ( + "id" TEXT NOT NULL, + "job_id" TEXT NOT NULL, + "request_id" TEXT NOT NULL, + "tier_classification" TEXT, + "real_model" TEXT NOT NULL, + "shadow_model" TEXT NOT NULL, + "judge_preference" TEXT NOT NULL, + "judge_confidence" DOUBLE PRECISION, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "LiteLLM_ShadowEvalVerdict_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "LiteLLM_ShadowEvalJob_api_key_id_status_idx" ON "LiteLLM_ShadowEvalJob"("api_key_id", "status"); + +-- CreateIndex +CREATE INDEX "LiteLLM_ShadowEvalJob_status_idx" ON "LiteLLM_ShadowEvalJob"("status"); + +-- CreateIndex +CREATE INDEX "LiteLLM_ShadowEvalJob_created_at_idx" ON "LiteLLM_ShadowEvalJob"("created_at"); + +-- CreateIndex +CREATE INDEX "LiteLLM_ShadowEvalVerdict_job_id_idx" ON "LiteLLM_ShadowEvalVerdict"("job_id"); + +-- One active job per key, enforced by the database rather than a read-then-create in +-- the start endpoint, which races against a concurrent start on another pod. Partial +-- indexes are not expressible in schema.prisma, so this lives here only. +CREATE UNIQUE INDEX "LiteLLM_ShadowEvalJob_one_active_per_key" + ON "LiteLLM_ShadowEvalJob"("api_key_id") WHERE status IN ('pending', 'running'); + diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index 854602f53804..6e0ba43f9676 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -1450,6 +1450,50 @@ model LiteLLM_AutoRouterSession { @@index([last_turn_at], map: "idx_autorouter_session_last_turn") } +// Shadow Eval: pre-adoption evaluation of an auto-router against a key's live traffic. +// A sampled slice of requests is duplicated through the router in a detached task and +// an LLM judge compares real vs shadow responses blind; verdicts stratify by tier. +model LiteLLM_ShadowEvalJob { + id String @id @default(cuid()) + api_key_id String // the hashed virtual key whose traffic is shadowed + router_name String // the auto-router config to shadow through + shadow_percentage Float + judge_model String + + status String @default("pending") // pending | running | completed + + request_count Int @default(0) // requests seen on the key while active + completed_count Int @default(0) // verdicts written + failed_count Int @default(0) // shadow or judge calls that errored + last_error String? + + cost_estimate Float? // upfront judge-spend estimate shown at start + cost_actual Float @default(0) // running judge-call spend + + created_at DateTime @default(now()) + created_by String? + ends_at DateTime? + completed_at DateTime? + + @@index([api_key_id, status]) + @@index([status]) + @@index([created_at]) +} + +model LiteLLM_ShadowEvalVerdict { + id String @id @default(cuid()) + job_id String + request_id String // the judged real request + tier_classification String? // the router's tier for the prompt, when classified + real_model String // model that actually served the request + shadow_model String // model the router picked + judge_preference String // real | shadow | tie + judge_confidence Float? + created_at DateTime @default(now()) + + @@index([job_id]) +} + // --------------------------------------------------------------------------- // Workflow Run Tracking // diff --git a/litellm/integrations/shadow_eval_logger.py b/litellm/integrations/shadow_eval_logger.py new file mode 100644 index 000000000000..7deac938fc61 --- /dev/null +++ b/litellm/integrations/shadow_eval_logger.py @@ -0,0 +1,669 @@ +"""Shadow Eval Logger: pre-adoption evaluation of an auto-router against live traffic. + +For each successful request on a key with an active shadow-eval job, a sampled slice of +requests is duplicated through the auto-router (the user never sees the shadow response), +an LLM judge compares the two responses blind with A/B labels randomized, and the verdict +is stored stratified by the router's own tier classification. + +The whole pipeline runs in a detached background task; the success hook itself is one +dict lookup against a job snapshot. Shadow and judge calls carry the shadowed key's +identity metadata (their provider spend bills to that key) and an +``internal_call_origin`` stamp, which the hook also skips on, so the logger can never +recurse on its own traffic. A shadow/judge pair is skipped outright if the shadowed key +or its team is already at or over budget. + +Job lifecycle (counter flushes, snapshot refresh, stopping a job at its ``ends_at`` or +judge-spend cap) runs on a periodic loop owned by logger registration, never on the +request path: an idle key's job still ends on schedule, and the final counter batch +lands without needing another request to arrive. +""" + +import asyncio +import hashlib +import random +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from types import MappingProxyType +from typing import TYPE_CHECKING, Final + +from pydantic import BaseModel + +from litellm._logging import verbose_logger +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.internal_call_metadata import sanitized_forwardable_call_metadata +from litellm.litellm_core_utils.llm_judge import ( + default_router_provider, + extract_text_from_content, + judge_acompletion, + parse_json_verdict, +) +from litellm.litellm_core_utils.redact_messages import should_redact_message_logging +from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN + +if TYPE_CHECKING: + from litellm.proxy.utils import PrismaClient + from litellm.router import Router + from litellm.types.utils import StandardLoggingPayload + +# Cadence of the lifecycle loop: snapshot refresh, counter flush, and job finalization +# all run on this tick, so a job starting or stopping takes up to one tick to be noticed. +_LIFECYCLE_TICK_SECONDS: Final = 10.0 + +# Concurrent shadow+judge pipelines per pod: a traffic spike turns into skipped samples +# rather than an unbounded task pileup. +_MAX_CONCURRENT_SHADOW_TASKS: Final = 16 + +# Total character budget for the judge's user prompt, however long the conversation and +# the two responses are, so the prompt can never overflow a judge model's context window. +_MAX_JUDGE_RESPONSE_CHARS: Final = 8_000 +_MAX_JUDGE_PROMPT_CHARS: Final = 24_000 + +# The judge answers with a small JSON object; a tighter budget truncates the JSON +# mid-object and the verdict is lost to failed_count. +JUDGE_MAX_OUTPUT_TOKENS: Final = 500 + +_MAX_LAST_ERROR_CHARS: Final = 500 + +# A job stops sampling once its judge spend reaches this multiple of the estimate shown +# at start. The headroom absorbs an estimate that undershot the real traffic mix; the +# floor keeps a cent-sized estimate from stopping a job on its first verdict. cost_actual +# is read from the job snapshot, so overshoot is bounded by one lifecycle tick. +_SPEND_CAP_MULTIPLIER: Final = 1.5 +_SPEND_CAP_FLOOR_USD: Final = 1.0 + +_EMPTY_METADATA: Final[Mapping[str, object]] = MappingProxyType({}) + +PAIRWISE_JUDGE_SYSTEM_PROMPT: Final = """You are an impartial quality judge comparing two responses to the same conversation. + +The responses are labeled A and B in random order. You do not know which system produced which. + +Criteria: correctness, completeness, clarity, conciseness. + +Return ONLY valid JSON in this exact format, no other text: +{ + "preference": "A" | "B" | "tie", + "confidence": <0.0 to 1.0>, + "reasoning": "" +}""" + + +class PairwiseVerdict(BaseModel): + """The judge's blind A/B verdict, validated at the parse boundary.""" + + preference: str = "tie" + confidence: float = 0.0 + + +def _sample_hits(request_id: str, job_id: str, percentage: float) -> bool: + """Deterministically decide whether a request falls in the shadowed slice: hash-based + rather than random so retries sample the same way and pods agree without coordination.""" + digest: Final = hashlib.sha256(f"{job_id}:{request_id}".encode()).digest() + bucket: Final = int.from_bytes(digest[:8], "big") / float(2**64) + return bucket * 100.0 < percentage + + +def _judge_call_cost(response: object) -> float: + """Price a judge call, treating an unmapped judge model as free rather than fatal.""" + import litellm + + try: + return litellm.completion_cost(completion_response=response) or 0.0 + except Exception: # noqa: BLE001 # unmapped judge model: verdict still counts, cost stays 0 + return 0.0 + + +def _unmask_preference(raw_preference: str, real_is_a: bool) -> str: + """Map the judge's blind A/B/tie verdict back to real/shadow/tie.""" + normalized: Final = raw_preference.strip().lower() + if normalized == "a": + return "real" if real_is_a else "shadow" + if normalized == "b": + return "shadow" if real_is_a else "real" + return "tie" + + +def _judge_user_prompt(conversation: str, response_a: str, response_b: str) -> str: + """The judge prompt under one total character budget: each response is capped, and + the conversation tail gets whatever budget the responses left over.""" + a: Final = response_a[:_MAX_JUDGE_RESPONSE_CHARS] + b: Final = response_b[:_MAX_JUDGE_RESPONSE_CHARS] + conversation_budget: Final = _MAX_JUDGE_PROMPT_CHARS - len(a) - len(b) + return ( + f"Conversation:\n{conversation[-conversation_budget:]}\n\n" + f"Response A:\n{a}\n\n" + f"Response B:\n{b}\n\n" + "Which response is better?" + ) + + +@dataclass(frozen=True, slots=True) +class _CallFailure: + """A shadow or judge call that produced no usable response, with why.""" + + error: str + + +@dataclass(frozen=True, slots=True) +class _ShadowResponse: + """A successful shadow call, with what the verdict row records.""" + + text: str + model: str + tier: str | None + + +@dataclass(frozen=True, slots=True) +class _JudgeVerdict: + """A parsed judge verdict, unmasked back to real/shadow/tie.""" + + preference: str + confidence: float + cost: float + + +@dataclass(frozen=True, slots=True) +class ActiveShadowEvalJob: + """The subset of a shadow-eval job row the request path actually needs.""" + + id: str + router_name: str + shadow_percentage: float + judge_model: str + status: str + cost_estimate: float | None = None + cost_actual: float = 0.0 + ends_at: datetime | None = None + + +def _as_utc(value: object) -> datetime | None: + if not isinstance(value, datetime): + return None + return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value + + +def _job_is_past_its_end(job: ActiveShadowEvalJob) -> bool: + """An eval prices a fixed window at start time, so sampling past ends_at would bill + traffic the estimate never covered.""" + return job.ends_at is not None and datetime.now(timezone.utc) >= job.ends_at + + +def _job_is_over_spend_cap(job: ActiveShadowEvalJob) -> bool: + """Budgets bound what the key may spend; this bounds what a single eval may spend + even under a generous budget, so a bad estimate or a traffic spike cannot quietly + turn a small eval into a much larger bill.""" + if job.cost_estimate is None: + return False + return job.cost_actual >= max(job.cost_estimate * _SPEND_CAP_MULTIPLIER, _SPEND_CAP_FLOOR_USD) + + +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. + + Advisory and fail-open: real traffic on an over-budget key is already rejected at + auth (so nothing reaches the success hook), and this gate only closes the race + where the key crosses its budget while a request is in flight. + """ + try: + from litellm.exceptions import BudgetExceededError + from litellm.proxy._types import UserAPIKeyAuth + from litellm.proxy.auth.auth_checks import ( + _team_max_budget_check, + _virtual_key_max_budget_check, + get_team_object, + ) + from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache + except ImportError: + return False + + auth: Final = metadata.get("user_api_key_auth") + if not isinstance(auth, UserAPIKeyAuth): + return False + try: + await _virtual_key_max_budget_check(valid_token=auth, proxy_logging_obj=proxy_logging_obj) + if auth.team_id: + team: Final = await get_team_object( + team_id=auth.team_id, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + check_cache_only=True, + ) + await _team_max_budget_check(team_object=team, valid_token=auth, proxy_logging_obj=proxy_logging_obj) + except BudgetExceededError: + return True + except Exception as e: # noqa: BLE001 # advisory gate: a failed read must not block sampling + verbose_logger.debug("shadow_eval: budget read failed: %s", e) + return False + + +def _request_was_routed_by(request_metadata: Mapping[str, object], router_name: str) -> bool: + """Duplicating a request the shadowed router already served compares the router to + itself: guaranteed ties, judge spend for zero information.""" + decision: Final = request_metadata.get("routing_decision") + if not isinstance(decision, Mapping): + return False + return decision.get("router_model_name") == router_name + + +class ShadowEvalLogger(CustomLogger): + """Fires blind pairwise shadow evaluations for keys with an active shadow-eval job.""" + + def __init__( + self, + router_provider: Callable[[], "Router | None"] | None = None, + prisma_provider: Callable[[], "PrismaClient | None"] | None = None, + ) -> None: + """Providers are callables so the proxy's lazily-initialized globals are resolved + at call time, not at logger construction.""" + self._router_provider = router_provider or default_router_provider + self._prisma_provider = prisma_provider or _default_prisma_provider + # Snapshot of every active job, keyed by shadowed api_key_id, refreshed as a + # whole by the lifecycle loop: one find_many per pod per tick keeps DB load flat + # no matter how many distinct keys the proxy serves. + self._jobs_by_key: dict[str, ActiveShadowEvalJob] = {} # mutable-ok: loop-refreshed snapshot + self._inflight_shadow_tasks: int = 0 + self._pending_seen: dict[str, int] = {} # mutable-ok: flush buffer + self._lifecycle_task: asyncio.Task[None] | None = None + + def start_lifecycle_loop(self) -> None: + """Idempotently start the loop that owns job lifecycle off the request path. + Without it, jobs never finalize and counters never flush, so a caller outside a + running event loop gets a warning rather than a silent no-op.""" + if self._lifecycle_task is not None and not self._lifecycle_task.done(): + return + try: + self._lifecycle_task = asyncio.create_task(self._lifecycle_loop()) + except RuntimeError: + verbose_logger.warning( + "shadow_eval: no running event loop; lifecycle loop not started, jobs will not finalize on this process" + ) + + async def _lifecycle_loop(self) -> None: + while True: + try: + await self._lifecycle_tick() + except Exception as e: # noqa: BLE001 # the loop must survive any single tick failing + verbose_logger.debug("shadow_eval: lifecycle tick failed: %s", e) + await asyncio.sleep(_LIFECYCLE_TICK_SECONDS) + + async def _lifecycle_tick(self) -> None: + """Flush counters while jobs are still active, refresh, then finalize, so an + expiring job's last counter batch lands while its row still passes the + active-status guard.""" + await self._flush_seen_counts() + await self._refresh_active_jobs() + for job in tuple(self._jobs_by_key.values()): + if _job_is_past_its_end(job): + await self._finalize_job(job, "reached its scheduled end") + elif _job_is_over_spend_cap(job): + await self._finalize_job( + job, + f"spend ${job.cost_actual:.4f} reached the cap for its ${job.cost_estimate or 0.0:.4f} estimate", + ) + + #### hook #### + + async def async_log_success_event( + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: object, + end_time: object, + ) -> None: + try: + payload: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object") # pyright: ignore[reportAssignmentType] # untyped callback kwargs + if payload is None: + return + metadata: Final = payload.get("metadata") or _EMPTY_METADATA + litellm_params: Final = kwargs.get("litellm_params") + raw_request_metadata: Final = ( + litellm_params.get("metadata") if isinstance(litellm_params, Mapping) else None + ) + request_metadata: Final = ( + raw_request_metadata if isinstance(raw_request_metadata, Mapping) else _EMPTY_METADATA + ) + if request_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY): + return # internal sub-call (our own shadow/judge, a classifier), not user traffic + api_key_hash: Final = metadata.get("user_api_key_hash") + if not api_key_hash: + return + job: Final = self._jobs_by_key.get(str(api_key_hash)) + if job is None: + return + if _job_is_past_its_end(job) or _job_is_over_spend_cap(job): + return # stop sampling now; the lifecycle loop finalizes the row + request_id: Final = payload.get("id") or "" + if not request_id: + return + # The job tracks every request it saw, sampled or not, so the UI can show + # "N of M requests shadowed". Flushed by the lifecycle loop. + self._pending_seen[job.id] = self._pending_seen.get(job.id, 0) + 1 + if not _sample_hits(request_id, job.id, job.shadow_percentage): + return + if payload.get("call_type") not in ("completion", "acompletion"): + return # only known chat-shaped traffic is comparable; unknown or missing types fail closed + if _request_was_routed_by(request_metadata, job.router_name): + return + if self._inflight_shadow_tasks >= _MAX_CONCURRENT_SHADOW_TASKS: + return + # Redaction rewrites the logged messages and response before callbacks run, + # so a redacted request offers this hook only placeholders: evaluating them + # would produce garbage verdicts, and the caller opted that content out of + # logging anyway. The redactor's own predicate decides, so every redaction + # source (dynamic param, headers, global setting) is honored. + if should_redact_message_logging(dict(kwargs)): # mutable-ok: predicate takes a plain dict + return + raw_messages: Final = kwargs.get("messages") + self._inflight_shadow_tasks += 1 + task: Final = asyncio.create_task( + self._run_shadow_eval( + job=job, + request_id=request_id, + messages=tuple(m for m in raw_messages if isinstance(m, Mapping)) + if isinstance(raw_messages, Sequence) + else (), + response_obj=response_obj, + real_model=payload.get("model") or "", + model_parameters=MappingProxyType( + dict(payload.get("model_parameters") or {}) # mutable-ok: frozen snapshot + ), + parent_metadata=MappingProxyType(dict(request_metadata)), # mutable-ok: frozen snapshot + ) + ) + task.add_done_callback(self._release_shadow_slot) + except Exception as e: # noqa: BLE001 # logging hooks must never fail the request + verbose_logger.debug("shadow_eval: failed to schedule task: %s", e) + + def _release_shadow_slot(self, _task: "asyncio.Task[None]") -> None: + self._inflight_shadow_tasks -= 1 + + #### job lifecycle #### + + async def _refresh_active_jobs(self) -> None: + """Reload the active-job set. On a DB blip the stale snapshot is kept and the + next tick retries, so a blip degrades freshness rather than turning the feature off.""" + prisma: Final = self._prisma_provider() + if prisma is None: + return + try: + records: Final = await prisma.db.litellm_shadowevaljob.find_many( + where={"status": {"in": ["pending", "running"]}}, # mutable-ok: Prisma filter + order={"created_at": "desc"}, # mutable-ok: Prisma order + ) + except Exception as e: # noqa: BLE001 # a DB blip must not break request logging + verbose_logger.debug("shadow_eval: active-job refresh failed: %s", e) + return + jobs_by_key: Final[dict[str, ActiveShadowEvalJob]] = {} # mutable-ok: building the new snapshot + for record in reversed(records or []): + jobs_by_key[str(record.api_key_id)] = ActiveShadowEvalJob( + id=str(record.id), + router_name=str(record.router_name), + shadow_percentage=float(record.shadow_percentage), + judge_model=str(record.judge_model), + status=str(record.status), + cost_estimate=float(record.cost_estimate) if record.cost_estimate is not None else None, + cost_actual=float(record.cost_actual or 0.0), + ends_at=_as_utc(getattr(record, "ends_at", None)), + ) + self._jobs_by_key = jobs_by_key # mutable-ok: atomic snapshot swap + + async def _finalize_job(self, job: ActiveShadowEvalJob, reason: str) -> None: + """Flip a finished job to completed, keeping the verdicts it already produced. + Guarded on the job still being active so two pods finishing the same job cannot + resurrect one an admin stopped in between.""" + prisma: Final = self._prisma_provider() + if prisma is None: + return + self._jobs_by_key = { # mutable-ok: atomic snapshot swap + k: v for k, v in self._jobs_by_key.items() if v.id != job.id + } + verbose_logger.info("shadow_eval: stopping job %s: %s", job.id, reason) + try: + await prisma.db.litellm_shadowevaljob.update_many( + where={ # mutable-ok: Prisma filter + "id": job.id, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, + data={ # mutable-ok: Prisma payload + "status": "completed", + "completed_at": datetime.now(timezone.utc), + }, + ) + except Exception as e: # noqa: BLE001 # the lifecycle loop must survive a failed write + verbose_logger.debug("shadow_eval: failed to stop job %s: %s", job.id, e) + + async def _flush_seen_counts(self) -> None: + """Write the buffered request counts, guarded on the job still being active, so + stopping a job freezes its counter: a pod on a stale snapshot keeps buffering for + up to one tick, and this write drops those increments instead of growing a + stopped job's request_count.""" + prisma: Final = self._prisma_provider() + if prisma is None or not self._pending_seen: + return + pending: Final = self._pending_seen + self._pending_seen = {} # mutable-ok: fresh flush buffer + for job_id, count in pending.items(): + try: + await prisma.db.litellm_shadowevaljob.update_many( + where={ # mutable-ok: Prisma filter + "id": job_id, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, + data={"request_count": {"increment": count}}, # mutable-ok: Prisma payload + ) + except Exception as e: # noqa: BLE001 # counter drift is acceptable; failing the loop is not + verbose_logger.debug("shadow_eval: request_count flush failed: %s", e) + + #### the shadow pipeline #### + + async def _run_shadow_eval( + self, + job: ActiveShadowEvalJob, + request_id: str, + messages: Sequence[Mapping[str, object]], + response_obj: object, + real_model: str, + model_parameters: Mapping[str, object], + parent_metadata: Mapping[str, object], + ) -> None: + """Detached background task: budget gate -> shadow call -> blind judge -> verdict. + + The prisma gate sits above the shadow and judge dispatch so no provider spend + happens without a place to record the verdict, and the budget read lives here, + not in the success hook, because get_current_spend can fall back to an + authoritative DB read that the production callback must not absorb. + """ + prisma: Final = self._prisma_provider() + try: + if prisma is None: + return + real_text: Final = self._extract_response_text(response_obj) + if not real_text or not messages: + return + if await _key_or_team_is_over_budget(parent_metadata): + return + + shadow: Final = await self._call_router_shadow(job.router_name, messages, model_parameters, parent_metadata) + if isinstance(shadow, _CallFailure): + await self._bump_failed(job.id, shadow.error) + return + + verdict: Final = await self._call_judge( + judge_model=job.judge_model, + messages=messages, + real_text=real_text, + shadow_text=shadow.text, + parent_metadata=parent_metadata, + ) + if isinstance(verdict, _CallFailure): + await self._bump_failed(job.id, verdict.error) + return + # One transaction records the outcome, so every pipeline lands in exactly + # one bucket and a job's counts always match its stored verdicts: the + # status-guarded counter update decides whether the verdict lands (a job + # stopped mid-flight matches zero rows and stores nothing), and a failed + # verdict write rolls the counters back before the outer handler files the + # pipeline under failed_count. + async with prisma.tx() as transaction: + counted: Final = await transaction.litellm_shadowevaljob.update_many( + where={ # mutable-ok: Prisma filter + "id": job.id, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, + data={ # mutable-ok: Prisma payload + "completed_count": {"increment": 1}, # mutable-ok: Prisma operator + "cost_actual": {"increment": verdict.cost}, # mutable-ok: Prisma operator + "status": "running", + }, + ) + if counted: + await transaction.litellm_shadowevalverdict.create( + data={ # mutable-ok: Prisma payload + "job_id": job.id, + "request_id": request_id, + "tier_classification": shadow.tier, + "real_model": real_model, + "shadow_model": shadow.model, + "judge_preference": verdict.preference, + "judge_confidence": verdict.confidence, + } + ) + except Exception as e: # noqa: BLE001 # detached task: log, count, never raise + verbose_logger.debug("shadow_eval: pipeline failed for %s: %s", request_id, e) + await self._bump_failed(job.id, f"pipeline error: {e}") + + async def _bump_failed(self, job_id: str, error: str) -> None: + prisma: Final = self._prisma_provider() + if prisma is None: + return + try: + await prisma.db.litellm_shadowevaljob.update_many( + where={ # mutable-ok: Prisma filter + "id": job_id, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, + data={ # mutable-ok: Prisma payload + "failed_count": {"increment": 1}, # mutable-ok: Prisma operator + "last_error": error[:_MAX_LAST_ERROR_CHARS], + }, + ) + except Exception as e: # noqa: BLE001 # counter drift is acceptable + verbose_logger.debug("shadow_eval: failed_count increment failed: %s", e) + + async def _call_router_shadow( + self, + router_name: str, + messages: Sequence[Mapping[str, object]], + model_parameters: Mapping[str, object], + parent_metadata: Mapping[str, object], + ) -> "_ShadowResponse | _CallFailure": + """Send the prompt through the auto-router being evaluated. The metadata carries + the shadowed key's identity (spend attribution) and receives the router's routing + decision write-back, read back for tier attribution.""" + router: Final = self._router_provider() + if router is None: + return _CallFailure("no router configured on this pod") + shadow_metadata: Final[dict[str, object]] = ( # mutable-ok: router writes its routing decision back + sanitized_forwardable_call_metadata( # mutable-ok: router writes back + parent_metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN + ) + ) + shadow_params: Final = { # mutable-ok: splatted as kwargs + k: v for k, v in model_parameters.items() if k not in ("stream", "metadata") + } + try: + response: Final = await router.acompletion( + model=router_name, + messages=messages, # pyright: ignore[reportArgumentType] # snapshot of the SDK's own message dicts + metadata=shadow_metadata, + num_retries=0, + fallbacks=[], # mutable-ok: SDK kwarg; a failed shadow is a counted miss, never a spend multiplier + **shadow_params, + ) + except Exception as e: # noqa: BLE001 # provider errors are a counted failure, not a crash + verbose_logger.debug("shadow_eval: router call failed: %s", e) + return _CallFailure(f"shadow router call failed: {e}") + text: Final = self._extract_response_text(response) + if not text: + return _CallFailure("shadow router returned an empty response") + raw_decision: Final = shadow_metadata.get("routing_decision") + routing_decision: Final = raw_decision if isinstance(raw_decision, Mapping) else _EMPTY_METADATA + raw_tier: Final = routing_decision.get("tier_label") or routing_decision.get("tier") + return _ShadowResponse( + text=text, + model=str(getattr(response, "model", None) or routing_decision.get("routed_model") or ""), + tier=str(raw_tier) if raw_tier is not None else None, + ) + + async def _call_judge( + self, + judge_model: str, + messages: Sequence[Mapping[str, object]], + real_text: str, + shadow_text: str, + parent_metadata: Mapping[str, object], + ) -> "_JudgeVerdict | _CallFailure": + """Blind pairwise judge with A/B labels randomized to cancel position bias.""" + real_is_a: Final = random.random() < 0.5 + response_a: Final = real_text if real_is_a else shadow_text + response_b: Final = shadow_text if real_is_a else real_text + + conversation: Final = "\n".join( + f"{str(m.get('role', 'user')).upper()}: {extract_text_from_content(m.get('content'))}" + for m in messages + if m.get("content") is not None + ) + judge_metadata: Final = sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_JUDGE_CALL_ORIGIN) + judge_messages: Final = [ # mutable-ok: SDK takes a list + {"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT}, # mutable-ok: SDK message + { + "role": "user", + "content": _judge_user_prompt(conversation, response_a, response_b), + }, # mutable-ok: SDK message + ] + try: + response: Final = await judge_acompletion( + self._router_provider(), + judge_model, + judge_messages, # pyright: ignore[reportArgumentType] # plain SDK message dicts + temperature=0, + max_tokens=JUDGE_MAX_OUTPUT_TOKENS, + metadata=judge_metadata, + ) + except Exception as e: # noqa: BLE001 # judge outages are a counted failure, not a crash + verbose_logger.debug("shadow_eval: judge call failed: %s", e) + return _CallFailure(f"judge call failed: {e}") + try: + raw: Final = response["choices"][0]["message"]["content"] or "" + verdict: Final = PairwiseVerdict.model_validate(parse_json_verdict(raw)) + except Exception as e: # noqa: BLE001 # malformed verdicts are a counted failure + verbose_logger.debug("shadow_eval: unparseable judge verdict: %s", e) + return _CallFailure(f"unparseable judge verdict: {e}") + return _JudgeVerdict( + preference=_unmask_preference(verdict.preference, real_is_a), + confidence=max(0.0, min(1.0, verdict.confidence)), + cost=_judge_call_cost(response), + ) + + @staticmethod + def _extract_response_text(response_obj: object) -> str: + """Extract the assistant's text from a ModelResponse-shaped object or dict.""" + try: + content: Final = ( + response_obj["choices"][0]["message"]["content"] + if isinstance(response_obj, Mapping) + else response_obj.choices[0].message.content # pyright: ignore[reportAttributeAccessIssue] # duck-typed ModelResponse + ) + except (AttributeError, KeyError, IndexError, TypeError): + return "" + return extract_text_from_content(content) + + +def _default_prisma_provider() -> "PrismaClient | None": + try: + from litellm.proxy.proxy_server import prisma_client + except ImportError: + return None + return prisma_client diff --git a/litellm/litellm_core_utils/internal_call_metadata.py b/litellm/litellm_core_utils/internal_call_metadata.py new file mode 100644 index 000000000000..6815727de691 --- /dev/null +++ b/litellm/litellm_core_utils/internal_call_metadata.py @@ -0,0 +1,94 @@ +"""Metadata a request forwards to the internal LLM sub-calls it triggers. + +Internal features (the auto-router's classifier and embeddings, shadow eval's shadow and +judge calls) bill real provider spend that nobody typed a prompt for. That spend must land +on the same key/team/org/user as the request that caused it, so the sub-call carries the +caller's identity metadata, minus two things that must never be forwarded as-is: + +* ``user_api_key_budget_reservation`` (and the reservation nested inside + ``user_api_key_auth``) belongs to the parent completion. If a sub-call's cost callback + sees it, that callback finalizes the reservation and the parent's own callback then + skips incrementing the key/team budget counters, losing the parent's spend. + ``user_api_key_auth`` itself is kept, sanitized, because model access-group filtering + needs it. +* The sub-call is stamped with ``INTERNAL_CALL_ORIGIN_METADATA_KEY`` so its spend log row + records that it is not traffic the caller sent. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Final + +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY +from litellm.types.utils import InternalCallOrigin + +BUDGET_RESERVATION_METADATA_KEYS: Final = frozenset({"user_api_key_budget_reservation"}) + +_USER_API_KEY_AUTH_KEY: Final = "user_api_key_auth" + +FORWARDABLE_IDENTITY_METADATA_KEYS: Final = frozenset( + { + "user_api_key", + "user_api_key_hash", + "user_api_key_alias", + "user_api_key_team_id", + "user_api_key_org_id", + "user_api_key_user_id", + "user_api_key_end_user_id", + _USER_API_KEY_AUTH_KEY, + } +) +"""The caller-identity subset a detached sub-call needs to be attributed and +budget-checked like the request that spawned it. Everything else on the parent's metadata +(routing decision, guardrail state, logging payload) describes the parent call and would +be a lie on a sub-call that runs after it returned.""" + + +def sanitize_user_api_key_auth(auth: object) -> object: + """Copy of the auth object with its budget reservation removed; the cost callback + falls back to reading the reservation from inside the auth object.""" + if isinstance(auth, dict): + return {k: v for k, v in auth.items() if k != "budget_reservation"} # mutable-ok: SDK metadata value + reservation: Final[object] = getattr(auth, "budget_reservation", None) + model_copy: Final[object] = getattr(auth, "model_copy", None) + if reservation is not None and callable(model_copy): + return model_copy(update={"budget_reservation": None}) # mutable-ok: pydantic update payload + return auth + + +def _sanitized(parent_metadata: Mapping[str, object]) -> dict[str, object]: # mutable-ok: SDK metadata kwarg + return { # mutable-ok: SDK metadata kwarg + k: sanitize_user_api_key_auth(v) if k == _USER_API_KEY_AUTH_KEY else v + for k, v in parent_metadata.items() + if k not in BUDGET_RESERVATION_METADATA_KEYS + } + + +def forwarded_internal_call_metadata( + parent_metadata: Mapping[str, object] | None, + call_origin: InternalCallOrigin, +) -> dict[str, object]: # mutable-ok: SDK metadata kwarg + """Parent metadata, minus its budget reservation, stamped with the sub-call's origin. + + For sub-calls made inside the parent request (classifier, embeddings), where the + parent's full context still describes the call being made. + """ + if not parent_metadata: + return {} # mutable-ok: SDK metadata kwarg + return _sanitized(parent_metadata) | { # mutable-ok: SDK metadata kwarg + INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin + } + + +def sanitized_forwardable_call_metadata( + parent_metadata: Mapping[str, object], + call_origin: InternalCallOrigin, +) -> dict[str, object]: # mutable-ok: SDK metadata kwarg + """Just the caller's identity, stamped with the sub-call's origin. + + For sub-calls detached from the parent request (shadow eval), which outlive it and + must not inherit per-request state such as its routing decision or logging payload. + """ + identity: Final = {k: v for k, v in parent_metadata.items() if k in FORWARDABLE_IDENTITY_METADATA_KEYS} + return _sanitized(identity) | {INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin} # mutable-ok: SDK metadata kwarg diff --git a/litellm/litellm_core_utils/llm_judge.py b/litellm/litellm_core_utils/llm_judge.py new file mode 100644 index 000000000000..51908f06338f --- /dev/null +++ b/litellm/litellm_core_utils/llm_judge.py @@ -0,0 +1,84 @@ +"""Shared primitives for LLM-judge features (llm_as_a_judge guardrail, shadow eval).""" + +from __future__ import annotations + +import json +import re +from typing import TYPE_CHECKING, Final + +import litellm + +if TYPE_CHECKING: + from litellm import Router + from litellm.types.llms.openai import AllMessageValues + from litellm.types.utils import ModelResponse + +JSON_FENCE_RE: Final = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE) + + +def default_router_provider() -> Router | None: + try: + from litellm.proxy.proxy_server import llm_router + except ImportError: + return None + + return llm_router + + +def parse_json_verdict(raw: str) -> dict[str, object]: # mutable-ok: plain parsed-JSON payload + """Parse a judge's JSON verdict, tolerating markdown fences and surrounding prose.""" + text = raw.strip() # rebind-ok: progressively narrowed to the JSON payload + fenced: Final = JSON_FENCE_RE.search(text) + if fenced is not None: + text = fenced.group(1).strip() # rebind-ok: progressively narrowed to the JSON payload + parsed: object + try: + parsed = json.loads(text) + except json.JSONDecodeError: + start: Final = text.find("{") + end: Final = text.rfind("}") + if start == -1 or end <= start: + raise + parsed = json.loads(text[start : end + 1]) + if not isinstance(parsed, dict): + raise ValueError("judge response is not a JSON object") + return {str(k): v for k, v in parsed.items()} # mutable-ok: plain parsed-JSON payload + + +def extract_text_from_content(content: object) -> str: + """Return plain text from a message content field (str or multimodal list).""" + if isinstance(content, str): + return content + if isinstance(content, list): + return " ".join( + str(part.get("text", "")) for part in content if isinstance(part, dict) and part.get("type") == "text" + ) + return "" + + +def router_resolves_model(router: Router | None, model: str) -> bool: + """Whether the model name resolves through the proxy's router (configured deployment + or model-group alias), the same check the judge dispatch itself makes, so start-time + validation cannot accept a name the call path then fails on.""" + return router is not None and bool(model in router.model_group_alias or router.get_model_list(model_name=model)) + + +async def judge_acompletion( + router: Router | None, + judge_model: str, + messages: list[AllMessageValues], # mutable-ok: the SDK acompletion signature takes a list + **params: object, +) -> ModelResponse: + """Dispatch a judge call through the proxy's router when the judge model is a + configured deployment (DB-stored credentials work), through the SDK for + provider-qualified public names. The router path never retries or falls back: + a failed judge call is the caller's counted failure, not a spend multiplier.""" + if router_resolves_model(router, judge_model): + return await router.acompletion( # pyright: ignore[reportOptionalMemberAccess] # router_resolves_model implies router is not None + model=judge_model, + messages=messages, + num_retries=0, + fallbacks=[], + **params, + ) + return await litellm.acompletion(model=judge_model, messages=messages, num_retries=0, **params) diff --git a/litellm/proxy/db/autorouter_session_rollup.py b/litellm/proxy/db/autorouter_session_rollup.py index 9732b1d74027..96192b884d83 100644 --- a/litellm/proxy/db/autorouter_session_rollup.py +++ b/litellm/proxy/db/autorouter_session_rollup.py @@ -24,6 +24,7 @@ from typing import TYPE_CHECKING, Final, NamedTuple from litellm._logging import verbose_proxy_logger +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.proxy._types import DB_RETRY_SAFE_ERROR_TYPES if TYPE_CHECKING: @@ -180,12 +181,17 @@ def build_autorouter_turn_transaction( The routing_decision record is what says a request was auto-routed at all, so a request without one (including the auto-router's own classifier sub-calls) never - reaches the rollup. Failed requests served nothing and are excluded. Cache facts - are derived from the payload's own usage record through the savings owner, never - handed in beside it. + reaches the rollup. Internal sub-calls that DO carry one (a shadow eval's duplicate + of a request through the router) are excluded by their internal_call_origin stamp: + they are not traffic a user sent, so counting them would manufacture sessions and + savings in the adoption metrics. Failed requests served nothing and are excluded. + Cache facts are derived from the payload's own usage record through the savings + owner, never handed in beside it. """ if payload.get("status") != "success": return None + if metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY): + return None routing_decision: Final = metadata.get("routing_decision") if not isinstance(routing_decision, Mapping) or not routing_decision: return None diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py index b0130db232ad..b2b72c1cac42 100644 --- a/litellm/proxy/db/db_spend_update_writer.py +++ b/litellm/proxy/db/db_spend_update_writer.py @@ -21,6 +21,7 @@ from litellm.constants import ( DB_DAILY_TAG_SPEND_UPDATE_JOB_NAME, DB_SPEND_UPDATE_JOB_NAME, + INTERNAL_CALL_ORIGIN_METADATA_KEY, ) from litellm.litellm_core_utils.safe_json_loads import safe_json_loads from litellm.proxy._types import ( @@ -1794,6 +1795,7 @@ async def _common_add_spend_log_transaction_to_daily_transaction( if call_type: endpoint = ROUTE_ENDPOINT_MAPPING.get(call_type, None) + is_internal_call: Final = bool(_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY)) cache_read_input_tokens: Final = extract_cache_read_tokens(usage_obj) compression_saved_tokens: Final = extract_compression_saved_tokens(_metadata) savings_spend: Final = compute_savings_spend( @@ -1818,15 +1820,20 @@ async def _common_add_spend_log_transaction_to_daily_transaction( prompt_tokens=payload["prompt_tokens"], completion_tokens=payload["completion_tokens"], spend=payload["spend"], - api_requests=1, - successful_requests=1 if request_status == "success" else 0, - failed_requests=1 if request_status != "success" else 0, + # Internal sub-calls (auto-router classifier, shadow eval's shadow and + # judge) bill real spend and tokens to the key, but they are not + # requests the caller made: counting them inflates request-volume + # readers, and an auto-router savings figure computed on a shadow + # duplicate credits savings for traffic no user sent. + api_requests=0 if is_internal_call else 1, + successful_requests=1 if not is_internal_call and request_status == "success" else 0, + failed_requests=1 if not is_internal_call and request_status != "success" else 0, cache_read_input_tokens=cache_read_input_tokens, cache_creation_input_tokens=extract_cache_creation_tokens(usage_obj), compression_saved_tokens=compression_saved_tokens, compression_savings_spend=savings_spend.compression, prompt_caching_savings_spend=savings_spend.prompt_caching, - autorouter_savings_spend=savings_spend.autorouter, + autorouter_savings_spend=0.0 if is_internal_call else savings_spend.autorouter, ) return daily_transaction except Exception as e: diff --git a/litellm/proxy/guardrails/guardrail_hooks/llm_as_a_judge/__init__.py b/litellm/proxy/guardrails/guardrail_hooks/llm_as_a_judge/__init__.py index 1907bb19abf1..e3f67f0024ba 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/llm_as_a_judge/__init__.py +++ b/litellm/proxy/guardrails/guardrail_hooks/llm_as_a_judge/__init__.py @@ -1,16 +1,20 @@ """LLM-as-a-Judge guardrail: uses an LLM to score responses against weighted criteria.""" -import json -import re from collections.abc import Callable from datetime import datetime -from typing import TYPE_CHECKING, Any, Final, Literal, Optional, cast +from typing import TYPE_CHECKING, Any, Final, Literal, Optional from fastapi import HTTPException import litellm from litellm._logging import verbose_logger from litellm.integrations.custom_guardrail import CustomGuardrail +from litellm.litellm_core_utils.llm_judge import ( + default_router_provider, + extract_text_from_content, + judge_acompletion, + parse_json_verdict, +) from litellm.types.guardrails import GuardrailEventHooks, SupportedGuardrailIntegrations from litellm.types.utils import GenericGuardrailAPIInputs, GuardrailStatus @@ -32,50 +36,9 @@ _VALID_ON_FAILURE: Final = frozenset({"block", "log"}) - -def _default_router_provider() -> "Router | None": - try: - from litellm.proxy.proxy_server import llm_router - except ImportError: - return None - - return llm_router - - -_JSON_FENCE_RE: Final = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE) - - -def _parse_judge_verdict(raw: str) -> dict[str, Any]: - """Parse the judge's JSON verdict, tolerating markdown fences and surrounding prose.""" - text = raw.strip() - fenced: Final = _JSON_FENCE_RE.search(text) - if fenced is not None: - text = fenced.group(1).strip() - parsed: object - try: - parsed = json.loads(text) - except json.JSONDecodeError: - start: Final = text.find("{") - end: Final = text.rfind("}") - if start == -1 or end <= start: - raise - parsed = json.loads(text[start : end + 1]) - if not isinstance(parsed, dict): - raise ValueError("judge response is not a JSON object") - return cast(dict[str, Any], parsed) # cast-ok: narrowed to dict by the isinstance guard above - - -def _extract_text_from_content(content: Any) -> str: - """Return plain text from a message content field (str or multimodal list).""" - if isinstance(content, str): - return content - if isinstance(content, list): - parts: Final = [] - for part in content: - if isinstance(part, dict) and part.get("type") == "text": - parts.append(part.get("text", "")) - return " ".join(parts) - return "" +_default_router_provider: Final = default_router_provider +_parse_judge_verdict: Final = parse_json_verdict +_extract_text_from_content: Final = extract_text_from_content def _get_litellm_param( @@ -168,25 +131,13 @@ async def _run_judge( "content": _build_judge_prompt(self.criteria, messages, response_text), }, ] - router: Final = self._router_provider() - if router is not None and ( - self.judge_model in router.model_group_alias or router.get_model_list(model_name=self.judge_model) - ): - response = await router.acompletion( - model=self.judge_model, - messages=judge_messages, - response_format={"type": "json_object"}, - temperature=0, - num_retries=0, - fallbacks=[], - ) - else: - response = await litellm.acompletion( - model=self.judge_model, - messages=judge_messages, - response_format={"type": "json_object"}, - temperature=0, - ) + response: Final = await judge_acompletion( + self._router_provider(), + self.judge_model, + judge_messages, + response_format={"type": "json_object"}, + temperature=0, + ) raw: Final = response.choices[0].message.content or "{}" return _parse_judge_verdict(raw) diff --git a/litellm/proxy/management_endpoints/auto_router_endpoints.py b/litellm/proxy/management_endpoints/auto_router_endpoints.py index 8b6aafea751c..337dcd7d32a0 100644 --- a/litellm/proxy/management_endpoints/auto_router_endpoints.py +++ b/litellm/proxy/management_endpoints/auto_router_endpoints.py @@ -11,8 +11,11 @@ from pydantic import BaseModel, TypeAdapter +import litellm from litellm._logging import verbose_proxy_logger from litellm.exceptions import BudgetExceededError +from litellm.integrations.shadow_eval_logger import JUDGE_MAX_OUTPUT_TOKENS +from litellm.litellm_core_utils.llm_judge import router_resolves_model from litellm.proxy._types import ( CommonProxyErrors, LiteLLM_TeamTable, @@ -38,12 +41,18 @@ AutoRouterCacheStats, AutoRouterRoutingTestRequest, AutoRouterRoutingTestResponse, + GetShadowEvalJobResponse, RequestComplexityRouterConfig, + ShadowEvalResult, + ShadowEvalSlice, + StartShadowEvalRequest, + StartShadowEvalResponse, ) if TYPE_CHECKING: from fastapi import APIRouter, Depends, HTTPException, Query, status + from litellm.proxy.utils import PrismaClient from litellm.router import Router else: try: @@ -388,14 +397,7 @@ async def get_auto_router_benchmarks( """ from litellm.proxy.proxy_server import prisma_client - if user_api_key_dict.user_role not in ( - LitellmUserRoles.PROXY_ADMIN, - LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, - ): - raise HTTPException( - status_code=403, - detail="Only proxy admin roles can view auto-router benchmarks across the deployment", - ) + _require_admin_viewer(user_api_key_dict, "view auto-router benchmarks across the deployment") if prisma_client is None: raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) @@ -430,3 +432,371 @@ async def get_auto_router_benchmarks( totals=_benchmark_totals(_summed_agg_row(rows)), groups=groups, ) + + +# --------------------------------------------------------------------------- +# Shadow eval: pre-adoption evaluation of an auto-router against live traffic. +# --------------------------------------------------------------------------- + +# Judge price assumption for the upfront estimate: the conversation plus both responses +# on the prompt side, the judge's own bounded output budget on the completion side. +_FALLBACK_JUDGE_COST_PER_CALL: Final = 0.01 +_JUDGE_PROMPT_TOKENS_ESTIMATE: Final = 4000 + +# The estimate projects from the key's request volume over this many trailing days, +# read from the LiteLLM_DailyUserSpend rollup (a handful of indexed rows per key/day), +# never LiteLLM_SpendLogs (unbounded per-request rows with no api_key index). +_ESTIMATE_LOOKBACK_DAYS: Final = 7 + +_ESTIMATE_VOLUME_SQL: Final = """ +SELECT COALESCE(SUM(api_requests), 0)::bigint AS request_count +FROM "LiteLLM_DailyUserSpend" +WHERE api_key = $1 AND date >= $2 +""" + + +class _EstimateVolumeRow(BaseModel): + request_count: int + + +_ESTIMATE_VOLUME_ROWS: Final = TypeAdapter(list[_EstimateVolumeRow]) + + +async def _recent_request_volume(prisma_client: "PrismaClient", api_key_id: str) -> int: + lookback_date: Final = (datetime.now(timezone.utc) - timedelta(days=_ESTIMATE_LOOKBACK_DAYS)).strftime("%Y-%m-%d") + raw_rows: Final = await prisma_client.db.query_raw(_ESTIMATE_VOLUME_SQL, api_key_id, lookback_date) + rows: Final = _ESTIMATE_VOLUME_ROWS.validate_python(raw_rows or []) + return rows[0].request_count if rows else 0 + + +def _is_unique_violation(error: Exception) -> bool: + """Whether a Prisma create failed on a unique index. The one-active-job-per-key + guarantee lives in a partial unique index (raw SQL in the migration; schema.prisma + cannot express partial indexes), so the read-then-create check above it is advisory: + two concurrent starts pass the read, and the loser must surface as the same 409.""" + try: + from prisma.errors import UniqueViolationError + except ImportError: + return "unique constraint" in str(error).lower() or "P2002" in str(error) + return isinstance(error, UniqueViolationError) + + +def _require_admin_viewer(user_api_key_dict: UserAPIKeyAuth, action: str) -> None: + if user_api_key_dict.user_role not in ( + LitellmUserRoles.PROXY_ADMIN, + LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, + ): + raise HTTPException(status_code=403, detail=f"Only proxy admin roles can {action}") + + +def _require_admin_writer(user_api_key_dict: UserAPIKeyAuth, action: str) -> None: + if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: + raise HTTPException(status_code=403, detail=f"Only a proxy admin can {action}") + + +def _is_configured_pre_routing_strategy(llm_router: "Router", router_name: str) -> bool: + return any( + router_name in registry + for registry in ( + llm_router.auto_routers, + llm_router.complexity_routers, + llm_router.adaptive_routers, + llm_router.quality_routers, + ) + ) + + +def _validate_judge_model(llm_router: "Router | None", judge_model: str) -> None: + """Reject a judge model the dispatch path cannot resolve, at start rather than as a + silently growing failed_count once the job is already sampling and billing.""" + if llm_router is not None and _is_configured_pre_routing_strategy(llm_router, judge_model): + raise HTTPException( + status_code=400, + detail=f"judge_model '{judge_model}' is an auto-router; the judge must be a plain model", + ) + if router_resolves_model(llm_router, judge_model): + return + try: + litellm.get_llm_provider(model=judge_model) + except Exception as e: + raise HTTPException( + status_code=400, + detail=( + f"judge_model '{judge_model}' is neither a model configured on this proxy nor a " + "provider-qualified public model name (e.g. 'anthropic/claude-sonnet-5')" + ), + ) from e + + +def _judge_pricing_model(llm_router: "Router | None", judge_model: str) -> str: + """The name to price the judge under: a configured deployment's underlying provider + model when the judge is a deployment (the deployment name itself is admin-arbitrary + and not a pricing key), the given name otherwise.""" + deployments: Final = llm_router.get_model_list(model_name=judge_model) if llm_router is not None else None + if deployments: + underlying: Final = deployments[0].get("litellm_params", {}).get("model") + if isinstance(underlying, str) and underlying: + return underlying + return judge_model + + +def _estimate_judge_cost_per_call(llm_router: "Router | None", judge_model: str) -> float: + pricing_model: Final = _judge_pricing_model(llm_router, judge_model) + try: + prompt_cost, completion_cost = litellm.cost_per_token( + model=pricing_model, prompt_tokens=_JUDGE_PROMPT_TOKENS_ESTIMATE, completion_tokens=JUDGE_MAX_OUTPUT_TOKENS + ) + estimated: Final = prompt_cost + completion_cost + if estimated > 0: + return estimated + except Exception as e: # noqa: BLE001 # unknown judge model: fall back to a flat per-call figure + verbose_proxy_logger.debug("shadow_eval: judge cost lookup failed for %s: %s", pricing_model, e) + return _FALLBACK_JUDGE_COST_PER_CALL + + +class _VerdictAggRow(BaseModel): + grp: str + turn_count: int + real_wins: int + shadow_wins: int + ties: int + avg_confidence: float | None + + +_VERDICT_AGG_ROWS: Final = TypeAdapter(list[_VerdictAggRow]) + +_VERDICT_AGG_SELECT: Final = """ + COUNT(*)::int AS turn_count, + COUNT(*) FILTER (WHERE judge_preference = 'real')::int AS real_wins, + COUNT(*) FILTER (WHERE judge_preference = 'shadow')::int AS shadow_wins, + COUNT(*) FILTER (WHERE judge_preference = 'tie')::int AS ties, + AVG(judge_confidence)::float AS avg_confidence +FROM "LiteLLM_ShadowEvalVerdict" +WHERE job_id = $1 +GROUP BY 1 +""" + +_VERDICT_AGG_BY_TIER_SQL: Final = "SELECT COALESCE(tier_classification, 'UNCLASSIFIED') AS grp," + _VERDICT_AGG_SELECT +_VERDICT_AGG_BY_MODEL_SQL: Final = "SELECT real_model AS grp," + _VERDICT_AGG_SELECT + + +def _slices(rows: Sequence[_VerdictAggRow]) -> tuple[ShadowEvalSlice, ...]: + return tuple( + ShadowEvalSlice( + group=row.grp, + turn_count=row.turn_count, + real_win_rate_pct=_pct(row.real_wins, row.turn_count), + shadow_win_rate_pct=_pct(row.shadow_wins, row.turn_count), + tie_rate_pct=_pct(row.ties, row.turn_count), + avg_judge_confidence=round(row.avg_confidence or 0.0, 3), + ) + for row in sorted(rows, key=lambda r: r.turn_count, reverse=True) + ) + + +async def _shadow_eval_results(prisma_client: "PrismaClient", job_id: str) -> ShadowEvalResult | None: + """Both stratifications of one job's verdicts. Tier answers "where does the router + do well"; current-model answers "which of the models this key uses today would the + router beat". Row counts are bounded by the job's own verdicts, via the job_id index.""" + by_tier: Final = _VERDICT_AGG_ROWS.validate_python( + await prisma_client.db.query_raw(_VERDICT_AGG_BY_TIER_SQL, job_id) or () + ) + if not by_tier: + return None + by_model: Final = _VERDICT_AGG_ROWS.validate_python( + await prisma_client.db.query_raw(_VERDICT_AGG_BY_MODEL_SQL, job_id) or () + ) + total_turns: Final = sum(r.turn_count for r in by_tier) + return ShadowEvalResult( + by_tier=_slices(by_tier), + by_current_model=_slices(by_model), + overall_shadow_win_rate_pct=_pct(sum(r.shadow_wins for r in by_tier), total_turns), + overall_tie_rate_pct=_pct(sum(r.ties for r in by_tier), total_turns), + ) + + +def _job_to_response(record: object, results: ShadowEvalResult | None) -> GetShadowEvalJobResponse: + return GetShadowEvalJobResponse.model_validate(record, from_attributes=True).model_copy( + update={"results": results} # mutable-ok: pydantic update payload + ) + + +@router.post( + "/auto_router/shadow_eval/start", + tags=("auto router",), + dependencies=(Depends(user_api_key_auth),), + response_model=StartShadowEvalResponse, + status_code=status.HTTP_201_CREATED, +) +async def start_shadow_eval( + data: StartShadowEvalRequest, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], +) -> StartShadowEvalResponse: + """ + Start a pre-adoption shadow eval: duplicate a sampled slice of a key's live + traffic through an auto-router, judge real vs. shadow responses blind, and + stratify win rates by the router's tier classification. + + The shadow responses are never served to users. The job samples traffic for + duration_days (or until stopped via /auto_router/shadow_eval/{job_id}/stop), + then completes itself. Judge calls bill to the shadowed key; the estimate + returned here prices them from the key's trailing request volume scaled to + the requested duration. + """ + from litellm.proxy.proxy_server import llm_router, prisma_client + + _require_admin_writer(user_api_key_dict, "start a shadow eval") + if prisma_client is None: + raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) + if llm_router is None or not _is_configured_pre_routing_strategy(llm_router, data.router_name): + raise HTTPException( + status_code=400, + detail=f"'{data.router_name}' is not a configured auto-router", + ) + _validate_judge_model(llm_router, data.judge_model) + + key_row: Final = await prisma_client.db.litellm_verificationtoken.find_unique( + where={"token": data.api_key_id} # mutable-ok: Prisma filter + ) + if key_row is None: + raise HTTPException( + status_code=400, + detail=( + f"api_key_id '{data.api_key_id}' is not a key on this proxy; pass the key's token hash, " + "the value the key list and key info endpoints report" + ), + ) + + existing: Final = await prisma_client.db.litellm_shadowevaljob.find_first( + where={ # mutable-ok: Prisma filter + "api_key_id": data.api_key_id, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, + ) + if existing is not None: + raise HTTPException( + status_code=409, + detail=f"Key already has an active shadow eval job ({existing.id}). Stop it first.", + ) + + recent_requests: Final = await _recent_request_volume(prisma_client, data.api_key_id) + sampled: Final = int( + recent_requests * (data.duration_days / _ESTIMATE_LOOKBACK_DAYS) * data.shadow_percentage / 100.0 + ) + per_call: Final = _estimate_judge_cost_per_call(llm_router, data.judge_model) + estimated_cost: Final = round(sampled * per_call, 2) + ends_at: Final = datetime.now(timezone.utc) + timedelta(days=data.duration_days) + + try: + job: Final = await prisma_client.db.litellm_shadowevaljob.create( + data={ # mutable-ok: Prisma payload + "api_key_id": data.api_key_id, + "router_name": data.router_name, + "shadow_percentage": data.shadow_percentage, + "judge_model": data.judge_model, + "status": "pending", + "cost_estimate": estimated_cost, + "created_by": user_api_key_dict.user_id, + "ends_at": ends_at, + } + ) + except Exception as e: + if not _is_unique_violation(e): + raise + raise HTTPException( + status_code=409, + detail="Key already has an active shadow eval job (started concurrently). Stop it first.", + ) from e + return StartShadowEvalResponse( + job_id=job.id, + status="pending", + estimated_request_count=sampled, + estimated_cost=estimated_cost, + ) + + +@router.get( + "/auto_router/shadow_eval", + tags=("auto router",), + dependencies=(Depends(user_api_key_auth),), + response_model=list[GetShadowEvalJobResponse], +) +async def list_shadow_eval_jobs( + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], + api_key_id: Annotated[str | None, Query(description="Filter to jobs shadowing this key")] = None, + limit: Annotated[int, Query(ge=1, le=200, description="Newest jobs to return")] = 50, +) -> tuple[GetShadowEvalJobResponse, ...]: + """List shadow eval jobs, newest first. Results are omitted; fetch a single job for them.""" + from litellm.proxy.proxy_server import prisma_client + + _require_admin_viewer(user_api_key_dict, "view shadow evals") + if prisma_client is None: + raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) + + where: Final = {"api_key_id": api_key_id} if api_key_id else {} # mutable-ok: Prisma filter + records: Final = await prisma_client.db.litellm_shadowevaljob.find_many( + where=where, + order={"created_at": "desc"}, # mutable-ok: Prisma order + take=limit, + ) + return tuple(_job_to_response(record, results=None) for record in records or ()) + + +@router.get( + "/auto_router/shadow_eval/{job_id}", + tags=("auto router",), + dependencies=(Depends(user_api_key_auth),), + response_model=GetShadowEvalJobResponse, +) +async def get_shadow_eval_job( + job_id: str, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], +) -> GetShadowEvalJobResponse: + """Status, counters, and stratified results of one shadow eval job.""" + from litellm.proxy.proxy_server import prisma_client + + _require_admin_viewer(user_api_key_dict, "view shadow evals") + if prisma_client is None: + raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) + + record: Final = await prisma_client.db.litellm_shadowevaljob.find_unique( + where={"id": job_id} # mutable-ok: Prisma filter + ) + if record is None: + raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}") + return _job_to_response(record, results=await _shadow_eval_results(prisma_client, job_id)) + + +@router.post( + "/auto_router/shadow_eval/{job_id}/stop", + tags=("auto router",), + dependencies=(Depends(user_api_key_auth),), + response_model=GetShadowEvalJobResponse, +) +async def stop_shadow_eval_job( + job_id: str, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], +) -> GetShadowEvalJobResponse: + """Stop an active shadow eval job. Existing verdicts are kept; sampling halts within ~10s.""" + from litellm.proxy.proxy_server import prisma_client + + _require_admin_writer(user_api_key_dict, "stop a shadow eval") + if prisma_client is None: + raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) + + record: Final = await prisma_client.db.litellm_shadowevaljob.find_unique( + where={"id": job_id} # mutable-ok: Prisma filter + ) + if record is None: + raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}") + if record.status not in ("pending", "running"): + raise HTTPException(status_code=400, detail=f"Job {job_id} is already {record.status}") + + updated: Final = await prisma_client.db.litellm_shadowevaljob.update( + where={"id": job_id}, # mutable-ok: Prisma filter + data={ # mutable-ok: Prisma payload + "status": "completed", + "completed_at": datetime.now(timezone.utc), + }, + ) + return _job_to_response(updated, results=await _shadow_eval_results(prisma_client, job_id)) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 0f0079c542f5..312f9fd434a7 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -2296,6 +2296,22 @@ def cost_tracking(): if prisma_client is not None: litellm.logging_callback_manager.add_litellm_callback(_ProxyDBLogger()) litellm.logging_callback_manager.add_litellm_async_success_callback(_ProxyDBLogger()) + _register_shadow_eval_logger() + + +def _register_shadow_eval_logger() -> None: + """Register the shadow-eval success hook and start its lifecycle loop on the one + instance that is registered. Registration owns the loop start so the two cannot be + reordered apart, and the isinstance guard keeps a second cost_tracking() call from + constructing a duplicate whose loop would poll the DB from an unregistered instance. + Cheap when idle: with no active job rows the hook is one dict lookup per request.""" + from litellm.integrations.shadow_eval_logger import ShadowEvalLogger + + if any(isinstance(callback, ShadowEvalLogger) for callback in litellm.callbacks): + return + shadow_eval_logger: Final = ShadowEvalLogger() + litellm.logging_callback_manager.add_litellm_callback(shadow_eval_logger) + shadow_eval_logger.start_lifecycle_loop() # Bounds authoritative DB re-reads when enforcing a budget against a diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index 854602f53804..6e0ba43f9676 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -1450,6 +1450,50 @@ model LiteLLM_AutoRouterSession { @@index([last_turn_at], map: "idx_autorouter_session_last_turn") } +// Shadow Eval: pre-adoption evaluation of an auto-router against a key's live traffic. +// A sampled slice of requests is duplicated through the router in a detached task and +// an LLM judge compares real vs shadow responses blind; verdicts stratify by tier. +model LiteLLM_ShadowEvalJob { + id String @id @default(cuid()) + api_key_id String // the hashed virtual key whose traffic is shadowed + router_name String // the auto-router config to shadow through + shadow_percentage Float + judge_model String + + status String @default("pending") // pending | running | completed + + request_count Int @default(0) // requests seen on the key while active + completed_count Int @default(0) // verdicts written + failed_count Int @default(0) // shadow or judge calls that errored + last_error String? + + cost_estimate Float? // upfront judge-spend estimate shown at start + cost_actual Float @default(0) // running judge-call spend + + created_at DateTime @default(now()) + created_by String? + ends_at DateTime? + completed_at DateTime? + + @@index([api_key_id, status]) + @@index([status]) + @@index([created_at]) +} + +model LiteLLM_ShadowEvalVerdict { + id String @id @default(cuid()) + job_id String + request_id String // the judged real request + tier_classification String? // the router's tier for the prompt, when classified + real_model String // model that actually served the request + shadow_model String // model the router picked + judge_preference String // real | shadow | tie + judge_confidence Float? + created_at DateTime @default(now()) + + @@index([job_id]) +} + // --------------------------------------------------------------------------- // Workflow Run Tracking // diff --git a/litellm/router_strategy/complexity_router/complexity_router.py b/litellm/router_strategy/complexity_router/complexity_router.py index 32d252f3f680..8811c6b9a05a 100644 --- a/litellm/router_strategy/complexity_router/complexity_router.py +++ b/litellm/router_strategy/complexity_router/complexity_router.py @@ -26,8 +26,9 @@ from pydantic import BaseModel, create_model from litellm._logging import verbose_router_logger -from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY, RETURN_RAW_MODEL_NAME_METADATA_KEY +from litellm.constants import RETURN_RAW_MODEL_NAME_METADATA_KEY from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.internal_call_metadata import forwarded_internal_call_metadata from litellm.llms.base_llm.base_utils import type_to_response_format_param from litellm.types.utils import ( AUTOROUTER_CLASSIFIER_CALL_ORIGIN, @@ -172,40 +173,6 @@ def _append_custom_keywords(base_keywords: list[str], custom_keywords: list[str] return [*base_keywords, *deduped_custom.values()] -# Metadata keys that carry only the parent request's budget reservation state. These -# must not reach internal sub-calls (classifier, embedding): the reservation belongs to -# the routed completion being decided on, not to the sub-call itself, and forwarding it -# would let the sub-call's cost callback finalize the reservation, causing the routed -# completion's callback to skip incrementing key/team budget counters. -# -# Note: user_api_key_auth itself is intentionally kept; it is required by -# _filter_deployments_by_model_access_groups to scope embedding/classifier model -# selection to the caller's authorized access groups. It is forwarded as a sanitized -# copy with its budget_reservation sub-field removed, because the proxy cost callback -# (_get_budget_reservation_from_metadata) falls back to reading the reservation from -# inside the auth object when the top-level key is absent; forwarding it unsanitized -# would re-create the exact double-finalization this stripping exists to prevent. -_BUDGET_RESERVATION_METADATA_KEYS: Final = frozenset({"user_api_key_budget_reservation"}) - - -def _sanitize_user_api_key_auth(auth: Any) -> Any: - if isinstance(auth, dict): - return {k: v for k, v in auth.items() if k != "budget_reservation"} - if getattr(auth, "budget_reservation", None) is not None and hasattr(auth, "model_copy"): - return auth.model_copy(update={"budget_reservation": None}) - return auth - - -def _classifier_call_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]: - if not metadata: - return {} - return { - k: _sanitize_user_api_key_auth(v) if k == "user_api_key_auth" else v - for k, v in metadata.items() - if k not in _BUDGET_RESERVATION_METADATA_KEYS - } | {INTERNAL_CALL_ORIGIN_METADATA_KEY: AUTOROUTER_CLASSIFIER_CALL_ORIGIN} - - def _parent_session_kwargs(request_kwargs: Mapping[str, Any] | None) -> Mapping[str, Any]: kwargs: Final = request_kwargs or {} return {k: kwargs[k] for k in ("litellm_session_id", "litellm_trace_id") if kwargs.get(k) is not None} @@ -1043,7 +1010,7 @@ async def _classify_with_llm( ) request_metadata = (request_kwargs or {}).get("litellm_metadata") or (request_kwargs or {}).get("metadata") - metadata: Final = _classifier_call_metadata(request_metadata) + metadata: Final = forwarded_internal_call_metadata(request_metadata, AUTOROUTER_CLASSIFIER_CALL_ORIGIN) turn_off_message_logging: Final = _effective_turn_off_message_logging(request_kwargs) labeled_tiers: Final = self.config.labeled_tiers() @@ -1535,8 +1502,12 @@ async def _semantic_tier_override(self, user_message: str, request_kwargs: dict) # embedding call. Forwarding it would let the embedding's cost callback finalize the # reservation, so the routed completion's own callback then skips incrementing the # key/team budget. Key/team attribution fields are preserved for spend logging. - metadata: Final = _classifier_call_metadata(request_kwargs.get("metadata")) - litellm_metadata: Final = _classifier_call_metadata(request_kwargs.get("litellm_metadata")) + metadata: Final = forwarded_internal_call_metadata( + request_kwargs.get("metadata"), AUTOROUTER_CLASSIFIER_CALL_ORIGIN + ) + litellm_metadata: Final = forwarded_internal_call_metadata( + request_kwargs.get("litellm_metadata"), AUTOROUTER_CLASSIFIER_CALL_ORIGIN + ) turn_off_message_logging: Final = _effective_turn_off_message_logging(request_kwargs) proxy_server_request: Final = {"body": {"model": self.config.embedding_model, "input": [user_message]}} query_vector: Final = ( diff --git a/litellm/types/management_endpoints/auto_router_endpoints.py b/litellm/types/management_endpoints/auto_router_endpoints.py index 269d9b50414c..c35532a5be67 100644 --- a/litellm/types/management_endpoints/auto_router_endpoints.py +++ b/litellm/types/management_endpoints/auto_router_endpoints.py @@ -3,9 +3,10 @@ """ from collections.abc import Mapping -from typing import Final +from datetime import datetime +from typing import Final, Literal, TypeAlias -from pydantic import BaseModel, Field, field_validator +from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator from litellm.router_strategy.complexity_router.config import ComplexityRouterConfig from litellm.types.utils import StandardLoggingRoutingDecision @@ -141,3 +142,106 @@ class AutoRouterBenchmarksResponse(BaseModel): routers_in_scope: int totals: AutoRouterBenchmarkTotals groups: tuple[AutoRouterBenchmarkGroup, ...] + + +ShadowEvalStatus: TypeAlias = Literal["pending", "running", "completed"] + +DEFAULT_SHADOW_EVAL_JUDGE_MODEL: Final[str] = "anthropic/claude-sonnet-5" + + +class StartShadowEvalRequest(BaseModel): + """Start shadowing a key's traffic through an auto-router for blind comparison.""" + + api_key_id: str = Field( + description=( + "The hashed virtual key whose traffic will be shadowed. Shadow evaluation runs ONLY on this " + "key's traffic; requests made with any other key are not sampled." + ) + ) + router_name: str = Field(description="The auto-router config to shadow requests through") + shadow_percentage: float = Field( + ge=0.1, + le=100.0, + description="Percentage of the key's requests to duplicate through the router", + ) + judge_model: str = Field( + default=DEFAULT_SHADOW_EVAL_JUDGE_MODEL, + description=( + "Model used to blindly judge real vs. shadow responses. The judge only compares two answers, so a " + "mid-tier model (Claude Sonnet or GPT-4o class) is the sweet spot: small/nano-class models produce " + "unreliable or malformed verdicts, while frontier reasoning models add cost without changing outcomes." + ), + ) + duration_days: int = Field( + default=7, + ge=1, + le=30, + description="How many days the job samples traffic before stopping itself", + ) + + @field_validator("shadow_percentage") + @classmethod + def _round_percentage(cls, value: float) -> float: + return round(value, 2) + + +class StartShadowEvalResponse(BaseModel): + """Acknowledgement that a shadow-eval job was created, with an upfront cost estimate.""" + + job_id: str + status: ShadowEvalStatus + estimated_request_count: int = Field( + description="Requests expected to be shadowed, based on the key's recent request volume" + ) + estimated_cost: float = Field(description="Estimated dollar cost of the judge calls this job will make") + + +class ShadowEvalSlice(BaseModel): + """Judge outcomes for one slice of a job's verdicts (a router tier, or one of the + models the shadowed key currently uses).""" + + group: str + turn_count: int + real_win_rate_pct: float = Field(description="Share of judged turns where the real (control) model won") + shadow_win_rate_pct: float = Field(description="Share of judged turns where the shadowed router's pick won") + tie_rate_pct: float + avg_judge_confidence: float + + +class ShadowEvalResult(BaseModel): + """Stratified results of a shadow-eval job's verdicts so far.""" + + by_tier: tuple[ShadowEvalSlice, ...] + by_current_model: tuple[ShadowEvalSlice, ...] + overall_shadow_win_rate_pct: float + overall_tie_rate_pct: float + + +class GetShadowEvalJobResponse(BaseModel): + """Status and, once available, results of a shadow-eval job. + + Validates directly from the prisma job record (job_id reads the row's id), so the + endpoint needs no hand-written row-to-response mapping. + """ + + model_config = ConfigDict(from_attributes=True, populate_by_name=True) + + job_id: str = Field(validation_alias=AliasChoices("id", "job_id")) + status: ShadowEvalStatus + router_name: str + api_key_id: str = Field(description="The hashed virtual key whose traffic this job evaluates, and only that key's") + shadow_percentage: float + request_count: int = Field(description="Total requests observed on the shadowed key since the job started") + completed_count: int = Field(description="Verdicts written so far") + failed_count: int = Field(description="Shadow or judge calls that errored and were skipped") + last_error: str | None = Field( + default=None, description="The most recent shadow or judge failure, so a growing failed_count is diagnosable" + ) + results: ShadowEvalResult | None = Field( + default=None, description="Present once at least one verdict has been recorded" + ) + cost_estimate: float | None = None + cost_actual: float = Field(default=0.0, description="Running total of judge-call spend for this job") + created_at: datetime + ends_at: datetime | None = Field(default=None, description="When the job stops sampling on its own") + completed_at: datetime | None = None diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 354857f8d72d..d9ef538d530f 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -2782,11 +2782,13 @@ class StandardLoggingRoutingDecisionTierBoundaries(TypedDict): ] -InternalCallOrigin = Literal["autorouter_classifier"] +InternalCallOrigin = Literal["autorouter_classifier", "shadow_eval_router", "shadow_eval_judge"] """Which internal litellm feature originated a billed sub-call, so a spend log row records that it is not traffic the caller sent.""" AUTOROUTER_CLASSIFIER_CALL_ORIGIN: Final[InternalCallOrigin] = "autorouter_classifier" +SHADOW_EVAL_ROUTER_CALL_ORIGIN: Final[InternalCallOrigin] = "shadow_eval_router" +SHADOW_EVAL_JUDGE_CALL_ORIGIN: Final[InternalCallOrigin] = "shadow_eval_judge" class StandardLoggingRoutingDecision(TypedDict, total=False): diff --git a/schema.prisma b/schema.prisma index 854602f53804..6e0ba43f9676 100644 --- a/schema.prisma +++ b/schema.prisma @@ -1450,6 +1450,50 @@ model LiteLLM_AutoRouterSession { @@index([last_turn_at], map: "idx_autorouter_session_last_turn") } +// Shadow Eval: pre-adoption evaluation of an auto-router against a key's live traffic. +// A sampled slice of requests is duplicated through the router in a detached task and +// an LLM judge compares real vs shadow responses blind; verdicts stratify by tier. +model LiteLLM_ShadowEvalJob { + id String @id @default(cuid()) + api_key_id String // the hashed virtual key whose traffic is shadowed + router_name String // the auto-router config to shadow through + shadow_percentage Float + judge_model String + + status String @default("pending") // pending | running | completed + + request_count Int @default(0) // requests seen on the key while active + completed_count Int @default(0) // verdicts written + failed_count Int @default(0) // shadow or judge calls that errored + last_error String? + + cost_estimate Float? // upfront judge-spend estimate shown at start + cost_actual Float @default(0) // running judge-call spend + + created_at DateTime @default(now()) + created_by String? + ends_at DateTime? + completed_at DateTime? + + @@index([api_key_id, status]) + @@index([status]) + @@index([created_at]) +} + +model LiteLLM_ShadowEvalVerdict { + id String @id @default(cuid()) + job_id String + request_id String // the judged real request + tier_classification String? // the router's tier for the prompt, when classified + real_model String // model that actually served the request + shadow_model String // model the router picked + judge_preference String // real | shadow | tie + judge_confidence Float? + created_at DateTime @default(now()) + + @@index([job_id]) +} + // --------------------------------------------------------------------------- // Workflow Run Tracking // diff --git a/tests/test_litellm/integrations/test_shadow_eval_logger.py b/tests/test_litellm/integrations/test_shadow_eval_logger.py new file mode 100644 index 000000000000..f500ff783973 --- /dev/null +++ b/tests/test_litellm/integrations/test_shadow_eval_logger.py @@ -0,0 +1,494 @@ +"""Unit tests for the shadow-eval logger: sampling, unmasking, the success hook's skip +paths, the detached pipeline, and the lifecycle loop's flush/finalize behavior.""" + +import asyncio +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY +from litellm.integrations.shadow_eval_logger import ( + _MAX_CONCURRENT_SHADOW_TASKS, + _MAX_JUDGE_PROMPT_CHARS, + JUDGE_MAX_OUTPUT_TOKENS, + ActiveShadowEvalJob, + ShadowEvalLogger, + _CallFailure, + _judge_user_prompt, + _sample_hits, + _unmask_preference, +) +from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN + + +def _job(**overrides) -> ActiveShadowEvalJob: + defaults = dict( + id="job-1", + router_name="my-router", + shadow_percentage=100.0, + judge_model="judge-model", + status="running", + cost_estimate=5.0, + cost_actual=0.0, + ends_at=datetime.now(timezone.utc) + timedelta(days=1), + ) + return ActiveShadowEvalJob(**{**defaults, **overrides}) + + +def _prisma() -> MagicMock: + """Prisma wrapper mock. Lifecycle/counter writes use prisma.db; the pipeline's + recording runs inside prisma.tx(), whose statements land on prisma.tx_mock.""" + prisma = MagicMock() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_shadowevaljob.update_many = AsyncMock(return_value=1) + prisma.db.litellm_shadowevalverdict.create = AsyncMock() + tx = MagicMock() + tx.litellm_shadowevaljob.update_many = AsyncMock(return_value=1) + tx.litellm_shadowevalverdict.create = AsyncMock() + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=tx) + ctx.__aexit__ = AsyncMock(return_value=False) + prisma.tx = MagicMock(return_value=ctx) + prisma.tx_mock = tx + return prisma + + +def _router(shadow_text="shadow answer", judge_json='{"preference": "A", "confidence": 0.9, "reasoning": "x"}'): + """One mock router serving the shadow call first, the judge call second. The shadow + call's metadata receives the routing decision write-back, like the real router.""" + router = MagicMock() + router.model_group_alias = {} + router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "openai/gpt-4o-mini"}}]) + + async def acompletion(**kwargs): + if kwargs["model"] == "my-router": + kwargs["metadata"]["routing_decision"] = {"tier_label": "SIMPLE", "routed_model": "cheap-model"} + return {"choices": [{"message": {"content": shadow_text}}], "usage": {"completion_tokens": 5}} + return {"choices": [{"message": {"content": judge_json}}]} + + router.acompletion = MagicMock(side_effect=acompletion) + return router + + +def _logger(router=None, prisma=None) -> ShadowEvalLogger: + return ShadowEvalLogger( + router_provider=lambda: router, + prisma_provider=lambda: prisma, + ) + + +def _success_kwargs(request_id="req-1", api_key_hash="key-hash", request_metadata=None, call_type="acompletion"): + return { + "standard_logging_object": { + "id": request_id, + "call_type": call_type, + "model": "claude-opus", + "metadata": {"user_api_key_hash": api_key_hash}, + "model_parameters": {"temperature": 0.5, "stream": True}, + }, + "litellm_params": {"metadata": request_metadata or {}}, + "messages": [{"role": "user", "content": "what is 2+2"}], + } + + +RESPONSE = {"choices": [{"message": {"content": "real answer"}}]} + + +async def _drain(logger: ShadowEvalLogger): + for _ in range(100): + if logger._inflight_shadow_tasks == 0: + return + await asyncio.sleep(0.01) + raise AssertionError("shadow tasks never drained") + + +class TestSampling: + def test_boundaries_and_determinism(self): + assert not any(_sample_hits(f"req-{i}", "job", 0.0) for i in range(100)) + assert all(_sample_hits(f"req-{i}", "job", 100.0) for i in range(100)) + assert len({_sample_hits("req-1", "job-1", 50.0) for _ in range(10)}) == 1 + + def test_distribution_close_to_percentage(self): + hits = sum(_sample_hits(f"req-{i}", "job-x", 10.0) for i in range(10_000)) + assert 800 < hits < 1200 + + def test_different_jobs_sample_independently(self): + agreements = sum( + _sample_hits(f"req-{i}", "job-a", 50.0) == _sample_hits(f"req-{i}", "job-b", 50.0) for i in range(1000) + ) + assert 300 < agreements < 700 + + +@pytest.mark.parametrize( + "raw,real_is_a,expected", + [ + ("A", True, "real"), + ("a", True, "real"), + ("A", False, "shadow"), + ("B", True, "shadow"), + ("B", False, "real"), + ("tie", True, "tie"), + ("garbage", True, "tie"), + ("", False, "tie"), + ], +) +def test_unmask_preference(raw, real_is_a, expected): + assert _unmask_preference(raw, real_is_a) == expected + + +def test_judge_prompt_is_bounded_however_large_the_inputs(): + prompt = _judge_user_prompt("c" * 200_000, "a" * 200_000, "b" * 200_000) + assert len(prompt) < _MAX_JUDGE_PROMPT_CHARS + 100 + assert prompt.endswith("Which response is better?") + small = _judge_user_prompt("conv", "alpha", "beta") + assert "conv" in small and "alpha" in small and "beta" in small + + +@pytest.mark.asyncio +class TestSuccessHookSkips: + """Every skip path must leave no scheduled task; the sampled path must schedule one.""" + + async def test_happy_path_writes_a_verdict_and_bumps_counters(self, monkeypatch: pytest.MonkeyPatch): + import litellm as litellm_module + + monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005) + prisma = _prisma() + router = _router() + logger = _logger(router=router, prisma=prisma) + logger._jobs_by_key = {"key-hash": _job()} + + await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None) + await _drain(logger) + + create_kwargs = prisma.tx_mock.litellm_shadowevalverdict.create.call_args.kwargs["data"] + assert create_kwargs["job_id"] == "job-1" + assert create_kwargs["request_id"] == "req-1" + assert create_kwargs["tier_classification"] == "SIMPLE" + assert create_kwargs["real_model"] == "claude-opus" + assert create_kwargs["shadow_model"] == "cheap-model" + assert create_kwargs["judge_preference"] in ("real", "shadow") + assert create_kwargs["judge_confidence"] == 0.9 + counter_update = prisma.tx_mock.litellm_shadowevaljob.update_many.call_args.kwargs + assert counter_update["data"]["completed_count"] == {"increment": 1} + assert counter_update["data"]["cost_actual"] == {"increment": 0.005} + assert counter_update["where"]["status"] == {"in": ["pending", "running"]} + assert logger._pending_seen == {"job-1": 1} + + @pytest.mark.parametrize( + "kwargs_mutation,job_mutation", + [ + ({"request_metadata": {INTERNAL_CALL_ORIGIN_METADATA_KEY: "shadow_eval_router"}}, {}), + ({"api_key_hash": "other-key"}, {}), + ({"call_type": "aembedding"}, {}), + ({"call_type": None}, {}), + ({"request_metadata": {"routing_decision": {"router_model_name": "my-router"}}}, {}), + ({}, {"ends_at": datetime.now(timezone.utc) - timedelta(seconds=1)}), + ({}, {"cost_estimate": 1.0, "cost_actual": 99.0}), + ], + ids=["internal-origin", "no-job-for-key", "non-chat", "missing-call-type", "self-shadow", "past-end", "over-spend-cap"], + ) + async def test_skip_paths_schedule_nothing(self, kwargs_mutation, job_mutation): + prisma = _prisma() + logger = _logger(router=_router(), prisma=prisma) + logger._jobs_by_key = {"key-hash": _job(**job_mutation)} + + await logger.async_log_success_event(_success_kwargs(**kwargs_mutation), RESPONSE, None, None) + await _drain(logger) + + prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called() + assert logger._inflight_shadow_tasks == 0 + + async def test_inflight_cap_sheds_instead_of_queueing(self): + prisma = _prisma() + logger = _logger(router=_router(), prisma=prisma) + logger._jobs_by_key = {"key-hash": _job()} + logger._inflight_shadow_tasks = _MAX_CONCURRENT_SHADOW_TASKS + + await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None) + + assert logger._inflight_shadow_tasks == _MAX_CONCURRENT_SHADOW_TASKS + prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called() + + async def test_sampled_out_request_still_counts_toward_request_count(self): + logger = _logger(router=_router(), prisma=_prisma()) + logger._jobs_by_key = {"key-hash": _job(shadow_percentage=0.1)} + + for i in range(20): + await logger.async_log_success_event(_success_kwargs(request_id=f"req-{i}"), RESPONSE, None, None) + await _drain(logger) + + assert logger._pending_seen["job-1"] == 20 + + +@pytest.mark.asyncio +class TestShadowPipeline: + async def test_no_prisma_means_no_provider_spend(self): + """The prisma gate sits above the shadow and judge dispatch: no verdict store, + no spend.""" + router = _router() + logger = _logger(router=router, prisma=None) + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={}, + parent_metadata={}, + ) + + router.acompletion.assert_not_called() + + async def test_over_budget_key_skips_before_any_call(self, monkeypatch: pytest.MonkeyPatch): + """The gate delegates to the auth path's own budget owner, so an over-budget + verdict there (BudgetExceededError) skips the shadow before any provider call.""" + import litellm.proxy.auth.auth_checks as auth_checks + from litellm.exceptions import BudgetExceededError + from litellm.proxy._types import UserAPIKeyAuth + + monkeypatch.setattr( + auth_checks, + "_virtual_key_max_budget_check", + AsyncMock(side_effect=BudgetExceededError(current_cost=11.0, max_budget=10.0)), + ) + router = _router() + logger = _logger(router=router, prisma=_prisma()) + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={}, + parent_metadata={"user_api_key_auth": UserAPIKeyAuth(api_key="sk-abc", max_budget=10.0)}, + ) + + router.acompletion.assert_not_called() + + async def test_shadow_failure_bumps_failed_count_with_last_error(self): + prisma = _prisma() + router = MagicMock() + router.model_group_alias = {} + router.get_model_list = MagicMock(return_value=None) + router.acompletion = AsyncMock(side_effect=RuntimeError("provider exploded")) + logger = _logger(router=router, prisma=prisma) + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={}, + parent_metadata={}, + ) + + prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called() + bump = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs + assert bump["data"]["failed_count"] == {"increment": 1} + assert "provider exploded" in bump["data"]["last_error"] + + async def test_unparseable_judge_verdict_is_a_counted_failure(self): + prisma = _prisma() + logger = _logger(router=_router(judge_json="I prefer response A, definitely"), prisma=prisma) + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={}, + parent_metadata={}, + ) + + prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called() + bump = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs + assert bump["data"]["failed_count"] == {"increment": 1} + + async def test_redacted_requests_are_never_shadowed(self): + """Redaction rewrites the logged messages before callbacks run, so this hook + only ever sees placeholders for opted-out traffic: sampling it would judge + garbage and put content the caller opted out of logging into sub-call rows. + The skip uses the redactor's own predicate, so every redaction source counts.""" + prisma = _prisma() + router = _router() + logger = _logger(router=router, prisma=prisma) + logger._jobs_by_key = {"key-hash": _job()} + + hook_kwargs = _success_kwargs() + hook_kwargs["standard_callback_dynamic_params"] = {"turn_off_message_logging": True} + await logger.async_log_success_event(hook_kwargs, RESPONSE, None, None) + await _drain(logger) + + router.acompletion.assert_not_called() + prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called() + + async def test_sub_calls_carry_identity_and_origin_but_never_parent_request_state(self): + prisma = _prisma() + router = _router() + logger = _logger(router=router, prisma=prisma) + parent_metadata = { + "user_api_key_hash": "key-hash", + "user_api_key_team_id": "team-1", + "user_api_key_budget_reservation": {"amount": 1.0}, + "routing_decision": {"router_model_name": "other-router"}, + } + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={"stream": True, "temperature": 0.2, "metadata": {"x": 1}}, + parent_metadata=parent_metadata, + ) + + shadow_call = router.acompletion.call_args_list[0].kwargs + judge_call = router.acompletion.call_args_list[1].kwargs + for call in (shadow_call, judge_call): + assert call["num_retries"] == 0 + assert call["fallbacks"] == [] + assert shadow_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN + assert judge_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_JUDGE_CALL_ORIGIN + for call in (shadow_call, judge_call): + assert call["metadata"]["user_api_key_hash"] == "key-hash" + assert call["metadata"]["user_api_key_team_id"] == "team-1" + assert "user_api_key_budget_reservation" not in call["metadata"] + assert "routing_decision" not in judge_call["metadata"] + assert "stream" not in shadow_call + assert shadow_call["temperature"] == 0.2 + assert judge_call["max_tokens"] == JUDGE_MAX_OUTPUT_TOKENS + + + + async def test_job_stopped_mid_flight_drops_the_verdict(self): + """The status-guarded counter update decides whether the verdict lands, so a + completed job's results can never disagree with its frozen counts.""" + prisma = _prisma() + prisma.tx_mock.litellm_shadowevaljob.update_many = AsyncMock(return_value=0) + logger = _logger(router=_router(), prisma=prisma) + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={}, + parent_metadata={}, + ) + + prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called() + + async def test_failed_verdict_write_files_the_pipeline_under_failed_once(self): + """A create that raises inside the transaction rolls the counters back, and the + pipeline lands in failed_count exactly once, never in both buckets.""" + prisma = _prisma() + prisma.tx_mock.litellm_shadowevalverdict.create = AsyncMock(side_effect=RuntimeError("db write failed")) + logger = _logger(router=_router(), prisma=prisma) + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={}, + parent_metadata={}, + ) + + bumps = [c.kwargs for c in prisma.db.litellm_shadowevaljob.update_many.call_args_list] + assert len(bumps) == 1 + assert bumps[0]["data"]["failed_count"] == {"increment": 1} + assert "db write failed" in bumps[0]["data"]["last_error"] + + async def test_recording_runs_inside_one_transaction(self): + """Counter and verdict must ride prisma.tx(), never separate prisma.db writes, + so a failed verdict write cannot leave a counted-but-missing verdict.""" + prisma = _prisma() + logger = _logger(router=_router(), prisma=prisma) + + await logger._run_shadow_eval( + job=_job(), + request_id="req-1", + messages=({"role": "user", "content": "hi"},), + response_obj=RESPONSE, + real_model="claude-opus", + model_parameters={}, + parent_metadata={}, + ) + + prisma.tx.assert_called_once() + prisma.tx_mock.litellm_shadowevaljob.update_many.assert_awaited_once() + prisma.tx_mock.litellm_shadowevalverdict.create.assert_awaited_once() + prisma.db.litellm_shadowevaljob.update_many.assert_not_called() + prisma.db.litellm_shadowevalverdict.create.assert_not_called() + +@pytest.mark.asyncio +class TestLifecycle: + async def test_flush_is_status_guarded_and_buffer_resets(self): + prisma = _prisma() + logger = _logger(prisma=prisma) + logger._pending_seen = {"job-1": 7} + + await logger._flush_seen_counts() + + flush = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs + assert flush["where"] == {"id": "job-1", "status": {"in": ["pending", "running"]}} + assert flush["data"] == {"request_count": {"increment": 7}} + assert logger._pending_seen == {} + + async def test_tick_finalizes_expired_and_overspent_jobs(self): + prisma = _prisma() + expired = _job(id="job-expired", ends_at=datetime.now(timezone.utc) - timedelta(seconds=1)) + overspent = _job(id="job-overspent", cost_estimate=1.0, cost_actual=99.0) + active = _job(id="job-active") + prisma.db.litellm_shadowevaljob.find_many = AsyncMock( + return_value=[ + MagicMock( + id=j.id, + api_key_id=f"key-{j.id}", + router_name=j.router_name, + shadow_percentage=j.shadow_percentage, + judge_model=j.judge_model, + status=j.status, + cost_estimate=j.cost_estimate, + cost_actual=j.cost_actual, + ends_at=j.ends_at, + ) + for j in (expired, overspent, active) + ] + ) + logger = _logger(prisma=prisma) + + await logger._lifecycle_tick() + + finalized = {c.kwargs["where"]["id"] for c in prisma.db.litellm_shadowevaljob.update_many.call_args_list} + assert finalized == {"job-expired", "job-overspent"} + for call in prisma.db.litellm_shadowevaljob.update_many.call_args_list: + assert call.kwargs["where"]["status"] == {"in": ["pending", "running"]} + assert call.kwargs["data"]["status"] == "completed" + assert set(logger._jobs_by_key) == {"key-job-active"} + + async def test_refresh_keeps_stale_snapshot_on_db_blip(self): + prisma = _prisma() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=RuntimeError("db blip")) + logger = _logger(prisma=prisma) + logger._jobs_by_key = {"key-hash": _job()} + + await logger._refresh_active_jobs() + + assert set(logger._jobs_by_key) == {"key-hash"} + + async def test_start_lifecycle_loop_is_idempotent(self): + logger = _logger(prisma=_prisma()) + logger.start_lifecycle_loop() + first: asyncio.Task = logger._lifecycle_task + logger.start_lifecycle_loop() + assert logger._lifecycle_task is first + first.cancel() diff --git a/tests/test_litellm/litellm_core_utils/test_internal_call_metadata.py b/tests/test_litellm/litellm_core_utils/test_internal_call_metadata.py new file mode 100644 index 000000000000..73923dc75a5f --- /dev/null +++ b/tests/test_litellm/litellm_core_utils/test_internal_call_metadata.py @@ -0,0 +1,121 @@ +"""Unit tests for internal-call metadata forwarding: budget-reservation stripping and origin stamping.""" + +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY +from litellm.litellm_core_utils.internal_call_metadata import ( + forwarded_internal_call_metadata, + sanitized_forwardable_call_metadata, +) +from litellm.types.utils import SHADOW_EVAL_ROUTER_CALL_ORIGIN + +PARENT = { + "user_api_key": "sk-hash", + "user_api_key_hash": "sk-hash", + "user_api_key_team_id": "team-1", + "user_api_key_budget_reservation": {"amount": 1.0}, + "user_api_key_auth": {"api_key": "sk-hash", "budget_reservation": {"amount": 1.0}}, + "routing_decision": {"router_model_name": "my-router"}, + "headers": {"x-request-id": "abc"}, +} + + +def test_forwarded_metadata_strips_reservation_everywhere_and_stamps_origin(): + result = forwarded_internal_call_metadata(PARENT, "autorouter_classifier") + + assert result[INTERNAL_CALL_ORIGIN_METADATA_KEY] == "autorouter_classifier" + assert "user_api_key_budget_reservation" not in result + assert result["user_api_key_auth"] == {"api_key": "sk-hash"} + assert result["routing_decision"] == {"router_model_name": "my-router"} + assert PARENT["user_api_key_auth"]["budget_reservation"] is not None + + +def test_forwarded_metadata_empty_parent_stays_unstamped(): + assert forwarded_internal_call_metadata(None, "autorouter_classifier") == {} + assert forwarded_internal_call_metadata({}, "autorouter_classifier") == {} + + +def test_sanitized_forwardable_metadata_keeps_only_identity_and_always_stamps(): + result = sanitized_forwardable_call_metadata(PARENT, SHADOW_EVAL_ROUTER_CALL_ORIGIN) + + assert result[INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN + assert result["user_api_key"] == "sk-hash" + assert result["user_api_key_team_id"] == "team-1" + assert result["user_api_key_auth"] == {"api_key": "sk-hash"} + assert "routing_decision" not in result + assert "headers" not in result + assert "user_api_key_budget_reservation" not in result + + assert sanitized_forwardable_call_metadata({}, SHADOW_EVAL_ROUTER_CALL_ORIGIN) == { + INTERNAL_CALL_ORIGIN_METADATA_KEY: SHADOW_EVAL_ROUTER_CALL_ORIGIN + } + + +class TestSubCallMetadataSanitization: + """The proxy cost callback must not be able to recover the parent budget reservation + from sub-call metadata, in either of the shapes it knows how to read.""" + + def test_cost_callback_cannot_recover_reservation_from_sanitized_metadata(self): + from litellm.proxy._types import UserAPIKeyAuth + from litellm.proxy.hooks.proxy_track_cost_callback import ( + _get_budget_reservation_from_metadata, + ) + + reservation = {"reserved_cost": 1.0} + auth_shapes = ( + {"models": ["gpt-4o"], "budget_reservation": dict(reservation)}, + UserAPIKeyAuth(api_key="sk-abc", budget_reservation=dict(reservation)), + ) + for auth in auth_shapes: + metadata = { + "user_api_key_hash": "hash-abc", + "user_api_key_budget_reservation": dict(reservation), + "user_api_key_auth": auth, + } + assert _get_budget_reservation_from_metadata(metadata) == reservation + + sanitized = forwarded_internal_call_metadata(metadata, "autorouter_classifier") + assert sanitized is not None + assert sanitized["user_api_key_auth"] is not None + assert _get_budget_reservation_from_metadata(sanitized) is None + + def test_classifier_buckets_keep_non_spend_fields_on_a_chat_completions_parent(self): + """Drives the real resolver over the buckets the embedding classifier builds. + + An absent bucket must stay empty rather than carry a lone origin stamp: + get_litellm_metadata_from_kwargs prefers litellm_metadata whenever truthy, so an + origin-only dict would make an empty litellm_metadata win and silently drop + requester_ip_address, tags and spend_logs_metadata from the classifier's row.""" + from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs + + parent = { + "user_api_key": "sk-abc", + "requester_ip_address": "10.0.0.1", + "spend_logs_metadata": {"team_note": "keep me"}, + "tags": ["prod"], + } + resolved = get_litellm_metadata_from_kwargs( + { + "litellm_params": { + "metadata": forwarded_internal_call_metadata(parent, "autorouter_classifier"), + "litellm_metadata": forwarded_internal_call_metadata(None, "autorouter_classifier"), + } + } + ) + assert resolved["internal_call_origin"] == "autorouter_classifier" + assert resolved["requester_ip_address"] == "10.0.0.1" + assert resolved["spend_logs_metadata"] == {"team_note": "keep me"} + assert resolved["tags"] == ["prod"] + + def test_sanitized_auth_keeps_access_group_fields_and_leaves_original_untouched(self): + from litellm.proxy._types import UserAPIKeyAuth + + auth = UserAPIKeyAuth( + api_key="sk-abc", + team_id="team-1", + budget_reservation={"reserved_cost": 1.0}, + ) + sanitized = forwarded_internal_call_metadata({"user_api_key_auth": auth}, "autorouter_classifier") + sanitized_auth = sanitized["user_api_key_auth"] + assert sanitized_auth.budget_reservation is None + assert sanitized_auth.team_id == "team-1" + assert sanitized_auth.api_key == auth.api_key + assert auth.budget_reservation == {"reserved_cost": 1.0} diff --git a/tests/test_litellm/litellm_core_utils/test_llm_judge.py b/tests/test_litellm/litellm_core_utils/test_llm_judge.py new file mode 100644 index 000000000000..ff7ec0c19011 --- /dev/null +++ b/tests/test_litellm/litellm_core_utils/test_llm_judge.py @@ -0,0 +1,88 @@ +"""Unit tests for the shared LLM-judge primitives: verdict parsing, router resolution, dispatch.""" + +import json +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from litellm.litellm_core_utils.llm_judge import ( + extract_text_from_content, + judge_acompletion, + parse_json_verdict, + router_resolves_model, +) + + +@pytest.mark.parametrize( + "raw,expected", + [ + ('{"preference": "A", "confidence": 0.9}', "A"), + ('Here it is:\n```json\n{"preference": "B"}\n```\nDone.', "B"), + ('```\n{"preference": "tie"}\n```', "tie"), + ('Verdict: {"preference": "A", "confidence": 0.5} final.', "A"), + ], +) +def test_parse_json_verdict_tolerates_fences_and_prose(raw, expected): + assert parse_json_verdict(raw)["preference"] == expected + + +def test_parse_json_verdict_rejects_non_object(): + with pytest.raises(ValueError): + parse_json_verdict('["not", "an", "object"]') + with pytest.raises((json.JSONDecodeError, ValueError)): + parse_json_verdict("no json here at all") + + +@pytest.mark.parametrize( + "content,expected", + [ + ("hello", "hello"), + ([{"type": "text", "text": "a"}, {"type": "image_url", "image_url": {}}, {"type": "text", "text": "b"}], "a b"), + (42, ""), + (None, ""), + ], +) +def test_extract_text_from_content(content, expected): + assert extract_text_from_content(content) == expected + + +def _router(alias=(), deployments=False) -> MagicMock: + router = MagicMock() + router.model_group_alias = dict.fromkeys(alias, "x") + router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "openai/gpt-4o"}}] if deployments else None) + router.acompletion = AsyncMock(return_value={"choices": [{"message": {"content": "router answer"}}]}) + return router + + +def test_router_resolves_model_matrix(): + assert router_resolves_model(None, "gpt-4o") is False + assert router_resolves_model(_router(), "gpt-4o") is False + assert router_resolves_model(_router(alias=("gpt-4o",)), "gpt-4o") is True + assert router_resolves_model(_router(deployments=True), "gpt-4o") is True + + +@pytest.mark.asyncio +async def test_judge_acompletion_prefers_router_and_disables_retries(): + router = _router(deployments=True) + response = await judge_acompletion(router, "judge-model", [{"role": "user", "content": "hi"}], temperature=0) + assert response == {"choices": [{"message": {"content": "router answer"}}]} + _, kwargs = router.acompletion.call_args + assert kwargs["num_retries"] == 0 + assert kwargs["fallbacks"] == [] + assert kwargs["temperature"] == 0 + + +@pytest.mark.asyncio +async def test_judge_acompletion_falls_back_to_sdk_for_unconfigured_model(monkeypatch: pytest.MonkeyPatch): + import litellm as litellm_module + + sdk = AsyncMock(return_value={"choices": [{"message": {"content": "sdk answer"}}]}) + monkeypatch.setattr(litellm_module, "acompletion", sdk) + router = _router() + + response = await judge_acompletion(router, "anthropic/claude-sonnet-5", [{"role": "user", "content": "hi"}]) + + assert response == {"choices": [{"message": {"content": "sdk answer"}}]} + router.acompletion.assert_not_called() + assert sdk.call_args.kwargs["model"] == "anthropic/claude-sonnet-5" + assert sdk.call_args.kwargs["num_retries"] == 0 diff --git a/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py b/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py index 0df11f224a24..95dce1ccb0a7 100644 --- a/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py +++ b/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py @@ -279,3 +279,10 @@ def test_every_drain_trigger_reads_the_one_queue_census_owner(): 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__ + + +def test_internal_call_origin_never_reaches_the_rollup(): + """A shadow eval's duplicate carries a real routing_decision, so the decision-presence + gate alone would count it; the internal_call_origin stamp must exclude it.""" + assert _build(metadata=_metadata(internal_call_origin="shadow_eval_router")) is None + assert _build() is not None diff --git a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py index 2c426e0f071c..ca7d5fcd2739 100644 --- a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py +++ b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py @@ -2221,3 +2221,50 @@ async def test_commit_spend_updates_to_db_does_not_stamp_key_settings_updated_at assert call_kwargs["where"] == {"token": token} assert set(call_kwargs["data"]) == {"spend", "last_active"} assert call_kwargs["data"]["spend"] == {"increment": response_cost} + + +@pytest.mark.asyncio +async def test_daily_transaction_internal_call_keeps_spend_but_not_request_counts(): + """Internal sub-calls (auto-router classifier, shadow eval's shadow and judge) bill + spend and tokens to the key but are not requests the caller made: api_requests, + successful_requests, and autorouter_savings_spend must all stay zero for them.""" + writer = DBSpendUpdateWriter() + mock_prisma = MagicMock() + mock_prisma.get_request_status = MagicMock(return_value="success") + + def _payload(metadata: dict) -> dict: + return { + "request_id": "req-internal-1", + "user": "test-user", + "startTime": "2026-08-11T00:00:00", + "api_key": "test-key", + "model": "claude-sonnet-5", + "custom_llm_provider": "anthropic", + "model_group": "claude-sonnet-5", + "call_type": "acompletion", + "prompt_tokens": 100, + "completion_tokens": 10, + "spend": 0.05, + "metadata": json.dumps(metadata), + } + + internal = await writer._common_add_spend_log_transaction_to_daily_transaction( + payload=_payload({"internal_call_origin": "shadow_eval_judge"}), + prisma_client=mock_prisma, + type="user", + ) + user_sent = await writer._common_add_spend_log_transaction_to_daily_transaction( + payload=_payload({}), + prisma_client=mock_prisma, + type="user", + ) + + assert internal is not None and user_sent is not None + assert internal["spend"] == 0.05 + assert internal["prompt_tokens"] == 100 + assert internal["api_requests"] == 0 + assert internal["successful_requests"] == 0 + assert internal["failed_requests"] == 0 + assert internal["autorouter_savings_spend"] == 0.0 + assert user_sent["api_requests"] == 1 + assert user_sent["successful_requests"] == 1 diff --git a/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py index 3a995e276974..dd38a5223285 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py @@ -466,3 +466,251 @@ async def query_raw(self, sql: str, *params: object): end_date="2026-08-01", ) assert response.groups[0].tier_turns == expected + + +# --------------------------------------------------------------------------- +# Shadow eval endpoints +# --------------------------------------------------------------------------- + +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock + +from fastapi import HTTPException + +from litellm.proxy.management_endpoints.auto_router_endpoints import ( + _estimate_judge_cost_per_call, + _FALLBACK_JUDGE_COST_PER_CALL, + get_shadow_eval_job, + list_shadow_eval_jobs, + start_shadow_eval, + stop_shadow_eval_job, +) +from litellm.types.management_endpoints.auto_router_endpoints import StartShadowEvalRequest + +VIEWER = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, api_key="sk-view", user_id="viewer") +NON_ADMIN = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, api_key="sk-user", user_id="user") + + +def _shadow_router() -> MagicMock: + router = MagicMock() + router.auto_routers = {} + router.complexity_routers = {"my-router": [MagicMock()]} + router.adaptive_routers = {} + router.quality_routers = {} + router.model_group_alias = {} + router.get_model_list = MagicMock(return_value=None) + return router + + +def _job_record(**overrides: object) -> MagicMock: + record = MagicMock() + defaults = { + "id": "job-1", + "status": "running", + "router_name": "my-router", + "api_key_id": "key-hash", + "shadow_percentage": 10.0, + "request_count": 40, + "completed_count": 3, + "failed_count": 1, + "last_error": "judge call failed: boom", + "cost_estimate": 2.5, + "cost_actual": 0.03, + "created_at": datetime(2026, 8, 11, tzinfo=timezone.utc), + "ends_at": None, + "completed_at": None, + } + for key, value in {**defaults, **overrides}.items(): + setattr(record, key, value) + return record + + +def _shadow_prisma(existing_job=None, volume_rows=None, agg_rows=None) -> MagicMock: + prisma = MagicMock() + prisma.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=MagicMock()) + prisma.db.litellm_shadowevaljob.find_first = AsyncMock(return_value=existing_job) + prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=None) + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_shadowevaljob.create = AsyncMock(return_value=_job_record(status="pending")) + prisma.db.litellm_shadowevaljob.update = AsyncMock(return_value=_job_record(status="completed")) + + async def query_raw(sql: str, *params: object): + if "LiteLLM_DailyUserSpend" in sql: + return volume_rows if volume_rows is not None else [{"request_count": 0}] + return agg_rows if agg_rows is not None else [] + + prisma.db.query_raw = AsyncMock(side_effect=query_raw) + return prisma + + +def _start_request(**overrides: object) -> StartShadowEvalRequest: + payload = { + "api_key_id": "key-hash", + "router_name": "my-router", + "shadow_percentage": 10.0, + "judge_model": "anthropic/claude-sonnet-5", + "duration_days": 7, + } + payload.update(overrides) + return StartShadowEvalRequest.model_validate(payload) + + +@pytest.mark.asyncio +async def test_start_shadow_eval_creates_job_with_volume_scaled_estimate(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(volume_rows=[{"request_count": 7000}]) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + response = await start_shadow_eval(_start_request(), ADMIN) + + assert response.status == "pending" + assert response.estimated_request_count == 700 + per_call = _estimate_judge_cost_per_call(None, "anthropic/claude-sonnet-5") + assert response.estimated_cost == round(700 * per_call, 2) + create_data = prisma.db.litellm_shadowevaljob.create.call_args.kwargs["data"] + assert create_data["api_key_id"] == "key-hash" + assert create_data["created_by"] == "admin" + assert create_data["ends_at"] is not None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "caller,request_overrides,existing,expected_status", + [ + (NON_ADMIN, {}, None, 403), + (VIEWER, {}, None, 403), + (ADMIN, {"router_name": "not-a-router"}, None, 400), + (ADMIN, {"judge_model": "not/a real model!"}, None, 400), + (ADMIN, {"judge_model": "my-router"}, None, 400), + (ADMIN, {}, "existing", 409), + ], + ids=["non-admin", "view-only", "unknown-router", "unresolvable-judge", "router-as-judge", "already-active"], +) +async def test_start_shadow_eval_rejections( + monkeypatch: pytest.MonkeyPatch, caller, request_overrides, existing, expected_status +): + import litellm.proxy.proxy_server as proxy_server + + existing_job = _job_record() if existing else None + monkeypatch.setattr(proxy_server, "prisma_client", _shadow_prisma(existing_job=existing_job)) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + with pytest.raises(HTTPException) as exc: + await start_shadow_eval(_start_request(**request_overrides), caller) + assert exc.value.status_code == expected_status + + +@pytest.mark.asyncio +async def test_start_shadow_eval_rejects_a_key_this_proxy_does_not_know(monkeypatch: pytest.MonkeyPatch): + """A typo'd api_key_id would otherwise create a job no traffic can ever match.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma() + prisma.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + with pytest.raises(HTTPException) as exc: + await start_shadow_eval(_start_request(), ADMIN) + assert exc.value.status_code == 400 + assert "not a key on this proxy" in exc.value.detail + + +@pytest.mark.asyncio +async def test_start_shadow_eval_concurrent_unique_violation_is_a_409(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + from prisma.errors import UniqueViolationError + + prisma = _shadow_prisma() + prisma.db.litellm_shadowevaljob.create = AsyncMock( + side_effect=UniqueViolationError(MagicMock(message="unique constraint")) + ) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + with pytest.raises(HTTPException) as exc: + await start_shadow_eval(_start_request(), ADMIN) + assert exc.value.status_code == 409 + + +def test_estimate_falls_back_for_unpriced_judge_model(): + assert _estimate_judge_cost_per_call(None, "unknown/never-priced-model") == _FALLBACK_JUDGE_COST_PER_CALL + + +@pytest.mark.asyncio +async def test_get_shadow_eval_job_stratifies_by_tier_and_model(monkeypatch: pytest.MonkeyPatch): + 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}, + ] + prisma = _shadow_prisma(agg_rows=tier_rows) + prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record()) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + response = await get_shadow_eval_job("job-1", VIEWER) + + assert response.job_id == "job-1" + assert response.last_error == "judge call failed: boom" + assert response.results is not None + assert [s.group for s in response.results.by_tier] == ["SIMPLE", "REASONING"] + assert response.results.by_tier[0].shadow_win_rate_pct == 50.0 + assert response.results.overall_shadow_win_rate_pct == 40.0 + assert response.results.overall_tie_rate_pct == 20.0 + + +@pytest.mark.asyncio +async def test_get_shadow_eval_job_404s_and_gates_on_role(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + + monkeypatch.setattr(proxy_server, "prisma_client", _shadow_prisma()) + + with pytest.raises(HTTPException) as missing: + await get_shadow_eval_job("nope", VIEWER) + assert missing.value.status_code == 404 + + with pytest.raises(HTTPException) as forbidden: + await get_shadow_eval_job("job-1", NON_ADMIN) + assert forbidden.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_list_shadow_eval_jobs_omits_results(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record()]) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) + + assert len(jobs) == 1 + assert jobs[0].results is None + find_kwargs = prisma.db.litellm_shadowevaljob.find_many.call_args.kwargs + assert find_kwargs["take"] == 50 + + +@pytest.mark.asyncio +async def test_stop_shadow_eval_completes_active_job_and_rejects_finished(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma() + prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record(status="running")) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + stopped = await stop_shadow_eval_job("job-1", ADMIN) + assert stopped.status == "completed" + update_kwargs = prisma.db.litellm_shadowevaljob.update.call_args.kwargs + assert update_kwargs["data"]["status"] == "completed" + + prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record(status="completed")) + with pytest.raises(HTTPException) as exc: + await stop_shadow_eval_job("job-1", ADMIN) + assert exc.value.status_code == 400 + + with pytest.raises(HTTPException) as forbidden: + await stop_shadow_eval_job("job-1", VIEWER) + assert forbidden.value.status_code == 403 diff --git a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py index 40ca7e3a64e0..2b35e080056d 100644 --- a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py +++ b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py @@ -480,8 +480,9 @@ def test_load_from_azure_key_vault_missing_uri_failure_is_swallowed(monkeypatch) # --------------------------------------------------------------------------- -def test_cost_tracking_adds_two_callbacks_when_prisma_set(monkeypatch): +def test_cost_tracking_adds_db_and_shadow_eval_callbacks_when_prisma_set(monkeypatch): import litellm + from litellm.integrations.shadow_eval_logger import ShadowEvalLogger fake_prisma = MagicMock() monkeypatch.setattr(ps, "prisma_client", fake_prisma, raising=False) @@ -496,15 +497,53 @@ def test_cost_tracking_adds_two_callbacks_when_prisma_set(monkeypatch): observed = { "added_to_callbacks": len(litellm.callbacks) - before_callbacks, "added_to_async_success": len(litellm._async_success_callback) - before_async, + "shadow_eval_loggers": sum(isinstance(cb, ShadowEvalLogger) for cb in litellm.callbacks), "prisma_was_set": True, } assert normalize(observed) == { - "added_to_callbacks": 1, + "added_to_callbacks": 2, "added_to_async_success": 1, + "shadow_eval_loggers": 1, "prisma_was_set": True, } +def test_cost_tracking_twice_registers_one_shadow_eval_logger(monkeypatch): + import litellm + from litellm.integrations.shadow_eval_logger import ShadowEvalLogger + + monkeypatch.setattr(ps, "prisma_client", MagicMock(), raising=False) + monkeypatch.setattr(litellm, "callbacks", [], raising=False) + monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False) + + cost_tracking() + cost_tracking() + + assert sum(isinstance(cb, ShadowEvalLogger) for cb in litellm.callbacks) == 1 + + +@pytest.mark.asyncio +async def test_registration_starts_the_lifecycle_loop_on_the_registered_instance(monkeypatch): + """The loop must run on the instance that is actually registered; a loop on a + discarded duplicate would poll the DB while never seeing request traffic.""" + import litellm + from litellm.integrations.shadow_eval_logger import ShadowEvalLogger + + monkeypatch.setattr(ps, "prisma_client", MagicMock(), raising=False) + monkeypatch.setattr(litellm, "callbacks", [], raising=False) + monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False) + + cost_tracking() + registered = [cb for cb in litellm.callbacks if isinstance(cb, ShadowEvalLogger)] + assert len(registered) == 1 + task = registered[0]._lifecycle_task + assert task is not None and not task.done() + + cost_tracking() + assert registered[0]._lifecycle_task is task + task.cancel() + + def test_cost_tracking_no_op_when_prisma_missing(monkeypatch): """Without a prisma_client cost_tracking is a no-op — not an error.""" import litellm diff --git a/tests/test_litellm/router_strategy/test_complexity_router.py b/tests/test_litellm/router_strategy/test_complexity_router.py index 94b6b68855ba..6ec7aa581e8b 100644 --- a/tests/test_litellm/router_strategy/test_complexity_router.py +++ b/tests/test_litellm/router_strategy/test_complexity_router.py @@ -3222,98 +3222,6 @@ async def test_semantic_embedding_error_falls_back_to_scoring(self, mock_router_ assert result.model in {"gpt-4o-mini", "gpt-4o", "claude-sonnet-4-20250514", "o1-preview"} -class TestSubCallMetadataSanitization: - """The proxy cost callback must not be able to recover the parent budget reservation - from sub-call metadata, in either of the shapes it knows how to read.""" - - def test_cost_callback_cannot_recover_reservation_from_sanitized_metadata(self): - from litellm.proxy._types import UserAPIKeyAuth - from litellm.proxy.hooks.proxy_track_cost_callback import ( - _get_budget_reservation_from_metadata, - ) - from litellm.router_strategy.complexity_router.complexity_router import ( - _classifier_call_metadata, - ) - - reservation = {"reserved_cost": 1.0} - auth_shapes = ( - {"models": ["gpt-4o"], "budget_reservation": dict(reservation)}, - UserAPIKeyAuth(api_key="sk-abc", budget_reservation=dict(reservation)), - ) - for auth in auth_shapes: - metadata = { - "user_api_key_hash": "hash-abc", - "user_api_key_budget_reservation": dict(reservation), - "user_api_key_auth": auth, - } - assert _get_budget_reservation_from_metadata(metadata) == reservation - - sanitized = _classifier_call_metadata(metadata) - assert sanitized is not None - assert sanitized["user_api_key_auth"] is not None - assert _get_budget_reservation_from_metadata(sanitized) is None - - def test_absent_parent_bucket_stays_empty(self): - """An absent bucket must not be materialized just to carry the origin. - - The embedding path passes both buckets, and get_litellm_metadata_from_kwargs - prefers litellm_metadata whenever it is truthy, backfilling only user_api_key* - keys from metadata. Returning an origin-only dict here would make a chat - completions parent's empty litellm_metadata win and silently drop - requester_ip_address, tags and spend_logs_metadata from the classifier's row.""" - from litellm.router_strategy.complexity_router.complexity_router import ( - _classifier_call_metadata, - ) - - for absent in (None, {}): - assert _classifier_call_metadata(absent) == {} - - def test_classifier_buckets_keep_non_spend_fields_on_a_chat_completions_parent(self): - """Drives the real resolver over the buckets the embedding classifier builds.""" - from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs - from litellm.router_strategy.complexity_router.complexity_router import ( - _classifier_call_metadata, - ) - - parent = { - "user_api_key": "sk-abc", - "requester_ip_address": "10.0.0.1", - "spend_logs_metadata": {"team_note": "keep me"}, - "tags": ["prod"], - } - resolved = get_litellm_metadata_from_kwargs( - { - "litellm_params": { - "metadata": _classifier_call_metadata(parent), - "litellm_metadata": _classifier_call_metadata(None), - } - } - ) - assert resolved["internal_call_origin"] == "autorouter_classifier" - assert resolved["requester_ip_address"] == "10.0.0.1" - assert resolved["spend_logs_metadata"] == {"team_note": "keep me"} - assert resolved["tags"] == ["prod"] - - def test_sanitized_auth_keeps_access_group_fields_and_leaves_original_untouched(self): - from litellm.proxy._types import UserAPIKeyAuth - from litellm.router_strategy.complexity_router.complexity_router import ( - _classifier_call_metadata, - ) - - auth = UserAPIKeyAuth( - api_key="sk-abc", - team_id="team-1", - budget_reservation={"reserved_cost": 1.0}, - ) - sanitized = _classifier_call_metadata({"user_api_key_auth": auth}) - assert sanitized is not None - sanitized_auth = sanitized["user_api_key_auth"] - assert sanitized_auth.budget_reservation is None - assert sanitized_auth.team_id == "team-1" - assert sanitized_auth.api_key == auth.api_key - assert auth.budget_reservation == {"reserved_cost": 1.0} - - class TestRoutingDecisionCauseLogging: """The info log must name what drove each routing decision so an operator can tell a literal keyword match, a semantic keyword match, and the complexity scorer apart. diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 326dfb80a7e9..fc4dfc87a7c2 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -807,6 +807,94 @@ export interface paths { patch?: never; trace?: never; }; + "/auto_router/shadow_eval": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * List Shadow Eval Jobs + * @description List shadow eval jobs, newest first. Results are omitted; fetch a single job for them. + */ + get: operations["list_shadow_eval_jobs_auto_router_shadow_eval_get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/auto_router/shadow_eval/start": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Start Shadow Eval + * @description Start a pre-adoption shadow eval: duplicate a sampled slice of a key's live + * traffic through an auto-router, judge real vs. shadow responses blind, and + * stratify win rates by the router's tier classification. + * + * The shadow responses are never served to users. The job samples traffic for + * duration_days (or until stopped via /auto_router/shadow_eval/{job_id}/stop), + * then completes itself. Judge calls bill to the shadowed key; the estimate + * returned here prices them from the key's trailing request volume scaled to + * the requested duration. + */ + post: operations["start_shadow_eval_auto_router_shadow_eval_start_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/auto_router/shadow_eval/{job_id}": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Get Shadow Eval Job + * @description Status, counters, and stratified results of one shadow eval job. + */ + get: operations["get_shadow_eval_job_auto_router_shadow_eval__job_id__get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/auto_router/shadow_eval/{job_id}/stop": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Stop Shadow Eval Job + * @description Stop an active shadow eval job. Existing verdicts are kept; sampling halts within ~10s. + */ + post: operations["stop_shadow_eval_job_auto_router_shadow_eval__job_id__stop_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/auto_router/test_routing": { parameters: { query?: never; @@ -25270,6 +25358,73 @@ export interface components { /** Tools */ tools?: components["schemas"]["ChatCompletionToolParam"][]; }; + /** + * GetShadowEvalJobResponse + * @description Status and, once available, results of a shadow-eval job. + * + * Validates directly from the prisma job record (job_id reads the row's id), so the + * endpoint needs no hand-written row-to-response mapping. + */ + GetShadowEvalJobResponse: { + /** + * Api Key Id + * @description The hashed virtual key whose traffic this job evaluates, and only that key's + */ + api_key_id: string; + /** Completed At */ + completed_at?: string | null; + /** + * Completed Count + * @description Verdicts written so far + */ + completed_count: number; + /** + * Cost Actual + * @description Running total of judge-call spend for this job + * @default 0 + */ + cost_actual: number; + /** Cost Estimate */ + cost_estimate?: number | null; + /** + * Created At + * Format: date-time + */ + created_at: string; + /** + * Ends At + * @description When the job stops sampling on its own + */ + ends_at?: string | null; + /** + * Failed Count + * @description Shadow or judge calls that errored and were skipped + */ + failed_count: number; + /** Job Id */ + job_id: string; + /** + * Last Error + * @description The most recent shadow or judge failure, so a growing failed_count is diagnosable + */ + last_error?: string | null; + /** + * Request Count + * @description Total requests observed on the shadowed key since the job started + */ + request_count: number; + /** @description Present once at least one verdict has been recorded */ + results?: components["schemas"]["ShadowEvalResult"] | null; + /** Router Name */ + router_name: string; + /** Shadow Percentage */ + shadow_percentage: number; + /** + * Status + * @enum {string} + */ + status: "pending" | "running" | "completed"; + }; /** * GetTeamMemberPermissionsResponse * @description Response to get the team member permissions for a team @@ -32516,6 +32671,45 @@ export interface components { /** Timeout */ timeout?: number | null; }; + /** + * ShadowEvalResult + * @description Stratified results of a shadow-eval job's verdicts so far. + */ + ShadowEvalResult: { + /** By Current Model */ + by_current_model: components["schemas"]["ShadowEvalSlice"][]; + /** By Tier */ + by_tier: components["schemas"]["ShadowEvalSlice"][]; + /** Overall Shadow Win Rate Pct */ + overall_shadow_win_rate_pct: number; + /** Overall Tie Rate Pct */ + overall_tie_rate_pct: number; + }; + /** + * ShadowEvalSlice + * @description Judge outcomes for one slice of a job's verdicts (a router tier, or one of the + * models the shadowed key currently uses). + */ + ShadowEvalSlice: { + /** Avg Judge Confidence */ + avg_judge_confidence: number; + /** Group */ + group: string; + /** + * Real Win Rate Pct + * @description Share of judged turns where the real (control) model won + */ + real_win_rate_pct: number; + /** + * Shadow Win Rate Pct + * @description Share of judged turns where the shadowed router's pick won + */ + shadow_win_rate_pct: number; + /** Tie Rate Pct */ + tie_rate_pct: number; + /** Turn Count */ + turn_count: number; + }; /** * Skill * @description Represents a skill from the Anthropic Skills API @@ -32689,6 +32883,62 @@ export interface components { /** Simple Medium */ simple_medium: number; }; + /** + * StartShadowEvalRequest + * @description Start shadowing a key's traffic through an auto-router for blind comparison. + */ + StartShadowEvalRequest: { + /** + * Api Key Id + * @description The hashed virtual key whose traffic will be shadowed. Shadow evaluation runs ONLY on this key's traffic; requests made with any other key are not sampled. + */ + api_key_id: string; + /** + * Duration Days + * @description How many days the job samples traffic before stopping itself + * @default 7 + */ + duration_days: number; + /** + * Judge Model + * @description Model used to blindly judge real vs. shadow responses. The judge only compares two answers, so a mid-tier model (Claude Sonnet or GPT-4o class) is the sweet spot: small/nano-class models produce unreliable or malformed verdicts, while frontier reasoning models add cost without changing outcomes. + * @default anthropic/claude-sonnet-5 + */ + judge_model: string; + /** + * Router Name + * @description The auto-router config to shadow requests through + */ + router_name: string; + /** + * Shadow Percentage + * @description Percentage of the key's requests to duplicate through the router + */ + shadow_percentage: number; + }; + /** + * StartShadowEvalResponse + * @description Acknowledgement that a shadow-eval job was created, with an upfront cost estimate. + */ + StartShadowEvalResponse: { + /** + * Estimated Cost + * @description Estimated dollar cost of the judge calls this job will make + */ + estimated_cost: number; + /** + * Estimated Request Count + * @description Requests expected to be shadowed, based on the key's recent request volume + */ + estimated_request_count: number; + /** Job Id */ + job_id: string; + /** + * Status + * @enum {string} + */ + status: "pending" | "running" | "completed"; + }; /** * SuccessfulKeyUpdate * @description Successfully updated key with its updated information @@ -36964,6 +37214,135 @@ export interface operations { }; }; }; + list_shadow_eval_jobs_auto_router_shadow_eval_get: { + parameters: { + query?: { + /** @description Filter to jobs shadowing this key */ + api_key_id?: string | null; + /** @description Newest jobs to return */ + limit?: number; + }; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["GetShadowEvalJobResponse"][]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; + start_shadow_eval_auto_router_shadow_eval_start_post: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["StartShadowEvalRequest"]; + }; + }; + responses: { + /** @description Successful Response */ + 201: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["StartShadowEvalResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; + get_shadow_eval_job_auto_router_shadow_eval__job_id__get: { + parameters: { + query?: never; + header?: never; + path: { + job_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["GetShadowEvalJobResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; + stop_shadow_eval_job_auto_router_shadow_eval__job_id__stop_post: { + parameters: { + query?: never; + header?: never; + path: { + job_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["GetShadowEvalJobResponse"]; + }; + }; + /** @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;