リアルタイム音声エージェントを構築する場合、一部の関数呼び出しによってモデルの実行がブロックされ、音声ストリームが停止し、ユーザーが沈黙の中で待機することになります。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.
非同期関数呼び出しを実装する
このセクションでは、Agent Platform SDK の Python バージョンを使用して、Gemini Live API の非同期関数呼び出し機能を使用する、応答性の高い同時実行アーキテクチャを構築する一連の例を示します。この例は、次のタスクに分類されます。
ツールを定義する
非同期関数呼び出しはモデルレベルで有効になるため、Gemini Enterprise Agent Platform 呼び出しで標準の Gemini API と同様に、リクエスト構成で使用するツールを指定できます。これにより、ツールが実行されている間もモデルが会話を継続できます。
from google import genai
from google.genai import types
# 1. A tool that takes a long time to execute
search_live_flights = {
"name": "search_live_flights",
"description": "Searches airlines for current flight prices. Can take up to 10 seconds."
}
# 2. A tool that executes instantly
get_current_weather = {
"name": "get_current_weather",
"description": "Gets the current weather for a given city."
}
tools = [{"function_declarations": [search_live_flights, get_current_weather]}]
メッセージ ストリームからの関数呼び出しを処理する
モデルが 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"
}
}
scheduling フィールドを省略すると、Gemini Live API は下位互換性のために、関数レスポンスを処理する元のメソッドを使用します。
次の Python の例は、会話が自然に一時停止するまで結果を通知しないように、function_response
を scheduling="WHEN_IDLE" とともにフォーマットして送信する方法を示しています。
aearcync 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.
ベスト プラクティス
- 同時実行を考慮した設計: 外部 API のクエリや RAG パイプラインの実行など、遅いツールは常にバックエンドのバックグラウンド タスクにオフロードします。モデルがアクティブな音声ストリームを処理し続けるようにします。
- 必要な場合を除き、INTERRUPT を使用しない:
INTERRUPTは重要なアラートに使用します。 バックグラウンド タスクの場合、SILENTまたはWHEN_IDLEを使用すると、よりスムーズで丁寧なユーザー エクスペリエンスを提供できます。 - 独立したチャットターン: Gemini Live API では、ツールの実行は チャットターンとは完全に独立しています。ツールがバックグラウンドで処理されている間、会話は分岐、継続、自然な流れで進みます。
- 「Silent」の注意点:
SILENTとしてスケジュールされている場合でも、モデルがツールの実行を音声でナレーションしようとする場合があります。完全に無音にするには、システム手順に明示的なガードレールを適用します（例: 「[ツール名] を使用する場合は、SILENT EXECUTION を実行して何も言わない」）。または、FunctionResponseをモデルにまったく返さない「fire-and-forget」バックエンド パターンを使用します。