feat(telemetry): interruption detail, handoff span, fallback events, text input

livekit-agents/livekit/agents/llm/fallback_adapter.py

@@ -7,8 +7,11 @@
 from dataclasses import dataclass
 from typing import Any, ClassVar, Literal
 
+from opentelemetry import trace
+
 from .._exceptions import APIConnectionError, APIError
 from ..log import logger
+from ..telemetry import trace_types
 from ..types import DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, APIConnectOptions, NotGivenOr
 from .chat_context import ChatContext, MetricsMetadata
 from .llm import LLM, ChatChunk, LLMStream
@@ -83,13 +86,27 @@ def __init__(
         for llm_instance in self._llm_instances:
             llm_instance.on("metrics_collected", self._on_metrics_collected)
 
+    def _next_instance(self) -> LLM:
+        """The instance the next request goes to first: the first one marked available, or
+        the primary once all are down (they are then all retried, primary first). A failed
+        instance's recovery task flips it back to available, so a recovered primary is
+        reported again before it has served."""
+        for instance, status in zip(self._llm_instances, self._status, strict=True):
+            if status.available:
+                return instance
+        return self._llm_instances[0]
+
     @property
     def model(self) -> str:
-        return "FallbackAdapter"
+        """The model of the instance that serves next (see :meth:`_next_instance`). Spans and
+        metrics read this, so a failover shows the model expected to answer rather than the
+        adapter; the instance that actually served is stamped per request by the stream."""
+        return self._next_instance().model
 
     @property
     def provider(self) -> str:
-        return "livekit"
+        """The provider of the instance that serves next (see :attr:`model`)."""
+        return self._next_instance().provider
 
     @property
     def metrics_metadata(self) -> MetricsMetadata:
@@ -137,6 +154,21 @@ def _on_metrics_collected(self, *args: Any, **kwargs: Any) -> None:
         self.emit("metrics_collected", *args, **kwargs)
 
 
+def _provider_attr(llm: LLM) -> dict[str, str]:
+    normalized = trace_types.gen_ai_provider_name(llm.provider)
+    return {trace_types.ATTR_GEN_AI_PROVIDER_NAME: normalized} if normalized else {}
+
+
+def _fallback_attrs(llm: LLM, index: int) -> dict[str, Any]:
+    """The instance that served: its label, position, model and provider."""
+    return {
+        trace_types.ATTR_FALLBACK_LABEL: llm.label,
+        trace_types.ATTR_FALLBACK_INDEX: index,
+        trace_types.ATTR_GEN_AI_REQUEST_MODEL: llm.model,
+        **_provider_attr(llm),
+    }
+
+
 class FallbackLLMStream(LLMStream):
     _llm_request_span_name: ClassVar[str] = "llm_fallback_adapter"
 
@@ -153,6 +185,8 @@ def __init__(
     ) -> None:
         super().__init__(llm, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
         self._fallback_adapter = llm
+        # the span this request was made under (llm_node): told which instance served
+        self._caller_span = trace.get_current_span()
         self._parallel_tool_calls = parallel_tool_calls
         self._tool_choice = tool_choice
         self._extra_kwargs = extra_kwargs
@@ -286,6 +320,18 @@ async def _run(self) -> None:
 
                         self._event_ch.send_nowait(result)
 
+                    served = _fallback_attrs(llm, i)
+                    trace.get_current_span().set_attributes(served)
+                    # request-side attributes named the instance expected to serve; the
+                    # response side names the one that did (from `llm`, not the adapter:
+                    # concurrent requests may be served by different instances)
+                    response_attrs = {
+                        trace_types.ATTR_GEN_AI_RESPONSE_MODEL: llm.model,
+                        **_provider_attr(llm),
+                    }
+                    if self._llm_request_span is not None:
+                        self._llm_request_span.set_attributes(response_attrs)
+                    self._caller_span.set_attributes(response_attrs)
                     return
                 except Exception:  # exceptions already logged inside _try_generate
                     if llm_status.available:

livekit-agents/livekit/agents/stt/fallback_adapter.py

@@ -120,13 +120,27 @@ def __init__(
             stt_instance.on("metrics_collected", self._on_metrics_collected)
         self._recognize_metrics_needed = False  # don't emit metrics via fallback adapter
 
+    def _next_instance(self) -> STT:
+        """The instance the next request goes to first: the first one marked available, or
+        the primary once all are down (they are then all retried, primary first). A failed
+        instance's recovery task flips it back to available, so a recovered primary is
+        reported again before it has served."""
+        for instance, status in zip(self._stt_instances, self._status, strict=True):
+            if status.available:
+                return instance
+        return self._stt_instances[0]
+
     @property
     def model(self) -> str:
-        return "FallbackAdapter"
+        """The model of the instance that serves next (see :meth:`_next_instance`). Spans and
+        metrics read this, so a failover shows the model expected to answer rather than the
+        adapter; the instance that actually served is stamped per request by the stream."""
+        return self._next_instance().model
 
     @property
     def provider(self) -> str:
-        return "livekit"
+        """The provider of the instance that serves next (see :attr:`model`)."""
+        return self._next_instance().provider
 
     @property
     def metrics_metadata(self) -> MetricsMetadata:

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

@@ -58,6 +58,13 @@
 ATTR_JOB_DISPATCH_LATENCY = "lk.job.dispatch_latency"
 """Seconds from the availability request to the entrypoint running: the whole chain."""
 
+# keyterm detection (keyterm_detection span): counts only, the terms themselves are the
+# customer's vocabulary and travel as lk.pii.keyterms in the session report
+ATTR_KEYTERMS_COUNT = "lk.keyterms.count"
+"""Keyterms in effect after the pass (static + confirmed)."""
+ATTR_KEYTERMS_ADDED = "lk.keyterms.added"
+ATTR_KEYTERMS_REMOVED = "lk.keyterms.removed"
+
 # room connect / room io
 ATTR_ROOM_AUTO_SUBSCRIBE = "lk.room.auto_subscribe"
 ATTR_ROOM_E2EE = "lk.room.e2ee"
@@ -410,6 +417,21 @@ def gen_ai_provider_name(provider: str | None) -> str | None:
 ATTR_AMD_DELAY = "lk.amd.delay"
 ATTR_AMD_TRANSCRIPT = "lk.pii.amd.transcript"
 
+# Interruptions (agent_turn)
+ATTR_INTERRUPTION_SOURCE = "lk.interruption.source"
+"""What interrupted the speech: ``audio_activity`` (barge-in), ``user_turn`` (a committed
+turn preempting the reply), or ``programmatic`` (session.interrupt(), a tool, teardown)."""
+ATTR_PLAYOUT_POSITION = "lk.playout.position"
+"""Seconds of audio that had actually played when the speech was interrupted."""
+
+# Agent handoff (update_agent span)
+ATTR_PREVIOUS_AGENT_LABEL = "lk.previous_agent_label"
+
+# Fallback adapters (the attempt span)
+ATTR_FALLBACK_LABEL = "lk.fallback.label"
+"""Label of the provider that served the request."""
+ATTR_FALLBACK_INDEX = "lk.fallback.index"
+
 # Adaptive Interruption attributes
 ATTR_IS_INTERRUPTION = "lk.is_interruption"
 ATTR_INTERRUPTION_PROBABILITY = "lk.interruption.probability"

livekit-agents/livekit/agents/tts/fallback_adapter.py

@@ -7,11 +7,14 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
+from opentelemetry import trace
+
 from livekit import rtc
 
 from .. import utils
 from .._exceptions import APIConnectionError
 from ..log import logger
+from ..telemetry import trace_types
 from ..types import DEFAULT_API_CONNECT_OPTIONS, USERDATA_TIMED_TRANSCRIPT, APIConnectOptions
 from ..utils import aio
 from .stream_adapter import StreamAdapter
@@ -121,13 +124,27 @@ def __init__(
 
             t.on("metrics_collected", self._on_metrics_collected)
 
+    def _next_instance(self) -> TTS:
+        """The instance the next request goes to first: the first one marked available, or
+        the primary once all are down (they are then all retried, primary first). A failed
+        instance's recovery task flips it back to available, so a recovered primary is
+        reported again before it has served."""
+        for instance, status in zip(self._tts_instances, self._status, strict=True):
+            if status.available:
+                return instance
+        return self._tts_instances[0]
+
     @property
     def model(self) -> str:
-        return "FallbackAdapter"
+        """The model of the instance that serves next (see :meth:`_next_instance`). Spans and
+        metrics read this, so a failover shows the model expected to answer rather than the
+        adapter; the instance that actually served is stamped per request by the stream."""
+        return self._next_instance().model
 
     @property
     def provider(self) -> str:
-        return "livekit"
+        """The provider of the instance that serves next (see :attr:`model`)."""
+        return self._next_instance().provider
 
     @property
     def metrics_metadata(self) -> MetricsMetadata:
@@ -169,6 +186,33 @@ async def aclose(self) -> None:
             t.off("metrics_collected", self._on_metrics_collected)
 
 
+def _fallback_attrs(tts: TTS, index: int) -> dict[str, Any]:
+    """The instance that served: its label, position, model and provider."""
+    attrs: dict[str, Any] = {
+        trace_types.ATTR_FALLBACK_LABEL: tts.label,
+        trace_types.ATTR_FALLBACK_INDEX: index,
+        trace_types.ATTR_GEN_AI_REQUEST_MODEL: tts.model,
+    }
+    if (normalized := trace_types.gen_ai_provider_name(tts.provider)) is not None:
+        attrs[trace_types.ATTR_GEN_AI_PROVIDER_NAME] = normalized
+    return attrs
+
+
+def _record_fallback_served(tts: TTS, index: int, *spans: trace.Span | None) -> None:
+    """The instance that served: on the current (attempt) span, and as the response side of
+    ``spans`` (the adapter's request span and the caller's, tts_node). From ``tts``, not the
+    adapter: concurrent requests may be served by different instances."""
+    attrs = _fallback_attrs(tts, index)
+    trace.get_current_span().set_attributes(attrs)
+    response_attrs = {
+        trace_types.ATTR_GEN_AI_RESPONSE_MODEL: tts.model,
+        **{k: v for k, v in attrs.items() if k == trace_types.ATTR_GEN_AI_PROVIDER_NAME},
+    }
+    for span in spans:
+        if span is not None:
+            span.set_attributes(response_attrs)
+
+
 class FallbackChunkedStream(ChunkedStream):
     _tts_request_span_name: ClassVar[str] = "tts_fallback_adapter"
 
@@ -177,6 +221,8 @@ def __init__(
     ) -> None:
         super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
         self._fallback_adapter = tts
+        # the span this request was made under (tts_node); see _record_fallback_served
+        self._caller_span = trace.get_current_span()
 
     async def _metrics_monitor_task(self, event_aiter: AsyncIterable[SynthesizedAudio]) -> None:
         async for _ in event_aiter:
@@ -282,6 +328,7 @@ async def _run(self, output_emitter: AudioEmitter) -> None:
                         for rf in resampler.flush():
                             output_emitter.push(rf.data.tobytes())
 
+                    _record_fallback_served(tts, i, self._tts_request_span, self._caller_span)
                     return
                 except Exception:  # exceptions already logged inside _try_synthesize
                     if tts_status.available:
@@ -295,6 +342,8 @@ async def _run(self, output_emitter: AudioEmitter) -> None:
                         logger.warning(
                             f"{tts.label} already synthesized of audio, ignoring fallback"
                         )
+                        # the caller heard this instance's audio: it served, partially
+                        _record_fallback_served(tts, i, self._tts_request_span, self._caller_span)
                         return
 
             self._try_recovery(tts)
@@ -310,6 +359,7 @@ class FallbackSynthesizeStream(SynthesizeStream):
     def __init__(self, *, tts: FallbackAdapter, conn_options: APIConnectOptions):
         super().__init__(tts=tts, conn_options=conn_options)
         self._fallback_adapter = tts
+        self._caller_span = trace.get_current_span()
         self._pushed_tokens: list[str] = []
 
     async def _metrics_monitor_task(self, event_aiter: AsyncIterable[SynthesizedAudio]) -> None:
@@ -474,6 +524,7 @@ async def _forward_input_task() -> None:
                             else:
                                 output_emitter.push(synthesized_audio.frame.data.tobytes())
 
+                        _record_fallback_served(tts, i, self._tts_request_span, self._caller_span)
                         return
                     except Exception:
                         if tts_status.available:
@@ -487,6 +538,10 @@ async def _forward_input_task() -> None:
                             logger.warning(
                                 f"{tts.label} already synthesized of audio, ignoring the current segment for the tts fallback"  # noqa: E501
                             )
+                            # the caller heard this instance's audio: it served, partially
+                            _record_fallback_served(
+                                tts, i, self._tts_request_span, self._caller_span
+                            )
                             return
 
                 pending_recovery.append(tts)