Gemini 2.5 Flash、Veo 2、Live API のアップデートのご紹介詳細

このページは Cloud Translation API によって翻訳されました。

Live API

Live API を使用すると、Gemini との双方向の音声と動画による低レイテンシのやり取りが可能になります。Live API を使用すると、エンドユーザーに自然で人間のような音声会話のエクスペリエンスを提供できます。また、音声コマンドを使用してモデルのレスポンスを中断することもできます。このモデルは、テキスト、音声、および動画入力を処理でき、テキストおよび音声出力を生成できます。

Live API は Google AI Studio で試すことができます。

Live API を使用する

このセクションでは、Google の SDK のいずれかを使用して Live API を使用する方法について説明します。基盤となる WebSockets API の詳細については、WebSockets API リファレンスをご覧ください。

すべての機能を利用するには、最新バージョンの SDK をインストールしてください。pip install -U google-genai。

テキストの送受信

import asyncio
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["TEXT"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        while True:
            message = input("User> ")
            if message.lower() == "exit":
                break
            await session.send_client_content(
                turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
            )

            async for response in session.receive():
                if response.text is not None:
                    print(response.text, end="")

if __name__ == "__main__":
    asyncio.run(main())

音声を受信する

次の例は、音声データを受信して .wav ファイルに書き込む方法を示しています。

import asyncio
import wave
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["AUDIO"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        wf = wave.open("audio.wav", "wb")
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(24000)

        message = "Hello? Gemini are you there?"
        await session.send_client_content(
            turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
        )

        async for idx,response in async_enumerate(session.receive()):
            if response.data is not None:
                wf.writeframes(response.data)

            # Un-comment this code to print audio data info
            # if response.server_content.model_turn is not None:
            #      print(response.server_content.model_turn.parts[0].inline_data.mime_type)

        wf.close()

if __name__ == "__main__":
    asyncio.run(main())

オーディオ形式

Live API は、次の音声形式をサポートしています。

入力音声形式: RAW 16 ビット PCM 音声、16kHz、リトルエンディアン
出力音声形式: RAW 16 ビット PCM 音声、24kHz、リトルエンディアン

音声と動画をストリーミングする

システム指示

システム指示を使用すると、特定のニーズやユースケースに基づいてモデルの動作を制御できます。システム指示は設定構成で設定でき、セッション全体で有効になります。

from google.genai import types

config = {
    "system_instruction": types.Content(
        parts=[
            types.Part(
                text="You are a helpful assistant and answer in a friendly tone."
            )
        ]
    ),
    "response_modalities": ["TEXT"],
}

コンテンツの増分更新

増分更新を使用して、テキスト入力の送信、セッションコンテキストの確立、セッションコンテキストの復元を行います。コンテキストが短い場合は、ターンバイターンのインタラクションを送信して、イベントの正確なシーケンスを表すことができます。

Python

turns = [
    {"role": "user", "parts": [{"text": "What is the capital of France?"}]},
    {"role": "model", "parts": [{"text": "Paris"}]},
]

await session.send_client_content(turns=turns, turn_complete=False)

turns = [{"role": "user", "parts": [{"text": "What is the capital of Germany?"}]}]

await session.send_client_content(turns=turns, turn_complete=True)

JSON

{
  "clientContent": {
    "turns": [
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"user"
      },
      {
        "parts":[
          {
            "text": ""
          }
        ],
        "role":"model"
      }
    ],
    "turnComplete": true
  }
}

コンテキストが長い場合は、1 つのメッセージの概要を提供して、後続のインタラクション用にコンテキストウィンドウを空けておくことをおすすめします。

音声を変更する

Live API は、Puck、Charon、Kore、Fenrir、Aoede、Leda、Orus、Zephyr の音声をサポートしています。

音声を指定するには、セッション構成の一部として、speechConfig オブジェクト内に音声名を設定します。

Python

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore")
        )
    )
)

JSON

{
  "voiceConfig": {
    "prebuiltVoiceConfig": {
      "voiceName": "Kore"
    }
  }
}

