From f974c66e12070dcc3df5eabe7fe8ec5d393497e6 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 27 Oct 2025 16:36:50 -0400 Subject: [PATCH] Update `GeminiLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` --- CHANGELOG.md | 53 +++++ .../26a-gemini-live-transcription.py | 13 +- .../26b-gemini-live-function-calling.py | 13 +- .../foundational/26c-gemini-live-video.py | 13 +- examples/foundational/26d-gemini-live-text.py | 7 +- .../26e-gemini-live-google-search.py | 13 +- .../foundational/26f-gemini-live-files-api.py | 15 +- .../26g-gemini-live-groundingMetadata.py | 13 +- ...26h-gemini-live-vertex-function-calling.py | 17 +- .../26i-gemini-live-graceful-end.py | 13 +- examples/foundational/46-video-processing.py | 13 +- .../adapters/services/gemini_adapter.py | 28 ++- .../services/google/gemini_live/llm.py | 204 ++++++++++++++---- 13 files changed, 328 insertions(+), 87 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dd1091e87..518791bf2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/examples/foundational/26a-gemini-live-transcription.py b/examples/foundational/26a-gemini-live-transcription.py index de277156b..9ac22a814 100644 --- a/examples/foundational/26a-gemini-live-transcription.py +++ b/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() diff --git a/examples/foundational/26b-gemini-live-function-calling.py b/examples/foundational/26b-gemini-live-function-calling.py index 65d159bb0..fd2b36ab1 100644 --- a/examples/foundational/26b-gemini-live-function-calling.py +++ b/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( [ diff --git a/examples/foundational/26c-gemini-live-video.py b/examples/foundational/26c-gemini-live-video.py index 7b765075e..be036a557 100644 --- a/examples/foundational/26c-gemini-live-video.py +++ b/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( [ diff --git a/examples/foundational/26d-gemini-live-text.py b/examples/foundational/26d-gemini-live-text.py index 062c0231b..fc9f68bcb 100644 --- a/examples/foundational/26d-gemini-live-text.py +++ b/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( [ diff --git a/examples/foundational/26e-gemini-live-google-search.py b/examples/foundational/26e-gemini-live-google-search.py index 178fdd282..e80ed4536 100644 --- a/examples/foundational/26e-gemini-live-google-search.py +++ b/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( [ diff --git a/examples/foundational/26f-gemini-live-files-api.py b/examples/foundational/26f-gemini-live-files-api.py index eeda16f52..0091c01e6 100644 --- a/examples/foundational/26f-gemini-live-files-api.py +++ b/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( diff --git a/examples/foundational/26g-gemini-live-groundingMetadata.py b/examples/foundational/26g-gemini-live-groundingMetadata.py index bea1756b2..df86094da 100644 --- a/examples/foundational/26g-gemini-live-groundingMetadata.py +++ b/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( [ diff --git a/examples/foundational/26h-gemini-live-vertex-function-calling.py b/examples/foundational/26h-gemini-live-vertex-function-calling.py index c0344a052..126d85ad7 100644 --- a/examples/foundational/26h-gemini-live-vertex-function-calling.py +++ b/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( [ diff --git a/examples/foundational/26i-gemini-live-graceful-end.py b/examples/foundational/26i-gemini-live-graceful-end.py index e51bbb032..4e5933e8d 100644 --- a/examples/foundational/26i-gemini-live-graceful-end.py +++ b/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( [ diff --git a/examples/foundational/46-video-processing.py b/examples/foundational/46-video-processing.py index 62ea5debe..5f92139bc 100644 --- a/examples/foundational/46-video-processing.py +++ b/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() diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index 26f037127..1fa8d9e6f 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/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), diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 4b5f05209..cf340c017 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/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 + )