feat(voice): auto-disable realtime server-side turn detection

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

@@ -70,6 +70,9 @@ class RealtimeCapabilities:
     """Whether the model can produce audio output directly"""
     manual_function_calls: bool
     """Whether function call items already in the chat context can be resumed"""
+    can_disable_turn_detection: bool = False
+    """Whether server-side turn detection can be disabled for a session so the client drives
+    turn-taking. Set by plugins that implement ``session(turn_detection_disabled=True)``."""
     mutable_chat_context: bool = False
     """Whether the chat context can be updated mid-session"""
     mutable_instructions: bool = False
@@ -113,7 +116,12 @@ def label(self) -> str:
         return self._label
 
     @abstractmethod
-    def session(self) -> RealtimeSession: ...
+    def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
+        """Create a new session, optionally with server-side turn detection disabled.
+
+        ``turn_detection_disabled`` is honored only by plugins reporting
+        ``can_disable_turn_detection``; the model itself is left unchanged and reusable."""
+        ...
 
     @abstractmethod
     async def aclose(self) -> None: ...

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

@@ -52,6 +52,7 @@ class RealtimeAvailabilityChangedEvent:
     "mutable_tools",
     "per_response_tool_choice",
     "supports_say",
+    "can_disable_turn_detection",
 )
 
 # child events re-emitted on the wrapper
@@ -127,8 +128,8 @@ def model(self) -> str:
     def provider(self) -> str:
         return "livekit"
 
-    def session(self) -> _FallbackRealtimeSession:
-        sess = _FallbackRealtimeSession(self)
+    def session(self, *, turn_detection_disabled: bool = False) -> _FallbackRealtimeSession:
+        sess = _FallbackRealtimeSession(self, turn_detection_disabled=turn_detection_disabled)
         self._sessions.add(sess)
         return sess
 
@@ -150,9 +151,13 @@ async def aclose(self) -> None:
 class _FallbackRealtimeSession(RealtimeSession[Literal["realtime_availability_changed"]]):
     """Bound once by AgentActivity; swaps the inner child session internally."""
 
-    def __init__(self, adapter: RealtimeModelFallbackAdapter) -> None:
+    def __init__(
+        self, adapter: RealtimeModelFallbackAdapter, *, turn_detection_disabled: bool = False
+    ) -> None:
         super().__init__(adapter)
         self._adapter = adapter
+        # applied to every underlying session, including those brought up on a swap
+        self._turn_detection_disabled = turn_detection_disabled
 
         # session state replayed onto a new child on swap
         self._instructions: NotGivenOr[str] = NOT_GIVEN
@@ -184,7 +189,9 @@ def _forward(ev: object) -> None:
         self._swapping = False
 
         self._active_index = 0
-        self._active = adapter._models[0].session()
+        self._active = adapter._models[0].session(
+            turn_detection_disabled=self._turn_detection_disabled
+        )
         self._bind(self._active)
 
     def _bind(self, child: RealtimeSession) -> None:
@@ -287,7 +294,9 @@ async def _swap(self, target_index: int, was_speaking: bool) -> None:
             # return the error so the caller can try the next model
             async def _bring_up(index: int) -> Exception | None:
                 try:
