Merge pull request #4448 from pipecat-ai/pk/gemini-live-async-tool-support

feat: support cancel_on_interruption=False on Gemini Live (Gemini 2.x)
This commit is contained in:
kompfner
2026-05-08 16:57:32 -04:00
committed by GitHub
8 changed files with 321 additions and 104 deletions

1
changelog/4448.added.md Normal file
View File

@@ -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.)

View File

@@ -4,15 +4,25 @@
# SPDX-License-Identifier: BSD 2-Clause License # 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 os
import random
from datetime import datetime from datetime import datetime
from dotenv import load_dotenv from dotenv import load_dotenv
from loguru import logger from loguru import logger
from pipecat.adapters.schemas.function_schema import FunctionSchema 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.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.runner import PipelineRunner
@@ -31,33 +41,55 @@ load_dotenv(override=True)
async def fetch_weather_from_api(params: FunctionCallParams): 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( await params.result_callback(
{ {
"conditions": "nice", "conditions": "nice",
"temperature": temperature, "temperature": temperature,
"location": params.arguments["location"],
"format": params.arguments["format"], "format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
} }
) )
async def fetch_restaurant_recommendation(params: FunctionCallParams): weather_function = FunctionSchema(
await params.result_callback({"name": "The Golden Dragon"}) 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 = """ system_instruction = (
You are a helpful assistant who can answer questions and use tools. "You are a friendly assistant. The user and you will engage in a spoken "
"dialog exchanging the transcripts of a natural real-time conversation. "
You have three tools available to you: "Keep your responses short, generally two or three sentences for chatty "
1. get_current_weather: Use this tool to get the current weather in a specific location. "scenarios. When the user asks for the weather, call get_current_weather. "
2. get_restaurant_recommendation: Use this tool to get a restaurant recommendation in a specific location. "While you wait for the result, keep chatting with the user. When the "
3. google_search: Use this tool to search the web for information. "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 = { transport_params = {
"daily": lambda: DailyParams( "daily": lambda: DailyParams(
audio_in_enabled=True, audio_in_enabled=True,
@@ -77,42 +109,6 @@ transport_params = {
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot") 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( llm = GeminiLiveLLMService(
api_key=os.environ["GOOGLE_API_KEY"], api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings( settings=GeminiLiveLLMService.Settings(
@@ -121,13 +117,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
tools=tools, tools=tools,
) )
llm.register_function("get_current_weather", fetch_weather_from_api) llm.register_function(
llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) "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() context = LLMContext()
# Server-side VAD is enabled by default; no local VAD is added. # Server-side VAD is enabled by default; no local VAD is added.
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context) 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") @transport.event_handler("on_client_connected")
async def on_client_connected(transport, client): async def on_client_connected(transport, client):
logger.info(f"Client connected") logger.info(f"Client connected")
# Kick off the conversation.
context.add_message( context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."} {"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() await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task) await runner.run(task)

View File

@@ -6,10 +6,13 @@
import os import os
from datetime import datetime
from dotenv import load_dotenv from dotenv import load_dotenv
from loguru import logger 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.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner 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.types import RunnerArguments
from pipecat.runner.utils import create_transport from pipecat.runner.utils import create_transport
from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService 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.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
@@ -30,6 +34,32 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True) 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 # We use lambdas to defer transport parameter creation until the transport
# type is selected at runtime. # type is selected at runtime.
transport_params = { transport_params = {
@@ -51,23 +81,55 @@ transport_params = {
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot") 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( llm = GeminiLiveLLMService(
api_key=os.environ["GOOGLE_API_KEY"], api_key=os.environ["GOOGLE_API_KEY"],
settings=GeminiLiveLLMService.Settings( settings=GeminiLiveLLMService.Settings(
system_instruction=system_instruction,
voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede
# system_instruction="Talk like a pirate."
), ),
# inference_on_context_initialization=False, tools=tools,
) )
context = LLMContext( llm.register_function("get_current_weather", fetch_weather_from_api)
[ llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
{
"role": "user", context = LLMContext()
"content": "Say hello. Then ask if I want to hear a joke.",
},
],
)
# Server-side VAD is enabled by default; no local VAD is added. # Server-side VAD is enabled by default; no local VAD is added.
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context) 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): async def on_client_connected(transport, client):
logger.info(f"Client connected") logger.info(f"Client connected")
# Kick off the conversation. # Kick off the conversation.
context.add_message(
{"role": "developer", "content": "Please introduce yourself to the user."}
)
await task.queue_frames([LLMRunFrame()]) await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected") @transport.event_handler("on_client_disconnected")

View File

@@ -223,12 +223,11 @@ TESTS_REALTIME = [
# ("realtime/realtime-azure.py", EVAL_WEATHER), # ("realtime/realtime-azure.py", EVAL_WEATHER),
("realtime/realtime-openai-text.py", EVAL_WEATHER), ("realtime/realtime-openai-text.py", EVAL_WEATHER),
("realtime/realtime-openai-live-video.py", EVAL_VISION_CAMERA), ("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-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-video.py", EVAL_VISION_CAMERA),
("realtime/realtime-gemini-live-google-search.py", EVAL_ONLINE_SEARCH), ("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-aws-nova-sonic.py", EVAL_SIMPLE_MATH),
("realtime/realtime-ultravox.py", EVAL_ORDER), ("realtime/realtime-ultravox.py", EVAL_ORDER),
("realtime/realtime-grok.py", EVAL_WEATHER), ("realtime/realtime-grok.py", EVAL_WEATHER),

View File

@@ -139,6 +139,36 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
return formatted_standard_tools + custom_gemini_tools 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": <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]]: 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. """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": elif role == "tool":
role = "user" role = "user"
try: response_dict = self.to_function_response_dict(msg["content"])
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"]}
# Get function name from mapping using tool_call_id, or fallback # Get function name from mapping using tool_call_id, or fallback
tool_call_id = msg.get("tool_call_id") tool_call_id = msg.get("tool_call_id")

View File

@@ -14,6 +14,7 @@ voice transcription, streaming responses, and tool usage.
import asyncio import asyncio
import base64 import base64
import io import io
import json
import time import time
import uuid import uuid
from dataclasses import dataclass, field from dataclasses import dataclass, field
@@ -56,7 +57,8 @@ from pipecat.frames.frames import (
UserStoppedSpeakingFrame, UserStoppedSpeakingFrame,
) )
from pipecat.metrics.metrics import LLMTokenUsage 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.processors.frame_processor import FrameDirection
from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult
from pipecat.services.google.utils import update_google_client_http_options from pipecat.services.google.utils import update_google_client_http_options
@@ -372,6 +374,15 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
"""Check if the current model is a Gemini 3.x model.""" """Check if the current model is a Gemini 3.x model."""
return "gemini-3" in (assert_given(self._settings.model) or "") 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 __init__( def __init__(
self, self,
*, *,
@@ -557,6 +568,12 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
# Bookkeeping for tool calls # Bookkeeping for tool calls
self._completed_tool_calls = set() 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] = {}
self._async_tool_warning_logged: bool = False
def create_client(self): def create_client(self):
"""Create the Gemini API client instance. Subclasses can override this.""" """Create the Gemini API client instance. Subclasses can override this."""
@@ -841,28 +858,92 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
await self._process_completed_function_calls(send_new_results=True) await self._process_completed_function_calls(send_new_results=True)
async def _process_completed_function_calls(self, send_new_results: bool): 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 # Check for set of completed function calls in the context
adapter = self.get_llm_adapter() for message in self._context.get_messages():
messages = adapter.get_llm_invocation_params(self._context).get("messages", []) # LLMSpecificMessages are opaque provider-specific payloads, not
for message in messages: # standard tool-result messages — skip them.
if message.parts: if isinstance(message, LLMSpecificMessage):
for part in message.parts: continue
if part.function_response:
tool_call_id = part.function_response.id # Async-tool messages live alongside regular tool messages in the
tool_name = part.function_response.name # context; detect and route them before the regular logic so we
response = part.function_response.response # don't try to send the async-tool envelope JSON as a tool result.
if ( async_payload = async_tool_messages.parse_message(message)
tool_call_id if async_payload is not None:
and tool_call_id not in self._completed_tool_calls if async_payload.tool_call_id in self._completed_tool_calls:
and response continue
and response.get("value") != "IN_PROGRESS" if async_payload.kind == "started":
): # The provider already issued the tool call and natively
# Found a newly-completed function call - send the result to the service # awaits a result; nothing to send for the started marker.
if send_new_results: continue
await self._tool_result( if async_payload.kind == "intermediate":
tool_call_id, tool_name, part.function_response.response logger.error(
) f"{self}: Gemini Live does not support streamed async "
self._completed_tool_calls.add(tool_call_id) 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
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
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): async def _set_bot_is_responding(self, responding: bool):
if self._bot_is_responding == responding: if self._bot_is_responding == responding:
@@ -1047,6 +1128,27 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
logger.debug(f"Setting system instruction: {system_instruction}") logger.debug(f"Setting system instruction: {system_instruction}")
config.system_instruction = system_instruction config.system_instruction = system_instruction
if tools: 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}") logger.debug(f"Setting tools: {tools}")
config.tools = tools config.tools = tools
@@ -1193,6 +1295,8 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
await self._session.close() await self._session.close()
self._session = None self._session = None
self._completed_tool_calls = set() self._completed_tool_calls = set()
self._tool_call_id_to_name = {}
self._async_tool_warning_logged = False
self._ready_for_realtime_input = False self._ready_for_realtime_input = False
self._disconnecting = False self._disconnecting = False
except Exception as e: except Exception as e:
@@ -1420,9 +1524,26 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
if self._disconnecting or not self._session: if self._disconnecting or not self._session:
return return
logger.debug(
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 # For now we're shoving the name into the tool_call_id field, so this
# will work until we revisit that. # 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: try:
await self._session.send_tool_response(function_responses=response) await self._session.send_tool_response(function_responses=response)
@@ -1555,6 +1676,9 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
for f in function_calls 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) await self.run_function_calls(function_calls_llm)
@traced_gemini_live(operation="llm_response") @traced_gemini_live(operation="llm_response")

View File

@@ -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." "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 @staticmethod
def _get_credentials( def _get_credentials(
credentials: str | None, credentials_path: str | None credentials: str | None, credentials_path: str | None