Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 103 additions & 110 deletions litellm/proxy/hooks/parallel_request_limiter_v3.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@
ensure_response_additional_headers,
response_has_hidden_params,
)
from litellm.router_utils.common_utils import resolve_model_group_alias
from litellm.types.caching import RedisPipelineIncrementOperation
from litellm.types.llms.openai import BaseLiteLLMOpenAIResponseObject, ResponseAPIUsage
from litellm.types.utils import (
Expand Down Expand Up @@ -91,6 +92,26 @@
_REQUEST_RATE_LIMIT_DATA: Final = TypeAdapter(Mapping[str, object])


@dataclass(frozen=True, slots=True)
class RateLimitedModel:
requested: str
group: str

def limit_in(self, limits: Mapping[str, int] | None) -> int | None:
if limits is None:
return None
requested_limit: Final = limits.get(self.requested)
return requested_limit if requested_limit is not None else limits.get(self.group)


def _resolve_model_group_alias_via_proxy_router(model: str) -> str | None:
from litellm.proxy.proxy_server import llm_router

if llm_router is None:
return None
return resolve_model_group_alias(llm_router.model_group_alias, model)


def _sibling_counter_keys(window_key: str) -> tuple[str, str]:
prefix: Final = window_key.removesuffix(":window")
return f"{prefix}:requests", f"{prefix}:tokens"
Expand Down Expand Up @@ -626,9 +647,11 @@ def __init__(
self,
internal_usage_cache: InternalUsageCache,
time_provider: Callable[[], datetime] | None = None,
model_group_resolver: Callable[[str], str | None] = _resolve_model_group_alias_via_proxy_router,
):
self.internal_usage_cache = internal_usage_cache
self._time_provider = time_provider or datetime.now
self._model_group_resolver = model_group_resolver
if self.internal_usage_cache.dual_cache.redis_cache is not None:
self.batch_rate_limiter_script = self.internal_usage_cache.dual_cache.redis_cache.async_register_script(
BATCH_RATE_LIMITER_SCRIPT
Expand Down Expand Up @@ -2346,6 +2369,14 @@ async def enforce_project_io_token_quota_for_frame(
if response["overall_code"] == "OVER_LIMIT":
self._handle_rate_limit_error(response, descriptors, requested_model)

def _rate_limited_model(self, requested_model: str | None) -> RateLimitedModel | None:
if not requested_model:
return None
return RateLimitedModel(
requested=requested_model,
group=self._model_group_resolver(requested_model) or requested_model,
)

def create_organization_rate_limit_descriptor(
self, user_api_key_dict: UserAPIKeyAuth, requested_model: str | None = None
) -> list[RateLimitDescriptor]:
Expand All @@ -2367,43 +2398,28 @@ def create_organization_rate_limit_descriptor(
)
)

# Model specific org rate limits
if (
model: Final = self._rate_limited_model(requested_model)
if model is None:
return descriptors
model_specific_tpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "organization_metadata", "model_tpm_limit")
)
model_specific_rpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "organization_metadata", "model_rpm_limit")
is not None
or get_model_rate_limit_from_metadata(user_api_key_dict, "organization_metadata", "model_tpm_limit")
is not None
):
_tpm_limit_for_team_model: Final = (
get_model_rate_limit_from_metadata(user_api_key_dict, "organization_metadata", "model_tpm_limit") or {}
)
_rpm_limit_for_team_model: Final = (
get_model_rate_limit_from_metadata(user_api_key_dict, "organization_metadata", "model_rpm_limit") or {}
)
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,
},
)

should_check_rate_limit = False
if requested_model in _tpm_limit_for_team_model or requested_model in _rpm_limit_for_team_model:
should_check_rate_limit = True

