feat: pre-adoption shadow eval for the auto-router (blind pairwise judge, per-tier win rates)

This commit is contained in:
Tin Chi Lo 2026-08-11 13:04:57 -07:00 committed by Tin
parent e37ae033cc
commit 52ad2538df
24 changed files with 3008 additions and 216 deletions

View file

@ -0,0 +1,55 @@
-- CreateTable
CREATE TABLE "LiteLLM_ShadowEvalJob" (
"id" TEXT NOT NULL,
"api_key_id" TEXT NOT NULL,
"router_name" TEXT NOT NULL,
"shadow_percentage" DOUBLE PRECISION NOT NULL,
"judge_model" TEXT NOT NULL,
"status" TEXT NOT NULL DEFAULT 'pending',
"request_count" INTEGER NOT NULL DEFAULT 0,
"completed_count" INTEGER NOT NULL DEFAULT 0,
"failed_count" INTEGER NOT NULL DEFAULT 0,
"last_error" TEXT,
"cost_estimate" DOUBLE PRECISION,
"cost_actual" DOUBLE PRECISION NOT NULL DEFAULT 0,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"created_by" TEXT,
"ends_at" TIMESTAMP(3),
"completed_at" TIMESTAMP(3),
CONSTRAINT "LiteLLM_ShadowEvalJob_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "LiteLLM_ShadowEvalVerdict" (
"id" TEXT NOT NULL,
"job_id" TEXT NOT NULL,
"request_id" TEXT NOT NULL,
"tier_classification" TEXT,
"real_model" TEXT NOT NULL,
"shadow_model" TEXT NOT NULL,
"judge_preference" TEXT NOT NULL,
"judge_confidence" DOUBLE PRECISION,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "LiteLLM_ShadowEvalVerdict_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE INDEX "LiteLLM_ShadowEvalJob_api_key_id_status_idx" ON "LiteLLM_ShadowEvalJob"("api_key_id", "status");
-- CreateIndex
CREATE INDEX "LiteLLM_ShadowEvalJob_status_idx" ON "LiteLLM_ShadowEvalJob"("status");
-- CreateIndex
CREATE INDEX "LiteLLM_ShadowEvalJob_created_at_idx" ON "LiteLLM_ShadowEvalJob"("created_at");
-- CreateIndex
CREATE INDEX "LiteLLM_ShadowEvalVerdict_job_id_idx" ON "LiteLLM_ShadowEvalVerdict"("job_id");
-- One active job per key, enforced by the database rather than a read-then-create in
-- the start endpoint, which races against a concurrent start on another pod. Partial
-- indexes are not expressible in schema.prisma, so this lives here only.
CREATE UNIQUE INDEX "LiteLLM_ShadowEvalJob_one_active_per_key"
ON "LiteLLM_ShadowEvalJob"("api_key_id") WHERE status IN ('pending', 'running');

View file

@ -1450,6 +1450,50 @@ model LiteLLM_AutoRouterSession {
@@index([last_turn_at], map: "idx_autorouter_session_last_turn")
}
// Shadow Eval: pre-adoption evaluation of an auto-router against a key's live traffic.
// A sampled slice of requests is duplicated through the router in a detached task and
// an LLM judge compares real vs shadow responses blind; verdicts stratify by tier.
model LiteLLM_ShadowEvalJob {
id String @id @default(cuid())
api_key_id String // the hashed virtual key whose traffic is shadowed
router_name String // the auto-router config to shadow through
shadow_percentage Float
judge_model String
status String @default("pending") // pending | running | completed
request_count Int @default(0) // requests seen on the key while active
completed_count Int @default(0) // verdicts written
failed_count Int @default(0) // shadow or judge calls that errored
last_error String?
cost_estimate Float? // upfront judge-spend estimate shown at start
cost_actual Float @default(0) // running judge-call spend
created_at DateTime @default(now())
created_by String?
ends_at DateTime?
completed_at DateTime?
@@index([api_key_id, status])
@@index([status])
@@index([created_at])
}
model LiteLLM_ShadowEvalVerdict {
id String @id @default(cuid())
job_id String
request_id String // the judged real request
tier_classification String? // the router's tier for the prompt, when classified
real_model String // model that actually served the request
shadow_model String // model the router picked
judge_preference String // real | shadow | tie
judge_confidence Float?
created_at DateTime @default(now())
@@index([job_id])
}
// ---------------------------------------------------------------------------
// Workflow Run Tracking
//

View file

@ -0,0 +1,669 @@
"""Shadow Eval Logger: pre-adoption evaluation of an auto-router against live traffic.
For each successful request on a key with an active shadow-eval job, a sampled slice of
requests is duplicated through the auto-router (the user never sees the shadow response),
an LLM judge compares the two responses blind with A/B labels randomized, and the verdict
is stored stratified by the router's own tier classification.
The whole pipeline runs in a detached background task; the success hook itself is one
dict lookup against a job snapshot. Shadow and judge calls carry the shadowed key's
identity metadata (their provider spend bills to that key) and an
``internal_call_origin`` stamp, which the hook also skips on, so the logger can never
recurse on its own traffic. A shadow/judge pair is skipped outright if the shadowed key
or its team is already at or over budget.
Job lifecycle (counter flushes, snapshot refresh, stopping a job at its ``ends_at`` or
judge-spend cap) runs on a periodic loop owned by logger registration, never on the
request path: an idle key's job still ends on schedule, and the final counter batch
lands without needing another request to arrive.
"""
import asyncio
import hashlib
import random
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
from pydantic import BaseModel
from litellm._logging import verbose_logger
from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.internal_call_metadata import sanitized_forwardable_call_metadata
from litellm.litellm_core_utils.llm_judge import (
default_router_provider,
extract_text_from_content,
judge_acompletion,
parse_json_verdict,
)
from litellm.litellm_core_utils.redact_messages import should_redact_message_logging
from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient
from litellm.router import Router
from litellm.types.utils import StandardLoggingPayload
# Cadence of the lifecycle loop: snapshot refresh, counter flush, and job finalization
# all run on this tick, so a job starting or stopping takes up to one tick to be noticed.
_LIFECYCLE_TICK_SECONDS: Final = 10.0
# Concurrent shadow+judge pipelines per pod: a traffic spike turns into skipped samples
# rather than an unbounded task pileup.
_MAX_CONCURRENT_SHADOW_TASKS: Final = 16
# Total character budget for the judge's user prompt, however long the conversation and
# the two responses are, so the prompt can never overflow a judge model's context window.
_MAX_JUDGE_RESPONSE_CHARS: Final = 8_000
_MAX_JUDGE_PROMPT_CHARS: Final = 24_000
# The judge answers with a small JSON object; a tighter budget truncates the JSON
# mid-object and the verdict is lost to failed_count.
JUDGE_MAX_OUTPUT_TOKENS: Final = 500
_MAX_LAST_ERROR_CHARS: Final = 500
# A job stops sampling once its judge spend reaches this multiple of the estimate shown
# at start. The headroom absorbs an estimate that undershot the real traffic mix; the
# floor keeps a cent-sized estimate from stopping a job on its first verdict. cost_actual
# is read from the job snapshot, so overshoot is bounded by one lifecycle tick.
_SPEND_CAP_MULTIPLIER: Final = 1.5
_SPEND_CAP_FLOOR_USD: Final = 1.0
_EMPTY_METADATA: Final[Mapping[str, object]] = MappingProxyType({})
PAIRWISE_JUDGE_SYSTEM_PROMPT: Final = """You are an impartial quality judge comparing two responses to the same conversation.
The responses are labeled A and B in random order. You do not know which system produced which.
Criteria: correctness, completeness, clarity, conciseness.
Return ONLY valid JSON in this exact format, no other text:
{
"preference": "A" | "B" | "tie",
"confidence": <0.0 to 1.0>,
"reasoning": "<one sentence>"
}"""
class PairwiseVerdict(BaseModel):
"""The judge's blind A/B verdict, validated at the parse boundary."""
preference: str = "tie"
confidence: float = 0.0
def _sample_hits(request_id: str, job_id: str, percentage: float) -> bool:
"""Deterministically decide whether a request falls in the shadowed slice: hash-based
rather than random so retries sample the same way and pods agree without coordination."""
digest: Final = hashlib.sha256(f"{job_id}:{request_id}".encode()).digest()
bucket: Final = int.from_bytes(digest[:8], "big") / float(2**64)
return bucket * 100.0 < percentage
def _judge_call_cost(response: object) -> float:
"""Price a judge call, treating an unmapped judge model as free rather than fatal."""
import litellm
try:
return litellm.completion_cost(completion_response=response) or 0.0
except Exception: # noqa: BLE001 # unmapped judge model: verdict still counts, cost stays 0
return 0.0
def _unmask_preference(raw_preference: str, real_is_a: bool) -> str:
"""Map the judge's blind A/B/tie verdict back to real/shadow/tie."""
normalized: Final = raw_preference.strip().lower()
if normalized == "a":
return "real" if real_is_a else "shadow"
if normalized == "b":
return "shadow" if real_is_a else "real"
return "tie"
def _judge_user_prompt(conversation: str, response_a: str, response_b: str) -> str:
"""The judge prompt under one total character budget: each response is capped, and
the conversation tail gets whatever budget the responses left over."""
a: Final = response_a[:_MAX_JUDGE_RESPONSE_CHARS]
b: Final = response_b[:_MAX_JUDGE_RESPONSE_CHARS]
conversation_budget: Final = _MAX_JUDGE_PROMPT_CHARS - len(a) - len(b)
return (
f"Conversation:\n{conversation[-conversation_budget:]}\n\n"
f"Response A:\n{a}\n\n"
f"Response B:\n{b}\n\n"
"Which response is better?"
)
@dataclass(frozen=True, slots=True)
class _CallFailure:
"""A shadow or judge call that produced no usable response, with why."""
error: str
@dataclass(frozen=True, slots=True)
class _ShadowResponse:
"""A successful shadow call, with what the verdict row records."""
text: str
model: str
tier: str | None
@dataclass(frozen=True, slots=True)
class _JudgeVerdict:
"""A parsed judge verdict, unmasked back to real/shadow/tie."""
preference: str
confidence: float
cost: float
@dataclass(frozen=True, slots=True)
class ActiveShadowEvalJob:
"""The subset of a shadow-eval job row the request path actually needs."""
id: str
router_name: str
shadow_percentage: float
judge_model: str
status: str
cost_estimate: float | None = None
cost_actual: float = 0.0
ends_at: datetime | None = None
def _as_utc(value: object) -> datetime | None:
if not isinstance(value, datetime):
return None
return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value
def _job_is_past_its_end(job: ActiveShadowEvalJob) -> bool:
"""An eval prices a fixed window at start time, so sampling past ends_at would bill
traffic the estimate never covered."""
return job.ends_at is not None and datetime.now(timezone.utc) >= job.ends_at
def _job_is_over_spend_cap(job: ActiveShadowEvalJob) -> bool:
"""Budgets bound what the key may spend; this bounds what a single eval may spend
even under a generous budget, so a bad estimate or a traffic spike cannot quietly
turn a small eval into a much larger bill."""
if job.cost_estimate is None:
return False
return job.cost_actual >= max(job.cost_estimate * _SPEND_CAP_MULTIPLIER, _SPEND_CAP_FLOOR_USD)
async def _key_or_team_is_over_budget(metadata: Mapping[str, object]) -> bool:
"""Whether the shadowed key or its team is over budget, decided by the same owners
the request path uses, so counter keys and thresholds can never drift from auth's.
Advisory and fail-open: real traffic on an over-budget key is already rejected at
auth (so nothing reaches the success hook), and this gate only closes the race
where the key crosses its budget while a request is in flight.
"""
try:
from litellm.exceptions import BudgetExceededError
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.auth_checks import (
_team_max_budget_check,
_virtual_key_max_budget_check,
get_team_object,
)
from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache
except ImportError:
return False
auth: Final = metadata.get("user_api_key_auth")
if not isinstance(auth, UserAPIKeyAuth):
return False
try:
await _virtual_key_max_budget_check(valid_token=auth, proxy_logging_obj=proxy_logging_obj)
if auth.team_id:
team: Final = await get_team_object(
team_id=auth.team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
check_cache_only=True,
)
await _team_max_budget_check(team_object=team, valid_token=auth, proxy_logging_obj=proxy_logging_obj)
except BudgetExceededError:
return True
except Exception as e: # noqa: BLE001 # advisory gate: a failed read must not block sampling
verbose_logger.debug("shadow_eval: budget read failed: %s", e)
return False
def _request_was_routed_by(request_metadata: Mapping[str, object], router_name: str) -> bool:
"""Duplicating a request the shadowed router already served compares the router to
itself: guaranteed ties, judge spend for zero information."""
decision: Final = request_metadata.get("routing_decision")
if not isinstance(decision, Mapping):
return False
return decision.get("router_model_name") == router_name
class ShadowEvalLogger(CustomLogger):
"""Fires blind pairwise shadow evaluations for keys with an active shadow-eval job."""
def __init__(
self,
router_provider: Callable[[], "Router | None"] | None = None,
prisma_provider: Callable[[], "PrismaClient | None"] | None = None,
) -> None:
"""Providers are callables so the proxy's lazily-initialized globals are resolved
at call time, not at logger construction."""
self._router_provider = router_provider or default_router_provider
self._prisma_provider = prisma_provider or _default_prisma_provider
# Snapshot of every active job, keyed by shadowed api_key_id, refreshed as a
# whole by the lifecycle loop: one find_many per pod per tick keeps DB load flat
# no matter how many distinct keys the proxy serves.
self._jobs_by_key: dict[str, ActiveShadowEvalJob] = {} # mutable-ok: loop-refreshed snapshot
self._inflight_shadow_tasks: int = 0
self._pending_seen: dict[str, int] = {} # mutable-ok: flush buffer
self._lifecycle_task: asyncio.Task[None] | None = None
def start_lifecycle_loop(self) -> None:
"""Idempotently start the loop that owns job lifecycle off the request path.
Without it, jobs never finalize and counters never flush, so a caller outside a
running event loop gets a warning rather than a silent no-op."""
if self._lifecycle_task is not None and not self._lifecycle_task.done():
return
try:
self._lifecycle_task = asyncio.create_task(self._lifecycle_loop())
except RuntimeError:
verbose_logger.warning(
"shadow_eval: no running event loop; lifecycle loop not started, jobs will not finalize on this process"
)
async def _lifecycle_loop(self) -> None:
while True:
try:
await self._lifecycle_tick()
except Exception as e: # noqa: BLE001 # the loop must survive any single tick failing
verbose_logger.debug("shadow_eval: lifecycle tick failed: %s", e)
await asyncio.sleep(_LIFECYCLE_TICK_SECONDS)
async def _lifecycle_tick(self) -> None:
"""Flush counters while jobs are still active, refresh, then finalize, so an
expiring job's last counter batch lands while its row still passes the
active-status guard."""
await self._flush_seen_counts()
await self._refresh_active_jobs()
for job in tuple(self._jobs_by_key.values()):
if _job_is_past_its_end(job):
await self._finalize_job(job, "reached its scheduled end")
elif _job_is_over_spend_cap(job):
await self._finalize_job(
job,
f"spend ${job.cost_actual:.4f} reached the cap for its ${job.cost_estimate or 0.0:.4f} estimate",
)
#### hook ####
async def async_log_success_event(
self,
kwargs: Mapping[str, object],
response_obj: object,
start_time: object,
end_time: object,
) -> None:
try:
payload: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object") # pyright: ignore[reportAssignmentType] # untyped callback kwargs
if payload is None:
return
metadata: Final = payload.get("metadata") or _EMPTY_METADATA
litellm_params: Final = kwargs.get("litellm_params")
raw_request_metadata: Final = (
litellm_params.get("metadata") if isinstance(litellm_params, Mapping) else None
)
request_metadata: Final = (
raw_request_metadata if isinstance(raw_request_metadata, Mapping) else _EMPTY_METADATA
)
if request_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY):
return # internal sub-call (our own shadow/judge, a classifier), not user traffic
api_key_hash: Final = metadata.get("user_api_key_hash")
if not api_key_hash:
return
job: Final = self._jobs_by_key.get(str(api_key_hash))
if job is None:
return
if _job_is_past_its_end(job) or _job_is_over_spend_cap(job):
return # stop sampling now; the lifecycle loop finalizes the row
request_id: Final = payload.get("id") or ""
if not request_id:
return
# The job tracks every request it saw, sampled or not, so the UI can show
# "N of M requests shadowed". Flushed by the lifecycle loop.
self._pending_seen[job.id] = self._pending_seen.get(job.id, 0) + 1
if not _sample_hits(request_id, job.id, job.shadow_percentage):
return
if payload.get("call_type") not in ("completion", "acompletion"):
return # only known chat-shaped traffic is comparable; unknown or missing types fail closed
if _request_was_routed_by(request_metadata, job.router_name):
return
if self._inflight_shadow_tasks >= _MAX_CONCURRENT_SHADOW_TASKS:
return
# Redaction rewrites the logged messages and response before callbacks run,
# so a redacted request offers this hook only placeholders: evaluating them
# would produce garbage verdicts, and the caller opted that content out of
# logging anyway. The redactor's own predicate decides, so every redaction
# source (dynamic param, headers, global setting) is honored.
if should_redact_message_logging(dict(kwargs)): # mutable-ok: predicate takes a plain dict
return
raw_messages: Final = kwargs.get("messages")
self._inflight_shadow_tasks += 1
task: Final = asyncio.create_task(
self._run_shadow_eval(
job=job,
request_id=request_id,
messages=tuple(m for m in raw_messages if isinstance(m, Mapping))
if isinstance(raw_messages, Sequence)
else (),
response_obj=response_obj,
real_model=payload.get("model") or "",
model_parameters=MappingProxyType(
dict(payload.get("model_parameters") or {}) # mutable-ok: frozen snapshot
),
parent_metadata=MappingProxyType(dict(request_metadata)), # mutable-ok: frozen snapshot
)
)
task.add_done_callback(self._release_shadow_slot)
except Exception as e: # noqa: BLE001 # logging hooks must never fail the request
verbose_logger.debug("shadow_eval: failed to schedule task: %s", e)
def _release_shadow_slot(self, _task: "asyncio.Task[None]") -> None:
self._inflight_shadow_tasks -= 1
#### job lifecycle ####
async def _refresh_active_jobs(self) -> None:
"""Reload the active-job set. On a DB blip the stale snapshot is kept and the
next tick retries, so a blip degrades freshness rather than turning the feature off."""
prisma: Final = self._prisma_provider()
if prisma is None:
return
try:
records: Final = await prisma.db.litellm_shadowevaljob.find_many(
where={"status": {"in": ["pending", "running"]}}, # mutable-ok: Prisma filter
order={"created_at": "desc"}, # mutable-ok: Prisma order
)
except Exception as e: # noqa: BLE001 # a DB blip must not break request logging
verbose_logger.debug("shadow_eval: active-job refresh failed: %s", e)
return
jobs_by_key: Final[dict[str, ActiveShadowEvalJob]] = {} # mutable-ok: building the new snapshot
for record in reversed(records or []):
jobs_by_key[str(record.api_key_id)] = ActiveShadowEvalJob(
id=str(record.id),
router_name=str(record.router_name),
shadow_percentage=float(record.shadow_percentage),
judge_model=str(record.judge_model),
status=str(record.status),
cost_estimate=float(record.cost_estimate) if record.cost_estimate is not None else None,
cost_actual=float(record.cost_actual or 0.0),
ends_at=_as_utc(getattr(record, "ends_at", None)),
)
self._jobs_by_key = jobs_by_key # mutable-ok: atomic snapshot swap
async def _finalize_job(self, job: ActiveShadowEvalJob, reason: str) -> None:
"""Flip a finished job to completed, keeping the verdicts it already produced.
Guarded on the job still being active so two pods finishing the same job cannot
resurrect one an admin stopped in between."""
prisma: Final = self._prisma_provider()
if prisma is None:
return
self._jobs_by_key = { # mutable-ok: atomic snapshot swap
k: v for k, v in self._jobs_by_key.items() if v.id != job.id
}
verbose_logger.info("shadow_eval: stopping job %s: %s", job.id, reason)
try:
await prisma.db.litellm_shadowevaljob.update_many(
where={ # mutable-ok: Prisma filter
"id": job.id,
"status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter
},
data={ # mutable-ok: Prisma payload
"status": "completed",
"completed_at": datetime.now(timezone.utc),
},
)
except Exception as e: # noqa: BLE001 # the lifecycle loop must survive a failed write
verbose_logger.debug("shadow_eval: failed to stop job %s: %s", job.id, e)
async def _flush_seen_counts(self) -> None:
"""Write the buffered request counts, guarded on the job still being active, so
stopping a job freezes its counter: a pod on a stale snapshot keeps buffering for
up to one tick, and this write drops those increments instead of growing a
stopped job's request_count."""
prisma: Final = self._prisma_provider()
if prisma is None or not self._pending_seen:
return
pending: Final = self._pending_seen
self._pending_seen = {} # mutable-ok: fresh flush buffer
for job_id, count in pending.items():
try:
await prisma.db.litellm_shadowevaljob.update_many(
where={ # mutable-ok: Prisma filter
"id": job_id,
"status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter
},
data={"request_count": {"increment": count}}, # mutable-ok: Prisma payload
)
except Exception as e: # noqa: BLE001 # counter drift is acceptable; failing the loop is not
verbose_logger.debug("shadow_eval: request_count flush failed: %s", e)
#### the shadow pipeline ####
async def _run_shadow_eval(
self,
job: ActiveShadowEvalJob,
request_id: str,
messages: Sequence[Mapping[str, object]],
response_obj: object,
real_model: str,
model_parameters: Mapping[str, object],
parent_metadata: Mapping[str, object],
) -> None:
"""Detached background task: budget gate -> shadow call -> blind judge -> verdict.
The prisma gate sits above the shadow and judge dispatch so no provider spend
happens without a place to record the verdict, and the budget read lives here,
not in the success hook, because get_current_spend can fall back to an
authoritative DB read that the production callback must not absorb.
"""
prisma: Final = self._prisma_provider()
try:
if prisma is None:
return
real_text: Final = self._extract_response_text(response_obj)
if not real_text or not messages:
return
if await _key_or_team_is_over_budget(parent_metadata):
return
shadow: Final = await self._call_router_shadow(job.router_name, messages, model_parameters, parent_metadata)
if isinstance(shadow, _CallFailure):
await self._bump_failed(job.id, shadow.error)
return
verdict: Final = await self._call_judge(
judge_model=job.judge_model,
messages=messages,
real_text=real_text,
shadow_text=shadow.text,
parent_metadata=parent_metadata,
)
if isinstance(verdict, _CallFailure):
await self._bump_failed(job.id, verdict.error)
return
# One transaction records the outcome, so every pipeline lands in exactly
# one bucket and a job's counts always match its stored verdicts: the
# status-guarded counter update decides whether the verdict lands (a job
# stopped mid-flight matches zero rows and stores nothing), and a failed
# verdict write rolls the counters back before the outer handler files the
# pipeline under failed_count.
async with prisma.tx() as transaction:
counted: Final = await transaction.litellm_shadowevaljob.update_many(
where={ # mutable-ok: Prisma filter
"id": job.id,
"status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter
},
data={ # mutable-ok: Prisma payload
"completed_count": {"increment": 1}, # mutable-ok: Prisma operator
"cost_actual": {"increment": verdict.cost}, # mutable-ok: Prisma operator
"status": "running",
},
)
if counted:
await transaction.litellm_shadowevalverdict.create(
data={ # mutable-ok: Prisma payload
"job_id": job.id,
"request_id": request_id,
"tier_classification": shadow.tier,
"real_model": real_model,
"shadow_model": shadow.model,
"judge_preference": verdict.preference,
"judge_confidence": verdict.confidence,
}
)
except Exception as e: # noqa: BLE001 # detached task: log, count, never raise
verbose_logger.debug("shadow_eval: pipeline failed for %s: %s", request_id, e)
await self._bump_failed(job.id, f"pipeline error: {e}")
async def _bump_failed(self, job_id: str, error: str) -> None:
prisma: Final = self._prisma_provider()
if prisma is None:
return
try:
await prisma.db.litellm_shadowevaljob.update_many(
where={ # mutable-ok: Prisma filter
"id": job_id,
"status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter
},
data={ # mutable-ok: Prisma payload
"failed_count": {"increment": 1}, # mutable-ok: Prisma operator
"last_error": error[:_MAX_LAST_ERROR_CHARS],
},
)
except Exception as e: # noqa: BLE001 # counter drift is acceptable
verbose_logger.debug("shadow_eval: failed_count increment failed: %s", e)
async def _call_router_shadow(
self,
router_name: str,
messages: Sequence[Mapping[str, object]],
model_parameters: Mapping[str, object],
parent_metadata: Mapping[str, object],
) -> "_ShadowResponse | _CallFailure":
"""Send the prompt through the auto-router being evaluated. The metadata carries
the shadowed key's identity (spend attribution) and receives the router's routing
decision write-back, read back for tier attribution."""
router: Final = self._router_provider()
if router is None:
return _CallFailure("no router configured on this pod")
shadow_metadata: Final[dict[str, object]] = ( # mutable-ok: router writes its routing decision back
sanitized_forwardable_call_metadata( # mutable-ok: router writes back
parent_metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN
)
)
shadow_params: Final = { # mutable-ok: splatted as kwargs
k: v for k, v in model_parameters.items() if k not in ("stream", "metadata")
}
try:
response: Final = await router.acompletion(
model=router_name,
messages=messages, # pyright: ignore[reportArgumentType] # snapshot of the SDK's own message dicts
metadata=shadow_metadata,
num_retries=0,
fallbacks=[], # mutable-ok: SDK kwarg; a failed shadow is a counted miss, never a spend multiplier
**shadow_params,
)
except Exception as e: # noqa: BLE001 # provider errors are a counted failure, not a crash
verbose_logger.debug("shadow_eval: router call failed: %s", e)
return _CallFailure(f"shadow router call failed: {e}")
text: Final = self._extract_response_text(response)
if not text:
return _CallFailure("shadow router returned an empty response")
raw_decision: Final = shadow_metadata.get("routing_decision")
routing_decision: Final = raw_decision if isinstance(raw_decision, Mapping) else _EMPTY_METADATA
raw_tier: Final = routing_decision.get("tier_label") or routing_decision.get("tier")
return _ShadowResponse(
text=text,
model=str(getattr(response, "model", None) or routing_decision.get("routed_model") or ""),
tier=str(raw_tier) if raw_tier is not None else None,
)
async def _call_judge(
self,
judge_model: str,
messages: Sequence[Mapping[str, object]],
real_text: str,
shadow_text: str,
parent_metadata: Mapping[str, object],
) -> "_JudgeVerdict | _CallFailure":
"""Blind pairwise judge with A/B labels randomized to cancel position bias."""
real_is_a: Final = random.random() < 0.5
response_a: Final = real_text if real_is_a else shadow_text
response_b: Final = shadow_text if real_is_a else real_text
conversation: Final = "\n".join(
f"{str(m.get('role', 'user')).upper()}: {extract_text_from_content(m.get('content'))}"
for m in messages
if m.get("content") is not None
)
judge_metadata: Final = sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_JUDGE_CALL_ORIGIN)
judge_messages: Final = [ # mutable-ok: SDK takes a list
{"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT}, # mutable-ok: SDK message
{
"role": "user",
"content": _judge_user_prompt(conversation, response_a, response_b),
}, # mutable-ok: SDK message
]
try:
response: Final = await judge_acompletion(
self._router_provider(),
judge_model,
judge_messages, # pyright: ignore[reportArgumentType] # plain SDK message dicts
temperature=0,
max_tokens=JUDGE_MAX_OUTPUT_TOKENS,
metadata=judge_metadata,
)
except Exception as e: # noqa: BLE001 # judge outages are a counted failure, not a crash
verbose_logger.debug("shadow_eval: judge call failed: %s", e)
return _CallFailure(f"judge call failed: {e}")
try:
raw: Final = response["choices"][0]["message"]["content"] or ""
verdict: Final = PairwiseVerdict.model_validate(parse_json_verdict(raw))
except Exception as e: # noqa: BLE001 # malformed verdicts are a counted failure
verbose_logger.debug("shadow_eval: unparseable judge verdict: %s", e)
return _CallFailure(f"unparseable judge verdict: {e}")
return _JudgeVerdict(
preference=_unmask_preference(verdict.preference, real_is_a),
confidence=max(0.0, min(1.0, verdict.confidence)),
cost=_judge_call_cost(response),
)
@staticmethod
def _extract_response_text(response_obj: object) -> str:
"""Extract the assistant's text from a ModelResponse-shaped object or dict."""
try:
content: Final = (
response_obj["choices"][0]["message"]["content"]
if isinstance(response_obj, Mapping)
else response_obj.choices[0].message.content # pyright: ignore[reportAttributeAccessIssue] # duck-typed ModelResponse
)
except (AttributeError, KeyError, IndexError, TypeError):
return ""
return extract_text_from_content(content)
def _default_prisma_provider() -> "PrismaClient | None":
try:
from litellm.proxy.proxy_server import prisma_client
except ImportError:
return None
return prisma_client

View file

@ -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

View file

@ -0,0 +1,84 @@
"""Shared primitives for LLM-judge features (llm_as_a_judge guardrail, shadow eval)."""
from __future__ import annotations
import json
import re
from typing import TYPE_CHECKING, Final
import litellm
if TYPE_CHECKING:
from litellm import Router
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import ModelResponse
JSON_FENCE_RE: Final = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE)
def default_router_provider() -> Router | None:
try:
from litellm.proxy.proxy_server import llm_router
except ImportError:
return None
return llm_router
def parse_json_verdict(raw: str) -> dict[str, object]: # mutable-ok: plain parsed-JSON payload
"""Parse a judge's JSON verdict, tolerating markdown fences and surrounding prose."""
text = raw.strip() # rebind-ok: progressively narrowed to the JSON payload
fenced: Final = JSON_FENCE_RE.search(text)
if fenced is not None:
text = fenced.group(1).strip() # rebind-ok: progressively narrowed to the JSON payload
parsed: object
try:
parsed = json.loads(text)
except json.JSONDecodeError:
start: Final = text.find("{")
end: Final = text.rfind("}")
if start == -1 or end <= start:
raise
parsed = json.loads(text[start : end + 1])
if not isinstance(parsed, dict):
raise ValueError("judge response is not a JSON object")
return {str(k): v for k, v in parsed.items()} # mutable-ok: plain parsed-JSON payload
def extract_text_from_content(content: object) -> str:
"""Return plain text from a message content field (str or multimodal list)."""
if isinstance(content, str):
return content
if isinstance(content, list):
return " ".join(
str(part.get("text", "")) for part in content if isinstance(part, dict) and part.get("type") == "text"
)
return ""
def router_resolves_model(router: Router | None, model: str) -> bool:
"""Whether the model name resolves through the proxy's router (configured deployment
or model-group alias), the same check the judge dispatch itself makes, so start-time
validation cannot accept a name the call path then fails on."""
return router is not None and bool(model in router.model_group_alias or router.get_model_list(model_name=model))
async def judge_acompletion(
router: Router | None,
judge_model: str,
messages: list[AllMessageValues], # mutable-ok: the SDK acompletion signature takes a list
**params: object,
) -> ModelResponse:
"""Dispatch a judge call through the proxy's router when the judge model is a
configured deployment (DB-stored credentials work), through the SDK for
provider-qualified public names. The router path never retries or falls back:
a failed judge call is the caller's counted failure, not a spend multiplier."""
if router_resolves_model(router, judge_model):
return await router.acompletion( # pyright: ignore[reportOptionalMemberAccess] # router_resolves_model implies router is not None
model=judge_model,
messages=messages,
num_retries=0,
fallbacks=[],
**params,
)
return await litellm.acompletion(model=judge_model, messages=messages, num_retries=0, **params)

View file

@ -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

View file

@ -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:

View file

@ -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)

View file

@ -11,8 +11,11 @@ from typing import TYPE_CHECKING, Annotated, Final
from pydantic import BaseModel, TypeAdapter
import litellm
from litellm._logging import verbose_proxy_logger
from litellm.exceptions import BudgetExceededError
from litellm.integrations.shadow_eval_logger import JUDGE_MAX_OUTPUT_TOKENS
from litellm.litellm_core_utils.llm_judge import router_resolves_model
from litellm.proxy._types import (
CommonProxyErrors,
LiteLLM_TeamTable,
@ -38,12 +41,18 @@ from litellm.types.management_endpoints.auto_router_endpoints import (
AutoRouterCacheStats,
AutoRouterRoutingTestRequest,
AutoRouterRoutingTestResponse,
GetShadowEvalJobResponse,
RequestComplexityRouterConfig,
ShadowEvalResult,
ShadowEvalSlice,
StartShadowEvalRequest,
StartShadowEvalResponse,
)
if TYPE_CHECKING:
from fastapi import APIRouter, Depends, HTTPException, Query, status
from litellm.proxy.utils import PrismaClient
from litellm.router import Router
else:
try:
@ -388,14 +397,7 @@ async def get_auto_router_benchmarks(
"""
from litellm.proxy.proxy_server import prisma_client
if user_api_key_dict.user_role not in (
LitellmUserRoles.PROXY_ADMIN,
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
):
raise HTTPException(
status_code=403,
detail="Only proxy admin roles can view auto-router benchmarks across the deployment",
)
_require_admin_viewer(user_api_key_dict, "view auto-router benchmarks across the deployment")
if prisma_client is None:
raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
@ -430,3 +432,371 @@ async def get_auto_router_benchmarks(
totals=_benchmark_totals(_summed_agg_row(rows)),
groups=groups,
)
# ---------------------------------------------------------------------------
# Shadow eval: pre-adoption evaluation of an auto-router against live traffic.
# ---------------------------------------------------------------------------
# Judge price assumption for the upfront estimate: the conversation plus both responses
# on the prompt side, the judge's own bounded output budget on the completion side.
_FALLBACK_JUDGE_COST_PER_CALL: Final = 0.01
_JUDGE_PROMPT_TOKENS_ESTIMATE: Final = 4000
# The estimate projects from the key's request volume over this many trailing days,
# read from the LiteLLM_DailyUserSpend rollup (a handful of indexed rows per key/day),
# never LiteLLM_SpendLogs (unbounded per-request rows with no api_key index).
_ESTIMATE_LOOKBACK_DAYS: Final = 7
_ESTIMATE_VOLUME_SQL: Final = """
SELECT COALESCE(SUM(api_requests), 0)::bigint AS request_count
FROM "LiteLLM_DailyUserSpend"
WHERE api_key = $1 AND date >= $2
"""
class _EstimateVolumeRow(BaseModel):
request_count: int
_ESTIMATE_VOLUME_ROWS: Final = TypeAdapter(list[_EstimateVolumeRow])
async def _recent_request_volume(prisma_client: "PrismaClient", api_key_id: str) -> int:
lookback_date: Final = (datetime.now(timezone.utc) - timedelta(days=_ESTIMATE_LOOKBACK_DAYS)).strftime("%Y-%m-%d")
raw_rows: Final = await prisma_client.db.query_raw(_ESTIMATE_VOLUME_SQL, api_key_id, lookback_date)
rows: Final = _ESTIMATE_VOLUME_ROWS.validate_python(raw_rows or [])
return rows[0].request_count if rows else 0
def _is_unique_violation(error: Exception) -> bool:
"""Whether a Prisma create failed on a unique index. The one-active-job-per-key
guarantee lives in a partial unique index (raw SQL in the migration; schema.prisma
cannot express partial indexes), so the read-then-create check above it is advisory:
two concurrent starts pass the read, and the loser must surface as the same 409."""
try:
from prisma.errors import UniqueViolationError
except ImportError:
return "unique constraint" in str(error).lower() or "P2002" in str(error)
return isinstance(error, UniqueViolationError)
def _require_admin_viewer(user_api_key_dict: UserAPIKeyAuth, action: str) -> None:
if user_api_key_dict.user_role not in (
LitellmUserRoles.PROXY_ADMIN,
LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
):
raise HTTPException(status_code=403, detail=f"Only proxy admin roles can {action}")
def _require_admin_writer(user_api_key_dict: UserAPIKeyAuth, action: str) -> None:
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
raise HTTPException(status_code=403, detail=f"Only a proxy admin can {action}")
def _is_configured_pre_routing_strategy(llm_router: "Router", router_name: str) -> bool:
return any(
router_name in registry
for registry in (
llm_router.auto_routers,
llm_router.complexity_routers,
llm_router.adaptive_routers,
llm_router.quality_routers,
)
)
def _validate_judge_model(llm_router: "Router | None", judge_model: str) -> None:
"""Reject a judge model the dispatch path cannot resolve, at start rather than as a
silently growing failed_count once the job is already sampling and billing."""
if llm_router is not None and _is_configured_pre_routing_strategy(llm_router, judge_model):
raise HTTPException(
status_code=400,
detail=f"judge_model '{judge_model}' is an auto-router; the judge must be a plain model",
)
if router_resolves_model(llm_router, judge_model):
return
try:
litellm.get_llm_provider(model=judge_model)
except Exception as e:
raise HTTPException(
status_code=400,
detail=(
f"judge_model '{judge_model}' is neither a model configured on this proxy nor a "
"provider-qualified public model name (e.g. 'anthropic/claude-sonnet-5')"
),
) from e
def _judge_pricing_model(llm_router: "Router | None", judge_model: str) -> str:
"""The name to price the judge under: a configured deployment's underlying provider
model when the judge is a deployment (the deployment name itself is admin-arbitrary
and not a pricing key), the given name otherwise."""
deployments: Final = llm_router.get_model_list(model_name=judge_model) if llm_router is not None else None
if deployments:
underlying: Final = deployments[0].get("litellm_params", {}).get("model")
if isinstance(underlying, str) and underlying:
return underlying
return judge_model
def _estimate_judge_cost_per_call(llm_router: "Router | None", judge_model: str) -> float:
pricing_model: Final = _judge_pricing_model(llm_router, judge_model)
try:
prompt_cost, completion_cost = litellm.cost_per_token(
model=pricing_model, prompt_tokens=_JUDGE_PROMPT_TOKENS_ESTIMATE, completion_tokens=JUDGE_MAX_OUTPUT_TOKENS
)
estimated: Final = prompt_cost + completion_cost
if estimated > 0:
return estimated
except Exception as e: # noqa: BLE001 # unknown judge model: fall back to a flat per-call figure
verbose_proxy_logger.debug("shadow_eval: judge cost lookup failed for %s: %s", pricing_model, e)
return _FALLBACK_JUDGE_COST_PER_CALL
class _VerdictAggRow(BaseModel):
grp: str
turn_count: int
real_wins: int
shadow_wins: int
ties: int
avg_confidence: float | None
_VERDICT_AGG_ROWS: Final = TypeAdapter(list[_VerdictAggRow])
_VERDICT_AGG_SELECT: Final = """
COUNT(*)::int AS turn_count,
COUNT(*) FILTER (WHERE judge_preference = 'real')::int AS real_wins,
COUNT(*) FILTER (WHERE judge_preference = 'shadow')::int AS shadow_wins,
COUNT(*) FILTER (WHERE judge_preference = 'tie')::int AS ties,
AVG(judge_confidence)::float AS avg_confidence
FROM "LiteLLM_ShadowEvalVerdict"
WHERE job_id = $1
GROUP BY 1
"""
_VERDICT_AGG_BY_TIER_SQL: Final = "SELECT COALESCE(tier_classification, 'UNCLASSIFIED') AS grp," + _VERDICT_AGG_SELECT
_VERDICT_AGG_BY_MODEL_SQL: Final = "SELECT real_model AS grp," + _VERDICT_AGG_SELECT
def _slices(rows: Sequence[_VerdictAggRow]) -> tuple[ShadowEvalSlice, ...]:
return tuple(
ShadowEvalSlice(
group=row.grp,
turn_count=row.turn_count,
real_win_rate_pct=_pct(row.real_wins, row.turn_count),
shadow_win_rate_pct=_pct(row.shadow_wins, row.turn_count),
tie_rate_pct=_pct(row.ties, row.turn_count),
avg_judge_confidence=round(row.avg_confidence or 0.0, 3),
)
for row in sorted(rows, key=lambda r: r.turn_count, reverse=True)
)
async def _shadow_eval_results(prisma_client: "PrismaClient", job_id: str) -> ShadowEvalResult | None:
"""Both stratifications of one job's verdicts. Tier answers "where does the router
do well"; current-model answers "which of the models this key uses today would the
router beat". Row counts are bounded by the job's own verdicts, via the job_id index."""
by_tier: Final = _VERDICT_AGG_ROWS.validate_python(
await prisma_client.db.query_raw(_VERDICT_AGG_BY_TIER_SQL, job_id) or ()
)
if not by_tier:
return None
by_model: Final = _VERDICT_AGG_ROWS.validate_python(
await prisma_client.db.query_raw(_VERDICT_AGG_BY_MODEL_SQL, job_id) or ()
)
total_turns: Final = sum(r.turn_count for r in by_tier)
return ShadowEvalResult(
by_tier=_slices(by_tier),
by_current_model=_slices(by_model),
overall_shadow_win_rate_pct=_pct(sum(r.shadow_wins for r in by_tier), total_turns),
overall_tie_rate_pct=_pct(sum(r.ties for r in by_tier), total_turns),
)
def _job_to_response(record: object, results: ShadowEvalResult | None) -> GetShadowEvalJobResponse:
return GetShadowEvalJobResponse.model_validate(record, from_attributes=True).model_copy(
update={"results": results} # mutable-ok: pydantic update payload
)
@router.post(
"/auto_router/shadow_eval/start",
tags=("auto router",),
dependencies=(Depends(user_api_key_auth),),
response_model=StartShadowEvalResponse,
status_code=status.HTTP_201_CREATED,
)
async def start_shadow_eval(
data: StartShadowEvalRequest,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
) -> StartShadowEvalResponse:
"""
Start a pre-adoption shadow eval: duplicate a sampled slice of a key's live
traffic through an auto-router, judge real vs. shadow responses blind, and
stratify win rates by the router's tier classification.
The shadow responses are never served to users. The job samples traffic for
duration_days (or until stopped via /auto_router/shadow_eval/{job_id}/stop),
then completes itself. Judge calls bill to the shadowed key; the estimate
returned here prices them from the key's trailing request volume scaled to
the requested duration.
"""
from litellm.proxy.proxy_server import llm_router, prisma_client
_require_admin_writer(user_api_key_dict, "start a shadow eval")
if prisma_client is None:
raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
if llm_router is None or not _is_configured_pre_routing_strategy(llm_router, data.router_name):
raise HTTPException(
status_code=400,
detail=f"'{data.router_name}' is not a configured auto-router",
)
_validate_judge_model(llm_router, data.judge_model)
key_row: Final = await prisma_client.db.litellm_verificationtoken.find_unique(
where={"token": data.api_key_id} # mutable-ok: Prisma filter
)
if key_row is None:
raise HTTPException(
status_code=400,
detail=(
f"api_key_id '{data.api_key_id}' is not a key on this proxy; pass the key's token hash, "
"the value the key list and key info endpoints report"
),
)
existing: Final = await prisma_client.db.litellm_shadowevaljob.find_first(
where={ # mutable-ok: Prisma filter
"api_key_id": data.api_key_id,
"status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter
},
)
if existing is not None:
raise HTTPException(
status_code=409,
detail=f"Key already has an active shadow eval job ({existing.id}). Stop it first.",
)
recent_requests: Final = await _recent_request_volume(prisma_client, data.api_key_id)
sampled: Final = int(
recent_requests * (data.duration_days / _ESTIMATE_LOOKBACK_DAYS) * data.shadow_percentage / 100.0
)
per_call: Final = _estimate_judge_cost_per_call(llm_router, data.judge_model)
estimated_cost: Final = round(sampled * per_call, 2)
ends_at: Final = datetime.now(timezone.utc) + timedelta(days=data.duration_days)
try:
job: Final = await prisma_client.db.litellm_shadowevaljob.create(
data={ # mutable-ok: Prisma payload
"api_key_id": data.api_key_id,
"router_name": data.router_name,
"shadow_percentage": data.shadow_percentage,
"judge_model": data.judge_model,
"status": "pending",
"cost_estimate": estimated_cost,
"created_by": user_api_key_dict.user_id,
"ends_at": ends_at,
}
)
except Exception as e:
if not _is_unique_violation(e):
raise
raise HTTPException(
status_code=409,
detail="Key already has an active shadow eval job (started concurrently). Stop it first.",
) from e
return StartShadowEvalResponse(
job_id=job.id,
status="pending",
estimated_request_count=sampled,
estimated_cost=estimated_cost,
)
@router.get(
"/auto_router/shadow_eval",
tags=("auto router",),
dependencies=(Depends(user_api_key_auth),),
response_model=list[GetShadowEvalJobResponse],
)
async def list_shadow_eval_jobs(
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
api_key_id: Annotated[str | None, Query(description="Filter to jobs shadowing this key")] = None,
limit: Annotated[int, Query(ge=1, le=200, description="Newest jobs to return")] = 50,
) -> tuple[GetShadowEvalJobResponse, ...]:
"""List shadow eval jobs, newest first. Results are omitted; fetch a single job for them."""
from litellm.proxy.proxy_server import prisma_client
_require_admin_viewer(user_api_key_dict, "view shadow evals")
if prisma_client is None:
raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
where: Final = {"api_key_id": api_key_id} if api_key_id else {} # mutable-ok: Prisma filter
records: Final = await prisma_client.db.litellm_shadowevaljob.find_many(
where=where,
order={"created_at": "desc"}, # mutable-ok: Prisma order
take=limit,
)
return tuple(_job_to_response(record, results=None) for record in records or ())
@router.get(
"/auto_router/shadow_eval/{job_id}",
tags=("auto router",),
dependencies=(Depends(user_api_key_auth),),
response_model=GetShadowEvalJobResponse,
)
async def get_shadow_eval_job(
job_id: str,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
) -> GetShadowEvalJobResponse:
"""Status, counters, and stratified results of one shadow eval job."""
from litellm.proxy.proxy_server import prisma_client
_require_admin_viewer(user_api_key_dict, "view shadow evals")
if prisma_client is None:
raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
record: Final = await prisma_client.db.litellm_shadowevaljob.find_unique(
where={"id": job_id} # mutable-ok: Prisma filter
)
if record is None:
raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}")
return _job_to_response(record, results=await _shadow_eval_results(prisma_client, job_id))
@router.post(
"/auto_router/shadow_eval/{job_id}/stop",
tags=("auto router",),
dependencies=(Depends(user_api_key_auth),),
response_model=GetShadowEvalJobResponse,
)
async def stop_shadow_eval_job(
job_id: str,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
) -> GetShadowEvalJobResponse:
"""Stop an active shadow eval job. Existing verdicts are kept; sampling halts within ~10s."""
from litellm.proxy.proxy_server import prisma_client
_require_admin_writer(user_api_key_dict, "stop a shadow eval")
if prisma_client is None:
raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
record: Final = await prisma_client.db.litellm_shadowevaljob.find_unique(
where={"id": job_id} # mutable-ok: Prisma filter
)
if record is None:
raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}")
if record.status not in ("pending", "running"):
raise HTTPException(status_code=400, detail=f"Job {job_id} is already {record.status}")
updated: Final = await prisma_client.db.litellm_shadowevaljob.update(
where={"id": job_id}, # mutable-ok: Prisma filter
data={ # mutable-ok: Prisma payload
"status": "completed",
"completed_at": datetime.now(timezone.utc),
},
)
return _job_to_response(updated, results=await _shadow_eval_results(prisma_client, job_id))

