fix: fallback adapters report the active instance's model/provider

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

@@ -10,7 +10,7 @@
 from .._exceptions import APIConnectionError, APIError
 from ..log import logger
 from ..types import DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, APIConnectOptions, NotGivenOr
-from .chat_context import ChatContext
+from .chat_context import ChatContext, MetricsMetadata
 from .llm import LLM, ChatChunk, LLMStream
 from .tool_context import Tool, ToolChoice
 
@@ -77,6 +77,9 @@ def __init__(
             _LLMStatus(available=True, recovering_task=None) for _ in self._llm_instances
         ]
 
+        # the instance that most recently served a request; used to label metrics & traces
+        self._active_instance: LLM = self._llm_instances[0]
+
         for llm_instance in self._llm_instances:
             llm_instance.on("metrics_collected", self._on_metrics_collected)
 
@@ -88,6 +91,11 @@ def model(self) -> str:
     def provider(self) -> str:
         return "livekit"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata of the instance that most recently served a request (the primary before any traffic)."""  # noqa: E501
+        return self._active_instance.metrics_metadata
+
     def chat(
         self,
         *,
@@ -194,6 +202,7 @@ async def _try_generate(
                     if should_set_current:
                         should_set_current = False
                         self._current_stream = stream
+                        self._fallback_adapter._active_instance = llm
                     yield chunk
 
         except asyncio.TimeoutError:

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

@@ -28,7 +28,7 @@
     NotGivenOr,
 )
 from ..utils import aio
-from .chat_context import ChatContext, ChatRole
+from .chat_context import ChatContext, ChatRole, MetricsMetadata
 from .tool_context import Tool, ToolChoice
 
 
@@ -132,6 +132,11 @@ def provider(self) -> str:
         """
         return "unknown"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata used to label turn metrics emitted for this LLM instance."""
+        return {"model_name": self.model, "model_provider": self.provider}
+
     @abstractmethod
     def chat(
         self,

livekit-agents/livekit/agents/llm/realtime.py

@@ -15,7 +15,7 @@
 from ..log import logger
 from ..types import NOT_GIVEN, NotGivenOr
 from ..utils import is_given
-from .chat_context import ChatContext, ChatItem, FunctionCall
+from .chat_context import ChatContext, ChatItem, FunctionCall, MetricsMetadata
 from .tool_context import Tool, ToolChoice, ToolContext
 
 
@@ -107,6 +107,11 @@ def model(self) -> str:
     def provider(self) -> str:
         return "unknown"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata used to label turn metrics emitted for this realtime model."""
+        return {"model_name": self.model, "model_provider": self.provider}
+
     @property
     def capabilities(self) -> RealtimeCapabilities:
         return self._capabilities

livekit-agents/livekit/agents/llm/realtime_fallback_adapter.py

@@ -13,7 +13,7 @@
 from ..log import logger
 from ..types import NOT_GIVEN, NotGivenOr
 from ..utils import aio, is_given
