Implementation:BerriAI Litellm Lowest TPM RPM V2 Strategy

Attribute	Value
Sources	litellm/router_strategy/lowest_tpm_rpm_v2.py
Domains	Router, Strategy, Rate Limiting, Load Balancing
last_updated	2026-02-15 16:00 GMT

Overview

The Lowest TPM/RPM V2 Strategy is an updated deployment selection strategy that routes requests to the deployment with the lowest tokens-per-minute (TPM) usage, designed to work across multiple instances using Redis-backed batch operations.

Description

This module provides the LowestTPMLoggingHandler_v2 class, which extends both BaseRoutingStrategy and CustomLogger. Unlike the V1 strategy that stores aggregate model group counts, V2 caches individual model deployment TPM/RPM values with separate cache keys per deployment and minute window. This design enables efficient cross-instance coordination via redis.mget for batch reads and redis.incr for atomic increments. The class includes pre-call RPM limit enforcement (raising RateLimitError before the call is made), post-call TPM tracking, and deployment selection that picks the deployment with the lowest current TPM while respecting both TPM and RPM limits.

Usage

Import this class when configuring the LiteLLM Router with routing_strategy="usage-based-routing-v2" or the default usage-based routing. The router automatically registers it for logging callbacks and deployment selection.

Code Reference

Source Location

litellm/router_strategy/lowest_tpm_rpm_v2.py

Classes

class RoutingArgs(LiteLLMPydanticObjectBase):
    ttl: int = 1 * 60  # 1min (RPM/TPM expire key)

class LowestTPMLoggingHandler_v2(BaseRoutingStrategy, CustomLogger):
    test_flag: bool = False
    logged_success: int = 0
    logged_failure: int = 0
    default_cache_time_seconds: int = 1 * 60 * 60  # 1 hour

    def __init__(self, router_cache: DualCache, routing_args: dict = {}):

Key Methods

Method	Signature	Description
pre_call_check	def pre_call_check(self, deployment: Dict) -> Optional[Dict]	Sync pre-call RPM limit check; increments RPM counter and raises RateLimitError if over limit
async_pre_call_check	async def async_pre_call_check(self, deployment: Dict, parent_otel_span: Optional[Span]) -> Optional[Dict]	Async pre-call RPM limit check with OpenTelemetry span support
log_success_event	def log_success_event(self, kwargs, response_obj, start_time, end_time)	Sync callback that increments TPM counter after successful call
async_log_success_event	async def async_log_success_event(self, kwargs, response_obj, start_time, end_time)	Async callback that increments TPM counter after successful call
get_available_deployments	def get_available_deployments(self, model_group: str, healthy_deployments: list, messages=None, input=None, parent_otel_span=None)	Sync deployment selection using batch cache reads
async_get_available_deployments	async def async_get_available_deployments(self, model_group: str, healthy_deployments: list, messages=None, input=None)	Async deployment selection using batch cache reads

Import

from litellm.router_strategy.lowest_tpm_rpm_v2 import LowestTPMLoggingHandler_v2

I/O Contract

Inputs

Parameter	Type	Description
router_cache	DualCache	Shared dual cache (in-memory + optional Redis) for TPM/RPM counters
routing_args	dict	Configuration with optional ttl (default 60s) for cache key expiration
model_group	str	The model group name to select a deployment for
healthy_deployments	list	List of deployment dictionaries currently considered healthy
messages	Optional[List[Dict[str, str]]]	Messages for estimating input token count
input	Optional[Union[str, List]]	Text input for estimating input token count

Outputs

Return Type	Description
dict	The selected deployment dictionary with the lowest TPM usage
Raises RateLimitError	When no deployments are available within their TPM/RPM limits (async variant) or raises ValueError (sync variant)

Usage Examples

from litellm.caching.caching import DualCache
from litellm.router_strategy.lowest_tpm_rpm_v2 import LowestTPMLoggingHandler_v2

# Initialize with a shared router cache
cache = DualCache()
handler = LowestTPMLoggingHandler_v2(router_cache=cache, routing_args={"ttl": 60})

# Pre-call check to enforce RPM limits (raises RateLimitError if exceeded)
deployment = {"model_info": {"id": "abc123"}, "litellm_params": {"model": "gpt-4"}, "rpm": 100}
checked = handler.pre_call_check(deployment=deployment)

# Select the deployment with the lowest TPM usage
selected = handler.get_available_deployments(
    model_group="gpt-4",
    healthy_deployments=[
        {"model_info": {"id": "abc123"}, "litellm_params": {"model": "gpt-4"}, "tpm": 100000, "rpm": 500},
        {"model_info": {"id": "def456"}, "litellm_params": {"model": "gpt-4"}, "tpm": 200000, "rpm": 1000},
    ],
    messages=[{"role": "user", "content": "Hello"}],
)

Related Pages

• BerriAI_Litellm_Lowest_TPM_RPM_Strategy - Original V1 TPM/RPM routing strategy
• BerriAI_Litellm_Lowest_Cost_Strategy - Cost-based deployment selection strategy
• BerriAI_Litellm_Least_Busy_Strategy - Least busy deployment selection strategy
• BerriAI_Litellm_Simple_Shuffle_Strategy - Random/weighted shuffle deployment selection