言語を変更

Live API は複数の言語をサポートしています。

言語を変更するには、セッション構成の一部として speechConfig オブジェクト内に言語コードを設定します。

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        language_code="de-DE",
    )
)

ツールを使用する

Live API を使用すると、関数呼び出し、コード実行、Google 検索などのツールを定義できます。

関数呼び出しを使用する

関数宣言は、セッション構成の一部として定義できます。詳細については、関数呼び出しチュートリアルをご覧ください。

ツール呼び出しを受け取ったクライアントは、session.send_tool_response メソッドを使用して FunctionResponse オブジェクトのリストを返す必要があります。

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

# Simple function definitions
turn_on_the_lights = {"name": "turn_on_the_lights"}
turn_off_the_lights = {"name": "turn_off_the_lights"}

tools = [{"function_declarations": [turn_on_the_lights, turn_off_the_lights]}]
config = {"response_modalities": ["TEXT"], "tools": tools}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        prompt = "Turn on the lights please"
        await session.send_client_content(turns={"parts": [{"text": prompt}]})

        async for chunk in session.receive():
            if chunk.server_content:
                if chunk.text is not None:
                    print(chunk.text)
            elif chunk.tool_call:
                function_responses = []
                for fc in tool_call.function_calls:
                    function_response = types.FunctionResponse(
                        id=fc.id,
                        name=fc.name,
                        response={ "result": "ok" } # simple, hard-coded function response
                    )
                    function_responses.append(function_response)

                await session.send_tool_response(function_responses=function_responses)


if __name__ == "__main__":
    asyncio.run(main())

モデルは、単一のプロンプトから複数の関数呼び出しと、出力の連結に必要なコードを生成できます。このコードはサンドボックス環境で実行され、後続の BidiGenerateContentToolCall メッセージを生成します。各関数呼び出しの結果が表示されるまで実行は停止するため、順番どおりに処理が行われます。

音声入力と音声出力は、モデルが関数呼び出しを使用する機能に悪影響を及ぼします。

コード実行を使用する

コード実行は、セッション構成の一部として定義できます。詳細については、コード実行チュートリアルをご覧ください。

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

tools = [{'code_execution': {}}]
config = {"response_modalities": ["TEXT"], "tools": tools}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        prompt = "Compute the largest prime palindrome under 100000."
        await session.send_client_content(turns={"parts": [{"text": prompt}]})

        async for chunk in session.receive():
            if chunk.server_content:
                if chunk.text is not None:
                    print(chunk.text)
            
                model_turn = chunk.server_content.model_turn
                if model_turn:
                    for part in model_turn.parts:
                      if part.executable_code is not None:
                        print(part.executable_code.code)

                      if part.code_execution_result is not None:
                        print(part.code_execution_result.output)

if __name__ == "__main__":
    asyncio.run(main())

Google 検索によるグラウンディングを使用する

グラウンディングと Google 検索は、セッション構成の一部として有効にできます。詳しくは、グラウンディングチュートリアルをご覧ください。

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

tools = [{'google_search': {}}]
config = {"response_modalities": ["TEXT"], "tools": tools}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        prompt = "When did the last Brazil vs. Argentina soccer match happen?"
        await session.send_client_content(turns={"parts": [{"text": prompt}]})

        async for chunk in session.receive():
            if chunk.server_content:
                if chunk.text is not None:
                    print(chunk.text)

                # The model might generate and execute Python code to use Search
                model_turn = chunk.server_content.model_turn
                if model_turn:
                    for part in model_turn.parts:
                      if part.executable_code is not None:
                        print(part.executable_code.code)

                      if part.code_execution_result is not None:
                        print(part.code_execution_result.output)

if __name__ == "__main__":
    asyncio.run(main())

複数のツールを組み合わせる

Live API 内で複数のツールを組み合わせることができます。

prompt = """
Hey, I need you to do three things for me.

1. Compute the largest prime palindrome under 100000.
2. Then use Google Search to look up information about the largest earthquake in California the week of Dec 5 2024?
3. Turn on the lights

Thanks!
"""

