mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-21 00:21:49 +00:00
fix(timing): union provider timing windows and anchor detailed pre-processing at receive time
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry
parent
0f54d76079
commit
d74e1bb445
5 changed files with 142 additions and 28 deletions
|
|
@ -1,5 +1,6 @@
|
|||
import datetime
|
||||
from collections.abc import Mapping
|
||||
from functools import reduce
|
||||
from typing import Any, Final
|
||||
|
||||
import httpx
|
||||
|
|
@ -25,6 +26,34 @@ def _timing_window_start(
|
|||
return start_time, False
|
||||
|
||||
|
||||
def _union_duration_ms(windows: object, lower: float, upper: float) -> float | None:
|
||||
if not isinstance(windows, (list, tuple)):
|
||||
return None
|
||||
clipped: Final[tuple[tuple[float, float], ...]] = tuple(
|
||||
(max(lower, float(window[0])), min(upper, float(window[1])))
|
||||
for window in windows
|
||||
if isinstance(window, (list, tuple))
|
||||
and len(window) == 2
|
||||
and isinstance(window[0], (int, float))
|
||||
and isinstance(window[1], (int, float))
|
||||
and max(lower, float(window[0])) < min(upper, float(window[1]))
|
||||
)
|
||||
if not clipped:
|
||||
return None
|
||||
|
||||
ordered: Final[tuple[tuple[float, float], ...]] = tuple(sorted(clipped))
|
||||
|
||||
def merge_window(
|
||||
merged: tuple[tuple[float, float], ...], current: tuple[float, float]
|
||||
) -> tuple[tuple[float, float], ...]:
|
||||
if not merged or current[0] > merged[-1][1]:
|
||||
return (*merged, current)
|
||||
return (*merged[:-1], (merged[-1][0], max(merged[-1][1], current[1])))
|
||||
|
||||
merged: Final[tuple[tuple[float, float], ...]] = reduce(merge_window, ordered, ())
|
||||
return sum(end - start for start, end in merged) * 1000
|
||||
|
||||
|
||||
def response_timing_metrics(
|
||||
start_time: datetime.datetime,
|
||||
end_time: datetime.datetime,
|
||||
|
|
@ -54,20 +83,17 @@ def response_timing_metrics(
|
|||
if cache_duration_ms is not None:
|
||||
overhead_ms: float | None = total_response_time_ms - cache_duration_ms
|
||||
elif llm_api_duration_ms is not None:
|
||||
total_provider_duration_ms: Final = metadata.get("llm_api_duration_ms_total")
|
||||
provider_duration_ms: Final = (
|
||||
total_provider_duration_ms
|
||||
provider_duration_ms: Final[float | None] = (
|
||||
_union_duration_ms(
|
||||
metadata.get("llm_api_timing_windows"),
|
||||
window_start.timestamp(),
|
||||
end_time.timestamp(),
|
||||
)
|
||||
if receive_anchored
|
||||
and isinstance(total_provider_duration_ms, float)
|
||||
and isinstance(llm_api_duration_ms, (int, float))
|
||||
and total_provider_duration_ms >= llm_api_duration_ms
|
||||
else llm_api_duration_ms
|
||||
)
|
||||
overhead_ms = (
|
||||
round(total_response_time_ms - provider_duration_ms, 4)
|
||||
if isinstance(provider_duration_ms, (int, float))
|
||||
else None
|
||||
)
|
||||
effective: Final = provider_duration_ms if provider_duration_ms is not None else llm_api_duration_ms
|
||||
overhead_ms = round(total_response_time_ms - effective, 4) if isinstance(effective, (int, float)) else None
|
||||
else:
|
||||
overhead_ms = None
|
||||
if overhead_ms is None:
|
||||
|
|
@ -178,7 +204,8 @@ class ResponseMetadata:
|
|||
# pre-processing = time from request start to LLM API call start
|
||||
api_call_start: Final[datetime.datetime | None] = logging_obj.model_call_details.get("api_call_start_time")
|
||||
if api_call_start is not None and start_time is not None:
|
||||
pre_ms: Final = (api_call_start - start_time).total_seconds() * 1000
|
||||
anchor: Final = _timing_window_start(start_time, logging_obj)[0]
|
||||
pre_ms: Final = (api_call_start - anchor).total_seconds() * 1000
|
||||
detailed["timing_pre_processing_ms"] = round(pre_ms, 4)
|
||||
|
||||
# post-processing = total - pre - llm_api
|
||||
|
|
|
|||
|
|
@ -288,10 +288,19 @@ def _set_duration_in_model_call_details(
|
|||
if logging_obj and hasattr(logging_obj, "model_call_details"):
|
||||
logging_obj.model_call_details["llm_api_duration_ms"] = duration_ms
|
||||
metadata: Final[dict[str, object]] = get_litellm_metadata_from_kwargs(logging_obj.model_call_details)
|
||||
existing_total: Final = metadata.get("llm_api_duration_ms_total")
|
||||
metadata["llm_api_duration_ms_total"] = (
|
||||
existing_total if isinstance(existing_total, float) else 0.0
|
||||
) + duration_ms
|
||||
recorded: Final = metadata.get("llm_api_timing_windows")
|
||||
earlier: Final[tuple[tuple[float, float], ...]] = tuple(
|
||||
(float(window[0]), float(window[1]))
|
||||
for window in (recorded if isinstance(recorded, (list, tuple)) else ())
|
||||
if isinstance(window, (list, tuple))
|
||||
and len(window) == 2
|
||||
and isinstance(window[0], (int, float))
|
||||
and isinstance(window[1], (int, float))
|
||||
)
|
||||
metadata["llm_api_timing_windows"] = (
|
||||
*earlier,
|
||||
(start_time.timestamp(), end_time.timestamp()),
|
||||
)
|
||||
else:
|
||||
verbose_logger.debug("`logging_obj` not found - unable to track `llm_api_duration_ms")
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import litellm.proxy.common_request_processing as common_request_processing_mod
|
|||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.litellm_core_utils.llm_response_utils.response_metadata import (
|
||||
ResponseMetadata,
|
||||
_union_duration_ms,
|
||||
response_timing_metrics,
|
||||
update_response_metadata,
|
||||
)
|
||||
|
|
@ -234,7 +235,7 @@ class TestResponseTimingMetrics:
|
|||
def _make_logging_obj(
|
||||
self,
|
||||
llm_api_duration_ms: float | None = None,
|
||||
llm_api_duration_ms_total: float | None = None,
|
||||
llm_api_timing_windows: object = None,
|
||||
caching_details: dict[str, object] | None = None,
|
||||
received_at: datetime.datetime | str | None = None,
|
||||
) -> MagicMock:
|
||||
|
|
@ -242,12 +243,12 @@ class TestResponseTimingMetrics:
|
|||
logging_obj.model_call_details = {}
|
||||
if llm_api_duration_ms is not None:
|
||||
logging_obj.model_call_details["llm_api_duration_ms"] = llm_api_duration_ms
|
||||
if received_at is not None or llm_api_duration_ms_total is not None:
|
||||
if received_at is not None or llm_api_timing_windows is not None:
|
||||
metadata = {}
|
||||
if received_at is not None:
|
||||
metadata["litellm_received_at"] = received_at
|
||||
if llm_api_duration_ms_total is not None:
|
||||
metadata["llm_api_duration_ms_total"] = llm_api_duration_ms_total
|
||||
if llm_api_timing_windows is not None:
|
||||
metadata["llm_api_timing_windows"] = llm_api_timing_windows
|
||||
logging_obj.model_call_details["litellm_params"] = {"metadata": metadata}
|
||||
logging_obj.caching_details = caching_details
|
||||
return logging_obj
|
||||
|
|
@ -271,7 +272,10 @@ class TestResponseTimingMetrics:
|
|||
def test_receive_anchored_window_subtracts_all_provider_attempts(self):
|
||||
logging_obj = self._make_logging_obj(
|
||||
llm_api_duration_ms=300.0,
|
||||
llm_api_duration_ms_total=700.0,
|
||||
llm_api_timing_windows=(
|
||||
(self.START.timestamp(), self.START.timestamp() + 0.3),
|
||||
(self.START.timestamp() + 0.4, self.START.timestamp() + 0.8),
|
||||
),
|
||||
received_at=self.START,
|
||||
)
|
||||
|
||||
|
|
@ -283,7 +287,7 @@ class TestResponseTimingMetrics:
|
|||
def test_sdk_window_subtracts_current_provider_attempt(self):
|
||||
logging_obj = self._make_logging_obj(
|
||||
llm_api_duration_ms=300.0,
|
||||
llm_api_duration_ms_total=700.0,
|
||||
llm_api_timing_windows=((self.START.timestamp(), self.START.timestamp() + 0.3),),
|
||||
)
|
||||
|
||||
result = response_timing_metrics(self.START, self.END, logging_obj)
|
||||
|
|
@ -291,6 +295,38 @@ class TestResponseTimingMetrics:
|
|||
assert result["_response_ms"] == pytest.approx(1000.0)
|
||||
assert result["litellm_overhead_time_ms"] == pytest.approx(700.0)
|
||||
|
||||
def test_receive_anchored_window_unions_nested_and_retry_windows(self):
|
||||
windows = (
|
||||
(self.START.timestamp(), self.START.timestamp() + 0.3),
|
||||
(self.START.timestamp(), self.START.timestamp() + 0.3),
|
||||
(self.START.timestamp() + 0.4, self.START.timestamp() + 0.7),
|
||||
)
|
||||
logging_obj = self._make_logging_obj(
|
||||
llm_api_duration_ms=300.0,
|
||||
llm_api_timing_windows=windows,
|
||||
received_at=self.START,
|
||||
)
|
||||
|
||||
result = response_timing_metrics(self.START, self.END, logging_obj)
|
||||
|
||||
assert result["litellm_overhead_time_ms"] == pytest.approx(400.0)
|
||||
assert _union_duration_ms(windows, self.START.timestamp(), self.END.timestamp()) == pytest.approx(600.0)
|
||||
|
||||
def test_receive_anchored_window_ignores_seeded_windows_outside_window(self):
|
||||
windows = (
|
||||
(self.START.timestamp() - 10.0, self.START.timestamp() - 1.0),
|
||||
(self.END.timestamp() + 1.0, self.END.timestamp() + 2.0),
|
||||
)
|
||||
logging_obj = self._make_logging_obj(
|
||||
llm_api_duration_ms=300.0,
|
||||
llm_api_timing_windows=windows,
|
||||
received_at=self.START,
|
||||
)
|
||||
|
||||
result = response_timing_metrics(self.START, self.END, logging_obj)
|
||||
|
||||
assert result["litellm_overhead_time_ms"] == pytest.approx(700.0)
|
||||
|
||||
def test_receive_anchored_window_falls_back_to_current_provider_attempt(self):
|
||||
logging_obj = self._make_logging_obj(
|
||||
llm_api_duration_ms=300.0,
|
||||
|
|
@ -434,6 +470,27 @@ class TestDetailedTiming:
|
|||
assert hidden.get("timing_pre_processing_ms") == 20.0
|
||||
assert hidden.get("timing_post_processing_ms") == 10.0 # 530 - 20 - 500
|
||||
|
||||
def test_detailed_timing_pre_processing_uses_receive_anchor(self, monkeypatch):
|
||||
monkeypatch.setattr(response_metadata_mod, "LITELLM_DETAILED_TIMING", True)
|
||||
|
||||
result = ModelResponse()
|
||||
start = datetime.datetime(2025, 1, 1, 0, 0, 0)
|
||||
received_at = start - datetime.timedelta(milliseconds=200)
|
||||
end = start + datetime.timedelta(milliseconds=530)
|
||||
logging_obj = self._make_logging_obj(
|
||||
llm_api_duration_ms=500.0,
|
||||
api_call_start_time=start,
|
||||
)
|
||||
logging_obj.model_call_details["litellm_params"] = {"metadata": {"litellm_received_at": received_at}}
|
||||
|
||||
metadata = ResponseMetadata(result)
|
||||
metadata.set_timing_metrics(start, end, logging_obj)
|
||||
metadata.apply()
|
||||
|
||||
hidden = result._hidden_params
|
||||
assert hidden.get("timing_pre_processing_ms") == pytest.approx(200.0)
|
||||
assert hidden.get("timing_post_processing_ms") == pytest.approx(30.0)
|
||||
|
||||
def test_detailed_timing_absent_when_disabled(self, monkeypatch):
|
||||
"""When LITELLM_DETAILED_TIMING is false, no detailed timing keys."""
|
||||
monkeypatch.setattr(response_metadata_mod, "LITELLM_DETAILED_TIMING", False)
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ from litellm.litellm_core_utils.logging_utils import (
|
|||
|
||||
|
||||
class TestSetDurationInModelCallDetails:
|
||||
def test_accumulates_provider_attempts_in_shared_metadata(self):
|
||||
def test_records_provider_attempt_windows_in_shared_metadata(self):
|
||||
metadata = {"request_id": "test"}
|
||||
logging_obj = MagicMock()
|
||||
logging_obj.model_call_details = {"litellm_params": {"metadata": metadata}}
|
||||
|
|
@ -31,7 +31,10 @@ class TestSetDurationInModelCallDetails:
|
|||
_set_duration_in_model_call_details(logging_obj, first_start, first_end)
|
||||
_set_duration_in_model_call_details(logging_obj, second_start, second_end)
|
||||
|
||||
assert metadata["llm_api_duration_ms_total"] == pytest.approx(1000.0)
|
||||
assert metadata["llm_api_timing_windows"] == (
|
||||
(first_start.timestamp(), first_end.timestamp()),
|
||||
(second_start.timestamp(), second_end.timestamp()),
|
||||
)
|
||||
assert logging_obj.model_call_details["llm_api_duration_ms"] == pytest.approx(700.0)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ Regression tests for https://github.com/BerriAI/litellm/issues/21343
|
|||
import asyncio
|
||||
import datetime
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Final, cast
|
||||
from typing import Final
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
|
@ -21,6 +21,10 @@ import pytest
|
|||
import litellm
|
||||
from litellm import Router
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.litellm_core_utils.llm_response_utils.response_metadata import (
|
||||
_union_duration_ms,
|
||||
response_timing_metrics,
|
||||
)
|
||||
from litellm.litellm_core_utils.logging_utils import track_llm_api_timing
|
||||
from litellm.litellm_core_utils.rules import Rules
|
||||
from litellm.utils import function_setup
|
||||
|
|
@ -286,7 +290,11 @@ async def test_not_found_error_in_retry_loop_raises_immediately():
|
|||
|
||||
@pytest.mark.asyncio
|
||||
async def test_retry_attempts_accumulate_timing_in_shared_request_metadata():
|
||||
metadata: dict[str, object] = {"model_group": "test-model"}
|
||||
received_at: Final = datetime.datetime.now()
|
||||
metadata: dict[str, object] = {
|
||||
"model_group": "test-model",
|
||||
"litellm_received_at": received_at,
|
||||
}
|
||||
logging_obj_raw, _ = function_setup(
|
||||
"acompletion",
|
||||
Rules(),
|
||||
|
|
@ -297,7 +305,8 @@ async def test_retry_attempts_accumulate_timing_in_shared_request_metadata():
|
|||
litellm_call_id="retry-timing-test",
|
||||
is_async_call=True,
|
||||
)
|
||||
logging_obj: Final[Logging] = cast(Logging, logging_obj_raw)
|
||||
assert isinstance(logging_obj_raw, Logging)
|
||||
logging_obj: Final[Logging] = logging_obj_raw
|
||||
attempt_numbers: list[int] = []
|
||||
metadata_ids: list[int] = []
|
||||
|
||||
|
|
@ -334,8 +343,17 @@ async def test_retry_attempts_accumulate_timing_in_shared_request_metadata():
|
|||
)
|
||||
|
||||
request_metadata: Final = logging_obj.model_call_details["litellm_params"]["metadata"]
|
||||
windows: Final = request_metadata["llm_api_timing_windows"]
|
||||
end_time: Final = datetime.datetime.fromtimestamp(max(window[1] for window in windows))
|
||||
timing_metrics: Final = response_timing_metrics(received_at, end_time, logging_obj)
|
||||
assert result == "success"
|
||||
assert attempt_numbers == [1, 2]
|
||||
assert request_metadata is metadata
|
||||
assert metadata_ids == [id(metadata), id(metadata)]
|
||||
assert request_metadata["llm_api_duration_ms_total"] > logging_obj.model_call_details["llm_api_duration_ms"]
|
||||
assert len(windows) == 2
|
||||
union_duration_ms: Final = _union_duration_ms(windows, received_at.timestamp(), end_time.timestamp())
|
||||
assert union_duration_ms is not None
|
||||
total_response_time_ms: Final = (end_time.timestamp() - received_at.timestamp()) * 1000
|
||||
assert timing_metrics["litellm_overhead_time_ms"] == pytest.approx(
|
||||
round(total_response_time_ms - union_duration_ms, 4)
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue