mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-20 00:11:50 +00:00
refactor: satisfy strict lint and type-discipline gates in shadow eval
The lint job's strict-rule budget flagged 17 new violations. Rather than raise the ceiling, this types the code properly: - validate the judge verdict into a PairwiseVerdict pydantic model at the parse boundary instead of dict[str, Any] + cast, which also removes the defensive float()/str() coercion downstream - replace the untyped job dict with a frozen ActiveShadowEvalJob dataclass - validate the prisma job row into _ShadowEvalJobRow, replacing 11 no-op '# type: ignore[attr-defined]' comments - annotate the success hook and drop Any from the remaining signatures - mark the genuine third-party dict shapes (prisma filters/payloads, SDK message lists) with '# mutable-ok' reasons per the existing convention
This commit is contained in:
Abhimanyu Kapur
parent
1da3bdc5cc
commit
ba4b52162b
4 changed files with 247 additions and 143 deletions
|
|
@ -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":
|
||||
|
|
|
|||
|
|
@ -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 ())
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue