voiceassistant: avoid tiny frames on playout

.changeset/popular-ants-press.md

@@ -0,0 +1,5 @@
+---
+"livekit-agents": patch
+---
+
+voiceassistant: avoid tiny frames on playout

livekit-agents/livekit/agents/voice_assistant/agent_playout.py

@@ -15,16 +15,22 @@ class PlayoutHandle:
     def __init__(
         self,
         speech_id: str,
+        audio_source: rtc.AudioSource,
         playout_source: AsyncIterable[rtc.AudioFrame],
         transcription_fwd: transcription.TTSSegmentsForwarder,
     ) -> None:
         self._playout_source = playout_source
+        self._audio_source = audio_source
         self._tr_fwd = transcription_fwd
         self._interrupted = False
-        self._time_played = 0.0
+        self._int_fut = asyncio.Future[None]()
         self._done_fut = asyncio.Future[None]()
         self._speech_id = speech_id
 
+        self._pushed_duration = 0.0
+
+        self._total_played_time: float | None = None  # set whem the playout is done
+
     @property
     def speech_id(self) -> str:
         return self._speech_id
@@ -35,25 +41,29 @@ def interrupted(self) -> bool:
 
     @property
     def time_played(self) -> float:
-        return self._time_played
+        if self._total_played_time is not None:
+            return self._total_played_time
+
+        return self._pushed_duration - self._audio_source.queued_duration
 
     def done(self) -> bool:
-        return self._done_fut.done()
+        return self._done_fut.done() or self._interrupted
 
     def interrupt(self) -> None:
         if self.done():
             return
 
+        self._int_fut.set_result(None)
         self._interrupted = True
 
     def join(self) -> asyncio.Future:
         return self._done_fut
 
 
 class AgentPlayout(utils.EventEmitter[EventTypes]):
-    def __init__(self, *, source: rtc.AudioSource, alpha: float = 0.95) -> None:
+    def __init__(self, *, audio_source: rtc.AudioSource) -> None:
         super().__init__()
-        self._source = source
+        self._audio_source = audio_source
         self._target_volume = 1.0
         self._playout_atask: asyncio.Task[None] | None = None
         self._closed = False
@@ -90,6 +100,7 @@ def play(
 
         handle = PlayoutHandle(
             speech_id=speech_id,
+            audio_source=self._audio_source,
             playout_source=playout_source,
             transcription_fwd=transcription_fwd,
         )
@@ -103,12 +114,24 @@ def play(
     async def _playout_task(
         self, old_task: asyncio.Task[None] | None, handle: PlayoutHandle
     ) -> None:
-        first_frame = True
+        if old_task is not None:
+            await utils.aio.gracefully_cancel(old_task)
 
-        try:
-            if old_task is not None:
-                await utils.aio.gracefully_cancel(old_task)
+        if self._audio_source.queued_duration > 0:
+            # this should not happen, but log it just in case
+            logger.warning(
+                "new playout while the source is still playing",
+                extra={
+                    "speech_id": handle.speech_id,
+                    "queued_duration": self._audio_source.queued_duration,
+                },
+            )
 
+        first_frame = True
+
+        @utils.log_exceptions(logger=logger)
+        async def _capture_task():
+            nonlocal first_frame
             async for frame in handle._playout_source:
                 if first_frame:
                     handle._tr_fwd.segment_playout_started()
@@ -121,32 +144,28 @@ async def _playout_task(
                     self.emit("playout_started")
                     first_frame = False
 
-                if handle.interrupted:
-                    break
-
-                # divide the frame by chunks of 20ms
-                ms20 = frame.sample_rate // 50
-                i = 0
-                while i < len(frame.data):
-                    if handle.interrupted:
-                        break
-
-                    rem = min(ms20, len(frame.data) - i)
-                    data = frame.data[i : i + rem]
-                    i += rem
-
-                    chunk_frame = rtc.AudioFrame(
-                        data=data.tobytes(),
-                        sample_rate=frame.sample_rate,
-                        num_channels=frame.num_channels,
-                        samples_per_channel=rem,
-                    )
-                    await self._source.capture_frame(chunk_frame)
-                    handle._time_played += rem / frame.sample_rate
+                handle._pushed_duration += frame.samples_per_channel / frame.sample_rate
+                await self._audio_source.capture_frame(frame)
+
+            await self._audio_source.wait_for_playout()
+
+        capture_task = asyncio.create_task(_capture_task())
+        try:
+            await asyncio.wait(
+                [capture_task, handle._int_fut],
+                return_when=asyncio.FIRST_COMPLETED,
+            )
         finally:
+            await utils.aio.gracefully_cancel(capture_task)
+
+            handle._total_played_time = (
+                handle._pushed_duration - self._audio_source.queued_duration
+            )
             if not first_frame:
                 if not handle.interrupted:
                     handle._tr_fwd.segment_playout_finished()
+                else:
+                    self._audio_source.clear_queue()  # make sure to remove any queued frames
 
                 self.emit("playout_stopped", handle.interrupted)
 

livekit-agents/livekit/agents/voice_assistant/voice_assistant.py

@@ -456,7 +456,7 @@ async def _main_task(self) -> None:
             track, rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE)
         )
 
-        agent_playout = AgentPlayout(source=audio_source)
+        agent_playout = AgentPlayout(audio_source=audio_source)
         self._agent_output = AgentOutput(
             room=self._room,
             agent_playout=agent_playout,

livekit-agents/setup.py

@@ -48,7 +48,7 @@
     python_requires=">=3.9.0",
     install_requires=[
         "click~=8.1",
-        "livekit>=0.15.2",
+        "livekit>=0.16.2",
         "livekit-api~=0.6",
         "livekit-protocol~=0.6",
         "protobuf>=3",