Gemini Live APIがアップデート、強力な会話AIを簡単に構築できる!

# -*- 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())