Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/auto_router_endpoints.py: 54%
565 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2AUTO ROUTER MANAGEMENT ENDPOINTS
4POST /auto_router/test_routing - Route one request through an unsaved complexity-router config
5POST /auto_router/validate_complexity_router_config - Dry-run the complexity-router write gate without saving
6"""
8from collections.abc import Mapping, Sequence
9from datetime import datetime, timedelta, timezone
10from itertools import chain, groupby
11from types import MappingProxyType
12from typing import TYPE_CHECKING, Annotated, Final, Protocol
13from uuid import uuid4
15from pydantic import BaseModel, ConfigDict, TypeAdapter, field_validator
17import litellm
18from litellm._logging import verbose_proxy_logger
19from litellm.exceptions import BudgetExceededError
20from litellm.litellm_core_utils.llm_judge import judge_target
21from litellm.proxy._types import (
22 CommonProxyErrors,
23 LiteLLM_TeamTable,
24 LitellmUserRoles,
25 ProxyErrorTypes,
26 ProxyException,
27 UserAPIKeyAuth,
28)
29from litellm.proxy.auth.auth_checks import (
30 _virtual_key_max_budget_check,
31 can_key_call_resolved_model,
32)
33from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
34from litellm.proxy.db.autorouter_session_rollup import (
35 AUTOROUTER_BENCHMARKS_SQL,
36 bounded_session_id,
37)
38from litellm.proxy.litellm_pre_call_utils import (
39 LiteLLMProxyRequestSetup,
40 refresh_proxy_server_request_body_snapshot,
41)
42from litellm.proxy.management_endpoints.common_utils import (
43 _is_user_team_admin, # pyright: ignore[reportPrivateUsage] # shared owner of team-admin membership
44)
45from litellm.proxy.management_helpers.auto_router_permissions import (
46 authorize_member_auto_router_dependencies,
47 authorize_member_auto_router_team,
48 validate_member_auto_router_config,
49)
50from litellm.repositories.autorouter_session_repository import AutoRouterSessionRepository
51from litellm.repositories.base_repository import SupportsModelDump
52from litellm.repositories.team_repository import TeamRepository
53from litellm.router_strategy.complexity_router import ComplexityRouter
54from litellm.router_utils.auto_router_model_naming import (
55 StrategyRouterDependencyRole,
56 classify_strategy_router_model,
57 strategy_router_dependencies,
58)
59from litellm.types.management_endpoints.auto_router_endpoints import (
60 SHADOW_EVAL_TURN_VALVE,
61 AutoRouterAvailabilityRequest,
62 AutoRouterAvailabilityResponse,
63 AutoRouterBenchmarkGroup,
64 AutoRouterBenchmarksResponse,
65 AutoRouterBenchmarkTotals,
66 AutoRouterCacheBucket,
67 AutoRouterCacheStats,
68 AutoRouterRoutingTestRequest,
69 AutoRouterRoutingTestResponse,
70 AutoRouterSessionResponse,
71 ComplexityRouterConfigValidationRequest,
72 ComplexityRouterConfigValidationResponse,
73 RequestComplexityRouterConfig,
74 ShadowEvalDirection,
75 ShadowEvalJobResponse,
76 ShadowEvalJobTargetResponse,
77 ShadowEvalResult,
78 ShadowEvalSlice,
79 ShadowEvalTargetType,
80 StartShadowEvalRequest,
81)
83if TYPE_CHECKING: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true
84 from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
86 from litellm.proxy.utils import PrismaClient
87 from litellm.router import Router
88else:
89 try:
90 from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
91 except ImportError:
92 # fastapi is only required for proxy, not for SDK usage
93 pass
95router: Final = APIRouter()
98class _TeamTable(Protocol):
99 async def find_unique(self, *, where: Mapping[str, object]) -> SupportsModelDump | None: ... 99 ↛ exitline 99 didn't return from function 'find_unique' because
102class _VerificationTokenRow(Protocol):
103 @property
104 def token(self) -> str: ... 104 ↛ exitline 104 didn't return from function 'token' because
106 @property
107 def key_alias(self) -> str | None: ... 107 ↛ exitline 107 didn't return from function 'key_alias' because
109 @property
110 def key_name(self) -> str | None: ... 110 ↛ exitline 110 didn't return from function 'key_name' because
112 @property
113 def team_id(self) -> str | None: ... 113 ↛ exitline 113 didn't return from function 'team_id' because
116class _VerificationTokenTable(Protocol):
117 async def find_unique(self, *, where: Mapping[str, object]) -> _VerificationTokenRow | None: ... 117 ↛ exitline 117 didn't return from function 'find_unique' because
119 async def find_many(self, *, where: Mapping[str, object]) -> Sequence[_VerificationTokenRow]: ... 119 ↛ exitline 119 didn't return from function 'find_many' because
122class _TeamRow(Protocol):
123 @property
124 def team_id(self) -> str: ... 124 ↛ exitline 124 didn't return from function 'team_id' because
126 @property
127 def team_alias(self) -> str | None: ... 127 ↛ exitline 127 didn't return from function 'team_alias' because
130class _TeamRowsTable(Protocol):
131 async def find_many(self, *, where: Mapping[str, object]) -> Sequence[_TeamRow]: ... 131 ↛ exitline 131 didn't return from function 'find_many' because
134class _UserRow(Protocol):
135 @property
136 def user_id(self) -> str: ... 136 ↛ exitline 136 didn't return from function 'user_id' because
138 @property
139 def user_email(self) -> str | None: ... 139 ↛ exitline 139 didn't return from function 'user_email' because
142class _UserRowsTable(Protocol):
143 async def find_many(self, *, where: Mapping[str, object]) -> Sequence[_UserRow]: ... 143 ↛ exitline 143 didn't return from function 'find_many' because
146class _ShadowEvalJobRow(Protocol):
147 @property
148 def id(self) -> str: ... 148 ↛ exitline 148 didn't return from function 'id' because
150 @property
151 def group_id(self) -> str: ... 151 ↛ exitline 151 didn't return from function 'group_id' because
153 @property
154 def target_type(self) -> str: ... 154 ↛ exitline 154 didn't return from function 'target_type' because
156 @property
157 def target_id(self) -> str: ... 157 ↛ exitline 157 didn't return from function 'target_id' because
160class _ShadowEvalJobTable(Protocol):
161 async def find_many(self, *, where: Mapping[str, object]) -> Sequence[_ShadowEvalJobRow]: ... 161 ↛ exitline 161 didn't return from function 'find_many' because
163 async def create_many(self, data: Sequence[Mapping[str, object]]) -> int: ... 163 ↛ exitline 163 didn't return from function 'create_many' because
166class _ShadowEvalAttemptRow(Protocol):
167 @property
168 def error(self) -> str | None: ... 168 ↛ exitline 168 didn't return from function 'error' because
171class _ShadowEvalFunnelTable(Protocol):
172 async def create_many(self, data: Sequence[Mapping[str, object]], skip_duplicates: bool) -> int: ... 172 ↛ exitline 172 didn't return from function 'create_many' because
175class _ShadowEvalAttemptTable(Protocol):
176 async def find_first( 176 ↛ exitline 176 didn't return from function 'find_first' because
177 self, *, where: Mapping[str, object], order: Mapping[str, str]
178 ) -> _ShadowEvalAttemptRow | None: ...
181def _team_table(prisma_client: "PrismaClient") -> _TeamTable:
182 return TeamRepository(prisma_client).table
185def _verification_tokens(prisma_client: "PrismaClient") -> _VerificationTokenTable:
186 return prisma_client.db.litellm_verificationtoken
189def _team_rows(prisma_client: "PrismaClient") -> _TeamRowsTable:
190 return prisma_client.db.litellm_teamtable
193def _user_rows(prisma_client: "PrismaClient") -> _UserRowsTable:
194 return prisma_client.db.litellm_usertable
197def _shadow_eval_jobs(prisma_client: "PrismaClient") -> _ShadowEvalJobTable:
198 return prisma_client.db.litellm_shadowevaljob
201def _shadow_eval_funnel(prisma_client: "PrismaClient") -> _ShadowEvalFunnelTable:
202 return prisma_client.db.litellm_shadowevalfunnel # pyright: ignore[reportAttributeAccessIssue] # generated client
205def _shadow_eval_attempts(prisma_client: "PrismaClient") -> _ShadowEvalAttemptTable:
206 return prisma_client.db.litellm_shadowevalattempt
209async def _query_raw(prisma_client: "PrismaClient", query: str, *args: object) -> Sequence[Mapping[str, object]]:
210 return await prisma_client.db.query_raw(query, *args)
213async def _authorize_router_dry_run(user_api_key_dict: UserAPIKeyAuth, team_id: str | None) -> LiteLLM_TeamTable | None:
214 from litellm.proxy.management_endpoints.model_management_endpoints import (
215 ModelManagementAuthChecks,
216 )
217 from litellm.proxy.proxy_server import premium_user, prisma_client
219 if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN: 219 ↛ 222line 219 didn't jump to line 222 because the condition on line 219 was always true
220 return None
222 if team_id is None:
223 raise HTTPException(
224 status_code=403,
225 detail={ # mutable-ok: HTTPException detail must be a plain mapping to keep this route's {"error": ...} response shape
226 "error": f"User does not have permission to dry-run an auto router. Your role={user_api_key_dict.user_role}. Call as a PROXY_ADMIN, or as a team admin by specifying a team_id."
227 },
228 )
230 if prisma_client is None:
231 raise HTTPException(
232 status_code=500,
233 detail={ # mutable-ok: HTTPException detail must be a plain mapping
234 "error": CommonProxyErrors.db_not_connected_error.value
235 },
236 )
238 team_row: Final = await _team_table(prisma_client).find_unique(
239 where={"team_id": team_id}, # mutable-ok: Prisma query filters are dict-shaped
240 )
241 if team_row is None:
242 raise HTTPException(
243 status_code=400,
244 detail={ # mutable-ok: HTTPException detail must be a plain mapping
245 "error": f"Team id={team_id} does not exist in db"
246 },
247 )
249 team: Final = LiteLLM_TeamTable.model_validate(team_row.model_dump())
250 if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team):
251 ModelManagementAuthChecks.can_user_make_team_model_call(
252 team_id=team_id,
253 user_api_key_dict=user_api_key_dict,
254 team_obj=team,
255 premium_user=premium_user,
256 )
257 return None
258 authorize_member_auto_router_team(
259 user_api_key_dict=user_api_key_dict,
260 team=team,
261 premium_user=premium_user,
262 )
263 return team
266async def _authorize_member_dry_run_config(
267 *,
268 config: Mapping[str, object],
269 default_model: str | None,
270 user_api_key_dict: UserAPIKeyAuth,
271 team: LiteLLM_TeamTable,
272) -> UserAPIKeyAuth:
273 from litellm.proxy.proxy_server import llm_router, prisma_client
275 if prisma_client is None or llm_router is None:
276 raise HTTPException(status_code=503, detail="Cannot verify auto-router model access")
277 validated: Final = validate_member_auto_router_config(config)
278 scoped_actor: Final = user_api_key_dict.model_copy(
279 update=MappingProxyType({"team_id": team.team_id, "team_models": team.models, "org_id": team.organization_id})
280 )
281 await authorize_member_auto_router_dependencies(
282 config=validated,
283 default_model=default_model,
284 user_api_key_dict=scoped_actor,
285 team=team,
286 prisma_client=prisma_client,
287 llm_router=llm_router,
288 )
289 return scoped_actor
292def _models_this_test_can_call(config: RequestComplexityRouterConfig) -> tuple[str, ...]:
293 """The models the routing test itself would send a request to, and so spend on.
295 Excludes every tier's models: the prompt is never sent to the model it routed to.
296 """
297 return tuple(
298 dependency.model_name
299 for dependency in strategy_router_dependencies(
300 MappingProxyType(
301 {
302 "model": "auto_router/complexity_router",
303 "complexity_router_config": config.model_dump(exclude_none=True),
304 }
305 )
306 )
307 if dependency.role in ("classifier", "embedding", "evaluation")
308 )
311async def _authorize_models_this_test_can_call(
312 config: RequestComplexityRouterConfig,
313 user_api_key_dict: UserAPIKeyAuth,
314 llm_router: "Router",
315) -> None:
316 """Hold a classifier or embedding call to the caller's model access and key budget.
318 Those calls go through the router rather than through /v1/chat/completions, so the model
319 checks a real request gets in user_api_key_auth would otherwise be skipped, letting a
320 caller spend on a model their key cannot call, and this route is not an LLM API route, so
321 the key's own budget is not checked either. Test Connection gets both for free by routing
322 its calls through the proxy. Team and member budgets are already enforced on every route.
323 """
324 models: Final = _models_this_test_can_call(config)
325 if not models and config.classifier_type != "jev":
326 return
328 from litellm.proxy.proxy_server import proxy_logging_obj
330 for model in models:
331 await can_key_call_resolved_model(
332 model=model,
333 llm_model_list=llm_router.model_list,
334 valid_token=user_api_key_dict,
335 llm_router=llm_router,
336 )
338 try:
339 await _virtual_key_max_budget_check(
340 valid_token=user_api_key_dict,
341 proxy_logging_obj=proxy_logging_obj,
342 )
343 except BudgetExceededError as e:
344 raise ProxyException(
345 message=e.message,
346 type=ProxyErrorTypes.budget_exceeded,
347 param=None,
348 code=status.HTTP_400_BAD_REQUEST,
349 ) from e
351 if config.classifier_type == "jev" and user_api_key_dict.budget_throttle_pct is not None:
352 raise ProxyException(
353 message="Budget has been exceeded! JEV Test Routing requires available budget.",
354 type=ProxyErrorTypes.budget_exceeded,
355 param=None,
356 code=status.HTTP_400_BAD_REQUEST,
357 )
360@router.post(
361 "/auto_router/validate_complexity_router_config",
362 tags=["model management"], # mutable-ok: fastapi's decorator signature types tags as a list
363 dependencies=[Depends(user_api_key_auth)], # mutable-ok: fastapi's decorator signature types dependencies as a list
364 response_model=ComplexityRouterConfigValidationResponse,
365 status_code=status.HTTP_200_OK,
366)
367async def validate_complexity_router_config(
368 data: ComplexityRouterConfigValidationRequest,
369 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
370) -> ComplexityRouterConfigValidationResponse:
371 """
372 Validate a complexity-router config without saving it.
374 Runs the same check every write path runs (the router's own pydantic model), so a form can
375 show the backend's exact verdict while the operator is still editing rather than after a
376 rejected save. Uses the same team opt-in and model-access checks as configuration
377 writes for members. Nothing is created, routed, or billed.
378 """
379 member_team: Final = await _authorize_router_dry_run(user_api_key_dict=user_api_key_dict, team_id=data.team_id)
381 from litellm.router_utils.auto_router_model_naming import (
382 validate_complexity_router_config_write,
383 )
385 error: Final = validate_complexity_router_config_write(data.complexity_router_config)
386 if error is None and member_team is not None: 386 ↛ 387line 386 didn't jump to line 387 because the condition on line 386 was never true
387 await _authorize_member_dry_run_config(
388 config=data.complexity_router_config,
389 default_model=None,
390 user_api_key_dict=user_api_key_dict,
391 team=member_team,
392 )
393 return ComplexityRouterConfigValidationResponse(valid=error is None, error=error)
396@router.post(
397 "/auto_router/availability",
398 tags=["model management"], # mutable-ok: FastAPI requires a list
399 response_model=AutoRouterAvailabilityResponse,
400)
401async def get_auto_router_availability(
402 data: AutoRouterAvailabilityRequest,
403 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
404) -> AutoRouterAvailabilityResponse:
405 from litellm.proxy.management_helpers.auto_router_availability import auto_router_availability
406 from litellm.proxy.proxy_server import (
407 _license_check, # pyright: ignore[reportPrivateUsage] # same entitlement owner as the model write gate
408 heuristic_v1_tuning_baselines,
409 llm_router,
410 proxy_config,
411 )
413 member_team: Final = await _authorize_router_dry_run(user_api_key_dict, data.team_id)
414 rows: Final = proxy_config.auto_router_db_catalog
415 if rows is None or llm_router is None: 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true
416 raise HTTPException(status_code=503, detail="Auto-router availability is unavailable")
417 saved: Final = next((row for row in rows if row.model_id == data.saved_model_id), None)
418 if data.saved_model_id is not None:
419 if saved is None: 419 ↛ 421line 419 didn't jump to line 421 because the condition on line 419 was always true
420 raise HTTPException(status_code=404, detail="Saved auto router is unavailable")
421 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN and (
422 saved.team_id != data.team_id or (member_team is not None and saved.created_by != user_api_key_dict.user_id)
423 ):
424 raise HTTPException(status_code=403, detail="Cannot check another user's auto router")
425 existing: Final = saved.deployment if saved is not None else None
426 others: Final = tuple(row.deployment for row in rows if row is not saved) + tuple(llm_router.config_deployments())
427 candidate: Final = MappingProxyType(
428 {
429 "litellm_params": MappingProxyType(
430 {"model": "auto_router/complexity_router", "complexity_router_config": data.complexity_router_config}
431 ),
432 "model_info": MappingProxyType({"id": data.saved_model_id or "availability-new-router", "db_model": True}),
433 }
434 )
435 return auto_router_availability(
436 others=others,
437 existing=existing,
438 candidate=candidate,
439 baselines=heuristic_v1_tuning_baselines,
440 limit=_license_check.auto_router_capability_limit(),
441 )
444async def _resolve_saved_routing_test(
445 data: AutoRouterRoutingTestRequest,
446 user_api_key_dict: UserAPIKeyAuth,
447 llm_router: "Router",
448) -> AutoRouterRoutingTestRequest:
449 if data.saved_model_id is None: 449 ↛ 450line 449 didn't jump to line 450 because the condition on line 449 was never true
450 return data
451 deployment: Final = llm_router.get_deployment(data.saved_model_id)
452 if deployment is None or deployment.model_info.blocked: 452 ↛ 454line 452 didn't jump to line 454 because the condition on line 452 was always true
453 raise HTTPException(status_code=404, detail="Saved auto router is unavailable")
454 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN and deployment.model_info.team_id != data.team_id:
455 raise HTTPException(status_code=403, detail="Saved auto router belongs to a different team")
456 await can_key_call_resolved_model(
457 model=deployment.model_info.team_public_model_name or deployment.model_name,
458 llm_model_list=llm_router.model_list,
459 valid_token=user_api_key_dict,
460 llm_router=llm_router,
461 )
462 params: Final = deployment.litellm_params
463 if classify_strategy_router_model(params.model or "") != "complexity" or params.complexity_router_config is None:
464 raise HTTPException(status_code=400, detail="Saved deployment is not a complexity auto router")
465 return data.model_copy(
466 update=MappingProxyType(
467 {
468 "complexity_router_config": RequestComplexityRouterConfig.model_validate(
469 params.complexity_router_config
470 ),
471 "default_model": params.complexity_router_default_model,
472 "router_name": deployment.model_name,
473 }
474 )
475 )
478@router.post(
479 "/auto_router/test_routing",
480 tags=["model management"], # mutable-ok: fastapi's decorator signature types tags as a list
481 dependencies=[Depends(user_api_key_auth)], # mutable-ok: fastapi's decorator signature types dependencies as a list
482 response_model=AutoRouterRoutingTestResponse,
483 status_code=status.HTTP_200_OK,
484)
485async def preview_auto_router_routing(
486 data: AutoRouterRoutingTestRequest,
487 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
488 http_request: Request,
489) -> AutoRouterRoutingTestResponse:
490 """
491 Route a single request through a complexity-router config and report where it landed.
493 Answers "which model would this request get?" for a config that only exists in a form,
494 so an auto router can be checked before it is created. The request is classified by the
495 same pre-routing hook a live request runs, over the same messages, system prompt and tool
496 definitions, then dropped: nothing is sent to the model it routed to, and no auto router is
497 created. A heuristic config therefore spends nothing, while an `llm` classifier or semantic
498 keyword matching bills its classifier/embedding call to the calling key, like Test Connection
499 does.
501 Send `messages` to classify a real turn, with `system` and `tools` beside it when the surface
502 carries them top level, as Anthropic /v1/messages does. `prompt` is the single-ask shorthand and
503 routes as one user turn with nothing around it.
505 **Example Request:**
506 ```json
507 {
508 "messages": [
509 {"role": "system", "content": "You are a database migration assistant"},
510 {"role": "user", "content": "the index is not unique"},
511 {"role": "assistant", "content": "Then two workers can both insert. Add a unique index"},
512 {"role": "user", "content": "ok do it"}
513 ],
514 "tools": [{"type": "function", "function": {"name": "Bash", "description": "Run a command"}}],
515 "complexity_router_config": {
516 "tiers": {"SIMPLE": ["gpt-4o-mini"], "REASONING": ["o3"]},
517 "classifier_type": "heuristic"
518 }
519 }
520 ```
521 """
522 from litellm.proxy.proxy_server import (
523 general_settings,
524 llm_router,
525 prisma_client,
526 proxy_logging_obj,
527 user_api_key_cache,
528 user_model,
529 )
530 from litellm.proxy.utils import get_available_models_for_user
532 member_team: Final = await _authorize_router_dry_run(user_api_key_dict=user_api_key_dict, team_id=data.team_id)
533 if llm_router is None: 533 ↛ 534line 533 didn't jump to line 534 because the condition on line 533 was never true
534 raise HTTPException(
535 status_code=500,
536 detail={ # mutable-ok: HTTPException detail must be a plain mapping
537 "error": CommonProxyErrors.no_llm_router.value
538 },
539 )
540 resolved: Final = await _resolve_saved_routing_test(data, user_api_key_dict, llm_router)
541 actor: Final = (
542 await _authorize_member_dry_run_config(
543 config=resolved.complexity_router_config.model_dump(exclude_none=True),
544 default_model=resolved.default_model,
545 user_api_key_dict=user_api_key_dict,
546 team=member_team,
547 )
548 if member_team is not None
549 else user_api_key_dict
550 )
551 request_data: Final[dict[str, object]] = { # mutable-ok: auth and routing enrich this request in place
552 **resolved.wire_body(),
553 "metadata": {}, # mutable-ok: centralized auth and identity stamping share this metadata bucket
554 "proxy_server_request": {"body": None}, # mutable-ok: the snapshot owner fills this body in place
555 }
557 if member_team is not None and _models_this_test_can_call(resolved.complexity_router_config):
558 from litellm.proxy.auth.user_api_key_auth import (
559 _run_centralized_common_checks, # pyright: ignore[reportPrivateUsage] # reuse the serving admission policy
560 )
562 await _run_centralized_common_checks(
563 user_api_key_auth_obj=actor,
564 request=http_request,
565 request_data=request_data,
566 route="/auto_router/test_routing",
567 )
569 await _authorize_models_this_test_can_call(
570 config=resolved.complexity_router_config,
571 user_api_key_dict=actor,
572 llm_router=llm_router,
573 )
575 complexity_router: Final = ComplexityRouter(
576 model_name=resolved.router_name,
577 litellm_router_instance=llm_router,
578 complexity_router_config=resolved.complexity_router_config.model_dump(exclude_none=True),
579 default_model=resolved.default_model,
580 derive_savings_baseline=False,
581 )
583 request_kwargs: Final = LiteLLMProxyRequestSetup.add_user_api_key_auth_to_request_metadata(
584 data=request_data,
585 user_api_key_dict=actor,
586 _metadata_variable_name="metadata",
587 )
588 refresh_proxy_server_request_body_snapshot(request_kwargs)
590 try:
591 hook_response: Final = await complexity_router.async_pre_routing_hook(
592 model=resolved.router_name,
593 request_kwargs=request_kwargs,
594 messages=request_kwargs["messages"],
595 )
596 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
597 verbose_proxy_logger.exception("Auto router routing test failed. Due to error - %s", e)
598 raise HTTPException(
599 status_code=400,
600 detail={ # mutable-ok: HTTPException detail must be a plain mapping
601 "error": f"Could not route this prompt: {e}"
602 },
603 ) from e
605 if hook_response is None or hook_response.routing_decision is None:
606 raise HTTPException(
607 status_code=400,
608 detail={ # mutable-ok: HTTPException detail must be a plain mapping
609 "error": "The router made no decision for this prompt. Check that at least one tier has a model."
610 },
611 )
613 available_models: Final = await get_available_models_for_user(
614 user_api_key_dict=user_api_key_dict,
615 llm_router=llm_router,
616 general_settings=general_settings,
617 user_model=user_model,
618 prisma_client=prisma_client,
619 proxy_logging_obj=proxy_logging_obj,
620 team_id=data.team_id,
621 user_api_key_cache=user_api_key_cache,
622 )
623 return AutoRouterRoutingTestResponse(
624 routed_model=hook_response.model,
625 routed_model_configured=hook_response.model in frozenset(available_models),
626 routing_decision=hook_response.routing_decision,
627 )
630class _SessionAggRow(BaseModel):
631 router_name: str
632 router_type: str
633 tier_turns: Mapping[str, int]
634 sessions: int
635 turns: int
636 unordered_turns: int
637 covered_turns: int
638 cache_hits: int
639 same_model_turns: int
640 same_model_hits: int
641 first_visit_turns: int
642 first_visit_hits: int
643 return_turns: int
644 return_hits: int
645 return_expired_misses: int
646 return_within_ttl_misses: int
647 ttl_5m_turns: int
648 ttl_1h_turns: int
649 total_tokens: int
650 spend: float
651 saved_spend: float
652 savings_estimated_turns: int = 0
653 savings_estimated_actual_spend: float = 0.0
654 savings_estimated_saved_spend: float = 0.0
655 classifier_cost: float
656 classifier_cost_recorded_turns: int
657 session_seconds: float
660_SESSION_AGG_ROWS: Final = TypeAdapter(list[_SessionAggRow])
663def _parse_benchmark_day(value: str) -> datetime:
664 try:
665 parsed: Final = datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=timezone.utc)
666 except ValueError:
667 raise HTTPException(status_code=400, detail=f"Invalid date format: {value}. Expected: 'YYYY-MM-DD'")
668 return parsed.replace(tzinfo=None)
671def _pct(numerator: float, denominator: float) -> float:
672 if denominator <= 0: 672 ↛ 674line 672 didn't jump to line 674 because the condition on line 672 was always true
673 return 0.0
674 return round(100.0 * numerator / denominator, 1)
677def _cache_bucket(turns: int, hits: int) -> AutoRouterCacheBucket:
678 return AutoRouterCacheBucket(turns=turns, hits=hits, hit_rate_pct=_pct(hits, turns))
681def _savings_cohort(
682 turns: int, estimated_turns: int, actual_spend: float, saved_spend: float
683) -> tuple[float | None, float | None]:
684 if turns > 0 and estimated_turns == 0: 684 ↛ 685line 684 didn't jump to line 685 because the condition on line 684 was never true
685 return None, None
686 return saved_spend, actual_spend + saved_spend
689def _benchmark_totals(row: _SessionAggRow) -> AutoRouterBenchmarkTotals:
690 return_misses: Final = row.return_turns - row.return_hits
691 saved_spend, baseline_spend = _savings_cohort(
692 row.turns, row.savings_estimated_turns, row.savings_estimated_actual_spend, row.savings_estimated_saved_spend
693 )
694 sessions: Final = row.sessions
695 return AutoRouterBenchmarkTotals(
696 sessions=sessions,
697 turns=row.turns,
698 avg_turns_per_session=row.turns / sessions if sessions else 0.0,
699 avg_session_seconds=row.session_seconds / sessions if sessions else 0.0,
700 avg_tokens_per_session=row.total_tokens / sessions if sessions else 0.0,
701 spend=row.spend,
702 savings_estimated_turns=row.savings_estimated_turns,
703 savings_estimated_actual_spend=row.savings_estimated_actual_spend,
704 saved_spend=saved_spend,
705 classifier_cost=row.classifier_cost if row.classifier_cost_recorded_turns == row.turns else None,
706 baseline_spend=baseline_spend,
707 saved_pct=_pct(saved_spend, baseline_spend) if saved_spend is not None and baseline_spend is not None else None,
708 saved_per_session=(row.savings_estimated_saved_spend / sessions if sessions else 0.0)
709 if row.savings_estimated_turns == row.turns
710 else None,
711 cache=AutoRouterCacheStats(
712 coverage_pct=_pct(row.covered_turns, row.turns),
713 hit_rate_pct=_pct(row.cache_hits, row.covered_turns),
714 same_model=_cache_bucket(row.same_model_turns, row.same_model_hits),
715 first_visit=_cache_bucket(row.first_visit_turns, row.first_visit_hits),
716 return_to_tier=_cache_bucket(row.return_turns, row.return_hits),
717 unordered_turns=row.unordered_turns,
718 return_misses_expired=row.return_expired_misses,
719 return_misses_within_ttl=row.return_within_ttl_misses,
720 return_misses_unknown=max(return_misses - row.return_expired_misses - row.return_within_ttl_misses, 0),
721 ttl_5m_turns=row.ttl_5m_turns,
722 ttl_1h_turns=row.ttl_1h_turns,
723 ),
724 )
727def _benchmark_group(row: _SessionAggRow) -> AutoRouterBenchmarkGroup:
728 totals: Final = _benchmark_totals(row)
729 return AutoRouterBenchmarkGroup(
730 router_name=row.router_name,
731 router_type=row.router_type,
732 tier_turns=row.tier_turns,
733 sessions=totals.sessions,
734 turns=totals.turns,
735 avg_turns_per_session=totals.avg_turns_per_session,
736 avg_session_seconds=totals.avg_session_seconds,
737 avg_tokens_per_session=totals.avg_tokens_per_session,
738 spend=totals.spend,
739 saved_spend=totals.saved_spend,
740 savings_estimated_turns=totals.savings_estimated_turns,
741 savings_estimated_actual_spend=totals.savings_estimated_actual_spend,
742 classifier_cost=totals.classifier_cost,
743 baseline_spend=totals.baseline_spend,
744 saved_pct=totals.saved_pct,
745 saved_per_session=totals.saved_per_session,
746 cache=totals.cache,
747 )
750def _summed_agg_row(rows: Sequence[_SessionAggRow]) -> _SessionAggRow:
751 return _SessionAggRow(
752 router_name="",
753 router_type="",
754 tier_turns=MappingProxyType({}),
755 sessions=sum(row.sessions for row in rows),
756 turns=sum(row.turns for row in rows),
757 unordered_turns=sum(row.unordered_turns for row in rows),
758 covered_turns=sum(row.covered_turns for row in rows),
759 cache_hits=sum(row.cache_hits for row in rows),
760 same_model_turns=sum(row.same_model_turns for row in rows),
761 same_model_hits=sum(row.same_model_hits for row in rows),
762 first_visit_turns=sum(row.first_visit_turns for row in rows),
763 first_visit_hits=sum(row.first_visit_hits for row in rows),
764 return_turns=sum(row.return_turns for row in rows),
765 return_hits=sum(row.return_hits for row in rows),
766 return_expired_misses=sum(row.return_expired_misses for row in rows),
767 return_within_ttl_misses=sum(row.return_within_ttl_misses for row in rows),
768 ttl_5m_turns=sum(row.ttl_5m_turns for row in rows),
769 ttl_1h_turns=sum(row.ttl_1h_turns for row in rows),
770 total_tokens=sum(row.total_tokens for row in rows),
771 spend=sum(row.spend for row in rows),
772 saved_spend=sum(row.saved_spend for row in rows),
773 savings_estimated_turns=sum(row.savings_estimated_turns for row in rows),
774 savings_estimated_actual_spend=sum(row.savings_estimated_actual_spend for row in rows),
775 savings_estimated_saved_spend=sum(row.savings_estimated_saved_spend for row in rows),
776 classifier_cost=sum(row.classifier_cost for row in rows),
777 classifier_cost_recorded_turns=sum(row.classifier_cost_recorded_turns for row in rows),
778 session_seconds=sum(row.session_seconds for row in rows),
779 )
782def _strategy_router_key(deployment: object) -> tuple[str, str] | None:
783 """``(model_name, kind)`` for a deployment whose routing the session rollup records.
785 Kinds come from ``classify_strategy_router_model``, the same rule the Router registers a
786 deployment by, so this arm cannot disagree with the arm that stamped ``router_type`` onto
787 the session rows. Semantic auto-routers return None: they record no routing decision, so
788 they can never own a session row, and ``AutoRouterBenchmarkGroup.router_type`` has no
789 value for them. A permanent zero would read as "no traffic" rather than "not instrumented".
790 """
791 if not isinstance(deployment, Mapping): 791 ↛ 792line 791 didn't jump to line 792 because the condition on line 791 was never true
792 return None
793 litellm_params: Final = deployment.get("litellm_params")
794 router_name: Final = deployment.get("model_name")
795 if not (isinstance(litellm_params, Mapping) and isinstance(router_name, str) and router_name): 795 ↛ 796line 795 didn't jump to line 796 because the condition on line 795 was never true
796 return None
797 model: Final = litellm_params.get("model")
798 if not isinstance(model, str): 798 ↛ 799line 798 didn't jump to line 799 because the condition on line 798 was never true
799 return None
800 kind: Final = classify_strategy_router_model(model)
801 return None if kind is None or kind == "semantic" else (router_name, kind)
804def _idle_router_groups(
805 llm_router: "Router | None", covered: frozenset[tuple[str, str]]
806) -> tuple[AutoRouterBenchmarkGroup, ...]:
807 """Zeroed groups for configured strategy routers the window's traffic did not cover.
809 The dashboard's router picker has to list a router the moment it is created rather than
810 once it has spent something, so the registry drives the list and the rollup only supplies
811 the measures. ``_summed_agg_row`` over no sessions is already the zero element of the
812 fold, so a group with every measure at zero costs one relabel rather than a literal that
813 would go stale the next time the response grows a field.
814 """
815 if llm_router is None: 815 ↛ 816line 815 didn't jump to line 816 because the condition on line 815 was never true
816 return ()
817 zero: Final = _summed_agg_row(())
818 idle: Final = frozenset(
819 key
820 for key in (_strategy_router_key(deployment) for deployment in llm_router.model_list or ())
821 if key is not None and key not in covered
822 )
823 return tuple(
824 _benchmark_group(zero.model_copy(update=MappingProxyType({"router_name": name, "router_type": kind})))
825 for name, kind in sorted(idle)
826 )
829@router.get(
830 "/auto_router/benchmarks",
831 tags=("auto router",),
832 dependencies=(Depends(user_api_key_auth),),
833 response_model=AutoRouterBenchmarksResponse,
834)
835async def get_auto_router_benchmarks(
836 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
837 start_date: Annotated[
838 str | None, Query(description="YYYY-MM-DD UTC, inclusive (defaults to 30 days before end_date)")
839 ] = None,
840 end_date: Annotated[str | None, Query(description="YYYY-MM-DD UTC, inclusive (defaults to today)")] = None,
841 api_key: Annotated[str | None, Query(description="Filter to one virtual key token hash")] = None,
842 user_id: Annotated[
843 str | None, Query(min_length=1, description="Filter to one canonical internal user recorded on each turn")
844 ] = None,
845) -> AutoRouterBenchmarksResponse:
846 """
847 Benchmarks for the auto-router dashboard: session shape, savings against the configured
848 baseline, and prompt-caching behaviour bucketed by what the router did.
850 Reads session rollups folded once per request at spend-write time, so this endpoint
851 never scans LiteLLM_SpendLogs. A user filter selects only turns attributed to that
852 internal user when written; older key-only history remains outside user views. A session
853 is in the window when it overlaps it: its last turn is on or after start_date and its first turn is on or before
854 end_date. Overall hit rate is over telemetry-bearing turns; each bucket's hit rate is
855 over that bucket's turns.
857 The rollup supplies the measures, never the list. Which routers appear comes from the
858 model registry, so one shows up as soon as it is configured and reads zero until it
859 serves traffic, and `routers_in_scope` counts those too rather than only the routers the
860 window recorded.
861 """
862 from litellm.proxy.proxy_server import llm_router, prisma_client
864 _require_admin_viewer(user_api_key_dict, "view auto-router benchmarks across the deployment")
865 if prisma_client is None: 865 ↛ 866line 865 didn't jump to line 866 because the condition on line 865 was never true
866 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
868 end_day: Final = (
869 _parse_benchmark_day(end_date)
870 if end_date
871 else datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=None)
872 )
873 start_day: Final = _parse_benchmark_day(start_date) if start_date else end_day - timedelta(days=30)
874 if end_day < start_day: 874 ↛ 875line 874 didn't jump to line 875 because the condition on line 874 was never true
875 raise HTTPException(status_code=400, detail="end_date must not be earlier than start_date")
877 raw_rows: Final = await _query_raw(
878 prisma_client,
879 AUTOROUTER_BENCHMARKS_SQL,
880 start_day.isoformat(),
881 (end_day + timedelta(days=1)).isoformat(),
882 api_key,
883 user_id,
884 )
885 rows: Final = _SESSION_AGG_ROWS.validate_python(raw_rows or ())
886 groups: Final = (
887 *(_benchmark_group(row) for row in rows),
888 *_idle_router_groups(llm_router, frozenset((row.router_name, row.router_type) for row in rows)),
889 )
890 return AutoRouterBenchmarksResponse(
891 start_date=start_day.strftime("%Y-%m-%d"),
892 end_date=end_day.strftime("%Y-%m-%d"),
893 routers_in_scope=len(groups),
894 totals=_benchmark_totals(_summed_agg_row(rows)),
895 groups=groups,
896 )
899@router.get(
900 "/auto_router/session",
901 tags=("auto router",),
902 dependencies=(Depends(user_api_key_auth),),
903 response_model=AutoRouterSessionResponse,
904)
905async def get_auto_router_session(
906 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
907 session_id: Annotated[
908 str, Query(description="The client session id (x-*-session-id header) the turns were sent under")
909 ],
910) -> AutoRouterSessionResponse:
911 """
912 One auto-routed session, for the key that ran it: the model its last turn was routed to and the
913 session's spend against the router's savings baseline. Built for a coding agent's status line
914 or stop hook, so any virtual key may call it and only ever sees rows written under its own
915 key hash. Reads the LiteLLM_AutoRouterSession rollup, which the asynchronous spend flush
916 fills a moment after each turn; a session with no flushed auto-routed turn yet is a 404. The
917 id is bounded the way the writer bounded it, so an oversized client id still finds its row.
918 """
919 from litellm.proxy.proxy_server import prisma_client
921 if prisma_client is None: 921 ↛ 922line 921 didn't jump to line 922 because the condition on line 921 was never true
922 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
923 row: Final = await AutoRouterSessionRepository(prisma_client).find_latest_for_key(
924 user_api_key_dict.api_key, bounded_session_id(session_id)
925 )
926 if row is None: 926 ↛ 930line 926 didn't jump to line 930 because the condition on line 926 was always true
927 raise HTTPException(
928 status_code=404, detail=f"No auto-routed turns recorded for session {session_id!r} under this key"
929 )
930 saved_spend, baseline_spend = _savings_cohort(
931 row.turns, row.savings_estimated_turns, row.savings_estimated_actual_spend, row.savings_estimated_saved_spend
932 )
933 return AutoRouterSessionResponse(
934 session_id=session_id,
935 router_name=row.router_name,
936 router_type=row.router_type,
937 turns=row.turns,
938 last_model=row.last_model,
939 spend=row.spend,
940 savings_estimated_turns=row.savings_estimated_turns,
941 savings_estimated_actual_spend=row.savings_estimated_actual_spend,
942 saved_spend=saved_spend,
943 baseline_spend=baseline_spend if row.savings_estimated_turns == row.turns else None,
944 savings_estimated_baseline_spend=baseline_spend,
945 baseline_model=row.baseline_model,
946 baseline_models=row.savings_estimated_baseline_models,
947 )
950# ---------------------------------------------------------------------------
951# Shadow eval: pre-adoption evaluation of an auto-router against live traffic.
952# The job row is immutable config plus stopped_at; status, counts, spend, and errors
953# are derived from the append-only attempt rows, so reads here are aggregations
954# bounded by each job's max_turns through the attempt table's job_id index.
955# ---------------------------------------------------------------------------
958def _require_admin_viewer(user_api_key_dict: UserAPIKeyAuth, action: str) -> None:
959 if user_api_key_dict.user_role not in ( 959 ↛ 963line 959 didn't jump to line 963 because the condition on line 959 was never true
960 LitellmUserRoles.PROXY_ADMIN,
961 LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
962 ):
963 raise HTTPException(status_code=403, detail=f"Only proxy admin roles can {action}")
966def _require_admin_writer(user_api_key_dict: UserAPIKeyAuth, action: str) -> None:
967 if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: 967 ↛ 968line 967 didn't jump to line 968 because the condition on line 967 was never true
968 raise HTTPException(status_code=403, detail=f"Only a proxy admin can {action}")
971def _is_configured_pre_routing_strategy(llm_router: "Router", router_name: str) -> bool:
972 return any(
973 router_name in registry
974 for registry in (
975 llm_router.auto_routers,
976 llm_router.complexity_routers,
977 llm_router.adaptive_routers,
978 llm_router.quality_routers,
979 )
980 )
983def _sdk_model_is_missing_anthropic_credentials(model: str) -> bool:
984 _, provider, _, _ = litellm.get_llm_provider(model=model)
985 if provider != "anthropic" or litellm.anthropic_key or litellm.api_key:
986 return False
987 from litellm.llms.anthropic.common_utils import AnthropicModelInfo
988 from litellm.secret_managers.main import secret_manager_would_be_consulted
990 if AnthropicModelInfo.get_api_key() or AnthropicModelInfo.get_auth_token():
991 return False
992 return not any(
993 secret_manager_would_be_consulted(secret_name) for secret_name in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN")
994 )
997def _validate_plain_model(
998 llm_router: "Router | None", model: str, field_name: str, team_ids: Sequence[str | None]
999) -> None:
1000 """Reject a model the dispatch path cannot resolve, at start rather than as a silently
1001 growing error count once the job is already sampling and billing. Both the judge and a
1002 reverse job's baseline must be plain models: an auto-router in either slot would
1003 re-route per turn, so the comparison would have no fixed arm to attribute results to.
1005 Resolvability is asked once per team the job samples for, because that is the identity
1006 the call carries: a name only one team can reach fails every turn for the other keys,
1007 which is the growing error count this check exists to prevent."""
1008 if llm_router is not None and _is_configured_pre_routing_strategy(llm_router, model):
1009 raise HTTPException(
1010 status_code=400,
1011 detail=f"{field_name} '{model}' is an auto-router; it must be a plain model",
1012 )
1013 targets: Final = tuple((team, judge_target(llm_router, model, team)) for team in team_ids)
1014 unreachable: Final = tuple(team for team, target in targets if target.via == "nothing")
1015 if unreachable:
1016 raise HTTPException(
1017 status_code=400,
1018 detail=(
1019 f"{field_name} '{model}' is neither a model configured on this proxy nor a "
1020 "provider-qualified public model name (e.g. 'anthropic/claude-sonnet-5')" + _for_teams(unreachable)
1021 ),
1022 )
1023 sdk_teams: Final = tuple(team for team, target in targets if target.via == "sdk")
1024 if not sdk_teams:
1025 return
1026 if not _sdk_model_is_missing_anthropic_credentials(model):
1027 return
1028 raise HTTPException(
1029 status_code=400,
1030 detail=(
1031 f"{field_name} '{model}' uses the LiteLLM SDK but required credentials are not configured: "
1032 "ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN" + _for_teams(sdk_teams)
1033 ),
1034 )
1037def _for_teams(team_ids: Sequence[str | None]) -> str:
1038 """Name the teams a fault applies to, when it does not apply to every key alike."""
1039 named: Final = tuple(sorted(team for team in team_ids if team is not None))
1040 return f" for team {', '.join(named)}" if named else ""
1043def _validate_model_scope(llm_router: "Router | None", models: Sequence[str]) -> None:
1044 """Reject a scope naming a model no request on this proxy could carry, at start rather
1045 than as a job that silently samples nothing. The question is "could any caller ask for
1046 this name", not "does it resolve for the job's teams": a user target's traffic can arrive
1047 on any team's key, so a team-public name is a legitimate scope for it, and an auto-router
1048 is one too (a forward job on router A scoped to router B samples what B serves today).
1049 Nothing here is ever dispatched to."""
1050 unreachable: Final = tuple(
1051 model
1052 for model in models
1053 if judge_target(llm_router, model).via == "nothing"
1054 and (llm_router is None or model not in llm_router.team_public_model_names)
1055 )
1056 if unreachable:
1057 raise HTTPException(
1058 status_code=400,
1059 detail="models not served by this proxy: " + ", ".join(f"'{model}'" for model in unreachable),
1060 )
1063_JUDGED_ROLES: Final[frozenset[StrategyRouterDependencyRole]] = frozenset({"tier", "default"})
1066def _router_arm_models(llm_router: "Router | None", router_name: str) -> tuple[tuple[str, str], ...]:
1067 """``(role, model_name)`` for every model the router under evaluation can answer with.
1069 Drawn from ``strategy_router_dependencies``, the single answer to "what does this router
1070 call", so this cannot disagree with the health check's reading of the same deployment.
1071 Only the roles that SERVE are arms: the classifier and embedding models pick the tier,
1072 they never produce a response anyone judges, so a judge sharing them carries no
1073 self-preference.
1075 A semantic auto-router keeps its routes in an opaque config blob or a file, so only its
1076 default model is enumerable and the guard below is incomplete for it. That direction is
1077 deliberate: it can miss a collision, never invent one.
1079 Which tiers a router declares is a property of its config and not of who is calling, so
1080 this lookup is unscoped; what each tier NAME resolves to is the team-dependent half, and
1081 it belongs to the caller that compares them.
1082 """
1083 deployments: Final = llm_router.get_model_list(model_name=router_name) if llm_router is not None else None
1084 return tuple(
1085 dict.fromkeys(
1086 (dependency.role, dependency.model_name)
1087 for deployment in deployments or ()
1088 for dependency in strategy_router_dependencies(deployment["litellm_params"])
1089 if dependency.role in _JUDGED_ROLES
1090 )
1091 )
1094def _judge_collisions_for_team(
1095 llm_router: "Router | None", data: StartShadowEvalRequest, team_id: str | None
1096) -> tuple[tuple[str, str], ...]:
1097 """``(role, model_name)`` for each arm the judge would also be, as one team's keys see it.
1099 Both sides resolve under the SAME team, since two names are the same model only for a
1100 caller who can reach both; resolving the judge for one team against an arm for another
1101 invents a collision no request could produce.
1102 """
1103 judge: Final = judge_target(llm_router, data.judge_model, team_id).models
1104 return tuple(
1105 (role, model)
1106 for role, model in (
1107 *(arm for name in data.router_names for arm in _router_arm_models(llm_router, name)),
1108 *((("baseline", data.baseline_model),) if data.baseline_model is not None else ()),
1109 )
1110 if judge & judge_target(llm_router, model, team_id).models
1111 )
1114def _validate_judge_is_not_a_candidate(
1115 llm_router: "Router | None", data: StartShadowEvalRequest, team_ids: Sequence[str | None]
1116) -> None:
1117 """Reject a judge that is one of the two arms it grades.
1119 A judge scores its own output higher than a rival's, so a run whose judge also serves an
1120 arm reports a win rate for that arm that measures the judge rather than the models, and
1121 the whole job's spend buys a result that has to be discarded. Both arms are in scope: the
1122 router answers with a tier or default model in either direction, and a reverse job's
1123 ``baseline_model`` is the fixed arm the router is compared against.
1125 Names are compared by what would ANSWER them, not by spelling: the shipped default judge
1126 ``anthropic/claude-sonnet-5`` collides with a tier deployment an admin named
1127 ``sonnet-tier``, and an alias collides with its target, neither of which a string
1128 comparison sees.
1130 A collision for ONE team is a collision for the job, because the verdicts every key
1131 produces land in the same win rates.
1132 """
1133 collisions: Final = tuple(
1134 dict.fromkeys(
1135 collision for team_id in team_ids for collision in _judge_collisions_for_team(llm_router, data, team_id)
1136 )
1137 )
1138 if not collisions:
1139 return
1140 raise HTTPException(
1141 status_code=400,
1142 detail=(
1143 f"judge_model '{data.judge_model}' is also an arm this job would judge: "
1144 + ", ".join(f"{role} model '{model}'" for role, model in collisions)
1145 + ". A judge scores its own answers higher than a rival's, so the win rates would "
1146 "measure the judge; pick a judge that serves neither arm"
1147 ),
1148 )
1151def _is_unique_violation(error: Exception) -> bool:
1152 """Whether a Prisma create failed on a unique index. One active job per target and
1153 direction lives in a partial unique index (raw SQL in the migration; schema.prisma
1154 cannot express partial indexes), so the read-then-create check above it is advisory:
1155 two concurrent starts pass the read, and the loser must surface as the same 409
1156 rather than a 500."""
1157 try:
1158 from prisma.errors import UniqueViolationError
1159 except ImportError:
1160 return "unique constraint" in str(error).lower() or "P2002" in str(error)
1161 return isinstance(error, UniqueViolationError)
1164class _AttemptAggRow(BaseModel):
1165 grp: str
1166 turn_count: int
1167 real_wins: int
1168 shadow_wins: int
1169 ties: int
1170 avg_confidence: float | None
1171 real_spend: float
1172 shadow_spend: float
1173 cache_hit_turns: int
1176_ATTEMPT_AGG_ROWS: Final = TypeAdapter(list[_AttemptAggRow])
1178_ATTEMPT_AGG_COLUMNS: Final = """
1179 COUNT(*)::int AS turn_count,
1180 COUNT(*) FILTER (WHERE outcome = 'real')::int AS real_wins,
1181 COUNT(*) FILTER (WHERE outcome = 'shadow')::int AS shadow_wins,
1182 COUNT(*) FILTER (WHERE outcome = 'tie')::int AS ties,
1183 AVG(confidence)::float AS avg_confidence,
1184 COALESCE(SUM(real_cost + real_classifier_cost) FILTER (WHERE real_cost IS NOT NULL AND NOT real_cache_hit), 0)::float AS real_spend,
1185 COALESCE(SUM(shadow_cost + shadow_classifier_cost) FILTER (WHERE real_cost IS NOT NULL AND NOT real_cache_hit), 0)::float AS shadow_spend,
1186 COUNT(*) FILTER (WHERE real_cache_hit)::int AS cache_hit_turns
1187"""
1189_ATTEMPT_AGG_SELECT: Final = (
1190 _ATTEMPT_AGG_COLUMNS
1191 + """
1192FROM "LiteLLM_ShadowEvalAttempt"
1193WHERE job_id = ANY($1::text[]) AND outcome != 'error'
1194GROUP BY 1
1195"""
1196)
1198_ATTEMPT_AGG_BY_TIER_SQL: Final = "SELECT COALESCE(tier, 'UNCLASSIFIED') AS grp," + _ATTEMPT_AGG_SELECT
1199_ATTEMPT_AGG_BY_MODEL_SQL: Final = "SELECT COALESCE(real_model, 'unknown') AS grp," + _ATTEMPT_AGG_SELECT
1200_ATTEMPT_AGG_BY_LEG_SQL: Final = "SELECT job_id AS grp," + _ATTEMPT_AGG_SELECT
1202# Attempt rows from before arm stamping carry no router_name; they belong to the job's
1203# own router, which the join reads off the leg.
1204_ATTEMPT_AGG_BY_ROUTER_SQL: Final = (
1205 "SELECT COALESCE(a.router_name, j.router_name) AS grp,"
1206 + _ATTEMPT_AGG_COLUMNS
1207 + """
1208FROM "LiteLLM_ShadowEvalAttempt" a
1209JOIN "LiteLLM_ShadowEvalJob" j ON j.id = a.job_id
1210WHERE a.job_id = ANY($1::text[]) AND a.outcome != 'error'
1211GROUP BY 1
1212"""
1213)
1215# These guards derive spend from attempt rows, the cross-pod authority; the sampler also
1216# reads the live counter, so admission can stop before a row-based guard would fire (safe
1217# direction, and mid-deploy rows from old pods price as judge-only until the deploy ends).
1218_SWEEP_FINISHED_JOBS_SQL: Final = """
1219UPDATE "LiteLLM_ShadowEvalJob" j SET stopped_at = (NOW() AT TIME ZONE 'utc')
1220WHERE j.target_type = $2 AND j.target_id = ANY($1::text[]) AND j.stopped_at IS NULL
1221 AND (
1222 j.ends_at <= (NOW() AT TIME ZONE 'utc')
1223 OR (SELECT COUNT(*) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = j.id) >= j.max_turns
1224 OR (
1225 j.max_budget IS NOT NULL
1226 AND (SELECT COALESCE(SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost), 0) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = j.id) >= j.max_budget
1227 )
1228 )
1229"""
1231_ATTEMPT_TOTALS_SQL: Final = """
1232SELECT
1233 COUNT(*) FILTER (WHERE outcome != 'error')::int AS judged_count,
1234 COUNT(*) FILTER (WHERE outcome = 'error')::int AS error_count,
1235 COALESCE(SUM(judge_cost), 0)::float AS judge_spend
1236FROM "LiteLLM_ShadowEvalAttempt"
1237WHERE job_id = ANY($1::text[])
1238"""
1240_ATTEMPT_COUNTS_SQL: Final = """
1241SELECT a.job_id, COUNT(*)::int AS attempt_count, COALESCE(SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost), 0)::float AS spend
1242FROM "LiteLLM_ShadowEvalAttempt" a
1243JOIN "LiteLLM_ShadowEvalJob" j ON j.id = a.job_id
1244WHERE a.job_id = ANY($1::text[]) AND (j.stopped_at IS NULL OR a.created_at <= j.stopped_at)
1245GROUP BY a.job_id
1246"""
1248_FUNNEL_TOTALS_SQL: Final = """
1249SELECT COUNT(*)::int AS legs_with_rows,
1250 COALESCE(SUM(not_sampled), 0)::int AS not_sampled,
1251 COALESCE(SUM(unjudgeable), 0)::int AS unjudgeable,
1252 COALESCE(SUM(shed), 0)::int AS shed,
1253 COALESCE(SUM(withheld), 0)::int AS withheld
1254FROM "LiteLLM_ShadowEvalFunnel"
1255WHERE job_id = ANY($1::text[])
1256"""
1259_STOP_JOB_SQL: Final = """
1260UPDATE "LiteLLM_ShadowEvalJob"
1261SET stopped_by = $2, stopped_at = COALESCE(stopped_at, $3::timestamp)
1262WHERE group_id = $1 AND stopped_by IS NULL
1263 AND ends_at > (NOW() AT TIME ZONE 'utc')
1264 AND EXISTS (
1265 SELECT 1 FROM "LiteLLM_ShadowEvalJob" k
1266 WHERE k.group_id = $1 AND k.stopped_at IS NULL
1267 AND (SELECT COUNT(*) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = k.id) < k.max_turns
1268 AND (
1269 k.max_budget IS NULL
1270 OR (SELECT COALESCE(SUM(a.judge_cost + a.shadow_cost + a.shadow_classifier_cost), 0) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = k.id) < k.max_budget
1271 )
1272 )
1273"""
1276class _FunnelTotalsRow(BaseModel):
1277 legs_with_rows: int
1278 not_sampled: int
1279 unjudgeable: int
1280 shed: int
1281 withheld: int
1284class _AttemptCountRow(BaseModel):
1285 job_id: str
1286 attempt_count: int
1287 spend: float
1290_ATTEMPT_COUNT_ROWS: Final = TypeAdapter(list[_AttemptCountRow])
1293_LIST_LEGS_SQL: Final = """
1294SELECT * FROM "LiteLLM_ShadowEvalJob"
1295WHERE group_id IN (
1296 SELECT group_id FROM "LiteLLM_ShadowEvalJob"
1297 GROUP BY group_id ORDER BY MAX(created_at) DESC LIMIT $1::int
1298)
1299"""
1301_LIST_LEGS_BY_TARGET_SQL: Final = """
1302SELECT * FROM "LiteLLM_ShadowEvalJob"
1303WHERE group_id IN (
1304 SELECT group_id FROM "LiteLLM_ShadowEvalJob" WHERE target_type = $2 AND target_id = $3
1305 GROUP BY group_id ORDER BY MAX(created_at) DESC LIMIT $1::int
1306)
1307"""
1310class _AttemptTotalsRow(BaseModel):
1311 judged_count: int
1312 error_count: int
1313 judge_spend: float
1316_ATTEMPT_TOTALS_ROWS: Final = TypeAdapter(list[_AttemptTotalsRow])
1319def _pct_of(numerator: int, denominator: int) -> float:
1320 return _pct(numerator, denominator)
1323def _slices(rows: Sequence[_AttemptAggRow]) -> tuple[ShadowEvalSlice, ...]:
1324 return tuple(
1325 ShadowEvalSlice(
1326 group=row.grp,
1327 turn_count=row.turn_count,
1328 real_win_rate_pct=_pct_of(row.real_wins, row.turn_count),
1329 shadow_win_rate_pct=_pct_of(row.shadow_wins, row.turn_count),
1330 tie_rate_pct=_pct_of(row.ties, row.turn_count),
1331 avg_judge_confidence=round(row.avg_confidence or 0.0, 3),
1332 real_spend=row.real_spend,
1333 shadow_spend=row.shadow_spend,
1334 cache_hit_turns=row.cache_hit_turns,
1335 )
1336 for row in sorted(rows, key=lambda r: r.turn_count, reverse=True)
1337 )
1340def _leg_group_id(leg: "_LegRow") -> str:
1341 return leg.group_id
1344class _LegRow(BaseModel):
1345 """One LiteLLM_ShadowEvalJob row, validated off the untyped prisma record. A row is
1346 one target's leg of a job; the legs of a job share group_id and identical config,
1347 written together by one create_many. The API's job id is the group id, so leg ids
1348 never leave the server (attempts reference them internally)."""
1350 model_config = ConfigDict(from_attributes=True)
1352 id: str
1353 group_id: str
1354 target_type: ShadowEvalTargetType
1355 target_id: str
1356 router_name: str
1357 router_names: tuple[str, ...] = ()
1358 models: tuple[str, ...] = ()
1359 direction: ShadowEvalDirection
1360 baseline_model: str | None = None
1361 judge_model: str
1362 shadow_percentage: float
1363 max_turns: int
1364 max_budget: float | None = None
1365 created_at: datetime
1366 ends_at: datetime
1367 stopped_at: datetime | None = None
1368 stopped_by: str | None = None
1370 @property
1371 def arm_router_names(self) -> tuple[str, ...]:
1372 """The job's full router set; rows from before router_names existed hold it in
1373 router_name alone. The one place that reading lives on the endpoint side."""
1374 return self.router_names or (self.router_name,)
1376 @field_validator("created_at", "ends_at", "stopped_at")
1377 @classmethod
1378 def _as_aware_utc(cls, value: datetime | None) -> datetime | None:
1379 """The columns store naive UTC wall time (prisma's convention); prisma reads hand
1380 back aware datetimes while raw SQL reads hand back naive ones, so this boundary
1381 makes every read aware UTC before anything compares or serializes them."""
1382 if value is None or value.tzinfo is not None:
1383 return value
1384 return value.replace(tzinfo=timezone.utc)
1387_LEG_ROWS: Final = TypeAdapter(list[_LegRow])
1390async def _leg_attempt_counts(prisma_client: "PrismaClient", legs: Sequence[_LegRow]) -> Mapping[str, _AttemptCountRow]:
1391 """Each leg's attempt count and recorded spend by leg id, judged and errored alike, in
1392 one grouped read. They are the same figures the sampler budgets against max_turns and
1393 max_budget, so the derived status flips to completed exactly when sampling actually
1394 ends. A stamped leg's figures freeze at its stopped_at: in-flight attempts that land
1395 after the stamp are excluded, so they can never reclassify a leg that was stopped
1396 under budget as budget-spent."""
1397 if not legs: 1397 ↛ 1399line 1397 didn't jump to line 1399 because the condition on line 1397 was always true
1398 return MappingProxyType({})
1399 rows: Final = _ATTEMPT_COUNT_ROWS.validate_python(
1400 await _query_raw(prisma_client, _ATTEMPT_COUNTS_SQL, [leg.id for leg in legs]) # mutable-ok: query param
1401 or ()
1402 )
1403 return MappingProxyType({row.job_id: row for row in rows})
1406def _group_response(
1407 group_id: str, legs: Sequence[_LegRow], attempt_counts: Mapping[str, _AttemptCountRow]
1408) -> ShadowEvalJobResponse:
1409 """The one constructor of a job response: the caller names the group and passes that
1410 group's legs. Config is read off the first leg because every leg carries the same copy,
1411 written by one create_many. No caller may serialize a raw row (that would leak a leg id
1412 as the job id)."""
1413 first: Final = legs[0]
1414 return ShadowEvalJobResponse(
1415 job_id=group_id,
1416 targets=tuple(
1417 ShadowEvalJobTargetResponse(
1418 target_type=leg.target_type,
1419 target_id=leg.target_id,
1420 max_turns=leg.max_turns,
1421 max_budget=leg.max_budget,
1422 stopped_at=leg.stopped_at,
1423 attempt_count=stats.attempt_count if (stats := attempt_counts.get(leg.id)) else 0,
1424 spend=round(stats.spend, 6) if stats else 0.0,
1425 )
1426 for leg in sorted(legs, key=lambda leg: (leg.target_type, leg.target_id))
1427 ),
1428 router_names=first.arm_router_names,
1429 models=first.models,
1430 direction=first.direction,
1431 baseline_model=first.baseline_model,
1432 judge_model=first.judge_model,
1433 shadow_percentage=first.shadow_percentage,
1434 created_at=first.created_at,
1435 ends_at=first.ends_at,
1436 stopped_by=next((leg.stopped_by for leg in legs if leg.stopped_by is not None), None),
1437 )
1440_NO_TARGET_LABELS: Final[tuple[str | None, str | None]] = (None, None)
1443def _target_labels(
1444 key_rows: Sequence[_VerificationTokenRow],
1445 team_rows: Sequence[_TeamRow],
1446 user_rows: Sequence[_UserRow],
1447) -> Mapping[tuple[str, str], tuple[str | None, str | None]]:
1448 """Display labels by (target_type, target_id): a key's (alias, masked name), a
1449 team's (alias, None), a user's (email, None)."""
1450 return MappingProxyType(
1451 {
1452 key: value
1453 for key, value in chain(
1454 ((("key", row.token), (row.key_alias, row.key_name)) for row in key_rows),
1455 ((("team", row.team_id), (row.team_alias, None)) for row in team_rows),
1456 ((("user", row.user_id), (row.user_email, None)) for row in user_rows),
1457 )
1458 }
1459 )
1462def _target_ids_of(responses: Sequence[ShadowEvalJobResponse], target_type: ShadowEvalTargetType) -> tuple[str, ...]:
1463 return tuple(
1464 sorted(
1465 frozenset(
1466 target.target_id
1467 for response in responses
1468 for target in response.targets
1469 if target.target_type == target_type
1470 )
1471 )
1472 )
1475async def _with_target_labels(
1476 prisma_client: "PrismaClient", responses: Sequence[ShadowEvalJobResponse]
1477) -> tuple[ShadowEvalJobResponse, ...]:
1478 """Resolve every scoped target's id to a display label in one batched read per kind,
1479 so the UI can say whose traffic a job shadows: a key's alias and masked name, a
1480 team's alias, a user's email. Deleted targets resolve to None."""
1481 if not responses: 1481 ↛ 1483line 1481 didn't jump to line 1483 because the condition on line 1481 was always true
1482 return ()
1483 tokens: Final = _target_ids_of(responses, "key")
1484 team_ids: Final = _target_ids_of(responses, "team")
1485 user_ids: Final = _target_ids_of(responses, "user")
1486 key_rows: Final = (
1487 await _verification_tokens(prisma_client).find_many(
1488 where={"token": {"in": list(tokens)}} # mutable-ok: Prisma filter
1489 )
1490 if tokens
1491 else ()
1492 )
1493 team_rows: Final = (
1494 await _team_rows(prisma_client).find_many(
1495 where={"team_id": {"in": list(team_ids)}} # mutable-ok: Prisma filter
1496 )
1497 if team_ids
1498 else ()
1499 )
1500 user_rows: Final = (
1501 await _user_rows(prisma_client).find_many(
1502 where={"user_id": {"in": list(user_ids)}} # mutable-ok: Prisma filter
1503 )
1504 if user_ids
1505 else ()
1506 )
1507 labels: Final = _target_labels(key_rows or (), team_rows or (), user_rows or ())
1508 return tuple(
1509 response.model_copy(
1510 update={ # mutable-ok: pydantic update payload
1511 "targets": tuple(
1512 target.model_copy(
1513 update={ # mutable-ok: pydantic update payload
1514 "target_alias": labels.get((target.target_type, target.target_id), _NO_TARGET_LABELS)[0],
1515 "key_name": labels.get((target.target_type, target.target_id), _NO_TARGET_LABELS)[1],
1516 }
1517 )
1518 for target in response.targets
1519 )
1520 }
1521 )
1522 for response in responses
1523 )
1526async def _shadow_eval_results(
1527 prisma_client: "PrismaClient", legs: Sequence[_LegRow]
1528) -> tuple[ShadowEvalResult | None, Mapping[tuple[str, str], ShadowEvalSlice]]:
1529 """One job's stratified verdicts, plus each target's own slice keyed by the
1530 (target_type, target_id) pair so a key, team, and user sharing an id can never
1531 collapse into one entry. Tier answers "where does the router do well"; the model
1532 stratification groups by whichever model served the real arm, so it answers "which
1533 of the models these targets use today would the router beat" forward, and "for the
1534 turns the router sent to X, did X beat the baseline" in reverse; the per-target
1535 slices answer "which target's traffic does the router suit". Reads are bounded by
1536 the job's own attempts (<= the sum of its targets' max_turns) via the job_id index."""
1537 leg_ids: Final = [leg.id for leg in legs] # mutable-ok: query param
1538 by_tier: Final = _ATTEMPT_AGG_ROWS.validate_python(
1539 await _query_raw(prisma_client, _ATTEMPT_AGG_BY_TIER_SQL, leg_ids) or ()
1540 )
1541 if not by_tier:
1542 return None, MappingProxyType({})
1543 by_model: Final = _ATTEMPT_AGG_ROWS.validate_python(
1544 await _query_raw(prisma_client, _ATTEMPT_AGG_BY_MODEL_SQL, leg_ids) or ()
1545 )
1546 target_by_leg: Final = MappingProxyType({leg.id: (leg.target_type, leg.target_id) for leg in legs})
1547 by_leg: Final = _ATTEMPT_AGG_ROWS.validate_python(
1548 await _query_raw(prisma_client, _ATTEMPT_AGG_BY_LEG_SQL, leg_ids) or ()
1549 )
1550 verdicts_by_target: Final[Mapping[tuple[str, str], ShadowEvalSlice]] = MappingProxyType(
1551 {
1552 target_by_leg[slice.group]: slice.model_copy(
1553 update={"group": target_by_leg[slice.group][1]} # mutable-ok: pydantic update payload
1554 )
1555 for slice in _slices(by_leg)
1556 }
1557 )
1558 by_router: Final = _ATTEMPT_AGG_ROWS.validate_python(
1559 await _query_raw(prisma_client, _ATTEMPT_AGG_BY_ROUTER_SQL, leg_ids) or ()
1560 )
1561 total_turns: Final = sum(r.turn_count for r in by_tier)
1562 funnel_rows: Final = await _query_raw(prisma_client, _FUNNEL_TOTALS_SQL, leg_ids)
1563 counted: Final = _FunnelTotalsRow.model_validate(funnel_rows[0]) if funnel_rows else None
1564 # Coverage only when EVERY leg has a funnel row: a partial seed (one leg's insert
1565 # failed) must read as unknown, not as job-level counts missing a leg's traffic.
1566 funnel: Final = counted if counted is not None and counted.legs_with_rows == len(leg_ids) else None
1567 result: Final = ShadowEvalResult(
1568 by_tier=_slices(by_tier),
1569 by_current_model=_slices(by_model),
1570 by_router=_slices(by_router),
1571 overall_shadow_win_rate_pct=_pct_of(sum(r.shadow_wins for r in by_tier), total_turns),
1572 overall_tie_rate_pct=_pct_of(sum(r.ties for r in by_tier), total_turns),
1573 sampled_real_spend=sum(r.real_spend for r in by_tier),
1574 sampled_shadow_spend=sum(r.shadow_spend for r in by_tier),
1575 not_sampled_count=funnel.not_sampled if funnel is not None else None,
1576 unjudgeable_count=funnel.unjudgeable if funnel is not None else None,
1577 shed_count=funnel.shed if funnel is not None else None,
1578 withheld_count=funnel.withheld if funnel is not None else None,
1579 )
1580 return result, verdicts_by_target
1583@router.post(
1584 "/auto_router/shadow_eval/start",
1585 tags=("auto router",),
1586 dependencies=(Depends(user_api_key_auth),),
1587 response_model=ShadowEvalJobResponse,
1588 status_code=status.HTTP_201_CREATED,
1589)
1590async def start_shadow_eval(
1591 data: StartShadowEvalRequest,
1592 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
1593) -> ShadowEvalJobResponse:
1594 """
1595 Start a shadow eval: duplicate a sampled slice of one or more targets' live traffic
1596 against a second arm, judge the two responses blind, and stratify win rates by tier,
1597 by the model that served the real arm, and by target.
1599 A target is a virtual key, a team, or a user. Team and user targets match on the
1600 identity every request resolves to at auth time, so they cover JWT-authenticated
1601 traffic, which presents no virtual key; a user target samples that user's traffic
1602 across all their teams, whether it arrives on a JWT or a key they own. models narrows
1603 every target to requests for those model groups, so a user plus one model samples that
1604 user's traffic on that model across every key they own; it is forward-only, since a
1605 reverse job already samples exactly the traffic its own router served.
1607 A forward job answers whether the targets should adopt router_name: it samples the
1608 requests the router did not serve and duplicates them through it. A reverse job
1609 answers whether a target already on the router still gains from it: it samples the
1610 requests the router did serve and duplicates them against baseline_model. A target
1611 can hold one active job per direction, so both questions can run at once, and a
1612 request matching several jobs' targets (say its key and its team) is sampled by
1613 each, separately budgeted.
1615 Shadow responses are never served to users. Each target samples until its recorded
1616 eval spend, the shadow and judge calls' own cost, reaches max_budget dollars, the
1617 job's window ends, or the job is stopped, so one target running out of budget does
1618 not end sampling for the others; sampling changes propagate to pods within about 10
1619 seconds. Shadow and judge calls bill to the sampled request's own identity but are
1620 excluded from request counts and auto-router adoption metrics.
1621 """
1622 from litellm.proxy.proxy_server import llm_router, prisma_client
1624 _require_admin_writer(user_api_key_dict, "start a shadow eval")
1625 if prisma_client is None:
1626 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
1627 unconfigured: Final = tuple(
1628 name
1629 for name in data.router_names
1630 if llm_router is None or not _is_configured_pre_routing_strategy(llm_router, name)
1631 )
1632 if unconfigured:
1633 raise HTTPException(
1634 status_code=400, detail=f"Not a configured auto-router: {', '.join(repr(n) for n in unconfigured)}"
1635 )
1636 token_rows: Final = (
1637 await _verification_tokens(prisma_client).find_many(
1638 where={"token": {"in": list(data.api_key_ids)}} # mutable-ok: Prisma filter
1639 )
1640 if data.api_key_ids
1641 else ()
1642 )
1643 team_rows: Final = (
1644 await _team_rows(prisma_client).find_many(
1645 where={"team_id": {"in": list(data.team_ids)}} # mutable-ok: Prisma filter
1646 )
1647 if data.team_ids
1648 else ()
1649 )
1650 user_rows: Final = (
1651 await _user_rows(prisma_client).find_many(
1652 where={"user_id": {"in": list(data.user_ids)}} # mutable-ok: Prisma filter
1653 )
1654 if data.user_ids
1655 else ()
1656 )
1657 unknown_keys: Final = sorted(frozenset(data.api_key_ids) - frozenset(row.token for row in token_rows or ()))
1658 unknown_teams: Final = sorted(frozenset(data.team_ids) - frozenset(row.team_id for row in team_rows or ()))
1659 unknown_users: Final = sorted(frozenset(data.user_ids) - frozenset(row.user_id for row in user_rows or ()))
1660 unknown_parts: Final = tuple(
1661 part
1662 for part in (
1663 (
1664 f"api_key_ids not on this proxy: {', '.join(unknown_keys)}; pass each key's token hash, "
1665 "the value the key list and key info endpoints report"
1666 )
1667 if unknown_keys
1668 else None,
1669 f"team_ids not on this proxy: {', '.join(unknown_teams)}" if unknown_teams else None,
1670 f"user_ids not on this proxy: {', '.join(unknown_users)}" if unknown_users else None,
1671 )
1672 if part is not None
1673 )
1674 if unknown_parts:
1675 raise HTTPException(status_code=400, detail=". ".join(unknown_parts))
1677 # Every model check below runs once per team the job samples for, since that is the
1678 # identity the shadow and judge calls carry and therefore what the router selects on.
1679 # A user target's traffic can span teams, so it validates unscoped (None); each
1680 # sampled attempt still resolves the judge under its own request's team at eval time.
1681 team_ids: Final = tuple(
1682 dict.fromkeys(
1683 (
1684 *(row.team_id for row in token_rows or ()),
1685 *data.team_ids,
1686 *((None,) if data.user_ids else ()),
1687 )
1688 )
1689 )
1690 _validate_plain_model(llm_router, data.judge_model, "judge_model", team_ids)
1691 if data.baseline_model is not None:
1692 _validate_plain_model(llm_router, data.baseline_model, "baseline_model", team_ids)
1693 _validate_judge_is_not_a_candidate(llm_router, data, team_ids)
1694 _validate_model_scope(llm_router, data.models)
1696 requested_targets: Final[tuple[tuple[ShadowEvalTargetType, str], ...]] = (
1697 *(("key", key) for key in data.api_key_ids),
1698 *(("team", team) for team in data.team_ids),
1699 *(("user", user) for user in data.user_ids),
1700 )
1701 requested_by_type: Final[tuple[tuple[ShadowEvalTargetType, tuple[str, ...]], ...]] = tuple(
1702 (target_type, ids)
1703 for target_type, ids in (("key", data.api_key_ids), ("team", data.team_ids), ("user", data.user_ids))
1704 if ids
1705 )
1706 # A job whose window passed or whose budget ran out stopped sampling on its own,
1707 # but its legs still hold their slots in the per-target, per-direction partial unique
1708 # index until stamped; free them so a new eval can start. Sweeping both directions is
1709 # deliberate. Sweep and claim filter on exact (target_type, id) pairs so a team id
1710 # that happens to equal a key hash never matches the other kind's slot.
1711 for target_type, ids in requested_by_type:
1712 await prisma_client.db.execute_raw(_SWEEP_FINISHED_JOBS_SQL, list(ids), target_type) # mutable-ok: query param
1713 claimed: Final = await _shadow_eval_jobs(prisma_client).find_many(
1714 where={ # mutable-ok: Prisma filter
1715 "OR": [ # mutable-ok: Prisma filter
1716 {"target_type": target_type, "target_id": {"in": list(ids)}} # mutable-ok: Prisma filter
1717 for target_type, ids in requested_by_type
1718 ],
1719 "direction": data.direction,
1720 "stopped_at": None,
1721 },
1722 )
1723 if claimed:
1724 raise HTTPException(
1725 status_code=409,
1726 detail=(
1727 f"Already in an active {data.direction} shadow eval job: "
1728 + ", ".join(sorted(f"{row.target_type} {row.target_id} (job {row.group_id})" for row in claimed))
1729 + ". Stop it first."
1730 ),
1731 )
1732 now: Final = datetime.now(timezone.utc)
1733 group_id: Final = str(uuid4())
1734 ends_at: Final = now + timedelta(days=data.duration_days)
1735 shared_config: Final = { # mutable-ok: Prisma payload
1736 "group_id": group_id,
1737 # a pre-router_names pod samples router_name alone, so it must be a real arm
1738 "router_name": data.router_names[0],
1739 "router_names": list(data.router_names), # mutable-ok: Prisma payload
1740 "models": list(data.models), # mutable-ok: Prisma payload
1741 "direction": data.direction,
1742 "baseline_model": data.baseline_model,
1743 "judge_model": data.judge_model,
1744 "shadow_percentage": data.shadow_percentage,
1745 "max_turns": SHADOW_EVAL_TURN_VALVE,
1746 "max_budget": data.max_budget,
1747 "created_by": user_api_key_dict.user_id,
1748 "created_at": now,
1749 "ends_at": ends_at,
1750 }
1751 try:
1752 # Leg ids are minted here rather than by the DB default so the funnel seed below
1753 # writes from the same values with no read-back, which a lagging read replica
1754 # (DATABASE_URL_READ_REPLICA) could otherwise return empty.
1755 leg_ids: Final = tuple(str(uuid4()) for _ in requested_targets)
1756 await _shadow_eval_jobs(prisma_client).create_many(
1757 data=[ # mutable-ok: Prisma payload
1758 { # mutable-ok: Prisma payload
1759 **shared_config,
1760 "id": leg_id,
1761 "target_type": target_type,
1762 "target_id": target_id,
1763 }
1764 for leg_id, (target_type, target_id) in zip(leg_ids, requested_targets)
1765 ]
1766 )
1767 except Exception as e:
1768 if not _is_unique_violation(e):
1769 raise
1770 raise HTTPException(
1771 status_code=409,
1772 detail=(
1773 f"A requested target was claimed by another {data.direction} shadow eval job concurrently. "
1774 "Stop it first."
1775 ),
1776 ) from e
1777 # Seed a zero funnel row per leg NOW: a fully covered job never skips a request, so
1778 # waiting for the first skip would leave it indistinguishable from a pre-funnel job
1779 # (null coverage). A failed seed degrades this job to exactly that, nothing worse.
1780 try:
1781 await _shadow_eval_funnel(prisma_client).create_many(
1782 data=[{"job_id": leg_id} for leg_id in leg_ids], # mutable-ok: Prisma payload
1783 skip_duplicates=True,
1784 )
1785 except Exception as seed_err: # noqa: BLE001 # coverage is advisory; the job must still start
1786 verbose_proxy_logger.error("shadow_eval: funnel seed failed for job %s: %s", group_id, seed_err)
1787 labels: Final = _target_labels(token_rows or (), team_rows or (), user_rows or ())
1788 return ShadowEvalJobResponse(
1789 job_id=group_id,
1790 targets=tuple(
1791 ShadowEvalJobTargetResponse(
1792 target_type=target_type,
1793 target_id=target_id,
1794 max_turns=SHADOW_EVAL_TURN_VALVE,
1795 max_budget=data.max_budget,
1796 target_alias=labels.get((target_type, target_id), _NO_TARGET_LABELS)[0],
1797 key_name=labels.get((target_type, target_id), _NO_TARGET_LABELS)[1],
1798 )
1799 for target_type, target_id in sorted(requested_targets)
1800 ),
1801 router_names=data.router_names,
1802 models=data.models,
1803 direction=data.direction,
1804 baseline_model=data.baseline_model,
1805 judge_model=data.judge_model,
1806 shadow_percentage=data.shadow_percentage,
1807 created_at=now,
1808 ends_at=ends_at,
1809 )
1812@router.get(
1813 "/auto_router/shadow_eval",
1814 tags=("auto router",),
1815 dependencies=(Depends(user_api_key_auth),),
1816 response_model=list[ShadowEvalJobResponse],
1817)
1818async def list_shadow_eval_jobs(
1819 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
1820 target_type: Annotated[
1821 ShadowEvalTargetType | None, Query(description="Kind of target to filter on; requires target_id")
1822 ] = None,
1823 target_id: Annotated[
1824 str | None, Query(description="Filter to jobs that shadow this target, alone or alongside others")
1825 ] = None,
1826 limit: Annotated[int, Query(ge=1, le=200, description="Newest jobs to return")] = 50,
1827) -> tuple[ShadowEvalJobResponse, ...]:
1828 """List shadow eval jobs, newest first, each target with its attempt count so status
1829 is accurate. Judged counts, spend, and results ride the detail endpoint only."""
1830 from litellm.proxy.proxy_server import prisma_client
1832 _require_admin_viewer(user_api_key_dict, "view shadow evals")
1833 if prisma_client is None: 1833 ↛ 1834line 1833 didn't jump to line 1834 because the condition on line 1833 was never true
1834 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
1835 filter_type: Final = target_type if isinstance(target_type, str) else None
1836 filter_id: Final = target_id if isinstance(target_id, str) else None
1837 if (filter_type is None) != (filter_id is None):
1838 raise HTTPException(status_code=400, detail="target_type and target_id filter together; pass both or neither")
1839 legs: Final = _LEG_ROWS.validate_python(
1840 (
1841 await _query_raw(prisma_client, _LIST_LEGS_BY_TARGET_SQL, limit, filter_type, filter_id)
1842 if filter_type and filter_id
1843 else await _query_raw(prisma_client, _LIST_LEGS_SQL, limit)
1844 )
1845 or ()
1846 )
1847 by_group: Final[Mapping[str, tuple[_LegRow, ...]]] = MappingProxyType(
1848 {group_id: tuple(group) for group_id, group in groupby(sorted(legs, key=_leg_group_id), key=_leg_group_id)}
1849 )
1850 newest_first: Final = sorted(
1851 by_group, key=lambda group_id: max(leg.created_at for leg in by_group[group_id]), reverse=True
1852 )
1853 counts: Final = await _leg_attempt_counts(prisma_client, legs)
1854 return await _with_target_labels(
1855 prisma_client, tuple(_group_response(group_id, by_group[group_id], counts) for group_id in newest_first)
1856 )
1859@router.get(
1860 "/auto_router/shadow_eval/{job_id}",
1861 tags=("auto router",),
1862 dependencies=(Depends(user_api_key_auth),),
1863 response_model=ShadowEvalJobResponse,
1864)
1865async def get_shadow_eval_job(
1866 job_id: str,
1867 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
1868) -> ShadowEvalJobResponse:
1869 """One job with derived counts, judge spend, latest error, and stratified results."""
1870 from litellm.proxy.proxy_server import prisma_client
1872 _require_admin_viewer(user_api_key_dict, "view shadow evals")
1873 if prisma_client is None: 1873 ↛ 1874line 1873 didn't jump to line 1874 because the condition on line 1873 was never true
1874 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
1875 legs: Final = _LEG_ROWS.validate_python(
1876 await _shadow_eval_jobs(prisma_client).find_many(
1877 where={"group_id": job_id} # mutable-ok: Prisma filter
1878 )
1879 or ()
1880 )
1881 if not legs: 1881 ↛ 1883line 1881 didn't jump to line 1883 because the condition on line 1881 was always true
1882 raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}")
1883 leg_ids: Final = [leg.id for leg in legs] # mutable-ok: query param
1884 totals: Final = _ATTEMPT_TOTALS_ROWS.validate_python(
1885 await _query_raw(prisma_client, _ATTEMPT_TOTALS_SQL, leg_ids) or ()
1886 )
1887 latest_error: Final = await _shadow_eval_attempts(prisma_client).find_first(
1888 where={"job_id": {"in": leg_ids}, "outcome": "error"}, # mutable-ok: Prisma filter
1889 order={"created_at": "desc"}, # mutable-ok: Prisma order
1890 )
1891 labeled: Final = await _with_target_labels(
1892 prisma_client, (_group_response(job_id, legs, await _leg_attempt_counts(prisma_client, legs)),)
1893 )
1894 results, verdicts_by_target = await _shadow_eval_results(prisma_client, legs)
1895 return labeled[0].model_copy(
1896 update={ # mutable-ok: pydantic update payload
1897 "judged_count": totals[0].judged_count if totals else 0,
1898 "error_count": totals[0].error_count if totals else 0,
1899 "judge_spend": round(totals[0].judge_spend, 6) if totals else 0.0,
1900 "last_error": latest_error.error if latest_error else None,
1901 "results": results,
1902 "targets": tuple(
1903 target.model_copy(
1904 update={ # mutable-ok: pydantic update payload
1905 "verdicts": verdicts_by_target.get((target.target_type, target.target_id))
1906 }
1907 )
1908 for target in labeled[0].targets
1909 ),
1910 }
1911 )
1914@router.post(
1915 "/auto_router/shadow_eval/{job_id}/stop",
1916 tags=("auto router",),
1917 dependencies=(Depends(user_api_key_auth),),
1918 response_model=ShadowEvalJobResponse,
1919)
1920async def stop_shadow_eval_job(
1921 job_id: str,
1922 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
1923) -> ShadowEvalJobResponse:
1924 """Stop an active shadow eval job, every target it scopes at once. Attempts are kept;
1925 sampling halts within ~10s. Targets that already stopped on their own budget keep the
1926 stopped_at they earned. The statement is the whole state machine: it claims the job
1927 only while a leg still samples inside the window with no stop recorded, so a racing
1928 operator, a same-instant budget spend, and a repeat stop all read the same 400 with
1929 the status the job actually holds."""
1930 from litellm.proxy.proxy_server import prisma_client
1932 _require_admin_writer(user_api_key_dict, "stop a shadow eval")
1933 if prisma_client is None: 1933 ↛ 1934line 1933 didn't jump to line 1934 because the condition on line 1933 was never true
1934 raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
1935 stamp: Final = datetime.now(timezone.utc)
1936 operator: Final = user_api_key_dict.user_id or "operator"
1937 claimed: Final = await prisma_client.db.execute_raw(
1938 _STOP_JOB_SQL, job_id, operator, stamp.replace(tzinfo=None).isoformat()
1939 )
1940 legs: Final = _LEG_ROWS.validate_python(
1941 await _shadow_eval_jobs(prisma_client).find_many(
1942 where={"group_id": job_id} # mutable-ok: Prisma filter
1943 )
1944 or ()
1945 )
1946 if not legs: 1946 ↛ 1948line 1946 didn't jump to line 1948 because the condition on line 1946 was always true
1947 raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}")
1948 counts: Final = await _leg_attempt_counts(prisma_client, legs)
1949 current: Final = _group_response(job_id, legs, counts)
1950 if claimed == 0:
1951 raise HTTPException(status_code=400, detail=f"Job {job_id} is already {current.status}")
1952 labeled: Final = await _with_target_labels(prisma_client, (current,))
1953 return labeled[0]