MENU
合同会社JOBIRUNのHPです。
  1. ホーム
  2. AI関係
  3. AIツール
  4. ［開発者向け］ADKとGemini Liveで作る「沈黙しない」音声エージェントの実装パターン

［開発者向け］ADKとGemini Liveで作る「沈黙しない」音声エージェントの実装パターン

AIツール AI技術 エンジニア関連
目次

はじめに

　Googleは2026年10月19日、ADK（Agent Development Kit）とgemini-3.8-liveで、ツール実行中も黙り込まない音声エージェントを作るガイドを公開しました。本稿では、会話を止めないツール実装のパターンをコード付きで解説します。

参考記事

要点

  • 音声会話では遅いツールの実行待ちがそのまま無音となり、プロンプト指示では解消できない。
  • ADKでは `return` を `yield` に変えるだけで、ツールがNON_BLOCKINGのストリーミングツールとして扱われる。
  • 進捗を伝えないバックグラウンド実行は、単一yieldのジェネレーターかFunctionToolで実装できる。
  • `response_scheduling` のWHEN_IDLE・INTERRUPT・SILENTで、結果を会話に届けるタイミングを制御できる。
  • Googleの検証では、ストリーミングツール化で最初の発話までの時間が14.4秒から1.2秒に短縮された。

詳細解説

音声エージェントが黙り込む問題とADKのアプローチ

　元記事は、ADKのGemini Live API Toolkitにおけるストリーミングツール、ノンブロッキングな関数呼び出し、レスポンススケジューリングを扱うガイドとして位置づけられています。出発点となる問題設定は明快です。テキストチャットであれば、13秒かかるデータベースクエリやRAG（検索拡張生成）パイプラインはローディング表示で待ってもらえます。しかしリアルタイムの音声会話では、13秒間ツールの実行がブロックされると13秒間の無音（デッドエア）が生じ、ユーザーは通話が切れたと考えて電話を切ってしまうと元記事は指摘しています。

　バックエンドのクエリを常に1秒未満にできるとは限らないため、解決策はアーキテクチャ側に求める必要がある、というのが元記事の立場です。具体的には、遅いツールをバックグラウンドで動かしながらエージェントに会話を続けさせます。ADK（Googleのオープンソースのエージェント開発フレームワーク）とgemini-3.8-liveを使えば、非同期のツール状態を管理するためにクライアント側でWebSocketのイベントループを独自に組まなくても、Pythonでノンブロッキングな音声ツールをネイティブに実装できるとされています。

　Googleによれば、同社がベンチマークとして用いた売上レポートのワークフローでは、gemini-3.8-live上でブロッキングツールをストリーミングツールに切り替えたところ、エージェントが最初の言葉を発するまでの時間が14.4秒から1.2秒に短縮されました。元記事には、スケジュールされた応答の動作を紹介する1分間のデモ動画「ADK Live AI Concierge」も掲載されています。

　この数値は特定のワークフローでの計測結果であり、ツールの処理内容やネットワーク環境によって変わる可能性があります。また、短縮されたのは「最初の発話までの時間」であって、処理そのものの時間ではない点に注意が必要だと思います。本稿としては、音声UIでは処理時間よりも「反応が返ってくるまでの時間」が体験を左右するという設計上の論点を、具体的な数値で示した例と受け止めています。

ノンブロッキングツールで実現できること

　元記事は、ADKでノンブロッキングなツール呼び出しを実装すると、次のようなユーザー体験上の利点がすぐに得られると説明しています。

  • 無音の解消（Zero Dead Air）：会話の途中で固まる代わりに、エージェントは「今その数字を調べています」のようにリクエストをすぐ受け止め、対話を続けられます。
  • 進捗の音声フィードバック（Audible Progress Updates）：接続・クエリ・照合といった複数段階のワークフローでは、途中のマイルストーンを読み上げることで、待ち時間を「作業が進んでいる時間」として聞かせられます。
  • マルチタスクと並行対話（Multitasking and Parallel Dialogue）：長時間のクエリがバックグラウンドで動いている間も、ユーザーは追加の質問や無関係な質問ができ、セッションを止めずに回答を受け取れます。
  • 結果の届け方の細かな制御（Granular Delivery Control）：最終結果をいつ、どのように会話に入れるかを開発者が決められます。自然な話の切れ目を待つ、重大なアラートとして即座に割り込む、会話コンテキストを黙って更新する、といった選択肢があります。

　4つ目の項目は、後述する `response_scheduling` の3つのモードにそれぞれ対応していると読めます。本稿としては、最初の3項目を「待たせ方」、4つ目を「結果の渡し方」の話として整理すると、以降の内容を追いやすいと考えています。

前提条件・環境構成

　元記事の「Requirements & Next Steps」に記載されている要件は次のとおりです。

  • フレームワーク：`google-adk >= 2.10.0`
  • モデル：Gemini Enterprise上の `gemini-3.8-live`
  • API環境：ストリーミングツールとライブスケジューリングには、双方向のライブセッション（`run_live`）が必要

　`run_live` は、Gemini Live APIで音声の入出力をリアルタイムにやり取りする双方向（bidirectional）セッションを指します。リクエストを送って応答を待つ通常の実行形態では、本稿で扱うストリーミングツールやスケジューリングはそのまま使えない前提である点に注意してください。

　なお、後述の機能の一部にはバージョン条件があり、FunctionToolの `behavior` 属性はADK 2.10以降、更新ごとのスケジューリング指定はADK 2.8以降とされています。導入時はpip（Pythonのパッケージ管理ツール）などでインストール済みのgoogle-adkのバージョンを確認しておくと安心だと思います。

　注意: 元記事のコードブロックは「Plain text」として表示されていますが、内容はPythonのため、本稿では言語指定をpythonとして掲載します。コードは元記事のサンプルをそのまま転載したもので、本稿では実行による動作確認を行っていません。

出発点：標準ツールが会話を止める理由

　一般的なエージェント開発では、ツールは処理を終えてから単一の値を返す `async def` 関数として定義されます。次のコードは、地域ごとの四半期売上レポートを実行し、完了時に結果を1回だけ返す標準的なツールの例です。

from typing import Any

async def query_sales_database(region: str) -> dict[str, Any]:
    """Runs the quarterly sales report query for a region.

    Args:
      region: Sales region to report on, e.g. "EMEA" or "APAC".

    Returns:
      A dict containing report metrics upon completion.
    """
    report = await run_expensive_query(region)       # e.g., 13 seconds of work
    return {"status": "success", "report": report}   # Single return at the end

　`run_expensive_query` は、コード内のコメントにあるとおり13秒程度かかる重い処理を表すもので、元記事内では定義されていません。手元で試す場合は、`asyncio.sleep` などで代替する必要があると考えられます。

　この関数をADKの `Agent` に直接渡すと、ブロッキングなツール呼び出しとして扱われます。Gemini Live APIの双方向音声セッション（`run_live`）では、実行ループがツールの戻り値を待ってから次のモデルのターンを生成します。そのため、`query_sales_database` に13秒かかれば、エージェントは13秒間完全に沈黙します。元記事によれば、「処理中だとユーザーに伝えて」といったプロンプト指示ではこれを解決できません。モデルのターン自体が関数レスポンスのペイロードを待ってブロックされているためです。

　会話を滑らかに保つには、求めるユーザー体験に応じて、2つの主要なノンブロッキングパターンから選ぶことになります。本稿としては、この「プロンプトでは直せない」という指摘が実装上の重要な分岐点だと考えています。音声エージェントの無言状態は指示文の問題に見えがちですが、元記事の説明に従えば、ツール呼び出しの実行モデルの問題として扱う必要があるという整理です。

パターン1：ストリーミングツールで進捗を語らせる

　データベースへの接続、レコードの集計、結果の整形のように、処理が意味のある段階で構成されている場合は、ストリーミングツールを使うとエージェントが進捗をその場で語れるようになります。元記事はこれを「進捗に物語がある場合（When Progress Has a Story）」のパターンとしています。

