Merge be33f1fd68 into 30ff3723b2

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(
                 {

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