-                    self._active = self._adapter._models[index].session()
+                    self._active = self._adapter._models[index].session(
+                        turn_detection_disabled=self._turn_detection_disabled
+                    )
                     self._active_index = index
                     self._bind(self._active)
                     await self._active._update_session(

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

@@ -221,11 +221,19 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
         self._on_enter_task: asyncio.Task | None = None
         self._on_exit_task: asyncio.Task | None = None
 
+        # session-scoped truth read by every server-side turn-detection check below
+        self._rt_turn_detection_enabled = self._resolve_rt_turn_detection_enabled()
         if (
             isinstance(self.llm, llm.RealtimeModel)
+            and not self._rt_turn_detection_enabled
             and self.llm.capabilities.turn_detection
-            and not self.allow_interruptions
         ):
+            logger.info(
+                "client-side turn-taking is configured, disabling realtime server-side "
+                "turn detection."
+            )
+
+        if self._rt_turn_detection_enabled and not self.allow_interruptions:
             raise ValueError(
                 "the RealtimeModel uses a server-side turn detection, "
                 "allow_interruptions cannot be False, disable turn_detection in "
@@ -263,6 +271,51 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
         # model to auto-generate a tool reply (auto_tool_reply_generation=True).
         self._pending_auto_tool_reply_fut: asyncio.Future[None] | None = None
 
+    def _resolve_rt_turn_detection_enabled(self) -> bool:
+        """Whether a realtime model's server-side turn detection is on for this session.
+
+        Off when the model can hand turn-taking to the client and the user configured
+        client-side turn-taking; otherwise the model's own setting stands.
+        """
+        if not isinstance(self.llm, llm.RealtimeModel):
+            return False
+
+        # a model that without turn_detection
+        # or can't hand turn-taking to the client keeps its own setting
+        if (
+            not (caps := self.llm.capabilities).turn_detection
+            or not caps.can_disable_turn_detection
+        ):
+            return caps.turn_detection
+
+        # resolve client-side turn detection (agent overrides session)
+        if is_given(self._agent.turn_detection):
+            client_td: TurnDetectionMode | None = self._agent.turn_detection
+            client_td_explicit = True
+        elif self._session._turn_detection_explicit:
+            client_td = self._session.turn_detection
+            client_td_explicit = True
+        else:
+            client_td = None
+            client_td_explicit = False
+
+        if client_td_explicit and client_td == "realtime_llm":
+            return True  # user wants the model to keep doing turn-taking
+        if client_td_explicit and client_td == "manual":
+            return False  # client commits turns manually, no VAD needed
+
+        # disable server-side turn detection if the user has a client-side turn detector or interruption detection
+        if self.vad is not None and (
+            (
+                client_td_explicit
+                and (isinstance(client_td, _StreamingTurnDetector) or client_td == "vad")
+            )
+            or is_given(self._agent.interruption_detection)
+            or is_given(self._session.interruption_detection)
+        ):
+            return False
+        return True
+
     def _validate_turn_detection(
         self, turn_detection: TurnDetectionMode | None
     ) -> TurnDetectionMode | None:
@@ -275,7 +328,7 @@ def _validate_turn_detection(
                     )
                     return None
 
-                if isinstance(self.llm, llm.RealtimeModel) and self.llm.capabilities.turn_detection:
+                if self._rt_turn_detection_enabled:
                     logger.warning(
                         "turn_detection is a TurnDetector, but the LLM is a RealtimeModel "
                         "with server-side turn detection enabled, ignoring the turn_detection setting"
@@ -304,7 +357,7 @@ def _validate_turn_detection(
             mode = None
 
         if isinstance(llm_model, llm.RealtimeModel):
-            if mode == "realtime_llm" and not llm_model.capabilities.turn_detection:
+            if mode == "realtime_llm" and not self._rt_turn_detection_enabled:
                 logger.warning(
                     "turn_detection is set to 'realtime_llm', but the LLM is not a RealtimeModel "
                     "or the server-side turn detection is not supported/enabled, "
@@ -319,7 +372,7 @@ def _validate_turn_detection(
                 )
                 mode = None
 
-            elif mode and mode != "realtime_llm" and llm_model.capabilities.turn_detection:
+            elif mode and mode != "realtime_llm" and self._rt_turn_detection_enabled:
                 logger.warning(
                     f"turn_detection is set to '{mode}', but the LLM "
                     "is a RealtimeModel and server-side turn detection enabled, "
@@ -329,7 +382,7 @@ def _validate_turn_detection(
 
             # fallback to VAD if server side turn detection is disabled and user supplied VAD is available
             if (
-                not llm_model.capabilities.turn_detection
+                not self._rt_turn_detection_enabled
                 and vad_model is not None
                 and not self.using_default_vad
                 and mode is None
@@ -544,6 +597,21 @@ def update_options(
             self._rt_session.update_options(tool_choice=self._tool_choice)
 
         if utils.is_given(turn_detection):
+            # a realtime model's server-side turn detection is resolved once at session start;
+            # runtime re-sync isn't supported yet, so warn when an explicit change would flip it.
+            if (
+                isinstance(self.llm, llm.RealtimeModel)
+                and self.llm.capabilities.can_disable_turn_detection
+                and turn_detection is not None
+                and self._resolve_rt_turn_detection_enabled() != self._rt_turn_detection_enabled
+            ):
+                logger.warning(
+                    "changing turn_detection at runtime does not update a realtime model's "
+                    "server-side turn detection (resolved at session start); it stays %s "
+                    "for this session.",
+                    "enabled" if self._rt_turn_detection_enabled else "disabled",
+                )
+
             turn_detection = self._validate_turn_detection(turn_detection)
 
             if (
@@ -807,6 +875,10 @@ async def _detach_reusable_resources(self, new_activity: AgentActivity) -> _Reus
                     self.llm.capabilities.mutable_tools
                     or llm.ToolContext(self.tools) == llm.ToolContext(new_activity.tools)
                 )
+                # only reuse if the new activity resolves server-side turn detection the same way
+                reusable = reusable and (
+                    self._rt_turn_detection_enabled == new_activity._rt_turn_detection_enabled
+                )
 
                 if reusable:
                     # detach: remove event listeners but don't close the session
@@ -912,7 +984,13 @@ async def _start_session(self, *, reuse_resources: _ReusableResources | None = N
                 self._rt_session.interrupt()
                 self._rt_session.clear_audio()
             else:
-                self._rt_session = self.llm.session()
+                # disable only when we resolved it off AND the model can (guards a model whose
+                # turn detection is explicitly pinned off from a spurious disable)
+                turn_detection_disabled = (
+                    not self._rt_turn_detection_enabled
+                    and self.llm.capabilities.can_disable_turn_detection
+                )
+                self._rt_session = self.llm.session(turn_detection_disabled=turn_detection_disabled)
                 logger.debug("created new realtime session for activity, id=%s", self._rt_session)
 
             self._rt_session.on("generation_created", self._on_generation_created)
@@ -987,12 +1065,7 @@ async def _start_session(self, *, reuse_resources: _ReusableResources | None = N
         await self._resume_scheduling_task()
         # skip default vad when llm does not need it
         wired_vad = self.vad
-        if (
-            wired_vad is not None
-            and self.using_default_vad
-            and isinstance(self.llm, llm.RealtimeModel)
-            and self.llm.capabilities.turn_detection
-        ):
+        if wired_vad is not None and self.using_default_vad and self._rt_turn_detection_enabled:
             wired_vad = None
         self._audio_recognition = AudioRecognition(
             self._session,
@@ -1327,11 +1400,7 @@ def say(
                 "add a TTS model to AgentSession to enable say()"
             )
 
-        if (
-            isinstance(self.llm, llm.RealtimeModel)
-            and self.llm.capabilities.turn_detection
-            and allow_interruptions is False
-        ):
+        if self._rt_turn_detection_enabled and allow_interruptions is False:
             logger.warning(
                 "the RealtimeModel uses a server-side turn detection, allow_interruptions cannot be False when using VoiceAgent.say(), "  # noqa: E501
                 "disable turn_detection in the RealtimeModel and use VAD on the AgentTask/VoiceAgent instead"  # noqa: E501
@@ -1397,11 +1466,7 @@ def _generate_reply(
         schedule_speech: bool = True,
         input_details: InputDetails = DEFAULT_INPUT_DETAILS,
     ) -> SpeechHandle:
-        if (
-            isinstance(self.llm, llm.RealtimeModel)
-            and self.llm.capabilities.turn_detection
-            and allow_interruptions is False
-        ):
+        if self._rt_turn_detection_enabled and allow_interruptions is False:
             logger.warning(
                 "the RealtimeModel uses a server-side turn detection, allow_interruptions cannot be False when using VoiceAgent.generate_reply(), "  # noqa: E501
                 "disable turn_detection in the RealtimeModel and use VAD on the AgentTask/VoiceAgent instead"  # noqa: E501
@@ -1907,7 +1972,7 @@ def _interrupt_by_audio_activity(
             # disable interruption from audio activity while aec warmup is active
             return
 
-        if isinstance(self.llm, llm.RealtimeModel) and self.llm.capabilities.turn_detection:
+        if self._rt_turn_detection_enabled:
             # ignore if realtime model has turn detection enabled
             return
 
@@ -2249,7 +2314,7 @@ def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
             self.stt is None
             and self._turn_detection != "manual"
             and isinstance(self.llm, llm.RealtimeModel)
-            and not self.llm.capabilities.turn_detection
+            and not self._rt_turn_detection_enabled
             and self._interruption_detection_enabled
             and (
                 # confirmed backchannel for this turn (survives the agent stopping)
@@ -2311,7 +2376,7 @@ async def _user_turn_completed_task(
             user_message.metrics = metrics_report
 
         if isinstance(self.llm, llm.RealtimeModel):
-            if self.llm.capabilities.turn_detection:
+            if self._rt_turn_detection_enabled:
                 return
 
             if self._rt_session is not None:
@@ -4337,7 +4402,7 @@ def _resolve_interruption_detection(self) -> inference.AdaptiveInterruptionDetec
         realtime_llm = self.llm if isinstance(self.llm, llm.RealtimeModel) else None
         if realtime_llm is not None:
             # realtime commits turns manually; barge-in withholds the commit, so no STT is needed
-            can_gatekeep = not realtime_llm.capabilities.turn_detection
+            can_gatekeep = not self._rt_turn_detection_enabled
         else:
             # the STT pipeline gatekeeps by holding and flushing transcripts
             can_gatekeep = (

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

@@ -490,6 +490,8 @@ def __init__(
         )
 
         self._turn_detection = raw_turn_detection
+        # whether the user passed turn_detection, vs the eager TurnDetector() default
+        self._turn_detection_explicit = "turn_detection" in turn_handling
         self._interruption_detection = interruption.get("mode", NOT_GIVEN)
         self._mcp_servers = mcp_servers or None
         if self._mcp_servers:
@@ -1226,6 +1228,7 @@ def update_options(
 
         if is_given(turn_detection):
             self._turn_detection = turn_detection
+            self._turn_detection_explicit = turn_detection is not None
 
         if self._activity is not None:
             self._activity.update_options(

livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/experimental/realtime/realtime_model.py

@@ -472,8 +472,9 @@ def modalities(self) -> MODALITIES:
     def provider(self) -> str:
         return "Amazon"
 
-    def session(self) -> RealtimeSession:
+    def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
         """Return a new RealtimeSession bound to this model instance."""
+        # disabling server-side turn detection is unsupported (can_disable_turn_detection=False)
         sess = RealtimeSession(self)
         self._sessions.add(sess)
         return sess

livekit-plugins/livekit-plugins-google/livekit/plugins/google/realtime/realtime_api.py

@@ -407,7 +407,9 @@ def provider(self) -> str:
         else:
             return "Gemini"
 
-    def session(self) -> RealtimeSession:
+    def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
+        # Gemini drives manual turns via activity_start/activity_end, not commit_audio/clear_audio,
+        # so the pipeline can't gatekeep turns yet; keep can_disable_turn_detection=False for now
         sess = RealtimeSession(self)
         self._sessions.add(sess)
         return sess

livekit-plugins/livekit-plugins-nvidia/livekit/plugins/nvidia/experimental/realtime/realtime_model.py

@@ -154,7 +154,8 @@ def _ensure_http_session(self) -> aiohttp.ClientSession:
             self._http_session = utils.http_context.http_session()
         return self._http_session
 
-    def session(self) -> RealtimeSession:
+    def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
+        # disabling server-side turn detection is unsupported (can_disable_turn_detection=False)
         sess = RealtimeSession(realtime_model=self)
         self._sessions.add(sess)
         return sess