chore(live): move audio and transcription managers into google.adk.live

Migrate `AudioCacheManager`, `AudioCacheConfig`, `TranscriptionManager`, and `AudioTranscriber` from `google.adk.flows.llm_flows` to `google.adk.live`. Backward-compatible re-exports are maintained in `google.adk.flows.llm_flows` for all moved symbols.

`RealtimeCacheEntry` also moves, from `google.adk.agents.invocation_context` to `google.adk.live._audio_cache_manager`, because the cache manager is its only real owner. `invocation_context` re-exports it, so that import path keeps working too.

This is a pure move: the moved code, including `RealtimeCacheEntry`'s pydantic config, is unchanged, and nothing is added to `google.adk.live.__all__` -- the moved symbols are reached through their module path, so the move commits to no new public API. `google.adk.live` does not import from `google.adk.flows.llm_flows`, so the two modules that needed `require_agent_name` carry a private copy rather than importing upward.

Co-authored-by: Shangjie Chen <deanchen@google.com>
PiperOrigin-RevId: 972094217

Author: DeanChensj

src/google/adk/agents/invocation_context.py

@@ -32,6 +32,7 @@
 from ..auth.credential_service.base_credential_service import BaseCredentialService
 from ..events._branch_path import _BranchPath
 from ..events.event import Event
+from ..live._audio_cache_manager import RealtimeCacheEntry as RealtimeCacheEntry
 from ..live.live_request_queue import LiveRequestQueue
 from ..memory.base_memory_service import BaseMemoryService
 from ..plugins.plugin_manager import PluginManager
@@ -53,25 +54,6 @@ class LlmCallsLimitExceededError(Exception):
   """Error thrown when the number of LLM calls exceed the limit."""
 
 
-class RealtimeCacheEntry(BaseModel):
-  """Store audio data chunks for caching before flushing."""
-
-  model_config = ConfigDict(
-      arbitrary_types_allowed=True,
-      extra="forbid",
-  )
-  """The pydantic model config."""
-
-  role: str
-  """The role that created this audio data, typically "user" or "model"."""
-
-  data: types.Blob
-  """The audio data chunk."""
-
-  timestamp: float
-  """Timestamp when the audio chunk was received."""
-
-
 class _InvocationCostManager(BaseModel):
   """A container to keep track of the cost of invocation.
 

src/google/adk/flows/llm_flows/audio_cache_manager.py

@@ -12,269 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from __future__ import annotations
+"""Backward compatibility module for AudioCacheManager.
 
