diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py index 4fd49a35417c..87c1d55646a5 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py @@ -7,6 +7,9 @@ from litellm import verbose_logger from litellm._uuid import uuid +from litellm.llms.anthropic.experimental_pass_through.responses_adapters.transformation import ( + translate_responses_usage_to_anthropic_usage, +) class AnthropicResponsesStreamWrapper: @@ -226,10 +229,12 @@ def _process_event(self, event: Any) -> None: event.get("response") if isinstance(event, dict) else None ) stop_reason = "end_turn" - input_tokens = 0 - output_tokens = 0 - cache_creation_tokens = 0 - cache_read_tokens = 0 + anthropic_usage = { + "input_tokens": 0, + "output_tokens": 0, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 0, + } if response_obj is not None: status = getattr(response_obj, "status", None) @@ -237,13 +242,7 @@ def _process_event(self, event: Any) -> None: stop_reason = "max_tokens" usage = getattr(response_obj, "usage", None) if usage is not None: - input_tokens = getattr(usage, "input_tokens", 0) or 0 - output_tokens = getattr(usage, "output_tokens", 0) or 0 - cache_creation_tokens = getattr(usage, "input_tokens_details", None) # type: ignore[assignment] - cache_read_tokens = getattr(usage, "output_tokens_details", None) # type: ignore[assignment] - # Prefer direct cache fields if present - cache_creation_tokens = int(getattr(usage, "cache_creation_input_tokens", 0) or 0) - cache_read_tokens = int(getattr(usage, "cache_read_input_tokens", 0) or 0) + anthropic_usage = translate_responses_usage_to_anthropic_usage(usage) # Check if tool_use was in the output to override stop_reason if response_obj is not None: @@ -257,13 +256,13 @@ def _process_event(self, event: Any) -> None: break usage_delta: Dict[str, Any] = { - "input_tokens": input_tokens, - "output_tokens": output_tokens, + "input_tokens": anthropic_usage["input_tokens"], + "output_tokens": anthropic_usage["output_tokens"], } - if cache_creation_tokens: - usage_delta["cache_creation_input_tokens"] = cache_creation_tokens - if cache_read_tokens: - usage_delta["cache_read_input_tokens"] = cache_read_tokens + if anthropic_usage["cache_creation_input_tokens"]: + usage_delta["cache_creation_input_tokens"] = anthropic_usage["cache_creation_input_tokens"] + if anthropic_usage["cache_read_input_tokens"]: + usage_delta["cache_read_input_tokens"] = anthropic_usage["cache_read_input_tokens"] self._chunk_queue.append( { diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py index 172e54de98e9..a5a684952666 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py @@ -6,6 +6,7 @@ """ import json +from numbers import Number from typing import Any, Dict, List, Optional, Union, cast from litellm.litellm_core_utils.reasoning_effort_utils import ( @@ -32,6 +33,50 @@ from litellm.types.llms.openai import ResponsesAPIResponse +def _get_usage_value(usage: Any, key: str) -> Any: + if isinstance(usage, dict): + return usage.get(key) + return getattr(usage, key, None) + + +def _get_int_usage_value(usage: Any, key: str) -> int: + value = _get_usage_value(usage, key) + if isinstance(value, Number): + return int(value) + if isinstance(value, str) and value.strip().isdigit(): + return int(value) + return 0 + + +def _get_responses_cached_tokens(usage: Any) -> int: + input_tokens_details = _get_usage_value(usage, "input_tokens_details") + if input_tokens_details is None: + return 0 + return _get_int_usage_value(input_tokens_details, "cached_tokens") + + +def translate_responses_usage_to_anthropic_usage(usage: Any) -> AnthropicUsage: + input_tokens = _get_int_usage_value(usage, "input_tokens") + output_tokens = _get_int_usage_value(usage, "output_tokens") + cache_creation_tokens = _get_int_usage_value(usage, "cache_creation_input_tokens") + + responses_cached_tokens = _get_responses_cached_tokens(usage) + if responses_cached_tokens: + cache_read_tokens = responses_cached_tokens + # OpenAI Responses input_tokens includes cached tokens. Anthropic Messages + # input_tokens excludes cache-read tokens, so only subtract for this source. + input_tokens = max(0, input_tokens - responses_cached_tokens) + else: + cache_read_tokens = _get_int_usage_value(usage, "cache_read_input_tokens") + + return AnthropicUsage( + input_tokens=input_tokens, + output_tokens=output_tokens, + cache_creation_input_tokens=cache_creation_tokens, + cache_read_input_tokens=cache_read_tokens, + ) + + class LiteLLMAnthropicToResponsesAPIAdapter: """ Converts Anthropic /v1/messages requests to OpenAI Responses API format and @@ -396,8 +441,6 @@ def translate_response( ResponseReasoningItem, ) - from litellm.types.llms.openai import ResponseAPIUsage - content: List[Dict[str, Any]] = [] stop_reason: AnthropicFinishReason = "end_turn" @@ -463,15 +506,7 @@ def translate_response( if response.status == "incomplete": stop_reason = "max_tokens" - # usage - raw_usage: Optional[ResponseAPIUsage] = response.usage - input_tokens = int(getattr(raw_usage, "input_tokens", 0) or 0) - output_tokens = int(getattr(raw_usage, "output_tokens", 0) or 0) - - anthropic_usage = AnthropicUsage( - input_tokens=input_tokens, - output_tokens=output_tokens, - ) + anthropic_usage = translate_responses_usage_to_anthropic_usage(response.usage) return AnthropicMessagesResponse( id=response.id, diff --git a/tests/e2e/proxy_client.py b/tests/e2e/proxy_client.py index 6c6b948e29c4..b3fc85383223 100644 --- a/tests/e2e/proxy_client.py +++ b/tests/e2e/proxy_client.py @@ -75,12 +75,130 @@ RowsPredicate = Callable[[list[SpendLogRow]], bool] +# After /model/new, the control-plane writer reloads itself immediately, but every +# other gateway worker (and peer pod) only picks the model up on its add_deployment +# job. That job runs every proxy_config_reload_interval_seconds (product default 30). +# A single /v1/models hit can land on a hot worker while the next /chat hits a cold +# one ("Invalid model name"). Wait for first listing within MODEL_SERVABLE_TIMEOUT, +# then require continuous listing for MODEL_SERVABLE_DB_SYNC_SECONDS (the default +# reload interval) so every worker has had a chance to sync from the DB. +MODEL_SERVABLE_TIMEOUT = 40.0 +MODEL_SERVABLE_DB_SYNC_SECONDS = 30.0 +MODEL_SERVABLE_INTERVAL = 2.0 +# Cap each /v1/models poll so one slow request cannot outlast the remaining budget. +MODEL_SERVABLE_REQUEST_TIMEOUT = 5.0 + + +@dataclass(frozen=True, slots=True) +class Servable: + """The data plane listed the model within the deadline.""" + + +@dataclass(frozen=True, slots=True) +class NotServable: + """The deadline passed without the data plane listing the model. + + `last_result` is the final /v1/models read, so the caller can tell "the proxy + answered but omitted the model" (propagation) from "the read itself failed" + (network/auth) when reporting.""" + + last_result: Result[ModelsListResponse] | None + + +ServableOutcome = Servable | NotServable + + +def await_servable( + list_models: Callable[[float], Result[ModelsListResponse]], + *, + model_name: str, + timeout: float, + interval: float, + request_timeout: float, + db_sync_seconds: float, + now: Callable[[], float], + sleep: Callable[[float], None], +) -> ServableOutcome: + """Poll until `model_name` is listed long enough for every worker to DB-sync. + + First listing must happen within `timeout`. After that, the model must stay + listed continuously for `db_sync_seconds` (any miss resets the continuous + window). `db_sync_seconds=0` returns on the first listing. Each poll's request + timeout is clamped to the remaining budget. Sleeps only min(interval, time left) + so a final deadline-clamped poll is never skipped just because a full interval + does not fit. Clock and sleep are injected.""" + started = now() + first_seen_at: float | None = None + last_result: Result[ModelsListResponse] | None = None + while True: + t = now() + phase_deadline = ( + started + timeout if first_seen_at is None else first_seen_at + db_sync_seconds + ) + remaining = phase_deadline - t + if remaining <= 0: + if ( + last_result is not None + and first_seen_at is not None + and (db_sync_seconds <= 0 or t - first_seen_at >= db_sync_seconds) + ): + return Servable() + return NotServable(last_result=last_result) + + poll_timeout = min(request_timeout, remaining) + last_result = list_models(poll_timeout) + listed = isinstance(last_result, Success) and any( + entry.id == model_name for entry in last_result.data.data + ) + t = now() + if not listed: + first_seen_at = None + elif first_seen_at is None: + if t > started + timeout: + return NotServable(last_result=last_result) + first_seen_at = t + if db_sync_seconds <= 0: + return Servable() + elif t - first_seen_at >= db_sync_seconds: + return Servable() + + phase_deadline = ( + started + timeout if first_seen_at is None else first_seen_at + db_sync_seconds + ) + wait = min(interval, phase_deadline - now()) + if wait > 0: + sleep(wait) + + +def servable_timeout_message( + *, + model_name: str, + timeout: float, + db_sync_seconds: float, + last_result: Result[ModelsListResponse] | None, +) -> str: + last_error = ( + f"; last /v1/models poll did not succeed: {last_result}" + if last_result is not None and not isinstance(last_result, Success) + else "" + ) + return ( + f"model {model_name!r} was created but never became servable on the data " + f"plane within {timeout}s of first listing (plus {db_sync_seconds}s continuous " + f"DB sync) after /model/new (control/data-plane propagation or " + f"STORE_MODEL_IN_DB reload issue){last_error}" + ) + @dataclass(frozen=True, slots=True) class ProxyClient: transport: Transport poll_timeout: float = 120.0 poll_interval: float = 5.0 + model_servable_timeout: float = MODEL_SERVABLE_TIMEOUT + model_servable_db_sync_seconds: float = MODEL_SERVABLE_DB_SYNC_SECONDS + model_servable_interval: float = MODEL_SERVABLE_INTERVAL + model_servable_request_timeout: float = MODEL_SERVABLE_REQUEST_TIMEOUT # ---- keys / customers (satisfies lifecycle.ResourceClient) ---------- @@ -167,7 +285,12 @@ def create_model( this returns can race the reload and 400 with "Invalid model name passed". We therefore poll the data-plane /v1/models until the model appears before handing back, so callers can invoke it immediately. In the monolithic case - it is already present on the first poll, so this adds one request.""" + it is already present on the first poll, so this adds one request. + + First listing must arrive within `model_servable_timeout` (not the longer + spend `poll_timeout`). The model must then stay listed for + `model_servable_db_sync_seconds` (product default DB reload interval) so every + gateway worker has run add_deployment before callers use the model.""" model_id = unwrap( self.transport.post( "/model/new", @@ -184,33 +307,38 @@ def create_model( return model_id def _await_model_servable(self, model_name: str) -> None: - """Block until the data plane lists `model_name`, or fail loudly if it does - not within poll_timeout (a real propagation/config problem, surfaced here - instead of as a downstream "Invalid model name passed").""" - deadline = time.monotonic() + self.poll_timeout - last_result: Result[ModelsListResponse] | None = None - while time.monotonic() < deadline: - last_result = self.transport.get( + """Block until the data plane lists `model_name` long enough for DB sync. + + Fails if first listing misses model_servable_timeout, or if continuous listing + for model_servable_db_sync_seconds never holds (multi-worker / peer reload).""" + outcome = await_servable( + lambda poll_timeout: self.transport.get( "/v1/models", headers=self.transport.master, params=NoBody(), response_type=ModelsListResponse, - ) - if isinstance(last_result, Success) and any( - entry.id == model_name for entry in last_result.data.data - ): - return - time.sleep(self.poll_interval) - last_error = ( - f"; last /v1/models poll did not succeed: {last_result}" - if last_result is not None and not isinstance(last_result, Success) - else "" - ) - raise AssertionError( - f"model {model_name!r} was created but never became servable on the data " - f"plane within {self.poll_timeout}s of /model/new (control/data-plane " - f"propagation or STORE_MODEL_IN_DB reload issue){last_error}" + timeout=poll_timeout, + ), + model_name=model_name, + timeout=self.model_servable_timeout, + interval=self.model_servable_interval, + request_timeout=self.model_servable_request_timeout, + db_sync_seconds=self.model_servable_db_sync_seconds, + now=time.monotonic, + sleep=time.sleep, ) + match outcome: + case Servable(): + return + case NotServable(last_result=last_result): + raise AssertionError( + servable_timeout_message( + model_name=model_name, + timeout=self.model_servable_timeout, + db_sync_seconds=self.model_servable_db_sync_seconds, + last_result=last_result, + ) + ) def update_model(self, model_id: str, litellm_params: LiteLLMParamsBody) -> None: """Merge `litellm_params` over the deployment `model_id`'s stored params via diff --git a/tests/e2e/transport.py b/tests/e2e/transport.py index a6adf83ed1f3..27b11befc8ec 100644 --- a/tests/e2e/transport.py +++ b/tests/e2e/transport.py @@ -58,6 +58,7 @@ def get[R: BaseModel]( headers: BaseModel, params: BaseModel, response_type: type[R], + timeout: float | None = None, ) -> Result[R]: ... def delete[R: BaseModel]( @@ -136,13 +137,16 @@ def get[R: BaseModel]( headers: BaseModel, params: BaseModel, response_type: type[R], + timeout: float | None = None, ) -> Result[R]: + """`timeout` overrides the transport-wide request_timeout for this call, for + pollers whose own deadline is shorter than it.""" return e2e_http.get( self._url(path), headers=headers, params=params, response_type=response_type, - timeout=self.request_timeout, + timeout=self.request_timeout if timeout is None else timeout, ) def delete[R: BaseModel]( @@ -336,9 +340,14 @@ def get[R: BaseModel]( headers: BaseModel, params: BaseModel, response_type: type[R], + timeout: float | None = None, ) -> Result[R]: return self._route(path).get( - path, headers=headers, params=params, response_type=response_type + path, + headers=headers, + params=params, + response_type=response_type, + timeout=timeout, ) def delete[R: BaseModel]( diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index 9b5197d90289..d396098dc5e5 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py @@ -6,6 +6,7 @@ import asyncio import os import sys +from types import SimpleNamespace sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../.."))) @@ -75,6 +76,72 @@ def test_second_response_created_is_skipped(self): assert len(message_starts) == 1 +def _completed_event_with_usage(usage: object) -> SimpleNamespace: + return SimpleNamespace( + type="response.completed", + response=SimpleNamespace(status="completed", usage=usage, output=[]), + ) + + +def _message_delta_usage(chunks: list) -> dict: + for chunk in chunks: + if chunk.get("type") == "message_delta": + return chunk["usage"] + raise AssertionError("message_delta chunk not found") + + +class TestProcessEventUsageMapping: + """Responses usage is translated to Anthropic Messages usage.""" + + def test_openai_responses_cached_tokens_map_to_cache_read_tokens(self): + usage = SimpleNamespace( + input_tokens=1000, + output_tokens=75, + input_tokens_details=SimpleNamespace(cached_tokens=800), + ) + + chunks = _process_all([_completed_event_with_usage(usage)]) + + assert _message_delta_usage(chunks) == { + "input_tokens": 200, + "output_tokens": 75, + "cache_read_input_tokens": 800, + } + + def test_dict_responses_cached_tokens_map_to_cache_read_tokens(self): + usage = SimpleNamespace( + input_tokens=1000, + output_tokens=75, + input_tokens_details={"cached_tokens": 800}, + ) + + chunks = _process_all([_completed_event_with_usage(usage)]) + + assert _message_delta_usage(chunks) == { + "input_tokens": 200, + "output_tokens": 75, + "cache_read_input_tokens": 800, + } + + def test_anthropic_usage_fallback_does_not_double_subtract_input_tokens(self): + usage = SimpleNamespace( + input_tokens=200, + output_tokens=75, + input_tokens_details=None, + cache_creation_input_tokens=10, + cache_read_input_tokens=800, + ) + + chunks = _process_all([_completed_event_with_usage(usage)]) + + assert _message_delta_usage(chunks) == { + "input_tokens": 200, + "output_tokens": 75, + "cache_creation_input_tokens": 10, + "cache_read_input_tokens": 800, + } + + class TestProcessEventTextDeltaWithoutOutputItemAdded: """Streams that skip response.output_item.added (e.g. LMStudio) must still open a text block before any delta and never emit index -1.""" diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py index 606ff39b35e7..6c1e09009dac 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py @@ -838,6 +838,16 @@ def _make_mock_response( return resp +def _make_response_with_usage(usage: Any) -> MagicMock: + resp = MagicMock() + resp.id = "resp_usage" + resp.model = "gpt-4o" + resp.status = "completed" + resp.output = [] + resp.usage = usage + return resp + + def _make_output_message(texts: List[str]) -> MagicMock: """Build a mock ResponseOutputMessage with output_text parts.""" from openai.types.responses import ResponseOutputMessage # type: ignore[import] @@ -961,6 +971,55 @@ def test_usage_mapped_correctly(self): assert result["usage"]["input_tokens"] == 200 assert result["usage"]["output_tokens"] == 75 + def test_openai_responses_cached_tokens_map_to_anthropic_usage(self): + """Responses cached_tokens becomes Anthropic cache_read_input_tokens.""" + usage = MagicMock() + usage.input_tokens = 1000 + usage.output_tokens = 75 + usage.input_tokens_details = MagicMock() + usage.input_tokens_details.cached_tokens = 800 + + result: Any = _ADAPTER.translate_response(_make_response_with_usage(usage)) + assert result["usage"] == { + "input_tokens": 200, + "output_tokens": 75, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 800, + } + + def test_dict_responses_cached_tokens_map_to_anthropic_usage(self): + """Dict-shaped Responses usage is handled the same as SDK objects.""" + usage = { + "input_tokens": 1000, + "output_tokens": 75, + "input_tokens_details": {"cached_tokens": 800}, + } + + result: Any = _ADAPTER.translate_response(_make_response_with_usage(usage)) + assert result["usage"] == { + "input_tokens": 200, + "output_tokens": 75, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 800, + } + + def test_anthropic_usage_fallback_does_not_double_subtract_input_tokens(self): + """Anthropic-style usage already reports uncached input tokens.""" + usage = MagicMock() + usage.input_tokens = 200 + usage.output_tokens = 75 + usage.cache_read_input_tokens = 800 + usage.cache_creation_input_tokens = 10 + usage.input_tokens_details = None + + result: Any = _ADAPTER.translate_response(_make_response_with_usage(usage)) + assert result["usage"] == { + "input_tokens": 200, + "output_tokens": 75, + "cache_creation_input_tokens": 10, + "cache_read_input_tokens": 800, + } + def test_model_and_id_preserved(self): """Model and response ID from the Responses API are forwarded.""" response = _make_mock_response(