　ADKで標準ツールをストリーミングツールに変えるには、`return` を `yield` に変えるだけで済みます。関数を `AsyncGenerator`（非同期ジェネレーター：値を一度に返さず、少しずつ順番に送り出す非同期関数）として定義すると、ADKが自動的にツール宣言を `NON_BLOCKING` としてマークし、モデルにすぐ状態の受領通知を返します。次のコードは、4段階の進捗メッセージを順に送り出し、最後に結果の要約を返すツールと、それを使うエージェントの定義例です。

import asyncio
from typing import AsyncGenerator
from google.adk.agents import Agent

async def query_sales_database(region: str) -> AsyncGenerator[str, None]:
    """Runs the quarterly sales report. Call this ONCE; it pushes updates automatically.

    Args:
      region: Sales region to report on, e.g. "EMEA" or "APAC".

    Yields:
      Progress updates while the query runs, followed by the final summary.
    """
    stages = [
        ("Connecting to the sales warehouse...", 1.5),
        ("Scanning order records...", 3.5),
        ("Aggregating revenue by product line...", 3.5),
        ("Formatting the report...", 1.5),
    ]
    for message, delay in stages:
        yield f"PROGRESS: {message}"
        await asyncio.sleep(delay)

    yield f"DONE: {region} Q2 revenue reached $4.81M, up 12.4% year over year."

agent = Agent(
    name="sales_analyst",
    model="gemini-3.8-live",
    tools=[query_sales_database],
    instruction=(
        "You are a helpful sales assistant. When the user asks about sales data, "
        "call query_sales_database once. Acknowledge the request immediately, "
        "briefly paraphrase each progress update, and present the final figures "
        "when completed. Keep answers conversational and concise."
    ),
)

　`stages` の各要素は「読み上げ用のメッセージ」と「待機秒数」の組で、`asyncio.sleep` が各段階の処理時間を模しています。待機時間は合計10秒です。`PROGRESS:` と `DONE:` の接頭辞で途中経過と最終結果を区別しており、最後の売上値（$4.81M、前年比12.4%増）はコード内に固定で書かれたデモ用の文字列です。`instruction` では、`query_sales_database` を1回だけ呼ぶこと、リクエストをすぐ受け止めること、各進捗を短く言い換えること、完了時に数字を伝えることを指示しています。docstringの「Call this ONCE」という記述は、後述する重複呼び出し対策の一部でもあります。

　本稿としては、`PROGRESS:` のような接頭辞は、モデルが途中経過と最終結果を取り違えにくくするための手がかりとして機能すると考えています。実際の業務ツールに置き換える場合も、段階ごとのメッセージ形式をそろえておくと、言い換えの品質を安定させやすい可能性があります。

ストリーミングツールの実際の動き

　元記事では、このストリーミングツールが実際の会話の中で次のように動くと説明されています。

  1. 即時の受け止め：モデルが `query_sales_database` を呼び出すと、ADKはすぐに非同期の状態イベントをモデルに送ります。エージェントはその場で「EMEAの数字を今お調べしています」のように話し始めます。
  2. 部分的な更新：`yield` のたびに部分的な関数レスポンスがモデル側へ送られ、モデルは届いた更新を順に言い換えて伝えます（「倉庫に接続しています」「製品ライン別に集計しています」など）。ただし、読み上げはツールの実際の進行から数秒遅れることがあります。
  3. 途切れない並行対話（Seamless Multi-Threading）：クエリの実行中にユーザーが無関係な質問をしても、エージェントはバックグラウンドのタスクを取り消したり中断したりせずに答えられます。

　2番目の「数秒遅れることがある」という点は、進捗メッセージの粒度を設計するうえで参考になると思います。段階が細かすぎると読み上げが追いつかず、実際の進行と発話がずれる可能性があるため、元記事のサンプルのように数段階程度に抑えるのが無難だと考えられます。

ストリーミングツールの本番運用での考慮点

