Gemini 3.1 Flash Live APIガイド｜リアルタイム音声AI実装

2026.04.03 公開 22分で読める

#AIエージェント #Gemini #live-api #Python #リアルタイム音声

この記事の結論

Gemini Flash Liveのリアルタイム音声・映像APIをPythonで実装する方法。ストリーミング処理のコード例付き。

「STTとTTSを組み合わせても、どうしても会話が不自然になる…」

音声AIを開発したことがある人なら、この問題に直面したことがあるはずだ。Speech-to-Textでテキスト化 → LLMで処理 → Text-to-Speechで出力、というパイプラインは処理ごとにレイテンシが積み重なり、ユーザーが話し終わってから応答が来るまで2〜3秒かかることも珍しくない。

Googleが2026年3月末にリリースしたGemini 3.1 Flash Liveは、このパイプラインを根本から変える。音声が入力されると同時に音声が出力される「音声対音声（audio-to-audio）」の設計で、人間の会話に近いリアルタイムインタラクションが実現できる。しかも割り込みに対応しているため、ユーザーが話しかけると即座に応答を止めて聞き取り直す。

この記事では、Gemini 3.1 Flash Live APIの仕組みとPythonでの実装方法を、コピペで動くコード付きで解説する。

AIエージェントの基本的な構築パターンについては、AIエージェント構築完全ガイドもあわせて参照してほしい。

まず5分で試せる最小実装

まずは動かしてみよう。以下はテキストを送るだけのシンプルな接続コードだ。


# 動作環境: Python 3.11+, google-genai>=1.10.0
# pip install google-genai

import asyncio
from google import genai

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# APIキーは環境変数で管理すること（ハードコード禁止）

import os
GEMINI_API_KEY = os.environ["GEMINI_API_KEY"]
client = genai.Client(api_key=GEMINI_API_KEY)

MODEL = "gemini-3.1-flash-live-preview"

async def quick_start():
    """Gemini 3.1 Flash Live APIの最小接続例"""
    config = {
        "response_modalities": ["TEXT"],  # まずはテキスト応答で動作確認
    }

    async with client.aio.live.connect(model=MODEL, config=config) as session:
        print("接続完了。テキストを送信してみます...")

        await session.send_message("こんにちは！Gemini 3.1 Flash Liveのテストです。")

        async for response in session.receive():
            if response.text:
                print(f"応答: {response.text}")
            if response.server_content and response.server_content.turn_complete:
                break

        print("接続終了")

asyncio.run(quick_start())

ポイント:

`gemini-3.1-flash-live-preview` が正式なモデルIDだ。`gemini-3.1-flash-live` と省略すると404になるので注意
`response_modalities` で「TEXT」または「AUDIO」を指定する。音声出力が不要な場面ではTEXTで十分
セッションはWebSocketベース。`async with` 構文でコンテキストマネージャとして扱う

Gemini 3.1 Flash Liveの全体像

従来のパイプラインとの違い

項目	従来（STT+LLM+TTS）	Gemini 3.1 Flash Live
処理方式	テキスト変換 → 処理 → 音声合成	音声to音声（ネイティブ処理）
レイテンシ	2〜4秒（積み上がる）	会話速度（リアルタイム）
割り込み対応	通常はなし（実装が複雑）	ネイティブ対応（VAD内蔵）
接続方式	REST API（リクエスト/レスポンス）	WebSocket（双方向持続接続）
文脈保持	手動でセッション管理	セッション内で自動保持
コスト	各API呼び出しごと	入力トークン/出力音声秒数

最終確認日: 2026-03-31（Google AI公式ドキュメントより）

モデルスペック（2026年3月時点）

モデルID: `gemini-3.1-flash-live-preview`
コンテキストウィンドウ: 131,072トークン（入力）/ 65,536トークン（出力）
対応入力: テキスト・音声・画像・動画
対応出力: テキスト・音声
音声セッション制限: 最大15分（音声+動画は2分）
対応言語: 97言語（日本語含む）

音声ストリーミングの実装（実用コード）

実際の音声入出力を扱う実装を見てみよう。マイク入力をリアルタイムでGeminiに送り、音声で返答を受け取る。


# 動作環境: Python 3.11+, google-genai>=1.10.0, pyaudio>=0.2.14
# pip install google-genai pyaudio numpy