tools = [
    {"google_search": {}},
    {"code_execution": {}},
    {"function_declarations": [turn_on_the_lights, turn_off_the_lights]},
]

config = {"response_modalities": ["TEXT"], "tools": tools}

中断を処理する

モデルの出力はいつでも中断できます。音声アクティビティ検出（VAD）が中断を検出すると、進行中の生成はキャンセルされ、破棄されます。クライアントにすでに送信された情報だけが、セッション履歴に保持されます。その後、サーバーは中断を報告する BidiGenerateContentServerContent メッセージを送信します。

さらに、Gemini サーバーは保留中の関数呼び出しを破棄し、キャンセルされた呼び出しの ID を記載した BidiGenerateContentServerContent メッセージを送信します。

async for response in session.receive():
    if response.server_content.interrupted is True:
        # The generation was interrupted

音声アクティビティ検出（VAD）を構成する

音声アクティビティ検出（VAD）を構成または無効にできます。

自動 VAD を使用する

デフォルトでは、モデルは連続した音声入力ストリームに対して VAD を自動的に実行します。VAD は、設定構成の realtimeInputConfig.automaticActivityDetection フィールドで構成できます。

音声ストリームが 1 秒以上一時停止した場合（ユーザーがマイクをオフにした場合など）、キャッシュに保存されている音声をフラッシュするために audioStreamEnd イベントを送信する必要があります。クライアントはいつでも音声データの送信を再開できます。

# example audio file to try:
# URL = "https://storage.googleapis.com/generativeai-downloads/data/hello_are_you_there.pcm"
# !wget -q $URL -O sample.pcm
import asyncio
from pathlib import Path
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["TEXT"]}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        audio_bytes = Path("sample.pcm").read_bytes()

        await session.send_realtime_input(
            audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
        )

        # if stream gets paused, send:
        # await session.send_realtime_input(audio_stream_end=True)

        async for response in session.receive():
            if response.text is not None:
                print(response.text)

if __name__ == "__main__":
    asyncio.run(main())

send_realtime_input を使用すると、API は VAD に基づいて音声に自動的に応答します。send_client_content は順序に従ってモデルコンテキストにメッセージを追加しますが、send_realtime_input は確定的な順序付けを犠牲にして応答性を最適化します。

自動 VAD を構成する

VAD アクティビティをより細かく制御するには、次のパラメータを構成します。詳細については、API リファレンスをご覧ください。

from google.genai import types

config = {
    "response_modalities": ["TEXT"],
    "realtime_input_config": {
        "automatic_activity_detection": {
            "disabled": False, # default
            "start_of_speech_sensitivity": types.StartSensitivity.START_SENSITIVITY_LOW,
            "end_of_speech_sensitivity": types.EndSensitivity.END_SENSITIVITY_LOW,
            "prefix_padding_ms": 20,
            "silence_duration_ms": 100,
        }
    }
}

自動 VAD を無効にする

または、設定メッセージで realtimeInputConfig.automaticActivityDetection.disabled を true に設定して、自動 VAD を無効にすることもできます。この構成では、クライアントがユーザーの音声を検出し、適切なタイミングで activityStart メッセージと activityEnd メッセージを送信します。この構成では audioStreamEnd は送信されません。代わりに、ストリームの中断は activityEnd メッセージでマークされます。

config = {
    "response_modalities": ["TEXT"],
    "realtime_input_config": {"automatic_activity_detection": {"disabled": True}},
}

async with client.aio.live.connect(model=model, config=config) as session:
    # ...
    await session.send_realtime_input(activity_start=types.ActivityStart())
    await session.send_realtime_input(
        audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000")
    )
    await session.send_realtime_input(activity_end=types.ActivityEnd())
    # ...

トークン数を取得する

消費されたトークンの合計数は、返されたサーバーメッセージの usageMetadata フィールドで確認できます。