　元記事は、ストリーミングツールを本番で使う際の注意点として次の3点を挙げています。

  • 単一実行（Single-Flight）のツール呼び出し：ストリーミングツールは複数のターンにまたがって更新を送るため、繰り返し促されるとモデルがツールを再度呼び出そうとすることがあります。docstringに明確な「Call this ONCE」の指示を残し、本番での安全策としてツール実装側にインメモリのラッチ（実行中かどうかを示すフラグ）や冪等キー（同じリクエストの重複実行を防ぐ識別子）を導入することを検討するよう、元記事は勧めています。
  • `stop_streaming` によるユーザー側のキャンセル：ADKには、長時間動くストリーミングツール向けの組み込みの「脱出口」があります。`stop_streaming(function_name: str)` という名前の関数をエージェントのtoolsリストに含めると、ADKがその呼び出しを横取りし、ジェネレーターの裏で動く `asyncio.Task` をキャンセルします。関数内では `asyncio.CancelledError` が発生するため、データベース接続やファイルハンドルをきれいに閉じられます。
  • ユーザーの割り込み（バージイン）ではツールは止まらない：ユーザーがエージェントの発話にかぶせて話すと、Live APIはモデルの出力を止め、ADKは `interrupted=True` のイベントを発行します。クライアントはこれを受けて音声再生を止められます。ただし、実行中のストリーミングツールやバックグラウンドツールはキャンセルされず、その結果もそのまま届きます。ツールを止めるには、モデルが `stop_streaming` を呼び出す必要があります。

　3点目は見落としやすいと思います。ユーザーが話を遮ったからといって処理も止めたいとは限らないため、ツールが動き続ける設計自体は合理的だと考えられます。一方で、「やっぱりやめて」と言われた場合にツールを止めたいなら、`stop_streaming` をtoolsに含めておく必要があります。`asyncio.CancelledError` を受けたときの後始末（try/finallyでの接続クローズなど）を用意しておくことも、元記事の説明から自然に導かれる実装上の配慮だと思います。

パターン2：バックグラウンド実行で静かに待つ

　進捗を逐一実況することが、常にふさわしいとは限りません。ユーザーがレポートを頼んですぐ別の話題に移った場合、状態の更新で何度も割り込んでくるエージェントは押し付けがましく感じられることがあると元記事は述べています。待ち時間に語るべき物語がないなら、タスクをバックグラウンドで静かに実行する、というのがこのパターンの考え方で、元記事はその実装方法としてアプローチAとアプローチBの2つを示しています。

　本稿としては、パターン1との違いは技術的な仕組みよりも、「その待ち時間をユーザーがどう過ごしたいか」という体験設計の違いだと受け止めています。

アプローチA：一度だけyieldするジェネレーター（最速のセットアップ）

　継続的な進捗報告なしで、すぐに口頭での受け止めだけを得たい場合に有効なのが、完了時に1回だけyieldする非同期ジェネレーターです。次のコードは、重い処理の完了を待ってからレポートを1回だけ送り出す例です。

async def query_sales_database(region: str) -> AsyncGenerator[dict, None]:
    """Runs the quarterly sales report. Call this ONCE; results arrive when ready."""
    report = await run_expensive_query(region)   # Background execution (~13s)
    yield report                                 # Single yield upon completion

　これは非同期ジェネレーターなので、ADKはすぐに保留中（pending）のレスポンスを発行します。エージェントは「今そのレポートをまとめています」のようにすぐ受け止め、計算中は黙っていて、結果が揃った時点、またはユーザーが話し終えた時点で数字を伝えます。

　このスニペットには `AsyncGenerator` のimportや `run_expensive_query` の定義が含まれていないため、パターン1のコードと同じファイルに置く前提と読めます。単体で試す場合は、パターン1のimport文などを補う必要があると考えられます。パターン1との差は `yield` の回数だけで、同じNON_BLOCKINGの仕組みのまま「受け止め＋最終結果」に絞れる点が、元記事が「最速のセットアップ」と呼ぶ理由だと思います。

アプローチB：FunctionToolとresponse_scheduling

　標準的なコルーチン（`async def … -> dict` の形）のままにしておきたい場合は、関数を `FunctionTool` で包み、`response_scheduling` を設定します。次のコードは、標準ツール版の `query_sales_database` を `FunctionTool` でラップし、結果をWHEN_IDLEで届けるよう指定する例です。

from google.adk.tools import FunctionTool
from google.genai import types

sales_tool = FunctionTool(query_sales_database)
sales_tool.response_scheduling = types.FunctionResponseScheduling.WHEN_IDLE
(Note: In ADK 2.10+, you can also set `sales_tool.behavior = types.Behavior.NON_BLOCKING` to run a tool in the background without modifying scheduling mode.)

　最後の行の「(Note: …)」は、元記事のコードブロック内にそのまま書かれた注記です。Pythonの構文としては有効ではないため、実際のコードに含める際はコメントにするか取り除く必要があると考えられます。注記の内容は、ADK 2.10以降では `sales_tool.behavior = types.Behavior.NON_BLOCKING` を設定することで、スケジューリングのモードを変えずにツールをバックグラウンドで実行できる、という補足です。ここで渡している `query_sales_database` は、ベースラインで示した、値を1回だけ返す版を想定していると読めます。

　元記事は、このアプローチのトレードオフも説明しています。一度だけ値を返すコルーチンに `response_scheduling` を使う場合、関数全体が完了するまでモデルにはペイロードが届きません。そのため、ユーザーが話し続けない限り、モデルは通常、最初の口頭での受け止めをしません。このパターンは、ユーザーがすぐに別の質問を続け、結果を次の自然な切れ目で会話に織り込みたい場合に適しているとされています。

　アプローチAとBはどちらも「静かに待つ」パターンですが、受け止めの一言がすぐに入るか（A）、入らないか（B）が実用上の分かれ目と読めます。「承りました」の一言が安心感につながる場面ではA、ユーザーが自分から話を進めることが多い場面ではBが合う、という見方もできると思います。

response_schedulingによる結果の届け方の制御

　`response_scheduling` 属性は、届いたツールレスポンスにモデルが「いつ」反応するかを制御します。WHEN_IDLEは進行中のターンに割り込むことがなく、モデルがまだ話している、またはユーザーがまだ話している場合は、そのターンが終わるまで結果を待たせ、次の自然な切れ目でモデルが応答します。元記事によれば、ADKは3種類のスケジューリング動作をサポートしており、その比較は表（画像）で示されています。本稿では表の画像の細部を参照できないため、本文の記述をもとに3つのモードを整理します。

　Googleがgemini-3.8-liveで行ったベンチマークセッションでは、各モードは次のように振る舞いました。

  • WHEN_IDLE：届いた売上レポートは、モデルが別の20秒の質問への回答を終えるまで待機し、その後に数字を滑らかに伝えました。
  • INTERRUPT：届いた結果が話の途中で発話を遮り、その後モデルは回答を最後まで終えてから数字を付け加えました。
  • SILENT：結果は会話履歴に追加され、頼まれていない読み上げを促すことなく、以降のターンで利用できる状態になりました。

　これらは元記事のベンチマークセッションで観察された挙動であり、モデルのバージョンや会話の流れによって細部は変わる可能性があります。使い分けの目安としては、ノンブロッキングツールの利点で挙げられていた「自然な切れ目を待つ」「重大なアラートで即座に割り込む」「会話コンテキストを黙って更新する」が、それぞれWHEN_IDLE、INTERRUPT、SILENTに当たると読めます。SILENTはユーザーが後から尋ねたときに答えられればよい補助的な情報に向き、INTERRUPTは発話を遮るため、緊急性の高い通知などに絞って使うのが無難だと思います。

ストリーミングとスケジューリングの組み合わせ

　実行の形（ストリーミングか単一の戻り値か）と、届けるタイミング（`response_scheduling`）は互いに独立しており、組み合わせて使えます。ストリーミングツールに `response_scheduling` を設定すると、すべての部分更新にその動作が付与されます。

