feat(voice): implement blocking acknowledgment system

examples/voice_agents/README.md

@@ -35,6 +35,7 @@ This directory contains a comprehensive collection of voice-based agent examples
 - [`inactive_user.py`](./inactive_user.py) - Handling inactive users with the `user_state_changed` event hook
 - [`resume_interrupted_agent.py`](./resume_interrupted_agent.py) - Resuming agent speech after false interruption detection
 - [`toggle_io.py`](./toggle_io.py) - Dynamically toggling audio input/output during conversations
+- [`acknowledgment.py`](./acknowledgment.py) - Using blocking acknowledgment to provide immediate feedback before LLM responses
 
 ### 🤖 Multi-agent & AgentTask Use Cases
 

examples/voice_agents/acknowledgment.py

@@ -0,0 +1,100 @@
+import asyncio
+import logging
+
+from dotenv import load_dotenv
+
+from livekit.agents import (
+    Agent,
+    AgentServer,
+    AgentSession,
+    JobContext,
+    JobProcess,
+    cli,
+    llm,
+)
+from livekit.plugins import silero
+from livekit.plugins.turn_detector.multilingual import MultilingualModel
+
+logger = logging.getLogger("acknowledgment-agent")
+
+load_dotenv()
+
+
+class AcknowledgmentAgent(Agent):
+    def __init__(self) -> None:
+        super().__init__(
+            instructions="You are a helpful and friendly assistant. Keep your responses concise."
+        )
+
+    async def on_user_turn_completed(
+        self, turn_ctx: llm.ChatContext, new_message: llm.ChatMessage
+    ) -> None:
+        """
+        This callback is triggered when the user finishes speaking.
+        We use it to demonstrate the blocking acknowledgment feature.
+        """
+
+        # 1. Start waiting for acknowledgment.
+        # This blocks the speech queue with a high-priority handle.
+        # It ensures that even if the LLM responds, its speech won't start
+        # until the acknowledgment is either played or skipped.
+        # The timeout prevents the queue from being blocked indefinitely.
+        self.wait_for_acknowledgment(timeout=3.0)
+
+        # 2. Simulate some backend work (e.g., database lookup, API call).
+        # While this delay is happening, the AgentActivity is already starting the LLM generation in parallel.
+        await asyncio.sleep(0.8)
+
+        # 3. Set the acknowledgment message.
+        # This signals the acknowledgment task to synthesize and play this message immediately.
+        # If the LLM response had already arrived (very fast TTFT), this call would effectively
+        # do nothing because the acknowledgment would have been automatically skipped.
+        self.set_acknowledgment("Just a second, I'm checking that for you...")
+
+        # The full LLM response will follow automatically once the acknowledgment message
+        # finishes playing.
+
+
+server = AgentServer()
+
+
+def prewarm(proc: JobProcess):
+    proc.userdata["vad"] = silero.VAD.load()
+
+
+server.setup_fnc = prewarm
+
+
+@server.rtc_session()
+async def entrypoint(ctx: JobContext):
+    logger.info(f"Connecting to room {ctx.room.name}")
+    session = AgentSession(
+        stt="deepgram/nova-3",
+        llm="openai/gpt-4o-mini",
+        tts="cartesia/sonic-2",
+        turn_detection=MultilingualModel(),
+        vad=ctx.proc.userdata["vad"],
+    )
+
+    @session.on("speech_created")
+    def _on_speech_created(ev):
+        if ev.source == "acknowledgment":
+            logger.info("Acknowledgment speech handle created")
+
+            def _on_done(handle):
+                if handle.interrupted:
+                    logger.info("Acknowledgment was SKIPPED or TIMED OUT (LLM responded early)")
+                else:
+                    logger.info("Acknowledgment message PLAYED successfully")
+
+            ev.speech_handle.add_done_callback(_on_done)
+
+    await session.start(
+        agent=AcknowledgmentAgent(),
+        room=ctx.room,
+    )
+    logger.info("Agent started")
+
+
+if __name__ == "__main__":
+    cli.run_app(server)

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