async for message in session.receive():
    # The server will periodically send messages that include UsageMetadata.
    if message.usage_metadata:
        usage = message.usage_metadata
        print(
            f"Used {usage.total_token_count} tokens in total. Response token breakdown:"
        )
        for detail in usage.response_tokens_details:
            match detail:
                case types.ModalityTokenCount(modality=modality, token_count=count):
                    print(f"{modality}: {count}")

セッション継続時間を延長する

最大セッション時間は、次の 2 つのメカニズムを使用して無制限に延長できます。

コンテキストウィンドウ圧縮を有効にする
セッションの再開を構成する

また、セッションが終了する前に GoAway メッセージが届き、追加の操作を行えます。

コンテキストウィンドウの圧縮を有効にする

セッションを長くして、接続が突然終了しないようにするには、セッション構成の一部として contextWindowCompression フィールドを設定して、コンテキストウィンドウ圧縮を有効にします。

ContextWindowCompressionConfig で、スライディングウィンドウメカニズムと、圧縮をトリガーするトークン数を構成できます。

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    context_window_compression=(
        # Configures compression with default parameters.
        types.ContextWindowCompressionConfig(
            sliding_window=types.SlidingWindow(),
        )
    ),
)

セッションの再開を構成する

サーバーが WebSocket 接続を定期的にリセットするときにセッションが終了しないようにするには、設定構成で sessionResumption フィールドを構成します。

この構成を渡すと、サーバーは SessionResumptionUpdate メッセージを送信します。このメッセージは、最後の再開トークンを後続の接続の SessionResumptionConfig.handle として渡すことで、セッションを再開するために使用できます。

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

async def main():
    print(f"Connecting to the service with handle {previous_session_handle}...")
    async with client.aio.live.connect(
        model=model,
        config=types.LiveConnectConfig(
            response_modalities=["AUDIO"],
            session_resumption=types.SessionResumptionConfig(
                # The handle of the session to resume is passed here,
                # or else None to start a new session.
                handle=previous_session_handle
            ),
        ),
    ) as session:
        while True:
            await session.send_client_content(
                turns=types.Content(
                    role="user", parts=[types.Part(text="Hello world!")]
                )
            )
            async for message in session.receive():
                # Periodically, the server will send update messages that may
                # contain a handle for the current state of the session.
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        # The handle should be retained and linked to the session.
                        return update.new_handle

                # For the purposes of this example, placeholder input is continually fed
                # to the model. In non-sample code, the model inputs would come from
                # the user.
                if message.server_content and message.server_content.turn_complete:
                    break

if __name__ == "__main__":
    asyncio.run(main())

セッションが切断される前にメッセージを受信する

サーバーは、現在の接続がまもなく終了することを通知する GoAway メッセージを送信します。このメッセージには、残り時間を示す timeLeft が含まれており、接続が ABORTED として終了する前に追加のアクションを実行できます。

async for response in session.receive():
    if response.go_away is not None:
        # The connection will soon be terminated
        print(response.go_away.time_left)

生成が完了したときにメッセージを受け取る

サーバーは、モデルがレスポンスを生成し終えたことを通知する generationComplete メッセージを送信します。

async for response in session.receive():
    if response.server_content.generation_complete is True:
        # The generation is complete

メディアの解像度を変更する

入力メディアのメディア解像度を指定するには、セッション構成の一部として mediaResolution フィールドを設定します。

from google.genai import types

config = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    media_resolution=types.MediaResolution.MEDIA_RESOLUTION_LOW,
)

音声文字変換を受信する

モデルのオーディオ出力の音声文字変換を有効にできます。音声文字変換の言語は、モデルのレスポンスから推測されます。

import asyncio
from google import genai
from google.genai import types

client = genai.Client(api_key="GEMINI_API_KEY")
model = "gemini-2.0-flash-live-001"

config = {"response_modalities": ["AUDIO"],
          "output_audio_transcription": {}
}

