建構即時語音代理時,部分函式呼叫可能會阻礙模型執行,導致音訊串流中斷,使用者只能默默等待。使用 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) 的呼叫時,必須將其傳遞給背景 worker。如果您在主要訊息迴圈中直接使用 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」注意事項:即使排定為
SILENT,模型有時仍可能嘗試口頭敘述工具的執行過程。如要強制執行真正的靜音,請在系統指令中加入明確的防護措施 (例如「使用 [工具名稱] 時,請執行靜音執行,不要說任何話」),或使用「發射後不理」後端模式,完全不將FunctionResponse傳回模型。