feat(telemetry): one agent_turn span per speech handle

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

@@ -107,7 +107,12 @@
 
 # agent turn
 ATTR_AGENT_TURN_ID = "lk.generation_id"
+"""On ``agent_turn``: the latest generation (LLM step) of the speech; each step is also a
+``generation`` event carrying its own id."""
 ATTR_AGENT_PARENT_TURN_ID = "lk.parent_generation_id"
+ATTR_GENERATION_COUNT = "lk.generation_count"
+"""On ``agent_turn``: how many generations (LLM steps) the speech took; more than one means
+tool calls were executed before the final reply."""
 ATTR_USER_INPUT = "lk.pii.user_input"
 ATTR_INSTRUCTIONS = "lk.pii.instructions"
 ATTR_SPEECH_INTERRUPTED = "lk.interrupted"

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

@@ -6,7 +6,7 @@
 import heapq
 import json
 import time
-from collections.abc import AsyncGenerator, AsyncIterable, Coroutine
+from collections.abc import AsyncGenerator, AsyncIterable, Coroutine, Iterator
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Literal
 
@@ -245,11 +245,74 @@ def _record_interruption(speech_handle: SpeechHandle) -> None:
     if not speech_handle.interrupted or speech_handle._agent_turn_context is None:
         return
     span = trace.get_current_span(context=speech_handle._agent_turn_context)
+    if not span.is_recording():
+        return
     span.set_attribute(
         trace_types.ATTR_INTERRUPTION_SOURCE, speech_handle._interrupt_source or "programmatic"
     )
 
 
