リアルタイム音声エージェントを構築する際、一部の関数呼び出しがモデルの実行をブロックし、音声ストリームが静かになり、ユーザーが沈黙の中で待機することになることがあります。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])]
メッセージ ストリームからの関数呼び出しを処理する
モデルが 1 つ以上の関数を呼び出す必要がある場合、Gemini Live API はリアルタイム メッセージ ストリームを介して tool_call イベントを送信します。
モデルは実行を継続することを想定しているため、バックエンドはストリームをブロックしないでください。search_live_flights などの低速な関数の呼び出しを受け取った場合は、バックグラウンド ワーカーに渡す必要があります。10 秒のタスクのメイン メッセージ ループで await を直接使用すると、接続がフリーズします。高速タスク（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 の例は、scheduling="WHEN_IDLE" を使用して function_response をフォーマットして送信し、結果を通知する前に会話の自然な一時停止を待機する方法を示しています。
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 で個別に送信すると、モデルは複数の個別の音声ターンを生成します。1 つの統合レスポンスを配信するには、次のいずれかのパターンを適用します。
- 中間結果に SILENT を使用する: すべての中間ツール レスポンスを
scheduling="SILENT"で送信します。scheduling="WHEN_IDLE"を使用して最終レスポンスを送信し、モデルが収集されたすべてのデータを 1 ターンで要約するようにします。 - バッチ レスポンス: すべての同時ツールタスクが完了するのを待ち、
function_responses=[...]を使用して 1 つのsend_tool_response呼び出しでまとめて配信します。
ベスト プラクティス
- ツールの動作を明示的に宣言する: 状態を変更するツールや不可逆的なツールを
BLOCKINGに設定し、読み取り専用のルックアップやバックグラウンド ルックアップをNON_BLOCKINGに設定します。 - 同時実行を考慮した設計: 処理速度の遅いツール（外部 API のクエリや RAG パイプラインの実行など）は、常にバックエンドのバックグラウンド タスクにオフロードします。モデルがアクティブな音声ストリームの処理を継続できるようにします。
- 非ブロッキング レスポンスを意図的にスケジュールする: 中間レスポンスには
SILENTを、音声出力をトリガーする最終結果にはWHEN_IDLEを、重要なアラートにはINTERRUPTのみを使用します。schedulingを設定しない場合、デフォルトのWHEN_IDLEが使用され、レスポンスごとにモデルターンが開始されます。 - 独立したチャットターン: Gemini Live API では、ツールの実行はチャットターンから完全に独立しています。ツールがバックグラウンドで処理している間も、会話は自然に分岐、継続、進行できます。
- 「サイレント」の注意事項: モデルは、
SILENTとしてスケジュールされている場合でも、ツールの実行を音声で説明しようとすることがあります。真のサイレント モードを適用するには、システム指示に明示的なガードレールを追加するか（例: 「[ツール名] を使用する場合は、サイレント実行を行い、何も言わない」）、モデルにFunctionResponseをまったく返さない「ファイア アンド フォーゲット」バックエンド パターンを使用します。