取り組みの背景：なぜ「本番グレード」の音声AIが難しいのか
Gemini Live APIを使えば、数十行のPythonコードで音声会話AIを動かすことができます。しかし、そのプロトタイプを実際のサービスとして展開しようとした瞬間、多くの開発者は壁にぶつかります。
- 接続の安定性：WebSocketが予期せず切断される
- Function Callingの遅延：外部API呼び出し中に会話が途切れる
- スケーリング：同時接続数が増えるとコストと品質が両立しない
- デプロイ：ローカルでは動くのにCloud Runに乗せると動かない
前提知識: Gemini Live APIの基本については Gemini Live API 完全ガイド を参照してください。
前提知識・環境構築
必要なもの
- Python 3.11以上
- Google Cloud プロジェクト（Vertex AI API有効化済み）
google-genaiSDK 1.10.0以上google-adk0.4.0以上- Cloud Run / Docker（本番デプロイ用）
インストール
pip install google-genai>=1.10.0 google-adk>=0.4.0 \
fastapi uvicorn websockets aiohttp python-dotenv環境変数
# .env
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json
# またはAPIキー認証（開発時のみ推奨）
GEMINI_API_KEY=your-api-keyアーキテクチャの設計思想
3層構造で本番品質を確保する
本番グレードの音声AIエージェントには、以下の3層構造が有効です。
クライアント（ブラウザ/モバイル）
↕ WebSocket (wss://)
┌─────────────────────────────┐
│ FastAPI WebSocket Server │ ← 層1：通信管理
│ (Cloud Run / GKE) │
└──────────┬──────────────────┘
│
┌──────────▼──────────────────┐
│ Agent Orchestration Layer │ ← 層2：エージェント制御
│ (Google ADK + Router) │
└──────────┬──────────────────┘
│
┌──────────▼──────────────────┐
│ Gemini Live API + Tools │ ← 層3：AI推論 + ツール実行
│ (gemini-live-2.5-flash) │
└─────────────────────────────┘
この設計の重要なポイントは、通信管理とエージェント制御を分離していることです。Gemini Live APIのセッションが切断されても、WebSocketサーバー層でリトライやフォールバックを処理でき、クライアントには影響を与えません。
ステップ1：Google ADKによるエージェント定義
Google ADK（Agent Development Kit）は、Geminiエージェントの構造化された定義と実行フレームワークです。まずエージェントの「役割」とツールを定義します。
# agent.py
import asyncio
import json
from datetime import datetime
from google.adk.agents import LiveAgent
from google.adk.tools import function_tool
from google.genai.types import LiveConnectConfig, SpeechConfig, VoiceConfig
# --- Function Calling ツールの定義 ---
@function_tool
async def get_current_weather(city: str) -> dict:
"""指定された都市の現在の天気を取得する。
Args:
city: 天気を調べる都市名（例: "Tokyo", "New York"）
Returns:
気温・天気・湿度などの情報を含む辞書
"""
# 実際の実装では OpenWeatherMap API などを呼び出す
# ここではモックデータを返す
mock_weather = {
"Tokyo": {"temp": 18, "condition": "晴れ", "humidity": 55},
"Osaka": {"temp": 20, "condition": "くもり", "humidity": 60},
}
data = mock_weather.get(city, {"temp": 15, "condition": "不明", "humidity": 50})
return {
"city": city,
"temperature": data["temp"],
"condition": data["condition"],
"humidity": data["humidity"],
"timestamp": datetime.now().isoformat(),
}
@function_tool
async def search_knowledge_base(query: str, category: str = "general") -> dict:
"""社内ナレッジベースを検索して関連情報を返す。
Args:
query: 検索クエリ
category: 検索カテゴリ（general / product / support）
Returns:
検索結果のリスト
"""
# 実際はベクトルDB（Vertex AI Vector Search 等）を参照
return {
"query": query,
"category": category,
"results": [
{"title": f"「{query}」に関するドキュメント", "relevance": 0.92},
{"title": f"{category}カテゴリのFAQ", "relevance": 0.78},
],
}
@function_tool
async def create_support_ticket(
issue_summary: str,
priority: str = "medium",
user_id: str = ""
) -> dict:
"""サポートチケットを作成する。
Args:
issue_summary: 問題の概要
priority: 優先度（low / medium / high / urgent）
user_id: ユーザーID
Returns:
作成されたチケットの情報
"""
import random
ticket_id = f"TKT-{random.randint(10000, 99999)}"
return {
"ticket_id": ticket_id,
"summary": issue_summary,
"priority": priority,
"status": "created",
"estimated_response": "2時間以内" if priority in ["high", "urgent"] else "1営業日以内",
}
# --- ADKエージェントの設定 ---
def create_voice_agent() -> LiveAgent:
"""本番グレードの音声AIエージェントを作成する。"""
return LiveAgent(
model="gemini-live-2.5-flash-native-audio",
system_instruction="""あなたは丁寧で有能なカスタマーサポートAIアシスタントです。
主な役割:
- お客様の質問に音声で分かりやすく回答する
- 必要に応じて天気情報・ナレッジベース検索・チケット作成を実行する
- 長い説明は短く区切って、会話のリズムを保つ
- 不明な点は確認してから行動する
制約:
- 個人情報（パスワード、クレジットカード番号など）は扱わない
- 確認できない事実を断定的に述べない""",
tools=[get_current_weather, search_knowledge_base, create_support_ticket],
config=LiveConnectConfig(
response_modalities=["AUDIO"],
speech_config=SpeechConfig(
voice_config=VoiceConfig(
prebuilt_voice_config={"voice_name": "Aoede"}
)
),
),
)ポイント: ADKの @function_tool デコレータを使うと、Pythonのdocstringから自動的にFunction Callingのスキーマが生成されます。型アノテーションも活用されるため、手動でJSONスキーマを書く必要がありません。
ステップ2：WebSocketサーバーの実装（接続管理・再接続ロジック付き）
本番環境での最大の課題は接続の安定性です。Gemini Live APIのセッションには上限（デフォルト15分）があり、ネットワークの問題でも切断される場合があります。
# server.py
import asyncio
import logging
from typing import Optional
import websockets
from websockets.exceptions import ConnectionClosed
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from agent import create_voice_agent
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="Voice Agent API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 本番では特定ドメインに制限すること
allow_methods=["*"],
allow_headers=["*"],
)
class VoiceAgentSession:
"""音声エージェントの1セッションを管理するクラス。"""
MAX_RETRIES = 3
RETRY_DELAY = 1.5 # seconds
def __init__(self, session_id: str):
self.session_id = session_id
self.agent = create_voice_agent()
self._live_session = None
self._retry_count = 0
async def _connect_live_session(self):
"""Gemini Live APIへの接続（再試行ロジック付き）。"""
while self._retry_count < self.MAX_RETRIES:
try:
self._live_session = await self.agent.connect()
self._retry_count = 0 # 接続成功時にリセット
logger.info(f"[{self.session_id}] Live API connected")
return
except Exception as e:
self._retry_count += 1
wait = self.RETRY_DELAY * (2 ** (self._retry_count - 1)) # 指数バックオフ
logger.warning(
f"[{self.session_id}] Connection failed ({self._retry_count}/{self.MAX_RETRIES}): {e}. "
f"Retry in {wait:.1f}s"
)
await asyncio.sleep(wait)
raise RuntimeError(f"[{self.session_id}] Max retries exceeded")
async def handle(self, client_ws: WebSocket):
"""クライアントWebSocketとLive APIのブリッジ処理。"""
await client_ws.accept()
await self._connect_live_session()
send_task = asyncio.create_task(
self._forward_client_to_gemini(client_ws)
)
recv_task = asyncio.create_task(
self._forward_gemini_to_client(client_ws)
)
try:
done, pending = await asyncio.wait(
[send_task, recv_task],
return_when=asyncio.FIRST_EXCEPTION,
)
for task in pending:
task.cancel()
for task in done:
if exc := task.exception():
logger.error(f"[{self.session_id}] Task error: {exc}")
except WebSocketDisconnect:
logger.info(f"[{self.session_id}] Client disconnected")
finally:
await self._cleanup()
async def _forward_client_to_gemini(self, client_ws: WebSocket):
"""クライアントの音声データをGemini Live APIに転送。"""
async for message in client_ws.iter_bytes():
if self._live_session:
await self._live_session.send_audio(message)
async def _forward_gemini_to_client(self, client_ws: WebSocket):
"""GeminiのレスポンスをクライアントWebSocketに転送。"""
async for event in self._live_session.receive():
if event.type == "audio":
await client_ws.send_bytes(event.data)
elif event.type == "text":
# 字幕・ログ用テキストもJSONで送信
await client_ws.send_json({
"type": "transcript",
"text": event.text,
"turn": event.turn,
})
elif event.type == "tool_call":
logger.info(
f"[{self.session_id}] Tool call: {event.tool_name}({event.args})"
)
async def _cleanup(self):
"""セッションのクリーンアップ。"""
if self._live_session:
try:
await self._live_session.close()
except Exception:
pass
logger.info(f"[{self.session_id}] Session cleaned up")
@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
"""WebSocketエンドポイント。各接続で新しいエージェントセッションを生成。"""
session = VoiceAgentSession(session_id)
await session.handle(websocket)
@app.get("/health")
async def health_check():
return {"status": "ok", "model": "gemini-live-2.5-flash-native-audio"}ステップ3：並行Function Callingによる応答遅延の最小化
音声エージェントにおける最大のUX課題は「ツール呼び出し中の沈黙」です。Gemini Live APIはFunction Callingを実行している間、音声レスポンスが止まります。
これを解消するには、Function Callingを非同期・並行実行することが有効です。
# parallel_tools.py
import asyncio
from typing import Any
from google.adk.tools import function_tool
class ParallelToolExecutor:
"""複数のFunction Callingを並行実行するオーケストレーター。"""
def __init__(self, timeout_seconds: float = 5.0):
self.timeout = timeout_seconds
self._tool_registry: dict[str, callable] = {}
def register(self, name: str, fn: callable):
self._tool_registry[name] = fn
async def execute_parallel(
self, tool_calls: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""複数のツール呼び出しを並行実行し、結果をまとめて返す。
Args:
tool_calls: [{"name": "tool_name", "args": {...}}, ...]
Returns:
各ツールの実行結果リスト
"""
tasks = []
for call in tool_calls:
tool_fn = self._tool_registry.get(call["name"])
if tool_fn:
task = asyncio.create_task(
self._execute_with_timeout(tool_fn, call["args"])
)
tasks.append((call["name"], task))
results = []
for name, task in tasks:
try:
result = await task
results.append({"tool": name, "result": result, "status": "success"})
except asyncio.TimeoutError:
results.append({
"tool": name,
"result": {"error": "タイムアウト"},
"status": "timeout"
})
except Exception as e:
results.append({
"tool": name,
"result": {"error": str(e)},
"status": "error"
})
return results
async def _execute_with_timeout(self, fn: callable, args: dict) -> Any:
return await asyncio.wait_for(fn(**args), timeout=self.timeout)
# 使用例
executor = ParallelToolExecutor(timeout_seconds=4.0)
executor.register("get_current_weather", get_current_weather)
executor.register("search_knowledge_base", search_knowledge_base)
# 複数ツールを同時実行（天気 + ナレッジ検索を並行）
results = await executor.execute_parallel([
{"name": "get_current_weather", "args": {"city": "Tokyo"}},
{"name": "search_knowledge_base", "args": {"query": "返品ポリシー", "category": "support"}},
])期待される出力:
[
{
"tool": "get_current_weather",
"result": {"city": "Tokyo", "temperature": 18, "condition": "晴れ"},
"status": "success"
},
{
"tool": "search_knowledge_base",
"result": {"query": "返品ポリシー", "results": [...]},
"status": "success"
}
]応用パターン：会話コンテキストの永続化
複数回の接続をまたいでユーザーの会話履歴を保持する場合、Vertex AI Firestore または Memorystore（Redis）と組み合わせます。
# context_store.py
import json
from datetime import datetime, timedelta
from google.cloud import firestore
class ConversationContextStore:
"""ユーザーごとの会話コンテキストをFirestoreに永続化するストア。"""
TTL_HOURS = 24 # コンテキストの有効期限
def __init__(self, project_id: str):
self.db = firestore.AsyncClient(project=project_id)
async def save_turn(
self,
user_id: str,
turn: dict,
):
"""会話ターンを保存する。"""
doc_ref = self.db.collection("conversations").document(user_id)
await doc_ref.set(
{
"turns": firestore.ArrayUnion([{
**turn,
"timestamp": datetime.utcnow().isoformat(),
}]),
"last_updated": datetime.utcnow().isoformat(),
"expires_at": (datetime.utcnow() + timedelta(hours=self.TTL_HOURS)).isoformat(),
},
merge=True,
)
async def get_recent_context(
self,
user_id: str,
max_turns: int = 10
) -> list[dict]:
"""直近の会話ターンを取得する。"""
doc = await self.db.collection("conversations").document(user_id).get()
if not doc.exists:
return []
data = doc.to_dict()
turns = data.get("turns", [])
return turns[-max_turns:] # 直近N件のみ返すトラブルシューティング
問題1：WebSocketが数分で切断される
原因: Gemini Live APIのデフォルトセッション上限（約15分）または、Cloud Runのリクエストタイムアウト（デフォルト300秒）。
解決策:
- Cloud Runのタイムアウトを
--timeout=3600に設定 wrangler.tomlや Cloud Run設定でWebSocketのkeepaliveを有効化- クライアント側でping/pongを定期実行（30秒ごと推奨）
# クライアント側のkeepalive実装
async def keepalive_ping(websocket):
while True:
await asyncio.sleep(30)
try:
await websocket.ping()
except ConnectionClosed:
break問題2：Function Calling後に音声が途切れる
原因: ツール実行結果の返却が遅すぎる（3秒以上）とGeminiがタイムアウトする場合があります。
解決策:
- 重いAPIコールは事前にキャッシュ（Memorystore/Redis）
- Function Calling のタイムアウトを2〜3秒以内に設計
- ツール呼び出し中に「少々お待ちください」などの音声フィラーを先行再生
問題3：Cloud Runで PERMISSION_DENIED エラー
原因: サービスアカウントにVertex AIの必要なロールが付与されていません。
解決策:
# 必要なIAMロールを付与
gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
--member="serviceAccount:YOUR-SA@YOUR_PROJECT_ID.iam.gserviceaccount.com" \
--role="roles/aiplatform.user"Cloud Runへのデプロイ
Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Cloud Run は $PORT を設定する
ENV PORT=8080
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8080"]デプロイコマンド
# Artifact Registry にプッシュ
gcloud builds submit --tag gcr.io/YOUR_PROJECT_ID/voice-agent .
# Cloud Run にデプロイ（WebSocket対応の設定）
gcloud run deploy voice-agent \
--image gcr.io/YOUR_PROJECT_ID/voice-agent \
--platform managed \
--region us-central1 \
--timeout 3600 \
--concurrency 80 \
--min-instances 1 \
--max-instances 20 \
--set-env-vars GOOGLE_CLOUD_PROJECT=YOUR_PROJECT_ID \
--allow-unauthenticated重要: --min-instances 1 を設定することでコールドスタートによる初回レイテンシを回避します。音声エージェントは最初の100msの遅延がユーザー体験に直結するため、常に1インスタンス以上を起動しておくことを推奨します。
コスト・パフォーマンス考慮事項
料金の目安（2026年3月時点）
| 項目 | 単価 | 1,000セッション/月の試算 |
|---|---|---|
| Gemini Live API（音声入力） | $0.70/100万トークン | 〜$14 |
| Gemini Live API（音声出力） | $2.80/100万トークン | 〜$56 |
| Cloud Run（実行時間） | $0.00002400/vCPU秒 | 〜$24 |
| Firestore（読み書き） | $0.06〜0.18/100万ops | 〜$5 |
| 合計 | — | 〜$99 |
コスト最適化の3つの戦略
① VAD（音声区間検出）でトークン節約: 無音区間の音声データをGeminiに送らないようにすると、入力トークンを30〜50%削減できます。webrtcvad ライブラリが有効です。
② セッション共有（非推奨を理解した上で）: 同一ユーザーの短期間内の再接続では、新しいセッションを立てずにLive APIセッションを再利用できる場合があります。ただし会話のコンテキストが正しく引き継がれるか検証が必要です。
③ Flash vs Pro の使い分け: 汎用カスタマーサポートには gemini-live-2.5-flash-native-audio、高精度な専門知識が必要な場面では gemini-live-2.5-pro-native-audio（将来提供予定）を使い分けることでコスト効率を高められます。
まとめ
ここではGemini Live APIとGoogle ADKを組み合わせた本番グレード音声AIエージェントの構築方法を解説しました。
要点を整理すると：
- 3層アーキテクチャ（通信管理・エージェント制御・AI推論）で安定性を確保
- ADKの
@function_toolでFunction Callingの定義を宣言的かつシンプルに - 指数バックオフ付きの再接続ロジックで本番環境の接続安定性を向上
- 並行Function Callingでツール実行中の沈黙を最小化
- Cloud Run + min-instances 1でコールドスタートを排除
次のステップとして、Gemini エージェント本番システム設計ガイド や Gemini Firebase GenKit実践ガイド も参照して、より高度なエージェントシステムの構築に挑戦してみてください。