top of page

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

更新日:9月27日


ree

より強力な音声エージェントを構築する ― Gemini Live API 最新アップデート

Google DeepMind プロダクトマネージャー Ivan Solovyev、Valeria Wu、エンジニア Mingqiu Wang が、Gemini API の Live API における大幅アップデートを発表しました。新しいネイティブ音声モデルがプレビュー版として利用可能になり、より信頼性が高く、応答性に優れ、自然な会話体験を持つ音声エージェントを構築できるようになります。


今回のリリースでは、主に以下の2点に焦点を当てています。


  • より堅牢なファンクションコール:外部データやサービスへの接続を確実に行えるようにすること。

  • より自然な会話体験:中断や一時停止、サイド会話(雑談)に対して直感的に対応できるようにすること。


信頼性の大幅な向上

音声エージェントが最も強力で魅力的な体験を提供するのは、外部のデータやサービスと確実に接続できるときです。ユーザーはリアルタイムで情報を取得したり、予定を予約したり、取引を完了することができます。このとき重要になるのが ファンクションコールの正確性 です。音声インタラクションはリアルタイムで行われるため、失敗したリクエストをやり直す余裕はありません。


新しいモデルでは以下が大幅に改善されました:


  • 呼び出すべき正しい関数を認識する精度

  • 不要な関数を呼び出さない判断

  • 提供されたツールスキーマの一貫した遵守


内部ベンチマークによると、複雑なシナリオ(10以上の関数が同時稼働するケース)でも、正確な関数コール率が大幅に改善。前バージョンと比較して、単一コールでの成功率は 2倍、5〜10件のコールを含むテストでは 1.5倍 に増加しました。


これは音声アプリケーションにとって大きな前進であり、今後も開発者からのフィードバックを基に、特にマルチターンシナリオでの信頼性をさらに向上させる予定です。


Google AI Studio上で、このファンクションコール精度の改善を試すことができます。


より自然な会話へ

今回の今回のアップデートでは、音声対話がより人間らしくなるように プロアクティブな音声機能 が追加されました。


新しいモデルは以下を実現します:


  • 会話に関係のない雑談を無視する

  • ユーザーの自然な沈黙や中断を理解し、スムーズに再開する


たとえば、音声エージェントと会話中に誰かが部屋に入ってきて質問をしてきた場合、モデルは会話を一時停止し、その雑談を無視。ユーザーが戻ってきたときに、自然に会話を再開できます。


また、話すリズムや文脈をよりよく理解するため、ユーザーが複雑な内容を考えながら発話する沈黙や、カジュアルな会話の間でも適切に対応します。結果として、誤って会話を遮るケースが大幅に減少しました。


さらに、中断の検出精度も改善され、ユーザーが割り込んだ場合にそれを見逃す頻度も著しく減っています。


「思考」機能によるスマートな応答

今回のリリースに続いて、来週からは Gemini 2.5 FlashやProで導入された「Thinking」機能 がLive APIでも利用可能になります。


すべての質問が即答に適しているわけではありません。複雑な問いには、数秒の「思考時間」を設定してより深い推論を行えるようになります。このプロセスでは、モデルが思考の要約をテキストで返す仕組みも導入されます。


実際の活用例:家庭の「COO」になるAva

初期のアクセスパートナーとの協力を通じて、最新のAPI機能をテストしてきました。その多くが肯定的な成果を報告しています。


たとえば、AI搭載のファミリーOS Ava は、Live APIを利用して「家庭のCOO」として機能。学校からのメールやPDF、音声メモといった複雑な入力を処理し、予定表やタスクに変換します。


共同創業者・CTOの Joe Alicata 氏はこう語ります:


「自然な双方向の音声チャットは必須でした。新モデルのファンクションコール精度の改善は決定的な転換点でした。雑音の多い入力に対しても高い一次精度を実現し、脆弱なプロンプト回避策に頼らずに済むようになったことで、小規模なチームでも信頼性の高いマルチモーダル製品を迅速に開発できました。」


AICU AIDX Labによる実験

Google AI Studio からすぐに Live API を利用できます。Cookbookにはエンドツーエンドのコードサンプルも用意されています。



以下は公式が提供したPythonコードをAICU AIDX LabがMacのPython環境用に日本語で解説、改善したものです。


ソースコード: 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

#

#

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

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(

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キーが必要です

    • export GOOGLE_API_KEY=…

  • 確認: 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



Originally published at note.com/aicu on Sep 23, 2025.

コメント


bottom of page