feat: surface agent backchannel opportunity as a public event

livekit-agents/livekit/agents/__init__.py

@@ -79,6 +79,7 @@
 from .version import __version__
 from .voice import (
     Agent,
+    AgentBackchannelOpportunityEvent,
     AgentEvent,
     AgentFalseInterruptionEvent,
     AgentSession,
@@ -195,6 +196,7 @@ def __getattr__(name: str) -> typing.Any:
     "ConversationItemAddedEvent",
     "AgentStateChangedEvent",
     "AgentFalseInterruptionEvent",
+    "AgentBackchannelOpportunityEvent",
     "UserInputTranscribedEvent",
     "UserStateChangedEvent",
     "UserTranscriptionTimeoutEvent",

livekit-agents/livekit/agents/voice/__init__.py

@@ -10,6 +10,7 @@
 )
 from .audio_recognition import AudioRecognition
 from .events import (
+    AgentBackchannelOpportunityEvent,
     AgentEvent,
     AgentFalseInterruptionEvent,
     AgentStateChangedEvent,
@@ -74,6 +75,7 @@
     "AgentStateChangedEvent",
     "FunctionToolsExecutedEvent",
     "AgentFalseInterruptionEvent",
+    "AgentBackchannelOpportunityEvent",
     "RemoteSession",
     "ToolExecutionUpdatedEvent",
     "ToolCallStarted",

livekit-agents/livekit/agents/voice/agent.py

@@ -14,7 +14,7 @@
 from ..log import logger
 from ..types import NOT_GIVEN, FlushSentinel, NotGivenOr
 from ..utils import is_given, misc
-from .events import UserTurnExceededEvent
+from .events import AgentBackchannelOpportunityEvent, UserTurnExceededEvent
 from .speech_handle import SpeechHandle
 from .tool_executor import ToolHandlingOptions
 from .turn import TurnHandlingOptions, _migrate_turn_handling
@@ -352,6 +352,31 @@ async def on_user_turn_exceeded(self, ev: UserTurnExceededEvent) -> None:
             tool_choice="none",
         )
 
