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 292d2622c7f..923fad8ac3d 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 @@ -9,10 +9,54 @@ from typing import Any, Final from litellm import verbose_logger from litellm._uuid import uuid from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage +from litellm.types.llms.openai import ResponseAPIUsage from .transformation import LiteLLMAnthropicToResponsesAPIAdapter +def _get_field(obj: object, key: str, default: object = None) -> object: + if isinstance(obj, dict): + return obj.get(key, default) + return getattr(obj, key, default) + + +def _translate_usage(raw_usage: object) -> AnthropicUsage: + if raw_usage is None or isinstance(raw_usage, ResponseAPIUsage): + return LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage(raw_usage) + + input_tokens: Final = int(_get_field(raw_usage, "input_tokens", 0) or 0) + output_tokens: Final = int(_get_field(raw_usage, "output_tokens", 0) or 0) + input_tokens_details: Final = _get_field(raw_usage, "input_tokens_details") + cache_creation_tokens: Final = int(_get_field(raw_usage, "cache_creation_input_tokens", 0) or 0) or int( + _get_field(input_tokens_details, "cache_write_tokens", 0) or 0 + ) + cache_read_tokens: Final = int(_get_field(raw_usage, "cache_read_input_tokens", 0) or 0) or int( + _get_field(input_tokens_details, "cached_tokens", 0) or 0 + ) + uncached_input_tokens: Final = max(input_tokens - cache_read_tokens - cache_creation_tokens, 0) + + if cache_creation_tokens and cache_read_tokens: + return AnthropicUsage( + input_tokens=uncached_input_tokens, + output_tokens=output_tokens, + cache_creation_input_tokens=cache_creation_tokens, + cache_read_input_tokens=cache_read_tokens, + ) + if cache_creation_tokens: + return AnthropicUsage( + input_tokens=uncached_input_tokens, + output_tokens=output_tokens, + cache_creation_input_tokens=cache_creation_tokens, + ) + if cache_read_tokens: + return AnthropicUsage( + input_tokens=uncached_input_tokens, + output_tokens=output_tokens, + cache_read_input_tokens=cache_read_tokens, + ) + return AnthropicUsage(input_tokens=uncached_input_tokens, output_tokens=output_tokens) + + class AnthropicResponsesStreamWrapper: """ Wraps a Responses API streaming iterator and re-emits events in Anthropic SSE format. @@ -210,28 +254,25 @@ class AnthropicResponsesStreamWrapper: event.get("response") if isinstance(event, dict) else None ) stop_reason = "end_turn" - anthropic_usage: AnthropicUsage = AnthropicUsage(input_tokens=0, output_tokens=0) + raw_usage: Final = _get_field(response_obj, "usage") if response_obj is not None else None + anthropic_usage: Final = _translate_usage(raw_usage) if response_obj is not None: - status: Final = getattr(response_obj, "status", None) + status: Final = _get_field(response_obj, "status") if status == "incomplete": stop_reason = "max_tokens" - anthropic_usage = ( - LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage( - getattr(response_obj, "usage", None) - ) - ) # Check if tool_use was in the output to override stop_reason if response_obj is not None: - output: Final = getattr(response_obj, "output", []) or [] - for out_item in output: - out_type = getattr(out_item, "type", None) or ( - out_item.get("type") if isinstance(out_item, dict) else None - ) - if out_type == "function_call": - stop_reason = "tool_use" - break + output: Final = _get_field(response_obj, "output", []) + if isinstance(output, list): + for out_item in output: + out_type = getattr(out_item, "type", None) or ( + out_item.get("type") if isinstance(out_item, dict) else None + ) + if out_type == "function_call": + stop_reason = "tool_use" + break self._chunk_queue.append( { 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 aebbed88c70..71e16a15846 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 @@ -282,6 +282,82 @@ class TestProcessEventTextDeltaWithoutOutputItemAdded: ] +class TestDictShapedCompletedEvents: + def test_dict_usage_is_extracted(self): + chunks = _process_all( + [ + { + "type": "response.completed", + "response": { + "status": "completed", + "usage": { + "input_tokens": 11, + "output_tokens": 42, + "cache_read_input_tokens": 7, + }, + }, + } + ] + ) + assert chunks[0]["type"] == "message_delta" + assert chunks[0]["usage"] == { + "input_tokens": 4, + "output_tokens": 42, + "cache_read_input_tokens": 7, + } + + def test_openai_responses_cached_tokens_details_extracted(self): + chunks = _process_all( + [ + { + "type": "response.completed", + "response": { + "status": "completed", + "usage": { + "input_tokens": 100, + "output_tokens": 20, + "input_tokens_details": {"cached_tokens": 30}, + }, + }, + } + ] + ) + assert chunks[0]["usage"] == { + "input_tokens": 70, + "output_tokens": 20, + "cache_read_input_tokens": 30, + } + + def test_dict_incomplete_maps_to_max_tokens(self): + chunks = _process_all([{"type": "response.incomplete", "response": {"status": "incomplete"}}]) + assert chunks[0]["delta"]["stop_reason"] == "max_tokens" + + def test_dict_function_call_output_maps_to_tool_use(self): + chunks = _process_all( + [ + { + "type": "response.completed", + "response": { + "status": "completed", + "output": [{"type": "function_call", "name": "get_weather"}], + }, + } + ] + ) + assert chunks[0]["delta"]["stop_reason"] == "tool_use" + + def test_object_shaped_usage_still_extracted(self): + usage = SimpleNamespace( + input_tokens=3, + output_tokens=5, + cache_creation_input_tokens=0, + cache_read_input_tokens=0, + ) + response = SimpleNamespace(status="completed", usage=usage, output=[]) + chunks = _process_all([{"type": "response.completed", "response": response}]) + assert chunks[0]["usage"] == {"input_tokens": 3, "output_tokens": 5} + + class TestResponseCompletedUsage: """The Anthropic ``message_delta`` usage must report cache reads/writes and exclude them from ``input_tokens``, so spend is not billed at the uncached