From 1a4a6f4edfd2f1e1deb3ca9fc5007f1c381d59b3 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 8 May 2026 13:55:49 -0400 Subject: [PATCH 1/6] refactor(gemini-live): bring tool-result handling in line with the canonical realtime pattern MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lays groundwork for cancel_on_interruption=False support on Gemini Live by restructuring _process_completed_function_calls to match the shape used by AWSNovaSonicLLMService and OpenAIRealtimeLLMService in #4441: a single-pass forward iteration over raw context messages that detects async-tool messages via async_tool_messages.parse_message and routes them — started skipped silently, intermediate logged-as-error and surfaced via push_error, final delivered via the formal FunctionResponse channel. Replaces the prior two-pass structure that went through the adapter for sync results — the service now uses a lightweight self._tool_call_id_to_name map (populated when the model issues tool calls) for the name lookup the adapter used to provide. Extracts a new GeminiLLMAdapter.to_function_response_dict static method for the dict-coercion logic that wraps non-dict tool returns as {value: } for Gemini's FunctionResponse.response field; the adapter's existing inline copy in _from_standard_message uses it too. Example consolidation: - Folds realtime-gemini-live-function-calling.py into the base realtime-gemini-live.py example so the base exercises function calling out of the box (matching realtime-openai.py and realtime-aws-nova-sonic.py). - Renames realtime-gemini-live-vertex-function-calling.py to realtime-gemini-live-vertex.py, mirroring the consolidation. - Adds realtime-gemini-live-async-tool.py. - Updates scripts/evals/run-release-evals.py for the renames. This commit alone doesn't make cancel_on_interruption=False fully work on Gemini Live — additional investigation is pending. This is foundational work to be built on. --- ....py => realtime-gemini-live-async-tool.py} | 109 ++++++++---------- ...ling.py => realtime-gemini-live-vertex.py} | 0 examples/realtime/realtime-gemini-live.py | 85 ++++++++++++-- scripts/evals/run-release-evals.py | 5 +- .../adapters/services/gemini_adapter.py | 41 +++++-- .../services/google/gemini_live/llm.py | 90 +++++++++++---- 6 files changed, 227 insertions(+), 103 deletions(-) rename examples/realtime/{realtime-gemini-live-function-calling.py => realtime-gemini-live-async-tool.py} (58%) rename examples/realtime/{realtime-gemini-live-vertex-function-calling.py => realtime-gemini-live-vertex.py} (100%) diff --git a/examples/realtime/realtime-gemini-live-function-calling.py b/examples/realtime/realtime-gemini-live-async-tool.py similarity index 58% rename from examples/realtime/realtime-gemini-live-function-calling.py rename to examples/realtime/realtime-gemini-live-async-tool.py index d038a5c4f..23d9d8fa0 100644 --- a/examples/realtime/realtime-gemini-live-function-calling.py +++ b/examples/realtime/realtime-gemini-live-async-tool.py @@ -4,15 +4,25 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""Example: async function call with the Gemini Live LLM service. +The ``get_current_weather`` tool is registered with +``cancel_on_interruption=False`` and simulates a slow API call (10s sleep). +While the call is in flight the conversation continues; the result arrives +later via the async-tool mechanism and is forwarded to Gemini Live as a +FunctionResponse so the model can integrate it naturally into its next turn. +""" + +import asyncio import os +import random 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.adapters.schemas.tools_schema import ToolsSchema from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner @@ -31,33 +41,55 @@ load_dotenv(override=True) async def fetch_weather_from_api(params: FunctionCallParams): - temperature = 75 if params.arguments["format"] == "fahrenheit" else 24 + # Simulate a long-running API call so we can demonstrate that the + # conversation continues while the tool is in flight. + await asyncio.sleep(10) + temperature = ( + random.randint(60, 85) + if params.arguments["format"] == "fahrenheit" + else random.randint(15, 30) + ) await params.result_callback( { "conditions": "nice", "temperature": temperature, + "location": params.arguments["location"], "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"}) +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"], +) + +tools = ToolsSchema(standard_tools=[weather_function]) -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. google_search: Use this tool to search the web for information. -""" +system_instruction = ( + "You are a friendly assistant. The user and you will engage in a spoken " + "dialog exchanging the transcripts of a natural real-time conversation. " + "Keep your responses short, generally two or three sentences for chatty " + "scenarios. When the user asks for the weather, call get_current_weather. " + "While you wait for the result, keep chatting with the user. When the " + "result arrives, share it with the user naturally." +) -# We use lambdas to defer transport parameter creation until the transport -# type is selected at runtime. transport_params = { "daily": lambda: DailyParams( audio_in_enabled=True, @@ -77,42 +109,6 @@ transport_params = { 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"], - ) - search_tool = {"google_search": {}} - # KNOWN ISSUE: If using GeminiVertexLiveLLMService, it appears - # you cannot use the "google_search" tool alongside other tools. - # See https://github.com/googleapis/python-genai/issues/941. - tools = ToolsSchema( - standard_tools=[weather_function, restaurant_function], - custom_tools={AdapterType.GEMINI: [search_tool]}, - ) - llm = GeminiLiveLLMService( api_key=os.environ["GOOGLE_API_KEY"], settings=GeminiLiveLLMService.Settings( @@ -121,13 +117,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): tools=tools, ) - llm.register_function("get_current_weather", fetch_weather_from_api) - llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + llm.register_function( + "get_current_weather", + fetch_weather_from_api, + cancel_on_interruption=False, + ) - # You can provide the system instructions and tools in the context rather - # than as arguments to GeminiLiveLLMService, but note that doing so will - # trigger a (fast) reconnection when the GeminiLiveLLMService first - # receives the context (i.e. when we send the LLMRunFrame below). context = LLMContext() # Server-side VAD is enabled by default; no local VAD is added. user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context) @@ -154,7 +149,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") - # Kick off the conversation. context.add_message( {"role": "developer", "content": "Please introduce yourself to the user."} ) @@ -166,7 +160,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.cancel() runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) - await runner.run(task) diff --git a/examples/realtime/realtime-gemini-live-vertex-function-calling.py b/examples/realtime/realtime-gemini-live-vertex.py similarity index 100% rename from examples/realtime/realtime-gemini-live-vertex-function-calling.py rename to examples/realtime/realtime-gemini-live-vertex.py diff --git a/examples/realtime/realtime-gemini-live.py b/examples/realtime/realtime-gemini-live.py index 04fa9c625..6751a16a7 100644 --- a/examples/realtime/realtime-gemini-live.py +++ b/examples/realtime/realtime-gemini-live.py @@ -6,10 +6,13 @@ 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.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner @@ -23,6 +26,7 @@ from pipecat.processors.aggregators.llm_response_universal import ( 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.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 @@ -30,6 +34,32 @@ 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"}) + + +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. google_search: Use this tool to search the web for information. +""" + + # We use lambdas to defer transport parameter creation until the transport # type is selected at runtime. transport_params = { @@ -51,23 +81,55 @@ transport_params = { 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"], + ) + search_tool = {"google_search": {}} + # KNOWN ISSUE: If using GeminiVertexLiveLLMService, it appears + # you cannot use the "google_search" tool alongside other tools. + # See https://github.com/googleapis/python-genai/issues/941. + tools = ToolsSchema( + standard_tools=[weather_function, restaurant_function], + custom_tools={AdapterType.GEMINI: [search_tool]}, + ) + llm = GeminiLiveLLMService( api_key=os.environ["GOOGLE_API_KEY"], settings=GeminiLiveLLMService.Settings( + system_instruction=system_instruction, voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede - # system_instruction="Talk like a pirate." ), - # inference_on_context_initialization=False, + tools=tools, ) - context = LLMContext( - [ - { - "role": "user", - "content": "Say hello. Then ask if I want to hear a joke.", - }, - ], - ) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + context = LLMContext() # Server-side VAD is enabled by default; no local VAD is added. user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context) @@ -94,6 +156,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def on_client_connected(transport, client): logger.info(f"Client connected") # Kick off the conversation. + context.add_message( + {"role": "developer", "content": "Please introduce yourself to the user."} + ) await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 861367027..230b174d6 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -223,12 +223,11 @@ TESTS_REALTIME = [ # ("realtime/realtime-azure.py", EVAL_WEATHER), ("realtime/realtime-openai-text.py", EVAL_WEATHER), ("realtime/realtime-openai-live-video.py", EVAL_VISION_CAMERA), - ("realtime/realtime-gemini-live.py", EVAL_SIMPLE_MATH), + ("realtime/realtime-gemini-live.py", EVAL_WEATHER), ("realtime/realtime-gemini-live-local-vad.py", EVAL_SIMPLE_MATH), - ("realtime/realtime-gemini-live-function-calling.py", EVAL_WEATHER), ("realtime/realtime-gemini-live-video.py", EVAL_VISION_CAMERA), ("realtime/realtime-gemini-live-google-search.py", EVAL_ONLINE_SEARCH), - ("realtime/realtime-gemini-live-vertex-function-calling.py", EVAL_WEATHER), + ("realtime/realtime-gemini-live-vertex.py", EVAL_WEATHER), ("realtime/realtime-aws-nova-sonic.py", EVAL_SIMPLE_MATH), ("realtime/realtime-ultravox.py", EVAL_ORDER), ("realtime/realtime-grok.py", EVAL_WEATHER), diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index 4e9e20e14..bb3a71def 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -139,6 +139,36 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): return formatted_standard_tools + custom_gemini_tools + @staticmethod + def to_function_response_dict(content: Any) -> dict[str, Any]: + """Convert a tool-result content value to Gemini's FunctionResponse.response shape. + + Gemini's ``FunctionResponse.response`` field requires a dict, so + non-dict values (e.g. plain strings, JSON-encoded scalars, or + sentinel strings like ``"COMPLETED"`` used when a function returned + no value) are wrapped as ``{"value": }``. JSON strings that + decode to a dict are passed through as-is. + + Args: + content: The tool-result content. Typically the JSON-encoded + return value of a function, but can also be a plain string + (e.g. ``"COMPLETED"``) or already-parsed dict. + + Returns: + A dict suitable for ``FunctionResponse.response``. + """ + if isinstance(content, dict): + return content + if not isinstance(content, str): + return {"value": content} + try: + decoded = json.loads(content) + except (json.JSONDecodeError, ValueError): + return {"value": content} + if isinstance(decoded, dict): + return decoded + return {"value": decoded} + def get_messages_for_logging(self, context: LLMContext) -> list[dict[str, Any]]: """Get messages from a universal LLM context in a format ready for logging about Gemini. @@ -382,16 +412,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): ) elif role == "tool": role = "user" - try: - response = json.loads(msg["content"]) - if isinstance(response, dict): - response_dict = response - else: - response_dict = {"value": response} - except Exception as e: - # Response might not be JSON-deserializable. - # This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string. - response_dict = {"value": msg["content"]} + response_dict = self.to_function_response_dict(msg["content"]) # Get function name from mapping using tool_call_id, or fallback tool_call_id = msg.get("tool_call_id") diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index adcbe28b8..61aebb14e 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -14,6 +14,7 @@ voice transcription, streaming responses, and tool usage. import asyncio import base64 import io +import json import time import uuid from dataclasses import dataclass, field @@ -56,7 +57,8 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import LLMTokenUsage -from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators import async_tool_messages +from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage from pipecat.processors.frame_processor import FrameDirection from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult from pipecat.services.google.utils import update_google_client_http_options @@ -557,6 +559,11 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): # Bookkeeping for tool calls self._completed_tool_calls = set() + # tool_call_id -> tool_name, populated as the model issues tool + # calls. Used to look up the function name when sending an async + # tool's final result back to the provider, since the async-tool + # message in the context only carries the id. + self._tool_call_id_to_name: dict[str, str] = {} def create_client(self): """Create the Gemini API client instance. Subclasses can override this.""" @@ -842,27 +849,58 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): async def _process_completed_function_calls(self, send_new_results: bool): # Check for set of completed function calls in the context - adapter = 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: - tool_call_id = part.function_response.id - tool_name = part.function_response.name - response = part.function_response.response - if ( - tool_call_id - and tool_call_id not in self._completed_tool_calls - and response - and response.get("value") != "IN_PROGRESS" - ): - # Found a newly-completed function call - send the result to the service - 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) + for message in self._context.get_messages(): + # LLMSpecificMessages are opaque provider-specific payloads, not + # standard tool-result messages — skip them. + if isinstance(message, LLMSpecificMessage): + continue + + # Async-tool messages live alongside regular tool messages in the + # context; detect and route them before the regular logic so we + # don't try to send the async-tool envelope JSON as a tool result. + async_payload = async_tool_messages.parse_message(message) + if async_payload is not None: + if async_payload.tool_call_id in self._completed_tool_calls: + continue + if async_payload.kind == "started": + # The provider already issued the tool call and natively + # awaits a result; nothing to send for the started marker. + continue + if async_payload.kind == "intermediate": + logger.error( + f"{self}: Gemini Live does not support streamed async " + f"tool results; dropping intermediate result for " + f"tool_call_id={async_payload.tool_call_id}. Use a " + f"non-realtime LLM service if your tool needs to " + f"stream intermediate results." + ) + await self.push_error( + error_msg="Gemini Live does not support streamed async tool results.", + ) + continue + # kind == "final": deliver via the formal tool-response channel + # — same path as a synchronous tool result, just delayed. + tool_name = self._tool_call_id_to_name.get( + async_payload.tool_call_id, "tool_call_result" + ) + response_dict = GeminiLLMAdapter.to_function_response_dict(async_payload.result) + if send_new_results: + await self._tool_result(async_payload.tool_call_id, tool_name, response_dict) + self._completed_tool_calls.add(async_payload.tool_call_id) + continue + + # Look for newly-completed "regular" (as opposed to async-tool) results + if message.get("role") == "tool" and message.get("content") != "IN_PROGRESS": + tool_call_id = message.get("tool_call_id") + if tool_call_id and tool_call_id not in self._completed_tool_calls: + # Found a newly-completed function call - send the result to the service + tool_name = self._tool_call_id_to_name.get(tool_call_id, "tool_call_result") + response_dict = GeminiLLMAdapter.to_function_response_dict( + message.get("content") + ) + if send_new_results: + await self._tool_result(tool_call_id, tool_name, response_dict) + self._completed_tool_calls.add(tool_call_id) async def _set_bot_is_responding(self, responding: bool): if self._bot_is_responding == responding: @@ -1193,6 +1231,7 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): await self._session.close() self._session = None self._completed_tool_calls = set() + self._tool_call_id_to_name = {} self._ready_for_realtime_input = False self._disconnecting = False except Exception as e: @@ -1420,6 +1459,10 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): if self._disconnecting or not self._session: return + logger.debug( + f"Sending tool result to Gemini Live for tool_call_id={tool_call_id}, tool_result_message={tool_result_message}" + ) + # For now we're shoving the name into the tool_call_id field, so this # will work until we revisit that. response = FunctionResponse(name=tool_name, id=tool_call_id, response=tool_result_message) @@ -1555,6 +1598,9 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): for f in function_calls ] + for fc in function_calls_llm: + self._tool_call_id_to_name[fc.tool_call_id] = fc.function_name + await self.run_function_calls(function_calls_llm) @traced_gemini_live(operation="llm_response") From 9086a46900171edf850a4496359f2af5bfc6fa1b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 8 May 2026 15:13:49 -0400 Subject: [PATCH 2/6] feat(gemini-live): support cancel_on_interruption=False on supported models MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Honors cancel_on_interruption=False on Gemini Live for models that support Gemini's NON_BLOCKING tool mechanism (Gemini 2.x at the time of writing). Function declarations registered via register_function(..., cancel_on_interruption=False) are sent with behavior: NON_BLOCKING so the conversation continues while the tool runs; the matching FunctionResponse carries scheduling: WHEN_IDLE so the result lands at a graceful pause rather than mid-sentence. Synchronous (default) tools stay BLOCKING — applying NON_BLOCKING uniformly produced filler responses like "let me look that up for you" on regular calls, since the model knew it would have an opportunity to keep talking while waiting. A new _supports_non_blocking_tools property gates the flow. On models that don't support it (currently Gemini 3.x), the service falls back to plain blocking behavior and surfaces a one-time error + ErrorFrame the moment async-tool messages first appear in the context, explaining that the flag's intent is not achievable. Caveat (Gemini 2.5): an intermittent server-side 1008 "Operation is not implemented" error can fire when realtime input arrives during a pending tool call. We auto-reconnect, but the user may need to repeat what they were saying. The proposed mitigation (https://discuss.ai.google.dev/t/gemini-live-api-websocket-error-1008-operation-is-not-implemented-or-supported-or-enabled/114644/56) of gating realtime input during pending tool calls is fundamentally incompatible with NON_BLOCKING tool calling, so we don't apply it. --- .../services/google/gemini_live/llm.py | 85 ++++++++++++++++++- 1 file changed, 84 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 61aebb14e..86fcc9b97 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -374,6 +374,27 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): """Check if the current model is a Gemini 3.x model.""" return "gemini-3" in (assert_given(self._settings.model) or "") + @property + def _supports_non_blocking_tools(self) -> bool: + """Whether the current model supports the NON_BLOCKING tool behavior + scheduling hints. + + Gemini 3.x has not yet shipped support for NON_BLOCKING function + declarations or for the ``scheduling`` field on FunctionResponse. + """ + return not self._is_gemini_3 + + def _function_is_async(self, name: str) -> bool: + """Whether the named function was registered with cancel_on_interruption=False. + + Mirrors the lookup pattern in ``LLMService.run_function_calls``: + a name-specific registry entry takes precedence; if there isn't + one, fall back to the ``None``-keyed catch-all entry. + """ + item = self._functions.get(name) + if item is None: + item = self._functions.get(None) + return item is not None and not item.cancel_on_interruption + def __init__( self, *, @@ -564,6 +585,7 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): # tool's final result back to the provider, since the async-tool # message in the context only carries the id. self._tool_call_id_to_name: dict[str, str] = {} + self._async_tool_warning_logged: bool = False def create_client(self): """Create the Gemini API client instance. Subclasses can override this.""" @@ -848,6 +870,32 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): await self._process_completed_function_calls(send_new_results=True) async def _process_completed_function_calls(self, send_new_results: bool): + # If the user registered a function with cancel_on_interruption=False, + # the aggregator emits async-tool-style messages into the context. On + # models that don't support NON_BLOCKING tool calls, the conversation + # freezes during tool execution, so the "keep talking while the tool + # runs" intent of the flag is structurally not achievable. Surface a + # one-time warning so users see they're not getting what they expect. + if not self._supports_non_blocking_tools and not self._async_tool_warning_logged: + for message in self._context.get_messages(): + if isinstance(message, LLMSpecificMessage): + continue + if async_tool_messages.parse_message(message) is not None: + logger.error( + f"{self}: cancel_on_interruption=False is not properly supported " + f"by the current Gemini Live model. Use cancel_on_interruption=True " + f"(the default), or use a non-realtime LLM service if your tool " + f"needs the async semantics." + ) + await self.push_error( + error_msg=( + "cancel_on_interruption=False is not properly supported by " + "the current Gemini Live model." + ), + ) + self._async_tool_warning_logged = True + break + # Check for set of completed function calls in the context for message in self._context.get_messages(): # LLMSpecificMessages are opaque provider-specific payloads, not @@ -1085,6 +1133,27 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): logger.debug(f"Setting system instruction: {system_instruction}") config.system_instruction = system_instruction if tools: + # Tag function declarations registered with + # cancel_on_interruption=False as NON_BLOCKING so Gemini + # doesn't stall the conversation while the tool runs. + # Synchronous (default) tools stay BLOCKING so the model + # finishes its turn before the result lands — otherwise + # we get the "let me look that up for you" filler the + # model produces when it knows the result is async. + # https://ai.google.dev/gemini-api/docs/live-api/tools#async-function-calling + if self._supports_non_blocking_tools: + for tool in tools: + if not isinstance(tool, dict): + continue + decls = tool.get("function_declarations") + if not isinstance(decls, list): + continue + for decl in decls: + if not isinstance(decl, dict): + continue + name = decl.get("name") + if isinstance(name, str) and self._function_is_async(name): + decl["behavior"] = "NON_BLOCKING" logger.debug(f"Setting tools: {tools}") config.tools = tools @@ -1232,6 +1301,7 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): self._session = None self._completed_tool_calls = set() self._tool_call_id_to_name = {} + self._async_tool_warning_logged = False self._ready_for_realtime_input = False self._disconnecting = False except Exception as e: @@ -1463,9 +1533,22 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): f"Sending tool result to Gemini Live for tool_call_id={tool_call_id}, tool_result_message={tool_result_message}" ) + # Pair the NON_BLOCKING declaration on async tools with a + # scheduling hint on the response. WHEN_IDLE lets Gemini finish + # whatever it's currently saying before addressing the result, so + # we don't cut off mid-sentence when delayed results land. Only + # meaningful for NON_BLOCKING tools — synchronous tools never + # leave the model mid-turn — so we mirror the gating used at + # tool-declaration time. + # https://ai.google.dev/gemini-api/docs/live-api/tools#async-function-calling + if self._supports_non_blocking_tools and self._function_is_async(tool_name): + response_payload = {**tool_result_message, "scheduling": "WHEN_IDLE"} + else: + response_payload = tool_result_message + # For now we're shoving the name into the tool_call_id field, so this # will work until we revisit that. - response = FunctionResponse(name=tool_name, id=tool_call_id, response=tool_result_message) + response = FunctionResponse(name=tool_name, id=tool_call_id, response=response_payload) try: await self._session.send_tool_response(function_responses=response) From 6faeffb88485797ec4d1b22e8112932e68914569 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 8 May 2026 15:21:04 -0400 Subject: [PATCH 3/6] chore: add changelog entry for cancel_on_interruption=False on Gemini Live --- changelog/4448.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4448.added.md diff --git a/changelog/4448.added.md b/changelog/4448.added.md new file mode 100644 index 000000000..d10d67922 --- /dev/null +++ b/changelog/4448.added.md @@ -0,0 +1 @@ +- Added `cancel_on_interruption=False` support for `GeminiLiveLLMService` on models that support Gemini's NON_BLOCKING tool mechanism (currently Gemini 2.x); the conversation now continues while the tool runs. On models that don't yet support NON_BLOCKING (Gemini 3.x), the service surfaces a one-time warning explaining the limitation. (Note: an intermittent 1008 error can occasionally fire on Gemini 2.5 during long-running tool calls; we auto-reconnect.) From 53922819ed3d47d19ab0710f1b089516c150c212 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 8 May 2026 15:47:06 -0400 Subject: [PATCH 4/6] refactor: explicit kind=='final' check in async-tool routing (Gemini Live) Mirrors the same change applied to AWSNovaSonicLLMService and OpenAIRealtimeLLMService in #4441 / GrokRealtimeLLMService in #4447: replaces the implicit "final happens last" pattern in _process_completed_function_calls with an explicit `if async_payload.kind == "final":` block, plus a trailing defensive `continue` so async-tool messages with an unrecognized kind don't fall through to the regular tool-result handling block. --- .../services/google/gemini_live/llm.py | 25 ++++++++++++------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 86fcc9b97..8330f9137 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -926,15 +926,22 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): error_msg="Gemini Live does not support streamed async tool results.", ) continue - # kind == "final": deliver via the formal tool-response channel - # — same path as a synchronous tool result, just delayed. - tool_name = self._tool_call_id_to_name.get( - async_payload.tool_call_id, "tool_call_result" - ) - response_dict = GeminiLLMAdapter.to_function_response_dict(async_payload.result) - if send_new_results: - await self._tool_result(async_payload.tool_call_id, tool_name, response_dict) - self._completed_tool_calls.add(async_payload.tool_call_id) + if async_payload.kind == "final": + # Deliver via the formal tool-response channel — same + # path as a synchronous tool result, just delayed. + tool_name = self._tool_call_id_to_name.get( + async_payload.tool_call_id, "tool_call_result" + ) + response_dict = GeminiLLMAdapter.to_function_response_dict(async_payload.result) + if send_new_results: + await self._tool_result( + async_payload.tool_call_id, tool_name, response_dict + ) + self._completed_tool_calls.add(async_payload.tool_call_id) + continue + # Defensive: any async-tool message must not fall through + # to the regular tool-result block below, even if it + # carries a kind we don't recognize. continue # Look for newly-completed "regular" (as opposed to async-tool) results From e21180b9625791e60c899460d47d0628490e8305 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 8 May 2026 16:27:37 -0400 Subject: [PATCH 5/6] refactor(gemini-live): use inherited LLMService._function_is_async The same registry-lookup helper was hoisted to LLMService in #4447, so drop the local duplicate. Behavior unchanged. --- src/pipecat/services/google/gemini_live/llm.py | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 8330f9137..7194127b1 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -383,18 +383,6 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]): """ return not self._is_gemini_3 - def _function_is_async(self, name: str) -> bool: - """Whether the named function was registered with cancel_on_interruption=False. - - Mirrors the lookup pattern in ``LLMService.run_function_calls``: - a name-specific registry entry takes precedence; if there isn't - one, fall back to the ``None``-keyed catch-all entry. - """ - item = self._functions.get(name) - if item is None: - item = self._functions.get(None) - return item is not None and not item.cancel_on_interruption - def __init__( self, *, From 550937734443b98800a66e99ac4ae7ecb06a37e9 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 8 May 2026 16:54:15 -0400 Subject: [PATCH 6/6] fix(gemini-live-vertex): disable NON_BLOCKING tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GeminiLiveVertexLLMService overrides _supports_non_blocking_tools to return False — Vertex AI's Gemini Live endpoint doesn't yet accept the NON_BLOCKING behavior field on function declarations or the scheduling field on FunctionResponse, and sending either breaks tool calling. Effect: function declarations sent to Vertex no longer carry NON_BLOCKING; FunctionResponses no longer carry scheduling: WHEN_IDLE. Users registering a function with cancel_on_interruption=False against Vertex get the same one-time logger.error + push_error the base class surfaces on Gemini 3.x. --- .../services/google/gemini_live/vertex/llm.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/pipecat/services/google/gemini_live/vertex/llm.py b/src/pipecat/services/google/gemini_live/vertex/llm.py index b02c18a60..42bd34f9a 100644 --- a/src/pipecat/services/google/gemini_live/vertex/llm.py +++ b/src/pipecat/services/google/gemini_live/vertex/llm.py @@ -238,6 +238,20 @@ class GeminiLiveVertexLLMService(GeminiLiveLLMService): "When using Vertex AI, the recommended approach is to use Google Cloud Storage for file handling. The Gemini File API is not directly supported in this context." ) + @property + def _supports_non_blocking_tools(self) -> bool: + """Vertex AI's Gemini Live endpoint does not yet support NON_BLOCKING tool calls. + + Override the base ``GeminiLiveLLMService`` getter to disable the + NON_BLOCKING ``behavior`` field on function declarations and the + ``scheduling`` field on FunctionResponse for Vertex sessions — + sending either appears to break tool calling against Vertex. + Users hitting this on a function registered with + ``cancel_on_interruption=False`` will see the same one-time + warning the base class surfaces for unsupported models. + """ + return False + @staticmethod def _get_credentials( credentials: str | None, credentials_path: str | None