Update STT metrics to include token usage and emit gpt-realtime transcription STT token counts

livekit-agents/livekit/agents/metrics/base.py

@@ -40,15 +40,23 @@ class STTMetrics(_BaseMetrics):
     request_id: str
     timestamp: float
     duration: float
-    """The request duration in seconds, 0.0 if the STT is streaming."""
+    """The request duration in seconds,
+    0.0 if the STT is streaming.
+    0.0 if the duration is meaningless because there is no blocking request (e.g. gpt-realtime model).
+    """
     audio_duration: float
-    """The duration of the pushed audio in seconds."""
-    input_tokens: int = 0
-    """Input audio tokens (for token-based billing)."""
-    output_tokens: int = 0
-    """Output text tokens (for token-based billing)."""
+    """The duration of the pushed audio in seconds.
+    0.0 if duration irrelevant/ or unknown like in streaming STT or token billing."""
     streamed: bool
     """Whether the STT is streaming (e.g using websocket)."""
+    input_tokens: int = 0
+    """Total input tokens (for token-based billing)."""
+    output_tokens: int = 0
+    """Total output tokens (for token-based billing)."""
+    total_tokens: int = 0
+    """Sum of input and output tokens."""
+    input_audio_tokens: int = 0
+    """Number of audio input tokens billed by the ASR model."""
     acquire_time: float = 0.0
     """Time in seconds to acquire the connection. (WebSocket only)"""
     connection_reused: bool = False

livekit-agents/livekit/agents/metrics/usage_collector.py

@@ -29,6 +29,9 @@ class UsageSummary:
     tts_characters_count: int = 0
     tts_audio_duration: float = 0.0
     stt_audio_duration: float = 0.0
+    stt_input_tokens: int = 0
+    stt_output_tokens: int = 0
+    stt_input_audio_tokens: int = 0
 
     def __post_init__(self) -> None:
         warnings.warn(
@@ -114,6 +117,9 @@ def collect(self, metrics: AgentMetrics) -> None:
 
         elif isinstance(metrics, STTMetrics):
             self._summary.stt_audio_duration += metrics.audio_duration
+            self._summary.stt_input_tokens += metrics.input_tokens or 0
+            self._summary.stt_output_tokens += metrics.output_tokens or 0
+            self._summary.stt_input_audio_tokens += metrics.input_audio_tokens or 0
 
     def get_summary(self) -> UsageSummary:
         return deepcopy(self._summary)

livekit-agents/livekit/agents/metrics/utils.py

@@ -89,6 +89,10 @@ def log_metrics(metrics: AgentMetrics, *, logger: logging.Logger | None = None)
             extra=metadata
             | {
                 "audio_duration": round(metrics.audio_duration, 2),
+                "input_tokens": metrics.input_tokens,
+                "output_tokens": metrics.output_tokens,
+                "total_tokens": metrics.total_tokens,
+                "input_audio_tokens": metrics.input_audio_tokens,
             },
         )
     elif isinstance(metrics, InterruptionMetrics):

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/realtime_model.py

@@ -18,7 +18,7 @@
 
 from livekit import rtc
 from livekit.agents import APIConnectionError, APIError, io, llm, utils
-from livekit.agents.metrics import RealtimeModelMetrics
+from livekit.agents.metrics import RealtimeModelMetrics, STTMetrics
 from livekit.agents.metrics.base import Metadata
 from livekit.agents.types import (
     DEFAULT_API_CONNECT_OPTIONS,
@@ -1782,6 +1782,8 @@ def _handle_conversion_item_input_audio_transcription_completed(
             ),
         )
 
+        self._emit_transcription_metrics(event)
+
     def _handle_conversion_item_input_audio_transcription_failed(
         self, event: ConversationItemInputAudioTranscriptionFailedEvent
     ) -> None:
@@ -1790,6 +1792,57 @@ def _handle_conversion_item_input_audio_transcription_failed(
             extra={"error": event.error},
         )
 
+    def _emit_transcription_metrics(
+        self, event: ConversationItemInputAudioTranscriptionCompletedEvent
+    ) -> None:
+        from openai.types.realtime.conversation_item_input_audio_transcription_completed_event import (
+            UsageTranscriptTextUsageDuration,
+            UsageTranscriptTextUsageTokens,
+        )
+
+        transcription_opts = self._realtime_model._opts.input_audio_transcription
+        transcription_model = transcription_opts.model if transcription_opts else None
+        metadata = Metadata(
+            model_name=transcription_model,
+            model_provider=self._realtime_model.provider,
+        )
+
+        usage = event.usage
+        if isinstance(usage, UsageTranscriptTextUsageTokens):
+            details = usage.input_token_details
+            input_audio_tokens = (
+                details.audio_tokens if details and details.audio_tokens is not None else 0
+            )
+            stt_metrics = STTMetrics(
+                request_id=event.event_id,
+                timestamp=time.time(),
+                # Request processing time is 0 so duration is 0.0 (server side events
+                duration=0.0,
+                label=self._realtime_model.label,
+                audio_duration=0.0,
+                # uses websocket but results are not streamed incrementally so streamed=False
+                streamed=False,
+                input_tokens=usage.input_tokens,
+                output_tokens=usage.output_tokens,
+                total_tokens=usage.total_tokens,
+                input_audio_tokens=input_audio_tokens,
+                metadata=metadata,
+            )
+            self.emit("metrics_collected", stt_metrics)
+        elif isinstance(usage, UsageTranscriptTextUsageDuration):
+            stt_metrics = STTMetrics(
+                request_id=event.event_id,
+                timestamp=time.time(),
+                # Request processing time is 0 so duration is 0.0 (server side events)
+                duration=0.0,
+                label=self._realtime_model.label,
+                audio_duration=usage.seconds,
+                # uses websocket but results are not streamed incrementally so streamed=False
+                streamed=False,
+                metadata=metadata,
+            )
+            self.emit("metrics_collected", stt_metrics)
+
     def _handle_response_text_delta(self, event: ResponseTextDeltaEvent) -> None:
         assert self._current_generation is not None, "current_generation is None"
         item_generation = self._current_generation.messages[event.item_id]