?ref_src=twsrc%5Etfw">September 23,
2025
ソースコード: Get_started_LiveAPI_NativeAudio.py は文末にて！
まとめ
今回のLive APIのアップデートにより、
-
ファンクションコールの信頼性
-
人間らしい会話体験
-
さらに 「思考」機能 で複雑な質問にも対応可能
音声エージェント開発は新しいステージに進みました。Googleは「これにより、直感的で強力な音声体験の新しい可能性が開かれる」とコメントしており、今後さらに多くのアップデートが予定されています。
ソースコード: Get_started_LiveAPI_NativeAudio.py
# -*- coding: utf-8 -*-
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
このスクリプトは、Gemini のネイティブ音声モデルと Live API を使って、
マイクの音声をリアルタイムに送り、応答音声をそのまま再生します。
重要: 必ずヘッドホンを使用してください。エコーキャンセルは行いません。
スピーカーからの出力がマイクに回り込むと、モデルが自分の声を拾って
割り込みやハウリングの原因になります。
【セットアップ】
1) macOS の場合: brew install portaudio
2) Python 依存関係: pip install -U google-genai pyaudio
3) Python 3.11 未満なら: pip install taskgroup exceptiongroup
【API キー】
環境変数で API キーを設定してください。
- 推奨: export GOOGLE_API_KEY=あなたのキー
- 互換: export GEMINI_API_KEY=あなたのキー でも可（本スクリプトが自動で拾います）
【実行】
python Get_started_LiveAPI_NativeAudio.py
起動後に、手順と注意点を日本語で案内します。Ctrl+C で終了できます。
"""
from __future__ import annotations
import os
import sys
import asyncio
import traceback
from typing import Optional
try:
import pyaudio # type: ignore
except Exception as e: # pragma: no cover
print(
"[エラー] PyAudio がインポートできません。\n"
"- macOS: brew install portaudio\n"
"- その後: pip install -U pyaudio\n"
"- または Python 用の prebuilt wheel を利用してください。",
file=sys.stderr,
)
raise
from google import genai
from google.genai import types
# Python 3.11 未満では TaskGroup / ExceptionGroup を polyfill
if sys.version_info < (3, 11, 0):
try:
import taskgroup, exceptiongroup # type: ignore
except Exception as e: # pragma: no cover
print(
"[エラー] Python 3.11 未満かつ 'taskgroup' / 'exceptiongroup' が見つかりません。\n"
"pip install taskgroup exceptiongroup を実行してください。",
file=sys.stderr,
)
raise
asyncio.TaskGroup = taskgroup.TaskGroup # type: ignore[attr-defined]
asyncio.ExceptionGroup = exceptiongroup.ExceptionGroup # type: ignore[attr-defined]
FORMAT = pyaudio.paInt16
CHANNELS = 1
SEND_SAMPLE_RATE = 16000 # 送信は 16kHz PCM
RECEIVE_SAMPLE_RATE = 24000 # 応答は 24kHz PCM
# 実運用で安定しやすい 20ms 単位のフレーム長
# 16kHz の 20ms = 320 フレーム, 24kHz の 20ms = 480 フレーム
SEND_CHUNK_FRAMES = 320
RECV_CHUNK_FRAMES = 480
# 再生前にためるジッタバッファ（ミリ秒）。
# ネットワーク遅延や断続的な受信を吸収します。
PREBUFFER_MS = 120
pya = pyaudio.PyAudio()
def _build_client() -> genai.Client:
# 環境変数の両方に対応（GOOGLE_API_KEY 優先）
api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
if api_key:
# v1alpha でのプレビュー機能を使う場合の例
return genai.Client(api_key=api_key, http_options={"api_version": "v1alpha"})
# 未設定でも動く環境（ADC など）があればそれを使う
return genai.Client(http_options={"api_version": "v1alpha"})
SYSTEM_INSTRUCTION = """
You are a helpful and friendly AI assistant.
Your default tone is helpful, engaging, and clear, with a touch of optimistic wit.
Anticipate user needs by clarifying ambiguous questions and always conclude your responses
with an engaging follow-up question to keep the conversation flowing.
""".strip()
MODEL = "gemini-2.5-flash-native-audio-preview-09-2025"
BASE_CONFIG = {
"system_instruction": SYSTEM_INSTRUCTION,
"response_modalities": ["AUDIO"],
# プレビューのプロアクティブ音声を有効化（無視される環境もあります）
"proactivity": {"proactive_audio": True},
}
class AudioLoop:
def __init__(self):
self.audio_in_queue: Optional[asyncio.Queue[bytes]] = None
self.out_queue: Optional[asyncio.Queue[types.Blob]] = None
self.session = None
self.audio_stream = None
async def listen_audio(self, input_device_index: Optional[int] = None):
"""マイクから 16kHz/16bit/モノ の PCM を取り込み、送信用キューへ投入"""
mic_info = (
pya.get_device_info_by_index(input_device_index)
if input_device_index is not None
else pya.get_default_input_device_info()
)
stream = await asyncio.to_thread(
pya.open,
format=FORMAT,
channels=CHANNELS,
rate=SEND_SAMPLE_RATE,
input=True,
input_device_index=mic_info["index"],
frames_per_buffer=SEND_CHUNK_FRAMES,
)
self.audio_stream = stream
print(
f"[録音開始] デバイス: {mic_info.get('name')} / {SEND_SAMPLE_RATE}Hz / mono / 16-bit",
flush=True,
)
# デバッグ時はオーバーフローの例外を避けて継続
kwargs = {"exception_on_overflow": False} if __debug__ else {}
while True:
data: bytes = await asyncio.to_thread(stream.read, SEND_CHUNK_FRAMES, **kwargs)
# ライブラリ推奨の Blob 型で送る（MIME にサンプルレートを付与）
blob = types.Blob(data=data, mime_type="audio/pcm;rate=16000")
assert self.out_queue is not None
await self.out_queue.put(blob)
async def send_realtime(self):
assert self.out_queue is not None
while True:
blob = await self.out_queue.get()
await self.session.send_realtime_input(audio=blob)
async def receive_audio(self):
"""サーバからの応答（PCMチャンク/テキスト）を受信し再生キューへ"""
assert self.audio_in_queue is not None
while True:
# 1 ターンぶんのイベントを受信
turn = self.session.receive()
async for response in turn:
if response.data is not None:
self.audio_in_queue.put_nowait(response.data)
continue
# テキストが来る場合に備えて安全に取り出す
text = getattr(response, "text", None)
if text:
print(text, end="", flush=True)
# 以前は turn 完了時に未再生の音声を破棄していましたが、
# 不必要な音切れの原因になり得るためデフォルトでは破棄しません。
# （割り込み時の即時停止が必要なら実装を追加します）
async def play_audio(self, output_device_index: Optional[int] = None):
"""応答（24kHz/16bit/モノ PCM）をスピーカーへ再生"""
out_info = (
pya.get_device_info_by_index(output_device_index)
if output_device_index is not None
else pya.get_default_output_device_info()
)
stream = await asyncio.to_thread(
pya.open,
format=FORMAT,
channels=CHANNELS,
rate=RECEIVE_SAMPLE_RATE,
output=True,
frames_per_buffer=RECV_CHUNK_FRAMES,
output_device_index=out_info["index"],
)
print(
f"[再生開始] デバイス: {out_info.get('name')} / {RECEIVE_SAMPLE_RATE}Hz / mono / 16-bit",
flush=True,
)
# まずは少し貯めてから再生を開始（ジッタ吸収）
assert self.audio_in_queue is not None
bytes_per_sec = RECEIVE_SAMPLE_RATE * 2 # 16-bit mono
prebuffer_target = max(0, int(bytes_per_sec * (PREBUFFER_MS / 1000.0)))
prebuf = bytearray()
while len(prebuf) < prebuffer_target:
prebuf.extend(await self.audio_in_queue.get())
if prebuf:
await asyncio.to_thread(stream.write, bytes(prebuf))
# 通常再生ループ
while True:
bytestream = await self.audio_in_queue.get()
await asyncio.to_thread(stream.write, bytestream)
async def run(self, model: str, config: dict, *, in_dev: Optional[int], out_dev: Optional[int]):
client = _build_client()
print("\n=== 実行ガイド ===")
print("1) ヘッドホンを使用してください（エコー防止）")
print("2) マイクに向かって話しかけてください")
print("3) モデルの応答が聞こえます（割り込み可）")
print("4) 終了は Ctrl+C です\n")
print(f"Model: {model}")
print("Proactive audio:", config.get("proactivity"))
if "speech_config" in config:
print("Speech config:", config.get("speech_config"))
print("")
try:
async with (
client.aio.live.connect(model=model, config=config) as session,
asyncio.TaskGroup() as tg,
):
self.session = session
# 受信キューは多少余裕を持たせ、送信側の詰まりを緩和
self.audio_in_queue = asyncio.Queue(maxsize=200)
self.out_queue = asyncio.Queue(maxsize=20)
tg.create_task(self.send_realtime())
tg.create_task(self.listen_audio(input_device_index=in_dev))
tg.create_task(self.receive_audio())
tg.create_task(self.play_audio(output_device_index=out_dev))
except KeyboardInterrupt:
print("\n[終了] ユーザーにより中断されました。")
except asyncio.CancelledError:
pass
except asyncio.ExceptionGroup as eg: # type: ignore[attr-defined]
traceback.print_exception(eg)
finally:
# ストリームと PyAudio を確実に解放
try:
if self.audio_stream is not None:
self.audio_stream.close()
finally:
pya.terminate()
def _list_devices() -> None:
print("\n=== オーディオデバイス一覧 (index / name) ===")
host_count = pya.get_host_api_count()
for i in range(pya.get_device_count()):
info = pya.get_device_info_by_index(i)
name = info.get("name")
max_in = int(info.get("maxInputChannels", 0))
max_out = int(info.get("maxOutputChannels", 0))
default_sr = int(info.get("defaultSampleRate", 0))
print(f"[{i:02d}] {name} (in:{max_in} / out:{max_out} / {default_sr}Hz)")
print("")
def main(argv: list[str] | None = None) -> int:
import argparse
parser = argparse.ArgumentParser(description="Gemini Live API (Native Audio) シンプル実行スクリプト")
parser.add_argument("--model", default=MODEL, help="使用するモデル名")
parser.add_argument("--no-proactivity", action="store_true", help="プロアクティブ音声を無効化")
parser.add_argument("--list-devices", action="store_true", help="利用可能なオーディオデバイス一覧を表示して終了")
parser.add_argument("--in-dev", type=int, help="入力デバイス index（未指定ならデフォルト）")
parser.add_argument("--out-dev", type=int, help="出力デバイス index（未指定ならデフォルト）")
# 半カスケード（TTS）で有効な音声選択オプション
parser.add_argument("--tts-voice", help="TTS の voice_name（半カスケードモデルのみ有効） 例: ja-JP-Standard-A")
parser.add_argument("--tts-rate", type=float, help="TTS の speaking_rate（例: 0.9～1.1）")
parser.add_argument("--tts-pitch", type=float, help="TTS の pitch（セミ単位。例: -2.0 ～ +2.0）")
args = parser.parse_args(argv)
if args.list_devices:
_list_devices()
return 0
cfg = dict(BASE_CONFIG)
if args.no_proactivity:
cfg.pop("proactivity", None)
# 半カスケード（例: gemini-live-2.5-flash-preview, gemini-2.0-flash-live-001）で TTS 音声を指定
model_lower = (args.model or "").lower()
is_half_cascade = ("live-" in model_lower) or model_lower.endswith("-live-001")
if is_half_cascade and (args.tts_voice or args.tts_rate is not None or args.tts_pitch is not None):
speech_cfg = {}
if args.tts_voice:
speech_cfg["voice_name"] = args.tts_voice
if args.tts_rate is not None:
speech_cfg["speaking_rate"] = args.tts_rate
if args.tts_pitch is not None:
speech_cfg["pitch"] = args.tts_pitch
if speech_cfg:
cfg["speech_config"] = speech_cfg
elif not is_half_cascade and (args.tts_voice or args.tts_rate is not None or args.tts_pitch is not None):
print(
"[注意] 指定した --tts-* はネイティブ音声モデルでは無視される場合があります。\n"
" 声の種類を選びたい場合は半カスケードモデル（例: gemini-live-2.5-flash-preview）を --model で指定してください。",
file=sys.stderr,
)
loop = AudioLoop()
asyncio.run(loop.run(args.model, cfg, in_dev=args.in_dev, out_dev=args.out_dev))
return 0
if __name__ == "__main__":
raise SystemExit(main())
Python3.12環境推奨
brew install portaudio
python3 -m pip install --upgrade pip setuptools wheel
-
3.12で新規作成: /opt/homebrew/bin/python3.12 -m venv .venv
-
有効化: source .venv/bin/activate
-
確認: python -V（3.12.xと表示）
-
ツール更新: python -m pip install -U pip setuptools wheel
-
依存導入: python -m pip install -U google-genai pyaudio
-
Gemini APIキーが必要です
-
確認: python -c "from google import genai; import pyaudio; print('OK')"
-
実行: python Get_started_LiveAPI_NativeAudio.py --list-devices
python Get_started_LiveAPI_NativeAudio.py --list-devices
=== オーディオデバイス一覧 (index / name) ===
[00] AirPods Pro A3047 (in:1 / out:0 / 24000Hz)
[01] AirPods Pro A3047 (in:0 / out:2 / 48000Hz)
[02] MacBook Airのマイク (in:1 / out:0 / 48000Hz)
[03] MacBook Airのスピーカー (in:0 / out:2 / 48000Hz)
[04] iP16PMのマイク (in:1 / out:0 / 48000Hz)
python Get_started_LiveAPI_NativeAudio.py --in-dev 2 --out-dev 1
効果が足りない場合の追加チューニング
-
プレバッファを増やす: （遅延は増えるが安定）PREBUFFER_MS を 200〜300 に上げる
-
フレーム長を大きくする: SEND_CHUNK_FRAMES = 640（40ms）、RECV_CHUNK_FRAMES = 960 など
-
プロアクティブ音声を無効化（会話の割り込み挙動を抑える）--no-proactivity
