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

ライブセッションを開始して管理する

Live API は、セッションと呼ばれる音声またはテキストの継続的なストリームを処理して、人間のような音声レスポンスを即座に提供することで、低レイテンシの音声とテキストのやり取りを実現します。初期ハンドシェイクから正常な終了までのセッションライフサイクル管理は、デベロッパーによって制御されます。

このページでは、Live API を使用して Gemini モデルとの会話セッションを開始する方法について説明します。セッションは、Vertex AI Studio、Gen AI SDK、または WebSocket を使用して開始できます。

このページでは、次の方法についても説明します。

セッションをデフォルトの制限時間を超えて延長する
以前のセッションを再開する
セッション中にシステム指示を更新する
セッションのコンテキストウィンドウを構成する
セッションの文字起こしを有効にする

セッションの有効期間

圧縮なしの場合、音声のみのセッションは 15 分、音声と動画のセッションは 2 分に制限されます。これらの上限を超えるとセッションが終了しますが、コンテキストウィンドウの圧縮を使用すると、セッションを無制限に延長できます。

接続の有効期間は約 10 分に制限されています。接続が終了すると、セッションも終了します。この場合、セッションの再開を使用して、複数の接続でアクティブな状態を維持するように単一のセッションを構成できます。接続が終了する前に GoAway メッセージも届くため、さらに対処できます。

最大同時セッション数

従量課金制（PayGo）プランでは、プロジェクトごとに最大 1,000 の同時セッションを設定できます。この上限は、プロビジョンドスループットを使用しているお客様には適用されません。

セッションを開始する

次のタブでは、Vertex AI Studio、Gen AI SDK、または WebSocket を使用してライブ会話セッションを開始する方法について説明します。

コンソール

[Vertex AI Studio] > [リアルタイムストリーミング] を開きます。
[ セッションを開始] をクリックして会話を開始します。

セッションを終了するには、[ セッションを停止] をクリックします。

Python

始める前に、API キーまたはアプリケーションのデフォルト認証情報（ADC）を使用して Vertex AI に対する認証を行う必要があります。

gcloud auth application-default login

認証の設定の詳細については、クイックスタートをご覧ください。

import asyncio
from google import genai

# Replace the PROJECT_ID and LOCATION with your Project ID and location. 
client = genai.Client(vertexai=True, project="PROJECT_ID", location="LOCATION")

# Configuration
MODEL = "gemini-live-2.5-flash-preview-native-audio-09-2025"
config = {
   "response_modalities": ["audio"],
}

async def main():
   # Establish WebSocket session
   async with client.aio.live.connect(model=MODEL, config=config) as session:
       print("Session established. Sending audio...")

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

Python

WebSocket を使用する場合、接続は標準の WebSocket ハンドシェイクで確立されます。このエンドポイントはリージョンであり、認証に OAuth 2.0 ベアラートークンを使用します。このシナリオでは、通常、認証トークンは WebSocket ヘッダー（Authorization: Bearer [TOKEN] など）で渡されます。

import asyncio
import websockets

# Replace the PROJECT_ID and LOCATION with your Project ID and location. 
PROJECT_ID = "PROJECT_ID"
LOCATION = "LOCATION"

# Authentication
token_list = !gcloud auth application-default print-access-token
ACCESS_TOKEN = token_list[0]

# Configuration
MODEL_ID = "gemini-live-2.5-flash-preview-native-audio-09-2025"
MODEL = f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/{MODEL_ID}"
config = {
   "response_modalities": ["audio"],
}

# Construct the WSS URL
HOST = f"{LOCATION}-aiplatform.googleapis.com"
URI = f"wss://{HOST}/ws/google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"

async def main():
   headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
  
   async with websockets.connect(URI, additional_headers=headers) as ws:
       print("Session established.")

       # Send Setup (Handshake)
       await ws.send(json.dumps({
           "setup": {
               "model": MODEL,
               "generation_config": config
           }
       }))
    # Send audio/video ...

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

セッションを延長する

会話セッションのデフォルトの最大長は 10 分です。セッションが終了する 60 秒前に、goAway 通知（BidiGenerateContentServerMessage.goAway）がクライアントに送信されます。

Gen AI SDK を使用すると、セッションの長さを 10 分単位で延長できます。セッションを延長できる回数に制限はありません。例については、前のセッションを再開するをご覧ください。

セッションを延長するには:

Python

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)

以前のセッションを再開する

Live API はセッションの再開をサポートしているため、ユーザーは短時間の切断（Wi-Fi から 5G への切り替えなど）中に会話のコンテキストを失うことはありません。24 時間以内であれば、前のセッションを再開できます。セッション再開は、テキスト、動画、音声のプロンプト、モデル出力などのキャッシュに保存されたデータを保存することで実現されます。このキャッシュに保存されたデータには、プロジェクトレベルのプライバシーが適用されます。

デフォルトでは、セッションの再開は無効になっています。セッション再開を有効にするには、BidiGenerateContentSetup メッセージの sessionResumption フィールドを設定します。有効にすると、サーバーは session_id と再開トークンを含む SessionResumptionUpdate メッセージを定期的に送信します。WebSocket が切断された場合、クライアントは再接続して、新しい設定メッセージにこれらの認証情報を含めることができます。サーバーは前のコンテキストを復元し、会話をシームレスに続行できるようにします。

再開ウィンドウは有限です（通常は約 10 分）。この期間内にクライアントが再接続しない場合、セッション状態は破棄され、サーバーリソースが解放されます。

セッションの再開を有効にしてハンドル ID を取得する例を次に示します。

Python

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


# Replace the PROJECT_ID and LOCATION with your Project ID and location. 
client = genai.Client(vertexai=True, project="PROJECT_ID", location="LOCATION")

