From 95c0f9db7d92cefafa5e2d9799786b4931c9a390 Mon Sep 17 00:00:00 2001 From: Clement Date: Wed, 9 Sep 2026 01:05:23 +0800 Subject: [PATCH] fix(router): rank streaming latency routing by raw TTFT, not TTFT per token (#40202) * fix(router): rank streaming latency routing by raw TTFT, not TTFT per token Latency-based routing divided time-to-first-token by completion_tokens before storing it, so a deployment that streamed a long answer looked faster to first token than one that answered briefly. TTFT is now stored as plain seconds (first token time minus request start) in both the sync and async success handlers, which is what the routing decision compares. Non-streaming latency normalization per output token is unchanged. Claude-Session: https://claude.ai/code/session_01Ttd5Q9ZhRPB4ch5guos3rj * fix(router): store streaming TTFT under a seconds-only cache key Workers on the previous release keep writing seconds-per-token samples under "time_to_first_token" in the shared router cache during a rolling deploy, so mixing the new raw-seconds samples into the same list averaged incompatible units. Raw TTFT now lives under "time_to_first_token_seconds" and routing reads only that key. Also fix the regression test's token counts: with 50 tokens on the fast deployment and 500 on the slow one the old per-token formula picks the slow deployment, so the routing assertion now catches the bug. Claude-Session: https://claude.ai/code/session_01Ttd5Q9ZhRPB4ch5guos3rj * test(router): cover the TTFT sliding window from the unit-test shard Move the TTFT list trimming checks from the CircleCI-only suite into the mapped unit test file as one sync/async parametrized test, so the changed lines in lowest_latency.py are exercised by the GitHub unit-test shard that reports patch coverage. Claude-Session: https://claude.ai/code/session_01Ttd5Q9ZhRPB4ch5guos3rj --- litellm/router_strategy/lowest_latency.py | 54 +++---- .../test_lowest_latency_routing.py | 136 ------------------ .../router_strategy/test_lowest_latency.py | 127 ++++++++++++++++ 3 files changed, 149 insertions(+), 168 deletions(-) diff --git a/litellm/router_strategy/lowest_latency.py b/litellm/router_strategy/lowest_latency.py index bb6877a032b..805d4ff9080 100644 --- a/litellm/router_strategy/lowest_latency.py +++ b/litellm/router_strategy/lowest_latency.py @@ -32,6 +32,12 @@ def _average_latency(samples: Sequence[float]) -> float: return sum(samples) / len(samples) +def _ttft_seconds(elapsed: timedelta | float) -> float: + if isinstance(elapsed, timedelta): + return elapsed.total_seconds() + return float(elapsed) + + class LowestLatencyLoggingHandler(CustomLogger): test_flag: bool = False logged_success: int = 0 @@ -86,14 +92,13 @@ class LowestLatencyLoggingHandler(CustomLogger): # breaks JSON serialization when the router cache syncs to # Redis (issue #33169) response_ms = response_ms.total_seconds() - time_to_first_token_response_time = None + time_to_first_token: float | None = None if kwargs.get("stream", None) is not None and kwargs["stream"] is True: # only log ttft for streaming request - time_to_first_token_response_time = kwargs.get("completion_start_time", end_time) - start_time + time_to_first_token = _ttft_seconds(kwargs.get("completion_start_time", end_time) - start_time) final_value: float = response_ms - time_to_first_token: float | None = None total_tokens = 0 if isinstance(response_obj, ModelResponse): @@ -111,13 +116,6 @@ class LowestLatencyLoggingHandler(CustomLogger): else: final_value = response_seconds - if time_to_first_token_response_time is not None: - if isinstance(time_to_first_token_response_time, timedelta): - ttft_seconds = time_to_first_token_response_time.total_seconds() - else: - ttft_seconds = time_to_first_token_response_time - time_to_first_token = safe_divide_seconds(ttft_seconds, completion_tokens) - # ------------ # Update usage # ------------ @@ -138,14 +136,14 @@ class LowestLatencyLoggingHandler(CustomLogger): ## Time to first token if time_to_first_token is not None: if ( - len(request_count_dict[id].get("time_to_first_token", [])) + len(request_count_dict[id].get("time_to_first_token_seconds", [])) < self.routing_args.max_latency_list_size ): - request_count_dict[id].setdefault("time_to_first_token", []).append(time_to_first_token) + request_count_dict[id].setdefault("time_to_first_token_seconds", []).append(time_to_first_token) else: - request_count_dict[id]["time_to_first_token"] = request_count_dict[id]["time_to_first_token"][ - 1: - ] + [time_to_first_token] + request_count_dict[id]["time_to_first_token_seconds"] = request_count_dict[id][ + "time_to_first_token_seconds" + ][1:] + [time_to_first_token] if precise_minute not in request_count_dict[id]: request_count_dict[id][precise_minute] = {} @@ -252,7 +250,7 @@ class LowestLatencyLoggingHandler(CustomLogger): {model_group}_map: { id: { "latency": [..] - "time_to_first_token": [..] + "time_to_first_token_seconds": [..] f"{date:hour:minute}" : {"tpm": 34, "rpm": 3} } } @@ -273,14 +271,13 @@ class LowestLatencyLoggingHandler(CustomLogger): # breaks JSON serialization when the router cache syncs to # Redis (issue #33169) response_ms = response_ms.total_seconds() - time_to_first_token_response_time = None + time_to_first_token: float | None = None if kwargs.get("stream", None) is not None and kwargs["stream"] is True: # only log ttft for streaming request - time_to_first_token_response_time = kwargs.get("completion_start_time", end_time) - start_time + time_to_first_token = _ttft_seconds(kwargs.get("completion_start_time", end_time) - start_time) final_value: float = response_ms total_tokens = 0 - time_to_first_token: float | None = None if isinstance(response_obj, ModelResponse): _usage: Final = getattr(response_obj, "usage", None) @@ -296,13 +293,6 @@ class LowestLatencyLoggingHandler(CustomLogger): final_value = float(normalized_value) else: final_value = response_seconds - - if time_to_first_token_response_time is not None: - if isinstance(time_to_first_token_response_time, timedelta): - ttft_seconds = time_to_first_token_response_time.total_seconds() - else: - ttft_seconds = time_to_first_token_response_time - time_to_first_token = safe_divide_seconds(ttft_seconds, completion_tokens) # ------------ # Update usage # ------------ @@ -328,14 +318,14 @@ class LowestLatencyLoggingHandler(CustomLogger): ## Time to first token if time_to_first_token is not None: if ( - len(request_count_dict[id].get("time_to_first_token", [])) + len(request_count_dict[id].get("time_to_first_token_seconds", [])) < self.routing_args.max_latency_list_size ): - request_count_dict[id].setdefault("time_to_first_token", []).append(time_to_first_token) + request_count_dict[id].setdefault("time_to_first_token_seconds", []).append(time_to_first_token) else: - request_count_dict[id]["time_to_first_token"] = request_count_dict[id]["time_to_first_token"][ - 1: - ] + [time_to_first_token] + request_count_dict[id]["time_to_first_token_seconds"] = request_count_dict[id][ + "time_to_first_token_seconds" + ][1:] + [time_to_first_token] if precise_minute not in request_count_dict[id]: request_count_dict[id][precise_minute] = {} @@ -433,7 +423,7 @@ class LowestLatencyLoggingHandler(CustomLogger): or float("inf") ) item_latency = item_map.get("latency", []) - item_ttft_latency = item_map.get("time_to_first_token", []) + 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/local_testing/test_lowest_latency_routing.py b/tests/local_testing/test_lowest_latency_routing.py index 598b1dbcaf9..aba4500199a 100644 --- a/tests/local_testing/test_lowest_latency_routing.py +++ b/tests/local_testing/test_lowest_latency_routing.py @@ -1077,73 +1077,6 @@ async def test_latency_list_trimming_discards_oldest_entry_async(): ), f"Oldest latency {oldest_latency} should have been discarded" -def test_ttft_list_trimming_discards_oldest_entry(): - """ - The time_to_first_token list trims the oldest entry when full, matching - the behavior of the latency list. - """ - max_size = 3 - test_cache = DualCache() - lowest_latency_logger = LowestLatencyLoggingHandler( - router_cache=test_cache, routing_args={"max_latency_list_size": max_size} - ) - - model_group = "gpt-3.5-turbo" - deployment_id = "test-deployment" - - ttft_values = [] - for i in range(max_size + 1): - start_time = time.time() - expected_ttft = float(i + 1) * 0.1 # 0.1, 0.2, 0.3, 0.4 - completion_start_time = start_time + expected_ttft - end_time = start_time + float(i + 1) - ttft_values.append(expected_ttft) - - kwargs = { - "litellm_params": { - "metadata": { - "model_group": model_group, - "deployment": "azure/gpt-4.1-mini", - }, - "model_info": {"id": deployment_id}, - }, - "stream": True, - "completion_start_time": completion_start_time, - } - # TTFT is only recorded when response_obj is a ModelResponse. - response_obj = litellm.ModelResponse( - usage=litellm.Usage(completion_tokens=1, total_tokens=1) - ) - - lowest_latency_logger.log_success_event( - response_obj=response_obj, - kwargs=kwargs, - start_time=start_time, - end_time=end_time, - ) - - latency_key = f"{model_group}_map" - cached_data = test_cache.get_cache(key=latency_key) - ttft_list = cached_data[deployment_id].get("time_to_first_token", []) - - assert ( - len(ttft_list) == max_size - ), f"Expected {max_size} entries, got {len(ttft_list)}" - - newest_ttft = ttft_values[-1] - oldest_ttft = ttft_values[0] - tolerance = 0.05 - - assert ( - abs(ttft_list[-1] - newest_ttft) < tolerance - ), f"Newest TTFT {newest_ttft} should be at end of list" - - for ttft in ttft_list: - assert ( - abs(ttft - oldest_ttft) > tolerance - ), f"Oldest TTFT {oldest_ttft} should have been discarded" - - @pytest.mark.asyncio async def test_timeout_penalty_discards_oldest_entry(): """ @@ -1269,72 +1202,3 @@ def test_list_order_preserved_after_multiple_trims(): assert ( abs(latency_list[i] - expected) < tolerance ), f"At index {i}, expected ~{expected}, got {latency_list[i]}" - - -@pytest.mark.asyncio -async def test_ttft_list_trimming_discards_oldest_entry_async(): - """ - Async counterpart: the time_to_first_token list trims the oldest entry - when full. Exercises the async_log_success_event TTFT path, which only - runs when response_obj is a ModelResponse and the call is marked as - streaming with a completion_start_time. - """ - max_size = 3 - test_cache = DualCache() - lowest_latency_logger = LowestLatencyLoggingHandler( - router_cache=test_cache, routing_args={"max_latency_list_size": max_size} - ) - - model_group = "gpt-3.5-turbo" - deployment_id = "test-deployment" - - ttft_values = [] - for i in range(max_size + 1): - start_time = time.time() - expected_ttft = float(i + 1) * 0.1 # 0.1, 0.2, 0.3, 0.4 - completion_start_time = start_time + expected_ttft - end_time = start_time + float(i + 1) - ttft_values.append(expected_ttft) - - kwargs = { - "litellm_params": { - "metadata": { - "model_group": model_group, - "deployment": "azure/gpt-4.1-mini", - }, - "model_info": {"id": deployment_id}, - }, - "stream": True, - "completion_start_time": completion_start_time, - } - response_obj = litellm.ModelResponse( - usage=litellm.Usage(completion_tokens=1, total_tokens=1) - ) - - await lowest_latency_logger.async_log_success_event( - response_obj=response_obj, - kwargs=kwargs, - start_time=start_time, - end_time=end_time, - ) - - latency_key = f"{model_group}_map" - cached_data = await test_cache.async_get_cache(key=latency_key) - ttft_list = cached_data[deployment_id].get("time_to_first_token", []) - - assert ( - len(ttft_list) == max_size - ), f"Expected {max_size} entries, got {len(ttft_list)}" - - newest_ttft = ttft_values[-1] - oldest_ttft = ttft_values[0] - tolerance = 0.05 - - assert ( - abs(ttft_list[-1] - newest_ttft) < tolerance - ), f"Newest TTFT {newest_ttft} should be at end of list" - - for ttft in ttft_list: - assert ( - abs(ttft - oldest_ttft) > tolerance - ), f"Oldest TTFT {oldest_ttft} should have been discarded" diff --git a/tests/test_litellm/router_strategy/test_lowest_latency.py b/tests/test_litellm/router_strategy/test_lowest_latency.py index eb02459be68..812d7bbff32 100644 --- a/tests/test_litellm/router_strategy/test_lowest_latency.py +++ b/tests/test_litellm/router_strategy/test_lowest_latency.py @@ -165,6 +165,133 @@ def test_sync_chat_zero_completion_tokens_falls_back_to_seconds(): json.dumps({"latency": latencies}) +MODEL_GROUP = "gpt-4o-mini" +FAST_TTFT_ID = "fast-ttft-short-output" +SLOW_TTFT_ID = "slow-ttft-long-output" +STREAMING_DEPLOYMENTS = [ + {"model_info": {"id": FAST_TTFT_ID}, "litellm_params": {}}, + {"model_info": {"id": SLOW_TTFT_ID}, "litellm_params": {}}, +] + + +def _streaming_kwargs(deployment_id: str, start_time: datetime, ttft_seconds: float): + return { + "litellm_params": { + "metadata": {"model_group": MODEL_GROUP}, + "model_info": {"id": deployment_id}, + }, + "stream": True, + "completion_start_time": start_time + timedelta(seconds=ttft_seconds), + } + + +def _recorded_ttft(cache: DualCache, deployment_id: str): + cached = cache.get_cache(key=f"{MODEL_GROUP}_map") or {} + return cached.get(deployment_id, {}).get("time_to_first_token_seconds", []) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"]) +async def test_streaming_ttft_ranking_ignores_completion_length(sync_mode: bool): + """Deployment A: TTFT 1s, 50 completion tokens. Deployment B: TTFT 3s, 500 + completion tokens. Dividing TTFT by completion tokens made B look faster + (3/500 = 0.006 beats 1/50 = 0.02); actual TTFT must win.""" + cache = DualCache() + handler = LowestLatencyLoggingHandler(router_cache=cache) + start_time = datetime(2026, 1, 1, 12, 0, 0) + end_time = start_time + timedelta(seconds=10) + + samples = ( + (FAST_TTFT_ID, 1.0, 50), + (SLOW_TTFT_ID, 3.0, 500), + ) + for deployment_id, ttft, completion_tokens in samples: + kwargs = _streaming_kwargs(deployment_id, start_time, ttft) + response_obj = _chat_response(completion_tokens=completion_tokens) + if sync_mode: + handler.log_success_event( + response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=end_time + ) + else: + await handler.async_log_success_event( + response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=end_time + ) + + assert _recorded_ttft(cache, FAST_TTFT_ID) == [pytest.approx(1.0)] + assert _recorded_ttft(cache, SLOW_TTFT_ID) == [pytest.approx(3.0)] + + request_kwargs = {"stream": True, "metadata": {}} + if sync_mode: + picked = handler.get_available_deployments( + model_group=MODEL_GROUP, healthy_deployments=STREAMING_DEPLOYMENTS, request_kwargs=request_kwargs + ) + else: + picked = await handler.async_get_available_deployments( + model_group=MODEL_GROUP, healthy_deployments=STREAMING_DEPLOYMENTS, request_kwargs=request_kwargs + ) + + assert picked is not None + assert picked["model_info"]["id"] == FAST_TTFT_ID + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"]) +async def test_ttft_window_keeps_newest_samples_when_full(sync_mode: bool): + """Float timestamps, as the SDK passes them. Once max_latency_list_size + samples exist the oldest TTFT is dropped so the window slides.""" + max_size = 3 + cache = DualCache() + handler = LowestLatencyLoggingHandler(router_cache=cache, routing_args={"max_latency_list_size": max_size}) + start_time = 1_700_000_000.0 + ttfts = (0.1, 0.2, 0.3, 0.4) + + for ttft in ttfts: + kwargs = { + "litellm_params": { + "metadata": {"model_group": MODEL_GROUP}, + "model_info": {"id": FAST_TTFT_ID}, + }, + "stream": True, + "completion_start_time": start_time + ttft, + } + response_obj = _chat_response(completion_tokens=1) + if sync_mode: + handler.log_success_event( + response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=start_time + 1.0 + ) + else: + await handler.async_log_success_event( + response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=start_time + 1.0 + ) + + assert _recorded_ttft(cache, FAST_TTFT_ID) == [pytest.approx(ttft) for ttft in ttfts[-max_size:]] + + +@pytest.mark.asyncio +async def test_streaming_routing_ignores_per_token_ttft_samples_from_older_workers(): + """Workers on the previous release share the Redis map and keep writing + seconds-per-token under the old "time_to_first_token" key during a rolling + deploy. Those samples favor SLOW; routing must only read the seconds key.""" + cache = DualCache() + handler = LowestLatencyLoggingHandler(router_cache=cache) + cache.set_cache( + key=f"{MODEL_GROUP}_map", + value={ + FAST_TTFT_ID: {"time_to_first_token": [0.02], "time_to_first_token_seconds": [1.0]}, + SLOW_TTFT_ID: {"time_to_first_token": [0.006], "time_to_first_token_seconds": [3.0]}, + }, + ) + + picked = await handler.async_get_available_deployments( + model_group=MODEL_GROUP, + healthy_deployments=STREAMING_DEPLOYMENTS, + request_kwargs={"stream": True, "metadata": {}}, + ) + + assert picked is not None + assert picked["model_info"]["id"] == FAST_TTFT_ID + + @pytest.mark.asyncio @pytest.mark.parametrize( "cached_entry",