From 7e8bb10636d80e76f229bb1825a98160cc20d243 Mon Sep 17 00:00:00 2001 From: Ishaan Date: Wed, 29 Jul 2026 09:16:59 +0000 Subject: [PATCH] Support audioStreamEnd in send_realtime for realtimeInput The Gemini Live API's `send_realtime_input` accepts an `audio_stream_end` parameter (a boolean flag) that forces the model to flush any buffered audio immediately when VAD (voice activity detection) is enabled. This is useful when the client wants to signal the definitive end of an audio stream without waiting for VAD to detect silence. Signed-off-by: Ishaan --- src/google/adk/agents/live_request_queue.py | 16 ++++++++++++++++ src/google/adk/flows/llm_flows/base_llm_flow.py | 2 ++ src/google/adk/models/gemini_llm_connection.py | 5 ++++- .../models/test_gemini_llm_connection.py | 12 ++++++++++++ 4 files changed, 34 insertions(+), 1 deletion(-) diff --git a/src/google/adk/agents/live_request_queue.py b/src/google/adk/agents/live_request_queue.py index c9c7b82c685..2241b053622 100644 --- a/src/google/adk/agents/live_request_queue.py +++ b/src/google/adk/agents/live_request_queue.py @@ -53,6 +53,14 @@ class LiveRequest(BaseModel): activity_end: Optional[types.ActivityEnd] = None """If set, signal the end of user activity to the model. + When multiple fields are set, they are processed by priority (highest first): + activity_start > activity_end > blob > content. state_delta, if set, is always + applied regardless of the other fields. + """ + audio_stream_end: Optional[bool] = None + """If set, signal the end of the audio stream to force flush buffered audio + when VAD (voice activity detection) is enabled. + When multiple fields are set, they are processed by priority (highest first): activity_start > activity_end > blob > content. state_delta, if set, is always applied regardless of the other fields. @@ -92,6 +100,14 @@ def send_activity_end(self) -> None: """Sends an activity end signal to mark the end of user input.""" self._queue.put_nowait(LiveRequest(activity_end=types.ActivityEnd())) + def send_audio_stream_end(self) -> None: + """Sends an audio stream end signal to force flush buffered audio. + + Use this to signal the end of an audio stream when VAD is enabled, + which causes the model to process any buffered audio immediately. + """ + self._queue.put_nowait(LiveRequest(audio_stream_end=True)) + def send(self, req: LiveRequest) -> None: self._queue.put_nowait(req) diff --git a/src/google/adk/flows/llm_flows/base_llm_flow.py b/src/google/adk/flows/llm_flows/base_llm_flow.py index f215895c994..4485cb46148 100644 --- a/src/google/adk/flows/llm_flows/base_llm_flow.py +++ b/src/google/adk/flows/llm_flows/base_llm_flow.py @@ -827,6 +827,8 @@ async def _send_to_model( await llm_connection.send_realtime(types.ActivityStart()) elif live_request.activity_end: await llm_connection.send_realtime(types.ActivityEnd()) + elif live_request.audio_stream_end: + await llm_connection.send_realtime(True) elif live_request.blob: # Cache input audio chunks before flushing self.audio_cache_manager.cache_audio( diff --git a/src/google/adk/models/gemini_llm_connection.py b/src/google/adk/models/gemini_llm_connection.py index 66534b39fdc..717623bb8c7 100644 --- a/src/google/adk/models/gemini_llm_connection.py +++ b/src/google/adk/models/gemini_llm_connection.py @@ -29,7 +29,7 @@ logger = logging.getLogger('google_adk.' + __name__) -RealtimeInput = Union[types.Blob, types.ActivityStart, types.ActivityEnd] +RealtimeInput = Union[types.Blob, types.ActivityStart, types.ActivityEnd, bool] from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -173,6 +173,9 @@ async def send_realtime(self, input: RealtimeInput) -> None: elif isinstance(input, types.ActivityEnd): logger.debug('Sending LLM activity end signal.') await self._gemini_session.send_realtime_input(activity_end=input) + elif isinstance(input, bool) and input: + logger.debug('Sending LLM audio stream end signal.') + await self._gemini_session.send_realtime_input(audio_stream_end=True) else: raise ValueError('Unsupported input type: %s' % type(input)) diff --git a/tests/unittests/models/test_gemini_llm_connection.py b/tests/unittests/models/test_gemini_llm_connection.py index 5bded80a266..6d5beee3847 100644 --- a/tests/unittests/models/test_gemini_llm_connection.py +++ b/tests/unittests/models/test_gemini_llm_connection.py @@ -89,6 +89,18 @@ async def test_send_realtime_audio_uses_audio_channel_for_live_translate( ) +@pytest.mark.asyncio +async def test_send_realtime_audio_stream_end( + gemini_connection, mock_gemini_session +): + """Test send_realtime with audio_stream_end=True sends the flush signal.""" + await gemini_connection.send_realtime(True) + + mock_gemini_session.send_realtime_input.assert_called_once_with( + audio_stream_end=True + ) + + @pytest.mark.asyncio async def test_send_history(gemini_connection, mock_gemini_session): """Test send_history method."""