import asyncio
import base64
import io
import os
import numpy as np
import pyaudio
from google import genai

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# PyAudioはシステムのPortAudioライブラリに依存します。
# macOS: brew install portaudio
# Ubuntu: apt-get install python3-pyaudio

GEMINI_API_KEY = os.environ["GEMINI_API_KEY"]
client = genai.Client(api_key=GEMINI_API_KEY)
MODEL = "gemini-3.1-flash-live-preview"

# 音声設定（Gemini Live APIの必須フォーマット）
INPUT_SAMPLE_RATE = 16000   # Geminiが要求するサンプルレート
OUTPUT_SAMPLE_RATE = 24000  # 出力音声のサンプルレート
CHUNK_SIZE = 1024           # 一度に処理するサンプル数
AUDIO_FORMAT = pyaudio.paInt16

async def audio_voice_session():
    """マイク → Gemini → スピーカーのリアルタイム音声会話"""

    audio = pyaudio.PyAudio()

    # マイク入力ストリーム
    input_stream = audio.open(
        format=AUDIO_FORMAT,
        channels=1,
        rate=INPUT_SAMPLE_RATE,
        input=True,
        frames_per_buffer=CHUNK_SIZE
    )

    # スピーカー出力ストリーム
    output_stream = audio.open(
        format=AUDIO_FORMAT,
        channels=1,
        rate=OUTPUT_SAMPLE_RATE,
        output=True
    )

    config = {
        "response_modalities": ["AUDIO"],
        "speech_config": {
            "voice_config": {
                "prebuilt_voice_config": {"voice_name": "Kore"}  # 日本語に適した声
            }
        }
    }

    print("接続中... 話しかけてください（Ctrl+Cで終了）")

    async with client.aio.live.connect(model=MODEL, config=config) as session:

        async def send_audio():
            """マイク音声を継続的に送信"""
            while True:
                chunk = input_stream.read(CHUNK_SIZE, exception_on_overflow=False)
                # PCMデータをbase64エンコードして送信
                encoded = base64.b64encode(chunk).decode()
                await session.send_realtime_input(
                    audio=genai.types.Blob(
                        data=encoded,
                        mime_type=f"audio/pcm;rate={INPUT_SAMPLE_RATE}"
                    )
                )
                await asyncio.sleep(0)  # イベントループに制御を返す

        async def receive_audio():
            """音声レスポンスを受信してスピーカーで再生"""
            async for response in session.receive():
                if response.data:
                    # バイナリ音声データを直接スピーカーに出力
                    output_stream.write(response.data)

                if response.server_content:
                    if response.server_content.interrupted:
                        print("[割り込み検知] ユーザーが話し始めました")
                    if response.server_content.turn_complete:
                        print("[応答完了]")

        # 送受信を並列実行
        try:
            await asyncio.gather(send_audio(), receive_audio())
        except KeyboardInterrupt:
            print("終了します")
        finally:
            input_stream.stop_stream()
            input_stream.close()
            output_stream.stop_stream()
            output_stream.close()
            audio.terminate()

asyncio.run(audio_voice_session())

ポイント:

入力音声は「16kHz、16-bit PCM、little-endian」が必須。ブラウザ収録音声（48kHz float32）は変換が必要
`interrupted` フラグを監視することで、ユーザーが割り込んだタイミングを検知できる。再生バッファをフラッシュする処理をここで入れる
`send_audio()` と `receive_audio()` を `asyncio.gather()` で並列実行するのが基本パターン

ツール呼び出し（Function Calling）の実装

Live APIの真価は「会話中にリアルタイムでツールを呼び出せる」点だ。音声会話中に外部APIを叩いて情報を取得し、そのまま応答に組み込める。


# 動作環境: Python 3.11+, google-genai>=1.10.0
# pip install google-genai

import asyncio
import json
import os
from google import genai
from google.genai import types

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

GEMINI_API_KEY = os.environ["GEMINI_API_KEY"]
client = genai.Client(api_key=GEMINI_API_KEY)
MODEL = "gemini-3.1-flash-live-preview"

# ツール定義（天気取得の例）
TOOLS = [
    types.Tool(
        function_declarations=[
            types.FunctionDeclaration(
                name="get_weather",
                description="指定した都市の現在の天気情報を取得します",
                parameters=types.Schema(
                    type="OBJECT",
                    properties={
                        "city": types.Schema(
                            type="STRING",
                            description="天気を調べる都市名（例: 東京、大阪）"
                        )
                    },
                    required=["city"]
                )
            )
        ]
    )
]