async def main():
    async with client.aio.live.connect(model=model, config=config) as session:
        message = "Hello? Gemini are you there?"

        await session.send_client_content(
            turns={"role": "user", "parts": [{"text": message}]}, turn_complete=True
        )

        async for response in session.receive():
            if response.server_content.model_turn:
                print("Model turn:", response.server_content.model_turn)
            if response.server_content.output_transcription:
                print("Transcript:", response.server_content.output_transcription.text)


if __name__ == "__main__":
    asyncio.run(main())

制限事項

プロジェクトを計画する際は、Live API と Gemini 2.0 の次の制限事項を考慮してください。

レスポンスモダリティ

セッション構成で設定できるレスポンスモダリティ（TEXT または AUDIO）は、セッションごとに 1 つだけです。両方を設定しようとすると、構成エラーメッセージが表示されます。つまり、テキストまたは音声のいずれかで応答するようにモデルを構成できますが、同じセッションで両方を使用することはできません。

クライアント認証

Live API が提供するのはサーバー間認証のみであるため、クライアントが直接使用することはおすすめしません。クライアントの入力を Live API で安全に認証するには、中間アプリケーションサーバーを経由する必要があります。

セッション継続時間

セッションの圧縮を有効にすると、セッションの継続時間を無制限に延長できます。圧縮なしの場合、音声のみのセッションは 15 分に制限され、音声と動画のセッションは 2 分に制限されます。圧縮なしでこれらの上限を超えると、接続が終了します。

また、セッションの再開を構成して、クライアントが終了したセッションを再開できるようにすることもできます。

コンテキストウィンドウ

セッションのコンテキストウィンドウの上限は 32,000 トークンです。

サポートされている言語

Live API は次の言語をサポートしています。

言語	BCP-47 コード
ドイツ語（ドイツ）	de-DE
英語（オーストラリア）	en-AU
英語（英国）	en-GB
英語（インド）	en-IN
英語（米国）	en-US
スペイン語（米国）	es-US
フランス語（フランス）	fr-FR
ヒンディー語（インド）	hi-IN
ポルトガル語（ブラジル）	pt-BR
アラビア語（一般）	ar-XA
スペイン語（スペイン）	es-ES
フランス語（カナダ）	fr-CA
インドネシア語（インドネシア）	id-ID
イタリア語（イタリア）	it-IT
日本語（日本）	ja-JP
トルコ語（トルコ）	tr-TR
ベトナム語（ベトナム）	vi-VN
ベンガル語（インド）	bn-IN
グジャラト語（インド）	gu-IN
カンナダ語（インド）	kn-IN
マラヤーラム語（インド）	ml-IN
マラーティー語（インド）	mr-IN
タミル語（インド）	ta-IN
テルグ語（インド）	te-IN
オランダ語（オランダ）	nl-NL
韓国語（韓国）	ko-KR
北京語（中国）	cmn-CN
ポーランド語（ポーランド）	pl-PL
ロシア語（ロシア）	ru-RU
タイ語（タイ）	th-TH

サードパーティとの連携

ウェブアプリとモバイルアプリのデプロイでは、次のオプションを検討できます。

Live API

最新情報

Live API を使用する

テキストの送受信

音声を受信する

オーディオ形式

音声と動画をストリーミングする

システム指示

コンテンツの増分更新

Python

JSON

音声を変更する

Python

JSON

言語を変更

ツールを使用する

関数呼び出しを使用する

コード実行を使用する

Google 検索によるグラウンディングを使用する

複数のツールを組み合わせる

中断を処理する

音声アクティビティ検出（VAD）を構成する

自動 VAD を使用する

自動 VAD を構成する

自動 VAD を無効にする

トークン数を取得する

セッション継続時間を延長する

コンテキスト ウィンドウの圧縮を有効にする

セッションの再開を構成する

セッションが切断される前にメッセージを受信する

生成が完了したときにメッセージを受け取る

メディアの解像度を変更する

音声文字変換を受信する

制限事項

レスポンス モダリティ

クライアント認証

セッション継続時間

コンテキスト ウィンドウ

サポートされている言語

サードパーティとの連携

コンテキストウィンドウの圧縮を有効にする

レスポンスモダリティ

コンテキストウィンドウ