View file

@ -2296,6 +2296,22 @@ def cost_tracking():
if prisma_client is not None:
litellm.logging_callback_manager.add_litellm_callback(_ProxyDBLogger())
litellm.logging_callback_manager.add_litellm_async_success_callback(_ProxyDBLogger())
_register_shadow_eval_logger()
def _register_shadow_eval_logger() -> None:
"""Register the shadow-eval success hook and start its lifecycle loop on the one
instance that is registered. Registration owns the loop start so the two cannot be
reordered apart, and the isinstance guard keeps a second cost_tracking() call from
constructing a duplicate whose loop would poll the DB from an unregistered instance.
Cheap when idle: with no active job rows the hook is one dict lookup per request."""
from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
if any(isinstance(callback, ShadowEvalLogger) for callback in litellm.callbacks):
return
shadow_eval_logger: Final = ShadowEvalLogger()
litellm.logging_callback_manager.add_litellm_callback(shadow_eval_logger)
shadow_eval_logger.start_lifecycle_loop()
# Bounds authoritative DB re-reads when enforcing a budget against a

View file

@ -1450,6 +1450,50 @@ model LiteLLM_AutoRouterSession {
@@index([last_turn_at], map: "idx_autorouter_session_last_turn")
}
// Shadow Eval: pre-adoption evaluation of an auto-router against a key's live traffic.
// A sampled slice of requests is duplicated through the router in a detached task and
// an LLM judge compares real vs shadow responses blind; verdicts stratify by tier.
model LiteLLM_ShadowEvalJob {
id String @id @default(cuid())
api_key_id String // the hashed virtual key whose traffic is shadowed
router_name String // the auto-router config to shadow through
shadow_percentage Float
judge_model String
status String @default("pending") // pending | running | completed
request_count Int @default(0) // requests seen on the key while active
completed_count Int @default(0) // verdicts written
failed_count Int @default(0) // shadow or judge calls that errored
last_error String?
cost_estimate Float? // upfront judge-spend estimate shown at start
cost_actual Float @default(0) // running judge-call spend
created_at DateTime @default(now())
created_by String?
ends_at DateTime?
completed_at DateTime?
@@index([api_key_id, status])
@@index([status])
@@index([created_at])
}
model LiteLLM_ShadowEvalVerdict {
id String @id @default(cuid())
job_id String
request_id String // the judged real request
tier_classification String? // the router's tier for the prompt, when classified
real_model String // model that actually served the request
shadow_model String // model the router picked
judge_preference String // real | shadow | tie
judge_confidence Float?
created_at DateTime @default(now())
@@index([job_id])
}
// ---------------------------------------------------------------------------
// Workflow Run Tracking
//

