fix(responses): handle dict-backed streaming response logs

litellm/litellm_core_utils/litellm_logging.py

@@ -3170,21 +3170,21 @@ class Logging(LiteLLMLoggingBaseClass):
             (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent),
         ):
             ## return unified Usage object
-            if isinstance(result.response.usage, ResponseAPIUsage):
-                transformed_usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(
-                    result.response.usage
-                )
+            response = result.response
+            usage = response.get("usage") if isinstance(response, dict) else getattr(response, "usage", None)
+            if isinstance(usage, ResponseAPIUsage):
+                transformed_usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(usage)
                 # Set as dict instead of Usage object so model_dump() serializes it correctly
-                setattr(
-                    result.response,
-                    "usage",
-                    (
-                        transformed_usage.model_dump()
-                        if hasattr(transformed_usage, "model_dump")
-                        else dict(transformed_usage)
-                    ),
+                transformed_usage_dict = (
+                    transformed_usage.model_dump()
+                    if hasattr(transformed_usage, "model_dump")
+                    else dict(transformed_usage)
                 )
-            return result.response
+                if isinstance(response, dict):
+                    response["usage"] = transformed_usage_dict
+                else:
+                    setattr(response, "usage", transformed_usage_dict)
+            return response
         else:
             return None
 

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -2478,6 +2478,82 @@ def test_get_assembled_streaming_response_returns_result_for_streaming():
     assert assembled is result
 
 
+def _assemble_responses_terminal_event(response: object) -> object:
+    import datetime
+
+    from litellm.types.llms.openai import (
+        ResponseCompletedEvent,
+        ResponsesAPIStreamEvents,
+    )
+
+    event = ResponseCompletedEvent.model_construct(
+        type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
+        response=response,
+    )
+    return _make_logging_obj(stream=True)._get_assembled_streaming_response(
+        result=event,
+        start_time=datetime.datetime.now(),
+        end_time=datetime.datetime.now(),
+        is_async=True,
+        streaming_chunks=[],
+    )
+
+
+def test_get_assembled_streaming_response_accepts_dict_response_usage() -> None:
+    response = {
+        "id": "resp-1",
+        "created_at": 0,
+        "usage": {"input_tokens": 3, "output_tokens": 5, "total_tokens": 8},
+    }
+
+    assert _assemble_responses_terminal_event(response) is response
+    assert response["usage"] == {
+        "input_tokens": 3,
+        "output_tokens": 5,
+        "total_tokens": 8,
+    }
+
+
+def test_get_assembled_streaming_response_transforms_typed_dict_usage() -> None:
+    from litellm.types.llms.openai import ResponseAPIUsage
+
+    response = {
+        "id": "resp-1",
+        "created_at": 0,
+        "usage": ResponseAPIUsage(input_tokens=3, output_tokens=5, total_tokens=8),
+    }
+
+    assert _assemble_responses_terminal_event(response) is response
+    assert isinstance(response["usage"], dict)
+    assert response["usage"]["prompt_tokens"] == 3
+    assert response["usage"]["completion_tokens"] == 5
+    assert response["usage"]["total_tokens"] == 8
+
+
+def test_get_assembled_streaming_response_preserves_typed_responses_usage() -> None:
+    from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
+
+    response = ResponsesAPIResponse(
+        id="resp-1",
+        created_at=0,
+        output=[],
+        usage=ResponseAPIUsage(input_tokens=3, output_tokens=5, total_tokens=8),
+    )
+
+    assert _assemble_responses_terminal_event(response) is response
+    assert isinstance(response.usage, dict)
+    assert response.usage["prompt_tokens"] == 3
+    assert response.usage["completion_tokens"] == 5
+    assert response.usage["total_tokens"] == 8
+
+
+def test_get_assembled_streaming_response_preserves_dict_without_usage() -> None:
+    response = {"id": "resp-1", "created_at": 0}
+
+    assert _assemble_responses_terminal_event(response) is response
+    assert response == {"id": "resp-1", "created_at": 0}
+
+
 def test_streaming_success_handler_includes_vertex_ai_metadata_in_standard_logging():
     """Assembled streaming responses should include Vertex AI metadata in logging payload."""
     import datetime