fix(google): report realtime usage metrics once per generation

livekit-plugins/livekit-plugins-google/livekit/plugins/google/realtime/realtime_api.py

@@ -217,6 +217,10 @@ class _ResponseGeneration:
     """Whether the generation is done (set when the turn is complete)"""
     _extra_content_warned: bool = False
     """Whether we've warned about audio/text arriving after generation completed"""
+    _usage_metadata: types.UsageMetadata | None = None
+    """Latest cumulative usage metadata reported for this generation"""
+    _usage_emitted: bool = False
+    """Whether the usage metrics event was emitted for this generation"""
 
     def push_text(self, text: str) -> None:
         if self.text_ch.closed:
@@ -1251,14 +1255,17 @@ async def _recv_task(self, session: AsyncSession) -> None:
                                 ]
                             )
 
+                    # usage metadata must be recorded before the handlers that may end the
+                    # generation (server content with turn_complete, tool calls); the final
+                    # usage value is often bundled on the message that ends the turn
+                    if response.usage_metadata:
+                        self._handle_usage_metadata(response.usage_metadata)
                     if response.server_content:
                         self._handle_server_content(response.server_content)
                     if response.tool_call:
                         self._handle_tool_calls(response.tool_call)
                     if response.tool_call_cancellation:
                         self._handle_tool_call_cancellation(response.tool_call_cancellation)
-                    if response.usage_metadata:
-                        self._handle_usage_metadata(response.usage_metadata)
                     if response.go_away:
                         self._handle_go_away(response.go_away)
 
@@ -1517,6 +1524,7 @@ def _mark_current_generation_done(self) -> None:
 
         gen.function_ch.close()
         gen.message_ch.close()
+        self._report_usage_metrics(gen)
         gen._done = True
         if lk_google_debug:
             logger.debug("generation done", extra={"lk.pii.generation": str(gen)})
@@ -1613,9 +1621,25 @@ def _handle_usage_metadata(self, usage_metadata: types.UsageMetadata) -> None:
                 logger.warning("no active generation to report metrics for")
             return
 
