feat: add OpenAI Realtime support to live runs

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,129 @@ 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],
+      # 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:
+    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.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 []:
+              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.

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)