Update GeminiLLMService to work with LLMContext and LLMContextAggregatorPair

CHANGELOG.md

@@ -7,6 +7,59 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Expanded support for universal `LLMContext` to `GeminiLiveLLMService`.
+  As a reminder, the context-setup pattern when using `LLMContext` is:
+
+  ```python
+  context = LLMContext(messages, tools)
+  context_aggregator = LLMContextAggregatorPair(
+    context,
+    # This part is `GeminiLiveLLMService`-specific.
+    # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+    #  modality (the default).
+    assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+  )
+  ```
+
+  (Note that even though `GeminiLiveLLMService` now supports the universal
+  `LLMContext`, it is not meant to be swapped out for another LLM service at
+  runtime.)
+
+  Worth noting: whether or not you use the new context-setup pattern with
+  `GeminiLiveLLMService`, some types have changed under the hood:
+
+  ```python
+  ## BEFORE:
+
+  # Context aggregator type
+  context_aggregator: GeminiLiveContextAggregatorPair
+
+  # Context frame type
+  frame: OpenAILLMContextFrame
+
+  # Context type
+  context: GeminiLiveLLMContext
+  # or
+  context: OpenAILLMContext
+
+  ## AFTER:
+
+  # Context aggregator type
+  context_aggregator: LLMContextAggregatorPair
+
+  # Context frame type
+  frame: LLMContextFrame
+
+  # Context type
+  context: LLMContext
+  ```
+
+  Also note that `LLMTextFrame`s are no longer pushed from `GeminiLiveLLMService`
+  when it's configured with `modalities=GeminiModalities.AUDIO`. If you need
+  to process its output, listen for `TTSTextFrame`s instead.
+
 ### Changed
 
 - `FunctionFilter` now has a `filter_system_frames` arg, which controls whether

examples/foundational/26a-gemini-live-transcription.py

