scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-11 03:38:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 82

fix(complexity-router): read the Nadir verdict through a TypeAdapter

Browse source 
CodeQL flagged classify() for mixing explicit and implicit returns: it
does not treat the trailing `case _` arm of the verdict match as
exhaustive. Dropping that arm is not an option either, since basedpyright
then reports reportMatchNotExhaustive, a rule budgeted at zero.

The body is now validated as a Mapping[str, object] and the bucket read
from it, the boundary pattern the repo already uses, so every path returns
explicitly and the Any on the match subject is gone. Behavior is
unchanged: a non-mapping body, a missing bucket or a non-string bucket
still declines, which the existing unusable-verdict cases cover.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
dor 2026-09-25 09:29:55 -04:00
parent 6789b2c028
commit 51492b3c62
1 changed files with 8 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
litellm/router_strategy/complexity_router/nadir_classifier.py
View file

@ -53,6 +53,8 @@ from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
from pydantic import TypeAdapter, ValidationError
from litellm.constants import NADIR_DEFAULT_API_BASE
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, get_async_httpx_client
from litellm.secret_managers.main import get_secret_str
@ -64,6 +66,7 @@ DEFAULT_TIER_MAP: Final[Mapping[str, str]] = MappingProxyType(
)
_API_PATH: Final = "/v1/bucket"
_VERDICT: Final = TypeAdapter(Mapping[str, object])
def _bucket_url(api_base: str) -> str:
@ -134,11 +137,11 @@ class NadirComplexityClassifier:
client: Final = self._client or get_async_httpx_client(llm_provider=httpxSpecialProvider.ComplexityClassifier)
body: Final = {"messages": list(messages), "source": "litellm"} # mutable-ok: one request body
response: Final = await client.post(url=url, json=body, headers=self._headers(url))
match response.json():
case {"bucket": str(bucket)}:
return self._tier_map.get(bucket.strip().lower())
case _:
return None
try:
bucket: Final = _VERDICT.validate_python(response.json()).get("bucket")
except ValidationError:
return None
return self._tier_map.get(bucket.strip().lower()) if isinstance(bucket, str) else None
nadir_classifier: Final = NadirComplexityClassifier()