View file

@ -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,
@ -172,40 +173,6 @@ def _append_custom_keywords(base_keywords: list[str], custom_keywords: list[str]
return [*base_keywords, *deduped_custom.values()]
# Metadata keys that carry only the parent request's budget reservation state. These
# must not reach internal sub-calls (classifier, embedding): the reservation belongs to
# the routed completion being decided on, not to the sub-call itself, and forwarding it
# would let the sub-call's cost callback finalize the reservation, causing the routed
# completion's callback to skip incrementing key/team budget counters.
#
# Note: user_api_key_auth itself is intentionally kept; it is required by
# _filter_deployments_by_model_access_groups to scope embedding/classifier model
# selection to the caller's authorized access groups. It is forwarded as a sanitized
# copy with its budget_reservation sub-field removed, because the proxy cost callback
# (_get_budget_reservation_from_metadata) falls back to reading the reservation from
# inside the auth object when the top-level key is absent; forwarding it unsanitized
# would re-create the exact double-finalization this stripping exists to prevent.
_BUDGET_RESERVATION_METADATA_KEYS: Final = frozenset({"user_api_key_budget_reservation"})
def _sanitize_user_api_key_auth(auth: Any) -> Any:
if isinstance(auth, dict):
return {k: v for k, v in auth.items() if k != "budget_reservation"}
if getattr(auth, "budget_reservation", None) is not None and hasattr(auth, "model_copy"):
return auth.model_copy(update={"budget_reservation": None})
return auth
def _classifier_call_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
if not metadata:
return {}
return {
k: _sanitize_user_api_key_auth(v) if k == "user_api_key_auth" else v
for k, v in metadata.items()
if k not in _BUDGET_RESERVATION_METADATA_KEYS
} | {INTERNAL_CALL_ORIGIN_METADATA_KEY: AUTOROUTER_CLASSIFIER_CALL_ORIGIN}
def _parent_session_kwargs(request_kwargs: Mapping[str, Any] | None) -> Mapping[str, Any]:
kwargs: Final = request_kwargs or {}
return {k: kwargs[k] for k in ("litellm_session_id", "litellm_trace_id") if kwargs.get(k) is not None}
@ -1043,7 +1010,7 @@ 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()
@ -1535,8 +1502,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 = (

View file

@ -3,9 +3,10 @@ Types for auto-router management endpoints
"""
from collections.abc import Mapping
from typing import Final
from datetime import datetime
from typing import Final, Literal, TypeAlias
from pydantic import BaseModel, Field, field_validator
from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator
from litellm.router_strategy.complexity_router.config import ComplexityRouterConfig
from litellm.types.utils import StandardLoggingRoutingDecision
@ -141,3 +142,106 @@ class AutoRouterBenchmarksResponse(BaseModel):
routers_in_scope: int
totals: AutoRouterBenchmarkTotals
groups: tuple[AutoRouterBenchmarkGroup, ...]
ShadowEvalStatus: TypeAlias = Literal["pending", "running", "completed"]
DEFAULT_SHADOW_EVAL_JUDGE_MODEL: Final[str] = "anthropic/claude-sonnet-5"
class StartShadowEvalRequest(BaseModel):
"""Start shadowing a key's traffic through an auto-router for blind comparison."""
api_key_id: str = Field(
description=(
"The hashed virtual key whose traffic will be shadowed. Shadow evaluation runs ONLY on this "
"key's traffic; requests made with any other key are not sampled."
)
)
router_name: str = Field(description="The auto-router config to shadow requests through")
shadow_percentage: float = Field(
ge=0.1,
le=100.0,
description="Percentage of the key's requests to duplicate through the router",
)
judge_model: str = Field(
default=DEFAULT_SHADOW_EVAL_JUDGE_MODEL,
description=(
"Model used to blindly judge real vs. shadow responses. The judge only compares two answers, so a "
"mid-tier model (Claude Sonnet or GPT-4o class) is the sweet spot: small/nano-class models produce "
"unreliable or malformed verdicts, while frontier reasoning models add cost without changing outcomes."
),
)
duration_days: int = Field(
default=7,
ge=1,
le=30,
description="How many days the job samples traffic before stopping itself",
)
@field_validator("shadow_percentage")
@classmethod
def _round_percentage(cls, value: float) -> float:
return round(value, 2)
class StartShadowEvalResponse(BaseModel):
"""Acknowledgement that a shadow-eval job was created, with an upfront cost estimate."""
job_id: str
status: ShadowEvalStatus
estimated_request_count: int = Field(
description="Requests expected to be shadowed, based on the key's recent request volume"
)
estimated_cost: float = Field(description="Estimated dollar cost of the judge calls this job will make")
class ShadowEvalSlice(BaseModel):
"""Judge outcomes for one slice of a job's verdicts (a router tier, or one of the
models the shadowed key currently uses)."""
group: str
turn_count: int
real_win_rate_pct: float = Field(description="Share of judged turns where the real (control) model won")
shadow_win_rate_pct: float = Field(description="Share of judged turns where the shadowed router's pick won")
tie_rate_pct: float
avg_judge_confidence: float
class ShadowEvalResult(BaseModel):
"""Stratified results of a shadow-eval job's verdicts so far."""
by_tier: tuple[ShadowEvalSlice, ...]
by_current_model: tuple[ShadowEvalSlice, ...]
overall_shadow_win_rate_pct: float
overall_tie_rate_pct: float
class GetShadowEvalJobResponse(BaseModel):
"""Status and, once available, results of a shadow-eval job.
Validates directly from the prisma job record (job_id reads the row's id), so the
endpoint needs no hand-written row-to-response mapping.
"""
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
job_id: str = Field(validation_alias=AliasChoices("id", "job_id"))
status: ShadowEvalStatus
router_name: str
api_key_id: str = Field(description="The hashed virtual key whose traffic this job evaluates, and only that key's")
shadow_percentage: float
request_count: int = Field(description="Total requests observed on the shadowed key since the job started")
completed_count: int = Field(description="Verdicts written so far")
failed_count: int = Field(description="Shadow or judge calls that errored and were skipped")
last_error: str | None = Field(
default=None, description="The most recent shadow or judge failure, so a growing failed_count is diagnosable"
)
results: ShadowEvalResult | None = Field(
default=None, description="Present once at least one verdict has been recorded"
)
cost_estimate: float | None = None
cost_actual: float = Field(default=0.0, description="Running total of judge-call spend for this job")
created_at: datetime
ends_at: datetime | None = Field(default=None, description="When the job stops sampling on its own")
completed_at: datetime | None = None

View file

@ -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):

View file

@ -1450,6 +1450,50 @@ model LiteLLM_AutoRouterSession {
@@index([last_turn_at], map: "idx_autorouter_session_last_turn")
}
// Shadow Eval: pre-adoption evaluation of an auto-router against a key's live traffic.
// A sampled slice of requests is duplicated through the router in a detached task and
// an LLM judge compares real vs shadow responses blind; verdicts stratify by tier.
model LiteLLM_ShadowEvalJob {
id String @id @default(cuid())
api_key_id String // the hashed virtual key whose traffic is shadowed
router_name String // the auto-router config to shadow through
shadow_percentage Float
judge_model String
status String @default("pending") // pending | running | completed
request_count Int @default(0) // requests seen on the key while active
completed_count Int @default(0) // verdicts written
failed_count Int @default(0) // shadow or judge calls that errored
last_error String?
cost_estimate Float? // upfront judge-spend estimate shown at start
cost_actual Float @default(0) // running judge-call spend
created_at DateTime @default(now())
created_by String?
ends_at DateTime?
completed_at DateTime?
@@index([api_key_id, status])
@@index([status])
@@index([created_at])
}
model LiteLLM_ShadowEvalVerdict {
id String @id @default(cuid())
job_id String
request_id String // the judged real request
tier_classification String? // the router's tier for the prompt, when classified
real_model String // model that actually served the request
shadow_model String // model the router picked
judge_preference String // real | shadow | tie
judge_confidence Float?
created_at DateTime @default(now())
@@index([job_id])
}
// ---------------------------------------------------------------------------
// Workflow Run Tracking
//

View file

@ -0,0 +1,494 @@
"""Unit tests for the shadow-eval logger: sampling, unmasking, the success hook's skip
paths, the detached pipeline, and the lifecycle loop's flush/finalize behavior."""
import asyncio
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock
import pytest
from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY
from litellm.integrations.shadow_eval_logger import (
_MAX_CONCURRENT_SHADOW_TASKS,
_MAX_JUDGE_PROMPT_CHARS,
JUDGE_MAX_OUTPUT_TOKENS,
ActiveShadowEvalJob,
ShadowEvalLogger,
_CallFailure,
_judge_user_prompt,
_sample_hits,
_unmask_preference,
)
from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN
def _job(**overrides) -> ActiveShadowEvalJob:
defaults = dict(
id="job-1",
router_name="my-router",
shadow_percentage=100.0,
judge_model="judge-model",
status="running",
cost_estimate=5.0,
cost_actual=0.0,
ends_at=datetime.now(timezone.utc) + timedelta(days=1),
)
return ActiveShadowEvalJob(**{**defaults, **overrides})
def _prisma() -> MagicMock:
"""Prisma wrapper mock. Lifecycle/counter writes use prisma.db; the pipeline's
recording runs inside prisma.tx(), whose statements land on prisma.tx_mock."""
prisma = MagicMock()
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[])
prisma.db.litellm_shadowevaljob.update_many = AsyncMock(return_value=1)
prisma.db.litellm_shadowevalverdict.create = AsyncMock()
tx = MagicMock()
tx.litellm_shadowevaljob.update_many = AsyncMock(return_value=1)
tx.litellm_shadowevalverdict.create = AsyncMock()
ctx = MagicMock()
ctx.__aenter__ = AsyncMock(return_value=tx)
ctx.__aexit__ = AsyncMock(return_value=False)
prisma.tx = MagicMock(return_value=ctx)
prisma.tx_mock = tx
return prisma
def _router(shadow_text="shadow answer", judge_json='{"preference": "A", "confidence": 0.9, "reasoning": "x"}'):
"""One mock router serving the shadow call first, the judge call second. The shadow
call's metadata receives the routing decision write-back, like the real router."""
router = MagicMock()
router.model_group_alias = {}
router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "openai/gpt-4o-mini"}}])
async def acompletion(**kwargs):
if kwargs["model"] == "my-router":
kwargs["metadata"]["routing_decision"] = {"tier_label": "SIMPLE", "routed_model": "cheap-model"}
return {"choices": [{"message": {"content": shadow_text}}], "usage": {"completion_tokens": 5}}
return {"choices": [{"message": {"content": judge_json}}]}
router.acompletion = MagicMock(side_effect=acompletion)
return router
def _logger(router=None, prisma=None) -> ShadowEvalLogger:
return ShadowEvalLogger(
router_provider=lambda: router,
prisma_provider=lambda: prisma,
)
def _success_kwargs(request_id="req-1", api_key_hash="key-hash", request_metadata=None, call_type="acompletion"):
return {
"standard_logging_object": {
"id": request_id,
"call_type": call_type,
"model": "claude-opus",
"metadata": {"user_api_key_hash": api_key_hash},
"model_parameters": {"temperature": 0.5, "stream": True},
},
"litellm_params": {"metadata": request_metadata or {}},
"messages": [{"role": "user", "content": "what is 2+2"}],
}
RESPONSE = {"choices": [{"message": {"content": "real answer"}}]}
async def _drain(logger: ShadowEvalLogger):
for _ in range(100):
if logger._inflight_shadow_tasks == 0:
return
await asyncio.sleep(0.01)
raise AssertionError("shadow tasks never drained")
class TestSampling:
def test_boundaries_and_determinism(self):
assert not any(_sample_hits(f"req-{i}", "job", 0.0) for i in range(100))
assert all(_sample_hits(f"req-{i}", "job", 100.0) for i in range(100))
assert len({_sample_hits("req-1", "job-1", 50.0) for _ in range(10)}) == 1
def test_distribution_close_to_percentage(self):
hits = sum(_sample_hits(f"req-{i}", "job-x", 10.0) for i in range(10_000))
assert 800 < hits < 1200
def test_different_jobs_sample_independently(self):
agreements = sum(
_sample_hits(f"req-{i}", "job-a", 50.0) == _sample_hits(f"req-{i}", "job-b", 50.0) for i in range(1000)
)
assert 300 < agreements < 700
@pytest.mark.parametrize(
"raw,real_is_a,expected",
[
("A", True, "real"),
("a", True, "real"),
("A", False, "shadow"),
("B", True, "shadow"),
("B", False, "real"),
("tie", True, "tie"),
("garbage", True, "tie"),
("", False, "tie"),
],
)
def test_unmask_preference(raw, real_is_a, expected):
assert _unmask_preference(raw, real_is_a) == expected
def test_judge_prompt_is_bounded_however_large_the_inputs():
prompt = _judge_user_prompt("c" * 200_000, "a" * 200_000, "b" * 200_000)
assert len(prompt) < _MAX_JUDGE_PROMPT_CHARS + 100
assert prompt.endswith("Which response is better?")
small = _judge_user_prompt("conv", "alpha", "beta")
assert "conv" in small and "alpha" in small and "beta" in small
@pytest.mark.asyncio
class TestSuccessHookSkips:
"""Every skip path must leave no scheduled task; the sampled path must schedule one."""
async def test_happy_path_writes_a_verdict_and_bumps_counters(self, monkeypatch: pytest.MonkeyPatch):
import litellm as litellm_module
monkeypatch.setattr(litellm_module, "completion_cost", lambda completion_response: 0.005)
prisma = _prisma()
router = _router()
logger = _logger(router=router, prisma=prisma)
logger._jobs_by_key = {"key-hash": _job()}
await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None)
await _drain(logger)
create_kwargs = prisma.tx_mock.litellm_shadowevalverdict.create.call_args.kwargs["data"]
assert create_kwargs["job_id"] == "job-1"
assert create_kwargs["request_id"] == "req-1"
assert create_kwargs["tier_classification"] == "SIMPLE"
assert create_kwargs["real_model"] == "claude-opus"
assert create_kwargs["shadow_model"] == "cheap-model"
assert create_kwargs["judge_preference"] in ("real", "shadow")
assert create_kwargs["judge_confidence"] == 0.9
counter_update = prisma.tx_mock.litellm_shadowevaljob.update_many.call_args.kwargs
assert counter_update["data"]["completed_count"] == {"increment": 1}
assert counter_update["data"]["cost_actual"] == {"increment": 0.005}
assert counter_update["where"]["status"] == {"in": ["pending", "running"]}
assert logger._pending_seen == {"job-1": 1}
@pytest.mark.parametrize(
"kwargs_mutation,job_mutation",
[
({"request_metadata": {INTERNAL_CALL_ORIGIN_METADATA_KEY: "shadow_eval_router"}}, {}),
({"api_key_hash": "other-key"}, {}),
({"call_type": "aembedding"}, {}),
({"call_type": None}, {}),
({"request_metadata": {"routing_decision": {"router_model_name": "my-router"}}}, {}),
({}, {"ends_at": datetime.now(timezone.utc) - timedelta(seconds=1)}),
({}, {"cost_estimate": 1.0, "cost_actual": 99.0}),
],
ids=["internal-origin", "no-job-for-key", "non-chat", "missing-call-type", "self-shadow", "past-end", "over-spend-cap"],
)
async def test_skip_paths_schedule_nothing(self, kwargs_mutation, job_mutation):
prisma = _prisma()
logger = _logger(router=_router(), prisma=prisma)
logger._jobs_by_key = {"key-hash": _job(**job_mutation)}
await logger.async_log_success_event(_success_kwargs(**kwargs_mutation), RESPONSE, None, None)
await _drain(logger)
prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called()
assert logger._inflight_shadow_tasks == 0
async def test_inflight_cap_sheds_instead_of_queueing(self):
prisma = _prisma()
logger = _logger(router=_router(), prisma=prisma)
logger._jobs_by_key = {"key-hash": _job()}
logger._inflight_shadow_tasks = _MAX_CONCURRENT_SHADOW_TASKS
await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None)
assert logger._inflight_shadow_tasks == _MAX_CONCURRENT_SHADOW_TASKS
prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called()
async def test_sampled_out_request_still_counts_toward_request_count(self):
logger = _logger(router=_router(), prisma=_prisma())
logger._jobs_by_key = {"key-hash": _job(shadow_percentage=0.1)}
for i in range(20):
await logger.async_log_success_event(_success_kwargs(request_id=f"req-{i}"), RESPONSE, None, None)
await _drain(logger)
assert logger._pending_seen["job-1"] == 20
@pytest.mark.asyncio
class TestShadowPipeline:
async def test_no_prisma_means_no_provider_spend(self):
"""The prisma gate sits above the shadow and judge dispatch: no verdict store,
no spend."""
router = _router()
logger = _logger(router=router, prisma=None)
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={},
parent_metadata={},
)
router.acompletion.assert_not_called()
async def test_over_budget_key_skips_before_any_call(self, monkeypatch: pytest.MonkeyPatch):
"""The gate delegates to the auth path's own budget owner, so an over-budget
verdict there (BudgetExceededError) skips the shadow before any provider call."""
import litellm.proxy.auth.auth_checks as auth_checks
from litellm.exceptions import BudgetExceededError
from litellm.proxy._types import UserAPIKeyAuth
monkeypatch.setattr(
auth_checks,
"_virtual_key_max_budget_check",
AsyncMock(side_effect=BudgetExceededError(current_cost=11.0, max_budget=10.0)),
)
router = _router()
logger = _logger(router=router, prisma=_prisma())
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={},
parent_metadata={"user_api_key_auth": UserAPIKeyAuth(api_key="sk-abc", max_budget=10.0)},
)
router.acompletion.assert_not_called()
async def test_shadow_failure_bumps_failed_count_with_last_error(self):
prisma = _prisma()
router = MagicMock()
router.model_group_alias = {}
router.get_model_list = MagicMock(return_value=None)
router.acompletion = AsyncMock(side_effect=RuntimeError("provider exploded"))
logger = _logger(router=router, prisma=prisma)
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={},
parent_metadata={},
)
prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called()
bump = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs
assert bump["data"]["failed_count"] == {"increment": 1}
assert "provider exploded" in bump["data"]["last_error"]
async def test_unparseable_judge_verdict_is_a_counted_failure(self):
prisma = _prisma()
logger = _logger(router=_router(judge_json="I prefer response A, definitely"), prisma=prisma)
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={},
parent_metadata={},
)
prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called()
bump = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs
assert bump["data"]["failed_count"] == {"increment": 1}
async def test_redacted_requests_are_never_shadowed(self):
"""Redaction rewrites the logged messages before callbacks run, so this hook
only ever sees placeholders for opted-out traffic: sampling it would judge
garbage and put content the caller opted out of logging into sub-call rows.
The skip uses the redactor's own predicate, so every redaction source counts."""
prisma = _prisma()
router = _router()
logger = _logger(router=router, prisma=prisma)
logger._jobs_by_key = {"key-hash": _job()}
hook_kwargs = _success_kwargs()
hook_kwargs["standard_callback_dynamic_params"] = {"turn_off_message_logging": True}
await logger.async_log_success_event(hook_kwargs, RESPONSE, None, None)
await _drain(logger)
router.acompletion.assert_not_called()
prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called()
async def test_sub_calls_carry_identity_and_origin_but_never_parent_request_state(self):
prisma = _prisma()
router = _router()
logger = _logger(router=router, prisma=prisma)
parent_metadata = {
"user_api_key_hash": "key-hash",
"user_api_key_team_id": "team-1",
"user_api_key_budget_reservation": {"amount": 1.0},
"routing_decision": {"router_model_name": "other-router"},
}
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={"stream": True, "temperature": 0.2, "metadata": {"x": 1}},
parent_metadata=parent_metadata,
)
shadow_call = router.acompletion.call_args_list[0].kwargs
judge_call = router.acompletion.call_args_list[1].kwargs
for call in (shadow_call, judge_call):
assert call["num_retries"] == 0
assert call["fallbacks"] == []
assert shadow_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN
assert judge_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_JUDGE_CALL_ORIGIN
for call in (shadow_call, judge_call):
assert call["metadata"]["user_api_key_hash"] == "key-hash"
assert call["metadata"]["user_api_key_team_id"] == "team-1"
assert "user_api_key_budget_reservation" not in call["metadata"]
assert "routing_decision" not in judge_call["metadata"]
assert "stream" not in shadow_call
assert shadow_call["temperature"] == 0.2
assert judge_call["max_tokens"] == JUDGE_MAX_OUTPUT_TOKENS
async def test_job_stopped_mid_flight_drops_the_verdict(self):
"""The status-guarded counter update decides whether the verdict lands, so a
completed job's results can never disagree with its frozen counts."""
prisma = _prisma()
prisma.tx_mock.litellm_shadowevaljob.update_many = AsyncMock(return_value=0)
logger = _logger(router=_router(), prisma=prisma)
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={},
parent_metadata={},
)
prisma.tx_mock.litellm_shadowevalverdict.create.assert_not_called()
async def test_failed_verdict_write_files_the_pipeline_under_failed_once(self):
"""A create that raises inside the transaction rolls the counters back, and the
pipeline lands in failed_count exactly once, never in both buckets."""
prisma = _prisma()
prisma.tx_mock.litellm_shadowevalverdict.create = AsyncMock(side_effect=RuntimeError("db write failed"))
logger = _logger(router=_router(), prisma=prisma)
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={},
parent_metadata={},
)
bumps = [c.kwargs for c in prisma.db.litellm_shadowevaljob.update_many.call_args_list]
assert len(bumps) == 1
assert bumps[0]["data"]["failed_count"] == {"increment": 1}
assert "db write failed" in bumps[0]["data"]["last_error"]
async def test_recording_runs_inside_one_transaction(self):
"""Counter and verdict must ride prisma.tx(), never separate prisma.db writes,
so a failed verdict write cannot leave a counted-but-missing verdict."""
prisma = _prisma()
logger = _logger(router=_router(), prisma=prisma)
await logger._run_shadow_eval(
job=_job(),
request_id="req-1",
messages=({"role": "user", "content": "hi"},),
response_obj=RESPONSE,
real_model="claude-opus",
model_parameters={},
parent_metadata={},
)
prisma.tx.assert_called_once()
prisma.tx_mock.litellm_shadowevaljob.update_many.assert_awaited_once()
prisma.tx_mock.litellm_shadowevalverdict.create.assert_awaited_once()
prisma.db.litellm_shadowevaljob.update_many.assert_not_called()
prisma.db.litellm_shadowevalverdict.create.assert_not_called()
@pytest.mark.asyncio
class TestLifecycle:
async def test_flush_is_status_guarded_and_buffer_resets(self):
prisma = _prisma()
logger = _logger(prisma=prisma)
logger._pending_seen = {"job-1": 7}
await logger._flush_seen_counts()
flush = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs
assert flush["where"] == {"id": "job-1", "status": {"in": ["pending", "running"]}}
assert flush["data"] == {"request_count": {"increment": 7}}
assert logger._pending_seen == {}
async def test_tick_finalizes_expired_and_overspent_jobs(self):
prisma = _prisma()
expired = _job(id="job-expired", ends_at=datetime.now(timezone.utc) - timedelta(seconds=1))
overspent = _job(id="job-overspent", cost_estimate=1.0, cost_actual=99.0)
active = _job(id="job-active")
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(
return_value=[
MagicMock(
id=j.id,
api_key_id=f"key-{j.id}",
router_name=j.router_name,
shadow_percentage=j.shadow_percentage,
judge_model=j.judge_model,
status=j.status,
cost_estimate=j.cost_estimate,
cost_actual=j.cost_actual,
ends_at=j.ends_at,
)
for j in (expired, overspent, active)
]
)
logger = _logger(prisma=prisma)
await logger._lifecycle_tick()
finalized = {c.kwargs["where"]["id"] for c in prisma.db.litellm_shadowevaljob.update_many.call_args_list}
assert finalized == {"job-expired", "job-overspent"}
for call in prisma.db.litellm_shadowevaljob.update_many.call_args_list:
assert call.kwargs["where"]["status"] == {"in": ["pending", "running"]}
assert call.kwargs["data"]["status"] == "completed"
assert set(logger._jobs_by_key) == {"key-job-active"}
async def test_refresh_keeps_stale_snapshot_on_db_blip(self):
prisma = _prisma()
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=RuntimeError("db blip"))
logger = _logger(prisma=prisma)
logger._jobs_by_key = {"key-hash": _job()}
await logger._refresh_active_jobs()
assert set(logger._jobs_by_key) == {"key-hash"}
async def test_start_lifecycle_loop_is_idempotent(self):
logger = _logger(prisma=_prisma())
logger.start_lifecycle_loop()
first: asyncio.Task = logger._lifecycle_task
logger.start_lifecycle_loop()
assert logger._lifecycle_task is first
first.cancel()