def execute_tool(function_name: str, args: dict) -> str:
    """ツールを実際に実行する（本番ではAPIを呼び出す）"""
    if function_name == "get_weather":
        city = args.get("city", "不明")
        # 本番実装では天気APIを呼び出す
        return json.dumps({
            "city": city,
            "temperature": "22°C",
            "condition": "晴れ",
            "humidity": "45%"
        }, ensure_ascii=False)
    return json.dumps({"error": f"未知のツール: {function_name}"})

async def tool_calling_session():
    """ツール呼び出し対応の会話セッション"""
    config = {
        "response_modalities": ["TEXT"],
        "tools": TOOLS,
        "system_instruction": "あなたは親切なアシスタントです。天気について聞かれたら必ずget_weatherツールを使って正確な情報を取得してください。"
    }

    async with client.aio.live.connect(model=MODEL, config=config) as session:

        await session.send_message("東京と大阪の天気を教えてください")

        async for response in session.receive():
            # ツール呼び出しの検知
            if response.tool_call:
                for fc in response.tool_call.function_calls:
                    print(f"[ツール呼び出し] {fc.name}({fc.args})")

                    # ツールを実行して結果を返す
                    result = execute_tool(fc.name, fc.args)
                    print(f"[ツール結果] {result}")

                    # 結果をセッションに返す（Gemini 3.1はシーケンシャル処理）
                    await session.send_tool_response(
                        function_responses=[
                            types.FunctionResponse(
                                name=fc.name,
                                id=fc.id,
                                response=json.loads(result)
                            )
                        ]
                    )

            # テキスト応答の受信
            if response.text:
                print(f"応答: {response.text}", end="", flush=True)

            if response.server_content and response.server_content.turn_complete:
                print()
                break

asyncio.run(tool_calling_session())

重要な注意点: Gemini 3.1 Flash LiveのFunction Callingはシーケンシャル処理のみ。Gemini 2.5 Flash Liveは非同期処理（NON_BLOCKING）をサポートするが、3.1版では一度に1つずつ処理する。並列ツール呼び出しが必要な場合はGemini 2.5 Flash Liveを検討すること。

ユースケース別の実装パターン

ユースケース	推奨設定	実装のポイント
カスタマーサポートBot	AUDIO出力 + Function Calling	システムプロンプトで業務スコープを厳格に定義。エスカレーションルールを明記
音声制御コーディング支援	TEXT出力 + 長いコンテキスト	コード内容をシステムプロンプトに含め、音声でコードレビューを指示
リアルタイム翻訳	AUDIO出力 + 言語設定	入力音声の言語を自動検出。出力言語をspeech_configで固定
フィールド技術者支援	AUDIO + VIDEO入力	カメラ映像を1fps程度で送りながら音声で質問。視覚情報とのマルチモーダル活用
アクセシビリティ支援	AUDIO出力	画面内容や画像の説明を音声でリアルタイム提供

WebSocket直接実装（SDKを使わない場合）

google-genaiライブラリが使えない環境では、生のWebSocketで接続できる。


# 動作環境: Python 3.11+, websockets>=12.0
# pip install websockets

import asyncio
import base64
import json
import os
import websockets

# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

API_KEY = os.environ["GEMINI_API_KEY"]
MODEL = "gemini-3.1-flash-live-preview"
WS_URL = (
    f"wss://generativelanguage.googleapis.com/ws/"
    f"google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent"
    f"?key={API_KEY}"
)

async def websocket_session(user_message: str):
    """生WebSocketでGemini Live APIに接続するシンプルな例"""

    async with websockets.connect(WS_URL) as ws:
        # 1. セットアップメッセージを送信
        setup = {
            "setup": {
                "model": f"models/{MODEL}",
                "generation_config": {
                    "response_modalities": ["TEXT"]
                },
                "system_instruction": {
                    "parts": [{"text": "あなたは親切な日本語アシスタントです。"}]
                }
            }
        }
        await ws.send(json.dumps(setup))

        # 2. setupCompleteを受信するまで待機
        while True:
            msg = json.loads(await ws.recv())
            if "setupComplete" in msg:
                print("セットアップ完了")
                break

        # 3. テキストメッセージを送信
        text_message = {
            "clientContent": {
                "turns": [{
                    "role": "user",
                    "parts": [{"text": user_message}]
                }],
                "turnComplete": True
            }
        }
        await ws.send(json.dumps(text_message))

        # 4. レスポンスを受信
        full_response = ""
        async for raw_msg in ws:
            msg = json.loads(raw_msg)

            if "serverContent" in msg:
                sc = msg["serverContent"]

                # テキスト部分を抽出
                if "modelTurn" in sc:
                    for part in sc["modelTurn"].get("parts", []):
                        if "text" in part:
                            full_response += part["text"]

                # ターン完了を確認
                if sc.get("turnComplete"):
                    print(f"応答: {full_response}")
                    break