if should_check_rate_limit:
model_specific_tpm_limit = None
model_specific_rpm_limit = None
if requested_model in _tpm_limit_for_team_model:
model_specific_tpm_limit = _tpm_limit_for_team_model[requested_model]
if requested_model in _rpm_limit_for_team_model:
model_specific_rpm_limit = _rpm_limit_for_team_model[requested_model]
descriptors.append(
RateLimitDescriptor(
key="model_per_organization",
value=f"{user_api_key_dict.org_id}:{requested_model}",
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(
Expand All @@ -2425,34 +2441,22 @@ def _add_model_per_key_rate_limit_descriptor(
get_key_model_tpm_limit,
)

if not requested_model:
model: Final = self._rate_limited_model(requested_model)
if model is None:
return

_tpm_limit_for_key_model = get_key_model_tpm_limit(user_api_key_dict, model_name=requested_model)
_rpm_limit_for_key_model = get_key_model_rpm_limit(user_api_key_dict, model_name=requested_model)

if _tpm_limit_for_key_model is None and _rpm_limit_for_key_model is None:
return

_tpm_limit_for_key_model = _tpm_limit_for_key_model or {}
_rpm_limit_for_key_model = _rpm_limit_for_key_model or {}

# Check if model has any rate limits configured
should_check_rate_limit: Final = (
requested_model in _tpm_limit_for_key_model or requested_model in _rpm_limit_for_key_model
model_specific_tpm_limit: Final = model.limit_in(
get_key_model_tpm_limit(user_api_key_dict, model_name=model.group)
)

if not should_check_rate_limit:
model_specific_rpm_limit: Final = model.limit_in(
get_key_model_rpm_limit(user_api_key_dict, model_name=model.group)
)
if model_specific_tpm_limit is None and model_specific_rpm_limit is None:
return

# Get model-specific limits
model_specific_tpm_limit: Final[int | None] = _tpm_limit_for_key_model.get(requested_model)
model_specific_rpm_limit: Final[int | None] = _rpm_limit_for_key_model.get(requested_model)

descriptors.append(
RateLimitDescriptor(
key="model_per_key",
value=f"{user_api_key_dict.api_key}:{requested_model}",
value=f"{user_api_key_dict.api_key}:{model.group}",
rate_limit={
"requests_per_unit": model_specific_rpm_limit,
"tokens_per_unit": model_specific_tpm_limit,
Expand Down Expand Up @@ -2955,57 +2959,56 @@ def get_rate_limiter_for_call_type(self, call_type: str) -> CallTypeRateLimiter
def _key_owns_model_limit(
self,
user_api_key_dict: UserAPIKeyAuth,
requested_model: str,
model: RateLimitedModel,
rate_limit_key: Literal["model_rpm_limit", "model_tpm_limit"],
) -> bool:
key_own_limits: Final = get_key_own_model_rate_limit(user_api_key_dict, rate_limit_key)
return key_own_limits is not None and key_own_limits.get(requested_model) is not None
return model.limit_in(get_key_own_model_rate_limit(user_api_key_dict, rate_limit_key)) is not None

def _inherited_team_model_limit(
self,
user_api_key_dict: UserAPIKeyAuth,
requested_model: str,
model: RateLimitedModel,
rate_limit_key: Literal["model_rpm_limit", "model_tpm_limit"],
) -> int | None:
team_limits: Final = get_model_rate_limit_from_metadata(user_api_key_dict, "team_metadata", rate_limit_key)
team_limit: Final = team_limits.get(requested_model) if team_limits else None
if team_limit is None:
return None
if self._key_owns_model_limit(user_api_key_dict, requested_model, rate_limit_key):
team_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "team_metadata", rate_limit_key)
)
if team_limit is None or self._key_owns_model_limit(user_api_key_dict, model, rate_limit_key):
return None
return team_limit

def _key_owns_model_tpm_limit_from_request_metadata(
self,
request_metadata: Mapping[str, object],
model_group: str | None,
model: RateLimitedModel | None,
) -> bool:
if model_group is None:
if model is None:
return False
key_view: Final = UserAPIKeyAuth.model_validate(
{
"metadata": request_metadata.get("user_api_key_metadata") or {},
"model_max_budget": request_metadata.get("user_api_key_model_max_budget") or {},
}
)
return self._key_owns_model_limit(key_view, model_group, "model_tpm_limit")
return self._key_owns_model_limit(key_view, model, "model_tpm_limit")

def _add_team_model_rate_limit_descriptor_from_metadata(
self,
user_api_key_dict: UserAPIKeyAuth,
requested_model: str | None,
descriptors: list[RateLimitDescriptor],
) -> None:
if requested_model is None:
model: Final = self._rate_limited_model(requested_model)
if model is None:
return
team_rpm_limit: Final = self._inherited_team_model_limit(user_api_key_dict, requested_model, "model_rpm_limit")
team_tpm_limit: Final = self._inherited_team_model_limit(user_api_key_dict, requested_model, "model_tpm_limit")
team_rpm_limit: Final = self._inherited_team_model_limit(user_api_key_dict, model, "model_rpm_limit")
team_tpm_limit: Final = self._inherited_team_model_limit(user_api_key_dict, model, "model_tpm_limit")
if team_rpm_limit is None and team_tpm_limit is None:
return
descriptors.append(
RateLimitDescriptor(
key="model_per_team",
value=f"{user_api_key_dict.team_id}:{requested_model}",
value=f"{user_api_key_dict.team_id}:{model.group}",
rate_limit={
"requests_per_unit": team_rpm_limit,
"tokens_per_unit": team_tpm_limit,
Expand All @@ -3021,34 +3024,28 @@ def _add_project_model_rate_limit_descriptor_from_metadata(
descriptors: list[RateLimitDescriptor],
) -> None:
"""Add project model rate limit descriptor from project_metadata if applicable."""
if (
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_rpm_limit") is not None
or get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_tpm_limit") is not None
):
_tpm_limit_for_project_model: Final = (
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_tpm_limit") or {}
)
_rpm_limit_for_project_model: Final = (
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_rpm_limit") or {}
)
should_check_rate_limit: Final = (
requested_model in _tpm_limit_for_project_model or requested_model in _rpm_limit_for_project_model
model: Final = self._rate_limited_model(requested_model)
if model is None:
return
model_specific_tpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_tpm_limit")
)
model_specific_rpm_limit: Final = model.limit_in(
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,
},
)

if should_check_rate_limit and requested_model is not None:
model_specific_tpm_limit: Final = _tpm_limit_for_project_model.get(requested_model)
model_specific_rpm_limit: Final = _rpm_limit_for_project_model.get(requested_model)
descriptors.append(
RateLimitDescriptor(
key="model_per_project",
value=f"{user_api_key_dict.project_id}:{requested_model}",
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(
self,
Expand All @@ -3062,25 +3059,21 @@ def add_project_io_token_rate_limit_descriptors_from_metadata(
TPM descriptor above -- these give Bedrock Mantle-style separate input/output
token quotas at the project level.
"""
if requested_model is None or user_api_key_dict.project_id is None:
model: Final = self._rate_limited_model(requested_model)
if model is None or user_api_key_dict.project_id is None:
return

itpm_limit_for_project_model: Final = (
model_itpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_itpm_limit")
or {} # mutable-ok: metadata helper returns an optional mapping
)
otpm_limit_for_project_model: Final = (
model_otpm_limit: Final = model.limit_in(
get_model_rate_limit_from_metadata(user_api_key_dict, "project_metadata", "model_otpm_limit")
or {} # mutable-ok: metadata helper returns an optional mapping
)

model_itpm_limit: Final = itpm_limit_for_project_model.get(requested_model)
model_otpm_limit: Final = otpm_limit_for_project_model.get(requested_model)

if model_itpm_limit is None and model_otpm_limit is None:
return

descriptor_value: Final = f"{user_api_key_dict.project_id}:{requested_model}"
descriptor_value: Final = f"{user_api_key_dict.project_id}:{model.group}"
if model_itpm_limit is not None:
descriptors.append(
RateLimitDescriptor(
Expand Down Expand Up @@ -4516,7 +4509,7 @@ def _build_success_event_pipeline_operations(
# pre-call reservation incremented. If a reservation was made,
# ``reserved_model`` is authoritative; otherwise fall back to the
# router's ``model_group`` (covers the no-reservation charge path).
reconcile_model: Final = reserved_model or model_group
reconcile_model: Final = self._rate_limited_model(reserved_model or model_group)
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved

pipeline_operations: Final[list[RedisPipelineIncrementOperation]] = []

Expand All @@ -4534,7 +4527,7 @@ def _build_success_event_pipeline_operations(
targets: Final = self._collect_tpm_scope_targets(
standard_logging_metadata=standard_logging_metadata,
kwargs=kwargs,
model_group=reconcile_model,
model_group=reconcile_model.group if reconcile_model is not None else None,
)
charged_targets: Final = (
[target for target in targets if target[0] != "model_per_team"]
Expand Down
1 change: 1 addition & 0 deletions tests/e2e/coverage_registry/quota_management.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
- {id: quota_management.ratelimit.redis_backed.blocks_over_limit, module: quota_management, tier: P0, behavior: ratelimit, variant: redis_backed, assertions: [blocks_over_limit], exercised_on: [chat_completions], source: "parallel_request_limiter_v3.py", rationale: "With Redis configured, RPM still enforces 429 across the shared limiter path customers run multi-replica"}
- {id: quota_management.ratelimit.rpm.resets_after_window, module: quota_management, tier: P1, behavior: ratelimit, variant: rpm, assertions: [resets_after_window], exercised_on: [chat_completions], source: "parallel_request_limiter_v3.py", rationale: "Rate-limit window (LITELLM_RATE_LIMIT_WINDOW_SIZE, 60s default) expires; a blocked key serves again in the next window"}
- {id: quota_management.ratelimit.rpm.headers_report_remaining, module: quota_management, tier: P1, behavior: ratelimit, variant: rpm, assertions: [headers_report_remaining], exercised_on: [chat_completions], source: "parallel_request_limiter_v3.py async_post_call_success_hook", rationale: "Successful responses carry x-ratelimit-api_key-{limit,remaining}-{requests,tokens} so clients can pace"}
- {id: quota_management.ratelimit.model_group_alias.shares_bucket, module: quota_management, tier: P1, behavior: ratelimit, variant: model_group_alias, assertions: [shares_bucket], exercised_on: [chat_completions], source: "parallel_request_limiter_v3.py:_add_model_per_key_rate_limit_descriptor", rationale: "A model_group_alias draws on the same per-key deployment rpm bucket as the resolved model group, so alias plus real-name traffic cannot exceed the configured limit"}
- {id: quota_management.ratelimit.priority_generous.picks_under_tpm, module: quota_management, tier: P1, behavior: ratelimit, variant: priority_generous, assertions: [picks_under_tpm], exercised_on: [chat_completions, messages], source: "dynamic_rate_limiter_v3.py:36-52", rationale: "Generous mode (<80% sat) allows priority borrowing"}
- {id: quota_management.ratelimit.priority_strict.picks_under_tpm, module: quota_management, tier: P1, behavior: ratelimit, variant: priority_strict, assertions: [picks_under_tpm], exercised_on: [chat_completions, messages], source: "dynamic_rate_limiter_v3.py:53-71", rationale: "Strict mode (>=80% sat) enforces priority fairness"}
- {id: quota_management.budget.key.blocks_over_limit, module: quota_management, tier: P0, behavior: budget, variant: key, assertions: [blocks_over_limit], exercised_on: [chat_completions], source: "proxy/auth/auth_checks.py", rationale: "A key's max_budget blocks further paid calls once spend crosses it"}
Expand Down
7 changes: 7 additions & 0 deletions tests/e2e/gateway/stage_mirror_ci_config.yml
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ router_settings:
num_retries: 3
allowed_fails: 5
cooldown_time: 30
model_group_alias:
e2e-alias-rl-alias: e2e-alias-rl-target

model_list:
- model_name: gpt-5.5
Expand All @@ -60,6 +62,11 @@ model_list:
litellm_params:
model: gemini/gemini-2.5-flash
api_key: os.environ/GEMINI_API_KEY
- model_name: e2e-alias-rl-target
litellm_params:
model: anthropic/claude-haiku-4-5
api_key: os.environ/ANTHROPIC_API_KEY
default_api_key_rpm_limit: 3
- model_name: openai-text-embedding-3-small
litellm_params:
model: openai/text-embedding-3-small
Expand Down
Loading
Loading