View file

@ -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}

View file

@ -0,0 +1,88 @@
"""Unit tests for the shared LLM-judge primitives: verdict parsing, router resolution, dispatch."""
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
from litellm.litellm_core_utils.llm_judge import (
extract_text_from_content,
judge_acompletion,
parse_json_verdict,
router_resolves_model,
)
@pytest.mark.parametrize(
"raw,expected",
[
('{"preference": "A", "confidence": 0.9}', "A"),
('Here it is:\n```json\n{"preference": "B"}\n```\nDone.', "B"),
('```\n{"preference": "tie"}\n```', "tie"),
('Verdict: {"preference": "A", "confidence": 0.5} final.', "A"),
],
)
def test_parse_json_verdict_tolerates_fences_and_prose(raw, expected):
assert parse_json_verdict(raw)["preference"] == expected
def test_parse_json_verdict_rejects_non_object():
with pytest.raises(ValueError):
parse_json_verdict('["not", "an", "object"]')
with pytest.raises((json.JSONDecodeError, ValueError)):
parse_json_verdict("no json here at all")
@pytest.mark.parametrize(
"content,expected",
[
("hello", "hello"),
([{"type": "text", "text": "a"}, {"type": "image_url", "image_url": {}}, {"type": "text", "text": "b"}], "a b"),
(42, ""),
(None, ""),
],
)
def test_extract_text_from_content(content, expected):
assert extract_text_from_content(content) == expected
def _router(alias=(), deployments=False) -> MagicMock:
router = MagicMock()
router.model_group_alias = dict.fromkeys(alias, "x")
router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "openai/gpt-4o"}}] if deployments else None)
router.acompletion = AsyncMock(return_value={"choices": [{"message": {"content": "router answer"}}]})
return router
def test_router_resolves_model_matrix():
assert router_resolves_model(None, "gpt-4o") is False
assert router_resolves_model(_router(), "gpt-4o") is False
assert router_resolves_model(_router(alias=("gpt-4o",)), "gpt-4o") is True
assert router_resolves_model(_router(deployments=True), "gpt-4o") is True
@pytest.mark.asyncio
async def test_judge_acompletion_prefers_router_and_disables_retries():
router = _router(deployments=True)
response = await judge_acompletion(router, "judge-model", [{"role": "user", "content": "hi"}], temperature=0)
assert response == {"choices": [{"message": {"content": "router answer"}}]}
_, kwargs = router.acompletion.call_args
assert kwargs["num_retries"] == 0
assert kwargs["fallbacks"] == []
assert kwargs["temperature"] == 0
@pytest.mark.asyncio
async def test_judge_acompletion_falls_back_to_sdk_for_unconfigured_model(monkeypatch: pytest.MonkeyPatch):
import litellm as litellm_module
sdk = AsyncMock(return_value={"choices": [{"message": {"content": "sdk answer"}}]})
monkeypatch.setattr(litellm_module, "acompletion", sdk)
router = _router()
response = await judge_acompletion(router, "anthropic/claude-sonnet-5", [{"role": "user", "content": "hi"}])
assert response == {"choices": [{"message": {"content": "sdk answer"}}]}
router.acompletion.assert_not_called()
assert sdk.call_args.kwargs["model"] == "anthropic/claude-sonnet-5"
assert sdk.call_args.kwargs["num_retries"] == 0