-        duration = (
-            current_gen._completed_timestamp or time.time()
-        ) - current_gen._created_timestamp
+        if current_gen._done and current_gen._usage_emitted:
+            logger.debug("ignoring usage metadata; already reported for this generation")
+            return
+
+        # Gemini sends usageMetadata periodically during a turn, each one carrying the
+        # cumulative totals. Keep the latest value and report usage only once per
+        # generation: when the generation completes (see _mark_current_generation_done),
+        # or right away if it already completed and the final usage arrived on the
+        # message that ended the turn (or right after it).
+        current_gen._usage_metadata = usage_metadata
+        if current_gen._done:
+            self._report_usage_metrics(current_gen)
+
+    def _report_usage_metrics(self, gen: _ResponseGeneration) -> None:
+        usage_metadata = gen._usage_metadata
+        if usage_metadata is None or gen._usage_emitted:
+            return
+
+        duration = (gen._completed_timestamp or time.time()) - gen._created_timestamp
 
         def _token_details_map(
             token_details: list[types.ModalityTokenCount] | None,
@@ -1638,8 +1662,8 @@ def _token_details_map(
 
         metrics = RealtimeModelMetrics(
             label=self._realtime_model.label,
-            request_id=current_gen.response_id,
-            timestamp=current_gen._created_timestamp,
+            request_id=gen.response_id,
+            timestamp=gen._created_timestamp,
             duration=duration,
             # Gemini has no response-start event; generation creation can be triggered
             # by an input transcript or by the first output packet.
@@ -1668,6 +1692,7 @@ def _token_details_map(
                 model_name=self._realtime_model.model, model_provider=self._realtime_model.provider
             ),
         )
+        gen._usage_emitted = True
         self.emit("metrics_collected", metrics)
 
     def _handle_go_away(self, go_away: types.LiveServerGoAway) -> None:

livekit-plugins/livekit-plugins-google/tests/test_realtime_usage_metrics.py

@@ -0,0 +1,168 @@
+"""
+Regression tests for Gemini Live usage metrics emission.
+
+The Gemini Live API sends usageMetadata messages periodically, each carrying the
+token counts for the current generation request (accumulating within the turn),
+and a final one around turn completion. The realtime session used to emit a `metrics_collected` event for
+every one of these messages, so a single generation was reported (and summed by
+the usage collectors) multiple times.
+
+These tests pin the intended behavior: usage is reported exactly once per
+generation, with the final value seen for that generation (accumulated within
+the turn).
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from google.genai import types
+
+from livekit.agents import llm
+from livekit.agents.metrics import RealtimeModelMetrics
+from livekit.plugins.google.realtime.realtime_api import RealtimeSession
+
+pytestmark = pytest.mark.unit
+
+
+def _model_turn_message(text: str) -> types.LiveServerMessage:
+    return types.LiveServerMessage(
+        server_content=types.LiveServerContent(
+            model_turn=types.Content(parts=[types.Part(text=text)]),
+        )
+    )
+
+
+def _usage_message(
+    prompt_tokens: int, output_tokens: int, total_tokens: int
+) -> types.LiveServerMessage:
+    return types.LiveServerMessage(
+        usage_metadata=types.UsageMetadata(
+            prompt_token_count=prompt_tokens,
+            response_token_count=output_tokens,
+            total_token_count=total_tokens,
+        )
+    )
+
+
+def _turn_complete_message(
+    usage: types.UsageMetadata | None = None,
+) -> types.LiveServerMessage:
+    return types.LiveServerMessage(
+        server_content=types.LiveServerContent(turn_complete=True),
+        usage_metadata=usage,
+    )
+
+
+class _FakeAsyncSession:
+    """Yields a scripted sequence of LiveServerMessages, then stops the recv loop."""
+
+    def __init__(self, messages: list[types.LiveServerMessage], on_drained) -> None:
+        self._messages = list(messages)
+        self._on_drained = on_drained
+
+    async def receive(self):
+        for message in self._messages:
+            yield message
+        if self._on_drained:
+            self._on_drained()
+
+
+def _make_session(
+    messages: list[types.LiveServerMessage],
+) -> tuple[RealtimeSession, list[RealtimeModelMetrics]]:
+    model = MagicMock()
+    model.capabilities.audio_output = False
+    model.label = "google-realtime-test"
+    model.model = "gemini-test"
+    model.provider = "google"
+
+    session = object.__new__(RealtimeSession)
+    llm.RealtimeSession.__init__(session, model)  # EventEmitter + _realtime_model
+
+    session._opts = SimpleNamespace(tool_choice="auto", output_audio_transcription=None)
+    session._chat_ctx = llm.ChatContext.empty()
+    session._current_generation = None
+    session._pending_generation_fut = None
+    session._rejected_tool_calls = 0
+    session._num_retries = 0
+    session._session_error = None
+    session._session_resumption_handle = None
+    session._session_should_close = asyncio.Event()
+    session._session_lock = asyncio.Lock()
+
+    collected: list[RealtimeModelMetrics] = []
+    session.on("metrics_collected", collected.append)
+
+    session._active_session = _FakeAsyncSession(
+        messages, on_drained=session._session_should_close.set
+    )
+    return session, collected
+
+
+async def _run_recv_task(session: RealtimeSession) -> None:
+    await RealtimeSession._recv_task(session, session._active_session)
+
+
+async def test_usage_emitted_once_per_generation():
+    """One turn with an incremental and a final usageMetadata: a single emission."""
+    messages = [
+        _model_turn_message("hello"),
+        _usage_message(prompt_tokens=100, output_tokens=10, total_tokens=110),
+        _usage_message(prompt_tokens=100, output_tokens=40, total_tokens=140),
+        _turn_complete_message(),
+    ]
+
+    session, collected = _make_session(messages)
+    await _run_recv_task(session)
+
+    assert len(collected) == 1
+    assert collected[0].input_tokens == 100
+    assert collected[0].output_tokens == 40
+    assert collected[0].total_tokens == 140
+
+
+async def test_final_usage_bundled_with_turn_complete_is_reported():
+    """The final usageMetadata may arrive on the turnComplete message itself."""
+    messages = [
+        _model_turn_message("hello"),
+        _usage_message(prompt_tokens=100, output_tokens=10, total_tokens=110),
+        _turn_complete_message(
+            usage=types.UsageMetadata(
+                prompt_token_count=100,
+                response_token_count=40,
+                total_token_count=140,
+            )
+        ),
+    ]
+
+    session, collected = _make_session(messages)
+    await _run_recv_task(session)
+
+    assert len(collected) == 1
+    assert collected[0].output_tokens == 40
+    assert collected[0].total_tokens == 140
+
+
+async def test_usage_reported_once_per_turn_across_multiple_turns():
+    messages = [
+        _model_turn_message("hello"),
+        _usage_message(prompt_tokens=100, output_tokens=10, total_tokens=110),
+        _usage_message(prompt_tokens=100, output_tokens=40, total_tokens=140),
+        _turn_complete_message(),
+        _model_turn_message("bye"),
+        _usage_message(prompt_tokens=150, output_tokens=70, total_tokens=220),
+        _turn_complete_message(
+            usage=types.UsageMetadata(
+                prompt_token_count=150,
+                response_token_count=75,
+                total_token_count=225,
+            )
+        ),
+    ]
+
+    session, collected = _make_session(messages)
+    await _run_recv_task(session)
+
+    assert [metrics.output_tokens for metrics in collected] == [40, 75]

tests/test_plugin_google_realtime.py

@@ -273,6 +273,7 @@ async def test_ttft_is_unavailable_without_a_response_start_event(
         session._handle_usage_metadata(
             types.UsageMetadata(prompt_token_count=10, response_token_count=5, total_token_count=15)
         )
+        session._mark_current_generation_done()
 
         assert len(collected) == 1
         assert collected[0].ttft == -1