From 7b6edf635eb737efa2045db506384651fc10e738 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Wed, 27 May 2026 00:29:34 +0200 Subject: [PATCH 01/11] Use Stream-based calls for LemonSlice integration --- plugins/lemonslice/README.md | 23 +- .../example/lemonslice_avatar_example.py | 3 - plugins/lemonslice/pyproject.toml | 2 - .../tests/test_lemonslice_plugin.py | 23 +- plugins/lemonslice/tests/test_track.py | 139 ++++++ .../plugins/lemonslice/lemonslice_avatar.py | 33 +- .../plugins/lemonslice/lemonslice_client.py | 20 +- .../lemonslice/lemonslice_rtc_manager.py | 402 +++++++++--------- .../vision_agents/plugins/lemonslice/track.py | 75 ++++ uv.lock | 111 ++--- 10 files changed, 501 insertions(+), 330 deletions(-) create mode 100644 plugins/lemonslice/tests/test_track.py create mode 100644 plugins/lemonslice/vision_agents/plugins/lemonslice/track.py diff --git a/plugins/lemonslice/README.md b/plugins/lemonslice/README.md index 2efcad856..a0fdcf293 100644 --- a/plugins/lemonslice/README.md +++ b/plugins/lemonslice/README.md @@ -64,10 +64,9 @@ LEMONSLICE_AGENT_ID=your_agent_id # Or, instead of LEMONSLICE_AGENT_ID: # LEMONSLICE_AGENT_IMAGE_URL=https://example.com/avatar.png -# LemonSlice uses LiveKit as a transport for audio and video -LIVEKIT_URL=wss://your-livekit-server.com -LIVEKIT_API_KEY=your_livekit_api_key -LIVEKIT_API_SECRET=your_livekit_api_secret +# LemonSlice uses Stream as the transport for audio and video +STREAM_API_KEY=your_stream_api_key +STREAM_API_SECRET=your_stream_api_secret ``` ### Avatar Options @@ -79,18 +78,19 @@ lemonslice.Avatar( agent_prompt=None, # Prompt to influence avatar expressions/movements api_key=None, # Optional: override LEMONSLICE_API_KEY env var idle_timeout=None, # Session timeout in seconds - livekit_url=None, # Optional: override LIVEKIT_URL env var - livekit_api_key=None, # Optional: override LIVEKIT_API_KEY env var - livekit_api_secret=None, # Optional: override LIVEKIT_API_SECRET env var - width=1920, # Output video width in pixels - height=1080, # Output video height in pixels + stream_api_key=None, # Optional: override STREAM_API_KEY env var + stream_api_secret=None, # Optional: override STREAM_API_SECRET env var + width=1280, # Output video width in pixels + height=720, # Output video height in pixels + fps=30, # Output video frame rate + buffer_seconds=1.0, # Max video buffer depth in seconds ) ``` ## How It Works -1. **LemonSlice Session**: Creates a session via LemonSlice API, and joins the LiveKit room as a participant -2. **Audio Forwarding**: TTS audio is captured and sent to LemonSlice via the room +1. **LemonSlice Session**: Creates a session via LemonSlice API, and joins the Stream call as a participant +2. **Audio Forwarding**: TTS audio is captured and sent to LemonSlice via the Stream call 3. **Avatar Generation**: LemonSlice generates synchronized avatar video and audio 4. **Video Streaming**: Avatar video is streamed to call participants via GetStream Edge @@ -98,7 +98,6 @@ lemonslice.Avatar( - Python 3.10+ - LemonSlice API key (get one at [lemonslice.com](https://lemonslice.com)) -- LiveKit server (cloud or self-hosted) - GetStream account for video calls - TTS provider (Cartesia, ElevenLabs, etc.) or Realtime LLM diff --git a/plugins/lemonslice/example/lemonslice_avatar_example.py b/plugins/lemonslice/example/lemonslice_avatar_example.py index fd82e9154..ed0523e68 100644 --- a/plugins/lemonslice/example/lemonslice_avatar_example.py +++ b/plugins/lemonslice/example/lemonslice_avatar_example.py @@ -6,9 +6,6 @@ Required environment variables: LEMONSLICE_API_KEY LEMONSLICE_AGENT_ID (or LEMONSLICE_AGENT_IMAGE_URL) - LIVEKIT_URL - LIVEKIT_API_KEY - LIVEKIT_API_SECRET STREAM_API_KEY STREAM_API_SECRET """ diff --git a/plugins/lemonslice/pyproject.toml b/plugins/lemonslice/pyproject.toml index 1fb799cef..f3143f8a1 100644 --- a/plugins/lemonslice/pyproject.toml +++ b/plugins/lemonslice/pyproject.toml @@ -12,8 +12,6 @@ requires-python = ">=3.10" license = "MIT" dependencies = [ "vision-agents", - "livekit>=1.1.2,<2", - "livekit-api>=1.1.0,<2", "httpx>=0.28.1,<1", ] diff --git a/plugins/lemonslice/tests/test_lemonslice_plugin.py b/plugins/lemonslice/tests/test_lemonslice_plugin.py index 3069b4291..ac0d5db99 100644 --- a/plugins/lemonslice/tests/test_lemonslice_plugin.py +++ b/plugins/lemonslice/tests/test_lemonslice_plugin.py @@ -7,10 +7,8 @@ def _make_avatar(**overrides) -> LemonSliceAvatar: default_kwargs = { "agent_id": "test-agent", - "api_key": "ls-test-key", - "livekit_url": "wss://test.livekit.cloud", - "livekit_api_key": "devkey", - "livekit_api_secret": "devsecret", + "stream_api_key": "key", + "stream_api_secret": "secret", } return LemonSliceAvatar(**{**default_kwargs, **overrides}) @@ -32,20 +30,13 @@ async def test_init_missing_api_key_raises(self, monkeypatch: pytest.MonkeyPatch with pytest.raises(ValueError, match="API key required"): _make_avatar(api_key=None) - async def test_init_missing_livekit_url_raises( + async def test_init_missing_stream_secret_raises( self, monkeypatch: pytest.MonkeyPatch ): - monkeypatch.delenv("LIVEKIT_URL", raising=False) - with pytest.raises(ValueError, match="LiveKit URL required"): - _make_avatar(livekit_url=None) - - async def test_init_missing_livekit_secret_raises( - self, monkeypatch: pytest.MonkeyPatch - ): - monkeypatch.delenv("LIVEKIT_API_KEY", raising=False) - monkeypatch.delenv("LIVEKIT_API_SECRET", raising=False) - with pytest.raises(ValueError, match="LiveKit API key and secret required"): - _make_avatar(livekit_api_key=None, livekit_api_secret=None) + monkeypatch.delenv("STREAM_API_KEY", raising=False) + monkeypatch.delenv("STREAM_API_SECRET", raising=False) + with pytest.raises(ValueError, match="Stream API key and secret required"): + _make_avatar(stream_api_key=None, stream_api_secret=None) async def test_video_output(self): avatar = _make_avatar(width=640, height=480) diff --git a/plugins/lemonslice/tests/test_track.py b/plugins/lemonslice/tests/test_track.py new file mode 100644 index 000000000..e1dde6a02 --- /dev/null +++ b/plugins/lemonslice/tests/test_track.py @@ -0,0 +1,139 @@ +import time + +import numpy as np +import pytest +from aiortc.mediastreams import AUDIO_PTIME +from getstream.video.rtc.track_util import AudioFormat, PcmData +from vision_agents.plugins.lemonslice.track import AvatarInputTrack + +_SAMPLE_RATE = 16000 +_CHANNELS = 1 +_SAMPLES_PER_FRAME = int(AUDIO_PTIME * _SAMPLE_RATE) + + +def _audio(num_samples: int) -> PcmData: + return PcmData( + samples=np.zeros(num_samples, dtype=np.int16), + sample_rate=_SAMPLE_RATE, + format=AudioFormat.S16, + channels=_CHANNELS, + ) + + +class TestStampedAudioTrack: + @pytest.fixture + def track(self) -> AvatarInputTrack: + return AvatarInputTrack(sample_rate=_SAMPLE_RATE, channels=_CHANNELS) + + async def test_pts_zero_on_fresh_track(self, track: AvatarInputTrack) -> None: + assert await track.pts() == 0 + + async def test_pts_reflects_buffered_samples_before_any_recv( + self, track: AvatarInputTrack + ) -> None: + await track.write(_audio(_SAMPLES_PER_FRAME * 3)) + assert await track.pts() == _SAMPLES_PER_FRAME * 3 + + async def test_pts_after_emission_equals_last_buffered_frame_pts( + self, track: AvatarInputTrack + ) -> None: + # Once recv() has begun, pts() = _timestamp + queued_samples = the wire + # PTS the last buffered frame will carry when it's emitted. + await track.write(_audio(_SAMPLES_PER_FRAME * 3)) + await track.recv() # _timestamp=0, 2 frames queued + assert await track.pts() == _SAMPLES_PER_FRAME * 2 + await track.recv() # _timestamp=320, 1 frame queued + assert await track.pts() == _SAMPLES_PER_FRAME * 2 + + async def test_recv_frame_shape(self, track: AvatarInputTrack) -> None: + await track.write(_audio(_SAMPLES_PER_FRAME)) + frame = await track.recv() + assert frame.samples == _SAMPLES_PER_FRAME + assert frame.sample_rate == _SAMPLE_RATE + assert frame.pts == 0 + + async def test_pts_advances_by_samples_per_frame( + self, track: AvatarInputTrack + ) -> None: + await track.write(_audio(_SAMPLES_PER_FRAME * 5)) + pts_values = [(await track.recv()).pts for _ in range(5)] + assert pts_values == [i * _SAMPLES_PER_FRAME for i in range(5)] + + async def test_burst_drains_without_pacing(self, track: AvatarInputTrack) -> None: + # 1 second of audio = 50 frames; under wall-clock pacing this would take ~1s. + await track.write(_audio(_SAMPLE_RATE)) + await track.recv() # first recv initializes the timestamp baseline + start = time.monotonic() + for _ in range(49): + await track.recv() + elapsed = time.monotonic() - start + assert elapsed < 49 * AUDIO_PTIME * 0.5, ( + f"burst was paced: {elapsed:.4f}s for 49 frames" + ) + + async def test_silence_paces_at_frame_interval( + self, track: AvatarInputTrack + ) -> None: + await track.recv() # init (first silence frame, also sleeps once) + start = time.monotonic() + await track.recv() + elapsed = time.monotonic() - start + assert AUDIO_PTIME * 0.5 < elapsed < AUDIO_PTIME * 3, ( + f"silence pacing off: {elapsed:.4f}s vs target {AUDIO_PTIME:.4f}s" + ) + + async def test_silence_after_burst_does_not_sleep_through_backlog( + self, track: AvatarInputTrack + ) -> None: + # The drift edge case: PTS runs ~1s ahead of wall clock after the burst. + # If we used the parent's wall-clock-anchored sleep, the next silence + # frame would sleep ~1s to "catch up". We must sleep only AUDIO_PTIME. + await track.write(_audio(_SAMPLE_RATE)) # 50 frames + for _ in range(50): + await track.recv() + start = time.monotonic() + frame = await track.recv() + elapsed = time.monotonic() - start + assert elapsed < AUDIO_PTIME * 5, ( + f"silence after burst slept through backlog: {elapsed:.4f}s" + ) + assert frame.pts == 50 * _SAMPLES_PER_FRAME + + async def test_partial_frame_is_padded_and_bursted( + self, track: AvatarInputTrack + ) -> None: + await track.write(_audio(_SAMPLES_PER_FRAME // 2)) + start = time.monotonic() + frame = await track.recv() + elapsed = time.monotonic() - start + assert elapsed < AUDIO_PTIME * 0.5, f"partial frame was paced: {elapsed:.4f}s" + assert frame.samples == _SAMPLES_PER_FRAME + assert frame.pts == 0 + + async def test_pts_monotonic_across_data_silence_data_transitions( + self, track: AvatarInputTrack + ) -> None: + # data, then silence, then data — PTS continues advancing by samples_per_frame each step. + await track.write(_audio(_SAMPLES_PER_FRAME * 2)) + f0 = await track.recv() + f1 = await track.recv() + f2 = await track.recv() # silence + await track.write(_audio(_SAMPLES_PER_FRAME)) + f3 = await track.recv() # data again + assert [f0.pts, f1.pts, f2.pts, f3.pts] == [ + i * _SAMPLES_PER_FRAME for i in range(4) + ] + + async def test_long_silence_run_does_not_drift_to_zero_or_negative_sleep( + self, track: AvatarInputTrack + ) -> None: + # Many silence frames in a row — each should still sleep ~AUDIO_PTIME, + # not zero (no drift compensation collapsing the interval). + await track.recv() # init + start = time.monotonic() + for _ in range(5): + await track.recv() + elapsed = time.monotonic() - start + assert elapsed > 5 * AUDIO_PTIME * 0.5, ( + f"silence loop did not pace: {elapsed:.4f}s for 5 frames" + ) diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py index 6d726f771..145c977fd 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py @@ -15,7 +15,7 @@ from vision_agents.core.utils.video_track import QueuedVideoTrack from .lemonslice_client import LemonSliceClient -from .lemonslice_rtc_manager import LemonSliceRTCManager +from .lemonslice_rtc_manager import StreamRTCManager logger = logging.getLogger(__name__) @@ -32,7 +32,7 @@ def _task_done_callback(task: asyncio.Task[None]) -> None: class LemonSliceAvatar(Avatar): """LemonSlice avatar video and audio publisher. - Sends TTS audio to LemonSlice over LiveKit and receives synchronized + Sends TTS audio to LemonSlice over a Stream call and receives synchronized avatar video and audio back. For standard LLMs: LemonSlice provides both video and audio. @@ -49,9 +49,9 @@ def __init__( idle_timeout: int | None = None, api_key: str | None = None, base_url: str | None = None, - livekit_url: str | None = None, - livekit_api_key: str | None = None, - livekit_api_secret: str | None = None, + stream_api_key: str | None = None, + stream_api_secret: str | None = None, + stream_call_type: str = "default", width: int = 1280, height: int = 720, fps: int = 30, @@ -66,9 +66,13 @@ def __init__( idle_timeout: Seconds before an idle session is closed. api_key: LemonSlice API key. Uses LEMONSLICE_API_KEY env var if not provided. base_url: LemonSlice API base URL override. - livekit_url: LiveKit server URL. Uses LIVEKIT_URL env var if not provided. - livekit_api_key: LiveKit API key. Uses LIVEKIT_API_KEY env var if not provided. - livekit_api_secret: LiveKit API secret. Uses LIVEKIT_API_SECRET env var if not provided. + stream_api_key: Stream API key. Uses STREAM_API_KEY env var if not provided. + stream_api_secret: Stream API secret. Uses STREAM_API_SECRET env var if not provided. + stream_call_type: Stream call type controlling the default feature set and + per-role permissions for the call. The built-in "default" type is meant + for 1:1/group video+audio calls: it enables audio, video, screensharing, + recording, HLS broadcasting, transcription and ringing, and gives + admins/hosts elevated permissions over regular participants. width: Output video width in pixels. height: Output video height in pixels. fps: Output video frame rate. Must be > 0. @@ -94,13 +98,13 @@ def __init__( client_kwargs["base_url"] = base_url self._client = LemonSliceClient(**client_kwargs) - self._rtc_manager = LemonSliceRTCManager( + self._rtc_manager = StreamRTCManager( on_video=self._on_video_frame, on_audio=self._on_audio_frame, on_disconnect=self._on_disconnect, - livekit_url=livekit_url, - livekit_api_key=livekit_api_key, - livekit_api_secret=livekit_api_secret, + stream_api_secret=stream_api_secret, + stream_api_key=stream_api_key, + stream_call_type=stream_call_type, ) self._sync = AVSynchronizer( width=width, @@ -171,7 +175,10 @@ async def _connect(self) -> None: await self._rtc_manager.connect(credentials) try: await self._client.create_session( - credentials.livekit_url, credentials.livekit_token + call_id=credentials.call_id, + call_type=credentials.call_type, + token=credentials.avatar_token, + api_key=credentials.api_key, ) except Exception: await self._rtc_manager.close() diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py index 06066dd8b..72dfa2178 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py @@ -63,21 +63,27 @@ def __init__( def session_id(self) -> str | None: return self._session_id - async def create_session(self, livekit_url: str, livekit_token: str) -> str: + async def create_session( + self, call_id: str, call_type: str, token: str, api_key: str + ) -> str: """Create a new LemonSlice avatar session. Args: - livekit_url: LiveKit server URL for the avatar to connect to. - livekit_token: LiveKit access token for the avatar participant. + call_id: Stream call ID the avatar should join. + call_type: Stream call type (e.g. "default"). + token: Stream access token for the avatar participant. + api_key: Stream API key. Returns: The created session ID. """ payload: dict[str, object] = { - "transport_type": "livekit", + "transport_type": "stream", "properties": { - "livekit_url": livekit_url, - "livekit_token": livekit_token, + "call_id": call_id, + "call_type": call_type, + "token": token, + "api_key": api_key, }, } @@ -92,7 +98,7 @@ async def create_session(self, livekit_url: str, livekit_token: str) -> str: response = await self._http_client.post("/sessions", json=payload) - if response.status_code != 201: + if response.status_code >= 400: raise LemonSliceSessionError( f"Failed to create session: {response.status_code} - {response.text}", status_code=response.status_code, 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 b89ccb713..1f36be800 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py @@ -2,39 +2,47 @@ import logging from dataclasses import dataclass from os import getenv -from typing import Callable, Coroutine +from typing import Any, Callable, Coroutine from uuid import uuid4 import av +from getstream import AsyncStream +from getstream.video import rtc +from getstream.video.async_call import Call +from getstream.video.rtc.connection_manager import ConnectionManager 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 +from vision_agents.core.utils.utils import cancel_and_wait, get_vision_agents_version + +from .track import AvatarInputTrack logger = logging.getLogger(__name__) -_AUDIO_STREAM_TOPIC = "lk.audio_stream" -_AVATAR_IDENTITY = "avatar" -_PLUGIN_IDENTITY = "plugin" -_SAMPLE_RATE = 16000 -_NUM_CHANNELS = 1 +_AVATAR_AUDIO_SAMPLE_RATE = 16000 +_AVATAR_AUDIO_CHANNELS = 1 +_CUSTOM_EVENT_END_UTTERANCE = "lemonslice.end_utterance" +_CUSTOM_EVENT_INTERRUPT = "lemonslice.interrupt" @dataclass(frozen=True) -class ConnectionCredentials: - """All credentials needed for a LemonSlice LiveKit session.""" +class StreamConnectionCredentials: + """Credentials a LemonSlice avatar needs to join a Stream call.""" + + api_key: str + call_id: str + call_type: str + avatar_user_id: str + avatar_token: str - room_name: str - agent_token: str - livekit_url: str - livekit_token: str +class StreamRTCManager: + """Stream-backed RTC manager for the LemonSlice avatar. -class LemonSliceRTCManager: - """Manages a LiveKit room connection for LemonSlice avatar streaming. + Creates a Stream call, mints a call-scoped token for the avatar, joins as + the plugin participant, publishes outgoing TTS audio, and dispatches + incoming avatar audio and video to the supplied callbacks. - Creates a LiveKit room, sends TTS audio to LemonSlice via data streams, - and receives synchronized avatar video and audio tracks. + flush() emits end-of-utterance RPC event to the avatar call. + interrupt() emits the "interrupt" RPC event to stop the current avatar's utterance """ def __init__( @@ -42,240 +50,224 @@ def __init__( on_video: Callable[[av.VideoFrame], Coroutine[None, None, None]], on_audio: Callable[[PcmData], Coroutine[None, None, None]], on_disconnect: Callable[[], Coroutine[None, None, None]], - livekit_url: str | None = None, - livekit_api_key: str | None = None, - livekit_api_secret: str | None = None, + stream_api_key: str | None = None, + stream_api_secret: str | None = None, + stream_call_type: str = "default", ): - self._livekit_url = livekit_url or getenv("LIVEKIT_URL") or "" - if not self._livekit_url: - raise ValueError( - "LiveKit URL required. Set LIVEKIT_URL environment variable " - "or pass livekit_url parameter." - ) + """Create the RTC manager. - self._livekit_api_key = livekit_api_key or getenv("LIVEKIT_API_KEY") or "" - self._livekit_api_secret = ( - livekit_api_secret or getenv("LIVEKIT_API_SECRET") or "" - ) - if not self._livekit_api_key or not self._livekit_api_secret: + Args: + on_video: Async callback invoked for each avatar video frame. + on_audio: Async callback invoked for each avatar audio chunk. + on_disconnect: Async callback invoked when the avatar leaves or the call ends. + stream_api_key: Stream API key. Uses STREAM_API_KEY env var if not provided. + stream_api_secret: Stream API secret. Uses STREAM_API_SECRET env var if not provided. + stream_call_type: Stream call type controlling the default feature set and + per-role permissions for the call. The built-in "default" type is meant + for 1:1/group video+audio calls: it enables audio, video, screensharing, + recording, HLS broadcasting, transcription and ringing, and gives + admins/hosts elevated permissions over regular participants. + """ + stream_api_key = stream_api_key or getenv("STREAM_API_KEY") + stream_api_secret = stream_api_secret or getenv("STREAM_API_SECRET") + + if not stream_api_key or not stream_api_secret: raise ValueError( - "LiveKit API key and secret required. Set LIVEKIT_API_KEY and " - "LIVEKIT_API_SECRET environment variables or pass them as parameters." + "Stream API key and secret required. Set STREAM_API_KEY and " + "STREAM_API_SECRET environment variables or pass them as parameters." ) + self._stream_api_key = stream_api_key + self._stream_api_secret = stream_api_secret + self._stream_call_type = stream_call_type + self._on_video = on_video self._on_audio = on_audio self._on_disconnect = on_disconnect - self._room: rtc.Room | None = None - self._stream_writer: rtc.ByteStreamWriter | None = None + version = get_vision_agents_version() + client_kwargs: dict[str, Any] = { + "api_key": self._stream_api_key, + "api_secret": self._stream_api_secret, + "user_agent": f"stream-vision-agents-{version}", + } + self._client = AsyncStream(**client_kwargs) + + self._plugin_user_id = f"plugin-{uuid4()}" + self._avatar_user_id = f"avatar-{uuid4()}" + + self._call: Call | None = None + self._connection: ConnectionManager | None = None + self._input_track: AvatarInputTrack | None = None self._resampler = FrameResampler( rate=_SAMPLE_RATE, layout="mono", format="s16", frame_size=0 ) self._connected = False + self._event_id = 0 self._tasks: set[asyncio.Task[None]] = set() @property def is_connected(self) -> bool: return self._connected - def generate_credentials(self) -> ConnectionCredentials: - """Generate credentials for a new LiveKit room session. + def generate_credentials(self) -> StreamConnectionCredentials: + call_id = f"lemonslice-{uuid4()}" + call_cid = f"{self._stream_call_type}:{call_id}" + avatar_token = self._client.create_call_token( + self._avatar_user_id, + call_cids=[call_cid], + expiration=3600, + ) + return StreamConnectionCredentials( + api_key=self._stream_api_key, + call_id=call_id, + call_type=self._stream_call_type, + avatar_user_id=self._avatar_user_id, + avatar_token=avatar_token, + ) - Returns: - Credentials for both the agent and the LemonSlice participant. - """ - room_name = f"lemonslice-{uuid4()}" - agent_token = self._generate_token(room_name, _PLUGIN_IDENTITY, kind="agent") - lemonslice_token = self._generate_token( - room_name, _AVATAR_IDENTITY, kind="agent" + async def connect(self, credentials: StreamConnectionCredentials) -> None: + """Join the Stream call and publish an outgoing audio track.""" + await self._client.create_user( + id=self._plugin_user_id, name=self._plugin_user_id ) - return ConnectionCredentials( - room_name=room_name, - agent_token=agent_token, - livekit_url=self._livekit_url, - livekit_token=lemonslice_token, + await self._client.create_user( + id=self._avatar_user_id, name=self._avatar_user_id ) - async def connect(self, credentials: ConnectionCredentials) -> None: - """Connect to a LiveKit room. + call = self._client.video.call(credentials.call_type, credentials.call_id) + await call.get_or_create(data={"created_by_id": self._plugin_user_id}) + self._call = call - Args: - credentials: Connection credentials from generate_credentials(). - """ - room = rtc.Room() - - @room.on("connected") - def on_connected(): - logger.info("Room connected") - - @room.on("participant_connected") - def on_participant_connected(participant: rtc.RemoteParticipant): - if participant.identity == _AVATAR_IDENTITY: - logger.info("LemonSlice avatar entered the room") - - @room.on("track_subscribed") - def on_track_subscribed( - track: rtc.Track, - publication: rtc.RemoteTrackPublication, - participant: rtc.RemoteParticipant, - ) -> None: - if participant.identity == _AVATAR_IDENTITY: - if track.kind == rtc.TrackKind.KIND_VIDEO: - logger.info("Received video track from LemonSlice") - video_stream = rtc.VideoStream(track) - self._create_task(self._consume_video(video_stream)) - elif track.kind == rtc.TrackKind.KIND_AUDIO: - logger.info("Received audio track from LemonSlice") - audio_stream = rtc.AudioStream( - track, sample_rate=48000, num_channels=2 - ) - self._create_task(self._consume_audio(audio_stream)) - - @room.on("participant_disconnected") - def on_participant_disconnected(participant: rtc.RemoteParticipant) -> None: - logger.info( - f"Participant disconnected: {participant.identity}; " - f"reason: {participant.disconnect_reason}" + subscription_config = SubscriptionConfig( + default=TrackSubscriptionConfig( + track_types=[ + StreamTrackType.TRACK_TYPE_VIDEO, + StreamTrackType.TRACK_TYPE_AUDIO, + ] ) + ) + + connection = await rtc.join( + call, + self._plugin_user_id, + subscription_config=subscription_config, + ) + self._connection = connection + + input_track = AvatarInputTrack( + sample_rate=_AVATAR_AUDIO_SAMPLE_RATE, + channels=_AVATAR_AUDIO_CHANNELS, + ) + self._input_track = input_track + + @connection.on("track_added") + async def on_track_added(track_id: str, kind: str, user: Any) -> None: + if user is None or user.user_id != self._avatar_user_id: + return + + if kind == "video": + logger.info("Received video track from LemonSlice avatar") + track = connection.subscriber_pc.add_track_subscriber(track_id) + if track is not None: + self._create_task(self._consume_video(track)) + + @connection.on("audio") + async def on_audio(pcm: PcmData) -> None: + participant = pcm.participant + if participant is None or participant.user_id != self._avatar_user_id: + return + await self._on_audio(pcm) + + @connection.on("participant_left") + async def on_participant_left(event: events_pb2.ParticipantLeft) -> None: + if event.participant.user_id != self._avatar_user_id: + return + logger.info("LemonSlice avatar left the call") self._connected = False self._create_task(self._on_disconnect()) - if self._room is not None: - self._create_task(self._room.disconnect()) - - @room.on("disconnected") - def on_disconnected(reason: str) -> None: - # The "disconnected" callback may be triggered multiple times - # because we disconnect ourselves when the avatar leaves the call. - if self._connected: - logger.info(f"Room disconnected; reason: {reason}") - self._connected = False - self._create_task(self._on_disconnect()) - - logger.info(f"Connecting to LiveKit room {credentials.room_name}") - await room.connect(self._livekit_url, credentials.agent_token) - logger.info(f"Connected to LiveKit room {credentials.room_name}") - - room.local_participant.register_rpc_method( - "lk.playback_finished", self._rpc_on_playback_finished - ) - self._room = room + @connection.on("call_ended") + async def on_call_ended(event: Any) -> None: + if not self._connected: + return + logger.info("Stream call ended") + self._connected = False + self._create_task(self._on_disconnect()) + + logger.info( + f"Joining Stream call {credentials.call_type}:{credentials.call_id}" + ) + await connection.__aenter__() + await connection.add_tracks(audio=input_track) + await connection.republish_tracks() self._connected = True + logger.info("Connected to Stream call") async def send_audio(self, pcm: PcmData) -> None: - """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) - - 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(): + """Push a PCM chunk into the outgoing audio track.""" + if self._input_track is None or not self._connected: return - - 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(_SAMPLE_RATE), - "num_channels": str(_NUM_CHANNELS), - }, - ) - logger.debug("Opened audio byte stream to LemonSlice") - - await self._stream_writer.write(frame.to_ndarray().tobytes()) + await self._input_track.write(pcm) async def flush(self) -> None: - """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 - logger.debug("Closed audio byte stream (segment end)") + """Signal end of a TTS segment to the avatar via a custom call event.""" + if self._call is None or not self._connected or self._input_track is None: + return + pts = await self._input_track.pts() + await self._call.send_call_event( + user_id=self._plugin_user_id, + custom={ + "type": _CUSTOM_EVENT_END_UTTERANCE, + "pts": pts, + "event_id": self._next_event_id(), + }, + ) async def interrupt(self) -> None: - """Send clear_buffer RPC to interrupt avatar playback.""" - if self._room is None or not self._room.isconnected(): + """Clear pending outgoing audio and signal the avatar to stop playback.""" + if self._input_track is not None: + await self._input_track.flush() + if self._call is None or not self._connected: return - try: - await self._room.local_participant.perform_rpc( - destination_identity=_AVATAR_IDENTITY, - method="lk.clear_buffer", - payload="", - ) - except rtc.RpcError: - logger.warning("clear_buffer RPC failed", exc_info=True) + await self._call.send_call_event( + user_id=self._plugin_user_id, + custom={ + "type": _CUSTOM_EVENT_INTERRUPT, + "event_id": self._next_event_id(), + }, + ) async def close(self) -> None: - """Disconnect from the LiveKit room and clean up resources.""" + """Leave the Stream call and clean up resources.""" try: - if self._stream_writer is not None: - await self._stream_writer.aclose() - await cancel_and_wait(*self._tasks) self._tasks.clear() - if self._room is not None: - await self._room.disconnect() + if self._connection is not None: + await self._connection.leave() + + if self._call is not None: + await self._call.end() + await self._client.aclose() finally: - self._room = None - self._stream_writer = None + self._connection = None + self._call = None + self._input_track = None self._connected = False - logger.debug("LemonSlice RTC manager closed") - - async def _consume_video(self, video_stream: rtc.VideoStream) -> None: - async for event in video_stream: - lk_frame = event.frame.convert(rtc.VideoBufferType.RGBA) - img = Image.frombuffer( - "RGBA", (lk_frame.width, lk_frame.height), lk_frame.data - ) - frame = av.VideoFrame.from_image(img) - await self._on_video(frame) - - async def _consume_audio(self, audio_stream: rtc.AudioStream) -> None: - async for event in audio_stream: - frame = event.frame - pcm = PcmData.from_bytes( - frame.data, # type: ignore[arg-type] - sample_rate=frame.sample_rate, - format=AudioFormat.S16, - channels=frame.num_channels, - ) - await self._on_audio(pcm) + logger.debug("LemonSlice Stream RTC manager closed") - def _rpc_on_playback_finished(self, data: rtc.RpcInvocationData) -> str: - logger.debug( - "playback finished event received", - extra={"caller_identity": data.caller_identity}, - ) - return "ok" - - def _generate_token( - self, - room_name: str, - identity: str, - kind: api.AccessToken.ParticipantKind, - ) -> str: - token = ( - api.AccessToken(self._livekit_api_key, self._livekit_api_secret) - .with_kind(kind) - .with_identity(identity) - .with_name(identity) - .with_grants( - api.VideoGrants( - room_join=True, - room=room_name, - can_publish=True, - can_subscribe=True, - ) - ) - ) - return token.to_jwt() + async def _consume_video(self, track: aiortc.mediastreams.MediaStreamTrack) -> None: + while True: + frame = await track.recv() + if isinstance(frame, av.VideoFrame): + await self._on_video(frame) def _create_task(self, coro: Coroutine[None, None, None]) -> None: task: asyncio.Task[None] = asyncio.create_task(coro) self._tasks.add(task) task.add_done_callback(self._tasks.discard) + + def _next_event_id(self) -> int: + self._event_id += 1 + return self._event_id diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py new file mode 100644 index 000000000..1b0bda360 --- /dev/null +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py @@ -0,0 +1,75 @@ +import asyncio +import fractions +import time + +import aiortc +import av +from getstream.video.rtc import AudioStreamTrack + +__all__ = ["AvatarInputTrack"] + + +class AvatarInputTrack(AudioStreamTrack): + """ + An input audio track for the LemonSlice avatar. + + Key differences from the base AudioStreamTrack: + - Tracks the last produced "pts" and exposes it as public API. + - Returns all available data on "recv()" instead of real-time pacing + to reduce avatar latency. + """ + + _timestamp: int | None + _start: float | None + _last_frame_time: float | None + + async def pts(self) -> int: + async with self._buffer_lock: + # next-to-emit PTS + samples still queued = wire PTS of the last buffered sample + ts = self._timestamp or 0 + return ts + len(self._buffer) // self._bytes_per_sample + + async def recv(self) -> av.AudioFrame: + """Drain buffered audio without pacing; pace only when emitting silence.""" + if self.readyState != "live": + raise aiortc.mediastreams.MediaStreamError + + samples_per_frame = int(aiortc.mediastreams.AUDIO_PTIME * self.sample_rate) + + if self._timestamp is None: + self._start = time.time() + timestamp = 0 + else: + timestamp = self._timestamp + samples_per_frame + self._timestamp = timestamp + + async with self._buffer_lock: + if len(self._buffer) >= self._bytes_per_frame: + audio_bytes = bytes(self._buffer[: self._bytes_per_frame]) + del self._buffer[: self._bytes_per_frame] + has_data = True + elif len(self._buffer) > 0: + audio_bytes = bytes(self._buffer) + audio_bytes += bytes(self._bytes_per_frame - len(audio_bytes)) + self._buffer.clear() + has_data = True + else: + audio_bytes = bytes(self._bytes_per_frame) + has_data = False + + if not has_data: + # Per-frame sleep keeps silence at the frame cadence even when PTS runs ahead of wall clock after a burst. + await asyncio.sleep(aiortc.mediastreams.AUDIO_PTIME) + + self._last_frame_time = time.time() + + layout = "stereo" if self.channels == 2 else "mono" + av_format = "flt" if self.format == "f32" else "s16" + frame = av.AudioFrame( + format=av_format, layout=layout, samples=samples_per_frame + ) + frame.planes[0].update(audio_bytes) + frame.pts = timestamp + frame.sample_rate = self.sample_rate + frame.time_base = fractions.Fraction(1, self.sample_rate) + return frame diff --git a/uv.lock b/uv.lock index d0db9afe3..c8732c890 100644 --- a/uv.lock +++ b/uv.lock @@ -369,7 +369,7 @@ name = "apache-tvm-ffi" version = "0.1.12" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions" }, + { name = "typing-extensions", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ff/95/ef83880657e89a0ce0f1ad79cbff11698286d00522dbc290d34a8458e9c2/apache_tvm_ffi-0.1.12.tar.gz", hash = "sha256:2aa5c8ece3144dad11afd6d0f10191d03cdb368bbcd9c92f9fb919f35906223d", size = 2843816, upload-time = "2026-06-09T18:17:31.68Z" } wheels = [ @@ -1107,7 +1107,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" }, @@ -1134,34 +1134,34 @@ wheels = [ [package.optional-dependencies] cudart = [ - { name = "nvidia-cuda-runtime" }, + { name = "nvidia-cuda-runtime", marker = "sys_platform == 'linux'" }, ] cufft = [ - { name = "nvidia-cufft" }, + { name = "nvidia-cufft", marker = "sys_platform == 'linux'" }, ] cufile = [ - { name = "nvidia-cufile" }, + { name = "nvidia-cufile", marker = "sys_platform == 'linux'" }, ] cupti = [ - { name = "nvidia-cuda-cupti" }, + { name = "nvidia-cuda-cupti", marker = "sys_platform == 'linux'" }, ] curand = [ - { name = "nvidia-curand" }, + { name = "nvidia-curand", marker = "sys_platform == 'linux'" }, ] cusolver = [ - { name = "nvidia-cusolver" }, + { name = "nvidia-cusolver", marker = "sys_platform == 'linux'" }, ] cusparse = [ - { name = "nvidia-cusparse" }, + { name = "nvidia-cusparse", marker = "sys_platform == 'linux'" }, ] nvjitlink = [ - { name = "nvidia-nvjitlink" }, + { name = "nvidia-nvjitlink", marker = "sys_platform == 'linux'" }, ] nvrtc = [ - { name = "nvidia-cuda-nvrtc" }, + { name = "nvidia-cuda-nvrtc", marker = "sys_platform == 'linux'" }, ] nvtx = [ - { name = "nvidia-nvtx" }, + { name = "nvidia-nvtx", marker = "sys_platform == 'linux'" }, ] [[package]] @@ -2487,16 +2487,16 @@ name = "kestrel" version = "0.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "apache-tvm-ffi" }, - { name = "httpx" }, - { name = "huggingface-hub" }, - { name = "kestrel-native" }, - { name = "safetensors" }, - { name = "starlette" }, - { name = "tokenizers" }, - { name = "torch-c-dlpack-ext" }, - { name = "transformers" }, - { name = "uvicorn" }, + { name = "apache-tvm-ffi", marker = "sys_platform == 'win32'" }, + { name = "httpx", marker = "sys_platform == 'win32'" }, + { name = "huggingface-hub", marker = "sys_platform == 'win32'" }, + { name = "kestrel-native", marker = "sys_platform == 'win32'" }, + { name = "safetensors", marker = "sys_platform == 'win32'" }, + { name = "starlette", marker = "sys_platform == 'win32'" }, + { name = "tokenizers", marker = "sys_platform == 'win32'" }, + { name = "torch-c-dlpack-ext", marker = "sys_platform == 'win32'" }, + { name = "transformers", marker = "sys_platform == 'win32'" }, + { name = "uvicorn", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/9a/4c/130645e9115d5b12798e9e8882cba602435a55ddfaa50796a40153291ba1/kestrel-0.2.0.tar.gz", hash = "sha256:8b7295036939c238717496925ebc0f34b7edc4aac7d0db67fe7e0ed54ea9ee5e", size = 146019, upload-time = "2026-03-18T11:04:56.716Z" } wheels = [ @@ -2508,7 +2508,7 @@ name = "kestrel-native" version = "0.1.3" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "numpy" }, + { name = "numpy", marker = "sys_platform == 'win32'" }, ] wheels = [ { url = "https://files.pythonhosted.org/packages/c7/e1/13f99f087254019c06c2b03c0b43edd4f3e8f1b8e6d3664f2d3945b8e27b/kestrel_native-0.1.3-cp312-cp312-win_amd64.whl", hash = "sha256:8da9dc66e2196c8bd583fc2f80c133c64eef9f0bc22fbbb90d8ff4e4d5fd0729", size = 2001859, upload-time = "2026-02-20T16:44:28.222Z" }, @@ -2800,35 +2800,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/da/ce/a3d3e0566dbd2586c325240d44afec6d44421eb794bcf8dbaca15463a7b7/livekit-1.1.13-py3-none-win_amd64.whl", hash = "sha256:22dff7a39cb3d590a4757e20d3ce5d326ab882350386f5070f45cbd6d9ccf839", size = 10717013, upload-time = "2026-06-30T11:53:57.701Z" }, ] -[[package]] -name = "livekit-api" -version = "1.1.1" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "aiohttp" }, - { name = "livekit-protocol" }, - { name = "protobuf" }, - { name = "pyjwt" }, - { name = "types-protobuf" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/f8/03/00e0ec173f247e1f7ea63cb5591d5680a64c7a74ea4d5d558e5aed6cc399/livekit_api-1.1.1.tar.gz", hash = "sha256:70c7b80eecbc297b40756ebd76e4f52d00b0348fb7d212a21c1f69cc57fd9c83", size = 15196, upload-time = "2026-06-24T01:36:19.686Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/1e/c0/d5f3ff74ab5db2d06f173801ec934885d11a754b9fb9ad768c8ede0a6c89/livekit_api-1.1.1-py3-none-any.whl", hash = "sha256:ce8c327676c366e66cf68782934368dd0ba92b9d48f578275227e255c890fe88", size = 19471, upload-time = "2026-06-24T01:36:18.42Z" }, -] - -[[package]] -name = "livekit-protocol" -version = "1.1.18" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "protobuf" }, - { name = "types-protobuf" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/e7/88/64f2be01a630e249f1dbd0d51876f109b53b7899ae41246d2ca5b647086d/livekit_protocol-1.1.18.tar.gz", hash = "sha256:187af32ebf75333a62117b0db9e551c99060bd4e1f57cfc0fce73bcd7a671da8", size = 115802, upload-time = "2026-06-27T15:31:04.102Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/14/80/9cc33e4d0280132538850aaf9559d6b8aa9e670c5917f75dab996400ab84/livekit_protocol-1.1.18-py3-none-any.whl", hash = "sha256:30c539410fd3cfc2e551ca3a193aaaaacaaec6dd57dabe2c9be7c7c7d15f0e01", size = 143134, upload-time = "2026-06-27T15:31:02.686Z" }, -] - [[package]] name = "llvmlite" version = "0.47.0" @@ -3099,7 +3070,7 @@ resolution-markers = [ "python_full_version < '3.13' and sys_platform != 'win32'", ] dependencies = [ - { name = "pillow" }, + { name = "pillow", marker = "sys_platform != 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a5/d7/85e4d020c4d00f4842b35773e4442fe5cea310e4ebc6a1856e55d3e1a658/moondream-0.2.0.tar.gz", hash = "sha256:402655cc23b94490512caa1cf9f250fc34d133dfdbac201f78b32cbdeabdae0d", size = 97837, upload-time = "2025-11-25T18:22:04.477Z" } wheels = [ @@ -3115,8 +3086,8 @@ resolution-markers = [ "python_full_version < '3.13' and sys_platform == 'win32'", ] dependencies = [ - { name = "kestrel" }, - { name = "pillow" }, + { name = "kestrel", marker = "sys_platform == 'win32'" }, + { name = "pillow", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/13/b9/7e2b5704a580c44eca8ed8062b4bbbb4bcd230d38c192e432548bc915b62/moondream-0.2.1.tar.gz", hash = "sha256:23ce9a33118b9bced38ada63e27f1c49ec735b9e851f85f90b0f3778237c624a", size = 100001, upload-time = "2026-03-18T13:12:02.775Z" } wheels = [ @@ -3390,7 +3361,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" }, @@ -3429,7 +3400,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" }, @@ -3441,7 +3412,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" }, @@ -3471,9 +3442,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" }, @@ -3485,7 +3456,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" }, @@ -5274,8 +5245,8 @@ name = "secretstorage" version = "3.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "cryptography" }, - { name = "jeepney" }, + { name = "cryptography", marker = "sys_platform != 'win32'" }, + { name = "jeepney", marker = "sys_platform != 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1c/03/e834bcd866f2f8a49a85eaff47340affa3bfa391ee9912a952a1faa68c7b/secretstorage-3.5.0.tar.gz", hash = "sha256:f04b8e4689cbce351744d5537bf6b1329c6fc68f91fa666f60a380edddcd11be", size = 19884, upload-time = "2025-11-23T19:02:53.191Z" } wheels = [ @@ -5688,8 +5659,8 @@ name = "standard-aifc" version = "3.13.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "audioop-lts" }, - { name = "standard-chunk" }, + { name = "audioop-lts", marker = "python_full_version >= '3.13'" }, + { name = "standard-chunk", marker = "python_full_version >= '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/c4/53/6050dc3dde1671eb3db592c13b55a8005e5040131f7509cef0215212cb84/standard_aifc-3.13.0.tar.gz", hash = "sha256:64e249c7cb4b3daf2fdba4e95721f811bde8bdfc43ad9f936589b7bb2fae2e43", size = 15240, upload-time = "2024-10-30T16:01:31.772Z" } wheels = [ @@ -5710,7 +5681,7 @@ name = "standard-sunau" version = "3.13.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "audioop-lts" }, + { name = "audioop-lts", marker = "python_full_version >= '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/66/e3/ce8d38cb2d70e05ffeddc28bb09bad77cfef979eb0a299c9117f7ed4e6a9/standard_sunau-3.13.0.tar.gz", hash = "sha256:b319a1ac95a09a2378a8442f403c66f4fd4b36616d6df6ae82b8e536ee790908", size = 9368, upload-time = "2024-10-30T16:01:41.626Z" } wheels = [ @@ -6003,7 +5974,7 @@ name = "torch-c-dlpack-ext" version = "0.1.5" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "torch" }, + { name = "torch", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/37/de/921b6491efce5c389a5ef9bbed3d2d6660005840dae488124173180859ab/torch_c_dlpack_ext-0.1.5.tar.gz", hash = "sha256:d06f0357d575d22a168cc77acb9020fc4bae30968ceb6718a055dcbe92bacabe", size = 12913, upload-time = "2026-01-12T11:25:08.484Z" } wheels = [ @@ -7073,8 +7044,6 @@ name = "vision-agents-plugins-lemonslice" source = { editable = "plugins/lemonslice" } dependencies = [ { name = "httpx" }, - { name = "livekit" }, - { name = "livekit-api" }, { name = "vision-agents" }, ] @@ -7087,8 +7056,6 @@ dev = [ [package.metadata] requires-dist = [ { name = "httpx", specifier = ">=0.28.1,<1" }, - { name = "livekit", specifier = ">=1.1.2,<2" }, - { name = "livekit-api", specifier = ">=1.1.0,<2" }, { name = "vision-agents", editable = "agents-core" }, ] From 3d13915cfc9195e53485ee9948ec304073598797 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Thu, 28 May 2026 13:34:54 +0200 Subject: [PATCH 02/11] Document Stream call types and how they are supposed to be used with LemonSlice Avatar --- plugins/lemonslice/README.md | 33 +++++++++++++++++++ .../plugins/lemonslice/lemonslice_avatar.py | 5 +++ .../lemonslice/lemonslice_rtc_manager.py | 21 +++++++++++- 3 files changed, 58 insertions(+), 1 deletion(-) diff --git a/plugins/lemonslice/README.md b/plugins/lemonslice/README.md index a0fdcf293..4c8010341 100644 --- a/plugins/lemonslice/README.md +++ b/plugins/lemonslice/README.md @@ -94,6 +94,39 @@ lemonslice.Avatar( 3. **Avatar Generation**: LemonSlice generates synchronized avatar video and audio 4. **Video Streaming**: Avatar video is streamed to call participants via GetStream Edge +## Custom Stream Call Type (recommended) + +The plugin runs its own internal Stream call as a bridge between your process and the LemonSlice avatar service — this is separate from the user-facing call the agent joins. Only two users ever need to be on the bridge call: the plugin user and the avatar user. We recommend passing a custom `stream_call_type` whose permissions allow **only those two** to join, so no other token-holder in your app can accidentally enter the bridge. + +Reference docs: +- [Built-in call types](https://getstream.io/video/docs/api/call_types/builtin/) +- [Managing call types](https://getstream.io/video/docs/api/call_types/manage/) +- [Permissions & capabilities](https://getstream.io/video/docs/api/call_types/permissions/) + +The plugin attaches both users to the bridge call as members with `role="call_member"`. Configure your custom call type so the `call_member` role has exactly the capabilities the plugin needs — and no other role has `join-call`: + +```python +client.video.create_call_type( + name="lemonslice_bridge", + grants={ + # plugin + avatar — everything they need to bridge audio/video + "call_member": ["join-call", "read-call", "send-audio", "send-video"], + # everyone else — denied + "user": [], + "admin": [], + "host": [], + "moderator": [], + }, +) + +lemonslice.Avatar( + agent_id="your-avatar-id", + stream_call_type="lemonslice_bridge", +) +``` + +If you stick with the default `"default"` call type the plugin still works, but the bridge call uses the same broad permissions as any default Stream call. + ## Requirements - Python 3.10+ diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py index 145c977fd..992c01d27 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py @@ -73,6 +73,11 @@ def __init__( for 1:1/group video+audio calls: it enables audio, video, screensharing, recording, HLS broadcasting, transcription and ringing, and gives admins/hosts elevated permissions over regular participants. + If you pass a custom call type, it must grant the `call_member` role + the `join-call`, `read-call`, `send-audio`, and `send-video` + capabilities — the plugin and avatar users are attached as members + with that role so they can join regardless of the type's default + user-role grants. width: Output video width in pixels. height: Output video height in pixels. fps: Output video frame rate. Must be > 0. 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 1f36be800..e6fc81d2c 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py @@ -7,6 +7,7 @@ import av from getstream import AsyncStream +from getstream.models import CallRequest, MemberRequest from getstream.video import rtc from getstream.video.async_call import Call from getstream.video.rtc.connection_manager import ConnectionManager @@ -67,6 +68,13 @@ def __init__( for 1:1/group video+audio calls: it enables audio, video, screensharing, recording, HLS broadcasting, transcription and ringing, and gives admins/hosts elevated permissions over regular participants. + If you pass a custom call type, it must grant the `call_member` role + the `join-call`, `read-call`, `send-audio`, and `send-video` + capabilities — the plugin and avatar users are attached as members + with that role so they can join regardless of the type's default + user-role grants. + See https://getstream.io/video/docs/api/call_types/builtin/ and + https://getstream.io/video/docs/api/call_types/permissions/. """ stream_api_key = stream_api_key or getenv("STREAM_API_KEY") stream_api_secret = stream_api_secret or getenv("STREAM_API_SECRET") @@ -136,7 +144,18 @@ async def connect(self, credentials: StreamConnectionCredentials) -> None: ) call = self._client.video.call(credentials.call_type, credentials.call_id) - await call.get_or_create(data={"created_by_id": self._plugin_user_id}) + # Attach plugin + avatar users as members so they can join regardless of + # the call type's per-role grants. See README for the contract on custom + # call types. + await call.get_or_create( + data=CallRequest( + created_by_id=self._plugin_user_id, + members=[ + MemberRequest(user_id=self._plugin_user_id, role="call_member"), + MemberRequest(user_id=self._avatar_user_id, role="call_member"), + ], + ) + ) self._call = call subscription_config = SubscriptionConfig( From 687fea9049db51a96580903ffff9108bbc70e428 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Thu, 6 Aug 2026 15:53:02 +0200 Subject: [PATCH 03/11] lemonslice.Avatar: fixes after rebase on main --- .../lemonslice/lemonslice_rtc_manager.py | 10 ++- .../vision_agents/plugins/lemonslice/track.py | 63 ++++++++----------- 2 files changed, 33 insertions(+), 40 deletions(-) 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 e6fc81d2c..e8fddb0c3 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py @@ -5,13 +5,19 @@ from typing import Any, Callable, Coroutine from uuid import uuid4 +import aiortc import av from getstream import AsyncStream from getstream.models import CallRequest, MemberRequest from getstream.video import rtc from getstream.video.async_call import Call from getstream.video.rtc.connection_manager import ConnectionManager -from getstream.video.rtc.track_util import AudioFormat, FrameResampler, PcmData +from getstream.video.rtc.pb.stream.video.sfu.event import events_pb2 +from getstream.video.rtc.pb.stream.video.sfu.models.models_pb2 import ( + TrackType as StreamTrackType, +) +from getstream.video.rtc.track_util import FrameResampler, PcmData +from getstream.video.rtc.tracks import SubscriptionConfig, TrackSubscriptionConfig from vision_agents.core.utils.utils import cancel_and_wait, get_vision_agents_version from .track import AvatarInputTrack @@ -108,7 +114,7 @@ def __init__( self._connection: ConnectionManager | None = None self._input_track: AvatarInputTrack | None = None self._resampler = FrameResampler( - rate=_SAMPLE_RATE, layout="mono", format="s16", frame_size=0 + rate=_AVATAR_AUDIO_SAMPLE_RATE, layout="mono", format="s16", frame_size=0 ) self._connected = False self._event_id = 0 diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py index 1b0bda360..bdc790833 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py @@ -1,6 +1,5 @@ import asyncio import fractions -import time import aiortc import av @@ -19,57 +18,45 @@ class AvatarInputTrack(AudioStreamTrack): to reduce avatar latency. """ - _timestamp: int | None - _start: float | None - _last_frame_time: float | None + # PTS of the last emitted frame; None until the first recv(). + _timestamp: int | None = None async def pts(self) -> int: - async with self._buffer_lock: - # next-to-emit PTS + samples still queued = wire PTS of the last buffered sample + async with self._frame_lock: + # last emitted PTS + samples still queued = wire PTS of the last buffered sample ts = self._timestamp or 0 - return ts + len(self._buffer) // self._bytes_per_sample + return ts + self._buffered_samples async def recv(self) -> av.AudioFrame: """Drain buffered audio without pacing; pace only when emitting silence.""" if self.readyState != "live": raise aiortc.mediastreams.MediaStreamError - samples_per_frame = int(aiortc.mediastreams.AUDIO_PTIME * self.sample_rate) - if self._timestamp is None: - self._start = time.time() - timestamp = 0 + self._timestamp = 0 else: - timestamp = self._timestamp + samples_per_frame - self._timestamp = timestamp - - async with self._buffer_lock: - if len(self._buffer) >= self._bytes_per_frame: - audio_bytes = bytes(self._buffer[: self._bytes_per_frame]) - del self._buffer[: self._bytes_per_frame] - has_data = True - elif len(self._buffer) > 0: - audio_bytes = bytes(self._buffer) - audio_bytes += bytes(self._bytes_per_frame - len(audio_bytes)) - self._buffer.clear() - has_data = True - else: - audio_bytes = bytes(self._bytes_per_frame) - has_data = False - - if not has_data: + self._timestamp += self._samples_per_frame + + async with self._frame_lock: + if not self._frame_buffer: + # Starved: emit the resampler's partial tail instead of waiting for a full frame. + for tail in self._resampler.flush(): + self._frame_buffer.append(tail) + self._buffered_samples += tail.samples + frame = self._frame_buffer.popleft() if self._frame_buffer else None + if frame is not None: + self._buffered_samples -= frame.samples + + if frame is None: # Per-frame sleep keeps silence at the frame cadence even when PTS runs ahead of wall clock after a burst. await asyncio.sleep(aiortc.mediastreams.AUDIO_PTIME) + frame = av.AudioFrame.from_ndarray( + self._silence, format="s16", layout=self._layout + ) + elif frame.samples < self._samples_per_frame: + frame = self._pad_to_full_frame(frame) - self._last_frame_time = time.time() - - layout = "stereo" if self.channels == 2 else "mono" - av_format = "flt" if self.format == "f32" else "s16" - frame = av.AudioFrame( - format=av_format, layout=layout, samples=samples_per_frame - ) - frame.planes[0].update(audio_bytes) - frame.pts = timestamp + frame.pts = self._timestamp frame.sample_rate = self.sample_rate frame.time_base = fractions.Fraction(1, self.sample_rate) return frame From a5384a263ca0efbf7171c7a1d28bbb13f0d9f55b Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 10 Aug 2026 15:11:57 +0200 Subject: [PATCH 04/11] Remove pacing and silence padding from AvatarInputTrack --- plugins/lemonslice/tests/test_track.py | 111 ++++++++---------- .../vision_agents/plugins/lemonslice/track.py | 74 +++++++----- 2 files changed, 94 insertions(+), 91 deletions(-) diff --git a/plugins/lemonslice/tests/test_track.py b/plugins/lemonslice/tests/test_track.py index e1dde6a02..68fb8de4f 100644 --- a/plugins/lemonslice/tests/test_track.py +++ b/plugins/lemonslice/tests/test_track.py @@ -1,3 +1,4 @@ +import asyncio import time import numpy as np @@ -34,16 +35,14 @@ async def test_pts_reflects_buffered_samples_before_any_recv( await track.write(_audio(_SAMPLES_PER_FRAME * 3)) assert await track.pts() == _SAMPLES_PER_FRAME * 3 - async def test_pts_after_emission_equals_last_buffered_frame_pts( - self, track: AvatarInputTrack - ) -> None: - # Once recv() has begun, pts() = _timestamp + queued_samples = the wire - # PTS the last buffered frame will carry when it's emitted. + async def test_pts_unchanged_by_draining(self, track: AvatarInputTrack) -> None: + # pts() is the wire PTS the track will have reached once the buffer drains, + # so writing moves it and recv() does not. await track.write(_audio(_SAMPLES_PER_FRAME * 3)) - await track.recv() # _timestamp=0, 2 frames queued - assert await track.pts() == _SAMPLES_PER_FRAME * 2 - await track.recv() # _timestamp=320, 1 frame queued - assert await track.pts() == _SAMPLES_PER_FRAME * 2 + await track.recv() + assert await track.pts() == _SAMPLES_PER_FRAME * 3 + await track.recv() + assert await track.pts() == _SAMPLES_PER_FRAME * 3 async def test_recv_frame_shape(self, track: AvatarInputTrack) -> None: await track.write(_audio(_SAMPLES_PER_FRAME)) @@ -62,78 +61,68 @@ async def test_pts_advances_by_samples_per_frame( async def test_burst_drains_without_pacing(self, track: AvatarInputTrack) -> None: # 1 second of audio = 50 frames; under wall-clock pacing this would take ~1s. await track.write(_audio(_SAMPLE_RATE)) - await track.recv() # first recv initializes the timestamp baseline start = time.monotonic() - for _ in range(49): + for _ in range(50): await track.recv() elapsed = time.monotonic() - start - assert elapsed < 49 * AUDIO_PTIME * 0.5, ( - f"burst was paced: {elapsed:.4f}s for 49 frames" + assert elapsed < 50 * AUDIO_PTIME * 0.5, ( + f"burst was paced: {elapsed:.4f}s for 50 frames" ) - async def test_silence_paces_at_frame_interval( + async def test_recv_waits_for_audio_instead_of_emitting_silence( self, track: AvatarInputTrack ) -> None: - await track.recv() # init (first silence frame, also sleeps once) - start = time.monotonic() - await track.recv() - elapsed = time.monotonic() - start - assert AUDIO_PTIME * 0.5 < elapsed < AUDIO_PTIME * 3, ( - f"silence pacing off: {elapsed:.4f}s vs target {AUDIO_PTIME:.4f}s" - ) + pending = asyncio.create_task(track.recv()) + await asyncio.sleep(AUDIO_PTIME * 3) + assert not pending.done(), "recv() emitted a frame while starved" + + await track.write(_audio(_SAMPLES_PER_FRAME)) + frame = await asyncio.wait_for(pending, timeout=1) + assert frame.samples == _SAMPLES_PER_FRAME + assert frame.pts == 0 - async def test_silence_after_burst_does_not_sleep_through_backlog( + async def test_pts_contiguous_across_idle_gap( self, track: AvatarInputTrack ) -> None: - # The drift edge case: PTS runs ~1s ahead of wall clock after the burst. - # If we used the parent's wall-clock-anchored sleep, the next silence - # frame would sleep ~1s to "catch up". We must sleep only AUDIO_PTIME. - await track.write(_audio(_SAMPLE_RATE)) # 50 frames - for _ in range(50): - await track.recv() - start = time.monotonic() - frame = await track.recv() - elapsed = time.monotonic() - start - assert elapsed < AUDIO_PTIME * 5, ( - f"silence after burst slept through backlog: {elapsed:.4f}s" - ) - assert frame.pts == 50 * _SAMPLES_PER_FRAME + # An idle gap costs no PTS: the next utterance continues the timeline. + await track.write(_audio(_SAMPLES_PER_FRAME * 2)) + first = [(await track.recv()).pts for _ in range(2)] + await asyncio.sleep(AUDIO_PTIME * 3) + await track.write(_audio(_SAMPLES_PER_FRAME)) + resumed = await track.recv() + assert first == [0, _SAMPLES_PER_FRAME] + assert resumed.pts == _SAMPLES_PER_FRAME * 2 - async def test_partial_frame_is_padded_and_bursted( + async def test_partial_tail_is_emitted_unpadded( self, track: AvatarInputTrack ) -> None: await track.write(_audio(_SAMPLES_PER_FRAME // 2)) - start = time.monotonic() frame = await track.recv() - elapsed = time.monotonic() - start - assert elapsed < AUDIO_PTIME * 0.5, f"partial frame was paced: {elapsed:.4f}s" - assert frame.samples == _SAMPLES_PER_FRAME + assert frame.samples == _SAMPLES_PER_FRAME // 2 assert frame.pts == 0 - async def test_pts_monotonic_across_data_silence_data_transitions( + async def test_pts_advances_by_actual_samples_after_a_partial_tail( self, track: AvatarInputTrack ) -> None: - # data, then silence, then data — PTS continues advancing by samples_per_frame each step. - await track.write(_audio(_SAMPLES_PER_FRAME * 2)) - f0 = await track.recv() - f1 = await track.recv() - f2 = await track.recv() # silence + # A short tail must not consume a full frame slot, or the timeline gains a hole. + await track.write(_audio(_SAMPLES_PER_FRAME // 2)) + tail = await track.recv() await track.write(_audio(_SAMPLES_PER_FRAME)) - f3 = await track.recv() # data again - assert [f0.pts, f1.pts, f2.pts, f3.pts] == [ - i * _SAMPLES_PER_FRAME for i in range(4) - ] + following = await track.recv() + assert tail.pts == 0 + assert following.pts == _SAMPLES_PER_FRAME // 2 - async def test_long_silence_run_does_not_drift_to_zero_or_negative_sleep( + async def test_interrupt_drops_pending_audio_and_keeps_waiting( self, track: AvatarInputTrack ) -> None: - # Many silence frames in a row — each should still sleep ~AUDIO_PTIME, - # not zero (no drift compensation collapsing the interval). - await track.recv() # init - start = time.monotonic() - for _ in range(5): - await track.recv() - elapsed = time.monotonic() - start - assert elapsed > 5 * AUDIO_PTIME * 0.5, ( - f"silence loop did not pace: {elapsed:.4f}s for 5 frames" - ) + await track.write(_audio(_SAMPLES_PER_FRAME * 3)) + await track.recv() + await track.flush() + + pending = asyncio.create_task(track.recv()) + await asyncio.sleep(AUDIO_PTIME * 3) + assert not pending.done(), "flushed frames were still emitted" + + await track.write(_audio(_SAMPLES_PER_FRAME)) + frame = await asyncio.wait_for(pending, timeout=1) + assert frame.pts == _SAMPLES_PER_FRAME diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py index bdc790833..147b9ac5b 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py @@ -4,6 +4,7 @@ import aiortc import av from getstream.video.rtc import AudioStreamTrack +from getstream.video.rtc.track_util import AudioFormat, AudioFormatType, PcmData __all__ = ["AvatarInputTrack"] @@ -16,47 +17,60 @@ class AvatarInputTrack(AudioStreamTrack): - Tracks the last produced "pts" and exposes it as public API. - Returns all available data on "recv()" instead of real-time pacing to reduce avatar latency. + - Waits for the next write instead of synthesizing silence: an avatar has + nothing to animate during a gap, and the PTS timeline stays contiguous + across it. """ - # PTS of the last emitted frame; None until the first recv(). - _timestamp: int | None = None + def __init__( + self, + sample_rate: int = 48000, + channels: int = 1, + format: AudioFormatType = AudioFormat.S16, + audio_buffer_size_ms: int = 30000, + ): + super().__init__( + sample_rate=sample_rate, + channels=channels, + format=format, + audio_buffer_size_ms=audio_buffer_size_ms, + ) + # PTS the next emitted frame will carry, in samples. + self._next_pts = 0 + self._data_available = asyncio.Event() async def pts(self) -> int: async with self._frame_lock: - # last emitted PTS + samples still queued = wire PTS of the last buffered sample - ts = self._timestamp or 0 - return ts + self._buffered_samples + # next PTS to emit + samples still queued = wire PTS once the buffer drains + return self._next_pts + self._buffered_samples + + async def write(self, pcm: PcmData, final: bool = False) -> None: + await super().write(pcm, final) + self._data_available.set() async def recv(self) -> av.AudioFrame: - """Drain buffered audio without pacing; pace only when emitting silence.""" + """Drain buffered audio without pacing, waiting for more instead of emitting silence.""" if self.readyState != "live": raise aiortc.mediastreams.MediaStreamError - if self._timestamp is None: - self._timestamp = 0 - else: - self._timestamp += self._samples_per_frame + while True: + async with self._frame_lock: + if not self._frame_buffer: + # Starved: emit the resampler's partial tail instead of waiting for a full frame. + for tail in self._resampler.flush(): + self._frame_buffer.append(tail) + self._buffered_samples += tail.samples + if self._frame_buffer: + frame = self._frame_buffer.popleft() + self._buffered_samples -= frame.samples + break + self._data_available.clear() + await self._data_available.wait() - async with self._frame_lock: - if not self._frame_buffer: - # Starved: emit the resampler's partial tail instead of waiting for a full frame. - for tail in self._resampler.flush(): - self._frame_buffer.append(tail) - self._buffered_samples += tail.samples - frame = self._frame_buffer.popleft() if self._frame_buffer else None - if frame is not None: - self._buffered_samples -= frame.samples - - if frame is None: - # Per-frame sleep keeps silence at the frame cadence even when PTS runs ahead of wall clock after a burst. - await asyncio.sleep(aiortc.mediastreams.AUDIO_PTIME) - frame = av.AudioFrame.from_ndarray( - self._silence, format="s16", layout=self._layout - ) - elif frame.samples < self._samples_per_frame: - frame = self._pad_to_full_frame(frame) - - frame.pts = self._timestamp + frame.pts = self._next_pts + # Advance by the samples actually emitted; a short tail must not consume a + # full frame slot, or the encoder's resampler pads the gap with silence. + self._next_pts += frame.samples frame.sample_rate = self.sample_rate frame.time_base = fractions.Fraction(1, self.sample_rate) return frame From 0d5a0de80c11f42d82d14e21b86cbd1fc28a02ad Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 10 Aug 2026 16:40:54 +0200 Subject: [PATCH 05/11] Wait until the Lemonslice avatar participant joins the call before sending audio (default: 30s) --- plugins/lemonslice/README.md | 3 ++- .../lemonslice/tests/test_lemonslice_plugin.py | 1 + .../plugins/lemonslice/lemonslice_avatar.py | 5 +++++ .../lemonslice/lemonslice_rtc_manager.py | 18 ++++++++++++++++++ 4 files changed, 26 insertions(+), 1 deletion(-) diff --git a/plugins/lemonslice/README.md b/plugins/lemonslice/README.md index 4c8010341..78a41b8bd 100644 --- a/plugins/lemonslice/README.md +++ b/plugins/lemonslice/README.md @@ -84,12 +84,13 @@ lemonslice.Avatar( height=720, # Output video height in pixels fps=30, # Output video frame rate buffer_seconds=1.0, # Max video buffer depth in seconds + avatar_join_timeout=30.0, # Seconds to wait for the avatar to join the bridge call ) ``` ## How It Works -1. **LemonSlice Session**: Creates a session via LemonSlice API, and joins the Stream call as a participant +1. **LemonSlice Session**: Creates a session via LemonSlice API, and joins the Stream call as a participant. `agent.join()` blocks until the avatar is on the call, so no audio is sent before it can receive it — if the avatar does not show up within `avatar_join_timeout`, the connection is torn down and the error is raised 2. **Audio Forwarding**: TTS audio is captured and sent to LemonSlice via the Stream call 3. **Avatar Generation**: LemonSlice generates synchronized avatar video and audio 4. **Video Streaming**: Avatar video is streamed to call participants via GetStream Edge diff --git a/plugins/lemonslice/tests/test_lemonslice_plugin.py b/plugins/lemonslice/tests/test_lemonslice_plugin.py index ac0d5db99..365c5f150 100644 --- a/plugins/lemonslice/tests/test_lemonslice_plugin.py +++ b/plugins/lemonslice/tests/test_lemonslice_plugin.py @@ -7,6 +7,7 @@ def _make_avatar(**overrides) -> LemonSliceAvatar: default_kwargs = { "agent_id": "test-agent", + "api_key": "lemonslice-key", "stream_api_key": "key", "stream_api_secret": "secret", } diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py index 992c01d27..657b00fb7 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py @@ -56,6 +56,7 @@ def __init__( height: int = 720, fps: int = 30, buffer_seconds: float = 1.0, + avatar_join_timeout: float = 30.0, ): """Initialize the LemonSlice avatar publisher. @@ -83,6 +84,8 @@ def __init__( fps: Output video frame rate. Must be > 0. buffer_seconds: Max video buffer depth in seconds. Caps how many frames can be queued ahead of audio playback. Must be > 0. + avatar_join_timeout: Seconds to wait for the avatar participant to join + the call before failing the connection. """ super().__init__() if buffer_seconds <= 0: @@ -110,6 +113,7 @@ def __init__( stream_api_secret=stream_api_secret, stream_api_key=stream_api_key, stream_call_type=stream_call_type, + avatar_join_timeout=avatar_join_timeout, ) self._sync = AVSynchronizer( width=width, @@ -185,6 +189,7 @@ async def _connect(self) -> None: token=credentials.avatar_token, api_key=credentials.api_key, ) + await self._rtc_manager.wait_for_avatar() except Exception: await self._rtc_manager.close() raise 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 e8fddb0c3..af339c8bb 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py @@ -60,6 +60,7 @@ def __init__( stream_api_key: str | None = None, stream_api_secret: str | None = None, stream_call_type: str = "default", + avatar_join_timeout: float = 30.0, ): """Create the RTC manager. @@ -81,6 +82,8 @@ def __init__( user-role grants. See https://getstream.io/video/docs/api/call_types/builtin/ and https://getstream.io/video/docs/api/call_types/permissions/. + avatar_join_timeout: Seconds to wait for the avatar participant to join + the call before giving up. """ stream_api_key = stream_api_key or getenv("STREAM_API_KEY") stream_api_secret = stream_api_secret or getenv("STREAM_API_SECRET") @@ -94,6 +97,7 @@ def __init__( self._stream_api_key = stream_api_key self._stream_api_secret = stream_api_secret self._stream_call_type = stream_call_type + self._avatar_join_timeout = avatar_join_timeout self._on_video = on_video self._on_audio = on_audio @@ -117,6 +121,7 @@ def __init__( rate=_AVATAR_AUDIO_SAMPLE_RATE, layout="mono", format="s16", frame_size=0 ) self._connected = False + self._avatar_joined = asyncio.Event() self._event_id = 0 self._tasks: set[asyncio.Task[None]] = set() @@ -204,6 +209,13 @@ async def on_audio(pcm: PcmData) -> None: return await self._on_audio(pcm) + @connection.on("participant_joined") + async def on_participant_joined(event: events_pb2.ParticipantJoined) -> None: + if event.participant.user_id != self._avatar_user_id: + return + logger.info("LemonSlice avatar joined the call") + self._avatar_joined.set() + @connection.on("participant_left") async def on_participant_left(event: events_pb2.ParticipantLeft) -> None: if event.participant.user_id != self._avatar_user_id: @@ -229,6 +241,12 @@ async def on_call_ended(event: Any) -> None: self._connected = True logger.info("Connected to Stream call") + async def wait_for_avatar(self) -> None: + """Block until the avatar participant joins the call.""" + await asyncio.wait_for( + self._avatar_joined.wait(), timeout=self._avatar_join_timeout + ) + async def send_audio(self, pcm: PcmData) -> None: """Push a PCM chunk into the outgoing audio track.""" if self._input_track is None or not self._connected: From b6c7c72bdac8cda5603ac08bcdd6b3b84aed2742 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 10 Aug 2026 17:07:34 +0200 Subject: [PATCH 06/11] Accept arbitrary LemonSlice avatar params as a dict --- plugins/lemonslice/README.md | 1 + .../tests/test_lemonslice_plugin.py | 57 +++++++++++++++++++ .../plugins/lemonslice/lemonslice_avatar.py | 4 ++ .../plugins/lemonslice/lemonslice_client.py | 5 ++ 4 files changed, 67 insertions(+) diff --git a/plugins/lemonslice/README.md b/plugins/lemonslice/README.md index 78a41b8bd..ecb6c551d 100644 --- a/plugins/lemonslice/README.md +++ b/plugins/lemonslice/README.md @@ -85,6 +85,7 @@ lemonslice.Avatar( fps=30, # Output video frame rate buffer_seconds=1.0, # Max video buffer depth in seconds avatar_join_timeout=30.0, # Seconds to wait for the avatar to join the bridge call + lemonslice_properties=None, # Extra fields merged into the LemonSlice session request ) ``` diff --git a/plugins/lemonslice/tests/test_lemonslice_plugin.py b/plugins/lemonslice/tests/test_lemonslice_plugin.py index 365c5f150..07d115500 100644 --- a/plugins/lemonslice/tests/test_lemonslice_plugin.py +++ b/plugins/lemonslice/tests/test_lemonslice_plugin.py @@ -1,3 +1,6 @@ +import json + +import httpx import pytest from vision_agents.core.agents.inference import AudioOutputStream from vision_agents.core.utils.video_track import QueuedVideoTrack @@ -14,6 +17,20 @@ def _make_avatar(**overrides) -> LemonSliceAvatar: return LemonSliceAvatar(**{**default_kwargs, **overrides}) +@pytest.fixture +def session_requests() -> list[httpx.Request]: + return [] + + +@pytest.fixture +def session_transport(session_requests: list[httpx.Request]) -> httpx.MockTransport: + def handler(request: httpx.Request) -> httpx.Response: + session_requests.append(request) + return httpx.Response(200, json={"session_id": "session-1"}) + + return httpx.MockTransport(handler) + + class TestLemonSliceAvatar: async def test_init_with_agent_image_url_instead_of_id(self): avatar = _make_avatar( @@ -57,3 +74,43 @@ async def test_init_odd_height_raises(self): async def test_audio_output(self): avatar = _make_avatar() assert isinstance(avatar.audio_output(), AudioOutputStream) + + async def test_extra_params_are_sent_in_the_session_request( + self, + session_transport: httpx.MockTransport, + session_requests: list[httpx.Request], + ): + avatar = _make_avatar( + lemonslice_properties={"voice_id": "nova", "metadata": {"tier": "pro"}} + ) + avatar._client._http_client = httpx.AsyncClient( + base_url="https://lemonslice.test", transport=session_transport + ) + + await avatar._client.create_session( + call_id="call-1", call_type="default", token="token", api_key="stream-key" + ) + + payload = json.loads(session_requests[0].content) + assert payload["voice_id"] == "nova" + assert payload["metadata"] == {"tier": "pro"} + + async def test_extra_params_do_not_override_transport_fields( + self, + session_transport: httpx.MockTransport, + session_requests: list[httpx.Request], + ): + avatar = _make_avatar( + lemonslice_properties={"transport_type": "websocket", "properties": {}} + ) + avatar._client._http_client = httpx.AsyncClient( + base_url="https://lemonslice.test", transport=session_transport + ) + + await avatar._client.create_session( + call_id="call-1", call_type="default", token="token", api_key="stream-key" + ) + + payload = json.loads(session_requests[0].content) + assert payload["transport_type"] == "stream" + assert payload["properties"]["call_id"] == "call-1" diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py index 657b00fb7..5f4b371b9 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py @@ -57,6 +57,7 @@ def __init__( fps: int = 30, buffer_seconds: float = 1.0, avatar_join_timeout: float = 30.0, + lemonslice_properties: dict[str, Any] | None = None, ): """Initialize the LemonSlice avatar publisher. @@ -86,6 +87,8 @@ def __init__( can be queued ahead of audio playback. Must be > 0. avatar_join_timeout: Seconds to wait for the avatar participant to join the call before failing the connection. + lemonslice_properties: Extra fields added to the LemonSlice session + creation request. """ super().__init__() if buffer_seconds <= 0: @@ -96,6 +99,7 @@ def __init__( agent_image_url = agent_image_url or os.getenv("LEMONSLICE_AGENT_IMAGE_URL") client_kwargs: dict[str, Any] = { + "lemonslice_properties": lemonslice_properties, "agent_id": agent_id, "agent_image_url": agent_image_url, "agent_prompt": agent_prompt, diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py index 72dfa2178..49772a804 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py @@ -1,5 +1,6 @@ import logging from os import getenv +from typing import Any import httpx @@ -25,6 +26,7 @@ def __init__( idle_timeout: int | None = None, api_key: str | None = None, base_url: str = DEFAULT_BASE_URL, + lemonslice_properties: dict[str, Any] | None = None, ): """Initialize the LemonSlice client. @@ -35,6 +37,7 @@ def __init__( idle_timeout: Session timeout in seconds. api_key: LemonSlice API key. Uses LEMONSLICE_API_KEY env var if not provided. base_url: LemonSlice API base URL. + lemonslice_properties: Extra fields added to the session creation payload. """ if not agent_id and not agent_image_url: raise ValueError("Either agent_id or agent_image_url must be provided.") @@ -50,6 +53,7 @@ def __init__( self._agent_image_url = agent_image_url self._agent_prompt = agent_prompt self._idle_timeout = idle_timeout + self._lemonslice_properties = lemonslice_properties or {} self._session_id: str | None = None self._http_client = httpx.AsyncClient( base_url=base_url, @@ -78,6 +82,7 @@ async def create_session( The created session ID. """ payload: dict[str, object] = { + **self._lemonslice_properties, "transport_type": "stream", "properties": { "call_id": call_id, From db38985c950526e02d25a125f971eb7b32f3218d Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 10 Aug 2026 17:13:32 +0200 Subject: [PATCH 07/11] Accept full LemonSlice url for session endpoint, increase timeout --- plugins/lemonslice/tests/test_lemonslice_plugin.py | 8 ++------ .../plugins/lemonslice/lemonslice_avatar.py | 8 ++++---- .../plugins/lemonslice/lemonslice_client.py | 11 ++++++----- 3 files changed, 12 insertions(+), 15 deletions(-) diff --git a/plugins/lemonslice/tests/test_lemonslice_plugin.py b/plugins/lemonslice/tests/test_lemonslice_plugin.py index 07d115500..7ab20be52 100644 --- a/plugins/lemonslice/tests/test_lemonslice_plugin.py +++ b/plugins/lemonslice/tests/test_lemonslice_plugin.py @@ -83,9 +83,7 @@ async def test_extra_params_are_sent_in_the_session_request( avatar = _make_avatar( lemonslice_properties={"voice_id": "nova", "metadata": {"tier": "pro"}} ) - avatar._client._http_client = httpx.AsyncClient( - base_url="https://lemonslice.test", transport=session_transport - ) + avatar._client._http_client = httpx.AsyncClient(transport=session_transport) await avatar._client.create_session( call_id="call-1", call_type="default", token="token", api_key="stream-key" @@ -103,9 +101,7 @@ async def test_extra_params_do_not_override_transport_fields( avatar = _make_avatar( lemonslice_properties={"transport_type": "websocket", "properties": {}} ) - avatar._client._http_client = httpx.AsyncClient( - base_url="https://lemonslice.test", transport=session_transport - ) + avatar._client._http_client = httpx.AsyncClient(transport=session_transport) await avatar._client.create_session( call_id="call-1", call_type="default", token="token", api_key="stream-key" diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py index 5f4b371b9..ca971f7ad 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py @@ -48,7 +48,7 @@ def __init__( agent_prompt: str | None = None, idle_timeout: int | None = None, api_key: str | None = None, - base_url: str | None = None, + api_url: str | None = None, stream_api_key: str | None = None, stream_api_secret: str | None = None, stream_call_type: str = "default", @@ -67,7 +67,7 @@ def __init__( agent_prompt: Prompt describing the agent's persona. idle_timeout: Seconds before an idle session is closed. api_key: LemonSlice API key. Uses LEMONSLICE_API_KEY env var if not provided. - base_url: LemonSlice API base URL override. + api_url: Full URL of the LemonSlice session creation endpoint. stream_api_key: Stream API key. Uses STREAM_API_KEY env var if not provided. stream_api_secret: Stream API secret. Uses STREAM_API_SECRET env var if not provided. stream_call_type: Stream call type controlling the default feature set and @@ -106,8 +106,8 @@ def __init__( "idle_timeout": idle_timeout, "api_key": api_key, } - if base_url is not None: - client_kwargs["base_url"] = base_url + if api_url is not None: + client_kwargs["api_url"] = api_url self._client = LemonSliceClient(**client_kwargs) self._rtc_manager = StreamRTCManager( diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py index 49772a804..9144d1b96 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_client.py @@ -8,7 +8,7 @@ logger = logging.getLogger(__name__) -DEFAULT_BASE_URL = "https://lemonslice.com/api/liveai" +DEFAULT_API_URL = "https://lemonslice.com/api/liveai/sessions" class LemonSliceClient: @@ -25,7 +25,7 @@ def __init__( agent_prompt: str | None = None, idle_timeout: int | None = None, api_key: str | None = None, - base_url: str = DEFAULT_BASE_URL, + api_url: str = DEFAULT_API_URL, lemonslice_properties: dict[str, Any] | None = None, ): """Initialize the LemonSlice client. @@ -36,7 +36,7 @@ def __init__( agent_prompt: Prompt influencing avatar expressions and movements. idle_timeout: Session timeout in seconds. api_key: LemonSlice API key. Uses LEMONSLICE_API_KEY env var if not provided. - base_url: LemonSlice API base URL. + api_url: Full URL of the LemonSlice session creation endpoint. lemonslice_properties: Extra fields added to the session creation payload. """ if not agent_id and not agent_image_url: @@ -54,9 +54,10 @@ def __init__( self._agent_prompt = agent_prompt self._idle_timeout = idle_timeout self._lemonslice_properties = lemonslice_properties or {} + self._api_url = api_url self._session_id: str | None = None self._http_client = httpx.AsyncClient( - base_url=base_url, + timeout=120, headers={ "X-API-Key": self._api_key, "Content-Type": "application/json", @@ -101,7 +102,7 @@ async def create_session( if self._idle_timeout is not None: payload["idle_timeout"] = self._idle_timeout - response = await self._http_client.post("/sessions", json=payload) + response = await self._http_client.post(self._api_url, json=payload) if response.status_code >= 400: raise LemonSliceSessionError( From 2d209daeaf0289f5c7333c37691b697cca110018 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 10 Aug 2026 20:15:47 +0200 Subject: [PATCH 08/11] Increase lemonslice track buffer size to 180s --- plugins/lemonslice/vision_agents/plugins/lemonslice/track.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py index 147b9ac5b..b4173ec80 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/track.py @@ -27,7 +27,7 @@ def __init__( sample_rate: int = 48000, channels: int = 1, format: AudioFormatType = AudioFormat.S16, - audio_buffer_size_ms: int = 30000, + audio_buffer_size_ms: int = 180_000, ): super().__init__( sample_rate=sample_rate, From b5b5fedddd97fbf228976bcf44625981c7e678f6 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Mon, 10 Aug 2026 21:02:34 +0200 Subject: [PATCH 09/11] Do not send end-of-utterance to Lemonslice on flush --- .../tests/test_lemonslice_plugin.py | 55 ++++++++++++++++++- .../plugins/lemonslice/lemonslice_avatar.py | 5 +- 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/plugins/lemonslice/tests/test_lemonslice_plugin.py b/plugins/lemonslice/tests/test_lemonslice_plugin.py index 7ab20be52..ee2d1247c 100644 --- a/plugins/lemonslice/tests/test_lemonslice_plugin.py +++ b/plugins/lemonslice/tests/test_lemonslice_plugin.py @@ -1,10 +1,16 @@ +import asyncio import json import httpx +import numpy as np import pytest -from vision_agents.core.agents.inference import AudioOutputStream +from getstream import AsyncStream +from getstream.video.rtc.track_util import AudioFormat, PcmData +from vision_agents.core.agents.inference import AudioOutputFlush, AudioOutputStream +from vision_agents.core.utils.utils import cancel_and_wait from vision_agents.core.utils.video_track import QueuedVideoTrack from vision_agents.plugins.lemonslice.lemonslice_avatar import LemonSliceAvatar +from vision_agents.plugins.lemonslice.track import AvatarInputTrack def _make_avatar(**overrides) -> LemonSliceAvatar: @@ -31,6 +37,21 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.MockTransport(handler) +@pytest.fixture +def call_events() -> list[dict]: + return [] + + +@pytest.fixture +def call_event_transport(call_events: list[dict]) -> httpx.MockTransport: + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/event"): + call_events.append(json.loads(request.content)["custom"]) + return httpx.Response(200, json={"duration": "0ms"}) + + return httpx.MockTransport(handler) + + class TestLemonSliceAvatar: async def test_init_with_agent_image_url_instead_of_id(self): avatar = _make_avatar( @@ -110,3 +131,35 @@ async def test_extra_params_do_not_override_transport_fields( payload = json.loads(session_requests[0].content) assert payload["transport_type"] == "stream" assert payload["properties"]["call_id"] == "call-1" + + async def test_interrupt_does_not_announce_an_end_of_utterance( + self, call_events: list[dict], call_event_transport: httpx.MockTransport + ): + # An interruption discards the buffered audio, so announcing an + # end-of-utterance PTS that covers it points the avatar at audio that + # never arrives. + avatar = _make_avatar() + manager = avatar._rtc_manager + manager._client = AsyncStream( + api_key="key", api_secret="secret", transport=call_event_transport + ) + manager._call = manager._client.video.call("default", "call-1") + manager._input_track = AvatarInputTrack(sample_rate=16000, channels=1) + manager._connected = True + await manager._input_track.write( + PcmData( + samples=np.zeros(16000, dtype=np.int16), + sample_rate=16000, + format=AudioFormat.S16, + channels=1, + ) + ) + + stream = AudioOutputStream() + avatar.attach_audio_input(stream) + task = asyncio.create_task(avatar._process_audio_input()) + stream.send_nowait(AudioOutputFlush()) + await asyncio.sleep(0.05) + await cancel_and_wait(task) + + assert [event["type"] for event in call_events] == ["lemonslice.interrupt"] diff --git a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py index ca971f7ad..1ce7ebd5f 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_avatar.py @@ -175,8 +175,9 @@ async def _process_audio_input(self) -> None: await self._end_turn() elif isinstance(item, AudioOutputFlush): - # Audio was interrupted - await self._end_turn() + # Audio was interrupted. No end-of-utterance here: interrupt() + # discards the buffered audio, so a PTS covering it would point + # the avatar at audio that never arrives. await self._sync.flush() await self._rtc_manager.interrupt() From b531d99b003eff27fb405f16fe8dcca72ea0516b Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 11 Aug 2026 19:18:06 +0200 Subject: [PATCH 10/11] Send AudioOutputChunk(final=True) only when the TTS output is complete TranscribingInferenceFlow was emitting final chunks for every TTSOutputChunk with final=True, which may mean the end of a single sentence rather than the whole synthesis. --- .../agents/inference/transcribing_flow.py | 17 ++++-- .../test_inference/test_transcribing_flow.py | 60 +++++++++++++++++-- 2 files changed, 66 insertions(+), 11 deletions(-) diff --git a/agents-core/vision_agents/core/agents/inference/transcribing_flow.py b/agents-core/vision_agents/core/agents/inference/transcribing_flow.py index e86212ca4..17656351e 100644 --- a/agents-core/vision_agents/core/agents/inference/transcribing_flow.py +++ b/agents-core/vision_agents/core/agents/inference/transcribing_flow.py @@ -538,6 +538,7 @@ async def process_tts( if remainder: async for chunk in self._tts.send_iter(remainder): await tts_output.send(chunk) + await tts_output.send(TTSOutputEnd()) elif item.delta: # Chunk: accumulate and emit on sentence boundaries. text = self._tts_tokenizer.update(item.text) @@ -551,17 +552,26 @@ async def process_tts( if isinstance(item, TTSInput) and not item.delta: async for chunk in self._tts.send_iter(item.text): await tts_output.send(chunk) + await tts_output.send(TTSOutputEnd()) async def write_audio_output( self, tts_output: Stream[TTSOutputChunk | TTSOutputEnd], audio_output: AudioOutputStream, ): + # A streaming TTS speaks one reply as several segments, and each segment + # ends with a chunk where final=True. + # TTSOutputEnd is the only signal that the complete reply was synthesized, + # and it triggers AudioOutputChunk.final for downstream consumers. speaking = False async for item in tts_output: with log_exceptions(logger, "Error while processing TTS output"): if isinstance(item, TTSOutputEnd): if speaking: + # An interrupt discards the audio already queued, so + # there is no complete reply to mark. + if not item.interrupted: + await audio_output.send(AudioOutputChunk(final=True)) self.events.send( AgentTurnEndedEvent(interrupted=item.interrupted) ) @@ -570,9 +580,4 @@ async def write_audio_output( if not speaking: self.events.send(AgentTurnStartedEvent()) speaking = True - await audio_output.send( - AudioOutputChunk(data=item.data, final=item.final) - ) - if item.final: - self.events.send(AgentTurnEndedEvent()) - speaking = False + await audio_output.send(AudioOutputChunk(data=item.data)) diff --git a/tests/test_agents/test_inference/test_transcribing_flow.py b/tests/test_agents/test_inference/test_transcribing_flow.py index 350a0ed3b..a7c680ef6 100644 --- a/tests/test_agents/test_inference/test_transcribing_flow.py +++ b/tests/test_agents/test_inference/test_transcribing_flow.py @@ -1204,7 +1204,32 @@ async def test_streaming_input_end_flushes_remainder(self, flow_factory) -> None await stage chunks = tts_out.peek() - assert [c.text for c in chunks] == ["Hi"] + assert [c.text for c in chunks[:-1]] == ["Hi"] + assert chunks[-1] == TTSOutputEnd() + + async def test_streaming_multi_sentence_turn_ends_once(self, flow_factory) -> None: + tts = TTSStub(streaming=True) + flow = flow_factory(tts=tts) + tts_in: Stream[TTSInput | TTSInputEnd] = Stream() + tts_out: Stream[TTSOutputChunk | TTSOutputEnd] = Stream() + + stage = asyncio.create_task(flow.process_tts(tts_in, tts_out)) + + await tts_in.send(TTSInput(text="One. ", delta=True)) + await tts_in.send(TTSInput(text="Two. ", delta=True)) + await tts_in.send(TTSInput(text="Three", delta=True)) + await tts_in.send(TTSInputEnd()) + + tts_in.close() + await stage + + # Three synthesis segments, but a single end-of-turn sentinel. + assert tts_out.peek() == [ + TTSOutputChunk(text="One."), + TTSOutputChunk(text="Two."), + TTSOutputChunk(text="Three"), + TTSOutputEnd(), + ] async def test_streaming_ignores_delta_false_inputs(self, flow_factory) -> None: tts = TTSStub(streaming=True) @@ -1239,7 +1264,7 @@ async def test_non_streaming_emits_only_on_full_utterance( await stage chunks = tts_out.peek() - assert [c.text for c in chunks] == ["full"] + assert chunks == [TTSOutputChunk(text="full"), TTSOutputEnd()] async def test_non_streaming_passes_through_all_chunks(self, flow_factory) -> None: preloaded = [ @@ -1250,7 +1275,7 @@ async def test_non_streaming_passes_through_all_chunks(self, flow_factory) -> No tts = TTSStub(chunks=preloaded, streaming=False) flow = flow_factory(tts=tts) tts_in: Stream[TTSInput | TTSInputEnd] = Stream() - tts_out: Stream[TTSOutputChunk] = Stream() + tts_out: Stream[TTSOutputChunk | TTSOutputEnd] = Stream() stage = asyncio.create_task(flow.process_tts(tts_in, tts_out)) @@ -1259,7 +1284,7 @@ async def test_non_streaming_passes_through_all_chunks(self, flow_factory) -> No tts_in.close() await stage - assert tts_out.peek() == preloaded + assert tts_out.peek() == [*preloaded, TTSOutputEnd()] async def test_returns_early_when_tts_is_none(self, flow_factory) -> None: flow = flow_factory(tts=None) @@ -1277,7 +1302,7 @@ async def test_returns_early_when_tts_is_none(self, flow_factory) -> None: class TestWriteAudioOutput: - async def test_forwards_chunks_preserving_final_flag(self, flow_factory) -> None: + async def test_only_output_end_marks_the_end_of_a_turn(self, flow_factory) -> None: flow = flow_factory() started: list[AgentTurnStartedEvent] = [] ended: list[AgentTurnEndedEvent] = [] @@ -1295,14 +1320,22 @@ async def _on(event: AgentTurnStartedEvent | AgentTurnEndedEvent): # data=None lets the chunk pass through AudioOutputStream unchanged, # bypassing its 20ms re-chunking so the assertion stays focused. + # Two synthesis segments, each ending with its own final chunk. + await tts_out.send(TTSOutputChunk(data=None, final=False)) + await tts_out.send(TTSOutputChunk(data=None, final=True)) await tts_out.send(TTSOutputChunk(data=None, final=False)) await tts_out.send(TTSOutputChunk(data=None, final=True)) + await tts_out.send(TTSOutputEnd()) tts_out.close() await stage await flow.events.wait() + # Per-segment finals are not end-of-turn: exactly one final marker. assert audio_out.peek() == [ + AudioOutputChunk(data=None, final=False), + AudioOutputChunk(data=None, final=False), + AudioOutputChunk(data=None, final=False), AudioOutputChunk(data=None, final=False), AudioOutputChunk(data=None, final=True), ] @@ -1310,6 +1343,23 @@ async def _on(event: AgentTurnStartedEvent | AgentTurnEndedEvent): assert len(ended) == 1 assert ended[0].interrupted is False + async def test_interrupted_output_end_emits_no_final_marker( + self, flow_factory + ) -> None: + # An interrupt discards the audio already queued, so there is no + # complete reply to mark. + flow = flow_factory() + tts_out: Stream[TTSOutputChunk | TTSOutputEnd] = Stream() + audio_out = AudioOutputStream() + stage = asyncio.create_task(flow.write_audio_output(tts_out, audio_out)) + + await tts_out.send(TTSOutputChunk(data=None, final=False)) + await tts_out.send(TTSOutputEnd(interrupted=True)) + tts_out.close() + await stage + + assert audio_out.peek() == [AudioOutputChunk(data=None, final=False)] + async def test_tts_output_end_emits_interrupted_agent_turn_ended( self, flow_factory ) -> None: From ff1e90a15e385081507455ad9c44b48c576e0979 Mon Sep 17 00:00:00 2001 From: Daniil Gusev Date: Tue, 25 Aug 2026 13:21:33 +0200 Subject: [PATCH 11/11] Flush the resampler before sending "end_utterance" event Without flushing, a piece of audio was sitting in the resampler's buffer mixing up pts computation --- .../tests/test_lemonslice_plugin.py | 41 ++++++++++++++++--- .../lemonslice/lemonslice_rtc_manager.py | 8 ++++ 2 files changed, 44 insertions(+), 5 deletions(-) diff --git a/plugins/lemonslice/tests/test_lemonslice_plugin.py b/plugins/lemonslice/tests/test_lemonslice_plugin.py index ee2d1247c..a32c8c164 100644 --- a/plugins/lemonslice/tests/test_lemonslice_plugin.py +++ b/plugins/lemonslice/tests/test_lemonslice_plugin.py @@ -132,12 +132,9 @@ async def test_extra_params_do_not_override_transport_fields( assert payload["transport_type"] == "stream" assert payload["properties"]["call_id"] == "call-1" - async def test_interrupt_does_not_announce_an_end_of_utterance( + async def test_end_utterance_and_interrupt_events_respect_audio_boundaries( self, call_events: list[dict], call_event_transport: httpx.MockTransport ): - # An interruption discards the buffered audio, so announcing an - # end-of-utterance PTS that covers it points the avatar at audio that - # never arrives. avatar = _make_avatar() manager = avatar._rtc_manager manager._client = AsyncStream( @@ -146,6 +143,37 @@ async def test_interrupt_does_not_announce_an_end_of_utterance( manager._call = manager._client.video.call("default", "call-1") manager._input_track = AvatarInputTrack(sample_rate=16000, channels=1) manager._connected = True + + for _ in range(3): + await manager.send_audio( + PcmData( + samples=np.zeros(480, dtype=np.int16), + sample_rate=24000, + format=AudioFormat.S16, + channels=1, + ) + ) + + await manager.flush() + + emitted_samples = 0 + while True: + try: + emitted_samples += ( + await asyncio.wait_for(manager._input_track.recv(), timeout=0.05) + ).samples + except TimeoutError: + break + + assert call_events[0] == { + "type": "lemonslice.end_utterance", + "pts": emitted_samples, + "event_id": 1, + } + + # An interruption discards the buffered audio, so announcing an + # end-of-utterance PTS that covers it points the avatar at audio that + # never arrives. await manager._input_track.write( PcmData( samples=np.zeros(16000, dtype=np.int16), @@ -162,4 +190,7 @@ async def test_interrupt_does_not_announce_an_end_of_utterance( await asyncio.sleep(0.05) await cancel_and_wait(task) - assert [event["type"] for event in call_events] == ["lemonslice.interrupt"] + assert [event["type"] for event in call_events] == [ + "lemonslice.end_utterance", + "lemonslice.interrupt", + ] 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 af339c8bb..6da529bd2 100644 --- a/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py +++ b/plugins/lemonslice/vision_agents/plugins/lemonslice/lemonslice_rtc_manager.py @@ -257,6 +257,14 @@ async def flush(self) -> None: """Signal end of a TTS segment to the avatar via a custom call event.""" if self._call is None or not self._connected or self._input_track is None: return + await self._input_track.write( + PcmData( + sample_rate=self._input_track.sample_rate, + format=self._input_track.format, + channels=self._input_track.channels, + ), + final=True, + ) pts = await self._input_track.pts() await self._call.send_call_event( user_id=self._plugin_user_id,