View file

@ -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

View file

@ -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

View file

@ -466,3 +466,251 @@ class TestAutoRouterBenchmarks:
end_date="2026-08-01",
)
assert response.groups[0].tier_turns == expected
# ---------------------------------------------------------------------------
# Shadow eval endpoints
# ---------------------------------------------------------------------------
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock
from fastapi import HTTPException
from litellm.proxy.management_endpoints.auto_router_endpoints import (
_estimate_judge_cost_per_call,
_FALLBACK_JUDGE_COST_PER_CALL,
get_shadow_eval_job,
list_shadow_eval_jobs,
start_shadow_eval,
stop_shadow_eval_job,
)
from litellm.types.management_endpoints.auto_router_endpoints import StartShadowEvalRequest
VIEWER = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, api_key="sk-view", user_id="viewer")
NON_ADMIN = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, api_key="sk-user", user_id="user")
def _shadow_router() -> MagicMock:
router = MagicMock()
router.auto_routers = {}
router.complexity_routers = {"my-router": [MagicMock()]}
router.adaptive_routers = {}
router.quality_routers = {}
router.model_group_alias = {}
router.get_model_list = MagicMock(return_value=None)
return router
def _job_record(**overrides: object) -> MagicMock:
record = MagicMock()
defaults = {
"id": "job-1",
"status": "running",
"router_name": "my-router",
"api_key_id": "key-hash",
"shadow_percentage": 10.0,
"request_count": 40,
"completed_count": 3,
"failed_count": 1,
"last_error": "judge call failed: boom",
"cost_estimate": 2.5,
"cost_actual": 0.03,
"created_at": datetime(2026, 8, 11, tzinfo=timezone.utc),
"ends_at": None,
"completed_at": None,
}
for key, value in {**defaults, **overrides}.items():
setattr(record, key, value)
return record
def _shadow_prisma(existing_job=None, volume_rows=None, agg_rows=None) -> MagicMock:
prisma = MagicMock()
prisma.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=MagicMock())
prisma.db.litellm_shadowevaljob.find_first = AsyncMock(return_value=existing_job)
prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=None)
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[])
prisma.db.litellm_shadowevaljob.create = AsyncMock(return_value=_job_record(status="pending"))
prisma.db.litellm_shadowevaljob.update = AsyncMock(return_value=_job_record(status="completed"))
async def query_raw(sql: str, *params: object):
if "LiteLLM_DailyUserSpend" in sql:
return volume_rows if volume_rows is not None else [{"request_count": 0}]
return agg_rows if agg_rows is not None else []
prisma.db.query_raw = AsyncMock(side_effect=query_raw)
return prisma
def _start_request(**overrides: object) -> StartShadowEvalRequest:
payload = {
"api_key_id": "key-hash",
"router_name": "my-router",
"shadow_percentage": 10.0,
"judge_model": "anthropic/claude-sonnet-5",
"duration_days": 7,
}
payload.update(overrides)
return StartShadowEvalRequest.model_validate(payload)
@pytest.mark.asyncio
async def test_start_shadow_eval_creates_job_with_volume_scaled_estimate(monkeypatch: pytest.MonkeyPatch):
import litellm.proxy.proxy_server as proxy_server
prisma = _shadow_prisma(volume_rows=[{"request_count": 7000}])
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
monkeypatch.setattr(proxy_server, "llm_router", _shadow_router())
response = await start_shadow_eval(_start_request(), ADMIN)
assert response.status == "pending"
assert response.estimated_request_count == 700
per_call = _estimate_judge_cost_per_call(None, "anthropic/claude-sonnet-5")
assert response.estimated_cost == round(700 * per_call, 2)
create_data = prisma.db.litellm_shadowevaljob.create.call_args.kwargs["data"]
assert create_data["api_key_id"] == "key-hash"
assert create_data["created_by"] == "admin"
assert create_data["ends_at"] is not None
@pytest.mark.asyncio
@pytest.mark.parametrize(
"caller,request_overrides,existing,expected_status",
[
(NON_ADMIN, {}, None, 403),
(VIEWER, {}, None, 403),
(ADMIN, {"router_name": "not-a-router"}, None, 400),
(ADMIN, {"judge_model": "not/a real model!"}, None, 400),
(ADMIN, {"judge_model": "my-router"}, None, 400),
(ADMIN, {}, "existing", 409),
],
ids=["non-admin", "view-only", "unknown-router", "unresolvable-judge", "router-as-judge", "already-active"],
)
async def test_start_shadow_eval_rejections(
monkeypatch: pytest.MonkeyPatch, caller, request_overrides, existing, expected_status
):
import litellm.proxy.proxy_server as proxy_server
existing_job = _job_record() if existing else None
monkeypatch.setattr(proxy_server, "prisma_client", _shadow_prisma(existing_job=existing_job))
monkeypatch.setattr(proxy_server, "llm_router", _shadow_router())
with pytest.raises(HTTPException) as exc:
await start_shadow_eval(_start_request(**request_overrides), caller)
assert exc.value.status_code == expected_status
@pytest.mark.asyncio
async def test_start_shadow_eval_rejects_a_key_this_proxy_does_not_know(monkeypatch: pytest.MonkeyPatch):
"""A typo'd api_key_id would otherwise create a job no traffic can ever match."""
import litellm.proxy.proxy_server as proxy_server
prisma = _shadow_prisma()
prisma.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None)
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
monkeypatch.setattr(proxy_server, "llm_router", _shadow_router())
with pytest.raises(HTTPException) as exc:
await start_shadow_eval(_start_request(), ADMIN)
assert exc.value.status_code == 400
assert "not a key on this proxy" in exc.value.detail
@pytest.mark.asyncio
async def test_start_shadow_eval_concurrent_unique_violation_is_a_409(monkeypatch: pytest.MonkeyPatch):
import litellm.proxy.proxy_server as proxy_server
from prisma.errors import UniqueViolationError
prisma = _shadow_prisma()
prisma.db.litellm_shadowevaljob.create = AsyncMock(
side_effect=UniqueViolationError(MagicMock(message="unique constraint"))
)
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
monkeypatch.setattr(proxy_server, "llm_router", _shadow_router())
with pytest.raises(HTTPException) as exc:
await start_shadow_eval(_start_request(), ADMIN)
assert exc.value.status_code == 409
def test_estimate_falls_back_for_unpriced_judge_model():
assert _estimate_judge_cost_per_call(None, "unknown/never-priced-model") == _FALLBACK_JUDGE_COST_PER_CALL
@pytest.mark.asyncio
async def test_get_shadow_eval_job_stratifies_by_tier_and_model(monkeypatch: pytest.MonkeyPatch):
import litellm.proxy.proxy_server as proxy_server
tier_rows = [
{"grp": "SIMPLE", "turn_count": 8, "real_wins": 2, "shadow_wins": 4, "ties": 2, "avg_confidence": 0.8},
{"grp": "REASONING", "turn_count": 2, "real_wins": 2, "shadow_wins": 0, "ties": 0, "avg_confidence": 0.9},
]
prisma = _shadow_prisma(agg_rows=tier_rows)
prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record())
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
response = await get_shadow_eval_job("job-1", VIEWER)
assert response.job_id == "job-1"
assert response.last_error == "judge call failed: boom"
assert response.results is not None
assert [s.group for s in response.results.by_tier] == ["SIMPLE", "REASONING"]
assert response.results.by_tier[0].shadow_win_rate_pct == 50.0
assert response.results.overall_shadow_win_rate_pct == 40.0
assert response.results.overall_tie_rate_pct == 20.0
@pytest.mark.asyncio
async def test_get_shadow_eval_job_404s_and_gates_on_role(monkeypatch: pytest.MonkeyPatch):
import litellm.proxy.proxy_server as proxy_server
monkeypatch.setattr(proxy_server, "prisma_client", _shadow_prisma())
with pytest.raises(HTTPException) as missing:
await get_shadow_eval_job("nope", VIEWER)
assert missing.value.status_code == 404
with pytest.raises(HTTPException) as forbidden:
await get_shadow_eval_job("job-1", NON_ADMIN)
assert forbidden.value.status_code == 403
@pytest.mark.asyncio
async def test_list_shadow_eval_jobs_omits_results(monkeypatch: pytest.MonkeyPatch):
import litellm.proxy.proxy_server as proxy_server
prisma = _shadow_prisma()
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record()])
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50)
assert len(jobs) == 1
assert jobs[0].results is None
find_kwargs = prisma.db.litellm_shadowevaljob.find_many.call_args.kwargs
assert find_kwargs["take"] == 50
@pytest.mark.asyncio
async def test_stop_shadow_eval_completes_active_job_and_rejects_finished(monkeypatch: pytest.MonkeyPatch):
import litellm.proxy.proxy_server as proxy_server
prisma = _shadow_prisma()
prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record(status="running"))
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
stopped = await stop_shadow_eval_job("job-1", ADMIN)
assert stopped.status == "completed"
update_kwargs = prisma.db.litellm_shadowevaljob.update.call_args.kwargs
assert update_kwargs["data"]["status"] == "completed"
prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record(status="completed"))
with pytest.raises(HTTPException) as exc:
await stop_shadow_eval_job("job-1", ADMIN)
assert exc.value.status_code == 400
with pytest.raises(HTTPException) as forbidden:
await stop_shadow_eval_job("job-1", VIEWER)
assert forbidden.value.status_code == 403

