fix(voice): report e2e_latency for the first reply after a handoff

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -47,6 +47,7 @@
 from ._utils import _set_participant_attributes
 from .agent import (
     Agent,
+    AgentTask,
     ModelSettings,
     _get_activity_task_info,
     _set_activity_task_info,
@@ -904,11 +905,18 @@ async def start(self, *, reuse_resources: _ReusableResources | None = None) -> N
                 @utils.log_exceptions(logger=logger)
                 async def _traceable_on_enter() -> None:
                     data = _OnEnterData(session=self._session, agent=self._agent)
+                    # the turn this agent was entered on; a turn the user commits while on_enter
+                    # runs is not its to decline
+                    entered_on = self._session._unanswered_user_metrics
                     try:
                         tk = _OnEnterContextVar.set(data)
                         await self._agent.on_enter()
                     finally:
                         _OnEnterContextVar.reset(tk)
+                        # an on_enter that returns without speaking has declined the turn;
+                        # speeches it created already took their copy
+                        if self._session._unanswered_user_metrics is entered_on:
+                            self._session._unanswered_user_metrics = None
 
                 self._on_enter_task = task = self._create_speech_task(
                     _traceable_on_enter(), name="AgentTask_on_enter"
@@ -1564,6 +1572,7 @@ def say(
             "speech_created",
             SpeechCreatedEvent(speech_handle=handle, user_initiated=True, source="say"),
         )
+        user_metrics = self._take_on_enter_user_metrics()
 
         if (
             self._rt_session is not None
@@ -1594,6 +1603,7 @@ def say(
                     audio=audio or None,
                     add_to_chat_ctx=add_to_chat_ctx,
                     model_settings=ModelSettings(),
+                    _previous_user_metrics=user_metrics,
                 ),
                 speech_handle=handle,
                 name="AgentActivity.tts_say",
@@ -1659,6 +1669,7 @@ def _generate_reply(
             "speech_created",
             SpeechCreatedEvent(speech_handle=handle, user_initiated=True, source="generate_reply"),
         )
+        user_metrics = self._take_on_enter_user_metrics()
 
         if isinstance(self.llm, llm.RealtimeModel):
             self._create_speech_task(
@@ -1689,6 +1700,7 @@ def _generate_reply(
                         if utils.is_given(tool_choice) or self._tool_choice is None
                         else self._tool_choice
                     ),
+                    _previous_user_metrics=user_metrics,
                 ),
                 speech_handle=handle,
                 name="AgentActivity.pipeline_reply",
@@ -2885,6 +2897,7 @@ async def _tts_task(
         audio: AsyncIterable[rtc.AudioFrame] | None,
         add_to_chat_ctx: bool,
         model_settings: ModelSettings,
+        _previous_user_metrics: llm.MetricsReport | None = None,
     ) -> None:
         with tracer.start_as_current_span(
             "agent_turn", context=self._session._root_span_context
@@ -2909,6 +2922,7 @@ async def _tts_task(
                     audio=audio,
                     add_to_chat_ctx=add_to_chat_ctx,
                     model_settings=model_settings,
+                    _previous_user_metrics=_previous_user_metrics,
                 )
             finally:
                 otel_metrics.record_invoke_agent_duration(
@@ -2922,6 +2936,7 @@ async def _tts_task_impl(
         audio: AsyncIterable[rtc.AudioFrame] | None,
         add_to_chat_ctx: bool,
         model_settings: ModelSettings,
+        _previous_user_metrics: llm.MetricsReport | None = None,
     ) -> None:
         current_span = trace.get_current_span(context=speech_handle._agent_turn_context)
         current_span.set_attribute(trace_types.ATTR_SPEECH_ID, speech_handle.id)
@@ -3097,21 +3112,25 @@ def _on_first_frame(fut: asyncio.Future[float] | asyncio.Future[None]) -> None:
                 forwarded_text = ""
         current_span.set_attribute(trace_types.ATTR_RESPONSE_TEXT, forwarded_text)
 
-        if forwarded_text and add_to_chat_ctx:
-            assistant_metrics: llm.MetricsReport = {}
+        assistant_metrics: llm.MetricsReport = {}
 
-            if tts_gen_data and tts_gen_data.ttfb is not None:
-                assistant_metrics["tts_node_ttfb"] = tts_gen_data.ttfb
+        if tts_gen_data and tts_gen_data.ttfb is not None:
+            assistant_metrics["tts_node_ttfb"] = tts_gen_data.ttfb
 
-            if stopped_speaking_at and started_speaking_at:
-                assistant_metrics["started_speaking_at"] = started_speaking_at
-                assistant_metrics["stopped_speaking_at"] = stopped_speaking_at
+        if stopped_speaking_at and started_speaking_at:
+            assistant_metrics["started_speaking_at"] = started_speaking_at
+            assistant_metrics["stopped_speaking_at"] = stopped_speaking_at
 
-                if started_forwarding_at is not None:
-                    assistant_metrics["playback_latency"] = (
-                        started_speaking_at - started_forwarding_at
-                    )
+            if started_forwarding_at is not None:
+                assistant_metrics["playback_latency"] = started_speaking_at - started_forwarding_at
 
+            # the audio answers the user turn, stored message or not
+            if _previous_user_metrics and "stopped_speaking_at" in _previous_user_metrics:
+                e2e_latency = started_speaking_at - _previous_user_metrics["stopped_speaking_at"]
+                assistant_metrics["e2e_latency"] = e2e_latency
+                current_span.set_attribute(trace_types.ATTR_E2E_LATENCY, e2e_latency)
+
+        if forwarded_text and add_to_chat_ctx:
             msg = self._agent._chat_ctx.add_message(
                 role="assistant",
                 content=forwarded_text,
@@ -3149,6 +3168,19 @@ def _on_enter_ignored_tools(self, tool_ctx: llm.ToolContext) -> list[llm.Tool]:
             and tool.info.flags & ToolFlag.IGNORE_ON_ENTER
         ]
 
+    def _take_on_enter_user_metrics(self) -> llm.MetricsReport | None:
+        """Claim the unanswered user turn for a speech created inside on_enter."""
+        on_enter_data = _OnEnterContextVar.get(None)
+        if (
+            on_enter_data is None
+            or on_enter_data.agent != self._agent
+            or on_enter_data.session != self._session
+        ):
+            return None
+        metrics = self._session._unanswered_user_metrics
+        self._session._unanswered_user_metrics = None
+        return metrics
+
     @utils.log_exceptions(logger=logger)
     async def _pipeline_reply_task(
         self,
@@ -3395,6 +3427,7 @@ def _end_segment() -> None:
             self._agent._chat_ctx.insert(new_message)
             self._session._conversation_item_added(new_message)
             user_metrics = new_message.metrics
+            self._session._unanswered_user_metrics = user_metrics
 
         if speech_handle.interrupted:
             current_span.set_attribute(trace_types.ATTR_SPEECH_INTERRUPTED, True)
@@ -3596,6 +3629,9 @@ async def _next_segment() -> _SpeechSegment | None:
                 assistant_metrics["e2e_latency"] = e2e_latency
                 current_span.set_attribute(trace_types.ATTR_E2E_LATENCY, e2e_latency)
 
+            if self._session._unanswered_user_metrics is user_metrics:
+                self._session._unanswered_user_metrics = None
+
         current_span.set_attribute(trace_types.ATTR_SPEECH_INTERRUPTED, speech_handle.interrupted)
 
         forwarded_text = "".join(out.forwarded_text for out in segment_outputs)
@@ -3682,6 +3718,9 @@ async def _next_segment() -> _SpeechSegment | None:
 
         # important: no agent output should be used after this point
 
+        # the reply chain goes on through a tool reply, a handoff, or an inline task handing
+        # back to the tool that awaited it; otherwise it ends here and nothing answers the turn
+        chain_continues = False
         if len(tool_output.output) > 0:
             max_steps_reached = speech_handle.num_steps >= self._session.options.max_tool_steps + 1
 
@@ -3725,6 +3764,12 @@ async def _next_segment() -> _SpeechSegment | None:
                 self._session.update_agent(new_agent_task)
                 draining = True
 
+            chain_continues = (
+                fnc_executed_ev.has_tool_reply
+                or self._session._agent is not self._agent
+                or (isinstance(self._agent, AgentTask) and self._agent.done())
+            )
+
             tool_messages = new_calls + new_fnc_outputs
             # commit now so results survive even if the reply speech never runs (#3702)
             if tool_messages:
@@ -3770,9 +3815,10 @@ async def _next_segment() -> _SpeechSegment | None:
                             if max_steps_reached or draining or model_settings.tool_choice == "none"
                             else "auto",
                         ),
-                        # in case the current reply only generated tools (no speech), re-use the current user_metrics for the next
-                        # tool response generation
-                        _previous_user_metrics=user_metrics if not forwarded_text else None,
+                        # the tool reply answers whatever user turn is still unanswered: this
+                        # one if the reply only generated tools, or the last turn of a
+                        # sub-conversation an inline AgentTask ran inside the tool
+                        _previous_user_metrics=self._session._unanswered_user_metrics,
                     ),
                     speech_handle=speech_handle,
                     name="AgentActivity.pipeline_reply",
@@ -3782,6 +3828,9 @@ async def _next_segment() -> _SpeechSegment | None:
                     speech_handle, SpeechHandle.SPEECH_PRIORITY_NORMAL, force=True
                 )
 
+        if not chain_continues:
+            self._session._unanswered_user_metrics = None
+
     @utils.log_exceptions(logger=logger)
     async def _realtime_reply_task(
         self,

livekit-agents/livekit/agents/voice/agent_session.py

@@ -698,6 +698,8 @@ def __init__(
         self._foreground_guards: set[asyncio.Future[None]] = set()
         # TODO(theomonnom): need a better way to expose early assistant metrics
         self._early_assistant_metrics: MetricsReport | None = None
+        # the latest user turn no agent speech has reported e2e_latency for yet
+        self._unanswered_user_metrics: MetricsReport | None = None
 
         # trace
         self._user_speaking_span: trace.Span | None = None

tests/fake_vad.py

@@ -27,6 +27,9 @@ def __init__(
         self._fake_user_speeches = fake_user_speeches
         self._min_speech_duration = min_speech_duration
         self._min_silence_duration = min_silence_duration
+        # one clock for every stream: a handoff opens a new stream mid-run, timed against
+        # the same origin the fake STT uses
+        self._start_time: float | None = None
 
     def stream(self) -> VADStream:
         return FakeVADStream(self)
@@ -42,13 +45,16 @@ async def _main_task(self) -> None:
         if not self._vad._fake_user_speeches:
             return
 
-        async for input_frame in self._input_ch:
-            if isinstance(input_frame, rtc.AudioFrame):
-                break
-        else:
-            return
-
-        start_time = time.perf_counter()
+        if self._vad._start_time is None:
+            # the clock starts with the first frame; the fake input pushes one burst, so a
+            # later stream must not wait for another
+            async for input_frame in self._input_ch:
+                if isinstance(input_frame, rtc.AudioFrame):
+                    break
+            else:
+                return
+            self._vad._start_time = time.perf_counter()
+        start_time = self._vad._start_time
 
         def current_time() -> float:
             return time.perf_counter() - start_time
@@ -57,6 +63,9 @@ def current_time() -> float:
             next_start_of_speech_time = fake_speech.start_time + self._vad._min_speech_duration
             next_end_of_speech_time = fake_speech.end_time + self._vad._min_silence_duration
 
+            if current_time() >= next_end_of_speech_time:
+                continue  # already over before this stream opened
+
             if current_time() < next_start_of_speech_time:
                 await asyncio.sleep(next_start_of_speech_time - current_time())