# Configuration
MODEL = "gemini-live-2.5-flash-preview-native-audio-09-2025"

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

透過モードでシームレスなセッションの再開を有効にする

セッションの再開を有効にすると、透過モードも有効にして、ユーザーにとって再開プロセスをよりシームレスにすることができます。透過モードが有効になると、コンテキストスナップショットに対応するクライアントメッセージのインデックスが明示的に返されます。これにより、再開ハンドルからセッションを再開するときに、どのクライアントメッセージを再送信する必要があるかを特定できます。

透過モードを有効にするには:

Python

config = {
   "response_modalities": ["audio"],
   "session_resumption": {
     "session_resumption_config": {
        "transparent": True,
     }
   }
}

セッション中にシステム指示を更新する

Live API を使用すると、アクティブなセッション中にシステム指示を更新できます。これを使用して、レスポンスの言語の変更やトーンの変更など、モデルのレスポンスを調整します。

セッション中にシステム指示を更新するには、system ロールでテキストコンテンツを送信します。更新されたシステム指示は、残りのセッションで有効になります。

Python

session.send_client_content(
      turns=types.Content(
          role="system", parts=[types.Part(text="new system instruction")]
      ),
      turn_complete=False
  )

セッションのコンテキストウィンドウを構成する

Live API コンテキストウィンドウは、リアルタイムでストリーミングされたデータ（音声の場合は 1 秒あたり 25 トークン（TPS）、動画の場合は 258 TPS）や、テキスト入力やモデル出力などの他のコンテンツを保存するために使用されます。セッションのコンテキストウィンドウの上限は次のとおりです。

ネイティブオーディオモデルの 128,000 トークン
他の Live API モデルの場合は 32,000 トークン

長時間実行されるセッションでは、会話が進むにつれて、音声トークンとテキストトークンの履歴が蓄積されます。この履歴がモデルの上限を超えると、モデルがハルシネーションを起こしたり、処理が遅くなったり、セッションが強制終了されたりする可能性があります。セッションを長くするには、セッション構成の一部として contextWindowCompression フィールドを設定して、コンテキストウィンドウ圧縮を有効にします。

コンテキストウィンドウの圧縮では、サーバーサイドのスライディングウィンドウを使用して、有効にすると最も古いターンを切り捨てます。累積トークンが定義された最大長（Vertex AI Studio の [最大コンテンツサイズ] スライダーまたは API の trigger_tokens を使用して設定）を超えると、サーバーは最も古いターンを自動的にプルーニングするか、要約して、コンテキストを上限内に維持します。ContextWindowCompressionConfig では、スライディングウィンドウメカニズムと、圧縮をトリガーする target_tokens パラメータで定義されたトークンの数を構成できます。

これにより、ユーザーの視点から見ると、理論上はセッションの継続時間を無限にすることができます。これは、「メモリ」が常に管理されているためです。圧縮なしの場合、音声のみのセッションはハードリミットに達するまで約 15 分に制限されることがあります。

コンテキストの長さとターゲットサイズの最小長と最大長は次のとおりです。

設定（API フラグ）	最小値	最大値
コンテキストの最大長（`trigger_tokens`）	5,000	128,000
ターゲットコンテキストのサイズ（`target_tokens`）	0	128,000

コンテキストウィンドウを設定するには:

コンソール

[Vertex AI Studio] > [リアルタイムストリーミング] を開きます。
クリックして [詳細] メニューを開きます。
[セッションのコンテキスト] セクションで、[コンテキストの最大サイズ] スライダーを使用して、コンテキストサイズを 5,000～128,000 の値に設定します。
（省略可）同じセクションで、[ターゲットコンテキストのサイズ] スライダーを使用して、ターゲットサイズを 0～128,000 の範囲の値に設定します。

Python

セットアップメッセージで context_window_compression.trigger_tokens フィールドと context_window_compression.sliding_window.target_tokens フィールドを設定します。

config = {
   "response_modalities": ["audio"],
   # Configures compression
   "context_window_compression" : {
    "trigger_tokens": 10000,
    "sliding_window": {"target_tokens" : 512}
   }
}

セッションの音声文字変換を有効にする

入力音声と出力音声の両方で文字起こしを有効にできます。

文字起こしを受信するには、セッション構成を更新する必要があります。input_audio_transcription オブジェクトと output_audio_transcription オブジェクトを追加し、text が response_modalities に含まれていることを確認する必要があります。

config = {
    "response_modalities": ["audio", "text"],
    "input_audio_transcription": {},
    "output_audio_transcription": {},
}

レスポンスの処理

次のコードサンプルは、構成されたセッションを使用して接続し、音声データとともにテキスト部分（文字起こし）を抽出する方法を示しています。

# Receive Output Loop
async for message in session.receive():
    server_content = message.server_content
    if server_content:
        # Handle Model Turns (Audio + Text)
        model_turn = server_content.model_turn
        if model_turn and model_turn.parts:
            for part in model_turn.parts:
                # Handle Text (Transcriptions)
                if part.text:
                    print(f"Transcription: {part.text}")
                # Handle Audio
                if part.inline_data:
                    audio_data = part.inline_data.data
                    # Process audio bytes...
                    pass

        # Check for turn completion
        if server_content.turn_complete:
            print("Turn complete.")

ライブ セッションを開始して管理する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

セッションの有効期間

最大同時セッション数

セッションを開始する

コンソール

Python

Python

セッションを延長する

Python

以前のセッションを再開する

Python

透過モードでシームレスなセッションの再開を有効にする

Python

セッション中にシステム指示を更新する

Python

セッションのコンテキスト ウィンドウを構成する

コンソール

Python

セッションの音声文字変換を有効にする

レスポンスの処理

次のステップ

ライブセッションを開始して管理する

セッションのコンテキストウィンドウを構成する