-from .chat_context import ChatContext
+from .chat_context import ChatContext, MetricsMetadata
 from .realtime import (
     EventTypes,
     GenerationCreatedEvent,
@@ -120,6 +120,9 @@ def __init__(
         self._regenerate_on_swap = regenerate_on_swap
         self._sessions: weakref.WeakSet[_FallbackRealtimeSession] = weakref.WeakSet()
 
+        # the model currently serving sessions; used to label metrics & traces
+        self._active_instance: RealtimeModel = models[0]
+
     @property
     def model(self) -> str:
         return "RealtimeModelFallbackAdapter"
@@ -128,6 +131,11 @@ def model(self) -> str:
     def provider(self) -> str:
         return "livekit"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata of the model currently serving sessions (the primary until a swap)."""
+        return self._active_instance.metrics_metadata
+
     def session(self, *, turn_detection_disabled: bool = False) -> _FallbackRealtimeSession:
         sess = _FallbackRealtimeSession(self, turn_detection_disabled=turn_detection_disabled)
         self._sessions.add(sess)
@@ -192,6 +200,8 @@ def _forward(ev: object) -> None:
         self._active = adapter._models[0].session(
             turn_detection_disabled=self._turn_detection_disabled
         )
+        # a fresh session always starts on the primary, even after an earlier failover
+        adapter._active_instance = adapter._models[0]
         self._bind(self._active)
 
     def _bind(self, child: RealtimeSession) -> None:
@@ -304,6 +314,7 @@ async def _bring_up(index: int) -> Exception | None:
                     )
                     if is_given(self._tool_choice):
                         self._active.update_options(tool_choice=self._tool_choice)
+                    self._adapter._active_instance = self._adapter._models[index]
                     return None
                 except Exception as e:
                     logger.exception("failed to start realtime model on swap, trying next")

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

@@ -20,6 +20,7 @@
 from .stt import STT, RecognizeStream, SpeechEvent, SpeechEventType, STTCapabilities
 
 if TYPE_CHECKING:
+    from ..llm.chat_context import MetricsMetadata
     from ..voice.events import ConversationItemAddedEvent
 
 # don't retry when using the fallback adapter
@@ -106,6 +107,9 @@ def __init__(
             for _ in self._stt_instances
         ]
 
+        # the instance that most recently served a request; used to label metrics & traces
+        self._active_instance: STT = self._stt_instances[0]
+
         for stt_instance in self._stt_instances:
             stt_instance.on("metrics_collected", self._on_metrics_collected)
         self._recognize_metrics_needed = False  # don't emit metrics via fallback adapter
@@ -118,6 +122,11 @@ def model(self) -> str:
     def provider(self) -> str:
         return "livekit"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata of the instance that most recently served a request (the primary before any traffic)."""  # noqa: E501
+        return self._active_instance.metrics_metadata
+
     def _update_session_keyterms(self, keyterms: list[str]) -> None:
         # forward to every underlying STT; unsupported ones warn-and-skip internally
         for stt_instance in self._stt_instances:
@@ -242,13 +251,15 @@ async def _recognize_impl(
             stt_status = self._status[i]
             if stt_status.available or all_failed:
                 try:
-                    return await self._try_recognize(
+                    event = await self._try_recognize(
                         stt=stt,
                         buffer=buffer,
                         language=language,
                         conn_options=conn_options,
                         recovering=False,
                     )
+                    self._active_instance = stt
+                    return event
                 except Exception:  # exceptions already logged inside _try_recognize
                     if stt_status.available:
                         stt_status.available = False
@@ -375,8 +386,12 @@ async def _forward_input_task() -> None:
                         forward_input_task = asyncio.create_task(_forward_input_task())
 
                     try:
+                        should_set_active = True
                         async with main_stream:
                             async for ev in main_stream:
+                                if should_set_active:
+                                    should_set_active = False
+                                    self._fallback_adapter._active_instance = stt
                                 self._event_ch.send_nowait(ev)
 
                     except asyncio.TimeoutError:

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

@@ -29,6 +29,7 @@
 from ..utils.audio import calculate_audio_duration
 
 if TYPE_CHECKING:
+    from ..llm.chat_context import MetricsMetadata
     from ..voice.events import ConversationItemAddedEvent
 
 
@@ -190,6 +191,11 @@ def provider(self) -> str:
         """
         return "unknown"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata used to label turn metrics emitted for this STT instance."""
+        return {"model_name": self.model, "model_provider": self.provider}
+
     @property
     def capabilities(self) -> STTCapabilities:
         return self._capabilities

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

@@ -5,7 +5,7 @@
 import time
 from collections.abc import AsyncGenerator, AsyncIterable
 from dataclasses import dataclass
-from typing import Any, ClassVar, Literal
+from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
 from livekit import rtc
 
@@ -24,6 +24,9 @@
     TTSCapabilities,
 )
 
+if TYPE_CHECKING:
+    from ..llm.chat_context import MetricsMetadata
+
 # don't retry when using the fallback adapter
 DEFAULT_FALLBACK_API_CONNECT_OPTIONS = APIConnectOptions(
     max_retry=0, timeout=DEFAULT_API_CONNECT_OPTIONS.timeout
@@ -93,6 +96,9 @@ def __init__(
         self._tts_instances = tts
         self._max_retry_per_tts = max_retry_per_tts
 
+        # the instance that most recently served a request; used to label metrics & traces
+        self._active_instance: TTS = self._tts_instances[0]
+
         self._status: list[_TTSStatus] = []
         for t in tts:
             needs_resampling = sample_rate != t.sample_rate
@@ -113,6 +119,11 @@ def model(self) -> str:
     def provider(self) -> str:
         return "livekit"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata of the instance that most recently served a request (the primary before any traffic)."""  # noqa: E501
+        return self._active_instance.metrics_metadata
+
     def synthesize(
         self, text: str, *, conn_options: APIConnectOptions = DEFAULT_FALLBACK_API_CONNECT_OPTIONS
     ) -> FallbackChunkedStream:
@@ -164,7 +175,11 @@ async def _try_synthesize(
                     retry_interval=self._conn_options.retry_interval,
                 ),
             ) as stream:
+                should_set_active = not recovering
                 async for audio in stream:
+                    if should_set_active:
+                        should_set_active = False
+                        self._fallback_adapter._active_instance = tts
                     yield audio
 
         except Exception as e:
@@ -317,8 +332,12 @@ def _capture_started_time() -> None:
 
         try:
             async with stream:
+                should_set_active = not recovering
                 async for audio in stream:
                     _capture_started_time()
+                    if should_set_active:
+                        should_set_active = False
+                        self._fallback_adapter._active_instance = tts
                     yield audio
         except Exception as e:
             if recovering:

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

@@ -29,6 +29,7 @@
 from ..utils import aio, audio, codecs, log_exceptions, shortuuid
 
 if TYPE_CHECKING:
+    from ..llm.chat_context import MetricsMetadata
     from ..voice.agent_session import SpeechSteeringOptions
     from ..voice.io import TimedString
 
@@ -204,6 +205,11 @@ def provider(self) -> str:
         """
         return "unknown"
 
+    @property
+    def metrics_metadata(self) -> MetricsMetadata:
+        """Metadata used to label turn metrics emitted for this TTS instance."""
+        return {"model_name": self.model, "model_provider": self.provider}
+
     @property
     def capabilities(self) -> TTSCapabilities:
         return self._capabilities

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

@@ -3377,15 +3377,9 @@ async def _next_segment() -> _SpeechSegment | None:
         assistant_metrics: llm.MetricsReport = {}
 
         if self.llm:
-            assistant_metrics["llm_metadata"] = {
-                "model_name": self.llm.model,
-                "model_provider": self.llm.provider,
-            }
+            assistant_metrics["llm_metadata"] = self.llm.metrics_metadata
         if self.tts:
-            assistant_metrics["tts_metadata"] = {
-                "model_name": self.tts.model,
-                "model_provider": self.tts.provider,
-            }
+            assistant_metrics["tts_metadata"] = self.tts.metrics_metadata
 
         if llm_gen_data.ttft is not None:
             assistant_metrics["llm_node_ttft"] = llm_gen_data.ttft
@@ -4454,10 +4448,7 @@ def _fallback_to_vad_interruption(
     def _init_metrics_from_end_of_turn(self, info: _EndOfTurnInfo) -> llm.MetricsReport:
         metrics_report: llm.MetricsReport = {}
         if self.stt:
-            metrics_report["stt_metadata"] = {
-                "model_name": self.stt.model,
-                "model_provider": self.stt.provider,
-            }
+            metrics_report["stt_metadata"] = self.stt.metrics_metadata
         if info.metrics.started_speaking_at is not None:
             metrics_report["started_speaking_at"] = info.metrics.started_speaking_at