@@ -946,8 +946,7 @@ def merge_pydantic_models(prev: C, curr: C) -> C:
         and isinstance(curr_val, (int, float))
         and not isinstance(curr_val, bool)
         and (
-            isinstance(curr, types.GenerateContentResponseUsageMetadata)
-            or field_name == 'token_count'
+            field_name == 'token_count'
             or field_name.endswith('_count')
             or field_name.endswith('_sum')
         )
@@ -973,7 +972,18 @@ def merge_pydantic_models(prev: C, curr: C) -> C:
     else:
       merged_data[field_name] = curr_val
 
-  return curr.model_copy(update=merged_data)
+  merged = curr.model_copy(update=merged_data)
+  if (
+      isinstance(prev, types.GenerateContentResponseUsageMetadata)
+      and isinstance(curr, types.GenerateContentResponseUsageMetadata)
+      and isinstance(merged, types.GenerateContentResponseUsageMetadata)
+  ):
+    merged._initial_prompt_token_count = (
+        prev.initial_prompt_token_count
+        if prev.initial_prompt_token_count is not None
+        else curr.initial_prompt_token_count
+    )
+  return merged
 
 
 def merge_continuation_responses(
 
@@ -6147,6 +6147,9 @@ def _generate_content_stream_with_continuation(
     )
     continuation_token: Optional[bytes] = None
     hop_finish_reason: Optional[types.FinishReason] = None
+    accumulated_usage: Optional[types.GenerateContentResponseUsageMetadata] = (
+        None
+    )
     is_first_hop = True
 
     while is_first_hop or (
@@ -6160,6 +6163,7 @@ def _generate_content_stream_with_continuation(
       )
       hop_finish_reason = None
       continuation_token = None
+      hop_usage: Optional[types.GenerateContentResponseUsageMetadata] = None
 
       try:
         for chunk in self._generate_content_stream(
@@ -6171,6 +6175,15 @@ def _generate_content_stream_with_continuation(
               hop_finish_reason = candidate.finish_reason
             if candidate.continuation_token:
               continuation_token = candidate.continuation_token
+          if chunk.usage_metadata is not None:
+            if accumulated_usage is not None:
+              merged_usage = _extra_utils.merge_pydantic_models(
+                  accumulated_usage, chunk.usage_metadata
+              )
+              chunk = chunk.model_copy(update={'usage_metadata': merged_usage})
+              hop_usage = merged_usage
+            else:
+              hop_usage = chunk.usage_metadata
           yield chunk
       except Exception:  # pylint: disable=broad-exception-caught
         # If the stream fails after emitting a checkpoint continuation_token
@@ -6182,6 +6195,8 @@ def _generate_content_stream_with_continuation(
             and _extra_utils.is_resumable_finish_reason(hop_finish_reason)
         ):
           raise
+      if hop_usage is not None:
+        accumulated_usage = hop_usage
 
   def generate_content(
       self,
@@ -8390,6 +8405,9 @@ async def _generate_content_stream_with_continuation(
     )
     continuation_token: Optional[bytes] = None
     hop_finish_reason: Optional[types.FinishReason] = None
+    accumulated_usage: Optional[types.GenerateContentResponseUsageMetadata] = (
+        None
+    )
     is_first_hop = True
 
     while is_first_hop or (
@@ -8403,6 +8421,7 @@ async def _generate_content_stream_with_continuation(
       )
       hop_finish_reason = None
       continuation_token = None
+      hop_usage: Optional[types.GenerateContentResponseUsageMetadata] = None
 
       try:
         response_stream = await self._generate_content_stream(
@@ -8415,6 +8434,15 @@ async def _generate_content_stream_with_continuation(
               hop_finish_reason = candidate.finish_reason
             if candidate.continuation_token:
               continuation_token = candidate.continuation_token
+          if chunk.usage_metadata is not None:
+            if accumulated_usage is not None:
+              merged_usage = _extra_utils.merge_pydantic_models(
+                  accumulated_usage, chunk.usage_metadata
+              )
+              chunk = chunk.model_copy(update={'usage_metadata': merged_usage})
+              hop_usage = merged_usage
+            else:
+              hop_usage = chunk.usage_metadata
           yield chunk
       except Exception:  # pylint: disable=broad-exception-caught
         # If the stream fails after emitting a checkpoint continuation_token
@@ -8426,6 +8454,8 @@ async def _generate_content_stream_with_continuation(
             and _extra_utils.is_resumable_finish_reason(hop_finish_reason)
         ):
           raise
+      if hop_usage is not None:
+        accumulated_usage = hop_usage
 
   async def generate_content(
       self,