diff --git a/litellm/integrations/shadow_eval_logger.py b/litellm/integrations/shadow_eval_logger.py index fb4139f62a7..2a4a2067882 100644 --- a/litellm/integrations/shadow_eval_logger.py +++ b/litellm/integrations/shadow_eval_logger.py @@ -24,8 +24,12 @@ import hashlib import json import random import re -from collections.abc import Callable -from typing import TYPE_CHECKING, Any, Final, Optional, cast +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from types import MappingProxyType +from typing import TYPE_CHECKING, Final, TypeAlias + +from pydantic import BaseModel, TypeAdapter import litellm from litellm._logging import verbose_logger @@ -55,6 +59,8 @@ _MAX_JUDGE_CHARS: Final = 16_000 _SEEN_FLUSH_INTERVAL_SECONDS: Final = 10.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. @@ -69,36 +75,40 @@ Return ONLY valid JSON in this exact format, no other text: }""" -def _parse_pairwise_verdict(raw: str) -> dict[str, Any]: +class PairwiseVerdict(BaseModel): + """The judge's blind A/B verdict, validated at the parse boundary.""" + + preference: str = "tie" + confidence: float = 0.0 + reasoning: str = "" + + +_VERDICT_ADAPTER: Final = TypeAdapter(PairwiseVerdict) + + +def _parse_pairwise_verdict(raw: str) -> PairwiseVerdict: """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 + stripped: Final = raw.strip() + fenced: Final = _JSON_FENCE_RE.search(stripped) + text: Final = fenced.group(1).strip() if fenced is not None else stripped try: - parsed = json.loads(text) - except json.JSONDecodeError: + return _VERDICT_ADAPTER.validate_json(text) + except ValueError: 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 + return _VERDICT_ADAPTER.validate_json(text[start : end + 1]) -def _extract_text_from_content(content: Any) -> str: +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): - parts: Final = [] - for part in content: - if isinstance(part, dict) and part.get("type") == "text": - parts.append(part.get("text", "")) - return " ".join(parts) + if isinstance(content, Sequence): + return " ".join( + str(part.get("text", "")) for part in content if isinstance(part, Mapping) and part.get("type") == "text" + ) return "" @@ -113,6 +123,14 @@ def _sample_hits(request_id: str, job_id: str, percentage: float) -> bool: 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.""" + 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() @@ -123,33 +141,59 @@ def _unmask_preference(raw_preference: str, real_is_a: bool) -> str: return "tie" +@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 + + +_JobCache: TypeAlias = "dict[str, tuple[float, ActiveShadowEvalJob | None]]" # mutable-ok: TTL cache + + class ShadowEvalLogger(CustomLogger): """Fires blind pairwise shadow evaluations for keys with an active shadow-eval job.""" def __init__( self, - router_provider: Optional[Callable[[], "Router | None"]] = None, - prisma_provider: Optional[Callable[[], "PrismaClient | None"]] = None, - ): + 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 # api_key_hash -> (fetched_at_monotonic, job_record_or_None) - self._job_cache: dict[str, tuple[float, dict[str, Any] | None]] = {} + self._job_cache: _JobCache = {} # mutable-ok: a TTL cache is mutable state by definition self._inflight_shadow_tasks: int = 0 - self._pending_seen: dict[str, int] = {} + self._pending_seen: dict[str, int] = {} # mutable-ok: flush buffer self._last_seen_flush: float = 0.0 #### hook #### - async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + async def async_log_success_event( + self, + kwargs: Mapping[str, object], + response_obj: object, + start_time: object, + end_time: object, + ) -> None: try: - payload: Final[Optional["StandardLoggingPayload"]] = kwargs.get("standard_logging_object") + 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 {} - request_metadata: Final = (kwargs.get("litellm_params") or {}).get("metadata") or {} + 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(SHADOW_EVAL_INTERNAL_MARKER): return # our own shadow/judge traffic api_key_hash: Final = metadata.get("user_api_key_hash") @@ -163,26 +207,31 @@ class ShadowEvalLogger(CustomLogger): return # The job tracks every request it saw, sampled or not, so the UI can # show "N of M requests shadowed". - self._pending_seen[job["id"]] = self._pending_seen.get(job["id"], 0) + 1 + self._pending_seen[job.id] = self._pending_seen.get(job.id, 0) + 1 now: Final = asyncio.get_event_loop().time() if now - self._last_seen_flush >= _SEEN_FLUSH_INTERVAL_SECONDS: self._last_seen_flush = now asyncio.create_task(self._flush_seen_counts()) - if not _sample_hits(request_id, job["id"], float(job["shadow_percentage"])): + if not _sample_hits(request_id, job.id, job.shadow_percentage): return if payload.get("call_type") not in (None, "completion", "acompletion", "chat_completion"): return # only chat-shaped traffic is comparable if self._inflight_shadow_tasks >= _MAX_CONCURRENT_SHADOW_TASKS: return + raw_messages: Final = kwargs.get("messages") self._inflight_shadow_tasks += 1 - task = asyncio.create_task( + task: Final = asyncio.create_task( self._run_shadow_eval( - job=dict(job), + job=job, request_id=request_id, - messages=list(kwargs.get("messages") or []), + 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=dict(payload.get("model_parameters") or {}), + model_parameters=MappingProxyType( + dict(payload.get("model_parameters") or {}) # mutable-ok: frozen snapshot + ), ) ) task.add_done_callback(lambda _: setattr(self, "_inflight_shadow_tasks", self._inflight_shadow_tasks - 1)) @@ -191,7 +240,7 @@ class ShadowEvalLogger(CustomLogger): #### job lookup #### - async def _get_active_job(self, api_key_hash: str) -> dict[str, Any] | None: + async def _get_active_job(self, api_key_hash: str) -> ActiveShadowEvalJob | None: cached: Final = self._job_cache.get(api_key_hash) now: Final = asyncio.get_event_loop().time() if cached is not None and now - cached[0] < _JOB_CACHE_TTL_SECONDS: @@ -199,20 +248,25 @@ class ShadowEvalLogger(CustomLogger): prisma: Final = self._prisma_provider() if prisma is None: return None - job: dict[str, Any] | None = None try: record: Final = await prisma.db.litellm_shadowevaljob.find_first( - where={"api_key_id": api_key_hash, "status": {"in": ["pending", "running"]}}, - order={"created_at": "desc"}, + where={ # mutable-ok: Prisma filter + "api_key_id": api_key_hash, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, + order={"created_at": "desc"}, # mutable-ok: Prisma order + ) + job: Final = ( + 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), + ) + if record is not None + else None ) - if record is not None: - job = { - "id": record.id, - "router_name": record.router_name, - "shadow_percentage": record.shadow_percentage, - "judge_model": record.judge_model, - "status": record.status, - } except Exception as e: # noqa: BLE001 # a DB blip must not break request logging verbose_logger.debug("shadow_eval: job lookup failed: %s", e) return cached[1] if cached is not None else None @@ -224,12 +278,12 @@ class ShadowEvalLogger(CustomLogger): if prisma is None: return pending: Final = self._pending_seen - 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( - where={"id": job_id}, - data={"request_count": {"increment": count}}, + where={"id": job_id}, # 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) @@ -238,12 +292,12 @@ class ShadowEvalLogger(CustomLogger): async def _run_shadow_eval( self, - job: dict[str, Any], + job: ActiveShadowEvalJob, request_id: str, - messages: list[dict[str, Any]], - response_obj: Any, + messages: Sequence[Mapping[str, object]], + response_obj: object, real_model: str, - model_parameters: dict[str, Any], + model_parameters: Mapping[str, object], ) -> None: """Detached background task: shadow call -> blind judge -> verdict row.""" prisma: Final = self._prisma_provider() @@ -252,28 +306,28 @@ class ShadowEvalLogger(CustomLogger): if not real_text or not messages: return - shadow = await self._call_router_shadow(job["router_name"], messages, model_parameters) + shadow: Final = await self._call_router_shadow(job.router_name, messages, model_parameters) if shadow is None: - await self._bump_failed(job["id"]) + await self._bump_failed(job.id) return shadow_text, shadow_model, tier, shadow_tokens = shadow - verdict = await self._call_judge( - judge_model=job["judge_model"], + verdict: Final = await self._call_judge( + judge_model=job.judge_model, messages=messages, real_text=real_text, shadow_text=shadow_text, ) if verdict is None: - await self._bump_failed(job["id"]) + await self._bump_failed(job.id) return preference, confidence, reasoning, judge_cost = verdict if prisma is None: return await prisma.db.litellm_shadowevalverdict.create( - data={ - "job_id": job["id"], + data={ # mutable-ok: Prisma payload + "job_id": job.id, "request_id": request_id, "tier_classification": tier, "real_model": real_model, @@ -282,20 +336,23 @@ class ShadowEvalLogger(CustomLogger): "judge_preference": preference, "judge_confidence": confidence, "judge_reasoning": reasoning[:1000] if reasoning else None, - "judge_model": job["judge_model"], + "judge_model": job.judge_model, } ) await prisma.db.litellm_shadowevaljob.update_many( - where={"id": job["id"], "status": {"in": ["pending", "running"]}}, - data={ - "completed_count": {"increment": 1}, - "cost_actual": {"increment": judge_cost}, + 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": judge_cost}, # mutable-ok: Prisma operator "status": "running", }, ) 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"]) + await self._bump_failed(job.id) async def _bump_failed(self, job_id: str) -> None: prisma: Final = self._prisma_provider() @@ -303,14 +360,14 @@ class ShadowEvalLogger(CustomLogger): return try: await prisma.db.litellm_shadowevaljob.update( - where={"id": job_id}, - data={"failed_count": {"increment": 1}}, + where={"id": job_id}, # mutable-ok: Prisma filter + data={"failed_count": {"increment": 1}}, # mutable-ok: Prisma payload ) 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: list[dict[str, Any]], model_parameters: dict[str, Any] + self, router_name: str, messages: Sequence[Mapping[str, object]], model_parameters: Mapping[str, object] ) -> tuple[str, str, str | None, int | None] | None: """Send the prompt through the auto-router; return (text, model, tier, completion_tokens).""" router: Final = self._router_provider() @@ -319,10 +376,14 @@ class ShadowEvalLogger(CustomLogger): return None # The router's pre-routing hook writes its routing decision into this # metadata dict; read it back after the call for tier attribution. - shadow_metadata: dict[str, Any] = {SHADOW_EVAL_INTERNAL_MARKER: True} - shadow_params: Final = {k: v for k, v in model_parameters.items() if k not in ("stream", "metadata")} + shadow_metadata: Final[dict[str, object]] = { # mutable-ok: router writes back + SHADOW_EVAL_INTERNAL_MARKER: True + } + 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 = await router.acompletion( + response: Final = await router.acompletion( model=router_name, messages=messages, metadata=shadow_metadata, @@ -334,17 +395,20 @@ class ShadowEvalLogger(CustomLogger): text: Final = self._extract_response_text(response) if not text: return None - routing_decision: Final = shadow_metadata.get("routing_decision") or {} - tier: Final = routing_decision.get("tier_label") or routing_decision.get("tier") - model: Final = getattr(response, "model", None) or routing_decision.get("routed_model") or "" + 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") + tier: Final = str(raw_tier) if raw_tier is not None else None + model: Final = str(getattr(response, "model", None) or routing_decision.get("routed_model") or "") usage: Final = getattr(response, "usage", None) - completion_tokens: Final = getattr(usage, "completion_tokens", None) if usage is not None else None + raw_tokens: Final = getattr(usage, "completion_tokens", None) if usage is not None else None + completion_tokens: Final = int(raw_tokens) if isinstance(raw_tokens, int) else None return text, model, tier, completion_tokens async def _call_judge( self, judge_model: str, - messages: list[dict[str, Any]], + messages: Sequence[Mapping[str, object]], real_text: str, shadow_text: str, ) -> tuple[str, float, str, float] | None: @@ -354,7 +418,7 @@ class ShadowEvalLogger(CustomLogger): response_b: Final = shadow_text if real_is_a else real_text conversation: Final = "\n".join( - f"{m.get('role', 'user').upper()}: {_extract_text_from_content(m.get('content'))}" + 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 ) @@ -365,15 +429,15 @@ class ShadowEvalLogger(CustomLogger): "Which response is better?" ) try: - response = await litellm.acompletion( + response: Final = await litellm.acompletion( model=judge_model, - messages=[ - {"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT}, - {"role": "user", "content": user_prompt}, + messages=[ # mutable-ok: SDK takes a list + {"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT}, # mutable-ok: SDK message + {"role": "user", "content": user_prompt}, # mutable-ok: SDK message ], temperature=0, max_tokens=200, - metadata={SHADOW_EVAL_INTERNAL_MARKER: True}, + metadata={SHADOW_EVAL_INTERNAL_MARKER: True}, # mutable-ok: SDK 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) @@ -384,29 +448,22 @@ class ShadowEvalLogger(CustomLogger): except (KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as e: verbose_logger.debug("shadow_eval: unparseable judge verdict: %s", e) return None - preference: Final = _unmask_preference(str(verdict.get("preference", "tie")), real_is_a) - try: - confidence = max(0.0, min(1.0, float(verdict.get("confidence", 0.5)))) - except (TypeError, ValueError): - confidence = 0.5 - reasoning: Final = str(verdict.get("reasoning", "")) - try: - cost = litellm.completion_cost(completion_response=response) or 0.0 - except Exception: # noqa: BLE001 # unmapped judge model: verdict still counts, cost stays 0 - cost = 0.0 - return preference, confidence, reasoning, cost + preference: Final = _unmask_preference(verdict.preference, real_is_a) + confidence: Final = max(0.0, min(1.0, verdict.confidence)) + return preference, confidence, verdict.reasoning, _judge_call_cost(response) @staticmethod - def _extract_response_text(response_obj: Any) -> str: + def _extract_response_text(response_obj: object) -> str: """Extract the assistant's text from a ModelResponse-shaped object or dict.""" try: - if isinstance(response_obj, dict): - content = response_obj["choices"][0]["message"]["content"] - else: - content = response_obj.choices[0].message.content + 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) if not isinstance(content, str) else content + return _extract_text_from_content(content) def _default_router_provider() -> "Router | None": diff --git a/litellm/proxy/management_endpoints/auto_router_endpoints.py b/litellm/proxy/management_endpoints/auto_router_endpoints.py index 761da2f6412..799b752c4a2 100644 --- a/litellm/proxy/management_endpoints/auto_router_endpoints.py +++ b/litellm/proxy/management_endpoints/auto_router_endpoints.py @@ -9,7 +9,7 @@ from datetime import datetime, timedelta, timezone from types import MappingProxyType from typing import TYPE_CHECKING, Annotated, Final -from pydantic import BaseModel, TypeAdapter +from pydantic import BaseModel, ConfigDict, TypeAdapter from litellm._logging import verbose_proxy_logger from litellm.exceptions import BudgetExceededError @@ -40,6 +40,7 @@ from litellm.types.management_endpoints.auto_router_endpoints import ( GetShadowEvalJobResponse, RequestComplexityRouterConfig, ShadowEvalResult, + ShadowEvalStatus, ShadowEvalTierResult, StartShadowEvalRequest, StartShadowEvalResponse, @@ -91,7 +92,7 @@ async def _authorize_routing_test(user_api_key_dict: UserAPIKeyAuth, team_id: st ) team_row: Final = await TeamRepository(prisma_client).table.find_unique( - where={"team_id": team_id}, # mutable-ok: Prisma query filters are dict-shaped + where={"team_id": team_id}, # mutable-ok: Prisma filter ) if team_row is None: raise HTTPException( @@ -236,7 +237,7 @@ async def preview_auto_router_routing( model=data.router_name, request_kwargs=request_kwargs, messages=[ # mutable-ok: the routing hook's signature takes a list of message dicts - {"role": "user", "content": data.prompt}, # mutable-ok: a message is dict-shaped + {"role": "user", "content": data.prompt}, # mutable-ok: SDK message ], ) except Exception as e: # noqa: BLE001 -- surfaces any classifier/plugin failure to the caller as a 400 instead of a 500, since the config under test is caller input @@ -531,8 +532,8 @@ def _estimate_judge_cost_per_call(judge_model: str) -> float: estimated: Final = prompt_cost + completion_cost if estimated > 0: return estimated - except Exception: # noqa: BLE001 # unknown judge model: fall back to a flat per-call figure - pass + 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", judge_model, e) return _FALLBACK_JUDGE_COST_PER_CALL @@ -585,20 +586,42 @@ def _shadow_eval_results(rows: Sequence[_VerdictAggRow]) -> ShadowEvalResult | N ) +class _ShadowEvalJobRow(BaseModel): + """The prisma job record, validated into concrete types at the response boundary.""" + + model_config = ConfigDict(from_attributes=True) + + id: str + status: ShadowEvalStatus + router_name: str + shadow_percentage: float + request_count: int + completed_count: int + failed_count: int + cost_estimate: float | None = None + cost_actual: float = 0.0 + created_at: datetime + completed_at: datetime | None = None + + +_SHADOW_EVAL_JOB_ROW: Final = TypeAdapter(_ShadowEvalJobRow) + + def _job_to_response(record: object, results: ShadowEvalResult | None) -> GetShadowEvalJobResponse: + row: Final = _SHADOW_EVAL_JOB_ROW.validate_python(record, from_attributes=True) return GetShadowEvalJobResponse( - job_id=record.id, # type: ignore[attr-defined] - status=record.status, # type: ignore[attr-defined] - router_name=record.router_name, # type: ignore[attr-defined] - shadow_percentage=record.shadow_percentage, # type: ignore[attr-defined] - request_count=record.request_count, # type: ignore[attr-defined] - completed_count=record.completed_count, # type: ignore[attr-defined] - failed_count=record.failed_count, # type: ignore[attr-defined] + job_id=row.id, + status=row.status, + router_name=row.router_name, + shadow_percentage=row.shadow_percentage, + request_count=row.request_count, + completed_count=row.completed_count, + failed_count=row.failed_count, results=results, - cost_estimate=record.cost_estimate, # type: ignore[attr-defined] - cost_actual=record.cost_actual, # type: ignore[attr-defined] - created_at=record.created_at.isoformat(), # type: ignore[attr-defined] - completed_at=record.completed_at.isoformat() if record.completed_at else None, # type: ignore[attr-defined] + cost_estimate=row.cost_estimate, + cost_actual=row.cost_actual, + created_at=row.created_at.isoformat(), + completed_at=row.completed_at.isoformat() if row.completed_at else None, ) @@ -635,7 +658,10 @@ async def start_shadow_eval( ) existing: Final = await prisma_client.db.litellm_shadowevaljob.find_first( - where={"api_key_id": data.api_key_id, "status": {"in": ["pending", "running"]}}, + 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( @@ -645,14 +671,17 @@ async def start_shadow_eval( lookback_start: Final = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=_ESTIMATE_LOOKBACK_DAYS) recent_requests: Final = await prisma_client.db.litellm_spendlogs.count( - where={"api_key": data.api_key_id, "startTime": {"gte": lookback_start}}, + where={ # mutable-ok: Prisma filter + "api_key": data.api_key_id, + "startTime": {"gte": lookback_start}, # mutable-ok: Prisma filter + }, ) weekly_sampled: Final = int(recent_requests * data.shadow_percentage / 100.0) per_call: Final = _estimate_judge_cost_per_call(data.judge_model) estimated_cost: Final = round(weekly_sampled * per_call, 2) job: Final = await prisma_client.db.litellm_shadowevaljob.create( - data={ + data={ # mutable-ok: Prisma payload "api_key_id": data.api_key_id, "router_name": data.router_name, "shadow_percentage": data.shadow_percentage, @@ -688,11 +717,13 @@ async def list_shadow_eval_jobs( 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 {} + 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"}, take=50 + where=where, + order={"created_at": "desc"}, # mutable-ok: Prisma order + take=50, # mutable-ok: Prisma order ) - return [_job_to_response(record, results=None) for record in records or ()] + return [_job_to_response(record, results=None) for record in records or ()] # mutable-ok: FastAPI response_model @router.get( @@ -712,7 +743,9 @@ async def get_shadow_eval_job( 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}) + 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}") @@ -738,15 +771,20 @@ async def stop_shadow_eval_job( 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}) + 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}, - data={"status": "completed", "completed_at": datetime.now(timezone.utc)}, + where={"id": job_id}, # mutable-ok: Prisma filter + data={ # mutable-ok: Prisma payload + "status": "completed", + "completed_at": datetime.now(timezone.utc), + }, ) raw_rows: Final = await prisma_client.db.query_raw(_VERDICT_AGG_SQL, job_id) rows: Final = _VERDICT_AGG_ROWS.validate_python(raw_rows or ()) diff --git a/litellm/types/management_endpoints/auto_router_endpoints.py b/litellm/types/management_endpoints/auto_router_endpoints.py index 0646ea3b3b2..313a97a88e0 100644 --- a/litellm/types/management_endpoints/auto_router_endpoints.py +++ b/litellm/types/management_endpoints/auto_router_endpoints.py @@ -3,7 +3,7 @@ Types for auto-router management endpoints """ from collections.abc import Mapping -from typing import Final, Literal +from typing import Final, Literal, TypeAlias from pydantic import BaseModel, Field, field_validator @@ -149,8 +149,8 @@ class AutoRouterBenchmarksResponse(BaseModel): # router would have fared without ever serving its answer to a real user. # --------------------------------------------------------------------------- -ShadowEvalStatus = Literal["pending", "running", "completed", "failed"] -JudgePreference = Literal["real", "shadow", "tie"] +ShadowEvalStatus: TypeAlias = Literal["pending", "running", "completed", "failed"] +JudgePreference: TypeAlias = Literal["real", "shadow", "tie"] DEFAULT_SHADOW_EVAL_JUDGE_MODEL: Final[str] = "anthropic/claude-sonnet-5" diff --git a/tests/test_litellm/integrations/test_shadow_eval_logger.py b/tests/test_litellm/integrations/test_shadow_eval_logger.py index 845f696f9e7..dd542d1b528 100644 --- a/tests/test_litellm/integrations/test_shadow_eval_logger.py +++ b/tests/test_litellm/integrations/test_shadow_eval_logger.py @@ -8,6 +8,7 @@ import pytest from litellm.integrations.shadow_eval_logger import ( SHADOW_EVAL_INTERNAL_MARKER, + ActiveShadowEvalJob, ShadowEvalLogger, _parse_pairwise_verdict, _sample_hits, @@ -60,16 +61,16 @@ class TestUnmaskPreference: class TestParsePairwiseVerdict: def test_plain_json(self): v = _parse_pairwise_verdict('{"preference": "A", "confidence": 0.9, "reasoning": "clearer"}') - assert v["preference"] == "A" - assert v["confidence"] == 0.9 + assert v.preference == "A" + assert v.confidence == 0.9 def test_fenced_json(self): raw = 'Here is my verdict:\n```json\n{"preference": "B", "confidence": 0.7, "reasoning": "x"}\n```\nDone.' - assert _parse_pairwise_verdict(raw)["preference"] == "B" + assert _parse_pairwise_verdict(raw).preference == "B" def test_json_with_surrounding_prose(self): raw = 'Verdict: {"preference": "tie", "confidence": 0.5, "reasoning": "same"} — final.' - assert _parse_pairwise_verdict(raw)["preference"] == "tie" + assert _parse_pairwise_verdict(raw).preference == "tie" def test_non_object_raises(self): with pytest.raises(ValueError): @@ -99,10 +100,14 @@ class TestSuccessHookSkipPaths: prisma.db.litellm_shadowevaljob.find_first.assert_not_called() async def test_skips_own_internal_traffic(self): - job = {"id": "j1", "router_name": "r", "shadow_percentage": 100.0, "judge_model": "m", "status": "running"} + job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") logger, prisma, _ = _logger_with_mocks(job) kwargs = { - "standard_logging_object": {"id": "req-1", "model": "gpt-4o", "metadata": {"user_api_key_hash": "key-hash"}}, + "standard_logging_object": { + "id": "req-1", + "model": "gpt-4o", + "metadata": {"user_api_key_hash": "key-hash"}, + }, "litellm_params": {"metadata": {SHADOW_EVAL_INTERNAL_MARKER: True}}, "messages": [{"role": "user", "content": "hi"}], } @@ -113,7 +118,11 @@ class TestSuccessHookSkipPaths: logger, prisma, _ = _logger_with_mocks() prisma.db.litellm_shadowevaljob.find_first = AsyncMock(return_value=None) kwargs = { - "standard_logging_object": {"id": "req-1", "model": "gpt-4o", "metadata": {"user_api_key_hash": "key-hash"}}, + "standard_logging_object": { + "id": "req-1", + "model": "gpt-4o", + "metadata": {"user_api_key_hash": "key-hash"}, + }, "litellm_params": {"metadata": {}}, "messages": [{"role": "user", "content": "hi"}], } @@ -121,7 +130,7 @@ class TestSuccessHookSkipPaths: assert logger._pending_seen == {} async def test_counts_seen_for_active_job(self): - job = {"id": "j1", "router_name": "r", "shadow_percentage": 0.0, "judge_model": "m", "status": "running"} + job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=0.0, judge_model="m", status="running") logger, _, _ = _logger_with_mocks(job) kwargs = { "standard_logging_object": { @@ -177,7 +186,7 @@ class TestCallRouterShadowForwardsParameters: @pytest.mark.asyncio class TestInflightTaskBacklog: async def test_drops_sample_when_at_capacity(self): - job = {"id": "j1", "router_name": "r", "shadow_percentage": 100.0, "judge_model": "m", "status": "running"} + job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") logger, _, _ = _logger_with_mocks(job) logger._inflight_shadow_tasks = 999999 # simulate saturation regardless of the real cap @@ -197,7 +206,7 @@ class TestInflightTaskBacklog: assert logger._inflight_shadow_tasks == before async def test_schedules_and_decrements_when_under_capacity(self): - job = {"id": "j1", "router_name": "r", "shadow_percentage": 100.0, "judge_model": "m", "status": "running"} + job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") logger, _, router = _logger_with_mocks(job) router.acompletion = AsyncMock(side_effect=asyncio.sleep(0)) # keep the task alive briefly @@ -243,7 +252,7 @@ class TestStoppedJobCannotBeReactivated: logger._call_router_shadow = AsyncMock(return_value=("shadow text", "shadow-model", "SIMPLE", 10)) logger._call_judge = AsyncMock(return_value=("real", 0.9, "clearer", 0.01)) - job = {"id": "j1", "router_name": "r", "judge_model": "m", "shadow_percentage": 100.0, "status": "running"} + job = ActiveShadowEvalJob(id="j1", router_name="r", judge_model="m", shadow_percentage=100.0, status="running") await logger._run_shadow_eval( job=job, request_id="req-1", @@ -279,7 +288,7 @@ class TestVerdictWriteAccumulatesCost: logger._call_router_shadow = AsyncMock(return_value=("shadow text", "shadow-model", "SIMPLE", 10)) logger._call_judge = AsyncMock(return_value=("real", 0.9, "clearer", 0.05)) - job = {"id": "j1", "router_name": "r", "judge_model": "m", "shadow_percentage": 100.0, "status": "running"} + job = ActiveShadowEvalJob(id="j1", router_name="r", judge_model="m", shadow_percentage=100.0, status="running") await logger._run_shadow_eval( job=job, request_id="req-1",