fix(usage): correct calculate usage with cached tokens when use ChatCompletionUsageBlock

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -604,6 +604,8 @@ def _calculate_usage_per_chunk(
                 usage_chunk = chunk._hidden_params.get("usage", None)
 
             if usage_chunk is not None:
+                if isinstance(usage_chunk, dict):
+                    usage_chunk = Usage(**usage_chunk)
                 usage_chunk_dict = self._usage_chunk_calculation_helper(usage_chunk)
                 if (
                     usage_chunk_dict["prompt_tokens"] is not None

tests/test_litellm/litellm_core_utils/test_streaming_chunk_builder_utils.py

@@ -8,7 +8,8 @@
     0, os.path.abspath("../../..")
 )  # Adds the parent directory to the system path
 
-from litellm import stream_chunk_builder
+from litellm import ChatCompletionUsageBlock, stream_chunk_builder
+from litellm.types.utils import GenericStreamingChunk
 from litellm.litellm_core_utils.streaming_chunk_builder_utils import ChunkProcessor
 from litellm.types.utils import (
     ChatCompletionDeltaToolCall,
@@ -324,6 +325,42 @@ def test_cache_read_input_tokens_retained():
     assert usage.cache_read_input_tokens == 11775
     assert usage.prompt_tokens_details.cached_tokens == 11775
 
+def test_cache_read_input_tokens_retained_genericstreamingchunk():
+    chunk1 = GenericStreamingChunk(
+        text="Test1",
+        is_finished=False,
+        finish_reason="",
+        usage=None,
+        index=1,
+    )
+
+    chunk2 = GenericStreamingChunk(
+        text="Test2",
+        is_finished=True,
+        finish_reason="stop",
+        usage=ChatCompletionUsageBlock(
+            completion_tokens=5,
+            prompt_tokens=1234,
+            total_tokens=1239,
+            completion_tokens_details=None,
+            prompt_tokens_details=PromptTokensDetails(
+                audio_tokens=None, cached_tokens=543
+            ).model_dump(),
+        ),
+        index=2,
+    )
+
+    # Use dictionaries directly instead of ModelResponseStream
+    chunks = [chunk1, chunk2]
+    processor = ChunkProcessor(chunks=chunks)
+
+    usage = processor.calculate_usage(
+        chunks=chunks,
+        model="gpt-5.5",
+        completion_output="",
+    )
+
+    assert usage.prompt_tokens_details.cached_tokens == 543
 
 def test_stream_chunk_builder_litellm_usage_chunks():
     """