scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-07 07:58:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 150

feat: add OpenRouter-style -latest model aliases for first-party providers (#50315)

Browse source 
* Add first-party floating model aliases

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Align Moonshot alias parameter test with its target

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Share floating alias proxy budgets and rate limits with targets

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Prove target-only team budgets count and block floating aliases

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Share floating alias quotas with project and organization targets

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Coalesce colliding floating alias rate limit counters

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Fix latest alias Rust fallback and failure accounting

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: adapt latest alias checks to typed rate-limit store

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
moe-berri 2026-11-06 18:22:38 +08:00 • committed by GitHub
parent 80194a7d70
commit b02bb99b08
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 1076 additions and 104 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
litellm/litellm_core_utils/get_llm_provider_logic.py
View file

@ -12,6 +12,7 @@ from litellm.constants import (
from litellm.litellm_core_utils.fallback_generalizations import (
match_routing_generalization,
)
from litellm.litellm_core_utils.model_latest_aliases import resolve_model_latest_alias
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
from litellm.llms.opencode.common_utils import is_opencode_api_base
from litellm.secret_managers.main import get_secret, get_secret_str
@ -189,6 +190,23 @@ def get_llm_provider(
api_base: str | None = None,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | None = None,
) -> tuple[str, str, str | None, str | None]:
resolved_model, provider, dynamic_api_key, resolved_api_base = _get_llm_provider(
model, custom_llm_provider, api_base, api_key, litellm_params
)
explicit_provider: Final = custom_llm_provider or (litellm_params.custom_llm_provider if litellm_params else None)
target: Final = (
resolve_model_latest_alias(resolved_model, provider) if explicit_provider or "/" in model else resolved_model
)
return target, provider, dynamic_api_key, resolved_api_base
def _get_llm_provider(
model: str,
custom_llm_provider: str | None = None,
api_base: str | None = None,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | None = None,
) -> tuple[str, str, str | None, str | None]:
"""
Returns the provider for a given model name - e.g. 'azure/chatgpt-v-2' -> 'azure'

33
litellm/litellm_core_utils/model_latest_aliases.py Normal file
View file

@ -0,0 +1,33 @@
from collections.abc import Mapping
from functools import cache
from importlib.resources import files
from types import MappingProxyType
from typing import Final
from pydantic import TypeAdapter
_ALIASES_ADAPTER: Final = TypeAdapter(dict[str, dict[str, str]])
@cache
def get_model_latest_aliases() -> Mapping[str, Mapping[str, str]]:
aliases: Final = _ALIASES_ADAPTER.validate_json(files("litellm").joinpath("model_latest_aliases.json").read_bytes())
return MappingProxyType({provider: MappingProxyType(models) for provider, models in aliases.items()})
def resolve_model_latest_alias(model: str, provider: str) -> str:
aliases: Final = get_model_latest_aliases().get(provider)
return aliases.get(model, model) if aliases is not None else model
def expand_model_latest_alias(model: str) -> tuple[str, ...]:
provider, separator, name = model.partition("/")
target: Final = resolve_model_latest_alias(name, provider) if separator else name
return (model, f"{provider}/{target}") if separator and target != name else (model,)
def is_model_latest_alias(model: str, custom_llm_provider: str | None = None) -> bool:
prefix, separator, name = model.partition("/")
provider: Final = custom_llm_provider or prefix
candidate: Final = name if separator and prefix == provider else model
return bool(custom_llm_provider or separator) and resolve_model_latest_alias(candidate, provider) != candidate

21
litellm/model_latest_aliases.json Normal file
View file

@ -0,0 +1,21 @@
{
"openai": {
"gpt-luna-latest": "gpt-6-luna",
"gpt-sol-latest": "gpt-6.1-sol",
"gpt-terra-latest": "gpt-6.1-terra",
"gpt-astra-latest": "gpt-6-astra",
"gpt-mini-latest": "gpt-5.4-mini"
},
"anthropic": {
"claude-opus-latest": "claude-opus-5-5",
"claude-sonnet-latest": "claude-sonnet-5-5",
"claude-haiku-latest": "claude-haiku-5-5",
"claude-fable-latest": "claude-fable-5-1"
},
"xai": {"grok-latest": "grok-4.7"},
"moonshot": {"kimi-latest": "kimi-k3"},
"zai": {"glm-latest": "glm-5.3", "glm-flash-latest": "glm-5.3-flash"},
"deepseek": {"deepseek-pro-latest": "deepseek-v4-pro", "deepseek-flash-latest": "deepseek-v4-flash"},
"dashscope": {"qwen-max-latest": "qwen3.8-max", "qwen-flash-latest": "qwen3.8-flash"},
"minimax": {"minimax-latest": "MiniMax-M3"}
}

85
litellm/proxy/hooks/model_max_budget_limiter.py
View file

@ -27,6 +27,7 @@ from litellm.litellm_core_utils.batch_utils import (
)
from litellm.litellm_core_utils.duration_parser import duration_in_seconds
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
from litellm.litellm_core_utils.model_latest_aliases import expand_model_latest_alias
from litellm.llms.bedrock.common_utils import get_bedrock_base_model, get_bedrock_cross_region_inference_regions
from litellm.proxy._types import Litellm_EntityType, UserAPIKeyAuth
from litellm.proxy.hooks.model_budget_shared_windows import (
@ -523,6 +524,25 @@ def _usable_entry(entry: str, raw_budget_config: object) -> ResolvedModelBudget
def team_model_budget_entries(
model: str, team_model_max_budget: Mapping[str, object], key_model_max_budget: Mapping[str, object] | None
) -> Mapping[str, object]:
models: Final = expand_model_latest_alias(model)
if len(models) == 1:
return _single_team_model_budget_entries(model, team_model_max_budget, key_model_max_budget)
entries: Final = tuple(
_single_team_model_budget_entries(candidate, team_model_max_budget, key_model_max_budget)
for candidate in models
)
return MappingProxyType(
{
entry: config
for entry, config in team_model_max_budget.items()
if all(entry in mapping for mapping in entries)
}
)
def _single_team_model_budget_entries(
model: str, team_model_max_budget: Mapping[str, object], key_model_max_budget: Mapping[str, object] | None
) -> Mapping[str, object]:
"""The team entries that budget `model` for this key.
@ -780,6 +800,31 @@ def _as_float(value: object) -> float | None:
return None
def _entity_model_budget_entries(
model: str,
entity_budget: tuple[Litellm_EntityType, str | None, object],
model_access_groups: ModelAccessGroups,
team_id: str | None,
) -> tuple[tuple[Litellm_EntityType, str, ResolvedModelBudget], ...]:
entity_type, entity_id, model_max_budget = entity_budget
if entity_id is None or not isinstance(model_max_budget, Mapping) or not model_max_budget:
return ()
return tuple(
(entity_type, entity_id, resolved)
for candidate in expand_model_latest_alias(model)
if (
resolved := resolve_model_budget(
model=candidate,
model_max_budget=model_max_budget,
model_access_groups=model_access_groups,
team_id=team_id,
)
)
is not None
and resolved.budget_config.budget_duration is not None
)
def _resolve_entity_model_budgets(
model: str,
entity_budgets: Iterable[tuple[Litellm_EntityType, str | None, object]],
@ -787,19 +832,15 @@ def _resolve_entity_model_budgets(
team_id: str | None,
) -> tuple[tuple[Litellm_EntityType, str, ResolvedModelBudget], ...]:
"""Drop the scopes that do not budget `model`, keeping only what can be incremented."""
entries: Final = itertools.chain.from_iterable(
_entity_model_budget_entries(model, entity_budget, model_access_groups, team_id)
for entity_budget in entity_budgets
)
return tuple(
(entity_type, entity_id, resolved)
for entity_type, entity_id, model_max_budget in entity_budgets
if entity_id is not None and isinstance(model_max_budget, Mapping) and model_max_budget
for resolved in (
resolve_model_budget(
model=model,
model_max_budget=model_max_budget,
model_access_groups=model_access_groups,
team_id=team_id,
),
)
if resolved is not None and resolved.budget_config.budget_duration is not None
{
(entity_type, entity_id, resolved.budget_model): (entity_type, entity_id, resolved)
for entity_type, entity_id, resolved in entries
}.values()
)
@ -996,6 +1037,26 @@ class _PROXY_VirtualKeyModelMaxBudgetLimiter(RouterBudgetLimiting):
model_max_budget: Mapping[str, object] | None,
model: str,
exceeded_message: str,
) -> bool:
for candidate in expand_model_latest_alias(model):
await self._is_single_entity_within_model_budget(
entity_type=entity_type,
entity_id=entity_id,
team_id=team_id,
model_max_budget=model_max_budget,
model=candidate,
exceeded_message=exceeded_message,
)
return True
async def _is_single_entity_within_model_budget(
self,
entity_type: Litellm_EntityType,
entity_id: str | None,
team_id: str | None,
model_max_budget: Mapping[str, object] | None,
model: str,
exceeded_message: str,
) -> bool:
if not model_max_budget:
return True

257
litellm/proxy/hooks/parallel_request_limiter_v3.py
View file

@ -50,6 +50,7 @@ from litellm.litellm_core_utils.add_retry_fallback_headers import (
response_has_hidden_params,
)
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.model_latest_aliases import expand_model_latest_alias
from litellm.litellm_core_utils.prompt_templates.common_utils import (
get_str_from_messages,
)
@ -74,6 +75,7 @@ from litellm.proxy.auth.auth_utils import (
get_model_rate_limit_from_metadata,
)
from litellm.proxy.auth.budget_throttle import throttled_limit
from litellm.proxy.common_utils.callback_utils import get_model_group_from_litellm_kwargs
from litellm.proxy.common_utils.http_parsing_utils import get_tags_from_request_body
from litellm.proxy.common_utils.proxy_rate_limit_error import (
ProxyRateLimitError,
@ -348,6 +350,44 @@ class RateLimitDescriptor(TypedDict):
rate_limit: RateLimitDescriptorRateLimitObject | None
def _minimum_descriptor_limit(
descriptors: Sequence[RateLimitDescriptor],
metric: Literal["requests_per_unit", "tokens_per_unit", "max_parallel_requests"],
) -> int | None:
return min(
(limit for descriptor in descriptors if (limit := (descriptor["rate_limit"] or {}).get(metric)) is not None),
default=None,
)
def _merge_rate_limit_descriptors(descriptors: Sequence[RateLimitDescriptor]) -> RateLimitDescriptor:
first: Final = descriptors[0]
if len(descriptors) == 1:
return first
return RateLimitDescriptor(
key=first["key"],
value=first["value"],
rate_limit=RateLimitDescriptorRateLimitObject(
requests_per_unit=_minimum_descriptor_limit(descriptors, "requests_per_unit"),
tokens_per_unit=_minimum_descriptor_limit(descriptors, "tokens_per_unit"),
max_parallel_requests=_minimum_descriptor_limit(descriptors, "max_parallel_requests"),
window_size=(first["rate_limit"] or {}).get("window_size"),
),
)
def _coalesce_rate_limit_descriptors(descriptors: Sequence[RateLimitDescriptor]) -> Sequence[RateLimitDescriptor]:
identities: Final = tuple(dict.fromkeys((descriptor["key"], descriptor["value"]) for descriptor in descriptors))
if len(identities) == len(descriptors):
return descriptors
return tuple(
_merge_rate_limit_descriptors(
tuple(descriptor for descriptor in descriptors if (descriptor["key"], descriptor["value"]) == identity)
)
for identity in identities
)
class ParallelRequestGauge(TypedDict):
scope: RateLimitScope
limit: int
@ -2405,6 +2445,18 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
group=self._model_group_resolver(requested_model) or requested_model,
)
def _rate_limited_models(self, model: RateLimitedModel | None) -> tuple[RateLimitedModel, ...]:
if model is None:
return ()
return (
model,
*(
target
for candidate in expand_model_latest_alias(model.requested)[1:]
if (target := self._rate_limited_model(candidate)) is not None
),
)
def create_organization_rate_limit_descriptor(
self, user_api_key_dict: UserAPIKeyAuth, requested_model: str | None = None
) -> list[RateLimitDescriptor]:
@ -2426,9 +2478,16 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
)
)
model: Final = self._rate_limited_model(requested_model)
if model is None:
return descriptors
model_descriptors: Final = tuple(
descriptor
for model in self._rate_limited_models(self._rate_limited_model(requested_model))
if (descriptor := self._organization_model_rate_limit_descriptor(user_api_key_dict, model)) is not None
)
return [*descriptors, *_coalesce_rate_limit_descriptors(model_descriptors)]
def _organization_model_rate_limit_descriptor(
self, user_api_key_dict: UserAPIKeyAuth, model: RateLimitedModel
) -> RateLimitDescriptor | None:
model_specific_tpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "organization_metadata", "model_tpm_limit")
)
@ -2436,19 +2495,16 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
get_model_rate_limit_from_metadata(user_api_key_dict, "organization_metadata", "model_rpm_limit")
)
if model_specific_tpm_limit is None and model_specific_rpm_limit is None:
return descriptors
descriptors.append(
RateLimitDescriptor(
key="model_per_organization",
value=f"{user_api_key_dict.org_id}:{model.group}",
rate_limit={
"requests_per_unit": model_specific_rpm_limit,
"tokens_per_unit": model_specific_tpm_limit,
"window_size": self.window_size,
},
)
return None
return RateLimitDescriptor(
key="model_per_organization",
value=f"{user_api_key_dict.org_id}:{model.group}",
rate_limit={
"requests_per_unit": model_specific_rpm_limit,
"tokens_per_unit": model_specific_tpm_limit,
"window_size": self.window_size,
},
)
return descriptors
def _add_model_per_key_rate_limit_descriptor(
self,
@ -2935,11 +2991,13 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
# Model rate limits
requested_model: Final = data.get("model", None)
self._add_model_per_key_rate_limit_descriptor(
user_api_key_dict=user_api_key_dict,
requested_model=requested_model,
descriptors=descriptors,
)
model_candidates: Final = expand_model_latest_alias(requested_model if isinstance(requested_model, str) else "")
for candidate in model_candidates:
self._add_model_per_key_rate_limit_descriptor(
user_api_key_dict=user_api_key_dict,
requested_model=candidate,
descriptors=descriptors,
)
# Per-request-tag rate limits scoped to this key
self._add_tag_per_key_rate_limit_descriptor(
@ -2948,11 +3006,12 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
descriptors=descriptors,
)
self._add_team_model_rate_limit_descriptor_from_metadata(
user_api_key_dict=user_api_key_dict,
requested_model=requested_model if isinstance(requested_model, str) else None,
descriptors=descriptors,
)
for candidate in model_candidates:
self._add_team_model_rate_limit_descriptor_from_metadata(
user_api_key_dict=user_api_key_dict,
requested_model=candidate,
descriptors=descriptors,
)
self._add_team_ptu_rate_limit_descriptor(
user_api_key_dict=user_api_key_dict,
requested_model=requested_model if isinstance(requested_model, str) else None,
@ -2975,7 +3034,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
)
)
return descriptors
return descriptors if len(model_candidates) == 1 else list(_coalesce_rate_limit_descriptors(descriptors))
_create_rate_limit_descriptors = create_rate_limit_descriptors
@ -3126,12 +3185,26 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
self,
user_api_key_dict: UserAPIKeyAuth,
requested_model: str | None,
descriptors: list[RateLimitDescriptor],
descriptors: _RateLimitDescriptorSink,
) -> None:
candidates: Final = expand_model_latest_alias(requested_model) if requested_model else (None,)
model_descriptors: Final = tuple(
descriptor
for candidate in candidates
if (descriptor := self._project_model_rate_limit_descriptor(user_api_key_dict, candidate)) is not None
)
for descriptor in _coalesce_rate_limit_descriptors(model_descriptors):
descriptors.append(descriptor)
def _project_model_rate_limit_descriptor(
self,
user_api_key_dict: UserAPIKeyAuth,
requested_model: str | None,
) -> RateLimitDescriptor | None:
"""Add project model rate limit descriptor from project_metadata if applicable."""
model: Final = self._rate_limited_model(requested_model)
if model is None:
return
return None
model_specific_tpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_tpm_limit")
)
@ -3139,17 +3212,15 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_rpm_limit")
)
if model_specific_tpm_limit is None and model_specific_rpm_limit is None:
return
descriptors.append(
RateLimitDescriptor(
key="model_per_project",
value=f"{user_api_key_dict.project_id}:{model.group}",
rate_limit={
"requests_per_unit": model_specific_rpm_limit,
"tokens_per_unit": model_specific_tpm_limit,
"window_size": self.window_size,
},
)
return None
return RateLimitDescriptor(
key="model_per_project",
value=f"{user_api_key_dict.project_id}:{model.group}",
rate_limit={
"requests_per_unit": model_specific_rpm_limit,
"tokens_per_unit": model_specific_tpm_limit,
"window_size": self.window_size,
},
)
def add_project_io_token_rate_limit_descriptors_from_metadata(
@ -3158,6 +3229,20 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
requested_model: str | None,
descriptors: _RateLimitDescriptorSink,
) -> None:
candidates: Final = expand_model_latest_alias(requested_model) if requested_model else (None,)
model_descriptors: Final = tuple(
itertools.chain.from_iterable(
self._project_io_token_rate_limit_descriptors(user_api_key_dict, candidate) for candidate in candidates
)
)
for descriptor in _coalesce_rate_limit_descriptors(model_descriptors):
descriptors.append(descriptor)
def _project_io_token_rate_limit_descriptors(
self,
user_api_key_dict: UserAPIKeyAuth,
requested_model: str | None,
) -> tuple[RateLimitDescriptor, ...]:
"""Add project-scoped ITPM/OTPM descriptors from project_metadata.
Enforced independently of, and alongside, the combined ``model_per_project``
@ -3166,7 +3251,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
"""
model: Final = self._rate_limited_model(requested_model)
if model is None or user_api_key_dict.project_id is None:
return
return ()
model_itpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_itpm_limit")
@ -3175,34 +3260,23 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_otpm_limit")
)
if model_itpm_limit is None and model_otpm_limit is None:
return
descriptor_value: Final = f"{user_api_key_dict.project_id}:{model.group}"
if model_itpm_limit is not None:
descriptors.append(
RateLimitDescriptor(
key=PROJECT_ITPM_DESCRIPTOR_KEY,
value=descriptor_value,
rate_limit={
"requests_per_unit": None,
"tokens_per_unit": model_itpm_limit,
"window_size": self.window_size,
},
)
return tuple(
RateLimitDescriptor(
key=key,
value=descriptor_value,
rate_limit={
"requests_per_unit": None,
"tokens_per_unit": limit,
"window_size": self.window_size,
},
)
if model_otpm_limit is not None:
descriptors.append(
RateLimitDescriptor(
key=PROJECT_OTPM_DESCRIPTOR_KEY,
value=descriptor_value,
rate_limit={
"requests_per_unit": None,
"tokens_per_unit": model_otpm_limit,
"window_size": self.window_size,
},
)
for key, limit in (
(PROJECT_ITPM_DESCRIPTOR_KEY, model_itpm_limit),
(PROJECT_OTPM_DESCRIPTOR_KEY, model_otpm_limit),
)
if limit is not None
)
def _handle_rate_limit_error(
self,
@ -4416,6 +4490,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
kwargs: object,
model_group: str | None,
tpm_limited_tags: Set[str] = frozenset(),
model_groups: tuple[str, ...] = (),
) -> list[tuple[str, str]]:
"""
Enumerate every (scope_key, scope_value) pair that *might* carry a
@ -4449,24 +4524,25 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
if user_api_key_organization_id:
targets.append(("organization", user_api_key_organization_id))
if model_group:
if user_api_key:
targets.append(("model_per_key", f"{user_api_key}:{model_group}"))
if user_api_key_team_id:
targets.append(("model_per_team", f"{user_api_key_team_id}:{model_group}"))
if user_api_key_organization_id:
targets.append(
(
"model_per_organization",
f"{user_api_key_organization_id}:{model_group}",
for group in dict.fromkeys(model_groups or (model_group,)):
if user_api_key:
targets.append(("model_per_key", f"{user_api_key}:{group}"))
if user_api_key_team_id:
targets.append(("model_per_team", f"{user_api_key_team_id}:{group}"))
if user_api_key_organization_id:
targets.append(
(
"model_per_organization",
f"{user_api_key_organization_id}:{group}",
)
)
)
if user_api_key_project_id:
targets.append(
(
"model_per_project",
f"{user_api_key_project_id}:{model_group}",
if user_api_key_project_id:
targets.append(
(
"model_per_project",
f"{user_api_key_project_id}:{group}",
)
)
)
if agent_id:
targets.append(("agent", agent_id))
if session_id:
@ -4565,9 +4641,6 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
) -> list[ReservationOperation]:
"""Build Redis pipeline increment ops for TPM / parallel-request counters."""
from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs
from litellm.proxy.common_utils.callback_utils import (
get_model_group_from_litellm_kwargs,
)
# Get metadata from standard_logging_object - this correctly handles both
# 'metadata' and 'litellm_metadata' fields from litellm_params
@ -4617,6 +4690,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
# mid-flight cannot move the charge) is authoritative; otherwise fall
# back to the router's ``model_group`` (the no-reservation charge path).
reconcile_model: Final = reserved_model if reserved_model is not None else self._rate_limited_model(model_group)
reconcile_models: Final = self._rate_limited_models(reconcile_model)
pipeline_operations: Final[list[ReservationOperation]] = []
@ -4636,17 +4710,24 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
kwargs=kwargs,
tpm_limited_tags=_tpm_limited_tags(stash, kwargs),
model_group=reconcile_model.group if reconcile_model is not None else None,
model_groups=tuple(model.group for model in reconcile_models),
)
targets.extend(
scope
for scope in sorted(reserved_scopes)
if scope[0] in ("agent", "agent_session") and scope not in targets
)
charged_targets: Final = (
[target for target in targets if target[0] != "model_per_team"]
if self._key_owns_model_tpm_limit_from_request_metadata(request_metadata, reconcile_model)
else targets
logging_metadata: Final = _callback_kwargs_mapping(standard_logging_metadata)
key_owned_team_scopes: Final = frozenset(
("model_per_team", f"{logging_metadata.get('user_api_key_team_id')}:{model.group}")
for model in reconcile_models
if all(
self._key_owns_model_tpm_limit_from_request_metadata(request_metadata, candidate)
for candidate in reconcile_models
if candidate.group == model.group
)
)
charged_targets: Final = [target for target in targets if target not in key_owned_team_scopes]
if reserved_tokens > 0 and total_tokens < reserved_tokens:
verbose_proxy_logger.debug(
"Releasing unused TPM budget on success: reserved=%s, actual=%s, release=%s",

2
litellm/rust_bridge/catalog.py
View file

@ -110,6 +110,7 @@ class Feature(str, Enum):
"""A model_list that Python fans out across its deployments."""
MODEL_ALIAS_MAP = "model_alias_map"
"""A model that litellm.model_alias_map rewrites before the call."""
LATEST_ALIAS = "latest_alias"
@dataclass(frozen=True, slots=True)
@ -153,6 +154,7 @@ COMMON_PYTHON_FEATURES: Final = frozenset(
Feature.TIMEOUT_VARIANTS,
Feature.AGENTIC_LOOP_HOOK,
Feature.TLS_POLICY,
Feature.LATEST_ALIAS,
}
)

3
litellm/rust_bridge/chat_completions/detectors.py
View file

@ -38,7 +38,8 @@ def model_list(request: NativeCall) -> bool:
def model_alias_map(request: NativeCall) -> bool:
"""Python rewrites this model through the configured alias map."""
return bool(litellm.model_alias_map) and str(request.resolved["model"]) in litellm.model_alias_map
model: Final = str(request.resolved["model"])
return model in litellm.model_alias_map
def chat_retries_fallbacks(request: NativeCall) -> bool:

9
litellm/rust_bridge/detectors.py
View file

@ -12,6 +12,7 @@ import httpx
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.core_helpers import normalize_drop_params
from litellm.litellm_core_utils.model_latest_aliases import is_model_latest_alias
from litellm.rust_bridge.catalog import Feature
from litellm.rust_bridge.public_call import NativeCall, optional_mapping, optional_sequence, optional_str
@ -35,6 +36,13 @@ def mock_response(request: NativeCall) -> bool:
return present(request, "mock_response")
def latest_alias(request: NativeCall) -> bool:
return is_model_latest_alias(
optional_str(request.resolved.get("model")) or "",
optional_str(request.resolved.get("custom_llm_provider")),
)
def caller_client(request: NativeCall) -> bool:
"""Python sends through the caller's HTTP client or shared session."""
return present(request, "client") or present(request, "shared_session")
@ -179,5 +187,6 @@ COMMON: Final[Detectors] = MappingProxyType(
Feature.TIMEOUT_VARIANTS: timeout_variants,
Feature.AGENTIC_LOOP_HOOK: agentic_loop_hook,
Feature.TLS_POLICY: tls_policy,
Feature.LATEST_ALIAS: latest_alias,
}
)

33
tests/unit/litellm_core_utils/test_get_llm_provider_logic.py
View file

@ -19,6 +19,7 @@ from litellm.litellm_core_utils.get_llm_provider_logic import (
)
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.litellm_core_utils.model_latest_aliases import get_model_latest_aliases
from litellm.types.router import LiteLLM_Params
from litellm.utils import _invalidate_model_cost_lowercase_map
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
@ -27,6 +28,38 @@ from unittest.mock import patch
CUSTOM_PROVIDER: Final = "test-onprem-llm"
@pytest.mark.parametrize(
("model", "provider"),
(("openai/gpt-luna-latest", None), ("gpt-luna-latest", "openai")),
)
def test_latest_alias_resolves_to_the_bundled_target(model: str, provider: str | None) -> None:
target: Final = get_model_latest_aliases()["openai"]["gpt-luna-latest"]
resolved, resolved_provider, _, api_base = get_llm_provider(
model, custom_llm_provider=provider, api_base="https://api.openai.com/v1"
)
assert (resolved, resolved_provider, api_base) == (target, "openai", "https://api.openai.com/v1")
@pytest.mark.parametrize(
"model",
("gemini/gemini-pro-latest", "gemini/gemini-flash-latest", "dashscope/qwen-plus-latest", "mistral/mistral-large-latest"),
)
def test_provider_native_latest_alias_passes_through(model: str) -> None:
provider, name = model.split("/", 1)
assert get_llm_provider(model)[:2] == (name, provider)
def test_latest_alias_completion_reports_and_prices_the_target() -> None:
target: Final = get_model_latest_aliases()["anthropic"]["claude-opus-latest"]
response: Final = litellm.completion(
model="anthropic/claude-opus-latest", messages=[{"role": "user", "content": "hi"}], mock_response="hi"
)
assert response.model == target
cost: Final = litellm.completion_cost(completion_response=response)
assert cost is not None and cost > 0
assert cost == litellm.completion_cost(completion_response=response, model=target, custom_llm_provider="anthropic")
@pytest.fixture
def registered_custom_provider(monkeypatch: pytest.MonkeyPatch) -> str:
monkeypatch.setattr(litellm, "custom_provider_map", [{"provider": CUSTOM_PROVIDER, "custom_handler": CustomLLM()}])

47
tests/unit/litellm_core_utils/test_model_latest_aliases.py Normal file
View file

@ -0,0 +1,47 @@
from typing import Final
import pytest
import litellm
from litellm.litellm_core_utils.model_latest_aliases import (
expand_model_latest_alias,
get_model_latest_aliases,
is_model_latest_alias,
)
@pytest.mark.parametrize("model", ("", "gpt-luna-latest", "openai/unknown", "gemini/gemini-pro-latest"))
def test_non_alias_expansion_keeps_the_requested_name(model: str) -> None:
assert expand_model_latest_alias(model) == (model,)
def test_alias_expansion_keeps_requested_and_target_names() -> None:
for provider, aliases in get_model_latest_aliases().items():
for alias, target in aliases.items():
assert expand_model_latest_alias(f"{provider}/{alias}") == (f"{provider}/{alias}", f"{provider}/{target}")
def test_latest_alias_targets_have_chat_pricing_without_shadowing_native_models() -> None:
for provider, aliases in get_model_latest_aliases().items():
for alias, target in aliases.items():
target_key: Final = f"{provider}/{target}" if f"{provider}/{target}" in litellm.model_cost else target
entry: Final = litellm.model_cost[target_key]
assert (entry["litellm_provider"], entry["mode"]) == (provider, "chat")
assert f"{provider}/{alias}" not in litellm.model_cost
assert litellm.model_cost.get(alias, {}).get("litellm_provider") != provider
assert litellm.get_llm_provider(f"{provider}/{alias}")[:2] == (target, provider)
@pytest.mark.parametrize(
("model", "provider", "expected"),
(
("openai/gpt-luna-latest", None, True),
("gpt-luna-latest", "openai", True),
("gpt-luna-latest", None, False),
("gemini/gemini-pro-latest", None, False),
("", None, False),
("openai/unknown-model", None, False),
),
)
def test_latest_alias_detection_requires_an_explicit_provider(model: str, provider: str | None, expected: bool) -> None:
assert is_model_latest_alias(model, provider) is expected

10
tests/unit/llms/moonshot/test_moonshot_chat_transformation.py
View file

@ -5,6 +5,7 @@ These tests validate the MoonshotChatConfig class which extends OpenAIGPTConfig.
Moonshot AI is an OpenAI-compatible provider with minor customizations.
"""
from typing import Final
from unittest.mock import patch
@ -13,6 +14,7 @@ import pytest
import litellm
import litellm.utils
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.litellm_core_utils.model_latest_aliases import get_model_latest_aliases
from litellm.llms.moonshot.chat.transformation import MoonshotChatConfig
@ -719,10 +721,16 @@ class TestMoonshotReasoningEffort:
def test_reasoning_model_supports_reasoning_effort(self, model):
assert "reasoning_effort" in MoonshotChatConfig().get_supported_openai_params(model)
@pytest.mark.parametrize("model", ["moonshot-v1-8k", "kimi-latest", "kimi-k2-turbo-preview"])
@pytest.mark.parametrize("model", ["moonshot-v1-8k", "kimi-k2-turbo-preview"])
def test_non_reasoning_model_does_not_support_reasoning_effort(self, model):
assert "reasoning_effort" not in MoonshotChatConfig().get_supported_openai_params(model)
def test_latest_alias_uses_target_supported_params(self) -> None:
target: Final = get_model_latest_aliases()["moonshot"]["kimi-latest"]
assert MoonshotChatConfig().get_supported_openai_params(
"kimi-latest"
) == MoonshotChatConfig().get_supported_openai_params(target)
@pytest.mark.parametrize("effort", ["low", "high", "max"])
def test_declared_effort_reaches_optional_params(self, effort):
optional_params = litellm.get_optional_params(

88
tests/unit/proxy/hooks/test_model_max_budget_limiter.py
View file

@ -12,6 +12,7 @@ import pytest
import litellm
from litellm.caching.caching import DualCache
from litellm.litellm_core_utils.model_latest_aliases import get_model_latest_aliases
from litellm.proxy._types import Litellm_EntityType, UserAPIKeyAuth
from litellm.proxy.db.db_transaction_queue.window_spend_update_queue import WindowSpendTransaction
from litellm.proxy.hooks.model_budget_shared_windows import DatabaseSharedWindows, SharedWindows
@ -331,7 +332,6 @@ async def test_a_batch_polled_within_every_budget_window_is_never_charged_again(
assert _local_spend(limiter, KEY_SPEND_KEY) == pytest.approx(BATCH_COST)
async def _poll_video(
limiter: PROXY_VirtualKeyModelMaxBudgetLimiter, video_id: str, status: str, response_cost: float
) -> None:
@ -382,6 +382,7 @@ async def test_only_the_finished_poll_pays_the_job_and_only_a_video_repoll_pays_
assert await _spend(limiter, KEY_SPEND_KEY) == pytest.approx(BATCH_COST + guardrail_charges * guardrail_cost)
OPUS_GROUP: Final = "claude-opus"
OPUS_GROUP_MEMBERS: Final = MappingProxyType(
{"claude-opus-4-8": frozenset({OPUS_GROUP}), "claude-opus-5-5": frozenset({OPUS_GROUP})}
@ -487,6 +488,91 @@ def _key(model_max_budget: Mapping[str, object]) -> UserAPIKeyAuth:
return UserAPIKeyAuth(token=KEY_HASH, key_alias="opus-family", model_max_budget=dict(model_max_budget))
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ("key", "team"))
@pytest.mark.parametrize("prefix", ("", "anthropic/"))
@pytest.mark.parametrize("budget_alias", (False, True))
async def test_latest_alias_is_blocked_by_either_model_budget(scope: str, prefix: str, budget_alias: bool) -> None:
target: Final = get_model_latest_aliases()["anthropic"]["claude-opus-latest"]
name: Final = "claude-opus-latest" if budget_alias else target
budget: Final = {f"{prefix}{name}": {"max_budget": 0, "budget_duration": "1d"}}
limiter: Final = _family_limiter()
admission: Final = (
limiter.is_key_within_model_budget(_key(budget), "anthropic/claude-opus-latest")
if scope == "key"
else limiter.is_team_within_model_budget(TEAM_ID, budget, None, "anthropic/claude-opus-latest")
)
with pytest.raises(litellm.BudgetExceededError, match="exceeded budget"):
await admission
@pytest.mark.asyncio
@pytest.mark.parametrize("model", ("anthropic/claude-opus-5-5", "anthropic/unknown-model"))
async def test_latest_alias_budget_does_not_govern_other_requests(model: str) -> None:
limiter: Final = _family_limiter()
budget: Final = {"anthropic/claude-opus-latest": {"max_budget": 0, "budget_duration": "1d"}}
assert await limiter.is_key_within_model_budget(_key(budget), model)
assert await limiter.is_team_within_model_budget(TEAM_ID, budget, None, model)
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", (Litellm_EntityType.KEY, Litellm_EntityType.TEAM))
@pytest.mark.parametrize(
"entries",
(
("anthropic/claude-opus-latest", "claude-opus-5-5"),
("claude-opus*",),
("anthropic/claude-opus-5-5",),
("claude-opus-5-5",),
),
)
@pytest.mark.parametrize("requested", ("anthropic/claude-opus-latest", "anthropic/claude-opus-5-5"))
async def test_latest_alias_spend_charges_each_governing_counter_once(
scope: Litellm_EntityType, entries: tuple[str, ...], requested: str
) -> None:
budget: Final = {entry: FAMILY_CAP for entry in entries}
cache: Final = DualCache()
limiter: Final = _family_limiter(dual_cache=cache)
await _log_success(
limiter,
{
"call_type": "acompletion",
"standard_logging_object": {
"model": "claude-opus-5-5",
"model_group": requested,
"response_cost": CHAT_COST,
"metadata": {"user_api_key_hash": KEY_HASH, "user_api_key_team_id": TEAM_ID},
},
"litellm_params": {
"metadata": {
"user_api_key_model_max_budget": budget if scope == Litellm_EntityType.KEY else {},
"user_api_key_team_model_max_budget": budget if scope == Litellm_EntityType.TEAM else {},
}
},
},
)
for entry in entries:
spend_key: Final = model_budget_spend_cache_key(
entity_type=scope,
entity_id=KEY_HASH if scope == Litellm_EntityType.KEY else TEAM_ID,
budget_model=entry,
budget_duration="1d",
)
expected: Final = (
CHAT_COST
if requested == "anthropic/claude-opus-latest" or entry != "anthropic/claude-opus-latest"
else None
)
assert await cache.async_get_cache(spend_key, local_only=True) == expected
admission: Final = (
limiter.is_key_within_model_budget(_key(budget), requested)
if scope == Litellm_EntityType.KEY
else limiter.is_team_within_model_budget(TEAM_ID, budget, None, requested)
)
with pytest.raises(litellm.BudgetExceededError, match="exceeded budget"):
await admission
async def _spend_on(
limiter: PROXY_VirtualKeyModelMaxBudgetLimiter, model: str, model_max_budget: Mapping[str, object]
) -> None:

529
tests/unit/proxy/hooks/test_parallel_request_limiter_v3.py
View file

@ -28,6 +28,8 @@ from litellm.exceptions import ServiceUnavailableError
from litellm.integrations.custom_secret_manager import CustomSecretManager
from litellm.litellm_core_utils import ptu_pricing
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.model_latest_aliases import get_model_latest_aliases
from litellm.litellm_core_utils.routing_common_utils import resolve_model_group_alias
from litellm.litellm_core_utils.ptu_shares import PTUTeamCeiling
from litellm.llms.azure.ptu_capacity import AZURE_PTU_CAPACITY
from litellm.proxy._types import UserAPIKeyAuth
@ -133,6 +135,533 @@ def _isolated_request_stash():
_request_stash.reset(token)
@pytest.fixture
def reservation() -> bool:
return True
@pytest.fixture
def collision_limiter(
reservation: bool, monkeypatch: pytest.MonkeyPatch,
) -> tuple[DualCache, PROXY_MaxParallelRequestsHandler_v3, str, str]:
monkeypatch.setenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", str(reservation).lower())
alias: Final = "anthropic/claude-opus-latest"
target: Final = f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"
router: Final = Router(
model_list=[{"model_name": "anthropic/*", "litellm_params": {"model": "anthropic/*"}}],
model_group_alias={alias: target},
)
cache: Final = DualCache()
handler: Final = PROXY_MaxParallelRequestsHandler_v3(
internal_usage_cache=InternalUsageCache(cache),
model_group_resolver=partial(resolve_model_group_alias, router.registry().aliases()),
time_provider=lambda: datetime(2026, 1, 1, tzinfo=timezone.utc),
)
return cache, handler, alias, target
@pytest.mark.asyncio
@pytest.mark.parametrize(
"scope,metric",
tuple(
(scope, metric)
for scope in ("key", "team", "project", "organization")
for metric in ("model_rpm_limit", "model_tpm_limit")
) + (("project", "model_itpm_limit"), ("project", "model_otpm_limit")),
)
@pytest.mark.parametrize("limit_alias", (False, True))
async def test_latest_alias_collision_keeps_the_stricter_limit(
collision_limiter: tuple[DualCache, PROXY_MaxParallelRequestsHandler_v3, str, str],
scope: str, metric: str, limit_alias: bool,
) -> None:
cache, handler, alias, target = collision_limiter
limits: Final = {metric: {alias: 0 if limit_alias else 100, target: 100 if limit_alias else 0}}
auth: Final = UserAPIKeyAuth(
api_key="collision-key", team_id="collision-team", project_id="collision-project", org_id="collision-org",
**{("metadata" if scope == "key" else f"{scope}_metadata"): limits},
)
with pytest.raises(ProxyRateLimitError) as exc:
await handler.async_pre_call_hook(
auth, cache,
{"model": alias, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 1}, "acompletion",
)
assert exc.value.status_code == 429
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ("key", "team", "project", "organization"))
@pytest.mark.parametrize("reservation", (False, True))
@pytest.mark.parametrize("event", ("success", "failure", "partial"))
async def test_latest_alias_collision_charges_and_settles_one_counter_per_metric(
collision_limiter: tuple[DualCache, PROXY_MaxParallelRequestsHandler_v3, str, str],
scope: str, reservation: bool, event: str,
) -> None:
cache, handler, alias, target = collision_limiter
limits: Final = {"model_rpm_limit": {alias: 2}, "model_tpm_limit": {target: 10000}}
auth: Final = UserAPIKeyAuth(
api_key="collision-key", team_id="collision-team", project_id="collision-project", org_id="collision-org",
**{("metadata" if scope == "key" else f"{scope}_metadata"): limits},
)
await handler.async_pre_call_hook(
auth, cache, {"model": alias, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 64}, "acompletion"
)
entity: Final = {
"key": auth.api_key, "team": auth.team_id, "project": auth.project_id, "organization": auth.org_id
}[scope]
rpm_key: Final = _counter(f"model_per_{scope}", f"{entity}:{target}", "requests")
tpm_key: Final = _counter(f"model_per_{scope}", f"{entity}:{target}", "tokens")
stash: Final = get_request_stash()
assert stash is not None
assert await handler.rate_limits.get(rpm_key) == 1
assert await handler.rate_limits.get(tpm_key) == (stash.reserved_tokens if reservation else 1)
usage: Final = Usage(prompt_tokens=3, completion_tokens=2, total_tokens=5)
kwargs: Final = {
"standard_logging_object": {
"model_group": alias,
"metadata": {
"user_api_key_hash": auth.api_key, "user_api_key_team_id": auth.team_id,
"user_api_key_project_id": auth.project_id, "user_api_key_org_id": auth.org_id,
},
},
"litellm_params": {"metadata": {"model_group": alias, "user_api_key_metadata": auth.metadata}},
**({"combined_usage_object": usage} if event == "partial" else {}),
}
if event == "success":
await handler.async_log_success_event(kwargs, ModelResponse(model=target, usage=usage), None, None)
else:
await handler.async_log_failure_event(kwargs, None, None, None)
actual_tokens: Final = 5 if event == "success" or (event == "partial" and reservation) else 0
assert await handler.rate_limits.get(tpm_key) == actual_tokens + (0 if reservation else 1)
alias_key: Final = _counter(f"model_per_{scope}", f"{entity}:{alias}", "tokens")
assert await handler.rate_limits.get(alias_key) is None
assert await handler.async_pre_call_hook(auth, cache, {"model": alias}, "acompletion") is None
with pytest.raises(ProxyRateLimitError) as exc:
await handler.async_pre_call_hook(auth, cache, {"model": alias}, "acompletion")
assert exc.value.status_code == 429
@pytest.mark.asyncio
@pytest.mark.parametrize("event", ("success", "failure", "partial"))
async def test_latest_alias_collision_project_io_reservations_settle_once(
collision_limiter: tuple[DualCache, PROXY_MaxParallelRequestsHandler_v3, str, str], event: str,
) -> None:
cache, handler, alias, target = collision_limiter
auth: Final = UserAPIKeyAuth(
api_key="collision-io-key", project_id="collision-project",
project_metadata={metric: {alias: 10000, target: 10000} for metric in ("model_itpm_limit", "model_otpm_limit")},
)
await handler.async_pre_call_hook(
auth, cache, {"model": alias, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 64}, "acompletion"
)
stash: Final = get_request_stash()
assert stash is not None
for scope, reserved in (
("model_per_project_itpm", stash.itpm_reserved_tokens), ("model_per_project_otpm", stash.otpm_reserved_tokens),
):
assert reserved > 0
counter: Final = _counter(scope, f"{auth.project_id}:{target}", "tokens")
assert await handler.rate_limits.get(counter) == reserved
usage: Final = Usage(
prompt_tokens=3, completion_tokens=2, total_tokens=5, prompt_tokens_details={"cached_tokens": 1}
)
kwargs: Final = {
"standard_logging_object": {
"model_group": alias,
"metadata": {"user_api_key_hash": auth.api_key, "user_api_key_project_id": auth.project_id},
},
"litellm_params": {"metadata": {"model_group": alias}},
**({"combined_usage_object": usage} if event == "partial" else {}),
}
if event == "success":
await handler.async_log_success_event(kwargs, ModelResponse(model=target, usage=usage), None, None)
else:
await handler.async_log_failure_event(kwargs, None, None, None)
for scope in ("model_per_project_itpm", "model_per_project_otpm"):
counter: Final = _counter(scope, f"{auth.project_id}:{target}", "tokens")
assert int(await handler.rate_limits.get(counter) or 0) == (0 if event == "failure" else 2)
@pytest.mark.asyncio
@pytest.mark.parametrize("reservation", (False, True))
@pytest.mark.parametrize("event", ("success", "partial"))
async def test_latest_alias_collision_settles_the_target_team_limit_beside_the_key_alias_limit(
collision_limiter: tuple[DualCache, PROXY_MaxParallelRequestsHandler_v3, str, str],
reservation: bool, event: str,
) -> None:
cache, handler, alias, target = collision_limiter
auth: Final = UserAPIKeyAuth(
api_key="collision-inherited-key", team_id="collision-team",
metadata={"model_tpm_limit": {alias: 10000}}, team_metadata={"model_tpm_limit": {target: 10000}},
)
await handler.async_pre_call_hook(
auth, cache, {"model": alias, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 64}, "acompletion"
)
usage: Final = Usage(prompt_tokens=3, completion_tokens=2, total_tokens=5)
kwargs: Final = {
"standard_logging_object": {
"model_group": alias,
"metadata": {"user_api_key_hash": auth.api_key, "user_api_key_team_id": auth.team_id},
},
"litellm_params": {"metadata": {"model_group": alias, "user_api_key_metadata": auth.metadata}},
**({"combined_usage_object": usage} if event == "partial" else {}),
}
if event == "success":
await handler.async_log_success_event(kwargs, ModelResponse(model=target, usage=usage), None, None)
else:
await handler.async_log_failure_event(kwargs, None, None, None)
actual_tokens: Final = 5 if reservation or event == "success" else 0
for scope, entity in (("model_per_key", auth.api_key), ("model_per_team", auth.team_id)):
counter: Final = _counter(scope, f"{entity}:{target}", "tokens")
assert await handler.rate_limits.get(counter) == actual_tokens + (0 if reservation else 1)
@pytest.mark.asyncio
async def test_latest_alias_collision_project_frames_charge_once_and_enforce_the_lower_cap(
collision_limiter: tuple[DualCache, PROXY_MaxParallelRequestsHandler_v3, str, str],
) -> None:
cache, handler, alias, target = collision_limiter
auth: Final = UserAPIKeyAuth(
api_key="collision-frame-key", project_id="collision-project",
project_metadata={metric: {alias: 5, target: 100} for metric in ("model_itpm_limit", "model_otpm_limit")},
)
await handler.enforce_project_io_token_quota_for_frame(auth, alias, 3, 2)
with pytest.raises(ProxyRateLimitError) as exc:
await handler.enforce_project_io_token_quota_for_frame(auth, alias, 3, 2)
assert exc.value.status_code == 429
for scope, expected in (("model_per_project_itpm", 3), ("model_per_project_otpm", 2)):
counter: Final = _counter(scope, f"{auth.project_id}:{target}", "tokens")
assert await handler.rate_limits.get(counter) == expected
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ("key", "team", "project", "organization"))
@pytest.mark.parametrize("metric", ("model_rpm_limit", "model_tpm_limit"))
@pytest.mark.parametrize("budget_alias", (False, True))
async def test_latest_alias_is_blocked_by_either_rate_limit(scope: str, metric: str, budget_alias: bool) -> None:
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache))
target: Final = get_model_latest_aliases()["anthropic"]["claude-opus-latest"]
limited_model: Final = "anthropic/claude-opus-latest" if budget_alias else f"anthropic/{target}"
limits: Final = {metric: {limited_model: 0}}
auth: Final = UserAPIKeyAuth(
api_key="latest-rate-key",
team_id="latest-rate-team",
project_id="latest-rate-project",
org_id="latest-rate-org",
metadata=limits if scope == "key" else {},
team_metadata=limits if scope == "team" else {},
project_metadata=limits if scope == "project" else {},
organization_metadata=limits if scope == "organization" else {},
)
with pytest.raises(ProxyRateLimitError) as exc:
await handler.async_pre_call_hook(
auth,
cache,
{"model": "anthropic/claude-opus-latest", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 1},
"acompletion",
)
assert exc.value.status_code == 429
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ("key", "team", "project", "organization"))
@pytest.mark.parametrize("metric", ("model_rpm_limit", "model_tpm_limit"))
@pytest.mark.parametrize("model", ("anthropic/claude-opus-5-5", "anthropic/unknown-model", None))
async def test_latest_alias_rate_limit_does_not_govern_other_requests(scope: str, metric: str, model: str | None) -> None:
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache))
limits: Final = {metric: {"anthropic/claude-opus-latest": 0}}
auth: Final = UserAPIKeyAuth(
api_key="latest-other-key",
team_id="latest-other-team",
project_id="latest-other-project",
org_id="latest-other-org",
metadata=limits if scope == "key" else {},
team_metadata=limits if scope == "team" else {},
project_metadata=limits if scope == "project" else {},
organization_metadata=limits if scope == "organization" else {},
)
assert await handler.async_pre_call_hook(auth, cache, {"model": model}, "acompletion") is None
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ("key", "team", "project", "organization"))
async def test_latest_alias_rpm_draws_from_both_model_counters(scope: str) -> None:
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache))
alias: Final = "anthropic/claude-opus-latest"
target: Final = f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"
limits: Final = {"model_rpm_limit": {alias: 2, target: 1}}
auth: Final = UserAPIKeyAuth(
api_key="latest-rpm-key",
team_id="latest-rpm-team",
project_id="latest-rpm-project",
org_id="latest-rpm-org",
organization_rpm_limit=10 if scope == "organization" else None,
metadata=limits if scope == "key" else {},
team_metadata=limits if scope == "team" else {},
project_metadata=limits if scope == "project" else {},
organization_metadata=limits if scope == "organization" else {},
)
assert await handler.async_pre_call_hook(auth, cache, {"model": alias}, "acompletion") is None
descriptor: Final = f"model_per_{scope}"
entity: Final = {
"key": auth.api_key, "team": auth.team_id, "project": auth.project_id, "organization": auth.org_id
}[scope]
for model in (alias, target):
counter: Final = _counter(descriptor, f"{entity}:{model}", "requests")
assert await handler.rate_limits.get(counter) == 1
if scope == "organization":
global_counter: Final = _counter("organization", str(auth.org_id), "requests")
assert await handler.rate_limits.get(global_counter) == 1
with pytest.raises(ProxyRateLimitError) as exc:
await handler.async_pre_call_hook(auth, cache, {"model": alias}, "acompletion")
assert exc.value.status_code == 429
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ("key", "team", "project", "organization"))
@pytest.mark.parametrize("event", ("success", "failure", "partial"))
@pytest.mark.parametrize("reservation", (False, True))
async def test_latest_alias_tpm_settles_both_counters(
scope: str, event: str, reservation: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", str(reservation).lower())
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(
internal_usage_cache=InternalUsageCache(cache),
time_provider=lambda: datetime(2026, 1, 1, tzinfo=timezone.utc),
)
alias: Final = "anthropic/claude-opus-latest"
target: Final = f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"
limits: Final = {"model_tpm_limit": {alias: 10000, target: 10000}}
auth: Final = UserAPIKeyAuth(
api_key="latest-tpm-key",
team_id="latest-tpm-team",
project_id="latest-tpm-project",
org_id="latest-tpm-org",
tpm_limit=10000,
metadata=limits if scope == "key" else {},
team_metadata=limits if scope == "team" else {},
project_metadata=limits if scope == "project" else {},
organization_metadata=limits if scope == "organization" else {},
)
await handler.async_pre_call_hook(
auth, cache, {"model": alias, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 64}, "acompletion"
)
stash: Final = get_request_stash()
assert stash is not None
descriptor: Final = f"model_per_{scope}"
entity: Final = {
"key": auth.api_key, "team": auth.team_id, "project": auth.project_id, "organization": auth.org_id
}[scope]
keys: Final = tuple(
_counter(descriptor, f"{entity}:{model}", "tokens") for model in (alias, target)
)
for key in keys:
assert int(await handler.rate_limits.get(key) or 0) == (stash.reserved_tokens if reservation else 1)
kwargs: Final = {
"standard_logging_object": {
"model_group": alias,
"metadata": {
"user_api_key_hash": auth.api_key,
"user_api_key_team_id": auth.team_id,
"user_api_key_project_id": auth.project_id,
"user_api_key_org_id": auth.org_id,
},
},
"litellm_params": {"metadata": {"model_group": alias, "user_api_key_metadata": auth.metadata}},
**(
{"combined_usage_object": Usage(prompt_tokens=3, completion_tokens=2, total_tokens=5)}
if event == "partial"
else {}
),
}
if event == "success":
await handler.async_log_success_event(
kwargs,
ModelResponse(model=target, usage=Usage(prompt_tokens=3, completion_tokens=2, total_tokens=5)),
None,
None,
)
else:
await handler.async_log_failure_event(kwargs, None, None, None)
actual_tokens: Final = 5 if event == "success" or (event == "partial" and reservation) else 0
for key in (*keys, _counter("api_key", auth.api_key, "tokens")):
assert int(await handler.rate_limits.get(key) or 0) == actual_tokens + (0 if reservation else 1)
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ("key", "team", "project", "organization"))
@pytest.mark.parametrize("request_alias", (False, True))
@pytest.mark.parametrize("reservation_released", (False, True))
async def test_failed_latest_alias_without_open_reservation_matches_direct_model_accounting(
scope: str,
request_alias: bool,
reservation_released: bool,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", str(reservation_released).lower())
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(
internal_usage_cache=InternalUsageCache(cache),
time_provider=lambda: datetime(2026, 1, 1, tzinfo=timezone.utc),
)
alias: Final = "anthropic/claude-opus-latest"
target: Final = f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"
model: Final = alias if request_alias else target
limits: Final = {"model_tpm_limit": {alias: 10000, target: 10000}}
auth: Final = UserAPIKeyAuth(
api_key="failed-latest-key",
team_id="failed-latest-team",
project_id="failed-latest-project",
org_id="failed-latest-org",
tpm_limit=10000,
metadata=limits if scope == "key" else {},
team_metadata=limits if scope == "team" else {},
project_metadata=limits if scope == "project" else {},
organization_metadata=limits if scope == "organization" else {},
)
request_data: Final = {"model": model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 64}
await handler.async_pre_call_hook(auth, cache, request_data, "acompletion")
if reservation_released:
await handler.async_post_call_failure_hook(
request_data=request_data,
original_exception=Exception("rejected"),
user_api_key_dict=auth,
)
entity: Final = {
"key": auth.api_key,
"team": auth.team_id,
"project": auth.project_id,
"organization": auth.org_id,
}[scope]
models: Final = (alias, target) if request_alias else (target,)
keys: Final = (
_counter("api_key", auth.api_key, "tokens"),
*(_counter(f"model_per_{scope}", f"{entity}:{name}", "tokens") for name in models),
)
await handler.async_log_failure_event(
{
"standard_logging_object": {
"model_group": model,
"metadata": {
"user_api_key_hash": auth.api_key,
"user_api_key_team_id": auth.team_id,
"user_api_key_project_id": auth.project_id,
"user_api_key_org_id": auth.org_id,
},
},
"litellm_params": {"metadata": {"model_group": model, "user_api_key_metadata": auth.metadata}},
"combined_usage_object": Usage(prompt_tokens=3, completion_tokens=2, total_tokens=5),
},
None,
None,
None,
)
for key in keys:
assert int(await handler.rate_limits.get(key) or 0) == (0 if reservation_released else 1)
@pytest.mark.asyncio
@pytest.mark.parametrize("metric", ("model_itpm_limit", "model_otpm_limit"))
@pytest.mark.parametrize("limit_alias", (False, True))
async def test_latest_alias_project_io_limits_block_either_name(metric: str, limit_alias: bool) -> None:
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache))
alias: Final = "anthropic/claude-opus-latest"
target: Final = f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"
auth: Final = UserAPIKeyAuth(
api_key="latest-io-key",
project_id="latest-io-project",
project_metadata={metric: {alias if limit_alias else target: 0}},
)
with pytest.raises(ProxyRateLimitError) as exc:
await handler.async_pre_call_hook(
auth, cache, {"model": alias, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 1}, "acompletion"
)
assert exc.value.status_code == 429
@pytest.mark.asyncio
@pytest.mark.parametrize("metric", ("model_itpm_limit", "model_otpm_limit"))
@pytest.mark.parametrize("model", ("anthropic/claude-opus-5-5", "anthropic/unknown-model", None))
async def test_latest_alias_project_io_limits_leave_other_requests_unchanged(metric: str, model: str | None) -> None:
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache))
auth: Final = UserAPIKeyAuth(
api_key="latest-io-other-key",
project_id="latest-io-other-project",
project_metadata={metric: {"anthropic/claude-opus-latest": 0}},
)
assert await handler.async_pre_call_hook(auth, cache, {"model": model}, "acompletion") is None
@pytest.mark.asyncio
@pytest.mark.parametrize("event", ("success", "failure", "partial"))
async def test_latest_alias_project_io_reservations_settle_both_counters(event: str) -> None:
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(
internal_usage_cache=InternalUsageCache(cache),
time_provider=lambda: datetime(2026, 1, 1, tzinfo=timezone.utc),
)
alias: Final = "anthropic/claude-opus-latest"
target: Final = f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"
auth: Final = UserAPIKeyAuth(
api_key="latest-io-settle-key",
project_id="latest-io-settle-project",
project_metadata={metric: {alias: 10000, target: 10000} for metric in ("model_itpm_limit", "model_otpm_limit")},
)
await handler.async_pre_call_hook(
auth, cache, {"model": alias, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 64}, "acompletion"
)
stash: Final = get_request_stash()
assert stash is not None
for scope, reserved in (
("model_per_project_itpm", stash.itpm_reserved_tokens),
("model_per_project_otpm", stash.otpm_reserved_tokens),
):
assert reserved > 0
for model in (alias, target):
counter: Final = _counter(scope, f"{auth.project_id}:{model}", "tokens")
assert await handler.rate_limits.get(counter) == reserved
usage: Final = Usage(prompt_tokens=3, completion_tokens=2, total_tokens=5, prompt_tokens_details={"cached_tokens": 1})
kwargs: Final = {
"standard_logging_object": {
"model_group": alias,
"metadata": {"user_api_key_hash": auth.api_key, "user_api_key_project_id": auth.project_id},
},
"litellm_params": {"metadata": {"model_group": alias}},
**({"combined_usage_object": usage} if event == "partial" else {}),
}
if event == "success":
await handler.async_log_success_event(kwargs, ModelResponse(model=target, usage=usage), None, None)
else:
await handler.async_log_failure_event(kwargs, None, None, None)
for scope in ("model_per_project_itpm", "model_per_project_otpm"):
for model in (alias, target):
counter: Final = _counter(scope, f"{auth.project_id}:{model}", "tokens")
assert int(await handler.rate_limits.get(counter) or 0) == (0 if event == "failure" else 2)
@pytest.mark.asyncio
async def test_latest_alias_project_io_frame_reservations_charge_both_names() -> None:
cache: Final = DualCache()
handler: Final = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache))
alias: Final = "anthropic/claude-opus-latest"
target: Final = f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}"
auth: Final = UserAPIKeyAuth(
api_key="latest-frame-key",
project_id="latest-frame-project",
project_metadata={metric: {alias: 20, target: 20} for metric in ("model_itpm_limit", "model_otpm_limit")},
)
await handler.enforce_project_io_token_quota_for_frame(auth, alias, 3, 2)
for scope, expected in (("model_per_project_itpm", 3), ("model_per_project_otpm", 2)):
for model in (alias, target):
counter: Final = _counter(scope, f"{auth.project_id}:{model}", "tokens")
assert await handler.rate_limits.get(counter) == expected
@pytest.mark.parametrize(
"throttle_pct, expected_rpm, expected_tpm",
[

29
tests/unit/rust_bridge/test_catalog.py
View file

@ -5,7 +5,8 @@ from typing import Final
import pytest
from litellm.rust_bridge import catalog, configuration
from litellm.litellm_core_utils.model_latest_aliases import get_model_latest_aliases
from litellm.rust_bridge import catalog, configuration, detectors
from litellm.rust_bridge.catalog import (
Feature,
Route,
@ -15,6 +16,32 @@ from litellm.rust_bridge.catalog import (
Target,
)
from litellm.rust_bridge.configuration import Decision, Rollout
from litellm.rust_bridge.public_call import NativeCall
@pytest.mark.parametrize(
("model", "expected"),
(
("anthropic/claude-opus-latest", Rollout.PYTHON_ONLY),
(f"anthropic/{get_model_latest_aliases()['anthropic']['claude-opus-latest']}", Rollout.RUST_OPT_IN),
),
)
def test_latest_alias_messages_falls_back_without_changing_the_pinned_model_rollout(
model: str,
expected: Rollout,
) -> None:
request: Final = NativeCall(args=(), kwargs={"model": model}, base={})
context: Final = RouteContext(
Route.MESSAGES,
provider="anthropic",
model=model,
features=detectors.detected(detectors.COMMON, request),
)
rules: Final = (
RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=catalog.MESSAGES_PYTHON_FEATURES),
RouteRule(Route.MESSAGES, Rollout.RUST_OPT_IN, providers=frozenset({"anthropic"})),
)
assert catalog.rollout(context, rules) is expected
@pytest.fixture(autouse=True)

16
tests/unit/rust_bridge/test_detectors.py
View file

@ -45,6 +45,22 @@ def call(**kwargs: object) -> NativeCall:
return NativeCall(args=(), kwargs=kwargs, base={"model": "gpt-5"})
@pytest.mark.parametrize(
("model", "provider", "expected"),
(
("openai/gpt-luna-latest", None, True),
("gpt-luna-latest", "openai", True),
("gemini/gemini-pro-latest", None, False),
("gpt-luna-latest", None, False),
("anthropic/claude-opus-latest", None, True),
("claude-opus-latest", "anthropic", True),
("", None, False),
),
)
def test_floating_alias_requires_python(model: str, provider: str | None, expected: bool) -> None:
assert detectors.latest_alias(call(model=model, custom_llm_provider=provider)) is expected
@pytest.mark.parametrize(
("model", "provider", "tools", "expected"),
(