diff --git a/CHANGELOG.md b/CHANGELOG.md index 337e0ddd3..10cd8738f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Updated `GeminiMultimodalLiveLLMService` to use the `google-genai` library rather than use WebSockets directly. +### Fixed + +- `GeminiMultimodalLiveLLMService` will now end gracefully (i.e. after the bot + has finished) upon receiving an `EndFrame`. + +- `GeminiMultimodalLiveLLMService` will try to seamlessly reconnect when it + loses its connection. + ## [0.0.89] - 2025-10-07 ### Fixed diff --git a/examples/foundational/26i-gemini-multimodal-live-graceful-end.py b/examples/foundational/26i-gemini-multimodal-live-graceful-end.py new file mode 100644 index 000000000..e178997b8 --- /dev/null +++ b/examples/foundational/26i-gemini-multimodal-live-graceful-end.py @@ -0,0 +1,206 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import asyncio +import os +from datetime import datetime + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +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.frame_processor import FrameDirection +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +async def fetch_weather_from_api(params: FunctionCallParams): + temperature = 75 if params.arguments["format"] == "fahrenheit" else 24 + await params.result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": params.arguments["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await params.result_callback({"name": "The Golden Dragon"}) + + +async def end_conversation(params: FunctionCallParams): + await params.result_callback({"success": True}) + await params.llm.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM) + + +system_instruction = """ +You are a helpful assistant who can answer questions and use tools. + +You have three tools available to you: +1. get_current_weather: Use this tool to get the current weather in a specific location. +2. get_restaurant_recommendation: Use this tool to get a restaurant recommendation in a specific location. +3. end_conversation: Use this tool to gracefully end the conversation. + +After you've responded to the user three times, do two things, in order: +1. Politely let them know that that's all the time you have today and say goodbye. +2. Call the end_conversation tool to gracefully end the conversation. +""" + + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + # set stop_secs to something roughly similar to the internal setting + # of the Multimodal Live api, just to align events. This doesn't really + # matter because we can only use the Multimodal Live API's phrase + # endpointing, for now. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + # set stop_secs to something roughly similar to the internal setting + # of the Multimodal Live api, just to align events. This doesn't really + # matter because we can only use the Multimodal Live API's phrase + # endpointing, for now. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + # set stop_secs to something roughly similar to the internal setting + # of the Multimodal Live api, just to align events. This doesn't really + # matter because we can only use the Multimodal Live API's phrase + # endpointing, for now. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + end_conversation_function = FunctionSchema( + name="end_conversation", + description="Gracefully end the conversation", + properties={}, + required=[], + ) + search_tool = {"google_search": {}} + tools = ToolsSchema( + standard_tools=[weather_function, restaurant_function, end_conversation_function], + custom_tools={AdapterType.GEMINI: [search_tool]}, + ) + + llm = GeminiMultimodalLiveLLMService( + api_key=os.getenv("GOOGLE_API_KEY"), + system_instruction=system_instruction, + tools=tools, + ) + + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + llm.register_function("end_conversation", end_conversation) + + context = OpenAILLMContext( + [{"role": "user", "content": "Say hello."}], + ) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), + context_aggregator.user(), + llm, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index f48fd3324..65b5eb781 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -659,6 +659,9 @@ class GeminiMultimodalLiveLLMService(LLMService): # Session resumption self._session_resumption_handle: Optional[str] = None + # Bookkeeping for ending gracefully (i.e. after the bot is finished) + self._end_frame_pending_bot_turn_finished: Optional[EndFrame] = None + def _create_client(self, api_key: str, http_options: Optional[HttpOptions] = None): self._client = Client(api_key=api_key, http_options=http_options) @@ -771,7 +774,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # async def _handle_interruption(self): - self._bot_is_speaking = False + await self._set_bot_is_speaking(False) await self.push_frame(TTSStoppedFrame()) await self.push_frame(LLMFullResponseEndFrame()) @@ -802,6 +805,13 @@ class GeminiMultimodalLiveLLMService(LLMService): frame: The frame to process. direction: The frame processing direction. """ + # Defer EndFrame handling until after the bot turn is finished + if isinstance(frame, EndFrame): + if self._bot_is_speaking: + logger.debug("Deferring handling EndFrame until bot turn is finished") + self._end_frame_pending_bot_turn_finished = frame + return + await super().process_frame(frame, direction) if isinstance(frame, TranscriptionFrame): @@ -864,6 +874,16 @@ class GeminiMultimodalLiveLLMService(LLMService): else: await self.push_frame(frame, direction) + async def _set_bot_is_speaking(self, speaking: bool): + if self._bot_is_speaking == speaking: + return + + self._bot_is_speaking = speaking + + if not self._bot_is_speaking and self._end_frame_pending_bot_turn_finished: + await self.queue_frame(self._end_frame_pending_bot_turn_finished) + self._end_frame_pending_bot_turn_finished = None + async def _connect(self, session_resumption_handle: Optional[str] = None): """Establish client connection to Gemini Live API.""" if self._session: @@ -1271,7 +1291,7 @@ class GeminiMultimodalLiveLLMService(LLMService): return if not self._bot_is_speaking: - self._bot_is_speaking = True + await self._set_bot_is_speaking(True) await self.push_frame(TTSStartedFrame()) await self.push_frame(LLMFullResponseStartFrame()) @@ -1307,7 +1327,7 @@ class GeminiMultimodalLiveLLMService(LLMService): @traced_gemini_live(operation="llm_response") async def _handle_msg_turn_complete(self, message: LiveServerMessage): """Handle the turn complete message.""" - self._bot_is_speaking = False + await self._set_bot_is_speaking(False) text = self._bot_text_buffer # Trace the complete LLM response (this will be handled by the decorator)