From fabb175e5596117091e9b42c7c6cb007512c08e8 Mon Sep 17 00:00:00 2001 From: GitMarco27 Date: Mon, 3 Aug 2026 17:01:45 +0200 Subject: [PATCH 1/3] feat: add OpenAI Realtime live support --- src/google/adk/labs/openai/README.md | 125 +- src/google/adk/labs/openai/_openai_llm.py | 42 +- .../adk/labs/openai/_openai_realtime.py | 741 +++++++++++ .../labs/openai/test_openai_realtime.py | 1177 +++++++++++++++++ 4 files changed, 2082 insertions(+), 3 deletions(-) create mode 100644 src/google/adk/labs/openai/_openai_realtime.py create mode 100644 tests/unittests/labs/openai/test_openai_realtime.py diff --git a/src/google/adk/labs/openai/README.md b/src/google/adk/labs/openai/README.md index c40bedb8305..9777a3cfe5e 100644 --- a/src/google/adk/labs/openai/README.md +++ b/src/google/adk/labs/openai/README.md @@ -2,7 +2,14 @@ This folder contains an experimental integration for OpenAI models in ADK. -## Usage in Code +## Choosing an OpenAI API + +- `OpenAILlm` uses Chat Completions for regular (non-live) agent runs. +- `OpenAIResponsesLlm` uses the Responses API for regular agent runs. +- `OpenAILlm` with a Realtime model and `Runner.run_live()` uses the Realtime + API for bidirectional audio or text streaming. No separate runner is needed. + +## Chat Completions To use the OpenAI integration in your Python code, instantiate `OpenAILlm` and assign it to your agent's `model` field: @@ -21,4 +28,118 @@ agent = LlmAgent( ) ``` -Requires the `openai` Python package and `OPENAI_API_KEY` environment variable. +## Realtime + +The same `OpenAILlm` class supports OpenAI Realtime models through ADK's +standard live runner. The following example streams a raw PCM file to +`gpt-realtime` and writes the returned audio to another raw PCM file. + +Set `OPENAI_API_KEY` in the environment before running the example. The input +file must be headerless, little-endian PCM16, mono, at 24 kHz. + +```python +import asyncio +from contextlib import aclosing +from contextlib import suppress +from pathlib import Path + +from google.genai import types + +from google.adk.agents.live_request_queue import LiveRequestQueue +from google.adk.agents.llm_agent import Agent +from google.adk.agents.run_config import RunConfig +from google.adk.agents.run_config import StreamingMode +from google.adk.apps.app import App +from google.adk.labs.openai import OpenAILlm +from google.adk.runners import Runner +from google.adk.sessions.in_memory_session_service import InMemorySessionService + +APP_NAME = "openai_realtime_example" +USER_ID = "example_user" +SESSION_ID = "example_session" +INPUT_PCM = Path("input_24khz_mono_s16le.pcm") +OUTPUT_PCM = Path("output_24khz_mono_s16le.pcm") + +# 20 ms of mono PCM16 audio at 24 kHz. +CHUNK_BYTES = 24_000 * 2 * 20 // 1_000 + + +async def send_audio(queue: LiveRequestQueue) -> None: + with INPUT_PCM.open("rb") as input_file: + while chunk := input_file.read(CHUNK_BYTES): + queue.send_realtime( + types.Blob(data=chunk, mime_type="audio/pcm;rate=24000") + ) + await asyncio.sleep(0.02) + queue.send_audio_stream_end() + + +async def main() -> None: + agent = Agent( + name="openai_realtime_agent", + model=OpenAILlm(model="gpt-realtime"), + instruction="You are a concise and helpful voice assistant.", + ) + app = App(name=APP_NAME, root_agent=agent) + session_service = InMemorySessionService() + await session_service.create_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=SESSION_ID, + ) + queue = LiveRequestQueue() + run_config = RunConfig( + streaming_mode=StreamingMode.BIDI, + response_modalities=[types.Modality.AUDIO], + ) + + async with Runner(app=app, session_service=session_service) as runner: + sender = asyncio.create_task(send_audio(queue)) + try: + with OUTPUT_PCM.open("wb") as output_file: + async with aclosing( + runner.run_live( + user_id=USER_ID, + session_id=SESSION_ID, + live_request_queue=queue, + run_config=run_config, + ) + ) as events: + async for event in events: + if event.output_transcription and event.output_transcription.text: + print(event.output_transcription.text, end="", flush=True) + + for part in (event.content.parts or []) if event.content else []: + if ( + part.inline_data + and part.inline_data.mime_type.startswith("audio/pcm") + ): + output_file.write(part.inline_data.data or b"") + + if event.turn_complete: + break + finally: + queue.close() + if not sender.done(): + sender.cancel() + with suppress(asyncio.CancelledError): + await sender + + +asyncio.run(main()) +``` + +To receive text instead, set `response_modalities` to +`[types.Modality.TEXT]` and read `part.text` from the yielded events. Realtime +supports one output modality per run: audio or text. + +### Realtime Scope and Limitations + +- Realtime currently targets the public OpenAI API only. Azure OpenAI is not + supported by this integration. +- ADK session resumption is not mapped to OpenAI Realtime sessions. +- The integration does not receive client playout timing, so interruption does + not truncate conversation state to the exact amount of audio already played. + +The integration requires the `openai` Python package and the `OPENAI_API_KEY` +environment variable. diff --git a/src/google/adk/labs/openai/_openai_llm.py b/src/google/adk/labs/openai/_openai_llm.py index 524827fd6e1..1076f678038 100644 --- a/src/google/adk/labs/openai/_openai_llm.py +++ b/src/google/adk/labs/openai/_openai_llm.py @@ -16,6 +16,9 @@ from __future__ import annotations +from collections.abc import Awaitable +from collections.abc import Callable +import contextlib import copy from functools import cached_property import json @@ -41,9 +44,11 @@ ) from e from pydantic import BaseModel +from pydantic import Field from typing_extensions import override from ...models.base_llm import BaseLlm +from ...models.base_llm_connection import BaseLlmConnection from ...models.llm_request import LlmRequest from ...models.llm_response import LlmResponse from ._openai_schema import enforce_strict_openai_schema @@ -329,10 +334,17 @@ class OpenAILlm(BaseLlm): Attributes: model: The name of the OpenAI model. max_tokens: The maximum number of tokens to generate. + api_key: An OpenAI API key or asynchronous key provider. + client: A pre-configured asynchronous OpenAI client. When provided, this + takes precedence over ``api_key``. """ model: str = "gpt-4o" max_tokens: int = 4096 + api_key: str | Callable[[], Awaitable[str]] | None = Field( + default=None, exclude=True, repr=False + ) + client: AsyncOpenAI | None = Field(default=None, exclude=True, repr=False) @classmethod @override @@ -491,6 +503,34 @@ async def _generate_content_streaming( partial=False, ) + @contextlib.asynccontextmanager + async def connect( # type: ignore[override] + self, llm_request: LlmRequest + ) -> AsyncGenerator[BaseLlmConnection, None]: + """Connects to the OpenAI Realtime API. + + Args: + llm_request: The request whose live configuration is applied to the + Realtime session. + + Yields: + A live model connection driven by ADK's existing live flow. + """ + # Imported lazily to keep the unary Chat Completions integration isolated + # from the optional Realtime WebSocket dependency until live mode is used. + from ._openai_realtime import _OpenAIRealtimeLlmConnection + + model = llm_request.model or self.model + async with self._openai_client.realtime.connect(model=model) as session: + connection = _OpenAIRealtimeLlmConnection( + session, + model_version=model, + ) + await connection.configure(llm_request) + yield connection + @cached_property def _openai_client(self) -> AsyncOpenAI: - return AsyncOpenAI() + if self.client is not None: + return self.client + return AsyncOpenAI(api_key=self.api_key) diff --git a/src/google/adk/labs/openai/_openai_realtime.py b/src/google/adk/labs/openai/_openai_realtime.py new file mode 100644 index 00000000000..860022eb168 --- /dev/null +++ b/src/google/adk/labs/openai/_openai_realtime.py @@ -0,0 +1,741 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Private OpenAI Realtime connection used by :class:`OpenAILlm`.""" + +from __future__ import annotations + +import base64 +from collections.abc import AsyncGenerator +from collections.abc import Mapping +import json +import logging +import re +from typing import Any +from typing import cast +from typing import Union + +from google.genai import types +from openai.resources.realtime.realtime import AsyncRealtimeConnection +from openai.types.chat import ChatCompletionToolParam +from openai.types.realtime import session_update_event_param + +from ...models.base_llm_connection import BaseLlmConnection +from ...models.llm_request import LlmRequest +from ...models.llm_response import LlmResponse +from ._openai_llm import _function_declaration_to_openai_tool + +logger = logging.getLogger('google_adk.' + __name__) + +_AUDIO_SAMPLE_RATE_HZ = 24_000 +_AUDIO_MIME_TYPE = 'audio/pcm;rate=24000' +_REALTIME_TRANSCRIPTION_MODEL = 'gpt-live-transcribe' + +_RealtimeInput = Union[ + types.Blob, + types.ActivityStart, + types.ActivityEnd, + types.LiveClientRealtimeInput, +] + + +def _get_value(obj: object, key: str, default: Any = None) -> Any: + if obj is None: + return default + if isinstance(obj, Mapping): + return obj.get(key, default) + return getattr(obj, key, default) + + +def _text_of(content: types.Content) -> str: + return ''.join(part.text or '' for part in content.parts or []) + + +def _instruction_text(value: types.ContentUnion | None) -> str: + if isinstance(value, str): + return value + if isinstance(value, types.Content): + return _text_of(value) + if isinstance(value, types.Part): + return value.text or '' + if isinstance(value, Mapping): + return _instruction_text(types.Part(**value)) + if isinstance(value, list): + return ''.join( + _instruction_text(item) + for item in value + if isinstance(item, (str, types.Part, Mapping)) + ) + return '' + + +def _function_output(value: object) -> str: + if value is None: + return '' + if isinstance(value, str): + return value + return json.dumps(value, default=str, ensure_ascii=False) + + +def _function_arguments(value: str | None) -> dict[str, Any]: + if not value: + return {} + try: + result = json.loads(value) + except json.JSONDecodeError: + logger.warning('Failed to parse Realtime function arguments as JSON.') + return {} + return result if isinstance(result, dict) else {} + + +def _usage_metadata( + usage: object, +) -> types.GenerateContentResponseUsageMetadata: + input_details = _get_value(usage, 'input_token_details') + return types.GenerateContentResponseUsageMetadata( + prompt_token_count=_get_value(usage, 'input_tokens'), + candidates_token_count=_get_value(usage, 'output_tokens'), + total_token_count=_get_value(usage, 'total_tokens'), + cached_content_token_count=_get_value(input_details, 'cached_tokens'), + ) + + +def _voice_name(config: types.LiveConnectConfig) -> str | None: + speech = config.speech_config + voice_config = speech.voice_config if speech else None + prebuilt = voice_config.prebuilt_voice_config if voice_config else None + return prebuilt.voice_name if prebuilt else None + + +def _transcription_languages( + config: types.AudioTranscriptionConfig, +) -> list[str]: + hints = config.language_hints + languages = hints.language_codes if hints and hints.language_codes else None + languages = languages or config.language_codes or [] + normalized = [ + language.strip().lower().split('-', 1)[0] + for language in languages + if language.strip() + ] + return list(dict.fromkeys(normalized)) + + +def _input_transcription( + config: types.LiveConnectConfig, +) -> dict[str, Any] | None: + transcription_config = config.input_audio_transcription + if transcription_config is None: + return None + result: dict[str, Any] = {'model': _REALTIME_TRANSCRIPTION_MODEL} + languages = _transcription_languages(transcription_config) + if languages: + result['languages'] = languages + if transcription_config.adaptation_phrases: + result['keywords'] = transcription_config.adaptation_phrases + return result + + +def _automatic_activity_detection( + config: types.LiveConnectConfig, +) -> tuple[bool, dict[str, Any] | None]: + realtime = config.realtime_input_config + detection = realtime.automatic_activity_detection if realtime else None + if detection and detection.disabled: + return False, None + + result: dict[str, Any] = { + 'type': 'server_vad', + 'create_response': True, + 'interrupt_response': not ( + realtime + and realtime.activity_handling + == types.ActivityHandling.NO_INTERRUPTION + ), + } + if detection and detection.prefix_padding_ms is not None: + result['prefix_padding_ms'] = detection.prefix_padding_ms + if detection and detection.silence_duration_ms is not None: + result['silence_duration_ms'] = detection.silence_duration_ms + return True, result + + +def _interruptions_allowed(config: types.LiveConnectConfig) -> bool: + realtime = config.realtime_input_config + return not ( + realtime + and realtime.activity_handling == types.ActivityHandling.NO_INTERRUPTION + ) + + +def _realtime_tools(llm_request: LlmRequest) -> list[dict[str, Any]]: + tools: list[dict[str, Any]] = [] + for tool in llm_request.config.tools or []: + if not isinstance(tool, types.Tool): + continue + for declaration in tool.function_declarations or []: + chat_tool: ChatCompletionToolParam = _function_declaration_to_openai_tool( + declaration + ) + function = chat_tool['function'] + realtime_tool: dict[str, Any] = { + 'type': 'function', + 'name': function['name'], + 'description': function.get('description', ''), + 'parameters': function.get( + 'parameters', {'type': 'object', 'properties': {}} + ), + } + tools.append(realtime_tool) + return tools + + +def _session_payload( + llm_request: LlmRequest, +) -> tuple[dict[str, Any], bool, bool]: + config = llm_request.live_connect_config + instructions = _instruction_text(config.system_instruction) + if not instructions: + instructions = _instruction_text(llm_request.config.system_instruction) + + modalities = { + str(getattr(modality, 'value', modality)).lower() + for modality in config.response_modalities or [] + } + if not modalities or modalities == {'audio'}: + output_modalities = ['audio'] + elif modalities == {'text'}: + output_modalities = ['text'] + else: + raise ValueError( + 'OpenAI Realtime supports exactly one output modality: audio or text.' + ) + + automatic_detection, turn_detection = _automatic_activity_detection(config) + audio_input: dict[str, Any] = { + 'format': {'type': 'audio/pcm', 'rate': _AUDIO_SAMPLE_RATE_HZ}, + 'turn_detection': turn_detection, + } + transcription = _input_transcription(config) + if transcription: + audio_input['transcription'] = transcription + + audio_output: dict[str, Any] = { + 'format': {'type': 'audio/pcm', 'rate': _AUDIO_SAMPLE_RATE_HZ}, + } + voice = _voice_name(config) + if voice: + audio_output['voice'] = voice + + session: dict[str, Any] = { + 'type': 'realtime', + 'output_modalities': output_modalities, + 'audio': {'input': audio_input, 'output': audio_output}, + } + if instructions: + session['instructions'] = instructions + if config.max_output_tokens is not None: + session['max_output_tokens'] = config.max_output_tokens + tools = _realtime_tools(llm_request) + if tools: + session['tools'] = tools + session['tool_choice'] = 'auto' + return session, automatic_detection, _interruptions_allowed(config) + + +def _conversation_items(content: types.Content) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + text = _text_of(content) + role = 'assistant' if content.role in ('model', 'assistant') else 'user' + if text: + content_type = 'output_text' if role == 'assistant' else 'input_text' + items.append({ + 'type': 'message', + 'role': role, + 'content': [{'type': content_type, 'text': text}], + }) + + for part in content.parts or []: + if part.function_call: + call = part.function_call + items.append({ + 'type': 'function_call', + 'call_id': call.id or '', + 'name': call.name or '', + 'arguments': json.dumps( + call.args or {}, default=str, ensure_ascii=False + ), + }) + elif part.function_response: + response = part.function_response + items.append({ + 'type': 'function_call_output', + 'call_id': response.id or '', + 'output': _function_output(response.response), + }) + return items + + +def _pcm16_24khz(blob: types.Blob) -> bytes: + mime_type = (blob.mime_type or '').lower().replace(' ', '') + if not mime_type.startswith('audio/pcm'): + raise ValueError('OpenAI Realtime only supports PCM16 audio input.') + rate = re.search(r'(?:^|;)rate=(\d+)(?:;|$)', mime_type) + if not rate or int(rate.group(1)) != _AUDIO_SAMPLE_RATE_HZ: + raise ValueError( + 'OpenAI Realtime audio input must declare a 24000 Hz sample rate; ' + f'use {_AUDIO_MIME_TYPE!r}.' + ) + data = blob.data + if not isinstance(data, bytes): + raise ValueError('OpenAI Realtime audio input must contain bytes.') + if len(data) % 2: + raise ValueError('PCM16 audio input must contain complete 16-bit samples.') + return data + + +def _plain_text_only(content: types.Content) -> bool: + """Whether content can be safely merged with a partial text turn.""" + return all( + set(part.model_dump(exclude_none=True)) <= {'text'} + for part in content.parts or [] + ) + + +def _output_text(item: object) -> str: + text_parts: list[str] = [] + for content_part in _get_value(item, 'content', []) or []: + if _get_value(content_part, 'type') == 'output_text': + text = _get_value(content_part, 'text') + if text: + text_parts.append(str(text)) + return ''.join(text_parts) + + +def _output_index(event: object) -> int | None: + value = _get_value(event, 'output_index') + return ( + value if isinstance(value, int) and not isinstance(value, bool) else None + ) + + +class _OpenAIRealtimeLlmConnection(BaseLlmConnection): + """Maps ADK's live connection contract onto the OpenAI Realtime SDK.""" + + def __init__( + self, + session: AsyncRealtimeConnection, + *, + model_version: str, + ) -> None: + self._session = session + self._model_version = model_version + self._live_session_id: str | None = None + self._automatic_detection = True + self._interruptions_allowed = True + self._text_output = False + self._response_active = False + self._interruption_announced = False + self._announced_calls: set[str] = set() + self._pending_calls: dict[str, object] = {} + self._pending_text = '' + self._pending_text_output_index: int | None = None + self._partial_input_text = '' + self._closed = False + + async def configure(self, llm_request: LlmRequest) -> None: + """Applies ADK's live configuration to the connected session.""" + ( + session, + self._automatic_detection, + self._interruptions_allowed, + ) = _session_payload(llm_request) + self._text_output = session['output_modalities'] == ['text'] + # gpt-live-transcribe's ``languages`` and ``keywords`` fields can precede + # the SDK's generated TypedDicts. The SDK still validates and serializes + # this ordinary payload at the WebSocket boundary. + await self._session.session.update( + session=cast(session_update_event_param.Session, session) + ) + + async def send_history(self, history: list[types.Content]) -> None: + last_content_sent = False + for index, content in enumerate(history): + items = _conversation_items(content) + for item in items: + await self._session.conversation.item.create(item=cast(Any, item)) + if index == len(history) - 1: + last_content_sent = bool(items) + if history and history[-1].role == 'user' and last_content_sent: + await self._session.response.create() + + async def send_content(self, content: types.Content) -> None: + await self._send_content(content) + + async def _send_content( + self, content: types.Content, *, partial: bool = False + ) -> None: + if partial: + if not _plain_text_only(content): + raise ValueError( + 'Partial OpenAI Realtime content may only contain text parts.' + ) + self._partial_input_text += _text_of(content) + return + + items: list[dict[str, Any]] = [] + if self._partial_input_text: + if _plain_text_only(content): + content = types.Content( + role=content.role or 'user', + parts=[ + types.Part.from_text( + text=self._partial_input_text + _text_of(content) + ) + ], + ) + else: + items.extend( + _conversation_items( + types.Content( + role='user', + parts=[types.Part.from_text(text=self._partial_input_text)], + ) + ) + ) + self._partial_input_text = '' + items.extend(_conversation_items(content)) + for item in items: + await self._session.conversation.item.create(item=cast(Any, item)) + if items: + await self._session.response.create() + + async def send_realtime(self, input: _RealtimeInput) -> None: + if isinstance(input, types.Blob): + audio = _pcm16_24khz(input) + await self._session.input_audio_buffer.append( + audio=base64.b64encode(audio).decode('ascii') + ) + return + if isinstance(input, types.ActivityStart): + if self._response_active and self._interruptions_allowed: + await self._session.response.cancel() + if not self._automatic_detection: + await self._session.input_audio_buffer.clear() + return + if isinstance(input, types.ActivityEnd): + if not self._automatic_detection: + await self._commit_audio_and_respond() + return + if isinstance(input, types.LiveClientRealtimeInput): + if input.audio_stream_end: + await self._commit_audio_and_respond() + return + logger.warning('Unary LiveClientRealtimeInput not fully supported yet.') + return + raise ValueError(f'Unsupported input type: {type(input)}') + + async def _commit_audio_and_respond(self) -> None: + await self._session.input_audio_buffer.commit() + await self._session.response.create() + + async def receive(self) -> AsyncGenerator[LlmResponse, None]: + async for event in self._session: + event_type = str(_get_value(event, 'type', '')) + + if event_type in ('session.created', 'session.updated'): + session = _get_value(event, 'session') + self._live_session_id = _get_value(session, 'id') or ( + self._live_session_id + ) + continue + + if event_type == 'response.created': + self._response_active = True + self._interruption_announced = False + continue + + if event_type == 'response.output_audio.delta': + delta = _get_value(event, 'delta') + try: + audio = base64.b64decode(delta, validate=True) + except (TypeError, ValueError): + logger.warning('Discarding an invalid Realtime audio delta.') + continue + yield self._response( + content=types.Content( + role='model', + parts=[ + types.Part( + inline_data=types.Blob( + data=audio, mime_type=_AUDIO_MIME_TYPE + ) + ) + ], + ), + ) + continue + + if event_type == 'response.output_text.delta': + delta = str(_get_value(event, 'delta') or '') + if delta: + self._pending_text += delta + output_index = _output_index(event) + if output_index is not None: + self._pending_text_output_index = output_index + yield self._response( + content=types.Content( + role='model', parts=[types.Part.from_text(text=delta)] + ), + partial=True, + ) + continue + + if event_type == 'response.output_audio_transcript.delta': + text = str(_get_value(event, 'delta') or '') + if text: + yield self._response( + output_transcription=types.Transcription( + text=text, finished=False + ), + partial=True, + ) + continue + + if event_type == 'response.output_audio_transcript.done': + text = str(_get_value(event, 'transcript') or '') + yield self._response( + output_transcription=types.Transcription(text=text, finished=True), + partial=False, + ) + continue + + if event_type == 'conversation.item.input_audio_transcription.delta': + text = str(_get_value(event, 'delta') or '') + if text: + yield self._response( + input_transcription=types.Transcription( + text=text, finished=False + ), + partial=True, + ) + continue + + if event_type == 'conversation.item.input_audio_transcription.completed': + text = str(_get_value(event, 'transcript') or '') + yield self._response( + input_transcription=types.Transcription(text=text, finished=True), + partial=False, + ) + continue + + if event_type == 'response.function_call_arguments.done': + self._buffer_function_call(event) + continue + + if event_type == 'input_audio_buffer.speech_started': + interrupted = ( + self._response_active + and self._interruptions_allowed + and not self._interruption_announced + ) + yield self._response( + voice_activity=types.VoiceActivity( + voice_activity_type=types.VoiceActivityType.ACTIVITY_START, + audio_offset=f"{_get_value(event, 'audio_start_ms', 0)}ms", + ), + ) + if interrupted: + # Text deltas are consolidated into an interrupted final Content at + # response.done. For audio-only output, surface interruption now. + if not self._text_output: + self._interruption_announced = True + yield self._response(interrupted=True) + continue + + if event_type == 'input_audio_buffer.speech_stopped': + yield self._response( + voice_activity=types.VoiceActivity( + voice_activity_type=types.VoiceActivityType.ACTIVITY_END, + audio_offset=f"{_get_value(event, 'audio_end_ms', 0)}ms", + ) + ) + continue + + if event_type == 'error': + error = _get_value(event, 'error') + code = str( + _get_value(error, 'code') + or _get_value(error, 'type') + or 'realtime_error' + ) + yield self._response( + error_code=code, + error_message=f'OpenAI Realtime error ({code})', + finish_reason=types.FinishReason.OTHER, + ) + continue + + if event_type == 'response.done': + async for response in self._response_done(event): + yield response + return + + if not self._closed: + yield self._response( + go_away=types.LiveServerGoAway(time_left='0s'), + ) + + async def _response_done( + self, event: object + ) -> AsyncGenerator[LlmResponse, None]: + self._response_active = False + response = _get_value(event, 'response') + status = str(_get_value(response, 'status') or 'completed') + parts = self._final_response_parts( + response, include_function_calls=status == 'completed' + ) + self._pending_calls.clear() + self._pending_text = '' + self._pending_text_output_index = None + + interruption_emitted_with_content = False + if parts: + interruption_emitted_with_content = status == 'cancelled' + yield self._response( + content=types.Content(role='model', parts=parts), + partial=False, + interrupted=interruption_emitted_with_content or None, + ) + + interrupted = ( + status == 'cancelled' + and not self._interruption_announced + and not interruption_emitted_with_content + ) + usage = _get_value(response, 'usage') + status_details = _get_value(response, 'status_details') + incomplete_reason = _get_value(status_details, 'reason') + finish_reason = types.FinishReason.STOP + if status == 'incomplete': + finish_reason = ( + types.FinishReason.SAFETY + if incomplete_reason == 'content_filter' + else types.FinishReason.MAX_TOKENS + ) + kwargs: dict[str, Any] = { + 'turn_complete': True, + 'interrupted': interrupted, + 'finish_reason': finish_reason, + } + if usage: + kwargs['usage_metadata'] = _usage_metadata(usage) + if status == 'failed': + error = _get_value(status_details, 'error') + kwargs.update( + error_code=str(_get_value(error, 'code') or 'response_failed'), + error_message='OpenAI Realtime response failed', + finish_reason=types.FinishReason.OTHER, + ) + yield self._response(**kwargs) + self._interruption_announced = False + + def _final_response_parts( + self, response: object, *, include_function_calls: bool + ) -> list[types.Part]: + """Builds one ordered final content from a Realtime response.""" + ordered_parts: list[tuple[int, int, types.Part]] = [] + seen_call_ids: set[str] = set() + text_added = False + output = list(_get_value(response, 'output', []) or []) + + for sequence, item in enumerate(output): + item_type = _get_value(item, 'type') + if item_type == 'message': + text = _output_text(item) + if text: + ordered_parts.append( + (sequence, sequence, types.Part.from_text(text=text)) + ) + text_added = True + elif item_type == 'function_call' and include_function_calls: + function_part = self._function_call_part(item) + if function_part is not None: + call_id, part = function_part + if call_id in seen_call_ids: + continue + seen_call_ids.add(call_id) + ordered_parts.append((sequence, sequence, part)) + + fallback_position = len(output) + if self._pending_text and not text_added: + text_position = self._pending_text_output_index + ordered_parts.append(( + text_position if text_position is not None else fallback_position, + -1, + types.Part.from_text(text=self._pending_text), + )) + + if include_function_calls: + for sequence, pending_call in enumerate(self._pending_calls.values()): + function_part = self._function_call_part(pending_call) + if function_part is None: + continue + call_id, part = function_part + if call_id in seen_call_ids: + continue + call_position = _output_index(pending_call) + ordered_parts.append(( + call_position + if call_position is not None + else fallback_position + sequence, + sequence, + part, + )) + seen_call_ids.add(call_id) + + self._announced_calls.update(seen_call_ids) + ordered_parts.sort(key=lambda value: (value[0], value[1])) + return [part for _, _, part in ordered_parts] + + def _buffer_function_call(self, event: object) -> None: + call_id = str(_get_value(event, 'call_id') or _get_value(event, 'id') or '') + if call_id and call_id not in self._announced_calls: + self._pending_calls[call_id] = event + + def _function_call_part(self, event: object) -> tuple[str, types.Part] | None: + call_id = str(_get_value(event, 'call_id') or _get_value(event, 'id') or '') + if not call_id or call_id in self._announced_calls: + return None + part = types.Part( + function_call=types.FunctionCall( + id=call_id, + name=str(_get_value(event, 'name') or ''), + args=_function_arguments(_get_value(event, 'arguments')), + ) + ) + return call_id, part + + def _response(self, **kwargs: Any) -> LlmResponse: + return LlmResponse( + model_version=self._model_version, + live_session_id=self._live_session_id, + **kwargs, + ) + + async def close(self) -> None: + if self._closed: + return + self._closed = True + await self._session.close() diff --git a/tests/unittests/labs/openai/test_openai_realtime.py b/tests/unittests/labs/openai/test_openai_realtime.py new file mode 100644 index 00000000000..1f03c12c6d5 --- /dev/null +++ b/tests/unittests/labs/openai/test_openai_realtime.py @@ -0,0 +1,1177 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Hermetic tests for the OpenAI Realtime live-model connection.""" + +from __future__ import annotations + +import asyncio +import base64 +from collections import deque +from collections.abc import AsyncIterator +from types import SimpleNamespace +from typing import Any +from unittest import mock + +from google.adk import Runner +from google.adk.agents import Agent +from google.adk.agents import RunConfig +from google.adk.agents.live_request_queue import LiveRequestQueue +from google.adk.agents.run_config import StreamingMode +from google.adk.apps import App +from google.adk.labs.openai._openai_llm import OpenAILlm +from google.adk.labs.openai._openai_realtime import _OpenAIRealtimeLlmConnection +from google.adk.models.llm_request import LlmRequest +from google.adk.sessions import InMemorySessionService +from google.genai import types +from openai import AsyncOpenAI +import pytest + + +class _AsyncCall: + """Records an SDK resource method call.""" + + def __init__(self, session: _FakeRealtimeSession, name: str) -> None: + self._session = session + self._name = name + + async def __call__(self, **kwargs: Any) -> None: + await self._session.record_call(self._name, kwargs) + + +class _FakeRealtimeSession: + """Small in-memory double for ``AsyncRealtimeConnection``.""" + + def __init__(self, events: list[object] | None = None) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + self.close_count = 0 + self._events: deque[object] = deque(events or []) + self.session = SimpleNamespace(update=_AsyncCall(self, "session.update")) + self.conversation = SimpleNamespace( + item=SimpleNamespace( + create=_AsyncCall(self, "conversation.item.create") + ) + ) + self.input_audio_buffer = SimpleNamespace( + append=_AsyncCall(self, "input_audio_buffer.append"), + commit=_AsyncCall(self, "input_audio_buffer.commit"), + clear=_AsyncCall(self, "input_audio_buffer.clear"), + ) + self.response = SimpleNamespace( + create=_AsyncCall(self, "response.create"), + cancel=_AsyncCall(self, "response.cancel"), + ) + + def __aiter__(self) -> AsyncIterator[object]: + async def iterate() -> AsyncIterator[object]: + while self._events: + yield self._events.popleft() + + return iterate() + + async def close(self) -> None: + self.close_count += 1 + + async def record_call(self, name: str, kwargs: dict[str, Any]) -> None: + self.calls.append((name, kwargs)) + + def calls_named(self, name: str) -> list[dict[str, Any]]: + return [kwargs for call_name, kwargs in self.calls if call_name == name] + + +class _RoundTripRealtimeSession(_FakeRealtimeSession): + """Provider double that waits for a complete parallel-tool round trip.""" + + def __init__( + self, + first_turn: list[object], + second_turn: list[object], + *, + expected_tool_outputs: int, + ) -> None: + super().__init__() + self._first_turn = first_turn + self._second_turn = second_turn + self._expected_tool_outputs = expected_tool_outputs + self._tool_outputs_ready = asyncio.Event() + self._phase = 0 + self.response_created_before_all_tool_outputs = False + + def __aiter__(self) -> AsyncIterator[object]: + async def iterate() -> AsyncIterator[object]: + if self._phase == 0: + self._phase = 1 + for event in self._first_turn: + yield event + if self._phase == 1: + await self._tool_outputs_ready.wait() + self._phase = 2 + for event in self._second_turn: + yield event + await asyncio.Event().wait() + + return iterate() + + async def record_call(self, name: str, kwargs: dict[str, Any]) -> None: + await super().record_call(name, kwargs) + outputs = [ + call + for call in self.calls_named("conversation.item.create") + if call["item"]["type"] == "function_call_output" + ] + if name == "response.create" and len(outputs) < self._expected_tool_outputs: + self.response_created_before_all_tool_outputs = True + if ( + name == "response.create" + and len(outputs) == self._expected_tool_outputs + ): + self._tool_outputs_ready.set() + + +class _FakeConnectionManager: + + def __init__(self, realtime: _FakeRealtime, session: _FakeRealtimeSession): + self._realtime = realtime + self._session = session + + async def __aenter__(self) -> _FakeRealtimeSession: + self._realtime.enter_count += 1 + return self._session + + async def __aexit__(self, *unused_args: object) -> None: + self._realtime.exit_count += 1 + + +class _FakeRealtime: + + def __init__(self, session: _FakeRealtimeSession) -> None: + self._session = session + self.models: list[str] = [] + self.enter_count = 0 + self.exit_count = 0 + + def connect(self, *, model: str) -> _FakeConnectionManager: + self.models.append(model) + return _FakeConnectionManager(self, self._session) + + +def _openai_llm( + session: _FakeRealtimeSession, +) -> tuple[OpenAILlm, _FakeRealtime]: + realtime = _FakeRealtime(session) + client = mock.Mock(spec=AsyncOpenAI) + client.realtime = realtime + return OpenAILlm(model="gpt-realtime", client=client), realtime + + +def _live_config( + *, + automatic_detection: bool = True, + modalities: list[types.Modality] | None = None, +) -> types.LiveConnectConfig: + return types.LiveConnectConfig( + response_modalities=modalities or [types.Modality.AUDIO], + max_output_tokens=321, + speech_config=types.SpeechConfig( + voice_config=types.VoiceConfig( + prebuilt_voice_config=types.PrebuiltVoiceConfig( + voice_name="marin" + ) + ) + ), + input_audio_transcription=types.AudioTranscriptionConfig( + language_hints=types.LanguageHints(language_codes=["it", "it"]), + adaptation_phrases=["Agent Development Kit"], + ), + realtime_input_config=types.RealtimeInputConfig( + automatic_activity_detection=types.AutomaticActivityDetection( + disabled=not automatic_detection, + prefix_padding_ms=240, + silence_duration_ms=480, + ) + ), + ) + + +def _request( + *, + automatic_detection: bool = True, + modalities: list[types.Modality] | None = None, +) -> LlmRequest: + tool = types.Tool( + function_declarations=[ + types.FunctionDeclaration( + name="get_weather", + description="Read the weather", + parameters=types.Schema( + type=types.Type.OBJECT, + properties={ + "city": types.Schema(type=types.Type.STRING), + }, + required=["city"], + ), + ) + ] + ) + return LlmRequest( + model="gpt-realtime-test", + config=types.GenerateContentConfig( + system_instruction="Be concise.", tools=[tool] + ), + live_connect_config=_live_config( + automatic_detection=automatic_detection, + modalities=modalities, + ), + ) + + +def _connection( + session: _FakeRealtimeSession, +) -> _OpenAIRealtimeLlmConnection: + return _OpenAIRealtimeLlmConnection( + session, # type: ignore[arg-type] + model_version="gpt-realtime-test", + ) + + +async def _responses( + session: _FakeRealtimeSession, +) -> list[Any]: + responses = [] + async for response in _connection(session).receive(): + responses.append(response) + if response.turn_complete: + break + return responses + + +def _done( + *, + status: str = "completed", + **response: Any, +) -> dict[str, Any]: + return {"type": "response.done", "response": {"status": status, **response}} + + +async def test_configure_sends_openai_realtime_session_payload() -> None: + session = _FakeRealtimeSession() + + await _connection(session).configure(_request()) + + payload = session.calls_named("session.update")[0]["session"] + assert payload["type"] == "realtime" + assert payload["instructions"] == "Be concise." + assert payload["output_modalities"] == ["audio"] + assert payload["max_output_tokens"] == 321 + assert payload["audio"]["input"] == { + "format": {"type": "audio/pcm", "rate": 24000}, + "turn_detection": { + "type": "server_vad", + "create_response": True, + "interrupt_response": True, + "prefix_padding_ms": 240, + "silence_duration_ms": 480, + }, + "transcription": { + "model": "gpt-live-transcribe", + "languages": ["it"], + "keywords": ["Agent Development Kit"], + }, + } + assert payload["audio"]["output"]["voice"] == "marin" + assert payload["tools"] == [{ + "type": "function", + "name": "get_weather", + "description": "Read the weather", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + }] + assert payload["tool_choice"] == "auto" + + +async def test_configure_supports_text_output_and_manual_vad() -> None: + session = _FakeRealtimeSession() + + await _connection(session).configure( + _request( + automatic_detection=False, + modalities=[types.Modality.TEXT], + ) + ) + + payload = session.calls_named("session.update")[0]["session"] + assert payload["output_modalities"] == ["text"] + assert payload["audio"]["input"]["turn_detection"] is None + + +async def test_configure_defaults_empty_modalities_to_audio() -> None: + session = _FakeRealtimeSession() + request = _request() + request.live_connect_config.response_modalities = [] + + await _connection(session).configure(request) + + payload = session.calls_named("session.update")[0]["session"] + assert payload["output_modalities"] == ["audio"] + + +async def test_configure_rejects_multiple_or_unsupported_modalities() -> None: + for modalities in ( + [types.Modality.AUDIO, types.Modality.TEXT], + [types.Modality.IMAGE], + ): + request = _request() + request.live_connect_config.response_modalities = modalities + with pytest.raises(ValueError, match="exactly one output modality"): + await _connection(_FakeRealtimeSession()).configure(request) + + +async def test_no_interruption_activity_handling_configures_server_vad() -> ( + None +): + session = _FakeRealtimeSession() + request = _request() + request.live_connect_config.realtime_input_config.activity_handling = ( + types.ActivityHandling.NO_INTERRUPTION + ) + + await _connection(session).configure(request) + + payload = session.calls_named("session.update")[0]["session"] + detection = payload["audio"]["input"]["turn_detection"] + assert detection["interrupt_response"] is False + + +def test_async_api_key_provider_is_delegated_to_openai_sdk() -> None: + called = False + + async def provider() -> str: + nonlocal called + called = True + return "dynamic-key" + + with mock.patch( + "google.adk.labs.openai._openai_llm.AsyncOpenAI" + ) as client_class: + llm = OpenAILlm(model="gpt-realtime", api_key=provider) + _ = llm._openai_client + + assert called is False + client_class.assert_called_once_with(api_key=provider) + + +def test_credentials_and_injected_client_are_not_serialized_or_repr() -> None: + client = mock.Mock(spec=AsyncOpenAI) + llm = OpenAILlm( + model="gpt-realtime", + api_key="sk-test-secret", + client=client, + ) + + dumped = llm.model_dump() + rendered = repr(llm) + + assert "api_key" not in dumped + assert "client" not in dumped + assert "sk-test-secret" not in rendered + assert repr(client) not in rendered + + +async def test_openai_llm_connect_uses_request_model_and_exits_context() -> ( + None +): + for raise_inside in (False, True): + session = _FakeRealtimeSession() + llm, realtime = _openai_llm(session) + + async def use_connection() -> None: + async with llm.connect(_request()): + if raise_inside: + raise RuntimeError("caller failed") + + if raise_inside: + with pytest.raises(RuntimeError, match="caller failed"): + await use_connection() + else: + await use_connection() + + assert realtime.models == ["gpt-realtime-test"] + assert realtime.enter_count == 1 + assert realtime.exit_count == 1 + assert len(session.calls_named("session.update")) == 1 + + +async def test_send_history_replays_items_and_answers_last_user() -> None: + session = _FakeRealtimeSession() + connection = _connection(session) + call = types.Part.from_function_call( + name="get_weather", args={"city": "Rome"} + ) + call.function_call.id = "call-1" + history = [ + types.Content( + role="model", + parts=[types.Part.from_text(text="Hello"), call], + ), + types.Content(role="user", parts=[types.Part.from_text(text="Weather?")]), + ] + + await connection.send_history(history) + + items = [ + call["item"] for call in session.calls_named("conversation.item.create") + ] + assert items == [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hello"}], + }, + { + "type": "function_call", + "call_id": "call-1", + "name": "get_weather", + "arguments": '{"city": "Rome"}', + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "Weather?"}], + }, + ] + assert len(session.calls_named("response.create")) == 1 + + +async def test_send_history_does_not_answer_last_assistant() -> None: + session = _FakeRealtimeSession() + + await _connection(session).send_history( + [types.Content(role="model", parts=[types.Part.from_text(text="Done")])] + ) + + assert not session.calls_named("response.create") + + +async def test_partial_text_is_coalesced_before_sending() -> None: + session = _FakeRealtimeSession() + connection = _connection(session) + + await connection._send_content( + types.Content(parts=[types.Part.from_text(text="hel")]), partial=True + ) + await connection._send_content( + types.Content(parts=[types.Part.from_text(text="lo")]), partial=True + ) + assert not session.calls + + await connection._send_content( + types.Content(parts=[types.Part.from_text(text="!")]), partial=False + ) + + item = session.calls_named("conversation.item.create")[0]["item"] + assert item["content"] == [{"type": "input_text", "text": "hello!"}] + assert len(session.calls_named("response.create")) == 1 + + +async def test_function_output_is_sent_as_conversation_item() -> None: + session = _FakeRealtimeSession() + response = types.Part.from_function_response( + name="get_weather", response={"temperature": 25} + ) + response.function_response.id = "call-1" + + await _connection(session).send_content( + types.Content(role="tool", parts=[response]) + ) + + item = session.calls_named("conversation.item.create")[0]["item"] + assert item == { + "type": "function_call_output", + "call_id": "call-1", + "output": '{"temperature": 25}', + } + assert len(session.calls_named("response.create")) == 1 + + +async def test_partial_text_does_not_replace_following_function_output() -> ( + None +): + session = _FakeRealtimeSession() + connection = _connection(session) + response = types.Part.from_function_response( + name="get_weather", response={"temperature": 25} + ) + response.function_response.id = "call-1" + + await connection._send_content( + types.Content(parts=[types.Part.from_text(text="context")]), + partial=True, + ) + await connection._send_content( + types.Content(role="tool", parts=[response]), partial=False + ) + + items = [ + call["item"] for call in session.calls_named("conversation.item.create") + ] + assert items == [ + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "context"}], + }, + { + "type": "function_call_output", + "call_id": "call-1", + "output": '{"temperature": 25}', + }, + ] + assert len(session.calls_named("response.create")) == 1 + + +async def test_pcm16_audio_is_base64_encoded_for_openai() -> None: + session = _FakeRealtimeSession() + + await _connection(session).send_realtime( + types.Blob(data=b"\x00\x01\x02\x03", mime_type="audio/pcm; rate=24000") + ) + + assert session.calls_named("input_audio_buffer.append") == [ + {"audio": base64.b64encode(b"\x00\x01\x02\x03").decode("ascii")} + ] + + +@pytest.mark.parametrize( + ("blob", "message"), + [ + ( + types.Blob(data=b"\x00\x00", mime_type="audio/wav;rate=24000"), + "only supports PCM16", + ), + ( + types.Blob(data=b"\x00\x00", mime_type="audio/pcm;rate=16000"), + "24000 Hz", + ), + ( + types.Blob(data=b"\x00", mime_type="audio/pcm;rate=24000"), + "complete 16-bit samples", + ), + ], +) +async def test_invalid_pcm_audio_is_rejected( + blob: types.Blob, + message: str, +) -> None: + with pytest.raises(ValueError, match=message): + await _connection(_FakeRealtimeSession()).send_realtime(blob) + + +@pytest.mark.parametrize("response_active", [False, True]) +async def test_activity_start_cancels_only_an_active_response( + response_active: bool, +) -> None: + session = _FakeRealtimeSession() + connection = _connection(session) + connection._response_active = response_active + + await connection.send_realtime(types.ActivityStart()) + + assert bool(session.calls_named("response.cancel")) is response_active + assert not session.calls_named("input_audio_buffer.clear") + + +async def test_activity_start_clears_audio_buffer_with_manual_vad() -> None: + session = _FakeRealtimeSession() + connection = _connection(session) + connection._automatic_detection = False + + await connection.send_realtime(types.ActivityStart()) + + assert len(session.calls_named("input_audio_buffer.clear")) == 1 + + +async def test_no_interruption_prevents_manual_activity_start_cancel() -> None: + session = _FakeRealtimeSession() + connection = _connection(session) + request = _request(automatic_detection=False) + request.live_connect_config.realtime_input_config.activity_handling = ( + types.ActivityHandling.NO_INTERRUPTION + ) + await connection.configure(request) + connection._response_active = True + + await connection.send_realtime(types.ActivityStart()) + + assert not session.calls_named("response.cancel") + assert len(session.calls_named("input_audio_buffer.clear")) == 1 + + +@pytest.mark.parametrize("automatic_detection", [False, True]) +async def test_activity_end_commits_only_with_manual_vad( + automatic_detection: bool, +) -> None: + session = _FakeRealtimeSession() + connection = _connection(session) + connection._automatic_detection = automatic_detection + + await connection.send_realtime(types.ActivityEnd()) + + expected = not automatic_detection + assert bool(session.calls_named("input_audio_buffer.commit")) is expected + assert bool(session.calls_named("response.create")) is expected + + +async def test_audio_stream_end_commits_and_requests_response() -> None: + session = _FakeRealtimeSession() + + await _connection(session).send_realtime( + types.LiveClientRealtimeInput(audio_stream_end=True) + ) + + assert len(session.calls_named("input_audio_buffer.commit")) == 1 + assert len(session.calls_named("response.create")) == 1 + + +async def test_receive_maps_audio_and_streaming_text() -> None: + session = _FakeRealtimeSession([ + {"type": "session.created", "session": {"id": "session-1"}}, + {"type": "response.created"}, + { + "type": "response.output_audio.delta", + "delta": base64.b64encode(b"\x01\x02").decode("ascii"), + }, + {"type": "response.output_text.delta", "delta": "hel"}, + {"type": "response.output_text.delta", "delta": "lo"}, + _done(), + ]) + + responses = await _responses(session) + + audio = responses[0] + assert audio.content.parts[0].inline_data == types.Blob( # type: ignore[index,union-attr] + data=b"\x01\x02", mime_type="audio/pcm;rate=24000" + ) + assert [response.content.parts[0].text for response in responses[1:3]] == [ # type: ignore[index,union-attr] + "hel", + "lo", + ] + assert responses[3].partial is False + assert responses[3].content.parts[0].text == "hello" # type: ignore[index,union-attr] + assert responses[4].turn_complete is True + assert all(response.live_session_id == "session-1" for response in responses) + + +async def test_receive_discards_invalid_audio_delta() -> None: + session = _FakeRealtimeSession([ + {"type": "response.output_audio.delta", "delta": "not base64!"}, + _done(), + ]) + + responses = await _responses(session) + + assert len(responses) == 1 + assert responses[0].turn_complete is True + + +async def test_receive_maps_input_and_output_transcriptions() -> None: + session = _FakeRealtimeSession([ + { + "type": "conversation.item.input_audio_transcription.delta", + "delta": "ciao ", + }, + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "ciao mondo", + }, + {"type": "response.output_audio_transcript.delta", "delta": "salve "}, + { + "type": "response.output_audio_transcript.done", + "transcript": "salve a te", + }, + _done(), + ]) + + responses = await _responses(session) + + assert ( + responses[0].input_transcription.text, + responses[0].input_transcription.finished, + responses[0].partial, + ) == ("ciao ", False, True) + assert ( + responses[1].input_transcription.text, + responses[1].input_transcription.finished, + responses[1].partial, + ) == ("ciao mondo", True, False) + assert ( + responses[2].output_transcription.text, + responses[2].output_transcription.finished, + responses[2].partial, + ) == ("salve ", False, True) + assert ( + responses[3].output_transcription.text, + responses[3].output_transcription.finished, + responses[3].partial, + ) == ("salve a te", True, False) + + +async def test_tool_call_is_deduplicated_between_delta_done_and_response() -> ( + None +): + function_call = { + "type": "function_call", + "call_id": "call-1", + "name": "get_weather", + "arguments": "not-json", + } + session = _FakeRealtimeSession([ + { + **function_call, + "type": "response.function_call_arguments.done", + }, + _done(output=[function_call]), + ]) + + responses = await _responses(session) + + calls = [ + response.content.parts[0].function_call + for response in responses + if response.content and response.content.parts[0].function_call + ] + assert len(calls) == 1 + assert calls[0].id == "call-1" + assert calls[0].name == "get_weather" + assert calls[0].args == {} + + +async def test_completed_response_aggregates_parallel_calls_in_provider_order() -> ( + None +): + first_call = { + "type": "function_call", + "call_id": "call-1", + "name": "get_weather", + "arguments": '{"city": "Rome"}', + } + second_call = { + "type": "function_call", + "call_id": "call-2", + "name": "get_weather", + "arguments": '{"city": "Milan"}', + } + session = _FakeRealtimeSession([ + {"type": "response.output_text.delta", "delta": "Checking both."}, + { + **second_call, + "type": "response.function_call_arguments.done", + }, + { + **first_call, + "type": "response.function_call_arguments.done", + }, + _done( + output=[ + { + "type": "message", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "Checking both."} + ], + }, + first_call, + second_call, + ] + ), + ]) + + responses = await _responses(session) + + final_content = [ + response.content + for response in responses + if response.content and response.partial is False + ] + assert len(final_content) == 1 + parts = final_content[0].parts + assert [ + part.text or (part.function_call.id if part.function_call else None) + for part in parts + ] == ["Checking both.", "call-1", "call-2"] + assert [part.function_call.args for part in parts if part.function_call] == [ + {"city": "Rome"}, + {"city": "Milan"}, + ] + + +async def test_completed_response_maps_usage_including_cached_tokens() -> None: + session = _FakeRealtimeSession([ + _done( + usage={ + "input_tokens": 12, + "output_tokens": 7, + "total_tokens": 19, + "input_token_details": {"cached_tokens": 5}, + } + ) + ]) + + response = (await _responses(session))[0] + + assert response.finish_reason == types.FinishReason.STOP + assert response.usage_metadata.prompt_token_count == 12 + assert response.usage_metadata.candidates_token_count == 7 + assert response.usage_metadata.total_token_count == 19 + assert response.usage_metadata.cached_content_token_count == 5 + + +async def test_response_done_statuses_are_mapped() -> None: + cases = [ + (_done(status="incomplete"), types.FinishReason.MAX_TOKENS, False, None), + (_done(status="cancelled"), types.FinishReason.STOP, True, None), + ( + _done( + status="failed", + status_details={ + "error": {"code": "server_error", "message": "Try again"} + }, + ), + types.FinishReason.OTHER, + False, + "server_error", + ), + ] + for event, finish_reason, interrupted, error_code in cases: + response = (await _responses(_FakeRealtimeSession([event])))[0] + assert response.turn_complete is True + assert response.finish_reason == finish_reason + assert response.interrupted is interrupted + assert response.error_code == error_code + + +async def test_cancelled_response_consolidates_text_and_interrupts_once() -> ( + None +): + session = _FakeRealtimeSession([ + {"type": "response.created"}, + {"type": "response.output_text.delta", "delta": "cut off"}, + _done(status="cancelled"), + ]) + + responses = await _responses(session) + + content_responses = [response for response in responses if response.content] + assert len(content_responses) == 2 + assert content_responses[0].partial is True + assert content_responses[0].content.parts[0].text == "cut off" + assert content_responses[1].partial is False + assert content_responses[1].content.parts[0].text == "cut off" + assert content_responses[1].interrupted is True + terminal = responses[-1] + assert terminal.interrupted is False + assert terminal.turn_complete is True + assert terminal.content is None + assert sum(response.interrupted is True for response in responses) == 1 + + +async def test_text_interruption_waits_for_late_delta_and_emits_once() -> None: + session = _FakeRealtimeSession([ + {"type": "response.created"}, + {"type": "input_audio_buffer.speech_started", "audio_start_ms": 125}, + {"type": "response.output_text.delta", "delta": "late delta"}, + _done(status="cancelled"), + ]) + connection = _connection(session) + await connection.configure(_request(modalities=[types.Modality.TEXT])) + + responses = [response async for response in connection.receive()] + + assert responses[0].voice_activity.voice_activity_type == ( + types.VoiceActivityType.ACTIVITY_START + ) + assert sum(response.interrupted is True for response in responses) == 1 + final_content = [ + response + for response in responses + if response.content and response.partial is False + ] + assert len(final_content) == 1 + assert final_content[0].content.parts[0].text == "late delta" + assert final_content[0].interrupted is True + + +async def test_speech_started_interrupts_only_during_response() -> None: + for response_active in (False, True): + events: list[object] = [] + if response_active: + events.append({"type": "response.created"}) + events.extend([ + {"type": "input_audio_buffer.speech_started", "audio_start_ms": 125}, + _done(status="cancelled" if response_active else "completed"), + ]) + + responses = await _responses(_FakeRealtimeSession(events)) + response = responses[0] + assert response.voice_activity.voice_activity_type == ( + types.VoiceActivityType.ACTIVITY_START + ) + assert response.voice_activity.audio_offset == "125ms" + if response_active: + assert responses[1].interrupted is True + else: + assert response.interrupted is None + + +async def test_speech_stopped_reports_activity_end() -> None: + session = _FakeRealtimeSession([ + {"type": "input_audio_buffer.speech_stopped", "audio_end_ms": 925}, + _done(), + ]) + + response = (await _responses(session))[0] + + assert response.voice_activity.voice_activity_type == ( + types.VoiceActivityType.ACTIVITY_END + ) + assert response.voice_activity.audio_offset == "925ms" + + +async def test_no_interruption_speech_started_only_reports_activity() -> None: + session = _FakeRealtimeSession([ + {"type": "response.created"}, + {"type": "input_audio_buffer.speech_started", "audio_start_ms": 125}, + _done(), + ]) + connection = _connection(session) + request = _request() + request.live_connect_config.realtime_input_config.activity_handling = ( + types.ActivityHandling.NO_INTERRUPTION + ) + await connection.configure(request) + + responses = [] + async for response in connection.receive(): + responses.append(response) + if response.turn_complete: + break + + activity = responses[0] + assert activity.voice_activity.voice_activity_type == ( + types.VoiceActivityType.ACTIVITY_START + ) + assert all(response.interrupted is not True for response in responses) + + +async def test_provider_error_is_recoverable_within_session() -> None: + session = _FakeRealtimeSession([ + { + "type": "error", + "error": {"code": "invalid_value", "message": "Bad request"}, + }, + {"type": "response.output_text.delta", "delta": "recovered"}, + _done(), + ]) + + responses = await _responses(session) + + assert responses[0].error_code == "invalid_value" + assert responses[0].error_message == "OpenAI Realtime error (invalid_value)" + assert responses[0].finish_reason == types.FinishReason.OTHER + assert responses[1].content.parts[0].text == "recovered" # type: ignore[index,union-attr] + assert responses[-1].turn_complete is True + + +async def test_close_is_idempotent() -> None: + session = _FakeRealtimeSession() + connection = _connection(session) + + await connection.close() + await connection.close() + + assert session.close_count == 1 + + +async def test_clean_provider_eof_emits_one_go_away() -> None: + responses = [ + response + async for response in _connection(_FakeRealtimeSession()).receive() + ] + + assert len(responses) == 1 + assert responses[0].go_away == types.LiveServerGoAway(time_left="0s") + + +async def test_runner_run_live_consumes_openai_realtime_connection() -> None: + session = _FakeRealtimeSession([ + {"type": "session.created", "session": {"id": "session-runner"}}, + { + "type": "conversation.item.input_audio_transcription.completed", + "transcript": "hello", + }, + {"type": "response.output_text.delta", "delta": "Hi"}, + _done(), + ]) + llm, realtime = _openai_llm(session) + agent = Agent(name="assistant", model=llm) + session_service = InMemorySessionService() + await session_service.create_session( + app_name="realtime-test", user_id="user", session_id="session" + ) + runner = Runner( + app=App(name="realtime-test", root_agent=agent), + session_service=session_service, + ) + queue = LiveRequestQueue() + stream = runner.run_live( + user_id="user", + session_id="session", + live_request_queue=queue, + run_config=RunConfig( + streaming_mode=StreamingMode.BIDI, + response_modalities=[types.Modality.TEXT], + input_audio_transcription=types.AudioTranscriptionConfig(), + ), + ) + events = [] + + async def consume_turn() -> None: + async for event in stream: + events.append(event) + if event.turn_complete: + break + + try: + await asyncio.wait_for(consume_turn(), timeout=2) + finally: + queue.close() + await stream.aclose() + + assert realtime.enter_count == 1 + assert realtime.exit_count == 1 + assert any(event.author == "user" for event in events) + assert any( + event.content + and event.content.parts + and event.content.parts[0].text == "Hi" + for event in events + ) + assert any(event.turn_complete for event in events) + + +async def test_runner_round_trips_parallel_tools_before_next_response() -> None: + first_call = { + "type": "function_call", + "call_id": "call-rome", + "name": "get_weather", + "arguments": '{"city": "Rome"}', + } + second_call = { + "type": "function_call", + "call_id": "call-milan", + "name": "get_weather", + "arguments": '{"city": "Milan"}', + } + session = _RoundTripRealtimeSession( + [ + {"type": "response.created"}, + _done(output=[first_call, second_call]), + ], + [ + {"type": "response.created"}, + {"type": "response.output_text.delta", "delta": "Both checked."}, + _done( + output=[{ + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Both checked."}], + }] + ), + ], + expected_tool_outputs=2, + ) + tool_invocations: list[str] = [] + + def get_weather(city: str) -> dict[str, str]: + """Returns deterministic weather for a city.""" + tool_invocations.append(city) + return {"city": city, "condition": "sunny"} + + llm, _ = _openai_llm(session) + agent = Agent(name="assistant", model=llm, tools=[get_weather]) + session_service = InMemorySessionService() + await session_service.create_session( + app_name="realtime-tools", user_id="user", session_id="session" + ) + runner = Runner( + app=App(name="realtime-tools", root_agent=agent), + session_service=session_service, + ) + queue = LiveRequestQueue() + stream = runner.run_live( + user_id="user", + session_id="session", + live_request_queue=queue, + run_config=RunConfig( + streaming_mode=StreamingMode.BIDI, + response_modalities=[types.Modality.TEXT], + ), + ) + events = [] + final_text_seen = False + + async def consume_tool_round_trip() -> None: + nonlocal final_text_seen + async for event in stream: + events.append(event) + if event.content and any( + part.text == "Both checked." for part in event.content.parts or [] + ): + final_text_seen = True + if final_text_seen and event.turn_complete: + break + + try: + await asyncio.wait_for(consume_tool_round_trip(), timeout=2) + finally: + queue.close() + await stream.aclose() + + call_events = [event for event in events if event.get_function_calls()] + response_events = [ + event for event in events if event.get_function_responses() + ] + assert len(call_events) == 1 + assert [call.id for call in call_events[0].get_function_calls()] == [ + "call-rome", + "call-milan", + ] + assert len(response_events) == 1 + assert [ + response.id for response in response_events[0].get_function_responses() + ] == ["call-rome", "call-milan"] + assert sorted(tool_invocations) == ["Milan", "Rome"] + assert session.response_created_before_all_tool_outputs is False + assert len(session.calls_named("response.create")) == 1 + output_items = [ + call["item"] + for call in session.calls_named("conversation.item.create") + if call["item"]["type"] == "function_call_output" + ] + assert [item["call_id"] for item in output_items] == [ + "call-rome", + "call-milan", + ] From 5843f633b941051b05d8ef4f45a3318dc7364229 Mon Sep 17 00:00:00 2001 From: GitMarco27 Date: Mon, 3 Aug 2026 18:24:30 +0200 Subject: [PATCH 2/3] docs: make Realtime file streaming deterministic --- src/google/adk/labs/openai/README.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/google/adk/labs/openai/README.md b/src/google/adk/labs/openai/README.md index 9777a3cfe5e..723ac52b51d 100644 --- a/src/google/adk/labs/openai/README.md +++ b/src/google/adk/labs/openai/README.md @@ -91,6 +91,13 @@ async def main() -> None: run_config = RunConfig( streaming_mode=StreamingMode.BIDI, response_modalities=[types.Modality.AUDIO], + # Treat the complete file as one turn. This prevents pauses inside a + # prerecorded clip from triggering server-side voice activity detection. + realtime_input_config=types.RealtimeInputConfig( + automatic_activity_detection=types.AutomaticActivityDetection( + disabled=True + ) + ), ) async with Runner(app=app, session_service=session_service) as runner: From c8a30059d922bc90d3a828ce906beef7af423d0d Mon Sep 17 00:00:00 2001 From: GitMarco27 Date: Tue, 4 Aug 2026 09:57:26 +0200 Subject: [PATCH 3/3] docs: avoid duplicate Realtime transcript output --- src/google/adk/labs/openai/README.md | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/google/adk/labs/openai/README.md b/src/google/adk/labs/openai/README.md index 723ac52b51d..dfa6d4aeb45 100644 --- a/src/google/adk/labs/openai/README.md +++ b/src/google/adk/labs/openai/README.md @@ -113,7 +113,11 @@ async def main() -> None: ) ) as events: async for event in events: - if event.output_transcription and event.output_transcription.text: + if ( + event.output_transcription + and event.output_transcription.finished + and event.output_transcription.text + ): print(event.output_transcription.text, end="", flush=True) for part in (event.content.parts or []) if event.content else []: