feat(telemetry): trace the end-of-turn wait and the turn handoff

livekit-agents/livekit/agents/telemetry/trace_types.py

@@ -73,6 +73,17 @@
 ATTR_EOU_SOURCE = "lk.eou.source"
 ATTR_EOU_DETECTION_DELAY = "lk.eou.detection_delay"
 ATTR_EOU_FROM_CACHE = "lk.eou.from_cache"
+# eou_wait span: from the user's last speech to the turn decision
+ATTR_EOU_OUTCOME = "lk.eou.outcome"
+"""How the wait ended: ``committed``, ``user_resumed``, or ``dropped``."""
+ATTR_EOU_WAIT_DURATION = "lk.eou.wait_duration"
+"""Seconds from the end of the user's speech to the turn decision."""
+ATTR_EOU_REARM_COUNT = "lk.eou.rearm_count"
+"""Times the endpointing wait restarted on a later trigger (late transcript, VAD)."""
+
+# speech scheduling
+ATTR_SPEECH_QUEUE_WAIT = "lk.speech.queue_wait"
+"""Seconds a speech handle waited in the queue before generation was authorized."""
 
 # metrics
 ATTR_LLM_METRICS = "lk.llm_metrics"

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

@@ -214,6 +214,28 @@ class _PausedSpeechInfo:
 
 
 # NOTE: AgentActivity isn't exposed to the public API
+def _end_user_turn(info: _EndOfTurnInfo) -> None:
+    """End the ``user_turn`` span the activity adopted from recognition: the turn is over once
+    ``on_user_turn_completed`` has run (or the turn was skipped), just before any reply.
+
+    Module-level for the same reason as ``_record_queue_wait``."""
+    if info.user_turn_span_adopted and info.user_turn_span is not None:
+        if info.user_turn_span.is_recording():
+            info.user_turn_span.end()
+        info.user_turn_span_adopted = False
+
+
+def _record_queue_wait(speech_handle: SpeechHandle) -> None:
+    """Stamp how long the speech sat in the queue on its agent_turn span.
+
+    Module-level on purpose: the reply tasks are also driven with lightweight stand-ins for
+    the activity in tests, which must not need to know about telemetry helpers."""
+    if (queue_wait := speech_handle._queue_wait()) is None:
+        return
+    span = trace.get_current_span(context=speech_handle._agent_turn_context)
+    span.set_attribute(trace_types.ATTR_SPEECH_QUEUE_WAIT, queue_wait)
+
+
 class AgentActivity(RecognitionHooks):
     def __init__(self, agent: Agent, sess: AgentSession) -> None:
         self._agent, self._session = agent, sess
@@ -2511,6 +2533,8 @@ def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
             self._cancel_false_interruption_timer()
 
         old_task = self._user_turn_completed_atask
+        # the turn is not over until the user hook has run: take the span and end it there
+        info.user_turn_span_adopted = info.user_turn_span is not None
         self._user_turn_completed_atask = self._create_speech_task(
             self._user_turn_completed_task(old_task, info),
             name="AgentActivity._user_turn_completed_task",
@@ -2520,6 +2544,14 @@ def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
     @utils.log_exceptions(logger=logger)
     async def _user_turn_completed_task(
         self, old_task: asyncio.Task[None] | None, info: _EndOfTurnInfo
+    ) -> None:
+        try:
+            await self._user_turn_completed_impl(old_task, info)
+        finally:
+            _end_user_turn(info)
+
+    async def _user_turn_completed_impl(
+        self, old_task: asyncio.Task[None] | None, info: _EndOfTurnInfo
     ) -> None:
         if old_task is not None:
             # We never cancel user code as this is very confusing.
@@ -2601,16 +2633,36 @@ async def _user_turn_completed_task(
         # Agent.chat_ctx
         temp_mutable_chat_ctx = self._agent.chat_ctx.copy()
         start_time = time.perf_counter()
-        try:
-            await self._agent.on_user_turn_completed(
-                temp_mutable_chat_ctx, new_message=user_message
-            )
-        except StopResponse:
-            return  # ignore this turn
-        except Exception:
-            logger.exception("error occurred during on_user_turn_completed")
-            return
+        # user code that gates the reply: a slow hook here is a gap between user_turn and
+        # agent_turn that nothing else explains
+        with tracer.start_as_current_span(
+            "on_user_turn_completed",
+            context=(
+                trace.set_span_in_context(info.user_turn_span)
+                if info.user_turn_span_adopted and info.user_turn_span is not None
+                else self._session._root_span_context
+            ),
+            attributes={trace_types.ATTR_AGENT_LABEL: self._agent.label},
+        ) as hook_span:
+            try:
+                await self._agent.on_user_turn_completed(
+                    temp_mutable_chat_ctx, new_message=user_message
+                )
+            except StopResponse:
+                hook_span.add_event("stop_response")
+                return  # ignore this turn
+            except Exception as e:
+                # the hook is user code and its message can quote the transcript; honour a
+                # redaction switched on for this session alone as well as the job's
+                trace_utils.record_exception(
+                    hook_span,
+                    e,
+                    redacted=self._session._redaction_enabled or trace_utils.redaction_enabled(),
+                )
+                logger.exception("error occurred during on_user_turn_completed")
+                return
 
+        _end_user_turn(info)
         on_user_turn_completed_delay = time.perf_counter() - start_time
         metrics_report["on_user_turn_completed_delay"] = on_user_turn_completed_delay
 
@@ -2942,6 +2994,7 @@ async def _tts_task_impl(
             authorization_tasks.append(asyncio.ensure_future(self._user_silence_event.wait()))
         await speech_handle.wait_if_not_interrupted(authorization_tasks)
         speech_handle._clear_authorization()
+        _record_queue_wait(speech_handle)
 
         if speech_handle.interrupted:
             current_span.set_attribute(trace_types.ATTR_SPEECH_INTERRUPTED, True)
@@ -3418,6 +3471,7 @@ def _end_segment() -> None:
             authorization_tasks.append(asyncio.ensure_future(self._user_silence_event.wait()))
         await speech_handle.wait_if_not_interrupted(authorization_tasks)
         speech_handle._clear_authorization()
+        _record_queue_wait(speech_handle)
 
         if speech_handle.interrupted:
             current_span.set_attribute(trace_types.ATTR_SPEECH_INTERRUPTED, True)
@@ -3803,6 +3857,7 @@ async def _realtime_reply_task(
         if speech_handle.allow_interruptions:
             authorization_tasks.append(asyncio.ensure_future(self._user_silence_event.wait()))
         await speech_handle.wait_if_not_interrupted(authorization_tasks)
+        _record_queue_wait(speech_handle)
         if speech_handle.interrupted:
             await utils.aio.cancel_and_wait(*authorization_tasks)
             return
@@ -4044,6 +4099,7 @@ async def _watch_generation_end() -> None:
             authorization_tasks.append(asyncio.ensure_future(self._user_silence_event.wait()))
         await speech_handle.wait_if_not_interrupted(authorization_tasks)
         speech_handle._clear_authorization()
+        _record_queue_wait(speech_handle)
 
         if speech_handle.interrupted:
             # nothing was played, but the response may still be generating server-side