Fix Gemini audio (#165)

Author: tbarbugli

agents-core/pyproject.toml

@@ -21,7 +21,7 @@ classifiers = [
 
 requires-python = ">=3.10"
 dependencies = [
-    "getstream[webrtc,telemetry]>=2.5.11",
+    "getstream[webrtc,telemetry]>=2.5.14",
     "python-dotenv>=1.1.1",
     "pillow>=10.4.0",  # Compatible with moondream SDK (<11.0.0)
     "numpy>=1.24.0",
@@ -91,5 +91,5 @@ include = ["vision_agents"]
 #]
 # getstream = { git = "https://github.com/GetStream/stream-py.git", branch = "audio-more" }
 # for local development
-#getstream = { git = "https://github.com/GetStream/stream-py.git", rev = "85bd8ef00859ef6ed5ef4ffe7b7f40ae12d12973" }
+#getstream = { path = "../../stream-py/", editable = true }
 # aiortc = { path = "../stream-py/", editable = true }

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

@@ -31,6 +31,7 @@
     LLMResponseCompletedEvent,
     RealtimeUserSpeechTranscriptionEvent,
     RealtimeAgentSpeechTranscriptionEvent,
+    RealtimeAudioOutputEvent,
 )
 from ..llm.llm import AudioLLM, LLM, VideoLLM
 from ..llm.realtime import Realtime
@@ -70,7 +71,6 @@
 tracer: Tracer = trace.get_tracer("agents")
 
 
-
 class Agent:
     """
     Agent class makes it easy to build your own video AI.
@@ -227,7 +227,9 @@ def __init__(
 
     async def _finish_llm_turn(self):
         if self._pending_turn is None or self._pending_turn.response is None:
-            raise ValueError("Finish LLM turn should only be called after self._pending_turn is set")
+            raise ValueError(
+                "Finish LLM turn should only be called after self._pending_turn is set"
+            )
         turn = self._pending_turn
         self._pending_turn = None
         event = turn.response
@@ -252,6 +254,7 @@ def setup_event_handling(self):
         self.events.subscribe(self._on_turn_event)
 
         if self.stt:
+
             @self.stt.events.subscribe
             async def on_turn_ended(event: TurnEndedEvent):
                 logger.info("Received TurnEndedEvent %s", event)
@@ -322,10 +325,11 @@ async def on_stt_transcript_event_create_response(event: STTTranscriptEvent):
 
             # if turn detection is disabled, treat the transcript event as an end of turn
             if not self.turn_detection_enabled:
-                self.events.send(TurnEndedEvent(
-                    participant = event.participant,
-                ))
-
+                self.events.send(
+                    TurnEndedEvent(
+                        participant=event.participant,
+                    )
+                )
 
         # TODO: chat event handling needs work
 
@@ -634,7 +638,12 @@ async def _apply(self, function_name: str, *args, **kwargs):
             ):
                 func = getattr(subclass, function_name)
                 if func is not None:
-                    await func(*args, **kwargs)
+                    try:
+                        await func(*args, **kwargs)
+                    except Exception as e:
+                        self.logger.exception(
+                            f"Error calling {function_name} on {subclass.__class__.__name__}: {e}"
+                        )
 
     def _end_tracing(self):
         if self._root_span is not None:
@@ -879,7 +888,10 @@ async def _reply_to_audio_consumer(self) -> None:
                             pcm, participant, conversation=self.conversation
                         )
 
-                    if participant and getattr(participant, "user_id", None) != self.agent_user.id:
+                    if (
+                        participant
+                        and getattr(participant, "user_id", None) != self.agent_user.id
+                    ):
                         # first forward to processors
                         # Extract audio bytes for processors using the proper PCM data structure
                         # PCM data has: format, sample_rate, samples, pts, dts, time_base
@@ -1044,6 +1056,8 @@ async def _on_turn_event(self, event: TurnStartedEvent | TurnEndedEvent) -> None
                     self.logger.info(
                         f"👉 Turn started - participant speaking {participant_id} : {event.confidence}"
                     )
+                if self._audio_track is not None:
+                    await self._audio_track.flush()
             else:
                 # Agent itself started speaking - this is normal
                 participant_id = (
@@ -1078,9 +1092,15 @@ async def _on_turn_event(self, event: TurnStartedEvent | TurnEndedEvent) -> None
                 self._pending_user_transcripts[participant.user_id] = ""
                 # cancel the old task if the text changed in the meantime
 
-                if self._pending_turn is not None and self._pending_turn.input != transcript:
-                    logger.debug("Eager turn and completed turn didn't match. Cancelling in flight response. %s vs %s ",
-                                self._pending_turn.input, transcript)
+                if (
+                    self._pending_turn is not None
+                    and self._pending_turn.input != transcript
+                ):
+                    logger.debug(
+                        "Eager turn and completed turn didn't match. Cancelling in flight response. %s vs %s ",
+                        self._pending_turn.input,
+                        transcript,
+                    )
                     if self._pending_turn.task:
                         self._pending_turn.task.cancel()
 
@@ -1092,18 +1112,22 @@ async def _on_turn_event(self, event: TurnStartedEvent | TurnEndedEvent) -> None
                         input=transcript,
                         participant=event.participant,
                         started_at=datetime.datetime.now(),
-                        turn_finished=not event.eager_end_of_turn
+                        turn_finished=not event.eager_end_of_turn,
                     )
                     self._pending_turn = llm_turn
-                    task = asyncio.create_task(self.simple_response(transcript, event.participant))
+                    task = asyncio.create_task(
+                        self.simple_response(transcript, event.participant)
+                    )
                     llm_turn.task = task
                 elif self._pending_turn.input == transcript:
                     # same text as pending turn
                     is_finished = not event.eager_end_of_turn
                     now = datetime.datetime.now()
                     elapsed = now - self._pending_turn.started_at
-                    logger.debug("Marking eager turn as completed. Eager turn detection saved %.2f",
-                                elapsed.total_seconds() * 1000)
+                    logger.debug(
+                        "Marking eager turn as completed. Eager turn detection saved %.2f",
+                        elapsed.total_seconds() * 1000,
+                    )
 
                     if is_finished:
                         self._pending_turn.turn_finished = True
@@ -1113,8 +1137,9 @@ async def _on_turn_event(self, event: TurnStartedEvent | TurnEndedEvent) -> None
     @property
     def turn_detection_enabled(self):
         # return true if either turn detection or stt provide turn detection capabilities
-        return self.turn_detection is not None or (self.stt is not None and self.stt.turn_detection)
-
+        return self.turn_detection is not None or (
+            self.stt is not None and self.stt.turn_detection
+        )
 
     @property
     def publish_audio(self) -> bool:
@@ -1246,30 +1271,17 @@ def _validate_configuration(self):
     def _prepare_rtc(self):
         # Variables are now initialized in __init__
 
-        # Set up audio track if TTS is available
         if self.publish_audio:
-            if _is_audio_llm(self.llm):
-                self._audio_track = self.llm.output_audio_track
-                self.logger.info("🎵 Using Realtime provider output track for audio")
-            elif self.audio_publishers:
-                # Get the first audio publisher to create the track
-                audio_publisher = self.audio_publishers[0]
-                self._audio_track = audio_publisher.publish_audio_track()
-                self.logger.info("🎵 Audio track initialized from audio publisher")
-            else:
-                # Default to WebRTC-friendly format unless configured differently
-                framerate = 48000
-                stereo = True
-                self._audio_track = self.edge.create_audio_track(
-                    framerate=framerate, stereo=stereo
-                )
-                # Inform TTS of desired output format so it can resample accordingly
-                if self.tts:
-                    channels = 2 if stereo else 1
-                    self.tts.set_output_format(
-                        sample_rate=framerate,
-                        channels=channels,
-                    )
+            framerate = 48000
+            stereo = True
+            self._audio_track = self.edge.create_audio_track(
+                framerate=framerate, stereo=stereo
+            )
+
+            @self.events.subscribe
+            async def forward_audio(event: RealtimeAudioOutputEvent):
+                if self._audio_track is not None:
+                    await self._audio_track.write(event.data)
 
         # Set up video track if video publishers are available
         if self.publish_video:

agents-core/vision_agents/core/edge/edge_transport.py

@@ -35,7 +35,7 @@ def create_audio_track(self) -> OutputAudioTrack:
         pass
 
     @abc.abstractmethod
-    def close(self):
+    async def close(self):
         pass
 
     @abc.abstractmethod

agents-core/vision_agents/core/edge/types.py

@@ -45,3 +45,5 @@ class OutputAudioTrack(Protocol):
     async def write(self, data: PcmData) -> None: ...
 
     def stop(self) -> None: ...
+
+    async def flush(self) -> None: ...

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

@@ -24,7 +24,7 @@
     from vision_agents.core.agents.conversation import Conversation
 
 from getstream.video.rtc.pb.stream.video.sfu.models.models_pb2 import Participant
-from getstream.video.rtc import AudioStreamTrack, PcmData
+from getstream.video.rtc import PcmData
 from vision_agents.core.processors import Processor
 from vision_agents.core.utils.utils import parse_instructions
 from vision_agents.core.events.manager import EventManager
@@ -426,13 +426,6 @@ async def simple_audio_response(
             participant: Optional participant information for the audio source.
         """
 
-    @property
-    @abc.abstractmethod
-    def output_audio_track(self) -> AudioStreamTrack:
-        """
-        An output audio track from the LLM.
-        """
-
 
 class VideoLLM(LLM, metaclass=abc.ABCMeta):
     """

plugins/gemini/vision_agents/plugins/gemini/gemini_realtime.py

@@ -4,7 +4,6 @@
 from typing import Optional, List, Dict, Any
 
 import aiortc
-from getstream.video.rtc.audio_track import AudioStreamTrack
 from getstream.video.rtc.track_util import PcmData
 from google import genai
 from google.genai.live import AsyncSession
@@ -103,19 +102,12 @@ def __init__(
         self.client = client
         self.config: LiveConnectConfigDict = self._create_config(config)
         self.logger = logging.getLogger(__name__)
-        # Gemini generates at 24k. webrtc automatically translates it to 48khz
-        self._output_audio_track = AudioStreamTrack(
-            sample_rate=24000, channels=1, format="s16"
-        )
+
         self._video_forwarder: Optional[VideoForwarder] = None
         self._session_context: Optional[Any] = None
         self._session: Optional[AsyncSession] = None
         self._receive_task: Optional[asyncio.Task[Any]] = None
 
-    @property
-    def output_audio_track(self) -> AudioStreamTrack:
-        return self._output_audio_track
-
     async def simple_response(
         self,
         text: str,
@@ -315,7 +307,6 @@ async def _receive_loop(self):
                                         self._emit_audio_output_event(
                                             audio_data=pcm,
                                         )
-                                        await self._output_audio_track.write(pcm)
                                     elif (
                                         hasattr(typed_part, "function_call")
                                         and typed_part.function_call

plugins/getstream/vision_agents/plugins/getstream/stream_edge_transport.py

@@ -347,7 +347,7 @@ def _get_subscription_config(self):
             ]
         )
 
-    def close(self):
+    async def close(self):
         # Note: Not calling super().close() as it's an abstract method with trivial body
         pass