@@ -221,6 +221,31 @@ async def on_user_turn_completed(
         """
         pass
 
+    def wait_for_acknowledgment(self, *, timeout: float = 4.0) -> None:
+        """
+        Schedules a high-priority acknowledgment message to be played before the LLM response.
+        This call is non-blocking and will return immediately.
+
+        Args:
+            timeout (float): The maximum time to wait for `set_acknowledgment` before skipping.
+        """
+        self._get_activity_or_raise().wait_for_acknowledgment(timeout)
+
+    def set_acknowledgment(self, text: str) -> None:
+        """
+        Sets the message for a previously scheduled acknowledgment.
+
+        Args:
+            text (str): The text to be synthesized and played.
+        """
+        self._get_activity_or_raise().set_acknowledgment(text)
+
+    def skip_acknowledgment(self) -> None:
+        """
+        Skips the currently waiting acknowledgment message.
+        """
+        self._get_activity_or_raise().skip_acknowledgment()
+
     def stt_node(
         self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
     ) -> (

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

@@ -131,6 +131,10 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
 
         self._drain_blocked_tasks: list[asyncio.Task[Any]] = []
         self._mcp_tools: list[mcp.MCPTool] = []
+        self._acknowledgment_set = False
+        self._acknowledgment_skipped = False
+        self._acknowledgment_text: str | None = None
+        self._acknowledgment_event = asyncio.Event()
 
         self._on_enter_task: asyncio.Task | None = None
         self._on_exit_task: asyncio.Task | None = None
@@ -724,7 +728,10 @@ async def aclose(self) -> None:
             self._cancel_preemptive_generation()
 
             await self._close_session()
-            await asyncio.gather(*self._interrupt_background_speeches(force=False))
+            await asyncio.gather(*self._interrupt_background_speeches(force=True))
+
+            for task in self._speech_tasks:
+                task.cancel()
 
             if self._scheduling_atask is not None:
                 await utils.aio.cancel_and_wait(self._scheduling_atask)
@@ -1007,6 +1014,70 @@ def _schedule_speech(self, speech: SpeechHandle, priority: int, force: bool = Fa
         speech._mark_scheduled()
         self._wake_up_scheduling_task()
 
+    def wait_for_acknowledgment(self, timeout: float = 4.0) -> None:
+        handle = SpeechHandle.create(allow_interruptions=self.allow_interruptions)
+        self._session.emit(
+            "speech_created",
+            SpeechCreatedEvent(speech_handle=handle, user_initiated=False, source="acknowledgment"),
+        )
+        self._create_speech_task(
+            self._acknowledgment_speech_task(handle, timeout),
+            speech_handle=handle,
+            name="AgentActivity.acknowledgment",
+        )
+        self._schedule_speech(handle, SpeechHandle.SPEECH_PRIORITY_HIGH)
+
+    async def _acknowledgment_speech_task(self, handle: SpeechHandle, timeout: float) -> None:
+        event_wait_task = asyncio.ensure_future(self._acknowledgment_event.wait())
+        try:
+            # Wait for acknowledgment event, but also listen for interruptions
+            await asyncio.wait_for(
+                handle.wait_if_not_interrupted([event_wait_task]),
+                timeout=timeout,
+            )
+        except asyncio.TimeoutError:
+            self._acknowledgment_skipped = True
+        finally:
+            await utils.aio.cancel_and_wait(event_wait_task)
+
+        if handle.interrupted or self._acknowledgment_skipped or not self._acknowledgment_set:
+            if not handle.interrupted:
+                handle.interrupt(force=True)
+
+            handle._mark_done()
+            return
+
+        # Play the acknowledgment
+        await self._tts_task_impl(
+            speech_handle=handle,
+            text=self._acknowledgment_text or "",
+            audio=None,
+            add_to_chat_ctx=True,
+            model_settings=ModelSettings(),
+        )
+        handle._mark_done()
+
+    def set_acknowledgment(self, text: str) -> None:
+        if self._acknowledgment_skipped:
+            return
+
+        self._acknowledgment_text = text
+        self._acknowledgment_set = True
+        self._acknowledgment_event.set()
+
+    def skip_acknowledgment(self) -> None:
+        if self._acknowledgment_set or self._acknowledgment_skipped:
+            return
+
+        self._acknowledgment_skipped = True
+        self._acknowledgment_event.set()
+
+    def _reset_acknowledgment(self) -> None:
+        self._acknowledgment_set = False
+        self._acknowledgment_skipped = False
+        self._acknowledgment_text = None
+        self._acknowledgment_event.clear()
+
     @utils.log_exceptions(logger=logger)
     async def _scheduling_task(self) -> None:
         last_playout_ts = 0.0
@@ -1144,6 +1215,8 @@ def _on_generation_created(self, ev: llm.GenerationCreatedEvent) -> None:
             logger.warning("skipping new realtime generation, the speech scheduling is not running")
             return
 
+        self.skip_acknowledgment()
+
         handle = SpeechHandle.create(allow_interruptions=self.allow_interruptions)
         self._session.emit(
             "speech_created",
@@ -1413,6 +1486,8 @@ async def _user_turn_completed_task(
         # interrupt all background speeches and wait for them to finish to update the chat context
         await asyncio.gather(*self._interrupt_background_speeches(force=False))
 
+        self._reset_acknowledgment()
+
         if isinstance(self.llm, llm.RealtimeModel):
             if self.llm.capabilities.turn_detection:
                 return
@@ -1840,8 +1915,15 @@ async def _pipeline_reply_task_impl(
         )
         tasks.append(llm_task)
 
-        text_tee = utils.aio.itertools.tee(llm_gen_data.text_ch, 2)
-        tts_text_input, tr_input = text_tee
+        text_tee = utils.aio.itertools.tee(llm_gen_data.text_ch, 3)
+        tts_text_input, tr_input, monitoring_ch = text_tee
+
+        async def _skip_acknowledgment_task() -> None:
+            async for _ in monitoring_ch:
+                self.skip_acknowledgment()
+                break
+
+        tasks.append(asyncio.create_task(_skip_acknowledgment_task()))
 
         tts_task: asyncio.Task[bool] | None = None
         tts_gen_data: _TTSGenerationData | None = None

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

@@ -194,7 +194,7 @@ class SpeechCreatedEvent(BaseModel):
     type: Literal["speech_created"] = "speech_created"
     user_initiated: bool
     """True if the speech was created using public methods like `say` or `generate_reply`"""
-    source: Literal["say", "generate_reply"]
+    source: Literal["say", "generate_reply", "acknowledgment"]
     """Source indicating how the speech handle was created"""
     speech_handle: SpeechHandle = Field(..., exclude=True)
     """The speech handle that was created"""