+@contextlib.contextmanager
+def _agent_turn(
+    speech_handle: SpeechHandle,
+    *,
+    root_context: otel_context.Context | None,
+    agent_label: str,
+) -> Iterator[trace.Span]:
+    """The speech's ``agent_turn`` span, made current for one generation.
+
+    One speech handle is one agent turn, however many LLM steps it takes: the follow-up
+    generation after a tool call runs in a new task but continues the open span instead of
+    opening a second turn. Each generation is a ``generation`` event on the span, whose
+    ``lk.generation_id`` names the latest one and ``lk.generation_count`` how many there were.
+    The span ends with the speech (``SpeechHandle._mark_done``), not with the step.
+
+    Module-level for the same reason as ``_record_queue_wait``."""
+    span = speech_handle._agent_turn_span
+    if span is None:
+        span = tracer.start_span(
+            "agent_turn",
+            context=root_context,
+            attributes={trace_types.ATTR_SPEECH_ID: speech_handle.id},
+        )
+        # an agent turn is the convention's `invoke_agent`: the framework running the agent
+        # in-process, with the inference and tool spans nested underneath
+        gen_ai_telemetry.set_agent_attributes(
+            span,
+            operation=trace_types.GenAIOperationName.INVOKE_AGENT,
+            agent_name=agent_label,
+        )
+        speech_handle._agent_turn_span = span
+        speech_handle._agent_turn_context = trace.set_span_in_context(span)
+        speech_handle._agent_turn_started_at = time.perf_counter()
+        speech_handle._agent_turn_agent_name = agent_label
+
+    generation_attrs: dict[str, Any] = {
+        trace_types.ATTR_AGENT_TURN_ID: speech_handle._generation_id
+    }
+    if parent_id := speech_handle._parent_generation_id:
+        generation_attrs[trace_types.ATTR_AGENT_PARENT_TURN_ID] = parent_id
+    span.add_event("generation", generation_attrs)
+    span.set_attributes(
+        {
+            trace_types.ATTR_AGENT_TURN_ID: speech_handle._generation_id,
+            trace_types.ATTR_GENERATION_COUNT: speech_handle._num_steps,
+        }
+    )
+    with tracer.use_span(span, end_on_exit=False):
+        yield span
+
+
+def _continue_discarded_turn(discarded: SpeechHandle | None, successor: SpeechHandle) -> None:
+    """A preemptive generation discarded for ``successor`` (a newer attempt, or the real reply
+    after the transcript changed) hands its open ``agent_turn`` over, so one turn shows the
+    wasted generation and the one that answered. Module-level like ``_record_queue_wait``."""
+    if discarded is None or discarded is successor:
+        return
+    if (carry := discarded._take_agent_turn()) is not None:
+        successor._continue_agent_turn(carry, discarded=discarded)
+
+
 def _record_queue_wait(speech_handle: SpeechHandle) -> None:
     """Stamp how long the speech sat in the queue on its agent_turn span.
 
@@ -259,7 +322,8 @@ def _record_queue_wait(speech_handle: SpeechHandle) -> None:
     ) is None or speech_handle._agent_turn_context is None:
         return  # no agent_turn span yet: never fall back to whatever span is current
     span = trace.get_current_span(context=speech_handle._agent_turn_context)
-    span.set_attribute(trace_types.ATTR_SPEECH_QUEUE_WAIT, queue_wait)
+    if span.is_recording():
+        span.set_attribute(trace_types.ATTR_SPEECH_QUEUE_WAIT, queue_wait)
 
 
 # NOTE: AgentActivity isn't exposed to the public API
@@ -2473,6 +2537,13 @@ def on_preemptive_generation(self, info: _PreemptiveGenerationInfo) -> None:
         ):
             return
 
+        # a newer attempt supersedes the current one; if one is created below it continues the
+        # discarded attempt's agent_turn (the cancelled speech only ends once the loop runs)
+        discarded = (
+            self._preemptive_generation.speech_handle
+            if self._preemptive_generation is not None
+            else None
+        )
         self._cancel_preemptive_generation()
 
         if (
@@ -2500,6 +2571,7 @@ def on_preemptive_generation(self, info: _PreemptiveGenerationInfo) -> None:
             schedule_speech=False,
             input_details=InputDetails(modality="audio"),
         )
+        _continue_discarded_turn(discarded, speech_handle)
 
         self._preemptive_generation = _PreemptiveGeneration(
             speech_handle=speech_handle,
@@ -2739,6 +2811,7 @@ async def _user_turn_completed_impl(
             return
 
         speech_handle: SpeechHandle | None = None
+        discarded_preemptive: SpeechHandle | None = None
         if preemptive := self._preemptive_generation:
             # make sure the on_user_turn_completed didn't change some request parameters
             # otherwise invalidate the preemptive generation
@@ -2768,6 +2841,7 @@ async def _user_turn_completed_impl(
                     "preemptive generation invalidated after `on_user_turn_completed` because "
                     "the transcript, chat context, tools, or tool choice changed",
                 )
+                discarded_preemptive = preemptive.speech_handle
                 preemptive.speech_handle._cancel()
 
             self._preemptive_generation = None
@@ -2780,6 +2854,8 @@ async def _user_turn_completed_impl(
                 chat_ctx=temp_mutable_chat_ctx,
                 input_details=InputDetails(modality="audio"),
             )
+            # the invalidated preemptive attempt answered this same turn: one agent_turn
+            _continue_discarded_turn(discarded_preemptive, speech_handle)
 
         if self._user_turn_completed_atask != asyncio.current_task():
             # If a new user turn has already started, interrupt this one since it's now outdated
@@ -2995,35 +3071,19 @@ async def _tts_task(
         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
-        ) as current_span:
-            current_span.set_attribute(trace_types.ATTR_AGENT_TURN_ID, speech_handle._generation_id)
-            if parent_id := speech_handle._parent_generation_id:
-                current_span.set_attribute(trace_types.ATTR_AGENT_PARENT_TURN_ID, parent_id)
-            # an agent turn is the convention's `invoke_agent`: the framework running the
-            # agent in-process, with the inference and tool spans nested underneath
-            gen_ai_telemetry.set_agent_attributes(
-                current_span,
-                operation=trace_types.GenAIOperationName.INVOKE_AGENT,
-                agent_name=self._agent.label,
+        with _agent_turn(
+            speech_handle,
+            root_context=self._session._root_span_context,
+            agent_label=self._agent.label,
+        ):
+            await self._tts_task_impl(
+                speech_handle=speech_handle,
+                text=text,
+                audio=audio,
+                add_to_chat_ctx=add_to_chat_ctx,
+                model_settings=model_settings,
+                _previous_user_metrics=_previous_user_metrics,
             )
-            speech_handle._agent_turn_context = otel_context.get_current()
-            turn_started_at = time.perf_counter()
-
-            try:
-                await self._tts_task_impl(
-                    speech_handle=speech_handle,
-                    text=text,
-                    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(
-                    time.perf_counter() - turn_started_at, agent_name=self._agent.label
-                )
 
     async def _tts_task_impl(
         self,
@@ -3299,36 +3359,20 @@ async def _pipeline_reply_task(
         instructions: str | Instructions | None = None,
         _previous_user_metrics: llm.MetricsReport | None = None,
     ) -> None:
-        with tracer.start_as_current_span(
-            "agent_turn", context=self._session._root_span_context
-        ) as current_span:
-            current_span.set_attribute(trace_types.ATTR_AGENT_TURN_ID, speech_handle._generation_id)
-            if parent_id := speech_handle._parent_generation_id:
-                current_span.set_attribute(trace_types.ATTR_AGENT_PARENT_TURN_ID, parent_id)
-            # an agent turn is the convention's `invoke_agent`: the framework running the
-            # agent in-process, with the inference and tool spans nested underneath
-            gen_ai_telemetry.set_agent_attributes(
-                current_span,
-                operation=trace_types.GenAIOperationName.INVOKE_AGENT,
-                agent_name=self._agent.label,
+        with _agent_turn(
+            speech_handle,
+            root_context=self._session._root_span_context,
+            agent_label=self._agent.label,
+        ):
+            await self._pipeline_reply_task_impl(
+                speech_handle=speech_handle,
+                chat_ctx=chat_ctx,
+                tools=tools,
+                model_settings=model_settings,
+                new_message=new_message,
+                instructions=instructions,
+                _previous_user_metrics=_previous_user_metrics,
             )
-            speech_handle._agent_turn_context = otel_context.get_current()
-            turn_started_at = time.perf_counter()
-
-            try:
-                await self._pipeline_reply_task_impl(
-                    speech_handle=speech_handle,
-                    chat_ctx=chat_ctx,
-                    tools=tools,
-                    model_settings=model_settings,
-                    new_message=new_message,
-                    instructions=instructions,
-                    _previous_user_metrics=_previous_user_metrics,
-                )
-            finally:
-                otel_metrics.record_invoke_agent_duration(
-                    time.perf_counter() - turn_started_at, agent_name=self._agent.label
-                )
 
     async def _pipeline_reply_task_impl(
         self,
@@ -4091,22 +4135,11 @@ async def _realtime_generation_task(
         model_settings: ModelSettings,
         instructions: str | None = None,
     ) -> None:
-        with tracer.start_as_current_span(
-            "agent_turn", context=self._session._root_span_context
-        ) as current_span:
-            current_span.set_attribute(trace_types.ATTR_AGENT_TURN_ID, speech_handle._generation_id)
-            if parent_id := speech_handle._parent_generation_id:
-                current_span.set_attribute(trace_types.ATTR_AGENT_PARENT_TURN_ID, parent_id)
-            # an agent turn is the convention's `invoke_agent`: the framework running the
-            # agent in-process, with the inference and tool spans nested underneath
-            gen_ai_telemetry.set_agent_attributes(
-                current_span,
-                operation=trace_types.GenAIOperationName.INVOKE_AGENT,
-                agent_name=self._agent.label,
-            )
-            speech_handle._agent_turn_context = otel_context.get_current()
-            turn_started_at = time.perf_counter()
-
+        with _agent_turn(
+            speech_handle,
+            root_context=self._session._root_span_context,
+            agent_label=self._agent.label,
+        ):
             inference_span = tracer.start_span("realtime_inference")
             try:
                 await self._realtime_generation_task_impl(
@@ -4118,9 +4151,6 @@ async def _realtime_generation_task(
                 )
             finally:
                 inference_span.end()
-                otel_metrics.record_invoke_agent_duration(
-                    time.perf_counter() - turn_started_at, agent_name=self._agent.label
-                )
 
     async def _realtime_generation_task_impl(
         self,

livekit-agents/livekit/agents/voice/speech_handle.py

@@ -7,10 +7,11 @@
 from dataclasses import dataclass
 from typing import Any, Literal
 
-from opentelemetry import context as otel_context
+from opentelemetry import context as otel_context, trace
 
 from .. import llm, utils
 from ..log import logger
+from ..telemetry import trace_types
 
 INTERRUPTION_TIMEOUT = 5.0  # seconds
 
@@ -51,7 +52,12 @@ def __init__(
         self._tasks: list[asyncio.Task] = []
         self._chat_items: list[llm.ChatItem] = []
         self._num_steps = 1
+        # one agent_turn span for the whole speech, however many generations (LLM steps) it
+        # takes; opened by the first reply task, ended with the speech in _mark_done
+        self._agent_turn_span: trace.Span | None = None
         self._agent_turn_context: otel_context.Context | None = None
+        self._agent_turn_started_at: float | None = None
+        self._agent_turn_agent_name: str | None = None
         self._scheduled_at: float | None = None
         self._authorized_at: float | None = None
         self._interrupt_source: str | None = None  # set by whoever interrupts, for the trace
@@ -335,6 +341,7 @@ def _mark_done(self, error: BaseException | None = None) -> None:
             if error is not None:
                 self._error = error
             self._done_fut.set_result(None)
+        self._end_agent_turn(error)
 
         if self._generations:
             self._mark_generation_done()
@@ -343,6 +350,54 @@ def _mark_done(self, error: BaseException | None = None) -> None:
             self._interrupt_timeout_handle.cancel()
             self._interrupt_timeout_handle = None
 
+    def _take_agent_turn(self) -> tuple[trace.Span, float | None, str | None] | None:
+        """Detach this speech's open ``agent_turn`` so a successor can continue it.
+
+        Used when a preemptive generation is discarded for another speech answering the same
+        user turn: the wasted generation stays visible under the one turn instead of becoming
+        a turn of its own. After this the speech ends without touching the span."""
+        span = self._agent_turn_span
+        if span is None:
+            return None
+        carry = (span, self._agent_turn_started_at, self._agent_turn_agent_name)
+        self._agent_turn_span = None
+        self._agent_turn_context = None
+        self._agent_turn_started_at = None
+        self._agent_turn_agent_name = None
+        return carry
+
+    def _continue_agent_turn(
+        self, carry: tuple[trace.Span, float | None, str | None], *, discarded: SpeechHandle
+    ) -> None:
+        """Adopt the ``agent_turn`` taken from ``discarded`` (see ``_take_agent_turn``)."""
+        span, started_at, agent_name = carry
+        if not span.is_recording():
+            return
+        span.add_event(
+            "preemptive_generation_discarded", {trace_types.ATTR_SPEECH_ID: discarded.id}
+        )
+        span.set_attribute(trace_types.ATTR_SPEECH_ID, self.id)
+        self._agent_turn_span = span
+        self._agent_turn_context = trace.set_span_in_context(span)
+        self._agent_turn_started_at = started_at
+        self._agent_turn_agent_name = agent_name
+
+    def _end_agent_turn(self, error: BaseException | None) -> None:
+        """Close the speech's ``agent_turn`` span: the speech is done, whatever step it was on."""
+        span, self._agent_turn_span = self._agent_turn_span, None
+        if span is None or not span.is_recording():
+            return
+        from ..telemetry import otel_metrics, utils as trace_utils
+
+        if isinstance(error, Exception):
+            trace_utils.record_exception(span, error)
+        if self._agent_turn_started_at is not None and self._agent_turn_agent_name is not None:
+            otel_metrics.record_invoke_agent_duration(
+                time.perf_counter() - self._agent_turn_started_at,
+                agent_name=self._agent_turn_agent_name,
+            )
+        span.end()
+
     def _mark_scheduled(self) -> None:
         if self._scheduled_at is None:
             self._scheduled_at = time.perf_counter()