-import logging
-from typing import TYPE_CHECKING
+AudioCacheManager and AudioCacheConfig have been moved to
+``google.adk.live.audio_cache_manager``.
+"""
 
-from google.adk.platform import time as platform_time
-from google.genai import types
+from __future__ import annotations
 
-from ...agents.invocation_context import RealtimeCacheEntry
-from ...events.event import Event
-from ._invocation_utils import require_agent_name
+import logging
 
-if TYPE_CHECKING:
-  from ...agents.invocation_context import InvocationContext
+from ...live._audio_cache_manager import AudioCacheConfig as AudioCacheConfig
+from ...live._audio_cache_manager import AudioCacheManager as AudioCacheManager
+from ...live._audio_cache_manager import RealtimeCacheEntry as RealtimeCacheEntry
 
 logger = logging.getLogger('google_adk.' + __name__)
-
-
-def _require_audio_data(blob: types.Blob) -> bytes:
-  data = blob.data
-  if not isinstance(data, bytes):
-    raise ValueError('Audio blobs must contain byte data.')
-  return data
-
-
-class AudioCacheManager:
-  """Manages audio caching and flushing for live streaming flows."""
-
-  def __init__(self, config: AudioCacheConfig | None = None) -> None:
-    """Initialize the audio cache manager.
-
-    Args:
-      config: Configuration for audio caching behavior.
-    """
-    self.config = config or AudioCacheConfig()
-
-  def cache_audio(
-      self,
-      invocation_context: InvocationContext,
-      audio_blob: types.Blob,
-      cache_type: str,
-  ) -> None:
-    """Cache incoming user or outgoing model audio data.
-
-    Args:
-      invocation_context: The current invocation context.
-      audio_blob: The audio data to cache.
-      cache_type: Type of audio to cache, either 'input' or 'output'.
-
-    Raises:
-      ValueError: If cache_type is not 'input' or 'output'.
-    """
-    audio_data = _require_audio_data(audio_blob)
-    if cache_type == 'input':
-      if not invocation_context.input_realtime_cache:
-        invocation_context.input_realtime_cache = []
-      cache = invocation_context.input_realtime_cache
-      role = 'user'
-    elif cache_type == 'output':
-      if not invocation_context.output_realtime_cache:
-        invocation_context.output_realtime_cache = []
-      cache = invocation_context.output_realtime_cache
-      role = 'model'
-    else:
-      raise ValueError("cache_type must be either 'input' or 'output'")
-
-    audio_entry = RealtimeCacheEntry(
-        role=role, data=audio_blob, timestamp=platform_time.get_time()
-    )
-    cache.append(audio_entry)
-
-    logger.debug(
-        'Cached %s audio chunk: %d bytes, cache size: %d',
-        cache_type,
-        len(audio_data),
-        len(cache),
-    )
-
-  async def flush_caches(
-      self,
-      invocation_context: InvocationContext,
-      flush_user_audio: bool = True,
-      flush_model_audio: bool = True,
-  ) -> list[Event]:
-    """Flush audio caches to artifact services.
-
-    The multimodality data is saved in artifact service in the format of
-    audio file. The file data reference is added to the session as an event.
-    The audio file follows the naming convention: artifact_ref =
-    f"artifact://{invocation_context.app_name}/{invocation_context.user_id}/
-    {invocation_context.session.id}/_adk_live/{filename}#{revision_id}"
-
-    Note: video data is not supported yet.
-
-    Args:
-      invocation_context: The invocation context containing audio caches.
-      flush_user_audio: Whether to flush the input (user) audio cache.
-      flush_model_audio: Whether to flush the output (model) audio cache.
-
-    Returns:
-      A list of Event objects created from the flushed caches.
-    """
-    flushed_events: list[Event] = []
-    if flush_user_audio and invocation_context.input_realtime_cache:
-      audio_event = await self._flush_cache_to_services(
-          invocation_context,
-          invocation_context.input_realtime_cache,
-          'input_audio',
-      )
-      if audio_event:
-        flushed_events.append(audio_event)
-        invocation_context.input_realtime_cache = []
-
-    if flush_model_audio and invocation_context.output_realtime_cache:
-      logger.debug('Flushed output audio cache')
-      audio_event = await self._flush_cache_to_services(
-          invocation_context,
-          invocation_context.output_realtime_cache,
-          'output_audio',
-      )
-      if audio_event:
-        flushed_events.append(audio_event)
-        invocation_context.output_realtime_cache = []
-
-    return flushed_events
-
-  async def _flush_cache_to_services(
-      self,
-      invocation_context: InvocationContext,
-      audio_cache: list[RealtimeCacheEntry],
-      cache_type: str,
-  ) -> Event | None:
-    """Flush a list of audio cache entries to artifact services.
-
-    The artifact service stores the actual blob. The session stores the
-    reference to the stored blob.
-
-    Args:
-      invocation_context: The invocation context.
-      audio_cache: The audio cache to flush.
-      cache_type: Type identifier for the cache ('input_audio' or 'output_audio').
-
-    Returns:
-      The created Event if the cache was successfully flushed, None otherwise.
-    """
-    if not invocation_context.artifact_service or not audio_cache:
-      logger.debug('Skipping cache flush: no artifact service or empty cache')
-      return None
-
-    try:
-      # Combine audio chunks into a single file. Use join rather than repeated
-      # `+=`, which is O(n^2) over the total audio size.
-      mime_type = audio_cache[0].data.mime_type or 'audio/pcm'
-      combined_audio_data = b''.join(
-          entry.data.data or b'' for entry in audio_cache
-      )
-
-      # Generate filename with timestamp from first audio chunk (when recording started)
-      timestamp = int(audio_cache[0].timestamp * 1000)  # milliseconds
-      filename = f"adk_live_audio_storage_{cache_type}_{timestamp}.{mime_type.split('/')[-1]}"
-
-      # Save to artifact service
-      combined_audio_part = types.Part(
-          inline_data=types.Blob(data=combined_audio_data, mime_type=mime_type)
-      )
-
-      revision_id = await invocation_context.artifact_service.save_artifact(
-          app_name=invocation_context.app_name,
-          user_id=invocation_context.user_id,
-          session_id=invocation_context.session.id,
-          filename=filename,
-          artifact=combined_audio_part,
-      )
-
-      # Create artifact reference for session service
-      artifact_ref = f'artifact://{invocation_context.app_name}/{invocation_context.user_id}/{invocation_context.session.id}/_adk_live/{filename}#{revision_id}'
-
-      # Create event with file data reference to add to session
-      # For model events, author should be the agent name, not the role
-      author = (
-          require_agent_name(invocation_context)
-          if audio_cache[0].role == 'model'
-          else audio_cache[0].role
-      )
-      audio_event = Event(
-          id=Event.new_id(),
-          invocation_id=invocation_context.invocation_id,
-          author=author,
-          content=types.Content(
-              role=audio_cache[0].role,
-              parts=[
-                  types.Part(
-                      file_data=types.FileData(
-                          file_uri=artifact_ref, mime_type=mime_type
-                      )
-                  )
-              ],
-          ),
-          timestamp=audio_cache[0].timestamp,
-      )
-
-      logger.debug(
-          'Successfully flushed %s cache: %d chunks, %d bytes, saved as %s',
-          cache_type,
-          len(audio_cache),
-          len(combined_audio_data),
-          filename,
-      )
-      return audio_event
-
-    except Exception as e:
-      logger.error('Failed to flush %s cache: %s', cache_type, e)
-      return None
-
-  def get_cache_stats(
-      self, invocation_context: InvocationContext
-  ) -> dict[str, int]:
-    """Get statistics about current cache state.
-
-    Args:
-      invocation_context: The invocation context.
-
-    Returns:
-      Dictionary containing cache statistics.
-    """
-    input_count = len(invocation_context.input_realtime_cache or [])
-    output_count = len(invocation_context.output_realtime_cache or [])
-
-    input_bytes = sum(
-        len(_require_audio_data(entry.data))
-        for entry in invocation_context.input_realtime_cache or []
-    )
-    output_bytes = sum(
-        len(_require_audio_data(entry.data))
-        for entry in invocation_context.output_realtime_cache or []
-    )
-
-    return {
-        'input_chunks': input_count,
-        'output_chunks': output_count,
-        'input_bytes': input_bytes,
-        'output_bytes': output_bytes,
-        'total_chunks': input_count + output_count,
-        'total_bytes': input_bytes + output_bytes,
-    }
-
-
-class AudioCacheConfig:
-  """Configuration for audio caching behavior."""
-
-  def __init__(
-      self,
-      max_cache_size_bytes: int = 10 * 1024 * 1024,  # 10MB
-      max_cache_duration_seconds: float = 300.0,  # 5 minutes
-      auto_flush_threshold: int = 100,  # Number of chunks
-  ) -> None:
-    """Initialize audio cache configuration.
-
-    Args:
-      max_cache_size_bytes: Maximum cache size in bytes before auto-flush.
-      max_cache_duration_seconds: Maximum duration to keep data in cache.
-      auto_flush_threshold: Number of chunks that triggers auto-flush.
-    """
-    self.max_cache_size_bytes = max_cache_size_bytes
-    self.max_cache_duration_seconds = max_cache_duration_seconds
-    self.auto_flush_threshold = auto_flush_threshold