View file

@ -480,8 +480,9 @@ def test_load_from_azure_key_vault_missing_uri_failure_is_swallowed(monkeypatch)
# ---------------------------------------------------------------------------
def test_cost_tracking_adds_two_callbacks_when_prisma_set(monkeypatch):
def test_cost_tracking_adds_db_and_shadow_eval_callbacks_when_prisma_set(monkeypatch):
import litellm
from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
fake_prisma = MagicMock()
monkeypatch.setattr(ps, "prisma_client", fake_prisma, raising=False)
@ -496,15 +497,53 @@ def test_cost_tracking_adds_two_callbacks_when_prisma_set(monkeypatch):
observed = {
"added_to_callbacks": len(litellm.callbacks) - before_callbacks,
"added_to_async_success": len(litellm._async_success_callback) - before_async,
"shadow_eval_loggers": sum(isinstance(cb, ShadowEvalLogger) for cb in litellm.callbacks),
"prisma_was_set": True,
}
assert normalize(observed) == {
"added_to_callbacks": 1,
"added_to_callbacks": 2,
"added_to_async_success": 1,
"shadow_eval_loggers": 1,
"prisma_was_set": True,
}
def test_cost_tracking_twice_registers_one_shadow_eval_logger(monkeypatch):
import litellm
from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
monkeypatch.setattr(ps, "prisma_client", MagicMock(), raising=False)
monkeypatch.setattr(litellm, "callbacks", [], raising=False)
monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False)
cost_tracking()
cost_tracking()
assert sum(isinstance(cb, ShadowEvalLogger) for cb in litellm.callbacks) == 1
@pytest.mark.asyncio
async def test_registration_starts_the_lifecycle_loop_on_the_registered_instance(monkeypatch):
"""The loop must run on the instance that is actually registered; a loop on a
discarded duplicate would poll the DB while never seeing request traffic."""
import litellm
from litellm.integrations.shadow_eval_logger import ShadowEvalLogger
monkeypatch.setattr(ps, "prisma_client", MagicMock(), raising=False)
monkeypatch.setattr(litellm, "callbacks", [], raising=False)
monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False)
cost_tracking()
registered = [cb for cb in litellm.callbacks if isinstance(cb, ShadowEvalLogger)]
assert len(registered) == 1
task = registered[0]._lifecycle_task
assert task is not None and not task.done()
cost_tracking()
assert registered[0]._lifecycle_task is task
task.cancel()
def test_cost_tracking_no_op_when_prisma_missing(monkeypatch):
"""Without a prisma_client cost_tracking is a no-op — not an error."""
import litellm

