feat: expose raw usage metadata in LLMTokenUsage for Gemini Live and OpenAI Realtime

src/pipecat/metrics/metrics.py

@@ -11,6 +11,8 @@
 processing statistics.
 """
 
+from typing import Any, Dict, Optional
+
 from pydantic import BaseModel
 
 
@@ -60,9 +62,10 @@ class LLMTokenUsage(BaseModel):
     prompt_tokens: int
     completion_tokens: int
     total_tokens: int
-    cache_read_input_tokens: int | None = None
-    cache_creation_input_tokens: int | None = None
-    reasoning_tokens: int | None = None
+    cache_read_input_tokens: Optional[int] = None
+    cache_creation_input_tokens: Optional[int] = None
+    reasoning_tokens: Optional[int] = None
+    raw_usage_metadata: Optional[Dict[str, Any]] = None
 
 
 class LLMUsageMetricsData(MetricsData):

src/pipecat/services/google/gemini_live/llm.py

@@ -2041,12 +2041,48 @@ async def _handle_msg_usage_metadata(self, message: LiveServerMessage):
         completion_tokens = usage.response_token_count or 0
         total_tokens = usage.total_token_count or (prompt_tokens + completion_tokens)
 
+        # Extract full raw details dict from google's class model
+        raw_usage = None
+        if hasattr(usage, "to_dict"):
+            try:
+                raw_usage = usage.to_dict()
+            except Exception:
+                pass
+        if not raw_usage:
+            try:
+                raw_usage = {}
+                for attr in [
+                    "prompt_token_count", "response_token_count", "total_token_count",
+                    "cached_content_token_count", "thoughts_token_count",
+                    "prompt_tokens_details", "response_tokens_details",
+                    "tool_use_prompt_tokens_details", "cache_tokens_details"
+                ]:
+                    val = getattr(usage, attr, None)
+                    if val is not None:
+                        if isinstance(val, list):
+                            dict_list = []
+                            for item in val:
+                                if hasattr(item, "to_dict"):
+                                    dict_list.append(item.to_dict())
+                                elif hasattr(item, "__dict__"):
+                                    dict_list.append({k: v for k, v in vars(item).items() if not k.startswith("_")})
+                                else:
+                                    dict_list.append(item)
+                            raw_usage[attr] = dict_list
+                        elif hasattr(val, "to_dict"):
+                            raw_usage[attr] = val.to_dict()
+                        else:
+                            raw_usage[attr] = val
+            except Exception:
+                pass
+
         tokens = LLMTokenUsage(
             prompt_tokens=prompt_tokens,
             completion_tokens=completion_tokens,
             total_tokens=total_tokens,
             cache_read_input_tokens=usage.cached_content_token_count,
             reasoning_tokens=usage.thoughts_token_count,
+            raw_usage_metadata=raw_usage,
         )
 
         await self.start_llm_usage_metrics(tokens)

src/pipecat/services/openai/realtime/llm.py

@@ -951,11 +951,43 @@ async def _handle_evt_response_done(self, evt):
             and evt.response.usage.input_token_details
             else None
         )
+
+        raw_usage = None
+        if hasattr(evt.response.usage, "to_dict"):
+            try:
+                raw_usage = evt.response.usage.to_dict()
+            except Exception:
+                pass
+        elif hasattr(evt.response.usage, "model_dump"):
+            try:
+                raw_usage = evt.response.usage.model_dump()
+            except Exception:
+                pass
+
+        if not raw_usage:
+            try:
+                raw_usage = {}
+                for attr in [
+                    "input_tokens", "output_tokens", "total_tokens",
+                    "input_token_details", "output_token_details"
+                ]:
+                    val = getattr(evt.response.usage, attr, None)
+                    if val is not None:
+                        if hasattr(val, "to_dict"):
+                            raw_usage[attr] = val.to_dict()
+                        elif hasattr(val, "model_dump"):
+                            raw_usage[attr] = val.model_dump()
+                        else:
+                            raw_usage[attr] = val
+            except Exception:
+                pass
+
         tokens = LLMTokenUsage(
             prompt_tokens=evt.response.usage.input_tokens,
             completion_tokens=evt.response.usage.output_tokens,
             total_tokens=evt.response.usage.total_tokens,
             cache_read_input_tokens=cached_tokens,
+            raw_usage_metadata=raw_usage,
         )
         await self.start_llm_usage_metrics(tokens)
         await self.stop_processing_metrics()