From 230f653523fa55aa446e1a5b6469f0871cf7278d Mon Sep 17 00:00:00 2001 From: joshua-berri Date: Tue, 20 Oct 2026 22:50:28 -0700 Subject: [PATCH] fix(router): normalize native Responses latency per output token (#47727) * fix(router): normalize native Responses latency per output token * fix(router): isolate normalized latency samples during upgrades --------- Co-authored-by: Joshua Valluru <326636767+joshua-berri@users.noreply.github.com> --- litellm/router_strategy/lowest_latency.py | 61 +++-- .../router_strategy/test_lowest_latency.py | 222 ++++++++++++++++-- .../test_router_routing_groups.py | 4 +- 3 files changed, 249 insertions(+), 38 deletions(-) diff --git a/litellm/router_strategy/lowest_latency.py b/litellm/router_strategy/lowest_latency.py index 83bb22435e6..28ed062f1ec 100644 --- a/litellm/router_strategy/lowest_latency.py +++ b/litellm/router_strategy/lowest_latency.py @@ -14,7 +14,9 @@ from litellm._internal_context import with_service_target from litellm.caching.caching import DualCache from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.core_helpers import get_parent_otel_span_from_kwargs, safe_divide_seconds +from litellm.responses.utils import ResponseAPILoggingUtils from litellm.router_utils.batch_utils import is_batch_retrieve_call_type +from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.utils import LiteLLMPydanticObjectBase if TYPE_CHECKING: @@ -86,7 +88,7 @@ class LowestLatencyLoggingHandler(CustomLogger): { {model_group}_map: { id: { - "latency": [..] + "latency_normalized": [..] f"{date:hour:minute}" : {"tpm": 34, "rpm": 3} } } @@ -116,11 +118,16 @@ class LowestLatencyLoggingHandler(CustomLogger): final_value: float = response_ms total_tokens = 0 - if isinstance(response_obj, ModelResponse): - _usage: Final = getattr(response_obj, "usage", None) + if isinstance(response_obj, (ModelResponse, ResponsesAPIResponse)): + _usage: Final = ( + ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(response_obj.usage) + if isinstance(response_obj, ResponsesAPIResponse) + else getattr(response_obj, "usage", None) + ) if _usage is not None: completion_tokens: Final = _usage.completion_tokens - total_tokens = _usage.total_tokens + if isinstance(response_obj, ModelResponse): + total_tokens = _usage.total_tokens # response_ms is already normalized to float seconds above response_seconds: Final = response_ms @@ -143,10 +150,12 @@ class LowestLatencyLoggingHandler(CustomLogger): request_count_dict[id] = {} ## Latency - if len(request_count_dict[id].get("latency", [])) < self.routing_args.max_latency_list_size: - request_count_dict[id].setdefault("latency", []).append(final_value) + if len(request_count_dict[id].get("latency_normalized", [])) < self.routing_args.max_latency_list_size: + request_count_dict[id].setdefault("latency_normalized", []).append(final_value) else: - request_count_dict[id]["latency"] = request_count_dict[id]["latency"][1:] + [final_value] + request_count_dict[id]["latency_normalized"] = request_count_dict[id]["latency_normalized"][1:] + [ + final_value + ] ## Time to first token if time_to_first_token is not None: @@ -212,7 +221,7 @@ class LowestLatencyLoggingHandler(CustomLogger): { {model_group}_map: { id: { - "latency": [..] + "latency_normalized": [..] f"{date:hour:minute}" : {"tpm": 34, "rpm": 3} } } @@ -225,10 +234,15 @@ class LowestLatencyLoggingHandler(CustomLogger): request_count_dict[id] = {} ## Latency - give 1000s penalty for failing - if len(request_count_dict[id].get("latency", [])) < self.routing_args.max_latency_list_size: - request_count_dict[id].setdefault("latency", []).append(1000.0) + if ( + len(request_count_dict[id].get("latency_normalized", [])) + < self.routing_args.max_latency_list_size + ): + request_count_dict[id].setdefault("latency_normalized", []).append(1000.0) else: - request_count_dict[id]["latency"] = request_count_dict[id]["latency"][1:] + [1000.0] + request_count_dict[id]["latency_normalized"] = request_count_dict[id]["latency_normalized"][ + 1: + ] + [1000.0] await self.router_cache.async_set_cache( key=latency_key, @@ -270,7 +284,7 @@ class LowestLatencyLoggingHandler(CustomLogger): { {model_group}_map: { id: { - "latency": [..] + "latency_normalized": [..] "time_to_first_token_seconds": [..] f"{date:hour:minute}" : {"tpm": 34, "rpm": 3} } @@ -300,11 +314,16 @@ class LowestLatencyLoggingHandler(CustomLogger): final_value: float = response_ms total_tokens = 0 - if isinstance(response_obj, ModelResponse): - _usage: Final = getattr(response_obj, "usage", None) + if isinstance(response_obj, (ModelResponse, ResponsesAPIResponse)): + _usage: Final = ( + ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(response_obj.usage) + if isinstance(response_obj, ResponsesAPIResponse) + else getattr(response_obj, "usage", None) + ) if _usage is not None: completion_tokens: Final = _usage.completion_tokens - total_tokens = _usage.total_tokens + if isinstance(response_obj, ModelResponse): + total_tokens = _usage.total_tokens # response_ms is already normalized to float seconds above response_seconds: Final = response_ms @@ -331,10 +350,12 @@ class LowestLatencyLoggingHandler(CustomLogger): request_count_dict[id] = {} ## Latency - if len(request_count_dict[id].get("latency", [])) < self.routing_args.max_latency_list_size: - request_count_dict[id].setdefault("latency", []).append(final_value) + if len(request_count_dict[id].get("latency_normalized", [])) < self.routing_args.max_latency_list_size: + request_count_dict[id].setdefault("latency_normalized", []).append(final_value) else: - request_count_dict[id]["latency"] = request_count_dict[id]["latency"][1:] + [final_value] + request_count_dict[id]["latency_normalized"] = request_count_dict[id]["latency_normalized"][1:] + [ + final_value + ] ## Time to first token if time_to_first_token is not None: @@ -403,7 +424,7 @@ class LowestLatencyLoggingHandler(CustomLogger): ## if healthy deployment not yet used if d["model_info"]["id"] not in all_deployments: all_deployments[d["model_info"]["id"]] = { - "latency": [0], + "latency_normalized": [0], precise_minute: {"tpm": 0, "rpm": 0}, } @@ -443,7 +464,7 @@ class LowestLatencyLoggingHandler(CustomLogger): or _deployment.get("model_info", {}).get("rpm", None) or float("inf") ) - item_latency = item_map.get("latency", []) + item_latency = item_map.get("latency_normalized", []) item_ttft_latency = item_map.get("time_to_first_token_seconds", []) item_rpm = item_map.get(precise_minute, {}).get("rpm", 0) item_tpm = item_map.get(precise_minute, {}).get("tpm", 0) diff --git a/tests/unit/router_strategy/test_lowest_latency.py b/tests/unit/router_strategy/test_lowest_latency.py index 0a3a92691ba..9af5def0d56 100644 --- a/tests/unit/router_strategy/test_lowest_latency.py +++ b/tests/unit/router_strategy/test_lowest_latency.py @@ -17,8 +17,10 @@ from pydantic import ValidationError import litellm from litellm.caching.caching import DualCache from litellm.caching.in_memory_cache import InMemoryCache +from litellm.responses.utils import ResponseAPILoggingUtils from litellm.router import Router from litellm.router_strategy.lowest_latency import LowestLatencyLoggingHandler, RoutingArgs +from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse DEPLOYMENT_ID = "9876" KWARGS = { @@ -43,7 +45,7 @@ def _embedding_response() -> litellm.EmbeddingResponse: def _recorded_latencies(cache: DualCache): cached = cache.get_cache(key="gemini-embedding-001_map") or {} - return cached.get(DEPLOYMENT_ID, {}).get("latency", []) + return cached.get(DEPLOYMENT_ID, {}).get("latency_normalized", []) def test_sync_embedding_latency_is_json_serializable(): @@ -69,7 +71,7 @@ def test_sync_embedding_latency_is_json_serializable(): ) assert latencies[-1] == pytest.approx(2.0) # the exact failure mode from production: redis cache sync json.dumps - json.dumps({"latency": latencies}) + json.dumps({"latency_normalized": latencies}) @pytest.mark.asyncio @@ -94,7 +96,7 @@ async def test_async_embedding_latency_is_json_serializable(): f"raw timedelta leaked into latency list: {latencies}" ) assert latencies[-1] == pytest.approx(3.0) - json.dumps({"latency": latencies}) + json.dumps({"latency_normalized": latencies}) def _chat_response(completion_tokens: int) -> litellm.ModelResponse: @@ -131,7 +133,7 @@ async def test_async_chat_latency_normalized_per_token(): latencies = _recorded_latencies(cache) assert latencies and latencies[-1] == pytest.approx(0.5) # 2s / 4 tokens - json.dumps({"latency": latencies}) + json.dumps({"latency_normalized": latencies}) @pytest.mark.asyncio @@ -151,7 +153,7 @@ async def test_async_chat_zero_completion_tokens_falls_back_to_seconds(): latencies = _recorded_latencies(cache) assert latencies and latencies[-1] == pytest.approx(3.0) assert not isinstance(latencies[-1], timedelta) - json.dumps({"latency": latencies}) + json.dumps({"latency_normalized": latencies}) def test_sync_chat_zero_completion_tokens_falls_back_to_seconds(): @@ -168,7 +170,7 @@ def test_sync_chat_zero_completion_tokens_falls_back_to_seconds(): latencies = _recorded_latencies(cache) assert latencies and latencies[-1] == pytest.approx(2.0) assert not isinstance(latencies[-1], timedelta) - json.dumps({"latency": latencies}) + json.dumps({"latency_normalized": latencies}) MODEL_GROUP = "gpt-4o-mini" @@ -362,8 +364,8 @@ async def test_ttft_percentile_does_not_change_non_streaming_routing(): cache.set_cache( key=f"{MODEL_GROUP}_map", value={ - FAST_TTFT_ID: {"latency": [1.0], "time_to_first_token_seconds": [0.1]}, - SLOW_TTFT_ID: {"latency": [0.2], "time_to_first_token_seconds": [1.5]}, + FAST_TTFT_ID: {"latency_normalized": [1.0], "time_to_first_token_seconds": [0.1]}, + SLOW_TTFT_ID: {"latency_normalized": [0.2], "time_to_first_token_seconds": [1.5]}, }, ) @@ -380,7 +382,7 @@ async def test_ttft_percentile_does_not_change_non_streaming_routing(): @pytest.mark.asyncio @pytest.mark.parametrize( "cached_entry", - [{"latency": []}, {"2026-09-05-15-39": {"tpm": 28, "rpm": 1}}], + [{"latency_normalized": []}, {"2026-09-05-15-39": {"tpm": 28, "rpm": 1}}], ids=["empty_latency_list", "minute_bucket_only_as_cost_based_routing_writes"], ) async def test_async_get_available_deployments_treats_missing_samples_as_zero_latency(cached_entry): @@ -388,7 +390,7 @@ async def test_async_get_available_deployments_treats_missing_samples_as_zero_la handler = LowestLatencyLoggingHandler(router_cache=cache) cache.set_cache( key="gemini-embedding-001_map", - value={DEPLOYMENT_ID: cached_entry, "slower": {"latency": [0.5]}}, + value={DEPLOYMENT_ID: cached_entry, "slower": {"latency_normalized": [0.5]}}, ) healthy_deployments = [ {"model_info": {"id": DEPLOYMENT_ID}, "litellm_params": {}}, @@ -605,7 +607,7 @@ def test_latency_updated(): end_time=end_time, ) latency_key = f"{model_group}_map" - assert end_time - start_time == test_cache.get_cache(key=latency_key)[deployment_id]["latency"][0] + assert end_time - start_time == test_cache.get_cache(key=latency_key)[deployment_id]["latency_normalized"][0] @pytest.mark.usefixtures("frozen_latency_clock") @@ -1002,7 +1004,7 @@ def test_latency_list_trimming_discards_oldest_entry(): latency_key = f"{model_group}_map" cached_data = test_cache.get_cache(key=latency_key) - latency_list = cached_data[deployment_id]["latency"] + latency_list = cached_data[deployment_id]["latency_normalized"] assert len(latency_list) == max_size, f"Expected {max_size} entries, got {len(latency_list)}" @@ -1062,7 +1064,7 @@ async def test_latency_list_trimming_discards_oldest_entry_async(): latency_key = f"{model_group}_map" cached_data = await test_cache.async_get_cache(key=latency_key) - latency_list = cached_data[deployment_id]["latency"] + latency_list = cached_data[deployment_id]["latency_normalized"] assert len(latency_list) == max_size @@ -1129,7 +1131,7 @@ async def test_timeout_penalty_discards_oldest_entry(): latency_key = f"{model_group}_map" cached_data = await test_cache.async_get_cache(key=latency_key) - latency_list = cached_data[deployment_id]["latency"] + latency_list = cached_data[deployment_id]["latency_normalized"] assert len(latency_list) == max_size @@ -1181,7 +1183,7 @@ def test_list_order_preserved_after_multiple_trims(): latency_key = f"{model_group}_map" cached_data = test_cache.get_cache(key=latency_key) - latency_list = cached_data[deployment_id]["latency"] + latency_list = cached_data[deployment_id]["latency_normalized"] assert len(latency_list) == max_size @@ -1423,7 +1425,7 @@ def test_latency_cache_honors_custom_ttl() -> None: latency_key: Final = "ttl-group_map" cached: Final = cache.get_cache(key=latency_key) assert isinstance(cached, dict) - assert cached["deployment"]["latency"] == [1.0] + assert cached["deployment"]["latency_normalized"] == [1.0] assert cache.in_memory_cache.ttl_dict[latency_key] == 103.0 clock.return_value = 104.0 @@ -1461,3 +1463,191 @@ async def test_router_model_group_usage_increases_by_logged_tokens() -> None: updated_usage: Final = await router.get_model_group_usage(model_group=model_group) assert updated_usage[0] == initial_usage[0] + 15 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_callback", [False, True]) +@pytest.mark.parametrize("normalized_usage", [False, True]) +@pytest.mark.parametrize("native_seconds", [2, 3]) +async def test_responses_and_chat_latency_use_comparable_token_units( + async_callback: bool, normalized_usage: bool, native_seconds: int +) -> None: + cache: Final = DualCache() + handler: Final = LowestLatencyLoggingHandler(router_cache=cache) + usage: Final = ResponseAPIUsage(input_tokens=10, output_tokens=100, total_tokens=110) + native: Final = ResponsesAPIResponse(id="resp-latency", created_at=0, output=[], usage=usage) + logged_native: Final = ( + native.model_copy(update={"usage": ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage)}) + if normalized_usage + else native + ) + original_usage: Final = logged_native.usage + assert original_usage is not None + original_fields: Final = original_usage.model_dump() + start: Final = datetime(2026, 1, 1) + for deployment_id, response, seconds in ( + ("native", logged_native, native_seconds), + ("bridge", _chat_response(100), 3), + ): + kwargs: Final = { + "call_type": "aresponses", + "litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": deployment_id}}, + } + if async_callback: + await handler.async_log_success_event(kwargs, response, start, start + timedelta(seconds=seconds)) + else: + handler.log_success_event(kwargs, response, start, start + timedelta(seconds=seconds)) + recorded: Final = cache.get_cache(key="mixed_map") + assert recorded["native"]["latency_normalized"] == pytest.approx([native_seconds / 100]) + assert recorded["bridge"]["latency_normalized"] == pytest.approx([3 / 100]) + assert logged_native.usage is original_usage + assert logged_native.usage.model_dump() == original_fields + assert native.usage is usage + deployments: Final = [ + {"model_name": "mixed", "model_info": {"id": name}, "litellm_params": {}} for name in ("native", "bridge") + ] + picked: Final = handler.get_available_deployments( + model_group="mixed", healthy_deployments=deployments, request_kwargs={"stream": False, "metadata": {}} + ) + assert picked["model_info"]["id"] in ({"native"} if native_seconds < 3 else {"native", "bridge"}) + native_counts: Final = tuple(value for key, value in recorded["native"].items() if key != "latency_normalized") + assert native_counts == ({"tpm": 0, "rpm": 1},) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_callback", [False, True]) +@pytest.mark.parametrize( + "response", + [ + litellm.ModelResponse(), + ResponsesAPIResponse(id="resp-missing", created_at=0, output=[], usage=None), + ResponsesAPIResponse( + id="resp-zero", created_at=0, output=[], + usage=ResponseAPIUsage(input_tokens=10, output_tokens=0, total_tokens=10), + ), + ], +) +async def test_responses_without_output_tokens_keep_elapsed_seconds( + async_callback: bool, response: litellm.ModelResponse | ResponsesAPIResponse +) -> None: + cache: Final = DualCache() + handler: Final = LowestLatencyLoggingHandler(router_cache=cache) + original_usage: Final = getattr(response, "usage", None) + if async_callback: + await handler.async_log_success_event(KWARGS, response, 1.0, 3.0) + else: + handler.log_success_event(KWARGS, response, 1.0, 3.0) + assert _recorded_latencies(cache) == [2.0] + assert getattr(response, "usage", None) is original_usage + + +@pytest.mark.usefixtures("frozen_latency_clock") +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("preserve_new_samples", [False, True]) +async def test_legacy_latency_does_not_starve_deployments_after_upgrade( + async_mode: bool, preserve_new_samples: bool +) -> None: + cache: Final = DualCache() + handler: Final = LowestLatencyLoggingHandler(router_cache=cache) + minute: Final = FROZEN_NOW.strftime("%Y-%m-%d-%H-%M") + legacy: Final = { + "native": {"latency": [2.0], "time_to_first_token_seconds": [0.1], minute: {"tpm": 0, "rpm": 1}}, + "bridge": {"latency": [0.03], "time_to_first_token_seconds": [0.2], minute: {"tpm": 110, "rpm": 1}}, + } + cache.set_cache(key="mixed_map", value=copy.deepcopy(legacy)) + deployments: Final = [ + {"model_name": "mixed", "model_info": {"id": name}, "litellm_params": {}} for name in ("native", "bridge") + ] + native: Final = ResponsesAPIResponse( + id="resp-upgrade", created_at=0, output=[], + usage=ResponseAPIUsage(input_tokens=10, output_tokens=100, total_tokens=110), + ) + for iteration in range(2): + bridge_kwargs: Final = { + "litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": "bridge"}}, + } + if async_mode: + await handler.async_log_success_event(bridge_kwargs, _chat_response(100), 0.0, 3.0) + else: + handler.log_success_event(bridge_kwargs, _chat_response(100), 0.0, 3.0) + selected: Final = ( + await handler.async_get_available_deployments(model_group="mixed", healthy_deployments=deployments) + if async_mode + else handler.get_available_deployments(model_group="mixed", healthy_deployments=deployments) + ) + assert selected["model_info"]["id"] == "native" + native_kwargs: Final = { + "litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": "native"}}, + } + if async_mode: + await handler.async_log_success_event(native_kwargs, native, 0.0, 2.0) + else: + handler.log_success_event(native_kwargs, native, 0.0, 2.0) + recorded: Final = copy.deepcopy(cache.get_cache(key="mixed_map")) + assert recorded["native"]["latency_normalized"] == pytest.approx( + [0.02] * (iteration + 1 if preserve_new_samples else 1) + ) + assert recorded["native"]["latency"] == [2.0] + assert recorded["native"]["time_to_first_token_seconds"] == [0.1] + assert recorded["bridge"]["time_to_first_token_seconds"] == [0.2] + assert recorded["native"][minute] == {"tpm": 0, "rpm": iteration + 2} + assert recorded["bridge"][minute] == {"tpm": 110 + (iteration + 1) * 110, "rpm": iteration + 2} + streaming: Final = handler.get_available_deployments( + model_group="mixed", healthy_deployments=deployments, request_kwargs={"stream": True} + ) + assert streaming["model_info"]["id"] == "native" + cache.set_cache( + key="mixed_map", + value={ + name: {**(recorded[name] if preserve_new_samples else legacy[name]), minute: recorded[name][minute]} + for name in ("native", "bridge") + }, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_normalized_latency_window_evicts_samples_without_changing_legacy_data(async_mode: bool) -> None: + cache: Final = DualCache() + cache.set_cache(key="gemini-embedding-001_map", value={DEPLOYMENT_ID: {"latency": [2.0]}}) + handler: Final = LowestLatencyLoggingHandler(router_cache=cache, routing_args={"max_latency_list_size": 2}) + for elapsed, expected in ((2.0, [0.02]), (3.0, [0.02, 0.03]), (4.0, [0.03, 0.04])): + if async_mode: + await handler.async_log_success_event(KWARGS, _chat_response(100), 0.0, elapsed) + else: + handler.log_success_event(KWARGS, _chat_response(100), 0.0, elapsed) + recorded: Final = cache.get_cache(key="gemini-embedding-001_map")[DEPLOYMENT_ID] + assert recorded["latency_normalized"] == pytest.approx(expected) + assert recorded["latency"] == [2.0] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("existing_samples", [(), (0.02, 0.03)]) +async def test_timeout_penalty_updates_normalized_latency_and_avoids_failed_deployment( + existing_samples: tuple[float, ...], +) -> None: + cache: Final = DualCache() + handler: Final = LowestLatencyLoggingHandler(router_cache=cache, routing_args={"max_latency_list_size": 2}) + cache.set_cache( + key="mixed_map", + value={ + "native": {"latency": [0.001], "latency_normalized": list(existing_samples)}, + "bridge": {"latency": [2.0], "latency_normalized": [0.04]}, + }, + ) + await handler.async_log_failure_event( + kwargs={ + "exception": litellm.Timeout(message="request timed out", model="test", llm_provider="openai"), + "litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": "native"}}, + }, + response_obj=None, start_time=0.0, end_time=3.0, + ) + recorded: Final = cache.get_cache(key="mixed_map")["native"] + assert recorded["latency_normalized"] == [*existing_samples[-1:], 1000.0] + assert recorded["latency"] == [0.001] + selected: Final = await handler.async_get_available_deployments( + model_group="mixed", + healthy_deployments=[{"model_info": {"id": name}, "litellm_params": {}} for name in ("native", "bridge")], + ) + assert selected["model_info"]["id"] == "bridge" diff --git a/tests/unit/router_strategy/test_router_routing_groups.py b/tests/unit/router_strategy/test_router_routing_groups.py index 7e4b470f1c1..7235333f59f 100644 --- a/tests/unit/router_strategy/test_router_routing_groups.py +++ b/tests/unit/router_strategy/test_router_routing_groups.py @@ -1249,7 +1249,7 @@ async def test_latency_based_override_learns_from_the_overriding_requests(): ) def samples(deployment_id: str) -> list[float]: - recorded = (router.cache.get_cache("grp_map") or {}).get(deployment_id, {}).get("latency", []) + recorded = (router.cache.get_cache("grp_map") or {}).get(deployment_id, {}).get("latency_normalized", []) return [latency for latency in recorded if latency > 0] async def overriding_call() -> str: @@ -1316,7 +1316,7 @@ async def test_override_matching_a_routing_group_strategy_records_each_request_o deployment_id = response._hidden_params["model_id"] await _async_wait_until(lambda: (router.cache.get_cache("grp_map") or {}).get(deployment_id) is not None) - assert len(router.cache.get_cache("grp_map")[deployment_id]["latency"]) == 1 + assert len(router.cache.get_cache("grp_map")[deployment_id]["latency_normalized"]) == 1 assert request["litellm_logging_obj"].dynamic_success_callbacks is None