View file

@ -3222,98 +3222,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.

View file

@ -807,6 +807,94 @@ export interface paths {
patch?: never;
trace?: never;
};
"/auto_router/shadow_eval": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/**
* List Shadow Eval Jobs
* @description List shadow eval jobs, newest first. Results are omitted; fetch a single job for them.
*/
get: operations["list_shadow_eval_jobs_auto_router_shadow_eval_get"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/auto_router/shadow_eval/start": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Start Shadow Eval
* @description Start a pre-adoption shadow eval: duplicate a sampled slice of a key's live
* traffic through an auto-router, judge real vs. shadow responses blind, and
* stratify win rates by the router's tier classification.
*
* The shadow responses are never served to users. The job samples traffic for
* duration_days (or until stopped via /auto_router/shadow_eval/{job_id}/stop),
* then completes itself. Judge calls bill to the shadowed key; the estimate
* returned here prices them from the key's trailing request volume scaled to
* the requested duration.
*/
post: operations["start_shadow_eval_auto_router_shadow_eval_start_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/auto_router/shadow_eval/{job_id}": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/**
* Get Shadow Eval Job
* @description Status, counters, and stratified results of one shadow eval job.
*/
get: operations["get_shadow_eval_job_auto_router_shadow_eval__job_id__get"];
put?: never;
post?: never;
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/auto_router/shadow_eval/{job_id}/stop": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Stop Shadow Eval Job
* @description Stop an active shadow eval job. Existing verdicts are kept; sampling halts within ~10s.
*/
post: operations["stop_shadow_eval_job_auto_router_shadow_eval__job_id__stop_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/auto_router/test_routing": {
parameters: {
query?: never;
@ -25270,6 +25358,73 @@ export interface components {
/** Tools */
tools?: components["schemas"]["ChatCompletionToolParam"][];
};
/**
* GetShadowEvalJobResponse
* @description Status and, once available, results of a shadow-eval job.
*
* Validates directly from the prisma job record (job_id reads the row's id), so the
* endpoint needs no hand-written row-to-response mapping.
*/
GetShadowEvalJobResponse: {
/**
* Api Key Id
* @description The hashed virtual key whose traffic this job evaluates, and only that key's
*/
api_key_id: string;
/** Completed At */
completed_at?: string | null;
/**
* Completed Count
* @description Verdicts written so far
*/
completed_count: number;
/**
* Cost Actual
* @description Running total of judge-call spend for this job
* @default 0
*/
cost_actual: number;
/** Cost Estimate */
cost_estimate?: number | null;
/**
* Created At
* Format: date-time
*/
created_at: string;
/**
* Ends At
* @description When the job stops sampling on its own
*/
ends_at?: string | null;
/**
* Failed Count
* @description Shadow or judge calls that errored and were skipped
*/
failed_count: number;
/** Job Id */
job_id: string;
/**
* Last Error
* @description The most recent shadow or judge failure, so a growing failed_count is diagnosable
*/
last_error?: string | null;
/**
* Request Count
* @description Total requests observed on the shadowed key since the job started
*/
request_count: number;
/** @description Present once at least one verdict has been recorded */
results?: components["schemas"]["ShadowEvalResult"] | null;
/** Router Name */
router_name: string;
/** Shadow Percentage */
shadow_percentage: number;
/**
* Status
* @enum {string}
*/
status: "pending" | "running" | "completed";
};
/**
* GetTeamMemberPermissionsResponse
* @description Response to get the team member permissions for a team
@ -32516,6 +32671,45 @@ export interface components {
/** Timeout */
timeout?: number | null;
};
/**
* ShadowEvalResult
* @description Stratified results of a shadow-eval job's verdicts so far.
*/
ShadowEvalResult: {
/** By Current Model */
by_current_model: components["schemas"]["ShadowEvalSlice"][];
/** By Tier */
by_tier: components["schemas"]["ShadowEvalSlice"][];
/** Overall Shadow Win Rate Pct */
overall_shadow_win_rate_pct: number;
/** Overall Tie Rate Pct */
overall_tie_rate_pct: number;
};
/**
* ShadowEvalSlice
* @description Judge outcomes for one slice of a job's verdicts (a router tier, or one of the
* models the shadowed key currently uses).
*/
ShadowEvalSlice: {
/** Avg Judge Confidence */
avg_judge_confidence: number;
/** Group */
group: string;
/**
* Real Win Rate Pct
* @description Share of judged turns where the real (control) model won
*/
real_win_rate_pct: number;
/**
* Shadow Win Rate Pct
* @description Share of judged turns where the shadowed router's pick won
*/
shadow_win_rate_pct: number;
/** Tie Rate Pct */
tie_rate_pct: number;
/** Turn Count */
turn_count: number;
};
/**
* Skill
* @description Represents a skill from the Anthropic Skills API
@ -32689,6 +32883,62 @@ export interface components {
/** Simple Medium */
simple_medium: number;
};
/**
* StartShadowEvalRequest
* @description Start shadowing a key's traffic through an auto-router for blind comparison.
*/
StartShadowEvalRequest: {
/**
* Api Key Id
* @description The hashed virtual key whose traffic will be shadowed. Shadow evaluation runs ONLY on this key's traffic; requests made with any other key are not sampled.
*/
api_key_id: string;
/**
* Duration Days
* @description How many days the job samples traffic before stopping itself
* @default 7
*/
duration_days: number;
/**
* Judge Model
* @description Model used to blindly judge real vs. shadow responses. The judge only compares two answers, so a mid-tier model (Claude Sonnet or GPT-4o class) is the sweet spot: small/nano-class models produce unreliable or malformed verdicts, while frontier reasoning models add cost without changing outcomes.
* @default anthropic/claude-sonnet-5
*/
judge_model: string;
/**
* Router Name
* @description The auto-router config to shadow requests through
*/
router_name: string;
/**
* Shadow Percentage
* @description Percentage of the key's requests to duplicate through the router
*/
shadow_percentage: number;
};
/**
* StartShadowEvalResponse
* @description Acknowledgement that a shadow-eval job was created, with an upfront cost estimate.
*/
StartShadowEvalResponse: {
/**
* Estimated Cost
* @description Estimated dollar cost of the judge calls this job will make
*/
estimated_cost: number;
/**
* Estimated Request Count
* @description Requests expected to be shadowed, based on the key's recent request volume
*/
estimated_request_count: number;
/** Job Id */
job_id: string;
/**
* Status
* @enum {string}
*/
status: "pending" | "running" | "completed";
};
/**
* SuccessfulKeyUpdate
* @description Successfully updated key with its updated information
@ -36964,6 +37214,135 @@ export interface operations {
};
};
};
list_shadow_eval_jobs_auto_router_shadow_eval_get: {
parameters: {
query?: {
/** @description Filter to jobs shadowing this key */
api_key_id?: string | null;
/** @description Newest jobs to return */
limit?: number;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["GetShadowEvalJobResponse"][];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
start_shadow_eval_auto_router_shadow_eval_start_post: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody: {
content: {
"application/json": components["schemas"]["StartShadowEvalRequest"];
};
};
responses: {
/** @description Successful Response */
201: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["StartShadowEvalResponse"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
get_shadow_eval_job_auto_router_shadow_eval__job_id__get: {
parameters: {
query?: never;
header?: never;
path: {
job_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["GetShadowEvalJobResponse"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
stop_shadow_eval_job_auto_router_shadow_eval__job_id__stop_post: {
parameters: {
query?: never;
header?: never;
path: {
job_id: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["GetShadowEvalJobResponse"];
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
preview_auto_router_routing_auto_router_test_routing_post: {
parameters: {
query?: never;