Support the new stateful AudioStreamTrack from getstream

agents-core/pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
 
 requires-python = ">=3.10"
 dependencies = [
-    "getstream[webrtc,telemetry]>=3.4.0,<4",
+    "getstream[webrtc,telemetry]>=4.1.0,<5",
     "aiortc>=1.14.0,<1.15.0",
     "av>=14.2.0, <17",
     "python-dotenv>=1.1.1",

agents-core/vision_agents/core/agents/agents.py

@@ -74,6 +74,11 @@
 
 logger = logging.getLogger(__name__)
 
+# Empty PCM used to drain a track's resampler tail on a bare end-of-turn marker
+# (an AudioOutputChunk with final=True but no data). With no samples nothing is
+# resampled — write(..., final=True) only flushes the tail — so the rate is inert.
+_DRAIN_MARKER = PcmData(sample_rate=48000, format="s16", channels=1)
+
 tracer: Tracer = trace.get_tracer("agents")
 
 
@@ -934,8 +939,12 @@ async def _produce_audio_output(self):
         try:
             async for audio_output in stream:
                 match audio_output:
-                    case AudioOutputChunk(data=data) if data is not None:
-                        await self._audio_track.write(data)
+                    case AudioOutputChunk(data=data, final=final) if data is not None:
+                        await self._audio_track.write(data, final=final)
+                    case AudioOutputChunk(final=True):
+                        # Bare end-of-turn marker (no audio): drain the track's
+                        # resampler tail so the last samples play out.
+                        await self._audio_track.write(_DRAIN_MARKER, final=True)
                     case AudioOutputFlush():
                         await self._audio_track.flush()
 

plugins/anam/vision_agents/plugins/anam/anam_avatar.py

@@ -12,7 +12,7 @@
     PersonaConfig,
     Session,
 )
