From 599c50711ca256e150261ff9c10dfaba10d7e690 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 14 Jul 2026 17:17:49 +0200 Subject: [PATCH 1/8] Support the new stateful AudioStreamTrack from `getstream` - pass final=True to `audio_track.write()` to flush the resampler - use stateful resampling in Avatars, LocalEdge and TencentEdge - extra: add close timeout for Anam to prevent hanging when the call ends --- .../vision_agents/core/agents/agents.py | 13 ++++- .../vision_agents/plugins/anam/anam_avatar.py | 39 ++++++++++---- .../lemonslice/lemonslice_rtc_manager.py | 30 +++++------ .../liveavatar/liveavatar_websocket.py | 26 +++++++--- plugins/local/tests/test_tracks.py | 47 +++++++++++++++-- .../vision_agents/plugins/local/tracks.py | 43 ++++++++-------- plugins/tencent/Dockerfile | 3 +- plugins/tencent/tests/test_tracks.py | 7 +-- .../vision_agents/plugins/tencent/tracks.py | 51 +++++++++++-------- tests/test_agents/test_agents.py | 37 +++++++++++++- uv.lock | 34 ++++++------- 11 files changed, 227 insertions(+), 103 deletions(-) diff --git a/agents-core/vision_agents/core/agents/agents.py b/agents-core/vision_agents/core/agents/agents.py index 3d733e877..5da5c5bc1 100644 --- a/agents-core/vision_agents/core/agents/agents.py +++ b/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() diff --git a/plugins/anam/vision_agents/plugins/anam/anam_avatar.py b/plugins/anam/vision_agents/plugins/anam/anam_avatar.py index 129106d81..6cf718be4 100644 --- a/plugins/anam/vision_agents/plugins/anam/anam_avatar.py +++ b/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,8 +156,12 @@ async def close(self) -> None: self._sync.close() try: - await self._exit_stack.aclose() + # 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) await self._client.close() + except asyncio.TimeoutError: + logger.warning("Timed out closing Anam session") except Exception: logger.warning("Failed to close Anam avatar publisher", exc_info=True) finally: @@ -170,15 +182,18 @@ 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. On the final chunk, + # also flush the resampler tail so the utterance plays out. if item.data is not None: - await self._send_audio(item.data) + await self._send_audio(item.data, flush=item.final) # 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,18 +224,22 @@ 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: + async def _send_audio(self, pcm: PcmData, flush: bool = False) -> None: """ - Send audio to the avatar. + Resample agent audio to the avatar's rate and send it. + + When flush is True, also flush the resampler tail (end of utterance). """ 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, flush=flush): + await stream.send_audio_chunk(frame.to_ndarray().tobytes()) async def _end_turn(self) -> None: """ diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py index 59ac70d72..b89ccb713 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py +++ b/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 diff --git a/plugins/liveavatar/vision_agents/plugins/liveavatar/liveavatar_websocket.py b/plugins/liveavatar/vision_agents/plugins/liveavatar/liveavatar_websocket.py index 11275a33a..faa44f1d7 100644 --- a/plugins/liveavatar/vision_agents/plugins/liveavatar/liveavatar_websocket.py +++ b/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") diff --git a/plugins/local/tests/test_tracks.py b/plugins/local/tests/test_tracks.py index dd164f990..410100ff0 100644 --- a/plugins/local/tests/test_tracks.py +++ b/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]) diff --git a/plugins/local/vision_agents/plugins/local/tracks.py b/plugins/local/vision_agents/plugins/local/tracks.py index 2a1d0e33c..310280786 100644 --- a/plugins/local/vision_agents/plugins/local/tracks.py +++ b/plugins/local/vision_agents/plugins/local/tracks.py @@ -18,7 +18,7 @@ import numpy as np import sounddevice as sd from aiortc import AudioStreamTrack, VideoStreamTrack -from getstream.video.rtc.track_util import PcmData +from getstream.video.rtc.track_util import FrameResampler, PcmData from .devices import AudioOutputDevice @@ -57,6 +57,12 @@ def __init__(self, audio_output: AudioOutputDevice, buffer_limit: int = 20): self._running = False self._playback_task: asyncio.Task[None] | None = None self._write_lock = asyncio.Lock() + self._resampler = FrameResampler( + rate=audio_output.sample_rate, + layout="stereo" if audio_output.channels == 2 else "mono", + format="s16", + frame_size=0, + ) async def recv(self) -> av.AudioFrame: """Not supported — this is a write-only playback track.""" @@ -73,14 +79,18 @@ def start(self) -> None: self._running = True self._playback_task = asyncio.create_task(self._playback_loop()) - async def write(self, data: PcmData) -> None: - """Write PCM data to be played on the speaker.""" + async def write(self, data: PcmData, final: bool = False) -> None: + """Write PCM data to be played on the speaker. + + When final is True, flush the resampler tail so the utterance plays out. + """ if not self._running: return async with self._write_lock: - samples = self._process_audio(data) - await self._queue.put(samples) + samples = self._process_audio(data, flush=final) + if samples.size: + await self._queue.put(samples) async def flush(self) -> None: """Clear any pending audio data and abort OS-level playback.""" @@ -90,6 +100,7 @@ async def flush(self) -> None: self._queue.get_nowait() except asyncio.QueueEmpty: break + self._resampler.flush() self._audio_output.flush() def stop(self) -> None: @@ -110,7 +121,7 @@ def stop(self) -> None: self._audio_output.stop() async def _playback_loop(self) -> None: - """Async task that drains the queue into the AudioOutput backend.""" + """Async task that flushes the queue into the AudioOutput backend.""" try: while True: data = await self._queue.get() @@ -126,20 +137,12 @@ async def _playback_loop(self) -> None: except OSError: logger.exception("Audio playback device error") - def _process_audio(self, data: PcmData) -> np.ndarray: - """Resample and convert PcmData to flat int16 numpy for the backend.""" - target_rate = self._audio_output.sample_rate - target_channels = self._audio_output.channels - - if data.sample_rate != target_rate or data.channels != target_channels: - data = data.resample(target_rate, target_channels) - - samples = data.to_int16().samples - - if samples.ndim == 2: - samples = samples.T.flatten() - - return samples + def _process_audio(self, data: PcmData, flush: bool = False) -> np.ndarray: + """Resample PcmData to the backend rate and flatten to int16 for the device.""" + frames = self._resampler.resample(data, flush=flush) + if not frames: + return np.empty(0, dtype=np.int16) + return np.concatenate([f.to_ndarray().reshape(-1) for f in frames]) class LocalVideoTrack(VideoStreamTrack): diff --git a/plugins/tencent/Dockerfile b/plugins/tencent/Dockerfile index f7094d26c..58d873c2b 100644 --- a/plugins/tencent/Dockerfile +++ b/plugins/tencent/Dockerfile @@ -2,8 +2,7 @@ FROM python:3.12-slim # Pick up Debian security patches that aren't baked into the base tag yet. RUN apt-get update \ - && apt-get upgrade -y \ - && rm -rf /var/lib/apt/lists/* + && apt-get upgrade -y && apt-get install -y git && rm -rf /var/lib/apt/lists/* COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv diff --git a/plugins/tencent/tests/test_tracks.py b/plugins/tencent/tests/test_tracks.py index 896a7fb5d..c27b319bc 100644 --- a/plugins/tencent/tests/test_tracks.py +++ b/plugins/tencent/tests/test_tracks.py @@ -54,9 +54,10 @@ def test_empty_input_is_a_noop(self) -> None: def test_offsample_pcm_is_resampled_before_chunking(self) -> None: track = TencentAudioTrack() - # 48 kHz input is 3x the target rate, so 3 * 640 input bytes of - # silence should resample down to 640 bytes after the downmix. - track._write_sync(_pcm_of_size(BYTES_PER_20MS * 3, sample_rate=48000)) + # 48 kHz input is 3x the target rate, so 3 * 640 input bytes of silence + # resample down to 640 bytes at 16 kHz. drain=True flushes the stateful + # resampler's filter tail so the full 20 ms frame lands (end of utterance). + track._write_sync(_pcm_of_size(BYTES_PER_20MS * 3, sample_rate=48000), drain=True) # After resample we have exactly one 20 ms frame's worth of audio # at 16 kHz — no remainder, one chunk. assert len(track._queue) == 1 diff --git a/plugins/tencent/vision_agents/plugins/tencent/tracks.py b/plugins/tencent/vision_agents/plugins/tencent/tracks.py index bb05e2bf5..c2996a26e 100644 --- a/plugins/tencent/vision_agents/plugins/tencent/tracks.py +++ b/plugins/tencent/vision_agents/plugins/tencent/tracks.py @@ -12,14 +12,14 @@ import av from aiortc.mediastreams import MediaStreamError -from getstream.video.rtc.track_util import PcmData +from getstream.video.rtc.track_util import FrameResampler, PcmData from vision_agents.plugins.tencent.bindings import ( AUDIO_CODEC_TYPE_PCM, STREAM_TYPE_VIDEO_HIGH, - VIDEO_PIXEL_FORMAT_YUV420p, VIDEO_ROTATION_0, AudioFrame, PixelFrame, + VIDEO_PIXEL_FORMAT_YUV420p, ) from vision_agents.plugins.tencent.video_utils import ( av_frame_to_yuv420p, @@ -77,6 +77,9 @@ def __init__(self) -> None: self._write_executor = concurrent.futures.ThreadPoolExecutor( max_workers=1, thread_name_prefix="trtc-audio-write" ) + self._resampler = FrameResampler( + rate=SAMPLE_RATE, layout="mono", format="s16", frame_size=0 + ) def set_cloud(self, cloud: Any) -> None: self._cloud = cloud @@ -84,33 +87,35 @@ def set_cloud(self, cloud: Any) -> None: self._sender_thread = threading.Thread(target=self._send_loop, daemon=True) self._sender_thread.start() - async def write(self, pcm: PcmData) -> None: + async def write(self, pcm: PcmData, final: bool = False) -> None: if not self._running or pcm is None: return loop = asyncio.get_running_loop() - await loop.run_in_executor(self._write_executor, self._write_sync, pcm) - - def _write_sync(self, pcm: PcmData) -> None: - if pcm.sample_rate != SAMPLE_RATE or pcm.channels != CHANNELS: - pcm = pcm.resample(target_sample_rate=SAMPLE_RATE, target_channels=CHANNELS) - if pcm.samples is not None and pcm.samples.size > 0: - data = pcm.samples.tobytes() - else: - data = pcm.to_bytes() - if not data: + await loop.run_in_executor(self._write_executor, self._write_sync, pcm, final) + + def _write_sync(self, pcm: PcmData, flush: bool = False) -> None: + # Resample to 16 kHz mono; on the final chunk, flush the resampler tail + # so the utterance plays out completely. + frames = self._resampler.resample(pcm, flush=flush) + if not frames: return with self._lock: - if self._remainder: - data = bytes(self._remainder) + data - self._remainder.clear() - offset = 0 - while offset + BYTES_PER_20MS <= len(data): - self._queue.append(data[offset : offset + BYTES_PER_20MS]) - offset += BYTES_PER_20MS - if offset < len(data): - self._remainder.extend(data[offset:]) + for frame in frames: + self._chunk_into_queue(frame.to_ndarray().tobytes()) self._last_write_at = time.monotonic() + def _chunk_into_queue(self, data: bytes) -> None: + """Split resampled bytes into 20 ms frames; caller holds the lock.""" + if self._remainder: + data = bytes(self._remainder) + data + self._remainder.clear() + offset = 0 + while offset + BYTES_PER_20MS <= len(data): + self._queue.append(data[offset : offset + BYTES_PER_20MS]) + offset += BYTES_PER_20MS + if offset < len(data): + self._remainder.extend(data[offset:]) + def stop(self) -> None: self._running = False # Wait for the sender thread to notice _running=False so callers @@ -128,6 +133,8 @@ async def flush(self) -> None: await loop.run_in_executor(None, self._flush_sync) def _flush_sync(self) -> None: + # Discard the resampler tail so it doesn't bleed into the next turn. + self._resampler.flush() with self._lock: self._queue.clear() self._remainder.clear() diff --git a/tests/test_agents/test_agents.py b/tests/test_agents/test_agents.py index 78150115c..b6c693f3e 100644 --- a/tests/test_agents/test_agents.py +++ b/tests/test_agents/test_agents.py @@ -4,11 +4,12 @@ from uuid import uuid4 import aiortc +import numpy as np import pytest from getstream.video.rtc import AudioStreamTrack from getstream.video.rtc.track_util import PcmData from vision_agents.core import Agent, User -from vision_agents.core.agents.inference import AudioOutputStream +from vision_agents.core.agents.inference import AudioOutputChunk, AudioOutputStream from vision_agents.core.avatars import Avatar from vision_agents.core.edge import Call, EdgeTransport from vision_agents.core.events import EventManager @@ -246,6 +247,40 @@ async def close(self) -> None: ... class TestAgent: + async def test_bare_final_marker_drains_output_track(self): + # A bare end-of-turn marker (AudioOutputChunk with final=True but no data) + # must drain the output track's resampler tail, so the utterance plays out + # further than when no marker follows the audio. + wave = (10000 * np.sin(2 * np.pi * 1000 * np.arange(4800) / 24000)).astype( + np.int16 + ) + pcm = PcmData(samples=wave, sample_rate=24000, format="s16", channels=1) + + ends = [] + for send_marker in (False, True): + agent = Agent( + llm=DummyLLM(), + tts=DummyTTS(), + edge=DummyEdge(), + agent_user=User(name="test"), + ) + track = agent.audio_track + producer = asyncio.create_task(agent._produce_audio_output()) + + await agent._audio_output_stream.send(AudioOutputChunk(data=pcm)) + if send_marker: + await agent._audio_output_stream.send(AudioOutputChunk(final=True)) + agent._audio_output_stream.close() + await producer + + out = np.concatenate( + [(await track.recv()).to_ndarray().reshape(-1) for _ in range(14)] + ) + nonzero = np.nonzero(np.abs(out) > 1)[0] + ends.append(int(nonzero[-1] + 1) if len(nonzero) else 0) + + assert ends[1] > ends[0] + @pytest.mark.parametrize( "edge_params", [ diff --git a/uv.lock b/uv.lock index cf1f2fd37..f953ce872 100644 --- a/uv.lock +++ b/uv.lock @@ -1106,7 +1106,7 @@ name = "cuda-bindings" version = "13.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "cuda-pathfinder" }, + { name = "cuda-pathfinder", marker = "sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/ce/67/5e7dba1ba576dd73da5dee894ca076ca5e959450dfff66d6d510a255d1f7/cuda_bindings-13.3.1-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c7855c4868aabc0cfae28abbe83d56734bdfbd08f08fc234ac1912a12858bf49", size = 6025351, upload-time = "2026-05-29T23:11:49.685Z" }, @@ -1133,34 +1133,34 @@ wheels = [ [package.optional-dependencies] cudart = [ - { name = "nvidia-cuda-runtime", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cuda-runtime", marker = "sys_platform == 'linux'" }, ] cufft = [ - { name = "nvidia-cufft", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cufft", marker = "sys_platform == 'linux'" }, ] cufile = [ { name = "nvidia-cufile", marker = "sys_platform == 'linux'" }, ] cupti = [ - { name = "nvidia-cuda-cupti", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cuda-cupti", marker = "sys_platform == 'linux'" }, ] curand = [ - { name = "nvidia-curand", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-curand", marker = "sys_platform == 'linux'" }, ] cusolver = [ - { name = "nvidia-cusolver", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cusolver", marker = "sys_platform == 'linux'" }, ] cusparse = [ - { name = "nvidia-cusparse", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cusparse", marker = "sys_platform == 'linux'" }, ] nvjitlink = [ - { name = "nvidia-nvjitlink", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-nvjitlink", marker = "sys_platform == 'linux'" }, ] nvrtc = [ - { name = "nvidia-cuda-nvrtc", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-cuda-nvrtc", marker = "sys_platform == 'linux'" }, ] nvtx = [ - { name = "nvidia-nvtx", marker = "sys_platform == 'linux' or sys_platform == 'win32'" }, + { name = "nvidia-nvtx", marker = "sys_platform == 'linux'" }, ] [[package]] @@ -3389,7 +3389,7 @@ name = "nvidia-cublas" version = "13.1.1.3" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-cuda-nvrtc" }, + { name = "nvidia-cuda-nvrtc", marker = "sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/a7/a1/0bd24ee8c8d03adac032fd2909426a00c88f8c57961b1277ded97f91119f/nvidia_cublas-13.1.1.3-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:b7a210458267ac818974c53038fbec2e969d5c99f305ab15c72522fa9f001dd5", size = 542848918, upload-time = "2026-04-08T18:46:22.985Z" }, @@ -3428,7 +3428,7 @@ name = "nvidia-cudnn-cu13" version = "9.20.0.48" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-cublas" }, + { name = "nvidia-cublas", marker = "sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/56/c5/83384d846b2fd17c44bd499b36c75a45ed4f095fbbb2252294e89cea5c5c/nvidia_cudnn_cu13-9.20.0.48-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:e31454ae00094b0c55319d9d15b6fa2fc50a9e1c0f5c8c80fb75258234e731e1", size = 444574296, upload-time = "2026-03-09T19:28:27.751Z" }, @@ -3440,7 +3440,7 @@ name = "nvidia-cufft" version = "12.0.0.61" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-nvjitlink" }, + { name = "nvidia-nvjitlink", marker = "sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/8b/ae/f417a75c0259e85c1d2f83ca4e960289a5f814ed0cea74d18c353d3e989d/nvidia_cufft-12.0.0.61-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:2708c852ef8cd89d1d2068bdbece0aa188813a0c934db3779b9b1faa8442e5f5", size = 214053554, upload-time = "2025-09-04T08:31:38.196Z" }, @@ -3470,9 +3470,9 @@ name = "nvidia-cusolver" version = "12.0.4.66" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-cublas" }, - { name = "nvidia-cusparse" }, - { name = "nvidia-nvjitlink" }, + { name = "nvidia-cublas", marker = "sys_platform != 'win32'" }, + { name = "nvidia-cusparse", marker = "sys_platform != 'win32'" }, + { name = "nvidia-nvjitlink", marker = "sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/c8/c3/b30c9e935fc01e3da443ec0116ed1b2a009bb867f5324d3f2d7e533e776b/nvidia_cusolver-12.0.4.66-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:02c2457eaa9e39de20f880f4bd8820e6a1cfb9f9a34f820eb12a155aa5bc92d2", size = 223467760, upload-time = "2025-09-04T08:33:04.222Z" }, @@ -3484,7 +3484,7 @@ name = "nvidia-cusparse" version = "12.6.3.3" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "nvidia-nvjitlink" }, + { name = "nvidia-nvjitlink", marker = "sys_platform != 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/f8/94/5c26f33738ae35276672f12615a64bd008ed5be6d1ebcb23579285d960a9/nvidia_cusparse-12.6.3.3-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:80bcc4662f23f1054ee334a15c72b8940402975e0eab63178fc7e670aa59472c", size = 162155568, upload-time = "2025-09-04T08:33:42.864Z" }, From 0b71a7d9239db8ec0764c9c6c8a0be53a04baf47 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Sat, 18 Jul 2026 11:13:52 +0200 Subject: [PATCH 2/8] Bump getstream to v4.1.0+ --- agents-core/pyproject.toml | 2 +- plugins/getstream/pyproject.toml | 2 +- uv.lock | 10 +++++----- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/agents-core/pyproject.toml b/agents-core/pyproject.toml index bf0d72308..6de03eeb7 100644 --- a/agents-core/pyproject.toml +++ b/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", diff --git a/plugins/getstream/pyproject.toml b/plugins/getstream/pyproject.toml index 30c67274a..22cdacfc2 100644 --- a/plugins/getstream/pyproject.toml +++ b/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] diff --git a/uv.lock b/uv.lock index f953ce872..2c474c5dc 100644 --- a/uv.lock +++ b/uv.lock @@ -1732,7 +1732,7 @@ wheels = [ [[package]] name = "getstream" -version = "3.5.0" +version = "4.1.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "dataclasses-json" }, @@ -1747,9 +1747,9 @@ dependencies = [ { name = "python-dateutil" }, { name = "twirp" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/9a/9e/73fcc8da5aca867fa2812aff9854f87615f90856bcee08d6a07d015f5a10/getstream-3.5.0.tar.gz", hash = "sha256:35957036284653c6641a24c3f89250e87bab2b17d7609d0a810820a6d437f042", size = 581210, upload-time = "2026-06-22T14:24:13.524Z" } +sdist = { url = "https://files.pythonhosted.org/packages/42/5f/cbb67aee76f65e67b1e5bf45aee0a0d26a97bf0f4bcd7d899698f4f38f77/getstream-4.1.0.tar.gz", hash = "sha256:570b429554890e86302e66075e3fa94e0b3618e8fce8a51c2b781253a08b2fb2", size = 585941, upload-time = "2026-07-17T10:56:41.112Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/ef/37/00d488fda5e1ec954608df989c64f54d40611a706c1e63374f70d0af96c8/getstream-3.5.0-py3-none-any.whl", hash = "sha256:a6f89e4db72a08d16c677b56d03d86d2211755ad71c8fc939df3bbf0a7260fee", size = 368633, upload-time = "2026-06-22T14:24:12.012Z" }, + { url = "https://files.pythonhosted.org/packages/8e/0a/66a07ef0cbf5b43f7ca05892837ae0b830055dddf952c9065b0e36f977f3/getstream-4.1.0-py3-none-any.whl", hash = "sha256:65700fe11b199cf504184acfa5d81327f276f419652db34a79951dc739b4e47d", size = 373090, upload-time = "2026-07-17T10:56:39.57Z" }, ] [package.optional-dependencies] @@ -6554,7 +6554,7 @@ requires-dist = [ { name = "click", specifier = ">=8.1" }, { name = "colorlog", specifier = ">=6.10.1" }, { name = "fastapi", specifier = ">=0.135.1" }, - { name = "getstream", extras = ["telemetry", "webrtc"], specifier = ">=3.4.0,<4" }, + { name = "getstream", extras = ["telemetry", "webrtc"], specifier = ">=4.1.0,<5" }, { name = "jinja2", specifier = ">=3.1" }, { name = "mcp", specifier = ">=1.23.3,<2" }, { name = "mypy", marker = "extra == 'dev'" }, @@ -6923,7 +6923,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "getstream", extras = ["telemetry", "webrtc"], specifier = ">=3.4.0,<4" }, + { name = "getstream", extras = ["telemetry", "webrtc"], specifier = ">=4.1.0,<5" }, { name = "vision-agents", editable = "agents-core" }, ] From e9cdad876e44ed6128053ee79fe2d8336bd4ca46 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Sat, 18 Jul 2026 11:14:22 +0200 Subject: [PATCH 3/8] fix "drain" -> "flush" in tencent track tests --- plugins/tencent/tests/test_tracks.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/plugins/tencent/tests/test_tracks.py b/plugins/tencent/tests/test_tracks.py index c27b319bc..31c9489ac 100644 --- a/plugins/tencent/tests/test_tracks.py +++ b/plugins/tencent/tests/test_tracks.py @@ -2,7 +2,6 @@ import numpy as np from getstream.video.rtc.track_util import AudioFormat, PcmData - from vision_agents.plugins.tencent.tracks import ( BYTES_PER_20MS, CHANNELS, @@ -55,9 +54,11 @@ def test_empty_input_is_a_noop(self) -> None: def test_offsample_pcm_is_resampled_before_chunking(self) -> None: track = TencentAudioTrack() # 48 kHz input is 3x the target rate, so 3 * 640 input bytes of silence - # resample down to 640 bytes at 16 kHz. drain=True flushes the stateful + # resample down to 640 bytes at 16 kHz. flush=True flushes the stateful # resampler's filter tail so the full 20 ms frame lands (end of utterance). - track._write_sync(_pcm_of_size(BYTES_PER_20MS * 3, sample_rate=48000), drain=True) + track._write_sync( + _pcm_of_size(BYTES_PER_20MS * 3, sample_rate=48000), flush=True + ) # After resample we have exactly one 20 ms frame's worth of audio # at 16 kHz — no remainder, one chunk. assert len(track._queue) == 1 From 322d7820b92aea4dc1e8c11d18e86728ce4a4682 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Sat, 18 Jul 2026 11:17:39 +0200 Subject: [PATCH 4/8] tencent: revert Dockerfile changes --- plugins/tencent/Dockerfile | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/plugins/tencent/Dockerfile b/plugins/tencent/Dockerfile index 58d873c2b..7955173f6 100644 --- a/plugins/tencent/Dockerfile +++ b/plugins/tencent/Dockerfile @@ -2,7 +2,8 @@ FROM python:3.12-slim # Pick up Debian security patches that aren't baked into the base tag yet. RUN apt-get update \ - && apt-get upgrade -y && apt-get install -y git && rm -rf /var/lib/apt/lists/* + && apt-get upgrade -y \ + && rm -rf /var/lib/apt/lists/* \ COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv From 8c9dbf5d7067b1696b27d2d734934544a6d1f5ab Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Sat, 18 Jul 2026 11:20:18 +0200 Subject: [PATCH 5/8] tencent: revert Dockerfile changes --- plugins/tencent/Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/tencent/Dockerfile b/plugins/tencent/Dockerfile index 7955173f6..f7094d26c 100644 --- a/plugins/tencent/Dockerfile +++ b/plugins/tencent/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.12-slim # Pick up Debian security patches that aren't baked into the base tag yet. RUN apt-get update \ && apt-get upgrade -y \ - && rm -rf /var/lib/apt/lists/* \ + && rm -rf /var/lib/apt/lists/* COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv From 93d66a4f6f917ae0a4fe9a7da8e912b9ea0a1224 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Sat, 18 Jul 2026 12:50:46 +0200 Subject: [PATCH 6/8] tencent: serialize flush and audio writes in audio track --- plugins/tencent/vision_agents/plugins/tencent/tracks.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/plugins/tencent/vision_agents/plugins/tencent/tracks.py b/plugins/tencent/vision_agents/plugins/tencent/tracks.py index c2996a26e..a319aba1c 100644 --- a/plugins/tencent/vision_agents/plugins/tencent/tracks.py +++ b/plugins/tencent/vision_agents/plugins/tencent/tracks.py @@ -130,7 +130,9 @@ def stop(self) -> None: async def flush(self) -> None: loop = asyncio.get_running_loop() - await loop.run_in_executor(None, self._flush_sync) + # Route through the write executor so the stateful, non-thread-safe + # resampler is only ever touched from one thread (serialised with writes). + await loop.run_in_executor(self._write_executor, self._flush_sync) def _flush_sync(self) -> None: # Discard the resampler tail so it doesn't bleed into the next turn. From a1e2c57287f97664cb189c252204d8adec1490f0 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Sat, 18 Jul 2026 12:51:36 +0200 Subject: [PATCH 7/8] anam.Avatar: suppress client.close() errors, simplify audio flow --- .../vision_agents/plugins/anam/anam_avatar.py | 26 +++++++++++-------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/plugins/anam/vision_agents/plugins/anam/anam_avatar.py b/plugins/anam/vision_agents/plugins/anam/anam_avatar.py index 6cf718be4..61297e5c9 100644 --- a/plugins/anam/vision_agents/plugins/anam/anam_avatar.py +++ b/plugins/anam/vision_agents/plugins/anam/anam_avatar.py @@ -159,12 +159,15 @@ async def close(self) -> None: # 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) - await self._client.close() 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 @@ -182,10 +185,9 @@ 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. On the final chunk, - # also flush the resampler tail so the utterance plays out. + # Received normal audio, send it to the avatar. if item.data is not None: - await self._send_audio(item.data, flush=item.final) + await self._send_audio(item.data) # Received final audio chunk (end-of-utterance), flush avatar's audio if item.final: await self._end_turn() @@ -231,22 +233,24 @@ def _init_avatar_input_stream(self) -> AgentAudioInputStream: ) return self._audio_input_stream - async def _send_audio(self, pcm: PcmData, flush: bool = False) -> None: + async def _send_audio(self, pcm: PcmData) -> None: """ Resample agent audio to the avatar's rate and send it. - - When flush is True, also flush the resampler tail (end of utterance). """ stream = self._init_avatar_input_stream() - for frame in self._resampler.resample(pcm, flush=flush): + 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: From 417555ecaa37a983ed2c17c66c956f24a6c353ec Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Sat, 18 Jul 2026 12:53:04 +0200 Subject: [PATCH 8/8] anam.Avatar: fix ruff --- plugins/anam/vision_agents/plugins/anam/anam_avatar.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/plugins/anam/vision_agents/plugins/anam/anam_avatar.py b/plugins/anam/vision_agents/plugins/anam/anam_avatar.py index 61297e5c9..7393b942d 100644 --- a/plugins/anam/vision_agents/plugins/anam/anam_avatar.py +++ b/plugins/anam/vision_agents/plugins/anam/anam_avatar.py @@ -249,7 +249,9 @@ async def _end_turn(self) -> 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.send_audio_chunk( + frame.to_ndarray().tobytes() + ) await self._audio_input_stream.end_sequence() async def _connect(self) -> None: