diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260811172448_add_shadow_eval/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260811172448_add_shadow_eval/migration.sql new file mode 100644 index 00000000000..26932addb42 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260811172448_add_shadow_eval/migration.sql @@ -0,0 +1,49 @@ +-- CreateTable +CREATE TABLE "LiteLLM_ShadowEvalJob" ( + "id" TEXT NOT NULL, + "api_key_id" TEXT NOT NULL, + "router_name" TEXT NOT NULL, + "judge_model" TEXT NOT NULL, + "shadow_percentage" DOUBLE PRECISION NOT NULL, + "max_turns" INTEGER NOT NULL, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "created_by" TEXT, + "ends_at" TIMESTAMP(3) NOT NULL, + "stopped_at" TIMESTAMP(3), + + CONSTRAINT "LiteLLM_ShadowEvalJob_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "LiteLLM_ShadowEvalAttempt" ( + "id" TEXT NOT NULL, + "job_id" TEXT NOT NULL, + "request_id" TEXT NOT NULL, + "outcome" TEXT NOT NULL, + "tier" TEXT, + "real_model" TEXT, + "shadow_model" TEXT, + "confidence" DOUBLE PRECISION, + "judge_cost" DOUBLE PRECISION NOT NULL DEFAULT 0, + "error" TEXT, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "LiteLLM_ShadowEvalAttempt_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "LiteLLM_ShadowEvalJob_api_key_id_idx" ON "LiteLLM_ShadowEvalJob"("api_key_id"); + +-- CreateIndex +CREATE INDEX "LiteLLM_ShadowEvalJob_created_at_idx" ON "LiteLLM_ShadowEvalJob"("created_at"); + +-- CreateIndex +CREATE INDEX "LiteLLM_ShadowEvalAttempt_job_id_idx" ON "LiteLLM_ShadowEvalAttempt"("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. Active means not yet +-- stopped; the start endpoint stamps stopped_at on expired jobs before creating. +CREATE UNIQUE INDEX "LiteLLM_ShadowEvalJob_one_active_per_key" + ON "LiteLLM_ShadowEvalJob"("api_key_id") WHERE "stopped_at" IS NULL; diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index 854602f5380..79d778fb464 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -1450,6 +1450,44 @@ 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. The job row is immutable config plus +// stopped_at; every count, status, and spend figure is derived from the append-only +// attempt rows, so nothing can disagree across pods or stop races. +model LiteLLM_ShadowEvalJob { + id String @id @default(cuid()) + api_key_id String // hashed virtual key whose traffic is shadowed + router_name String + judge_model String + shadow_percentage Float + max_turns Int // sample budget: judge at most this many turns + created_at DateTime @default(now()) + created_by String? + ends_at DateTime + stopped_at DateTime? + + @@index([api_key_id]) + @@index([created_at]) +} + +// One row per sampled pipeline: a blind verdict (real | shadow | tie) or an error. +model LiteLLM_ShadowEvalAttempt { + id String @id @default(cuid()) + job_id String + request_id String // the judged real request + outcome String // real | shadow | tie | error + tier String? // router's tier for the prompt, when classified + real_model String? + shadow_model String? + confidence Float? + judge_cost Float @default(0) + error String? + 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 00000000000..c7b89e0e9b0 --- /dev/null +++ b/litellm/integrations/shadow_eval_logger.py @@ -0,0 +1,563 @@ +"""Shadow Eval Logger: samples a shadowed key's successful chat requests, duplicates each +through the auto-router in a detached task, blind-judges real vs shadow, and appends one +``LiteLLM_ShadowEvalAttempt`` row (verdict or error) as the feature's only hot-path write. +Counts, status, and spend derive from those rows at read time, so nothing can disagree +across pods or stop races; the hook reads active jobs through a short-TTL cache.""" + +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.caching.in_memory_cache import InMemoryCache +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY +from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs +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 + +# A job starting, stopping, or hitting its turn budget propagates to sampling within one +# TTL; the turn budget can overshoot by at most one TTL of in-flight samples per pod. +_JOBS_CACHE_TTL_SECONDS: Final = 10 + +# 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 attempt is lost to an error row. +JUDGE_MAX_OUTPUT_TOKENS: Final = 500 + +_MAX_ERROR_CHARS: Final = 500 + +_EMPTY_METADATA: Final[Mapping[str, object]] = MappingProxyType({}) + +_SAMPLED_CALL_TYPES: Final = frozenset({"completion", "acompletion"}) + +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: the 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?" + ) + + +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 + + +@dataclass(frozen=True, slots=True) +class _CallFailure: + """A shadow or judge call that produced no usable response. cost carries any judge + spend the failed attempt still billed, so job-level judge_spend never undercounts.""" + + error: str + cost: float = 0.0 + + +@dataclass(frozen=True, slots=True) +class _ShadowResponse: + """A successful shadow call, with what the attempt 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: + """One active job as the sampling path needs it: immutable config plus the attempt + count as of the cache fill (the turn budget's staleness is bounded by the cache TTL).""" + + id: str + router_name: str + shadow_percentage: float + judge_model: str + max_turns: int + ends_at: datetime + attempts: int + + +def _as_utc(value: datetime) -> datetime: + return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value + + +_jobs_cache: Final = InMemoryCache(max_size_in_memory=4, default_ttl=_JOBS_CACHE_TTL_SECONDS) +_JOBS_CACHE_KEY: Final = "shadow_eval:active_jobs" + + +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, + jobs_cache: InMemoryCache | 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 + self._jobs_cache = jobs_cache or _jobs_cache + self._inflight_shadow_tasks: int = 0 + # Starts per job since the last cache fill, never decremented within a + # generation; the refill absorbs written rows and resets. + self._job_starts: dict[str, int] = {} # mutable-ok: per-generation counter + + async def _active_jobs(self) -> Mapping[str, ActiveShadowEvalJob]: + """Active jobs by api_key_id, cache-first. A DB fault returns empty without + caching, so sampling pauses for that request and the next one retries.""" + cached: Final = await self._jobs_cache.async_get_cache(_JOBS_CACHE_KEY) + if cached is not None: + return cached # pyright: ignore[reportReturnType] # cache stores exactly this mapping shape + prisma: Final = self._prisma_provider() + if prisma is None: + return _EMPTY_JOBS + try: + records: Final = await prisma.db.litellm_shadowevaljob.find_many( + where={ # mutable-ok: Prisma filter + "stopped_at": None, + "ends_at": {"gt": datetime.now(timezone.utc)}, # mutable-ok: Prisma filter + }, + ) + grouped: Final = ( + await prisma.db.litellm_shadowevalattempt.group_by( + by=["job_id"], + count=True, + where={"job_id": {"in": [str(record.id) for record in records]}}, # mutable-ok: Prisma filter + ) + if records + else () + ) + attempt_counts: Final = {str(row["job_id"]): int(row["_count"]["_all"]) for row in grouped or []} + jobs: Final = { + 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), + max_turns=int(record.max_turns), + ends_at=_as_utc(record.ends_at), + attempts=attempt_counts.get(str(record.id), 0), + ) + for record in records or [] + } + await self._jobs_cache.async_set_cache(_JOBS_CACHE_KEY, jobs) + self._job_starts = {} # rebind-ok: new generation, counts absorbed into the fill + return jobs + except Exception as e: # noqa: BLE001 # a DB blip must never break request logging + verbose_logger.debug("shadow_eval: active-job read failed: %s", e) + return _EMPTY_JOBS + + #### 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 + raw_meta: Final = get_litellm_metadata_from_kwargs(dict(kwargs)) # mutable-ok: helper needs dict + request_metadata: Final = raw_meta if isinstance(raw_meta, 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 + # redaction rewrites logged content before callbacks run, so this hook + # only ever sees placeholders for a redacted request + if should_redact_message_logging(dict(kwargs)): # mutable-ok: predicate takes a plain dict + return + metadata: Final = payload.get("metadata") or _EMPTY_METADATA + api_key_hash: Final = metadata.get("user_api_key_hash") + if not api_key_hash: + return + job: Final = (await self._active_jobs()).get(str(api_key_hash)) + if job is None: + return + if datetime.now(timezone.utc) >= job.ends_at: + return + if job.attempts + self._job_starts.get(job.id, 0) >= job.max_turns: + return + request_id: Final = payload.get("id") or "" + if not request_id: + return + if not _sample_hits(request_id, job.id, job.shadow_percentage): + return + if payload.get("call_type") not in _SAMPLED_CALL_TYPES: + 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 + raw_messages: Final = kwargs.get("messages") + self._job_starts[job.id] = self._job_starts.get(job.id, 0) + 1 + 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 + + #### the detached pipeline: one attempt row per sampled request, verdict or error #### + + 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: + """Budget gate -> shadow call -> blind judge -> one attempt row. The prisma gate + sits above the dispatch so no provider spend happens without a place to record + the outcome, and the budget read lives here rather than in the success hook.""" + 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._record_attempt(prisma, job, request_id, outcome="error", error=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._record_attempt( + prisma, + job, + request_id, + outcome="error", + error=verdict.error, + shadow=shadow, + judge_cost=verdict.cost, + ) + return + await self._record_attempt( + prisma, + job, + request_id, + outcome=verdict.preference, + shadow=shadow, + real_model=real_model, + confidence=verdict.confidence, + judge_cost=verdict.cost, + ) + except Exception as e: # noqa: BLE001 # detached task: record what happened, never raise + verbose_logger.debug("shadow_eval: pipeline failed for %s: %s", request_id, e) + await self._record_attempt(prisma, job, request_id, outcome="error", error=f"pipeline error: {e}") + + @staticmethod + async def _record_attempt( + prisma: "PrismaClient | None", + job: ActiveShadowEvalJob, + request_id: str, + *, + outcome: str, + shadow: _ShadowResponse | None = None, + real_model: str = "", + confidence: float | None = None, + judge_cost: float = 0.0, + error: str | None = None, + ) -> None: + if prisma is None: + return + try: + await prisma.db.litellm_shadowevalattempt.create( + data={ # mutable-ok: Prisma payload + "job_id": job.id, + "request_id": request_id, + "outcome": outcome, + "tier": shadow.tier if shadow else None, + "real_model": real_model or None, + "shadow_model": shadow.model if shadow else None, + "confidence": confidence, + "judge_cost": judge_cost, + "error": error[:_MAX_ERROR_CHARS] if error else None, + } + ) + except Exception as e: # noqa: BLE001 # a lost row degrades sample size, nothing can disagree with it + verbose_logger.debug("shadow_eval: attempt write failed for %s: %s", request_id, 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(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 recorded error, never a spend multiplier + **shadow_params, + ) + except Exception as e: # noqa: BLE001 # provider errors become error rows, not crashes + verbose_logger.debug("shadow_eval: router call failed: %s", e) + return _CallFailure(f"shadow router call failed: {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 become error rows, not crashes + 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 become error rows + verbose_logger.debug("shadow_eval: unparseable judge verdict: %s", e) + return _CallFailure(f"unparseable judge verdict: {e}", cost=_judge_call_cost(response)) + 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) + + +_EMPTY_JOBS: Final[Mapping[str, ActiveShadowEvalJob]] = MappingProxyType({}) + + +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 00000000000..6815727de69 --- /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 00000000000..4ad8d719402 --- /dev/null +++ b/litellm/litellm_core_utils/llm_judge.py @@ -0,0 +1,87 @@ +"""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. + Sampling preferences are advisory: models that removed sampling params (e.g. + claude-sonnet-5) drop them instead of rejecting the judge call.""" + 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=[], + drop_params=True, + **params, + ) + return await litellm.acompletion(model=judge_model, messages=messages, num_retries=0, drop_params=True, **params) diff --git a/litellm/proxy/db/autorouter_session_rollup.py b/litellm/proxy/db/autorouter_session_rollup.py index 9732b1d7402..96192b884d8 100644 --- a/litellm/proxy/db/autorouter_session_rollup.py +++ b/litellm/proxy/db/autorouter_session_rollup.py @@ -24,6 +24,7 @@ from itertools import groupby 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 b0130db232a..b2b72c1cac4 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.caching import RedisCache 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 @@ class DBSpendUpdateWriter: 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 @@ class DBSpendUpdateWriter: 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 1907bb19abf..e3f67f0024b 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 @@ Return ONLY valid JSON in this exact format: _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 @@ class LLMAsAJudgeGuardrail(CustomGuardrail): "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/hooks/parallel_request_limiter_v3.py b/litellm/proxy/hooks/parallel_request_limiter_v3.py index 3fd2adda480..f62dbec2e85 100644 --- a/litellm/proxy/hooks/parallel_request_limiter_v3.py +++ b/litellm/proxy/hooks/parallel_request_limiter_v3.py @@ -24,7 +24,7 @@ from typing import ( from litellm import DualCache from litellm._logging import verbose_proxy_logger -from litellm.constants import DYNAMIC_RATE_LIMIT_ERROR_THRESHOLD_PER_MINUTE +from litellm.constants import DYNAMIC_RATE_LIMIT_ERROR_THRESHOLD_PER_MINUTE, INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.prompt_templates.common_utils import ( get_str_from_messages, @@ -2991,6 +2991,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): rate_limit_type: Literal["output", "input", "total"], ) -> list[RedisPipelineIncrementOperation]: """Build Redis pipeline increment ops for TPM / parallel-request counters.""" + from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs from litellm.proxy.common_utils.callback_utils import ( get_model_group_from_litellm_kwargs, ) @@ -2998,6 +2999,11 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): # Get metadata from standard_logging_object - this correctly handles both # 'metadata' and 'litellm_metadata' fields from litellm_params standard_logging_object: Final = kwargs.get("standard_logging_object") or {} + request_metadata: Final = get_litellm_metadata_from_kwargs(kwargs) + if request_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY): + # Internal sub-calls bill spend to the caller but are not the caller's + # traffic; charging them here would let background evals eat TPM headroom. + return [] standard_logging_metadata: Final = standard_logging_object.get("metadata") or {} model_group: Final = get_model_group_from_litellm_kwargs(kwargs) diff --git a/litellm/proxy/management_endpoints/auto_router_endpoints.py b/litellm/proxy/management_endpoints/auto_router_endpoints.py index 8b6aafea751..cb0e8dba62a 100644 --- a/litellm/proxy/management_endpoints/auto_router_endpoints.py +++ b/litellm/proxy/management_endpoints/auto_router_endpoints.py @@ -13,6 +13,7 @@ from pydantic import BaseModel, TypeAdapter from litellm._logging import verbose_proxy_logger from litellm.exceptions import BudgetExceededError +from litellm.litellm_core_utils.llm_judge import router_resolves_model from litellm.proxy._types import ( CommonProxyErrors, LiteLLM_TeamTable, @@ -39,11 +40,16 @@ from litellm.types.management_endpoints.auto_router_endpoints import ( AutoRouterRoutingTestRequest, AutoRouterRoutingTestResponse, RequestComplexityRouterConfig, + ShadowEvalJobResponse, + ShadowEvalResult, + ShadowEvalSlice, + StartShadowEvalRequest, ) 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 +394,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 +429,335 @@ 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. +# The job row is immutable config plus stopped_at; status, counts, spend, and errors +# are derived from the append-only attempt rows, so reads here are aggregations +# bounded by each job's max_turns through the attempt table's job_id index. +# --------------------------------------------------------------------------- + + +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 error 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 + import litellm + + 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 _is_unique_violation(error: Exception) -> bool: + """Whether a Prisma create failed on a unique index. One active job per key 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 rather than a 500.""" + try: + from prisma.errors import UniqueViolationError + except ImportError: + return "unique constraint" in str(error).lower() or "P2002" in str(error) + return isinstance(error, UniqueViolationError) + + +class _AttemptAggRow(BaseModel): + grp: str + turn_count: int + real_wins: int + shadow_wins: int + ties: int + avg_confidence: float | None + + +_ATTEMPT_AGG_ROWS: Final = TypeAdapter(list[_AttemptAggRow]) + +_ATTEMPT_AGG_SELECT: Final = """ + COUNT(*)::int AS turn_count, + COUNT(*) FILTER (WHERE outcome = 'real')::int AS real_wins, + COUNT(*) FILTER (WHERE outcome = 'shadow')::int AS shadow_wins, + COUNT(*) FILTER (WHERE outcome = 'tie')::int AS ties, + AVG(confidence)::float AS avg_confidence +FROM "LiteLLM_ShadowEvalAttempt" +WHERE job_id = $1 AND outcome != 'error' +GROUP BY 1 +""" + +_ATTEMPT_AGG_BY_TIER_SQL: Final = "SELECT COALESCE(tier, 'UNCLASSIFIED') AS grp," + _ATTEMPT_AGG_SELECT +_ATTEMPT_AGG_BY_MODEL_SQL: Final = "SELECT COALESCE(real_model, 'unknown') AS grp," + _ATTEMPT_AGG_SELECT + +_SWEEP_FINISHED_JOBS_SQL: Final = """ +UPDATE "LiteLLM_ShadowEvalJob" j SET stopped_at = NOW() +WHERE j.api_key_id = $1 AND j.stopped_at IS NULL + AND ( + j.ends_at <= NOW() + OR (SELECT COUNT(*) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = j.id) >= j.max_turns + ) +""" + +_ATTEMPT_TOTALS_SQL: Final = """ +SELECT + COUNT(*) FILTER (WHERE outcome != 'error')::int AS judged_count, + COUNT(*) FILTER (WHERE outcome = 'error')::int AS error_count, + COALESCE(SUM(judge_cost), 0)::float AS judge_spend +FROM "LiteLLM_ShadowEvalAttempt" +WHERE job_id = $1 +""" + + +class _AttemptTotalsRow(BaseModel): + judged_count: int + error_count: int + judge_spend: float + + +_ATTEMPT_TOTALS_ROWS: Final = TypeAdapter(list[_AttemptTotalsRow]) + + +def _pct_of(numerator: int, denominator: int) -> float: + return _pct(numerator, denominator) + + +def _slices(rows: Sequence[_AttemptAggRow]) -> tuple[ShadowEvalSlice, ...]: + return tuple( + ShadowEvalSlice( + group=row.grp, + turn_count=row.turn_count, + real_win_rate_pct=_pct_of(row.real_wins, row.turn_count), + shadow_win_rate_pct=_pct_of(row.shadow_wins, row.turn_count), + tie_rate_pct=_pct_of(row.ties, row.turn_count), + avg_judge_confidence=round(row.avg_confidence or 0.0, 3), + ) + 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". Reads are bounded by the job's own attempts (<= max_turns) via the job_id index.""" + by_tier: Final = _ATTEMPT_AGG_ROWS.validate_python( + await prisma_client.db.query_raw(_ATTEMPT_AGG_BY_TIER_SQL, job_id) or () + ) + if not by_tier: + return None + by_model: Final = _ATTEMPT_AGG_ROWS.validate_python( + await prisma_client.db.query_raw(_ATTEMPT_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_of(sum(r.shadow_wins for r in by_tier), total_turns), + overall_tie_rate_pct=_pct_of(sum(r.ties for r in by_tier), total_turns), + ) + + +@router.post( + "/auto_router/shadow_eval/start", + tags=("auto router",), + dependencies=(Depends(user_api_key_auth),), + response_model=ShadowEvalJobResponse, + status_code=status.HTTP_201_CREATED, +) +async def start_shadow_eval( + data: StartShadowEvalRequest, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], +) -> ShadowEvalJobResponse: + """ + 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 and by the incumbent model. + + Shadow responses are never served to users. The job samples until it has judged + max_turns turns, reaches the end of its window, or is stopped; sampling changes + propagate to pods within about 10 seconds. Shadow and judge calls bill to the + shadowed key but are excluded from request counts and auto-router adoption metrics. + """ + 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" + ), + ) + + # A job that expired or exhausted its turn budget stopped sampling on its own, but + # still holds the one-active-per-key partial unique index until stamped; free it so + # a new eval can start. + await prisma_client.db.execute_raw(_SWEEP_FINISHED_JOBS_SQL, data.api_key_id) + active: Final = await prisma_client.db.litellm_shadowevaljob.find_first( + where={"api_key_id": data.api_key_id, "stopped_at": None}, # mutable-ok: Prisma filter + ) + if active is not None: + raise HTTPException( + status_code=409, + detail=f"Key already has an active shadow eval job ({active.id}). Stop it first.", + ) + now: Final = datetime.now(timezone.utc) + 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, + "judge_model": data.judge_model, + "shadow_percentage": data.shadow_percentage, + "max_turns": data.max_turns, + "created_by": user_api_key_dict.user_id, + "ends_at": now + timedelta(days=data.duration_days), + } + ) + 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 ShadowEvalJobResponse.model_validate(job, from_attributes=True) + + +@router.get( + "/auto_router/shadow_eval", + tags=("auto router",), + dependencies=(Depends(user_api_key_auth),), + response_model=list[ShadowEvalJobResponse], +) +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[ShadowEvalJobResponse, ...]: + """List shadow eval jobs, newest first. Counts and results ride the detail endpoint only.""" + 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) + records: Final = await prisma_client.db.litellm_shadowevaljob.find_many( + where={"api_key_id": api_key_id} if api_key_id else {}, # mutable-ok: Prisma filter + order={"created_at": "desc"}, # mutable-ok: Prisma order + take=limit, + ) + return tuple(ShadowEvalJobResponse.model_validate(record, from_attributes=True) for record in records or ()) + + +@router.get( + "/auto_router/shadow_eval/{job_id}", + tags=("auto router",), + dependencies=(Depends(user_api_key_auth),), + response_model=ShadowEvalJobResponse, +) +async def get_shadow_eval_job( + job_id: str, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], +) -> ShadowEvalJobResponse: + """One job with derived counts, judge spend, latest error, and stratified results.""" + 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}") + totals: Final = _ATTEMPT_TOTALS_ROWS.validate_python( + await prisma_client.db.query_raw(_ATTEMPT_TOTALS_SQL, job_id) or () + ) + latest_error: Final = await prisma_client.db.litellm_shadowevalattempt.find_first( + where={"job_id": job_id, "outcome": "error"}, # mutable-ok: Prisma filter + order={"created_at": "desc"}, # mutable-ok: Prisma order + ) + return ShadowEvalJobResponse.model_validate(record, from_attributes=True).model_copy( + update={ # mutable-ok: pydantic update payload + "judged_count": totals[0].judged_count if totals else 0, + "error_count": totals[0].error_count if totals else 0, + "judge_spend": round(totals[0].judge_spend, 6) if totals else 0.0, + "last_error": latest_error.error if latest_error else None, + "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=ShadowEvalJobResponse, +) +async def stop_shadow_eval_job( + job_id: str, + user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], +) -> ShadowEvalJobResponse: + """Stop an active shadow eval job. Attempts 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}") + current: Final = ShadowEvalJobResponse.model_validate(record, from_attributes=True) + if current.status != "running": + raise HTTPException(status_code=400, detail=f"Job {job_id} is already {current.status}") + updated: Final = await prisma_client.db.litellm_shadowevaljob.update( + where={"id": job_id}, # mutable-ok: Prisma filter + data={"stopped_at": datetime.now(timezone.utc)}, # mutable-ok: Prisma payload + ) + return ShadowEvalJobResponse.model_validate(updated, from_attributes=True) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 64a91b880cd..968c645ffcd 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -2329,8 +2329,11 @@ def load_from_azure_key_vault(use_azure_key_vault: bool = False): def cost_tracking(): global prisma_client if prisma_client is not None: + from litellm.integrations.shadow_eval_logger import ShadowEvalLogger + litellm.logging_callback_manager.add_litellm_callback(_ProxyDBLogger()) litellm.logging_callback_manager.add_litellm_async_success_callback(_ProxyDBLogger()) + litellm.logging_callback_manager.add_litellm_callback(ShadowEvalLogger()) # Bounds authoritative DB re-reads when enforcing a budget against a diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index 854602f5380..79d778fb464 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -1450,6 +1450,44 @@ 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. The job row is immutable config plus +// stopped_at; every count, status, and spend figure is derived from the append-only +// attempt rows, so nothing can disagree across pods or stop races. +model LiteLLM_ShadowEvalJob { + id String @id @default(cuid()) + api_key_id String // hashed virtual key whose traffic is shadowed + router_name String + judge_model String + shadow_percentage Float + max_turns Int // sample budget: judge at most this many turns + created_at DateTime @default(now()) + created_by String? + ends_at DateTime + stopped_at DateTime? + + @@index([api_key_id]) + @@index([created_at]) +} + +// One row per sampled pipeline: a blind verdict (real | shadow | tie) or an error. +model LiteLLM_ShadowEvalAttempt { + id String @id @default(cuid()) + job_id String + request_id String // the judged real request + outcome String // real | shadow | tie | error + tier String? // router's tier for the prompt, when classified + real_model String? + shadow_model String? + confidence Float? + judge_cost Float @default(0) + error String? + 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 be466fcc575..9f634acfcdd 100644 --- a/litellm/router_strategy/complexity_router/complexity_router.py +++ b/litellm/router_strategy/complexity_router/complexity_router.py @@ -26,8 +26,9 @@ from typing import TYPE_CHECKING, Any, Final, Literal, NamedTuple, cast 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, @@ -206,40 +207,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} @@ -1069,7 +1036,7 @@ class ComplexityRouter(CustomLogger): ) 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() @@ -1562,8 +1529,12 @@ class ComplexityRouter(CustomLogger): # 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 269d9b50414..bf8a3d34098 100644 --- a/litellm/types/management_endpoints/auto_router_endpoints.py +++ b/litellm/types/management_endpoints/auto_router_endpoints.py @@ -3,9 +3,10 @@ Types for auto-router management endpoints """ from collections.abc import Mapping -from typing import Final +from datetime import datetime, timezone +from typing import Final, Literal, TypeAlias -from pydantic import BaseModel, Field, field_validator +from pydantic import AliasChoices, BaseModel, ConfigDict, Field, computed_field, field_validator from litellm.router_strategy.complexity_router.config import ComplexityRouterConfig from litellm.types.utils import StandardLoggingRoutingDecision @@ -141,3 +142,112 @@ class AutoRouterBenchmarksResponse(BaseModel): routers_in_scope: int totals: AutoRouterBenchmarkTotals groups: tuple[AutoRouterBenchmarkGroup, ...] + + +ShadowEvalStatus: TypeAlias = Literal["running", "completed", "stopped"] + +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 completing on its own", + ) + max_turns: int = Field( + default=200, + ge=1, + le=2000, + description=( + "Sample budget: the job judges at most this many turns, then completes. This is also the spend " + "bound; expected judge cost is roughly max_turns times one judge call" + ), + ) + + @field_validator("shadow_percentage") + @classmethod + def _round_percentage(cls, value: float) -> float: + return round(value, 2) + + +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 ShadowEvalJobResponse(BaseModel): + """A shadow-eval job. Validates directly from the prisma record (job_id reads the + row's id); status is derived from stopped_at and ends_at, never stored, so no writer + anywhere can produce an inconsistent one. Aggregate fields are populated by the + detail endpoint only and stay None on list responses.""" + + model_config = ConfigDict(from_attributes=True, populate_by_name=True) + + job_id: str = Field(validation_alias=AliasChoices("id", "job_id")) + api_key_id: str = Field(description="The hashed virtual key whose traffic this job evaluates, and only that key's") + router_name: str + judge_model: str + shadow_percentage: float + max_turns: int + created_at: datetime + ends_at: datetime + stopped_at: datetime | None = None + + judged_count: int | None = Field(default=None, description="Verdicts recorded; detail endpoint only") + error_count: int | None = Field(default=None, description="Sampled attempts that errored; detail endpoint only") + judge_spend: float | None = Field(default=None, description="Judge cost so far; detail endpoint only") + last_error: str | None = Field(default=None, description="Most recent attempt error; detail endpoint only") + results: ShadowEvalResult | None = Field(default=None, description="Stratified verdicts; detail endpoint only") + + @computed_field + @property + def status(self) -> ShadowEvalStatus: + """A job whose window has passed reads completed even if a later sweep stamped + stopped_at; stopped means sampling ended before the window did.""" + if datetime.now(timezone.utc) >= ( + self.ends_at if self.ends_at.tzinfo else self.ends_at.replace(tzinfo=timezone.utc) + ): + return "completed" + if self.stopped_at is not None: + return "stopped" + return "running" diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 354857f8d72..d9ef538d530 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -2782,11 +2782,13 @@ RoutingDecisionCause = Literal[ ] -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 854602f5380..79d778fb464 100644 --- a/schema.prisma +++ b/schema.prisma @@ -1450,6 +1450,44 @@ 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. The job row is immutable config plus +// stopped_at; every count, status, and spend figure is derived from the append-only +// attempt rows, so nothing can disagree across pods or stop races. +model LiteLLM_ShadowEvalJob { + id String @id @default(cuid()) + api_key_id String // hashed virtual key whose traffic is shadowed + router_name String + judge_model String + shadow_percentage Float + max_turns Int // sample budget: judge at most this many turns + created_at DateTime @default(now()) + created_by String? + ends_at DateTime + stopped_at DateTime? + + @@index([api_key_id]) + @@index([created_at]) +} + +// One row per sampled pipeline: a blind verdict (real | shadow | tie) or an error. +model LiteLLM_ShadowEvalAttempt { + id String @id @default(cuid()) + job_id String + request_id String // the judged real request + outcome String // real | shadow | tie | error + tier String? // router's tier for the prompt, when classified + real_model String? + shadow_model String? + confidence Float? + judge_cost Float @default(0) + error String? + 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 00000000000..e1c56db21af --- /dev/null +++ b/tests/test_litellm/integrations/test_shadow_eval_logger.py @@ -0,0 +1,466 @@ +"""Unit tests for the shadow-eval logger: sampling, unmasking, the hook's skip chain, +the detached pipeline's single attempt-row write, and the cache-first job lookup.""" + +import asyncio +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from litellm.caching.in_memory_cache import InMemoryCache +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, + _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", + max_turns=200, + ends_at=datetime.now(timezone.utc) + timedelta(days=1), + attempts=0, + ) + return ActiveShadowEvalJob(**{**defaults, **overrides}) + + +def _prisma(jobs=(), attempt_counts=()) -> MagicMock: + prisma = MagicMock() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=list(jobs)) + prisma.db.litellm_shadowevalattempt.group_by = AsyncMock( + return_value=[{"job_id": job_id, "_count": {"_all": count}} for job_id, count in attempt_counts] + ) + prisma.db.litellm_shadowevalattempt.create = AsyncMock() + return prisma + + +def _job_record(job: ActiveShadowEvalJob, api_key_id="key-hash") -> MagicMock: + record = MagicMock() + for field, value in dict( + id=job.id, + api_key_id=api_key_id, + router_name=job.router_name, + shadow_percentage=job.shadow_percentage, + judge_model=job.judge_model, + max_turns=job.max_turns, + ends_at=job.ends_at, + ).items(): + setattr(record, field, value) + return record + + +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, job=None) -> ShadowEvalLogger: + cache = InMemoryCache(max_size_in_memory=4, default_ttl=60) + logger = ShadowEvalLogger( + router_provider=lambda: router, + prisma_provider=lambda: prisma, + jobs_cache=cache, + ) + if job is not None: + cache.set_cache("shadow_eval:active_jobs", {"key-hash": job}) + return logger + + +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, target: int = 0): + for _ in range(100): + if logger._inflight_shadow_tasks == target: + 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 TestSuccessHookSkipChain: + async def test_happy_path_writes_exactly_one_attempt_row(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, job=_job()) + + await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None) + await _drain(logger) + + create = prisma.db.litellm_shadowevalattempt.create + create.assert_awaited_once() + row = create.call_args.kwargs["data"] + assert row["job_id"] == "job-1" + assert row["request_id"] == "req-1" + assert row["outcome"] in ("real", "shadow") + assert row["tier"] == "SIMPLE" + assert row["real_model"] == "claude-opus" + assert row["shadow_model"] == "cheap-model" + assert row["confidence"] == 0.9 + assert row["judge_cost"] == 0.005 + assert row["error"] is None + assert prisma.db.litellm_shadowevaljob.find_many.await_count == 0 + + @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)}), + ({}, {"attempts": 200}), + ({}, {"attempts": 199, "max_turns": 200, "_starts": 1}), + ], + ids=[ + "internal-origin", + "no-job-for-key", + "non-chat", + "missing-call-type", + "self-shadow", + "past-end", + "turn-budget-reached", + "budget-consumed-by-started-tasks", + ], + ) + async def test_skip_paths_store_nothing(self, kwargs_mutation, job_mutation): + starts = job_mutation.pop("_starts", 0) + prisma = _prisma() + logger = _logger(router=_router(), prisma=prisma, job=_job(**job_mutation)) + logger._job_starts = {"job-1": starts} + + await logger.async_log_success_event(_success_kwargs(**kwargs_mutation), RESPONSE, None, None) + await _drain(logger) + + prisma.db.litellm_shadowevalattempt.create.assert_not_called() + assert logger._job_starts.get("job-1", 0) == starts + + async def test_completed_pipelines_hold_turn_budget_within_a_cache_generation(self): + """A finished pipeline frees its concurrency slot but not its slice of the turn + budget; the budget only reopens when a cache refill absorbs the written rows.""" + prisma = _prisma() + logger = _logger(router=_router(), prisma=prisma, job=_job(attempts=199, max_turns=200)) + + await logger.async_log_success_event(_success_kwargs(request_id="req-1"), RESPONSE, None, None) + await _drain(logger) + await logger.async_log_success_event(_success_kwargs(request_id="req-2"), RESPONSE, None, None) + await _drain(logger) + + assert prisma.db.litellm_shadowevalattempt.create.await_count == 1 + + async def test_v1_messages_surface_forwards_identity_from_litellm_metadata(self): + """/v1/messages stores identity in litellm_params.litellm_metadata, so the hook + resolves the bucket through the shared helper; every surface forwards the same + identity to the shadow and judge calls.""" + prisma = _prisma() + router = _router() + logger = _logger(router=router, prisma=prisma, job=_job()) + + hook_kwargs = _success_kwargs() + hook_kwargs["litellm_params"] = { + "litellm_metadata": {"user_api_key_hash": "key-hash", "user_api_key_team_id": "team-1"} + } + await logger.async_log_success_event(hook_kwargs, RESPONSE, None, None) + await _drain(logger) + + shadow_call = router.acompletion.call_args_list[0].kwargs + assert shadow_call["metadata"]["user_api_key_hash"] == "key-hash" + assert shadow_call["metadata"]["user_api_key_team_id"] == "team-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; the skip uses the redactor's own + predicate, so every redaction source counts.""" + prisma = _prisma() + router = _router() + logger = _logger(router=router, prisma=prisma, job=_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.db.litellm_shadowevalattempt.create.assert_not_called() + + async def test_inflight_cap_sheds_instead_of_queueing(self): + prisma = _prisma() + logger = _logger(router=_router(), prisma=prisma, job=_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.db.litellm_shadowevalattempt.create.assert_not_called() + + +@pytest.mark.asyncio +class TestActiveJobsCache: + async def test_cache_miss_reads_db_once_then_serves_from_cache(self): + job = _job() + prisma = _prisma(jobs=[_job_record(job)], attempt_counts=[("job-1", 7)]) + logger = ShadowEvalLogger( + router_provider=lambda: None, + prisma_provider=lambda: prisma, + jobs_cache=InMemoryCache(max_size_in_memory=4, default_ttl=60), + ) + + first = await logger._active_jobs() + second = await logger._active_jobs() + + assert first["key-hash"].id == "job-1" + assert second["key-hash"].attempts == 7 + assert prisma.db.litellm_shadowevaljob.find_many.await_count == 1 + where = prisma.db.litellm_shadowevaljob.find_many.call_args.kwargs["where"] + assert where["stopped_at"] is None + assert "gt" in where["ends_at"] + count_where = prisma.db.litellm_shadowevalattempt.group_by.call_args.kwargs["where"] + assert count_where == {"job_id": {"in": ["job-1"]}} + + async def test_no_active_jobs_is_cached_too(self): + prisma = _prisma(jobs=[]) + logger = ShadowEvalLogger( + router_provider=lambda: None, + prisma_provider=lambda: prisma, + jobs_cache=InMemoryCache(max_size_in_memory=4, default_ttl=60), + ) + + assert await logger._active_jobs() == {} + assert await logger._active_jobs() == {} + assert prisma.db.litellm_shadowevaljob.find_many.await_count == 1 + prisma.db.litellm_shadowevalattempt.group_by.assert_not_called() + + async def test_db_fault_returns_empty_without_caching_the_fault(self): + prisma = _prisma() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=RuntimeError("db blip")) + logger = ShadowEvalLogger( + router_provider=lambda: None, + prisma_provider=lambda: prisma, + jobs_cache=InMemoryCache(max_size_in_memory=4, default_ttl=60), + ) + + assert await logger._active_jobs() == {} + assert await logger._active_jobs() == {} + assert prisma.db.litellm_shadowevaljob.find_many.await_count == 2 + + async def test_cache_refill_resets_the_starts_counter(self): + job = _job() + prisma = _prisma(jobs=[_job_record(job)], attempt_counts=[("job-1", 7)]) + logger = ShadowEvalLogger( + router_provider=lambda: None, + prisma_provider=lambda: prisma, + jobs_cache=InMemoryCache(max_size_in_memory=4, default_ttl=60), + ) + logger._job_starts = {"job-1": 5} + + await logger._active_jobs() + + assert logger._job_starts == {} + + +@pytest.mark.asyncio +class TestShadowPipeline: + async def test_no_prisma_means_no_provider_spend(self): + 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() + 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={"user_api_key_auth": UserAPIKeyAuth(api_key="sk-abc", max_budget=10.0)}, + ) + + router.acompletion.assert_not_called() + prisma.db.litellm_shadowevalattempt.create.assert_not_called() + + @pytest.mark.parametrize( + "router_factory,expected_error,expected_cost", + [ + (lambda: _failing_router(), "provider exploded", 0.0), + (lambda: _router(judge_json="I prefer response A, definitely"), "unparseable judge verdict", 0.007), + ], + ids=["shadow-call-fails", "judge-verdict-unparseable"], + ) + async def test_failures_become_error_rows_and_keep_billed_judge_cost( + self, router_factory, expected_error, expected_cost, monkeypatch: pytest.MonkeyPatch + ): + import litellm as litellm_module + + monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.007) + prisma = _prisma() + logger = _logger(router=router_factory(), 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={}, + ) + + row = prisma.db.litellm_shadowevalattempt.create.call_args.kwargs["data"] + assert row["outcome"] == "error" + assert expected_error in row["error"] + assert row["confidence"] is None + assert row["judge_cost"] == expected_cost + + 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 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 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 + 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 + + +def _failing_router(): + router = MagicMock() + router.model_group_alias = {} + router.get_model_list = MagicMock(return_value=None) + router.acompletion = AsyncMock(side_effect=RuntimeError("provider exploded")) + return router 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 00000000000..73923dc75a5 --- /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 00000000000..5c092caa7c3 --- /dev/null +++ b/tests/test_litellm/litellm_core_utils/test_llm_judge.py @@ -0,0 +1,92 @@ +"""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 + assert kwargs["drop_params"] is True + + +@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 + assert sdk.call_args.kwargs["drop_params"] is True 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 0df11f224a2..95dce1ccb0a 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 2c426e0f071..ca7d5fcd273 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/hooks/test_parallel_request_limiter_v3.py b/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py index 076151fcd3b..59434290c48 100644 --- a/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py +++ b/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py @@ -17,6 +17,7 @@ from fastapi import HTTPException import litellm from litellm import Router from litellm.caching.caching import DualCache +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.hooks.parallel_request_limiter_v3 import ( PARALLEL_REQUEST_SLOT_TTL_SECONDS, @@ -5554,3 +5555,41 @@ async def test_configured_estimate_blocks_the_overrun_the_static_floor_admits(mo assert await admitted({}) == 7 assert await admitted({"default_estimated_output_tokens": 3000}) == 2 + + +def test_internal_call_origin_success_ops_are_skipped(): + """Internal sub-calls (auto-router classifier, shadow eval shadow/judge) bill spend + to the caller's key but must not consume its TPM counters: the same kwargs charge + ops without the origin stamp and none with it.""" + handler = _PROXY_MaxParallelRequestsHandler( + internal_usage_cache=InternalUsageCache(DualCache()) + ) + response = ModelResponse( + id="internal-origin-tpm", + object="chat.completion", + created=int(datetime.now().timestamp()), + model="gpt-4o-mini", + usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150), + choices=[], + ) + + def _kwargs(metadata: Dict[str, Any]) -> Dict[str, Any]: + return { + "standard_logging_object": { + "metadata": {"user_api_key_hash": hash_token("sk-internal-origin")} + }, + "litellm_params": {"metadata": metadata}, + "model": "gpt-4o-mini", + } + + charged = handler._build_success_event_pipeline_operations( + kwargs=_kwargs({}), response_obj=response, rate_limit_type="output" + ) + skipped = handler._build_success_event_pipeline_operations( + kwargs=_kwargs({INTERNAL_CALL_ORIGIN_METADATA_KEY: "shadow_eval_judge"}), + response_obj=response, + rate_limit_type="output", + ) + + assert charged + assert skipped == [] 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 3a995e27697..16e82bc3bda 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,276 @@ class TestAutoRouterBenchmarks: end_date="2026-08-01", ) assert response.groups[0].tier_turns == expected + + +# --------------------------------------------------------------------------- +# Shadow eval endpoints +# --------------------------------------------------------------------------- + +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, MagicMock + +from fastapi import HTTPException + +from litellm.proxy.management_endpoints.auto_router_endpoints import ( + get_shadow_eval_job, + list_shadow_eval_jobs, + start_shadow_eval, + stop_shadow_eval_job, +) +from litellm.types.management_endpoints.auto_router_endpoints import ShadowEvalJobResponse, 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: + """Spec'd like a real prisma row: only the table's columns exist as attributes, so + from_attributes validation falls back to model defaults for everything else.""" + defaults = { + "id": "job-1", + "api_key_id": "key-hash", + "router_name": "my-router", + "judge_model": "anthropic/claude-sonnet-5", + "shadow_percentage": 10.0, + "max_turns": 200, + "created_at": datetime(2026, 8, 11, tzinfo=timezone.utc), + "ends_at": datetime.now(timezone.utc) + timedelta(days=7), + "stopped_at": None, + } + fields = {**defaults, **overrides} + record = MagicMock(spec=list(fields)) + for key, value in fields.items(): + setattr(record, key, value) + return record + + +def _shadow_prisma(active_job=None, agg_rows=None) -> MagicMock: + prisma = MagicMock() + prisma.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=MagicMock()) + prisma.db.execute_raw = AsyncMock(return_value=0) + prisma.db.litellm_shadowevaljob.find_first = AsyncMock(return_value=active_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()) + prisma.db.litellm_shadowevaljob.update = AsyncMock( + return_value=_job_record(stopped_at=datetime.now(timezone.utc)) + ) + prisma.db.litellm_shadowevalattempt.find_first = AsyncMock(return_value=None) + + async def query_raw(sql: str, *params: object): + if "FILTER (WHERE outcome != 'error')::int AS judged_count" in sql: + return [{"judged_count": 10, "error_count": 2, "judge_spend": 0.031}] + 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, + "max_turns": 200, + } + payload.update(overrides) + return StartShadowEvalRequest.model_validate(payload) + + +@pytest.mark.asyncio +async def test_start_shadow_eval_creates_job_and_frees_expired_or_exhausted_ones(monkeypatch: pytest.MonkeyPatch): + """Expiry and turn-budget exhaustion both end sampling on their own; either must + release the one-active-per-key index so a new eval can start.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma() + 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 == "running" + assert response.max_turns == 200 + assert response.judged_count is None + sweep_sql, sweep_key = prisma.db.execute_raw.call_args.args + assert "stopped_at IS NULL" in sweep_sql + assert "ends_at <= NOW()" in sweep_sql + assert ">= j.max_turns" in sweep_sql + assert sweep_key == "key-hash" + 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 "status" not in create_data + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "caller,request_overrides,active,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, {}, "active", 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, active, expected_status +): + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(active_job=_job_record() if active else 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(**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 + + +@pytest.mark.asyncio +async def test_get_shadow_eval_job_derives_counts_spend_and_stratified_results(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()) + prisma.db.litellm_shadowevalattempt.find_first = AsyncMock( + return_value=MagicMock(error="judge call failed: boom") + ) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + response = await get_shadow_eval_job("job-1", VIEWER) + + assert response.job_id == "job-1" + assert response.status == "running" + assert response.judged_count == 10 + assert response.error_count == 2 + assert response.judge_spend == 0.031 + assert response.last_error == "judge call failed: boom" + 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_returns_derived_status_without_aggregates(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(), + _job_record(id="job-2", ends_at=datetime.now(timezone.utc) - timedelta(days=1)), + _job_record(id="job-3", stopped_at=datetime.now(timezone.utc)), + ] + ) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) + + assert [job.status for job in jobs] == ["running", "completed", "stopped"] + swept = ShadowEvalJobResponse.model_validate( + _job_record( + id="job-4", + ends_at=datetime.now(timezone.utc) - timedelta(days=1), + stopped_at=datetime.now(timezone.utc), + ), + from_attributes=True, + ) + assert swept.status == "completed" + assert all(job.judged_count is None and job.results is None for job in jobs) + assert prisma.db.query_raw.await_count == 0 + + +@pytest.mark.asyncio +async def test_stop_shadow_eval_sets_stopped_at_and_rejects_non_running(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()) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + stopped = await stop_shadow_eval_job("job-1", ADMIN) + assert stopped.status == "stopped" + update = prisma.db.litellm_shadowevaljob.update.call_args.kwargs + assert set(update["data"]) == {"stopped_at"} + + prisma.db.litellm_shadowevaljob.find_unique = AsyncMock( + return_value=_job_record(ends_at=datetime.now(timezone.utc) - timedelta(days=1)) + ) + 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 ad41592db1e..a06e7142122 100644 --- a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py +++ b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py @@ -524,8 +524,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) @@ -535,16 +536,19 @@ def test_cost_tracking_adds_two_callbacks_when_prisma_set(monkeypatch): before_callbacks = len(litellm.callbacks) before_async = len(litellm._async_success_callback) + cost_tracking() cost_tracking() 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, } diff --git a/tests/test_litellm/router_strategy/test_complexity_router.py b/tests/test_litellm/router_strategy/test_complexity_router.py index e3b0872ee65..4f43567de36 100644 --- a/tests/test_litellm/router_strategy/test_complexity_router.py +++ b/tests/test_litellm/router_strategy/test_complexity_router.py @@ -3449,98 +3449,6 @@ class TestKeywordOverrideEdgeCases: 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/eslint-suppressions.json b/ui/litellm-dashboard/eslint-suppressions.json index c7e2b84032a..2f4def9ffa2 100644 --- a/ui/litellm-dashboard/eslint-suppressions.json +++ b/ui/litellm-dashboard/eslint-suppressions.json @@ -3799,11 +3799,6 @@ "count": 1 } }, - "src/components/view_logs/LogDetailsDrawer/JsonViewer.tsx": { - "no-restricted-imports": { - "count": 1 - } - }, "src/components/view_logs/LogDetailsDrawer/LogDetailContent.tsx": { "no-nested-ternary": { "count": 3 @@ -3843,21 +3838,6 @@ "count": 1 } }, - "src/components/view_logs/LogDetailsDrawer/SimpleMessageBlock.tsx": { - "no-restricted-imports": { - "count": 1 - } - }, - "src/components/view_logs/LogDetailsDrawer/SimpleToolCallBlock.tsx": { - "no-restricted-imports": { - "count": 1 - } - }, - "src/components/view_logs/LogDetailsDrawer/TokenFlow.tsx": { - "no-restricted-imports": { - "count": 1 - } - }, "src/components/view_logs/LogDetailsDrawer/TruncatedValue.tsx": { "no-restricted-imports": { "count": 1 diff --git a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/JsonViewer.test.tsx b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/JsonViewer.test.tsx new file mode 100644 index 00000000000..055e400167b --- /dev/null +++ b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/JsonViewer.test.tsx @@ -0,0 +1,28 @@ +import { render, screen } from "@testing-library/react"; +import { describe, expect, it } from "vitest"; +import { JsonViewer } from "./JsonViewer"; + +describe("JsonViewer", () => { + it("should render a placeholder and no tree when the log entry carries no payload", () => { + render(); + + expect(screen.getByText("No data")).toBeInTheDocument(); + expect(screen.queryByRole("tree")).not.toBeInTheDocument(); + }); + + it("should render the payload as a tree exposing its keys", () => { + render(); + + expect(screen.getByRole("tree")).toBeInTheDocument(); + expect(screen.getByText(/model/)).toBeInTheDocument(); + expect(screen.getByText(/stream/)).toBeInTheDocument(); + expect(screen.queryByText("No data")).not.toBeInTheDocument(); + }); + + it("should treat an empty payload as data rather than showing the placeholder", () => { + render(); + + expect(screen.getByRole("tree")).toBeInTheDocument(); + expect(screen.queryByText("No data")).not.toBeInTheDocument(); + }); +}); diff --git a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/JsonViewer.tsx b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/JsonViewer.tsx index 0a873a7e255..af99eee0d16 100644 --- a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/JsonViewer.tsx +++ b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/JsonViewer.tsx @@ -1,10 +1,7 @@ -import { Typography } from "antd"; import { JsonView, defaultStyles } from "react-json-view-lite"; import "react-json-view-lite/dist/index.css"; import { JSON_MAX_HEIGHT, COLOR_BG_LIGHT, SPACING_LARGE } from "./constants"; -const { Text } = Typography; - interface JsonViewerProps { data: any; mode: "formatted"; @@ -15,7 +12,7 @@ interface JsonViewerProps { * Uses an interactive tree component for easy navigation. */ export function JsonViewer({ data }: JsonViewerProps) { - if (!data) return No data; + if (!data) return No data; return (
- - {label} - +
+ {label} {displayContent && (
{displayContent}
diff --git a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/SimpleToolCallBlock.tsx b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/SimpleToolCallBlock.tsx index 42738794e2f..6d69d07682c 100644 --- a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/SimpleToolCallBlock.tsx +++ b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/SimpleToolCallBlock.tsx @@ -3,11 +3,9 @@ * Used in compact/tree views */ -import { Typography } from "antd"; +import { cn } from "@/lib/cva.config"; import { ToolCall } from "./prettyMessagesTypes"; -const { Text } = Typography; - interface SimpleToolCallBlockProps { tool: ToolCall; compact?: boolean; @@ -16,46 +14,24 @@ interface SimpleToolCallBlockProps { export function SimpleToolCallBlock({ tool, compact = false }: SimpleToolCallBlockProps) { return (
{/* Function badge */} -
+
function
- - {tool.name} - + {tool.name} {Object.keys(tool.arguments).length > 0 && (
{Object.entries(tool.arguments).map(([key, value]) => ( -
- - {key}:{" "} - - {JSON.stringify(value)} +
+ {key}: + {JSON.stringify(value)}
))}
diff --git a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/TokenFlow.test.tsx b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/TokenFlow.test.tsx new file mode 100644 index 00000000000..0e76b92da36 --- /dev/null +++ b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/TokenFlow.test.tsx @@ -0,0 +1,29 @@ +import { render, screen } from "@testing-library/react"; +import { describe, expect, it } from "vitest"; +import { TokenFlow } from "./TokenFlow"; + +const localised = (count: number) => count.toLocaleString(); + +describe("TokenFlow", () => { + it("should render the total followed by its prompt and completion breakdown", () => { + render(); + + expect(screen.getByText("12 (9 prompt tokens + 3 completion tokens)")).toBeInTheDocument(); + }); + + it("should group large counts the way the reader's locale does", () => { + render(); + + expect( + screen.getByText( + `${localised(1323579)} (${localised(1234567)} prompt tokens + ${localised(89012)} completion tokens)`, + ), + ).toBeInTheDocument(); + }); + + it("should fall back to zero for counts the log entry does not carry", () => { + render(); + + expect(screen.getByText("12 (0 prompt tokens + 0 completion tokens)")).toBeInTheDocument(); + }); +}); diff --git a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/TokenFlow.tsx b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/TokenFlow.tsx index 5eec3c0a5cb..2183ed53f35 100644 --- a/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/TokenFlow.tsx +++ b/ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/TokenFlow.tsx @@ -1,7 +1,3 @@ -import { Typography } from "antd"; - -const { Text } = Typography; - interface TokenFlowProps { prompt?: number; completion?: number; @@ -14,9 +10,9 @@ interface TokenFlowProps { */ export function TokenFlow({ prompt = 0, completion = 0, total = 0 }: TokenFlowProps) { return ( - + {total.toLocaleString()} ({prompt.toLocaleString()} prompt tokens + {completion.toLocaleString()} completion tokens) - + ); } diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 327304dc45f..a2a4c01c25f 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -807,6 +807,93 @@ 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. Counts and results ride the detail endpoint only. + */ + 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 and by the incumbent model. + * + * Shadow responses are never served to users. The job samples until it has judged + * max_turns turns, reaches the end of its window, or is stopped; sampling changes + * propagate to pods within about 10 seconds. Shadow and judge calls bill to the + * shadowed key but are excluded from request counts and auto-router adoption metrics. + */ + 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 One job with derived counts, judge spend, latest error, and stratified results. + */ + 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. Attempts 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; @@ -32557,6 +32644,110 @@ export interface components { /** Timeout */ timeout?: number | null; }; + /** + * ShadowEvalJobResponse + * @description A shadow-eval job. Validates directly from the prisma record (job_id reads the + * row's id); status is derived from stopped_at and ends_at, never stored, so no writer + * anywhere can produce an inconsistent one. Aggregate fields are populated by the + * detail endpoint only and stay None on list responses. + */ + ShadowEvalJobResponse: { + /** + * Api Key Id + * @description The hashed virtual key whose traffic this job evaluates, and only that key's + */ + api_key_id: string; + /** + * Created At + * Format: date-time + */ + created_at: string; + /** + * Ends At + * Format: date-time + */ + ends_at: string; + /** + * Error Count + * @description Sampled attempts that errored; detail endpoint only + */ + error_count?: number | null; + /** Job Id */ + job_id: string; + /** Judge Model */ + judge_model: string; + /** + * Judge Spend + * @description Judge cost so far; detail endpoint only + */ + judge_spend?: number | null; + /** + * Judged Count + * @description Verdicts recorded; detail endpoint only + */ + judged_count?: number | null; + /** + * Last Error + * @description Most recent attempt error; detail endpoint only + */ + last_error?: string | null; + /** Max Turns */ + max_turns: number; + /** @description Stratified verdicts; detail endpoint only */ + results?: components["schemas"]["ShadowEvalResult"] | null; + /** Router Name */ + router_name: string; + /** Shadow Percentage */ + shadow_percentage: number; + /** + * Status + * @description A job whose window has passed reads completed even if a later sweep stamped + * stopped_at; stopped means sampling ended before the window did. + * @enum {string} + */ + readonly status: "running" | "completed" | "stopped"; + /** Stopped At */ + stopped_at?: string | 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 @@ -32730,6 +32921,45 @@ 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 completing on its own + * @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; + /** + * Max Turns + * @description Sample budget: the job judges at most this many turns, then completes. This is also the spend bound; expected judge cost is roughly max_turns times one judge call + * @default 200 + */ + max_turns: number; + /** + * 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; + }; /** * SuccessfulKeyUpdate * @description Successfully updated key with its updated information @@ -37006,6 +37236,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"]["ShadowEvalJobResponse"][]; + }; + }; + /** @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"]["ShadowEvalJobResponse"]; + }; + }; + /** @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"]["ShadowEvalJobResponse"]; + }; + }; + /** @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"]["ShadowEvalJobResponse"]; + }; + }; + /** @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;