@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -72,7 +74,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         # inference_on_context_initialization=False,
     )
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [
             {
                 "role": "user",
@@ -90,7 +92,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             #     },
         ],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     transcript = TranscriptProcessor()
 

examples/foundational/26b-gemini-live-function-calling.py

@@ -19,7 +19,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -139,10 +141,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm.register_function("get_current_weather", fetch_weather_from_api)
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [{"role": "user", "content": "Say hello."}],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [

examples/foundational/26c-gemini-live-video.py

@@ -17,7 +17,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import (
     create_transport,
@@ -65,7 +67,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         # inference_on_context_initialization=False,
     )
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [
             {
                 "role": "user",
@@ -73,7 +75,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             },
         ],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [

examples/foundational/26d-gemini-live-text.py

@@ -16,7 +16,8 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -109,8 +110,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     # Set up conversation context and management
     # The context_aggregator will automatically collect conversation context
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    context = LLMContext(messages)
+    context_aggregator = LLMContextAggregatorPair(context)
 
     pipeline = Pipeline(
         [

examples/foundational/26e-gemini-live-google-search.py

@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -90,7 +92,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         tools=tools,
     )
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [
             {
                 "role": "user",
@@ -98,7 +100,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             }
         ],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [

examples/foundational/26f-gemini-live-files-api.py

@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -129,7 +131,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         mime_type = "text/plain"
 
         # Create context with file reference
-        context = OpenAILLMContext(
+        context = LLMContext(
             [
                 {
                     "role": "user",
@@ -152,7 +154,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     except Exception as e:
         logger.error(f"Error uploading file: {e}")
         # Continue with a basic context if file upload fails
-        context = OpenAILLMContext(
+        context = LLMContext(
             [
                 {
                     "role": "user",
@@ -162,7 +164,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         )
 
     # Create context aggregator
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     # Build the pipeline
     pipeline = Pipeline(

examples/foundational/26g-gemini-live-groundingMetadata.py

@@ -10,7 +10,9 @@ from pipecat.frames.frames import Frame, LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -124,8 +126,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     ]
 
     # Set up conversation context and management
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    context = LLMContext(messages)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [

examples/foundational/26h-gemini-live-vertex-function-calling.py

@@ -9,21 +9,21 @@ import os
 from datetime import datetime
 
 from dotenv import load_dotenv
-from google.genai.types import HttpOptions
 from loguru import logger
 
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
-from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
 from pipecat.services.google.gemini_live.llm_vertex import GeminiLiveVertexLLMService
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -139,10 +139,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm.register_function("get_current_weather", fetch_weather_from_api)
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
 
-    context = OpenAILLMContext(
-        [{"role": "user", "content": "Say hello."}],
+    context = LLMContext([{"role": "user", "content": "Say hello."}])
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
     )
-    context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [

examples/foundational/26i-gemini-live-graceful-end.py

@@ -18,7 +18,9 @@ from pipecat.frames.frames import EndTaskFrame, LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -152,10 +154,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
     llm.register_function("end_conversation", end_conversation)
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [{"role": "user", "content": "Say hello."}],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [

examples/foundational/46-video-processing.py

@@ -15,7 +15,9 @@ from pipecat.frames.frames import Frame, InputImageRawFrame, LLMRunFrame, Output
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor
 from pipecat.runner.types import RunnerArguments
@@ -108,8 +110,13 @@ async def run_bot(pipecat_transport):
         }
     ]
 
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    context = LLMContext(messages)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     # RTVI events for Pipecat client UI
     rtvi = RTVIProcessor()

src/pipecat/adapters/services/gemini_adapter.py

@@ -24,13 +24,7 @@ from pipecat.processors.aggregators.llm_context import (
 )
 
 try:
-    from google.genai.types import (
-        Blob,
-        Content,
-        FunctionCall,
-        FunctionResponse,
-        Part,
-    )
+    from google.genai.types import Blob, Content, FileData, FunctionCall, FunctionResponse, Part
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
@@ -309,6 +303,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                 parts.append(
                     Part(
                         function_call=FunctionCall(
+                            id=id,
                             name=name,
                             args=json.loads(tc["function"]["arguments"]),
                         )
@@ -334,9 +329,12 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                 function_name = params.tool_call_id_to_name_mapping[tool_call_id]
 
             parts.append(
-                Part.from_function_response(
-                    name=function_name,
-                    response=response_dict,
+                Part(
+                    function_response=FunctionResponse(
+                        id=tool_call_id,
+                        name=function_name,
+                        response=response_dict,
+                    )
                 )
             )
         elif isinstance(content, str):
@@ -358,6 +356,16 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                     input_audio = c["input_audio"]
                     audio_bytes = base64.b64decode(input_audio["data"])
                     parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes)))
+                elif c["type"] == "file_data":
+                    file_data = c["file_data"]
+                    parts.append(
+                        Part(
+                            file_data=FileData(
+                                mime_type=file_data.get("mime_type"),
+                                file_uri=file_data.get("file_uri"),
+                            )
+                        )
+                    )
 
         return self.MessageConversionResult(
             content=Content(role=role, parts=parts),

src/pipecat/services/google/gemini_live/llm.py

@@ -17,6 +17,7 @@ import json
 import random
 import time
 import uuid
+import warnings
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Dict, List, Optional, Union
@@ -56,10 +57,12 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
 )
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -219,6 +222,10 @@ class GeminiLiveContext(OpenAILLMContext):
 
     Provides Gemini-specific context management including system instruction
     extraction and message format conversion for the Live API.
+
+    .. deprecated:: 0.0.93
+        Gemini Live no longer uses `GeminiLiveContext` under the hood.
+        It now uses `LLMContext`.
     """
 
     @staticmethod
@@ -231,6 +238,22 @@ class GeminiLiveContext(OpenAILLMContext):
         Returns:
             The upgraded Gemini context instance.
         """
+        # This warning is here rather than `__init__` since `upgrade()` was the
+        # "main" way that GeminiLiveContext instances were created.
+        # Almost no users should be seeing this message anyway, as
+        # GeminiLiveContext instances were typically created under the hood:
+        # the user would pass an OpenAILLMContext instance, which would be
+        # upgraded without them necessarily knowing.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveContext is deprecated. "
+                "Gemini Live no longer uses GeminiLiveContext under the hood. "
+                "It now uses LLMContext.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
         if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GeminiLiveContext):
             logger.debug(f"Upgrading to Gemini Live Context: {obj}")
             obj.__class__ = GeminiLiveContext
@@ -328,8 +351,28 @@ class GeminiLiveUserContextAggregator(OpenAIUserContextAggregator):
 
     Extends OpenAI user aggregator to handle Gemini-specific message passing
     while maintaining compatibility with the standard aggregation pipeline.
+
+    .. deprecated:: 0.0.93
+        Gemini Live no longer expects a `GeminiLiveUserContextAggregator`.
+        It now expects a `LLMUserAggregator`.
     """
 
+    def __init__(self, *args, **kwargs):
+        """Initialize Gemini Live user context aggregator."""
+        # Almost no users should be seeing this message, as
+        # `GeminiLiveUserContextAggregator`` instances were typically created
+        # under the hood, as part of `llm.create_context_aggregator()`.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveUserContextAggregator is deprecated. "
+                "Gemini Live no longer expects a GeminiLiveUserContextAggregator. "
+                "It now expects a LLMUserAggregator.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        super().__init__(*args, **kwargs)
+
     async def process_frame(self, frame, direction):
         """Process incoming frames for user context aggregation.
 
@@ -349,8 +392,28 @@ class GeminiLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
     Handles assistant response aggregation while filtering out LLMTextFrames
     to prevent duplicate context entries, as Gemini Live pushes both
     LLMTextFrames and TTSTextFrames.
+
+    .. deprecated:: 0.0.93
+        Gemini Live no longer uses `GeminiLiveAssistantContextAggregator` under the hood.
+        It now uses `LLMAssistantAggregator`.
     """
 
+    def __init__(self, *args, **kwargs):
+        """Initialize Gemini Live assistant context aggregator."""
+        # Almost no users should be seeing this message, as
+        # `GeminiLiveAssistantContextAggregator` instances were typically
+        # created under the hood, as part of `llm.create_context_aggregator()`.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveAssistantContextAggregator is deprecated. "
+                "Gemini Live no longer uses GeminiLiveAssistantContextAggregator under the hood. "
+                "It now uses LLMAssistantAggregator.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        super().__init__(*args, **kwargs)
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process incoming frames for assistant context aggregation.
 
@@ -380,6 +443,10 @@ class GeminiLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
 class GeminiLiveContextAggregatorPair:
     """Pair of user and assistant context aggregators for Gemini Live.
 
+    .. deprecated:: 0.0.93
+        `GeminiLiveContextAggregatorPair` is deprecated.
+        Use `LLMContextAggregatorPair` instead.
+
     Parameters:
         _user: The user context aggregator instance.
         _assistant: The assistant context aggregator instance.
@@ -388,6 +455,19 @@ class GeminiLiveContextAggregatorPair:
     _user: GeminiLiveUserContextAggregator
     _assistant: GeminiLiveAssistantContextAggregator
 
+    def __post_init__(self):
+        # Almost no users should be seeing this message, as
+        # `GeminiLiveContextAggregatorPair` instances were typically created
+        # under the hood, with `llm.create_context_aggregator()`.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveContextAggregatorPair is deprecated. "
+                "Use LLMContextAggregatorPair instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
     def user(self) -> GeminiLiveUserContextAggregator:
         """Get the user context aggregator.
 
@@ -665,6 +745,9 @@ class GeminiLiveLLMService(LLMService):
         # Initialize the API client. Subclasses can override this if needed.
         self.create_client()
 
+        # Bookkeeping for tool calls
+        self._completed_tool_calls = set()
+
     def create_client(self):
         """Create the Gemini API client instance. Subclasses can override this."""
         self._client = Client(api_key=self._api_key, http_options=self._http_options)
@@ -787,9 +870,10 @@ class GeminiLiveLLMService(LLMService):
     #
 
     async def _handle_interruption(self):
-        await self._set_bot_is_speaking(False)
-        await self.push_frame(TTSStoppedFrame())
-        await self.push_frame(LLMFullResponseEndFrame())
+        if self._bot_is_speaking:
+            await self._set_bot_is_speaking(False)
+            await self.push_frame(TTSStoppedFrame())
+            await self.push_frame(LLMFullResponseEndFrame())
 
     async def _handle_user_started_speaking(self, frame):
         self._user_is_speaking = True
@@ -807,7 +891,6 @@ class GeminiLiveLLMService(LLMService):
 
     #
     # frame processing
-    #
     # StartFrame, StopFrame, CancelFrame implemented in base class
     #
 
@@ -829,22 +912,13 @@ class GeminiLiveLLMService(LLMService):
 
         if isinstance(frame, TranscriptionFrame):
             await self.push_frame(frame, direction)
-        elif isinstance(frame, OpenAILLMContextFrame):
-            context: GeminiLiveContext = GeminiLiveContext.upgrade(frame.context)
-            # For now, we'll only trigger inference here when either:
-            #   1. We have not seen a context frame before
-            #   2. The last message is a tool call result
-            if not self._context:
-                self._context = context
-                if frame.context.tools:
-                    self._tools = frame.context.tools
-                await self._create_initial_response()
-            elif context.messages and context.messages[-1].get("role") == "tool":
-                # Support just one tool call per context frame for now
-                tool_result_message = context.messages[-1]
-                await self._tool_result(tool_result_message)
-        elif isinstance(frame, LLMContextFrame):
-            raise NotImplementedError("Universal LLMContext is not yet supported for Gemini Live.")
+        elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
+            context = (
+                frame.context
+                if isinstance(frame, LLMContextFrame)
+                else LLMContext.from_openai_context(frame.context)
+            )
+            await self._handle_context(context)
         elif isinstance(frame, InputTextRawFrame):
             await self._send_user_text(frame.text)
             await self.push_frame(frame, direction)
@@ -883,6 +957,40 @@ class GeminiLiveLLMService(LLMService):
         else:
             await self.push_frame(frame, direction)
 
+    async def _handle_context(self, context: LLMContext):
+        if not self._context:
+            # We got our initial context
+            self._context = context
+            if context.tools:
+                self._tools = context.tools
+            # Initialize our bookkeeping of already-completed tool calls in
+            # the context
+            await self._process_completed_function_calls(send_new_results=False)
+            await self._create_initial_response()
+        else:
+            # We got an updated context.
+            # This may contain a new user message or tool call result.
+            self._context = context
+            # Send results for newly-completed function calls, if any.
+            await self._process_completed_function_calls(send_new_results=True)
+
+    async def _process_completed_function_calls(self, send_new_results: bool):
+        # Check for set of completed function calls in the context
+        adapter: GeminiLLMAdapter = self.get_llm_adapter()
+        messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
+        for message in messages:
+            if message.parts:
+                for part in message.parts:
+                    if part.function_response:
+                        # Found a newly-completed function call - send the result to the service
+                        tool_call_id = part.function_response.id
+                        tool_name = part.function_response.name
+                        if send_new_results:
+                            await self._tool_result(
+                                tool_call_id, tool_name, part.function_response.response
+                            )
+                        self._completed_tool_calls.add(tool_call_id)
+
     async def _set_bot_is_speaking(self, speaking: bool):
         if self._bot_is_speaking == speaking:
             return
@@ -1116,6 +1224,7 @@ class GeminiLiveLLMService(LLMService):
             if self._session:
                 await self._session.close()
                 self._session = None
+            self._completed_tool_calls = set()
             self._disconnecting = False
         except Exception as e:
             logger.error(f"{self} error disconnecting: {e}")
@@ -1195,7 +1304,8 @@ class GeminiLiveLLMService(LLMService):
             self._run_llm_when_session_ready = True
             return
 
-        messages = self._context.get_messages_for_initializing_history()
+        adapter: GeminiLLMAdapter = self.get_llm_adapter()
+        messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
         if not messages:
             return
 
@@ -1223,8 +1333,9 @@ class GeminiLiveLLMService(LLMService):
 
         # Create a throwaway context just for the purpose of getting messages
         # in the right format
-        context = GeminiLiveContext.upgrade(OpenAILLMContext(messages=messages_list))
-        messages = context.get_messages_for_initializing_history()
+        context = LLMContext(messages=messages_list)
+        adapter: GeminiLLMAdapter = self.get_llm_adapter()
+        messages = adapter.get_llm_invocation_params(context).get("messages", [])
 
         if not messages:
             return
@@ -1239,17 +1350,16 @@ class GeminiLiveLLMService(LLMService):
             await self._handle_send_error(e)
 
     @traced_gemini_live(operation="llm_tool_result")
-    async def _tool_result(self, tool_result_message):
+    async def _tool_result(
+        self, tool_call_id: str, tool_name: str, tool_result_message: Dict[str, Any]
+    ):
         """Send tool result back to the API."""
         if self._disconnecting or not self._session:
             return
 
         # For now we're shoving the name into the tool_call_id field, so this
         # will work until we revisit that.
-        id = tool_result_message.get("tool_call_id")
-        name = tool_result_message.get("tool_call_name")
-        result = json.loads(tool_result_message.get("content") or "")
-        response = FunctionResponse(name=name, id=id, response=result)
+        response = FunctionResponse(name=tool_name, id=tool_call_id, response=tool_result_message)
 
         try:
             await self._session.send_tool_response(function_responses=response)
@@ -1442,8 +1552,8 @@ class GeminiLiveLLMService(LLMService):
             return
 
         # This is the output transcription text when modalities is set to AUDIO.
-        # In this case, we push LLMTextFrame and TTSTextFrame to be handled by the
-        # downstream assistant context aggregator.
+        # In this case, we push TTSTextFrame to be handled by the downstream
+        # assistant context aggregator.
         text = message.server_content.output_transcription.text
 
         if not text:
@@ -1458,7 +1568,17 @@ class GeminiLiveLLMService(LLMService):
         # Collect text for tracing
         self._llm_output_buffer += text
 
-        await self.push_frame(LLMTextFrame(text=text))
+        # NOTE: Shoot. When using Vertex AI, output transcription messages
+        # arrive *before* the model_turn messages with audio, so we need to
+        # handle sending TTSStartedFrame and LLMFullResponseStartFrame here as
+        # well. These messages also contain much *more* text (it looks further
+        # ahead). That means that on an interruption our recorded context will
+        # contain some text that was actually never spoken.
+        if not self._bot_is_speaking:
+            await self._set_bot_is_speaking(True)
+            await self.push_frame(TTSStartedFrame())
+            await self.push_frame(LLMFullResponseStartFrame())
+
         await self.push_frame(TTSTextFrame(text=text))
 
     async def _handle_msg_grounding_metadata(self, message: LiveServerMessage):
@@ -1557,26 +1677,26 @@ class GeminiLiveLLMService(LLMService):
         *,
         user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
         assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
-    ) -> GeminiLiveContextAggregatorPair:
+    ) -> LLMContextAggregatorPair:
         """Create an instance of GeminiLiveContextAggregatorPair from an OpenAILLMContext.
 
         Constructor keyword arguments for both the user and assistant aggregators can be provided.
 
+        NOTE: this method exists only for backward compatibility. New code
+        should instead do:
+            context = LLMContext(...)
+            context_aggregator = LLMContextAggregatorPair(context)
+
         Args:
             context: The LLM context to use.
             user_params: User aggregator parameters. Defaults to LLMUserAggregatorParams().
             assistant_params: Assistant aggregator parameters. Defaults to LLMAssistantAggregatorParams().
 
         Returns:
-            GeminiLiveContextAggregatorPair: A pair of context
-            aggregators, one for the user and one for the assistant,
-            encapsulated in an GeminiLiveContextAggregatorPair.
+            A pair of user and assistant context aggregators.
         """
-        context.set_llm_adapter(self.get_llm_adapter())
-
-        GeminiLiveContext.upgrade(context)
-        user = GeminiLiveUserContextAggregator(context, params=user_params)
-
+        context = LLMContext.from_openai_context(context)
         assistant_params.expect_stripped_words = False
-        assistant = GeminiLiveAssistantContextAggregator(context, params=assistant_params)
-        return GeminiLiveContextAggregatorPair(_user=user, _assistant=assistant)
+        return LLMContextAggregatorPair(
+            context, user_params=user_params, assistant_params=assistant_params
+        )