+    async def on_backchannel_opportunity(self, ev: AgentBackchannelOpportunityEvent) -> None:
+        """Called when the turn detector predicts a natural mid-turn pause where
+        the agent could insert a short acknowledgment (*backchannel*) such as
+        "mm-hmm", "I see", "right", or "uh-huh".
+
+        Override this method to emit a backchannel phrase. The default
+        implementation does nothing so that agents that don't need backchannels
+        are unaffected.
+
+        Use ``ev.end_of_turn_margin`` to pick an appropriate phrase:
+        a large positive margin means the user is clearly still mid-turn
+        (affirmative words like "right" or "okay" are safe); a small or
+        negative margin means a reply is imminent, so prefer neutral sounds
+        like "hmm" or "uh-huh" that won't conflict with the upcoming reply.
+
+        Example::
+
+            async def on_backchannel_opportunity(
+                self, ev: AgentBackchannelOpportunityEvent
+            ) -> None:
+                phrase = "right" if ev.end_of_turn_margin > 0.3 else "mm-hmm"
+                await self.session.say(phrase, allow_interruptions=True)
+        """
+        pass
+
     def stt_node(
         self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
     ) -> (

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

@@ -54,6 +54,7 @@
 )
 from .endpointing import create_endpointing
 from .events import (
+    AgentBackchannelOpportunityEvent,
     AgentFalseInterruptionEvent,
     AgentState,
     AgentStateChangedEvent,
@@ -66,7 +67,6 @@
     UserInputTranscribedEvent,
     UserTranscriptionTimeoutEvent,
     UserTurnExceededEvent,
-    _AgentBackchannelOpportunityEvent,
 )
 from .generation import (
     ToolExecutionOutput,
@@ -2394,10 +2394,18 @@ def on_eot_prediction(self, ev: EotPredictionEvent) -> None:
         if (host := self._session._session_host) is not None:
             host._on_eot_prediction(ev)
 
-    def on_agent_backchannel_opportunity(self, ev: _AgentBackchannelOpportunityEvent) -> None:
-        # TODO: consume the backchannel opportunity internally (e.g. trigger a
-        # backchannel phrase). Kept internal for now — not surfaced as a public event.
-        pass
+    def on_agent_backchannel_opportunity(self, ev: AgentBackchannelOpportunityEvent) -> None:
+        self._session.emit("agent_backchannel_opportunity", ev)
+        self._create_speech_task(
+            self._agent_backchannel_opportunity_task(ev),
+            name="AgentActivity.on_backchannel_opportunity",
+        )
+
+    @utils.log_exceptions(logger=logger)
+    async def _agent_backchannel_opportunity_task(
+        self, ev: AgentBackchannelOpportunityEvent
+    ) -> None:
+        await self._agent.on_backchannel_opportunity(ev)
 
     def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
         # IMPORTANT: This method is sync to avoid it being cancelled by the AudioRecognition

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

@@ -35,9 +35,9 @@
 from ._utils import _set_participant_attributes
 from .endpointing import BaseEndpointing
 from .events import (
+    AgentBackchannelOpportunityEvent,
     EotPredictionEvent,
     UserTurnExceededEvent,
-    _AgentBackchannelOpportunityEvent,
 )
 from .turn import (
     TurnDetectionEvent,
@@ -143,7 +143,7 @@ def on_final_transcript(self, ev: stt.SpeechEvent, *, speaking: bool | None = No
     def on_transcription_timeout(self, *, speech_duration: float, turn_start: float) -> None: ...
     def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool: ...
     def on_eot_prediction(self, ev: EotPredictionEvent) -> None: ...
-    def on_agent_backchannel_opportunity(self, ev: _AgentBackchannelOpportunityEvent) -> None: ...
+    def on_agent_backchannel_opportunity(self, ev: AgentBackchannelOpportunityEvent) -> None: ...
     def on_preemptive_generation(self, info: _PreemptiveGenerationInfo) -> None: ...
     def on_user_turn_exceeded(self, ev: UserTurnExceededEvent) -> None: ...
     def retrieve_chat_ctx(self) -> llm.ChatContext: ...
@@ -286,6 +286,7 @@ def __init__(
         self._stt_pipeline: _STTPipeline | None = None
         self._vad_ch: aio.Chan[rtc.AudioFrame] | None = None
         self._vad_stream: VADStream | None = None
+        self._vad_generation = 0
 
         self._tasks: set[asyncio.Task[Any]] = set()
 
@@ -943,20 +944,24 @@ def _check_vad_silence_requirement(
     def _update_vad(self, vad: vad.VAD | None) -> None:
         self._vad = vad
         self._check_vad_silence_requirement()
-        if vad:
-            self._vad_stream = None
-            self._vad_ch = aio.Chan[rtc.AudioFrame]()
-            self._vad_atask = asyncio.create_task(
-                self._vad_task(vad, self._vad_ch, self._vad_atask)
-            )
-        elif self._vad_atask is not None:
+
+        self._vad_generation += 1
+        current_generation = self._vad_generation
+
+        if self._vad_atask is not None:
             task = asyncio.create_task(aio.cancel_and_wait(self._vad_atask))
             task.add_done_callback(lambda _: self._tasks.discard(task))
             self._tasks.add(task)
             self._vad_atask = None
             self._vad_ch = None
             self._vad_stream = None
 
+        if vad:
+            self._vad_ch = aio.Chan[rtc.AudioFrame]()
+            self._vad_atask = asyncio.create_task(
+                self._vad_task(vad, self._vad_ch, current_generation)
+            )
+
         self._interruption_enabled = (
             self._interruption_detection is not None and self._vad is not None
         )
@@ -1700,7 +1705,7 @@ async def _bounce_eou_task(
                                 and backchannel_probability >= backchannel_threshold
                             ):
                                 self._hooks.on_agent_backchannel_opportunity(
-                                    _AgentBackchannelOpportunityEvent(
+                                    AgentBackchannelOpportunityEvent(
                                         probability=backchannel_probability,
                                         threshold=backchannel_threshold,
                                         end_of_turn_probability=end_of_turn_probability,
@@ -1872,11 +1877,8 @@ async def _vad_task(
         self,
         vad: vad.VAD,
         audio_input: AsyncIterable[rtc.AudioFrame],
-        task: asyncio.Task[None] | None,
+        generation: int,
     ) -> None:
-        if task is not None:
-            await aio.cancel_and_wait(task)
-
         stream = vad.stream()
         self._vad_stream = stream
 
@@ -1889,6 +1891,8 @@ async def _forward() -> None:
 
         try:
             async for ev in stream:
+                if generation != self._vad_generation:
+                    break
                 await self._on_vad_event(ev)
         finally:
             await aio.cancel_and_wait(forward_task)
@@ -1897,11 +1901,12 @@ async def _forward() -> None:
                 self._vad_stream = None
 
             # reset the speaking state to prevent stuck user speaking state during handoff
-            if self._speaking:
-                with tracer.use_span(self._ensure_user_turn_span()):
-                    self._hooks.on_end_of_speech(None)
-                self._speaking = False
-                self._vad_speech_started = False
+            if generation == self._vad_generation:
+                if self._speaking:
+                    with tracer.use_span(self._ensure_user_turn_span()):
+                        self._hooks.on_end_of_speech(None)
+                    self._speaking = False
+                    self._vad_speech_started = False
 
     @utils.log_exceptions(logger=logger)
     async def _interruption_task(

livekit-agents/livekit/agents/voice/events.py

@@ -291,6 +291,7 @@ def _make_update_pair(
     "user_transcription_timeout",
     "conversation_item_added",
     "agent_false_interruption",
+    "agent_backchannel_opportunity",
     "overlapping_speech",
     "function_tools_executed",
     "metrics_collected",
@@ -352,26 +353,43 @@ class EotPredictionEvent(BaseModel):
     created_at: float = Field(default_factory=time.time)
 
 
-class _AgentBackchannelOpportunityEvent(BaseModel):
-    """Internal: a window in which the agent could backchannel (a short
-    acknowledgment such as "mm-hmm"), as predicted by the turn detector. Passed to
-    ``AgentActivity`` only — not surfaced as a public ``AgentSession`` event yet.
+class AgentBackchannelOpportunityEvent(BaseModel):
+    """Emitted when the turn detector predicts a natural mid-turn pause where the
+    agent could insert a short acknowledgment (a *backchannel*) such as
+    "mm-hmm", "I see", "right", or "okay".
 
-    ``AgentActivity`` owns the decision of what to do with it. The end-of-turn margin
-    (``end_of_turn_threshold - end_of_turn_probability``) gives a progressive risk axis:
-    a large positive margin means the user is clearly still going, so riskier
-    backchannels (yeah/okay/right) are safe; a small margin (or a negative one, where
-    ``end_of_turn_probability >= end_of_turn_threshold`` and a reply is imminent) calls
-    for safe, less ambiguous ones (hmm/uh-huh) that won't collide with the reply."""
+    The ``end_of_turn_margin`` property gives a progressive risk axis for choosing
+    an appropriate phrase:
+
+    * **Large positive margin** (user is clearly still mid-turn): riskier, more
+      affirmative phrases ("yeah", "okay", "right") are safe.
+    * **Small or negative margin** (reply may be imminent): prefer safer,
+      less-committal sounds ("hmm", "uh-huh") that won't semantically conflict
+      with the upcoming reply.
+
+    Subscribe via ``session.on("agent_backchannel_opportunity", handler)`` or
+    override ``Agent.on_backchannel_opportunity()``.
+    """
 
     type: Literal["agent_backchannel_opportunity"] = "agent_backchannel_opportunity"
     probability: float
+    """Backchannel probability predicted by the turn detector (0–1)."""
     threshold: float
+    """Minimum probability required before this event is fired."""
     end_of_turn_probability: float
+    """Probability that the user is *ending* their turn right now (0–1)."""
     end_of_turn_threshold: float
+    """Threshold above which the agent would start replying instead."""
     language: str | None = None
+    """BCP-47 language tag detected for the current utterance, if available."""
     created_at: float = Field(default_factory=time.time)
 
+    @property
+    def end_of_turn_margin(self) -> float:
+        """Positive margin → user is mid-turn (safe to backchannel).
+        Negative margin → reply may be imminent (use cautious phrases)."""
+        return self.end_of_turn_threshold - self.end_of_turn_probability
+
 
 class AgentFalseInterruptionEvent(BaseModel):
     type: Literal["agent_false_interruption"] = "agent_false_interruption"
@@ -585,6 +603,7 @@ class CloseEvent(BaseModel):
     | UserStateChangedEvent
     | AgentStateChangedEvent
     | AgentFalseInterruptionEvent
+    | AgentBackchannelOpportunityEvent
     | MetricsCollectedEvent
     | SessionUsageUpdatedEvent
     | ConversationItemAddedEvent