asyncio.run(websocket_session("AIエージェントの基本について教えてください"))

ポイント: WebSocketプロトコルでは、最初に`setup`メッセージを送り`setupComplete`を受信してから本文を送る。SDKはこのハンドシェイクを内部で処理しているが、直接WebSocketを使う場合は手動で実装する。

【要注意】よくあるエラーと対策

失敗1：音声形式の不一致

❌ ブラウザのMediaRecorder APIで録音した音声（WebM/Opus形式、48kHz）をそのまま送る
⭕ 送信前に16kHz、16-bit PCM（little-endian）にダウンサンプリングする

対策コード:


import numpy as np

def downsample_audio(audio_48k: bytes, original_rate=48000, target_rate=16000) -> bytes:
    """48kHz float32から16kHz int16に変換"""
    samples = np.frombuffer(audio_48k, dtype=np.float32)
    # 単純なダウンサンプリング（本番ではscipy.signal.resampleを推奨）
    ratio = target_rate / original_rate
    new_length = int(len(samples) * ratio)
    downsampled = np.interp(
        np.linspace(0, len(samples), new_length),
        np.arange(len(samples)),
        samples
    )
    return (downsampled * 32767).astype(np.int16).tobytes()

失敗2：セッション切断時のハンドリング漏れ

❌ WebSocket接続が切れた時に何も処理しない（音声セッションは最大15分で自動終了）
⭕ 再接続ロジックと指数バックオフを実装する

なぜ重要か: 音声セッションは最大15分。長時間の運用では必ず切断が発生する。再接続時は新しいセッションとなるため、必要なコンテキストを再設定する処理が必要だ。

失敗3：割り込み後のバッファ処理漏れ

❌ ユーザーが割り込んでも、再生バッファの音声が流れ続ける
⭕ `interrupted` フラグを検知したら即座に出力バッファをフラッシュする

失敗4：コンテキストウィンドウの無管理

❌ 長時間セッションで会話履歴を蓄積し続けてコンテキスト上限に達する
⭕ 15分のセッション制限を逆手に取り、セッションを区切りごとにリセット。要約を新セッションのシステムプロンプトに引き継ぐ

参考・出典

Gemini Live API overview — Google AI for Developers（参照日: 2026-03-31）
Gemini 3.1 Flash Live Preview — Model Card — Google AI for Developers（参照日: 2026-03-31）
Get started with Gemini Live API using WebSockets — Google AI for Developers（参照日: 2026-03-31）
Live API capabilities guide — Google AI for Developers（参照日: 2026-03-31）
Build real-time conversational agents with Gemini 3.1 Flash Live — Google（参照日: 2026-03-31）

まとめ：今日から始める3つのアクション

今日やること: 「5分で試せる最小実装」のコードを動かして接続確認。`response_modalities: [“TEXT”]` から始めると音声デバイス不要でハードルが低い
今週中: ユースケース別実装パターンの中から自分のプロジェクトに近いものを選び、ツール呼び出しを1つ実装してみる。天気APIや社内データベース検索が最初の題材として扱いやすい
今月中: 音声ストリーミング実装を本番環境に移行。セッション管理（最大15分制限への対応、再接続ロジック）と割り込み処理を必ず実装してから公開する

あわせて読みたい:

AIエージェント構築完全ガイド — エージェント設計の基礎から応用まで
AIエージェントツール徹底比較 — Gemini含む主要フレームワークの選び方

AIエージェント・音声AI開発の相談は、Uravation お問い合わせフォームからお気軽にどうぞ。100社以上のAI研修・導入支援実績があります。

この記事はAIgent Lab編集部がお届けしました。

Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

30分無料相談を予約支援実績を見る

※ 本記事の情報は2026年4月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。