scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-30 06:43:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 249

fix(router): read rpm through a typed mapping so the skip stays type-clean

Browse source
This commit is contained in:
TobiMayr 2026-10-14 12:58:50 +01:00
parent 01e81369bb
commit 6251142e38
1 changed files with 18 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
litellm/router.py
View file

@ -14022,6 +14022,16 @@ class Router:
if param in _PRE_CALL_CHECK_FILTERED_PARAMS and not isinstance(value, str)
)
def _deployment_configures_rpm(self, deployment: object) -> bool:
if not isinstance(deployment, dict):
return False
fields: Final[Mapping[str, object]] = cast(Mapping[str, object], deployment) # cast-ok: untyped dict
params: Final = fields.get("litellm_params")
if not isinstance(params, dict):
return False
parsed_params: Final[Mapping[str, object]] = cast(Mapping[str, object], params) # cast-ok: untyped dict
return parsed_params.get("rpm") is not None
def _pre_call_checks(
self,
model: str,
@ -14080,18 +14090,19 @@ class Router:
# Safe to check statically since `rpm` is a direct litellm_params setting, unlike
# max_input_tokens which get_router_model_info() can also derive from the cost map.
any_deployment_has_rpm: Final = self.routing_strategy != "usage-based-routing-v2" and any(
isinstance(litellm_params := d.get("litellm_params"), dict) and litellm_params.get("rpm") is not None
for d in _returned_deployments
self._deployment_configures_rpm(
deployment # pyright: ignore[reportUnknownArgumentType] # untyped deployment list
)
for deployment in _returned_deployments # pyright: ignore[reportUnknownVariableType] # untyped list
)
## get model group RPM ##
rpm_key: Final = f"{model}:rpm:{get_utc_datetime().strftime('%H-%M')}"
cached_model_group: Final = (
self.cache.get_cache(key=rpm_key, local_only=True, parent_otel_span=parent_otel_span)
model_group_cache: Final = (
(self.cache.get_cache(key=rpm_key, local_only=True, parent_otel_span=parent_otel_span) or {})
if any_deployment_has_rpm
else None
else {}
)
model_group_cache: Final = cached_model_group if isinstance(cached_model_group, dict) else {}
for idx, deployment in enumerate(_returned_deployments):
# Cache nested dict access to avoid repeated temporary dict allocations
_litellm_params = deployment.get("litellm_params", {})
@ -14115,7 +14126,7 @@ class Router:
continue
## RPM CHECK ##
if any_deployment_has_rpm:
if any_deployment_has_rpm and isinstance(model_group_cache, dict):
model_id = _model_info.get("id", "")
### get local router cache ###
current_request_cache_local = (