From 5daf27b185af58a62e5316ad077d3835ada9c5f9 Mon Sep 17 00:00:00 2001 From: lottielin <77655652+lottielin@users.noreply.github.com> Date: Thu, 30 Jul 2026 11:26:41 -0700 Subject: [PATCH] feat: support audio_stream_end for realtime input Merge https://github.com/google/adk-python/pull/4490 **Problem:** `send_realtime` method only accepted `Blob` (audio/video bytes), `ActivityStart`, and `ActivityEnd`. There's no mechanism to send the `audioStreamEnd` boolean field, which is required to flush cached audio when Voice Activity Detection ([VAD](https://ai.google.dev/gemini-api/docs/live-guide#interruptions)) is enabled. **Solution:** This PR updates the `GeminiLlmConnection`, `BaseLlmFlow`, and `LiveRequestQueue` to support sending generic `LiveClientRealtimeInput` messages with `audio_stream_end` field configured to the [Gemini Live API](https://googleapis.github.io/python-genai/genai.html#genai.live.AsyncSession.send_realtime_input). This closes #2887. **Unit Tests:** - added unit test `test_send_realtime_audiostreamend` - all unit tests passing locally - processing priority: activity_start > activity_end > audio_stream_end > blob > content - this is to follows the principal of control signals > data payloads, since `audio_stream_end` is a signal to flush the audio buffer Co-authored-by: Liang Wu COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/4490 from lottielin:support-audio-stream-end ea6d2dfae3b9f1c5456b8cd8c74d16163443c06c PiperOrigin-RevId: 956638801 --- src/google/adk/agents/live_request_queue.py | 34 ++++++++++--------- .../adk/flows/llm_flows/base_llm_flow.py | 4 +++ .../adk/models/gemini_llm_connection.py | 13 ++++++- .../models/test_gemini_llm_connection.py | 31 +++++++++++++++++ 4 files changed, 65 insertions(+), 17 deletions(-) diff --git a/src/google/adk/agents/live_request_queue.py b/src/google/adk/agents/live_request_queue.py index 9b698c81d6d..1722547f189 100644 --- a/src/google/adk/agents/live_request_queue.py +++ b/src/google/adk/agents/live_request_queue.py @@ -24,35 +24,33 @@ class LiveRequest(BaseModel): - """Request send to live agents.""" + """Request send to live agents. + + When multiple fields are set, they are processed by priority (highest first): + activity_start > activity_end > audio_stream_end > blob > content. + state_delta, if set, is always applied regardless of the other fields. + """ model_config = ConfigDict(ser_json_bytes='base64', val_json_bytes='base64') """The pydantic model config.""" content: Optional[types.Content] = None - """If set, send the content to the model in turn-by-turn mode. + """If set, send the content to the model in turn-by-turn mode.""" - When multiple fields are set, they are processed by priority (highest first): - activity_start > activity_end > blob > content. - """ blob: Optional[types.Blob] = None - """If set, send the blob to the model in realtime mode. + """If set, send the blob to the model in realtime mode.""" - When multiple fields are set, they are processed by priority (highest first): - activity_start > activity_end > blob > content. - """ activity_start: Optional[types.ActivityStart] = None - """If set, signal the start of user activity to the model. + """If set, signal the start of user activity to the model.""" - When multiple fields are set, they are processed by priority (highest first): - activity_start > activity_end > blob > content. - """ activity_end: Optional[types.ActivityEnd] = None - """If set, signal the end of user activity to the model. + """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. + audio_stream_end: bool = False + """If set, signal the end of the audio stream to the model. This is only used + when Voice Activity Detection is enabled. """ + close: bool = False """If set, close the queue. queue.shutdown() is only supported in Python 3.13+.""" @@ -80,6 +78,10 @@ def send_activity_end(self): """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 audio.""" + self._queue.put_nowait(LiveRequest(audio_stream_end=True)) + def send(self, req: LiveRequest): 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 e359cc56259..23ec1f19747 100644 --- a/src/google/adk/flows/llm_flows/base_llm_flow.py +++ b/src/google/adk/flows/llm_flows/base_llm_flow.py @@ -768,6 +768,10 @@ 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( + types.LiveClientRealtimeInput(audio_stream_end=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 6646fe9a798..634f8c9f316 100644 --- a/src/google/adk/models/gemini_llm_connection.py +++ b/src/google/adk/models/gemini_llm_connection.py @@ -29,7 +29,12 @@ logger = logging.getLogger('google_adk.' + __name__) -RealtimeInput = Union[types.Blob, types.ActivityStart, types.ActivityEnd] +RealtimeInput = Union[ + types.Blob, + types.ActivityStart, + types.ActivityEnd, + types.LiveClientRealtimeInput, +] from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -157,6 +162,12 @@ async def send_realtime(self, input: RealtimeInput): 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, types.LiveClientRealtimeInput): + if input.audio_stream_end: + logger.debug('Sending LLM audio stream end signal.') + await self._gemini_session.send_realtime_input(audio_stream_end=True) + else: + logger.warning('Unary LiveClientRealtimeInput not fully supported yet.') 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 5a39e4ae605..a3ac0daad4f 100644 --- a/tests/unittests/models/test_gemini_llm_connection.py +++ b/tests/unittests/models/test_gemini_llm_connection.py @@ -71,6 +71,37 @@ async def test_send_realtime_default_behavior( mock_gemini_session.send.assert_not_called() +@pytest.mark.asyncio +async def test_send_realtime_audio_stream_end( + gemini_connection, mock_gemini_session +): + """Test send_realtime with LiveClientRealtimeInput(audio_stream_end=True).""" + input_signal = types.LiveClientRealtimeInput(audio_stream_end=True) + await gemini_connection.send_realtime(input_signal) + + # Should call send_realtime_input with audio_stream_end=True + mock_gemini_session.send_realtime_input.assert_called_once_with( + audio_stream_end=True + ) + + +@pytest.mark.asyncio +async def test_send_realtime_unsupported_liveClientRealtimeInput( + gemini_connection, mock_gemini_session, caplog +): + """Test send_realtime with unsupported LiveClientRealtimeInput.""" + input_signal = types.LiveClientRealtimeInput() + + with caplog.at_level('WARNING'): + await gemini_connection.send_realtime(input_signal) + + # Should log a warning + assert 'Unary LiveClientRealtimeInput not fully supported yet.' in caplog.text + # Should not call send_realtime_input or send + mock_gemini_session.send_realtime_input.assert_not_called() + mock_gemini_session.send.assert_not_called() + + @pytest.mark.asyncio async def test_send_history(gemini_connection, mock_gemini_session): """Test send_history method."""