リアルタイム音声エージェントを構築する際に、一部の関数呼び出しによってモデルの実行がブロックされ、音声ストリームが停止してユーザーが沈黙の中で待機することになります。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 の例は、scheduling="WHEN_IDLE" を使用して function_response をフォーマットして送信し、会話の自然な一時停止を待ってから結果を発表する方法を示しています。
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を使用します。バックグラウンド タスクの場合、SILENTまたはWHEN_IDLEを使用すると、ユーザー エクスペリエンスがはるかにスムーズで丁寧になります。 - 独立したチャットターン: Gemini Live API では、ツールの実行はチャットターンとは完全に独立しています。ツールがバックグラウンドで処理されている間も、会話は分岐したり、継続したり、自然に流れたりします。
- 「サイレント」の注意点:
SILENTとしてスケジュールされている場合でも、モデルがツールの実行を音声で説明しようとすることがあります。完全にサイレントにするには、システム指示に明示的なガードレールを追加するか（例: 「[ツール名] を使用する場合は、サイレント実行を行い、何も言わない」）、FunctionResponseをモデルにまったく送信しない「ファイア アンド フォーゲット」バックエンド パターンを使用します。