　さらに、ADK 2.8以降では、ストリーミングツールが `types.FunctionResponse` をyieldすることで、更新ごとにスケジューリングを動的に指定できます。次のコードは、途中経過はSILENTで静かに渡し、最終結果はWHEN_IDLEで話者が話し終えたタイミングに届ける例です。

# Acknowledge and keep intermediate progress quiet:
yield types.FunctionResponse(
    response={"stage": "aggregating"},
    scheduling=types.FunctionResponseScheduling.SILENT,
)

# Deliver the final result when the speaker finishes:
yield types.FunctionResponse(
    response={"report": final_data},
    scheduling=types.FunctionResponseScheduling.WHEN_IDLE,
)

　このコードはストリーミングツールの関数本体に置く `yield` 文の抜粋で、`final_data` や `types` のimportは周辺のコードで用意されている前提と読めます。本稿としては、この組み合わせにより「受け止めは即座に、途中経過は黙って、結果は切れ目で」という、パターン1とパターン2の中間的な体験を1つのツールで組み立てられる点が実用的だと考えています。

アーキテクチャのまとめ：ツールパターンの選び方

　元記事では、ツールパターンの選び方を表（画像）でまとめたうえで、次の経験則を示しています。

　「待ち時間に語るべき物語があるならストリーミングにする。待ち時間がただの待ち時間なら、バックグラウンドで実行し、アイドル時に結果を届ける」

　表の画像の内容は本稿では参照できないため、以下は本文の記述をもとに本稿が整理した選択の目安です。

  • 処理に意味のある段階があり、進捗を伝えたい：ストリーミングツール（`return` を `yield` に変更）
  • すぐに受け止めの一言だけ返し、途中は黙って結果を待ちたい：単一yieldのジェネレーター（アプローチA）
  • 標準のコルーチンを維持し、結果を次の自然な切れ目で届けたい：FunctionTool＋response_scheduling（アプローチB）
  • 途中経過と最終結果で届け方を変えたい：ストリーミングツールで `types.FunctionResponse` をyieldし、更新ごとに `scheduling` を指定（ADK 2.8以降）

結論：変わるのは処理時間ではなく会話上の存在感

　元記事は、ノンブロッキングツールはバックエンドの実行時間を短くするものではないと強調しています。13秒のクエリは依然として13秒かかり、変わるのはエージェントの会話上の存在感です。読み込みに10秒かかるWebアプリケーションは「遅い」と受け取られますが、10秒黙り込む音声エージェントは「壊れている」と受け取られる、というのが元記事の対比です。そのうえで、ADKのストリーミングツールと柔軟なレスポンススケジューリングを、gemini-3.8-liveのリアルタイム機能と組み合わせることで、複雑な処理の最中も注意深く自然に応対し続ける音声アシスタントを構築できるとしています。

　元記事では、同じ13秒のクエリをすべてのツールパターンと `response_scheduling` のモードで実行し、違いを聞き比べられるインタラクティブデモも公開されています。

　本稿としては、14.4秒から1.2秒という数値も「処理が速くなった」ではなく「反応が速くなった」と読むのが、この結論と整合する解釈だと考えています。実際の導入では、バックエンド自体の高速化と、本稿で見た会話設計の両面から検討するのが現実的だと思います。

参考資料

　次の参考資料が挙げられています。

　ストリーミングツールの仕様はStreaming Tools、FunctionToolの定義方法はFunction toolsのドキュメントが、それぞれ次に読む資料の入口になると考えられます。

まとめ

　ADKとgemini-3.8-liveでは、ストリーミングツールやresponse_schedulingを使い分けることで、遅いツールの実行中も会話を止めない音声エージェントを構築できます。本稿としては、処理時間ではなく「待ち方」と「結果の届け方」を設計する視点が要点だと考えています。

AIツール AI技術 エンジニア関連
Gemini Google 音声

この記事が気に入ったら
フォローしてね！

  • URLをコピーしました！
  • URLをコピーしました！

関連記事

目次