-from getstream.video.rtc.track_util import PcmData
+from getstream.video.rtc.track_util import FrameResampler, PcmData
 from vision_agents.core.agents.inference import (
     AudioOutputChunk,
     AudioOutputFlush,
@@ -25,6 +25,11 @@
 
 logger = logging.getLogger(__name__)
 
+# Sample rate Anam expects for agent audio input.
+AVATAR_SAMPLE_RATE = 24000
+
+CLOSE_TIMEOUT = 5.0
+
 
 def _task_done_callback(task: asyncio.Task[None]) -> None:
     if not task.cancelled() and task.exception() is not None:
@@ -116,6 +121,9 @@ def __init__(
         self._exit_stack = contextlib.AsyncExitStack()
         self._real_session: Session | None = None
         self._audio_input_stream: AgentAudioInputStream | None = None
+        self._resampler = FrameResampler(
+            rate=AVATAR_SAMPLE_RATE, layout="mono", format="s16", frame_size=0
+        )
         self._audio_receiver_task: asyncio.Task[None] | None = None
         self._video_receiver_task: asyncio.Task[None] | None = None
         self._audio_input_task: asyncio.Task[None] | None = None
@@ -148,11 +156,18 @@ async def close(self) -> None:
 
         self._sync.close()
         try:
-            await self._exit_stack.aclose()
-            await self._client.close()
+            # aiortc/websocket teardown blocks forever when the call is already
+            # gone (no peer to ack the DTLS/ICE and WS close), so bound it.
+            await asyncio.wait_for(self._exit_stack.aclose(), timeout=CLOSE_TIMEOUT)
+        except asyncio.TimeoutError:
+            logger.warning("Timed out closing Anam session")
         except Exception:
             logger.warning("Failed to close Anam avatar publisher", exc_info=True)
         finally:
+            # Close the client even if the session teardown above timed out or
+            # failed, so its HTTP/WS resources don't leak in that scenario.
+            with contextlib.suppress(Exception):
+                await self._client.close()
             logger.debug("Anam avatar publisher closed")
 
     @property
@@ -170,15 +185,17 @@ async def _process_audio_input(self) -> None:
         self._init_avatar_input_stream()
         async for item in self.input_audio_stream:
             if isinstance(item, AudioOutputChunk):
-                # Received normal audio, send it to the avatar
+                # Received normal audio, send it to the avatar.
                 if item.data is not None:
                     await self._send_audio(item.data)
                 # Received final audio chunk (end-of-utterance), flush avatar's audio
                 if item.final:
                     await self._end_turn()
 
             elif isinstance(item, AudioOutputFlush):
-                # Audio was interrupted
+                # Audio was interrupted: discard the resampler tail so it doesn't
+                # bleed into the next turn.
+                self._resampler.flush()
                 await self._end_turn()
                 await self._sync.flush()
                 await self._session.interrupt()
@@ -209,25 +226,33 @@ def _init_avatar_input_stream(self) -> AgentAudioInputStream:
         if self._audio_input_stream is None:
             self._audio_input_stream = self._session.create_agent_audio_input_stream(
                 AgentAudioInputConfig(
-                    encoding="pcm_s16le", sample_rate=24000, channels=1
+                    encoding="pcm_s16le",
+                    sample_rate=AVATAR_SAMPLE_RATE,
+                    channels=1,
                 )
             )
         return self._audio_input_stream
 
     async def _send_audio(self, pcm: PcmData) -> None:
         """
-        Send audio to the avatar.
+        Resample agent audio to the avatar's rate and send it.
         """
         stream = self._init_avatar_input_stream()
-        pcm = pcm.resample(target_channels=1, target_sample_rate=24000)
-        await stream.send_audio_chunk(pcm.to_bytes())
+        for frame in self._resampler.resample(pcm):
+            await stream.send_audio_chunk(frame.to_ndarray().tobytes())
 
     async def _end_turn(self) -> None:
         """
-        Signal end of the turn to the avatar.
+        Drain the resampler tail and signal end of the turn to the avatar.
         """
-        if self._audio_input_stream is not None:
-            await self._audio_input_stream.end_sequence()
+        if self._audio_input_stream is None:
+            return
+        # Flush the resampler tail so the last partial frame isn't dropped.
+        for frame in self._resampler.flush():
+            await self._audio_input_stream.send_audio_chunk(
+                frame.to_ndarray().tobytes()
+            )
+        await self._audio_input_stream.end_sequence()
 
     async def _connect(self) -> None:
         if self._real_session is None:

plugins/getstream/pyproject.toml

@@ -12,7 +12,7 @@ requires-python = ">=3.10"
 license = "MIT"
 dependencies = [
     "vision-agents",
-    "getstream[webrtc,telemetry]>=3.4.0,<4",
+    "getstream[webrtc,telemetry]>=4.1.0,<5",
 ]
 
 [project.urls]

plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py

@@ -6,7 +6,7 @@
 from uuid import uuid4
 
 import av
-from getstream.video.rtc.track_util import AudioFormat, PcmData
+from getstream.video.rtc.track_util import AudioFormat, FrameResampler, PcmData
 from livekit import api, rtc
 from PIL import Image
 from vision_agents.core.utils.utils import cancel_and_wait
@@ -69,6 +69,9 @@ def __init__(
 
         self._room: rtc.Room | None = None
         self._stream_writer: rtc.ByteStreamWriter | None = None
+        self._resampler = FrameResampler(
+            rate=_SAMPLE_RATE, layout="mono", format="s16", frame_size=0
+        )
         self._connected = False
         self._tasks: set[asyncio.Task[None]] = set()
 
@@ -161,36 +164,33 @@ def on_disconnected(reason: str) -> None:
         self._connected = True
 
     async def send_audio(self, pcm: PcmData) -> None:
-        """Send a PCM audio chunk to LemonSlice via a LiveKit byte stream.
+        """Resample a PCM audio chunk to 16 kHz mono and send it to LemonSlice."""
+        for frame in self._resampler.resample(pcm):
+            await self._write_frame(frame)
 
-        Args:
-            pcm: Audio data to send. Resampled to 16 kHz mono automatically.
-        """
+    async def _write_frame(self, frame: av.AudioFrame) -> None:
+        """Write one resampled audio frame to the LemonSlice byte stream."""
         if self._room is None or not self._room.isconnected():
             return
 
-        if pcm.sample_rate != _SAMPLE_RATE or pcm.channels != _NUM_CHANNELS:
-            pcm = pcm.resample(
-                target_sample_rate=_SAMPLE_RATE,
-                target_channels=_NUM_CHANNELS,
-            )
-
         if self._stream_writer is None:
             self._stream_writer = await self._room.local_participant.stream_bytes(
                 name=f"AUDIO_{uuid4()}",
                 topic=_AUDIO_STREAM_TOPIC,
                 destination_identities=[_AVATAR_IDENTITY],
                 attributes={
-                    "sample_rate": str(pcm.sample_rate),
-                    "num_channels": str(pcm.channels),
+                    "sample_rate": str(_SAMPLE_RATE),
+                    "num_channels": str(_NUM_CHANNELS),
                 },
             )
             logger.debug("Opened audio byte stream to LemonSlice")
 
-        await self._stream_writer.write(pcm.to_bytes())
+        await self._stream_writer.write(frame.to_ndarray().tobytes())
 
     async def flush(self) -> None:
-        """Close the current byte stream, signalling end of a TTS segment."""
+        """Flush the resampler tail and close the byte stream (segment end)."""
+        for frame in self._resampler.flush():
+            await self._write_frame(frame)
         if self._stream_writer is not None:
             await self._stream_writer.aclose()
             self._stream_writer = None

plugins/liveavatar/vision_agents/plugins/liveavatar/liveavatar_websocket.py

@@ -4,8 +4,9 @@
 import logging
 import uuid
 
+import av
 import websockets
-from getstream.video.rtc.track_util import PcmData
+from getstream.video.rtc.track_util import FrameResampler, PcmData
 from websockets.asyncio.client import ClientConnection
 from websockets.exceptions import ConnectionClosed
 
@@ -27,6 +28,12 @@ def __init__(
         self._ws: ClientConnection | None = None
         self._closed = False
         self._reconnect_lock = asyncio.Lock()
+        self._resampler = FrameResampler(
+            rate=sample_rate,
+            layout="stereo" if num_channels == 2 else "mono",
+            format="s16",
+            frame_size=0,
+        )
 
     @property
     def connected(self) -> bool:
@@ -65,21 +72,26 @@ async def close(self) -> None:
             self._ws = None
 
     async def send_audio_frame(self, pcm: PcmData) -> None:
-        pcm = pcm.resample(
-            target_sample_rate=self._sample_rate,
-            target_channels=self._num_channels,
-        )
-        b64 = base64.b64encode(pcm.to_bytes()).decode("ascii")
-        await self._send_json({"type": "agent.speak", "audio": b64})
+        for frame in self._resampler.resample(pcm):
+            await self._send_frame(frame)
 
     async def end_turn(self) -> None:
+        # Flush the resampler tail so the utterance plays out, then end the turn.
+        for frame in self._resampler.flush():
+            await self._send_frame(frame)
         await self._send_json({"type": "agent.speak_end"})
 
     async def interrupt(self) -> None:
+        # Discard the resampler tail so it doesn't bleed into the next turn.
+        self._resampler.flush()
         await self._send_json(
             {"type": "agent.interrupt", "event_id": str(uuid.uuid4())}
         )
 
+    async def _send_frame(self, frame: av.AudioFrame) -> None:
+        b64 = base64.b64encode(frame.to_ndarray().tobytes()).decode("ascii")
+        await self._send_json({"type": "agent.speak", "audio": b64})
+
     async def _send_json(self, msg: dict[str, object]) -> None:
         if self._closed:
             raise RuntimeError("liveavatar_ws is closed")

plugins/local/tests/test_tracks.py

@@ -33,7 +33,7 @@ async def test_audio_track_write(self) -> None:
         track = LocalOutputAudioTrack(audio_output=output)
         track.start()
 
-        samples = np.array([100, 200, 300, 400], dtype=np.int16)
+        samples = np.array([[100, 200], [300, 400]], dtype=np.int16)
         pcm = PcmData(
             samples=samples,
             sample_rate=48000,
@@ -62,7 +62,7 @@ async def test_audio_track_flush(self) -> None:
         track = LocalOutputAudioTrack(audio_output=output)
         track.start()
 
-        samples = np.array([100, 200, 300, 400], dtype=np.int16)
+        samples = np.array([[100, 200], [300, 400]], dtype=np.int16)
         pcm = PcmData(
             samples=samples,
             sample_rate=48000,
@@ -84,7 +84,7 @@ async def test_playback_task_processes_queue(self) -> None:
         track = LocalOutputAudioTrack(audio_output=output)
         track.start()
 
-        samples = np.array([100, 200, 300, 400], dtype=np.int16)
+        samples = np.array([[100, 200], [300, 400]], dtype=np.int16)
         pcm = PcmData(
             samples=samples,
             sample_rate=48000,
@@ -110,7 +110,9 @@ async def test_resampling(self) -> None:
         track = LocalOutputAudioTrack(audio_output=output)
         track.start()
 
-        samples = np.array([100, 200, 300, 400], dtype=np.int16)
+        # 320 samples ≈ 20ms @16k; the stateful resampler buffers sub-filter-length
+        # input, so a realistic chunk is needed for it to emit any output.
+        samples = np.tile(np.array([100, 200, 300, 400], dtype=np.int16), 80)
         pcm = PcmData(
             samples=samples,
             sample_rate=16000,
@@ -123,3 +125,40 @@ async def test_resampling(self) -> None:
         assert len(output.written) == 1
 
         track.stop()
+
+    async def test_final_drains_resampler_tail(self) -> None:
+        # 16k -> 48k leaves a small tail buffered in the resampler; final=True must
+        # drain it, so the utterance plays out longer than without it.
+        samples = np.tile(np.array([100, 200, 300, 400], dtype=np.int16), 80)
+
+        out_open = _FakeAudioOutput(sample_rate=48000, channels=1)
+        track_open = LocalOutputAudioTrack(audio_output=out_open)
+        track_open.start()
+        await track_open.write(
+            PcmData(
+                samples=samples,
+                sample_rate=16000,
+                format=AudioFormat.S16,
+                channels=1,
+            ),
+            final=False,
+        )
+        await out_open.wait_consumed()
+        track_open.stop()
+
+        out_final = _FakeAudioOutput(sample_rate=48000, channels=1)
+        track_final = LocalOutputAudioTrack(audio_output=out_final)
+        track_final.start()
+        await track_final.write(
+            PcmData(
+                samples=samples,
+                sample_rate=16000,
+                format=AudioFormat.S16,
+                channels=1,
+            ),
+            final=True,
+        )
+        await out_final.wait_consumed()
+        track_final.stop()
+
+        assert len(out_final.written[0]) > len(out_open.written[0])