Support adaptive interruption for realtime models

livekit-agents/livekit/agents/inference/interruption.py

@@ -136,6 +136,13 @@ class OverlappingSpeechEvent(BaseModel):
     is_interruption: bool = False
     """Whether interruption is detected."""
 
+    agent_ended: bool = False
+    """True when the overlap ended because the agent finished speaking rather than the user.
+
+    The user may still be talking, so ``is_interruption`` (always ``False`` here) is inconclusive
+    and must not be treated as a confirmed backchannel verdict.
+    """
+
     total_duration: float = 0.0
     """RTT (Round Trip Time) time taken to perform the inference, in seconds."""
 
@@ -180,6 +187,7 @@ def from_cache_entry(
         is_interruption: bool,
         started_at: float | None = None,
         ended_at: float | None = None,
+        agent_ended: bool = False,
     ) -> OverlappingSpeechEvent:
         """Initialize the event from a cache entry.
 
@@ -188,6 +196,7 @@ def from_cache_entry(
             is_interruption: Whether the interruption is detected.
             started_at: The timestamp when the overlap speech started.
             ended_at: The timestamp when the overlap speech ended.
+            agent_ended: Whether the overlap ended because the agent finished speaking.
 
         Returns:
             The initialized event.
@@ -196,6 +205,7 @@ def from_cache_entry(
             type="overlapping_speech",
             detected_at=ended_at or time.time(),
             is_interruption=is_interruption,
+            agent_ended=agent_ended,
             overlap_started_at=started_at,
             speech_input=entry.speech_input,
             probabilities=entry.probabilities,
@@ -232,8 +242,9 @@ def __init__(
 
 
 class _OverlapSpeechEndedSentinel:
-    def __init__(self, ended_at: float) -> None:
+    def __init__(self, ended_at: float, agent_ended: bool = False) -> None:
         self._ended_at = ended_at
+        self._agent_ended = agent_ended
 
 
 class _FlushSentinel:
@@ -623,6 +634,7 @@ async def _reset_state() -> None:
                             is_interruption=False,
                             started_at=self._overlap_started_at,
                             ended_at=input_frame._ended_at,
+                            agent_ended=input_frame._agent_ended,
                         )
                         ev.num_requests = await self._num_requests.get_and_reset()
                         self.send(ev)

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

@@ -1798,10 +1798,7 @@ def _on_error(
         self._session._on_error(error)
 
     def _on_overlap_speech_ended(self, ev: inference.OverlappingSpeechEvent) -> None:
-        if ev.is_interruption:
-            self._interruption_detected = True
-        else:
-            self._interruption_detected = False
+        self._interruption_detected = ev.is_interruption
         self._session.emit("overlapping_speech", ev)
 
     def _on_input_speech_started(self, _: llm.InputSpeechStartedEvent) -> None:
@@ -2057,6 +2054,15 @@ def on_vad_inference_done(self, ev: vad.VADEvent) -> None:
         else:
             self._user_silence_event.set()
 
+    def on_backchannel_confirmed(self) -> None:
+        # clear the buffered backchannel audio so it can't prefix the next committed turn
+        if (
+            self._interruption_detection_enabled
+            and self._rt_session is not None
+            and self._turn_detection not in ("manual", "realtime_llm")
+        ):
+            self._rt_session.clear_audio()
+
     def on_interruption(self, ev: inference.OverlappingSpeechEvent) -> None:
         # restore interruption by audio activity and then immediately interrupt
         self._restore_interruption_by_audio_activity()
@@ -2224,6 +2230,7 @@ def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
             # TODO(theomonnom): should we "forward" this new turn to the next agent/activity?
             return True
 
+        # avoid interruption if the new_transcript is too short
         if (
             self.stt is not None
             and self._turn_detection != "manual"
@@ -2235,7 +2242,31 @@ def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
             < self._session.options.interruption["min_words"]
         ):
             self._cancel_preemptive_generation()
-            # avoid interruption if the new_transcript is too short
+            return False
+
+        # avoid interruption if backchannel is detected with realtime model
+        if (
+            self.stt is None
+            and self._turn_detection != "manual"
+            and isinstance(self.llm, llm.RealtimeModel)
+            and not self.llm.capabilities.turn_detection
+            and self._interruption_detection_enabled
+            and (
+                # confirmed backchannel for this turn (survives the agent stopping)
+                info.backchannel_over_agent
+                # or agent still speaking with nothing flagged yet
+                or (
+                    not self._interruption_detected
+                    and self._current_speech is not None
+                    and not self._current_speech.interrupted
+                )
+            )
+        ):
+            self._cancel_preemptive_generation()
+            # no transcript to gatekeep for realtime barge-in — drop the backchannel turn
+            # and clear the buffered audio so it can't leak into the next committed turn
+            if self._rt_session is not None:
+                self._rt_session.clear_audio()
             return False
 
         old_task = self._user_turn_completed_atask
@@ -4303,13 +4334,22 @@ def using_default_vad(self) -> bool:
         return self._session._using_default_vad
 
     def _resolve_interruption_detection(self) -> inference.AdaptiveInterruptionDetector | None:
-        if not (
-            self.stt is not None
-            and self.stt.capabilities.aligned_transcript
-            and self.stt.capabilities.streaming
-            and self.vad is not None
-            and self._turn_detection not in ("manual", "realtime_llm")
-            and not isinstance(self.llm, llm.RealtimeModel)
+        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
+        else:
+            # the STT pipeline gatekeeps by holding and flushing transcripts
+            can_gatekeep = (
+                self.stt is not None
+                and self.stt.capabilities.aligned_transcript
+                and self.stt.capabilities.streaming
+            )
+
+        if (
+            not can_gatekeep
+            or self.vad is None
+            or self._turn_detection in ("manual", "realtime_llm")
         ):
             if (
                 is_given(self._agent.interruption_detection)

livekit-agents/livekit/agents/voice/audio_recognition.py

@@ -72,6 +72,8 @@ class _EndOfTurnInfo:
     new_transcript: str
     transcript_confidence: float
     metrics: _EndOfTurnMetrics
+    backchannel_over_agent: bool = False
+    """The turn's speech overlapped agent speech and was classified a backchannel by adaptive interruption."""
 
 
 def _compute_end_of_turn_metrics(
@@ -131,6 +133,7 @@ class _UserTurnTracker:
 
 class RecognitionHooks(Protocol):
     def on_interruption(self, ev: inference.OverlappingSpeechEvent) -> None: ...
+    def on_backchannel_confirmed(self) -> None: ...
     def on_start_of_speech(self, ev: vad.VADEvent | None, speech_start_time: float) -> None: ...
     def on_vad_inference_done(self, ev: vad.VADEvent) -> None: ...
     def on_end_of_speech(self, ev: vad.VADEvent | None) -> None: ...
@@ -293,6 +296,9 @@ def __init__(
         self._interruption_enabled: bool = interruption_detection is not None and vad is not None
         self._agent_speaking: bool = False
         self._agent_speech_started_at: float | None = None
+        # turn-scoped backchannel-over-agent verdict from adaptive interruption, consumed and reset at end of turn
+        self._overlap_in_current_turn: bool = False
+        self._turn_backchannel_over_agent: bool = False
 
         _backchannel_boundary: float | tuple[float, float] | None = (
             session.options.interruption.get("backchannel_boundary")
@@ -473,7 +479,7 @@ def _on_end_of_agent_speech(self, *, ignore_user_transcript_until: float) -> Non
         if self._agent_speaking:
             # no interruption is detected, end the inference (idempotent)
             if not is_given(self._ignore_user_transcript_until):
-                self._on_end_of_overlap_speech(ended_at=time.time())
+                self._on_end_of_overlap_speech(ended_at=time.time(), agent_ended=True)
 
             end_cooldown: float = (
                 self._backchannel_boundary[1] if self._backchannel_boundary else 0.0
@@ -509,8 +515,15 @@ def _on_start_of_speech(
         self._endpointing.on_start_of_speech(
             started_at=started_at, overlapping=self._agent_speaking
         )
+        # every speech onset clears the prior backchannel verdict; an overlap re-derives it below
+        self._turn_backchannel_over_agent = False
+        if not self._agent_speaking:
+            self._overlap_in_current_turn = False
+
         if not self._adaptive_interruption_active or not self._agent_speaking:
             return
+        # overlap over agent speech started this turn; gates verdict acceptance below
+        self._overlap_in_current_turn = True
         self._interruption_ch.send_nowait(  # type: ignore[union-attr]
             _OverlapSpeechStartedSentinel(
                 speech_duration=speech_duration,
@@ -538,8 +551,14 @@ def _on_end_of_overlap_speech(
         self,
         ended_at: float,
         user_speaking_span: trace.Span | None = None,
+        agent_ended: bool = False,
     ) -> None:
-        """End interruption inference when agent is speaking and overlap speech ends."""
+        """End interruption inference when agent is speaking and overlap speech ends.
+
+        agent_ended is True when the overlap is force-ended because the agent finished
+        speaking (the user may still be talking), in which case the synthesized verdict
+        is inconclusive and must not be treated as a confirmed backchannel.
+        """
         if not self._adaptive_interruption_active or not self._agent_speaking:
             return
 
@@ -555,7 +574,7 @@ def _on_end_of_overlap_speech(
                 user_speaking_span.set_attribute(trace_types.ATTR_IS_INTERRUPTION, "false")
 
         self._interruption_ch.send_nowait(  # type: ignore[union-attr]
-            _OverlapSpeechEndedSentinel(ended_at=ended_at or time.time())
+            _OverlapSpeechEndedSentinel(ended_at=ended_at or time.time(), agent_ended=agent_ended)
         )
 
     @property
@@ -1381,6 +1400,14 @@ async def _on_overlap_speech_event(self, ev: inference.OverlappingSpeechEvent) -
             )
             return
 
+        # only honor the verdict while this turn's overlap is unresolved so a late verdict
+        # can't leak into the next turn; an interruption supersedes a prior backchannel
+        if self._overlap_in_current_turn and not ev.agent_ended:
+            self._turn_backchannel_over_agent = not ev.is_interruption
+            # clear the backchannel audio, but only between segments — else we'd clip a real turn
+            if not ev.is_interruption and not self._speaking:
+                self._hooks.on_backchannel_confirmed()
+
         if ev.is_interruption:
             self._hooks.on_interruption(ev)
 
@@ -1629,6 +1656,7 @@ async def _bounce_eou_task(
                     new_transcript=self._audio_transcript,
                     transcript_confidence=confidence_avg,
                     metrics=metrics,
+                    backchannel_over_agent=self._turn_backchannel_over_agent,
                 )
             )
             if committed:
@@ -1677,6 +1705,9 @@ async def _bounce_eou_task(
                     self._turn_detector_prediction_fut = None
                     self._turn_detector_flushed = True
 
+            # reset turn-scoped barge-in state once per logical turn (commit or drop)
+            self._turn_backchannel_over_agent = False
+            self._overlap_in_current_turn = False
             self._user_turn_committed = False
 
         if self._end_of_turn_task is not None:

tests/test_agent_session.py

@@ -1777,6 +1777,9 @@ def __init__(self) -> None:
     def on_interruption(self, ev: inference.OverlappingSpeechEvent) -> None:
         self.interruptions.append(ev)
 
+    def on_backchannel_confirmed(self) -> None:
+        pass
+
     def on_start_of_speech(self, ev: object, speech_start_time: float) -> None:
         pass