在构建实时语音智能体时，某些函数调用可能会阻止模型执行，导致音频串流静音，用户只能在沉默中等待。借助 Gemini Live API，所有函数调用默认都是非阻塞的，这让您可以与主对话流程并行执行函数。此过程称为异步函数调用。您的后端可以在后台处理繁重的任务，例如搜索实时航班价格或查询复杂的外部 API，而模型则继续自然地倾听、说话并与用户对话。Gemini Live API 允许函数调用在后台处理，而不会中断用户与模型的互动，从而实现更流畅、更实时的互动。
借助异步函数调用，您无需暂停对话即可完成预约、设置提醒或提取数据等任务。例如，用户可以请求预订航班，并在后台处理预订时立即询问天气信息。
异步函数调用示例
以下示例演示了用户预订航班并询问纽约时间，同时 book_ticket 函数在后台异步运行：
User: Please book the 2:00 PM flight to New York for me.
Model: function_call: {name: "book_ticket"}
//(The "book_ticket" function call is sent to the client.)
//(Right after the "book_ticket" function call is received, the client sends a text message to the model: "repeat this sentence 'I'm booking your ticket now, please wait.'")
//(The client runs the function call asynchronously in the background.)
Model: I'm booking your ticket now, please wait.
User: What is the current time in New York?
Model: The current time in New York is 12:00pm.
//(Once the book_ticket function finishes, the client sends the result.)
Function_response: {name: "book_ticket", response: {booking_status: "booked"}}
Model: Your flight has been booked. Expect a confirmation text on your phone within 5 minutes.
实现异步函数调用
本部分提供了一系列示例，展示了如何使用 Python 版 Agent Platform SDK 构建一个高度响应的并发架构，该架构利用了 Gemini Live API 的异步函数调用功能。这些示例分为以下任务：
定义工具
您可以通过设置每个工具的 FunctionDeclaration 的 behavior 字段来控制其执行方式：
NON_BLOCKING（未设置behavior时的默认值）：模型在工具执行期间继续对话。对于只读查找或结果可以稍后交付的后台任务，请使用此设置。BLOCKING：模型会暂停并等待您的FunctionResponse，然后再生成更多输出。对于会改变状态或不可逆的操作（例如付款、预订或账号更新），请使用此设置，因为在这些操作中，模型在得知结果之前不得确认结果。如果在阻塞调用运行期间有新的用户输入，服务器会取消该调用，这需要后端具有幂等性。
如果您在 Gemini Live API 之上使用框架，请验证该框架是否在会话设置消息中传输 behavior；如果省略，工具默认以 NON_BLOCKING 身份运行。
以下 Python 示例展示了如何同时声明非阻塞工具和阻塞工具：
from google import genai
from google.genai import types
client = genai.Client()
# 1. A read-only lookup: safe to run in the background
search_live_flights = types.FunctionDeclaration(
name="search_live_flights",
description="Searches airlines for current flight prices. Can take up to 10 seconds.",
behavior=types.Behavior.NON_BLOCKING,
)
# 2. A state-changing operation: the model waits for the result before continuing
book_flight = types.FunctionDeclaration(
name="book_flight",
description="Books a flight ticket. Irreversible.",
behavior=types.Behavior.BLOCKING,
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"flight_id": types.Schema(type=types.Type.STRING),
},
required=["flight_id"],
),
)
tools = [types.Tool(function_declarations=[search_live_flights, book_flight])]
处理来自消息流的函数调用
当模型应调用一个或多个函数时，Gemini Live API 会通过实时消息流发送 tool_call 事件。
您的后端不得阻塞数据流，因为模型需要持续运行。当您收到对慢速函数（例如 search_live_flights）的调用时，必须将其传递给后台工作器。如果您在主消息循环中直接使用 await 执行 10 秒的任务，则会冻结连接。可以安全地等待快速任务（例如 get_current_weather）。
import asyncio
async def handle_stream(session):
async for response in session.receive():
# Check if the model is asking to use a tool
if response.tool_call is not None:
for fc in response.tool_call.function_calls:
if fc.name == "search_live_flights":
# Pass to a background task so we don't block the receive loop!
asyncio.create_task(background_flight_search(fc.id, fc.args, session))
elif fc.name == "get_current_weather":
# Instant lookups can be safely awaited directly
await instant_weather_lookup(fc.id, fc.args, session)
管理用户预期
为了在长时间运行的异步函数调用期间管理用户预期，请从客户端发送一条文本消息，提示系统告知用户请求正在处理中，并请他们耐心等待。例如，在客户端收到函数调用后，客户端可以向模型发送一条文本消息，例如：“重复这句话：‘我正在为您订票，请稍候。’”
以下示例对话框展示了此交流过程：
User: Please book the 2:00 PM flight to New York for me.
Model: function_call: {name: "book_ticket"}
//(The "book_ticket" function call is sent to the client.)
//(Right after the "book_ticket" function call is received, the client sends a text message to the model: "repeat this sentence 'I'm booking your ticket now, please wait.'")
//(The client runs the function call asynchronously in the background.)
Model: I'm booking your ticket now, please wait.
User: What is the current time in New York?
Model: The current time in New York is 12:00pm.
//(Once the "book_ticket" function call finishes, the client sends in the response.)
Function_response: {name: "book_ticket", response: {booking_status: "booked"}}
Model: Your flight has been booked. Expect a confirmation text on your phone within 5 minutes.
这种主动发送消息的策略具有以下优势：
- 告知用户当前系统操作，以便在长时间运行的函数调用期间管理预期。
- 减少多余的简短用户提示（例如“你好？”或“你在吗？”）的频率。这些情况通常发生在系统长时间处于静默状态，同时异步函数调用正在处理的过程中。这可以最大限度地降低由这些重复的用户查询触发的重复函数调用风险。
- 提供额外的系统提示可以降低在后续互动中创建重复调用的可能性。
处理重复的函数调用
在收到第一个调用的响应之前，模型有很小的概率会生成重复的函数调用。如果您的应用场景允许，当同一函数调用的响应仍处于待处理状态时，您的应用可以忽略重复的函数调用。
以下示例展示了客户端如何忽略重复的函数调用：
User: Please book the 2:00 PM flight to New York for me.
Model: function_call: {name: "book_ticket"}
//(The "book_ticket" function call is sent to the client. It is running asynchronously in the background.)
User: What is the current time in New York?
Model: The current time in New York is 12:00pm. + function_call: {name: "book_ticket"}
//(The duplicated "book_ticket" can be ignored by the client since the response for the first "book_ticket" has not been sent to the model yet.)
//(The first "book_ticket" function call finishes, and client sends in the response.)
Function_response: {name: "book_ticket", response: {booking_status: "booked"}}
Model: Your flight has been booked. Expect a confirmation text on your phone within 5 minutes.
处理异步函数响应
异步函数调用完成后，您的应用会将结果发送到 function_response 中的模型。当后端正在处理函数调用（例如搜索航班）时，用户可能会向模型提出一个完全不同的问题，例如“伦敦的天气怎么样？”。模型会实时响应请求，并与函数调用执行并行。由于用户在函数执行完成时可能正在与模型进行持续互动，因此您可以指定一项政策，用于定义模型应如何处理此传入响应。您可以指定以下政策之一：
如需指定政策，请在 function_response 载荷中添加 scheduling 字段：
{
"name": "book_ticket",
"scheduling": "WHEN_IDLE",
"response": {
"booking_status": "booked"
}
}
FunctionResponse.scheduling 的模型支持因模型而异：
Gemini 3.8 Live：支持scheduling（SILENT、WHEN_IDLE和INTERRUPT）。gemini-3.8-live-extended-thinking-preview：不支持显式调度配置。发送scheduling会导致会话关闭，并显示 WebSocket 错误 1007。对于此模型，请省略scheduling字段。
在支持的模型上省略 scheduling 时，Gemini Live API 默认使用 WHEN_IDLE。在用户说完后，每个传入的函数回答都会触发一个新的模型轮次。
以下 Python 示例展示了如何设置 function_response 的格式并将其与 scheduling="WHEN_IDLE" 一起发送，以便在宣布结果之前等待对话自然暂停：
async def background_flight_search(call_id, args, session):
# 1. Simulate a slow API call taking 5 seconds
await asyncio.sleep(5)
flight_data = ["Air Canada AC758: $350", "WestJet WS12: $290"]
# 2. Format the response
function_response = types.FunctionResponse(
id=call_id,
name="search_live_flights",
response={ "status": "success", "flights": flight_data },
scheduling="WHEN_IDLE" # Wait for a moment to tell the user
)
# 3. Send it back into the live session
await session.send_tool_response(function_responses=[function_response])
您可以在 scheduling 字段中指定以下政策，以管理函数响应：
SILENT 响应政策
使用 SILENT 政策时，函数回答会添加到模型的上下文中，但模型不会为其生成回答，并且任何正在进行的用户互动都不会中断。
User: Please book the 2:00 PM flight to New York for me.
Model: function_call: {name: "book_ticket"}
//(The book_ticket function call is sent to the client and starts running asynchronously in the background.)
User: What is the current time in New York?
Model: The current time in New York is 12:00pm.
//(The book_ticket function finishes, and client sends the result with scheduling: "SILENT".)
Function_response: {name: "book_ticket", scheduling: "SILENT", response: {booking_status: "booked"}}
//(The model doesn't generate a response for the function response.)
User: Is my flight ticket booked?
Model: Yes. Your flight has been booked.
WHEN_IDLE 响应政策
使用 WHEN_IDLE 政策时，模型仅在没有活跃用户互动时才会生成对函数响应的回答。如果用户正在进行互动，模型会等待互动完成，然后再生成回答，以避免中断。
User: Please book the 2:00 PM flight to New York for me.
Model: function_call: {name: "book_ticket"}
//(The book_ticket function call is sent to the client and starts running asynchronously in the background.)
User: What is the current time in New York?
//(The book_ticket function finishes, and client sends the result with scheduling: "WHEN_IDLE".)
Function_response: {name: "book_ticket", scheduling: "WHEN_IDLE", response: {booking_status: "booked"}}
//(The ongoing interaction about the time is not interrupted.)
Model: The current time in New York is 12:00pm.
//(After responding to the user's time query, the model issues the response for the book_ticket function.)
Model: Your flight has been booked. Expect a confirmation text on your phone within 5 minutes.
INTERRUPT 响应政策
使用 INTERRUPT 政策时，模型会立即生成对函数响应的回答，从而中断任何正在进行的用户互动。
User: Please book the 2:00 PM flight to New York for me.
Model: function_call: {name: "book_ticket"}
//(The book_ticket function call is sent to the client and starts running asynchronously in the background.)
User: What is the current time in New York?
//(The book_ticket function finishes, and client sends the result with scheduling: "INTERRUPT".)
Function_response: {name: "book_ticket", scheduling: "INTERRUPT", response: {booking_status: "booked"}}
//(The ongoing interaction about the time is interrupted, and model skips responding to it.)
Model: Your flight has been booked. Expect a confirmation text on your phone within 5 minutes.
汇总多个非阻塞回答
当模型同时调用多个非阻塞工具（例如并行数据查找）时，使用 WHEN_IDLE 单独发送每个结果会导致模型生成多个单独的语音轮次。如需提供单个合并响应，请应用以下模式之一：
- 对中间结果使用 SILENT：使用
scheduling="SILENT"发送所有中间工具回答。使用scheduling="WHEN_IDLE"发送最终回答，以便模型在一轮对话中总结所有收集到的数据。 - 批量回答：等待所有并发工具任务完成，然后使用
function_responses=[...]在单个send_tool_response调用中一起提供这些任务。
最佳做法
- 明确声明工具行为：将状态更改或不可逆工具设置为
BLOCKING，并将只读或后台查找设置为NON_BLOCKING。 - 针对并发进行设计：始终将慢速工具（例如查询外部 API 或运行 RAG 流水线）分流到后端中的后台任务。让模型继续处理活跃的音频串流。
- 谨慎安排非阻塞响应：使用
SILENT发送中间响应，使用WHEN_IDLE发送应触发语音输出的最终结果，仅在发送严重警报时使用INTERRUPT。如果未设置scheduling，则默认值为WHEN_IDLE，并为每个回答启动模型轮次。 - 独立的对话轮次：在 Gemini Live API 中，工具执行完全独立于对话轮次。当工具在后台处理时，对话可以分支、继续并自然流畅地进行。
- “静默”警告：即使工具已安排为
SILENT，模型仍可能偶尔尝试口头叙述工具的执行过程。如需强制执行真正的静默，请在系统指令中添加明确的护栏（例如，“使用 [工具名称] 时，执行静默执行，不发出任何声音”），或使用“即发即弃”后端模式，即根本不向模型发送FunctionResponse。