開始及管理直播工作階段

Gemini Live API 會處理連續的音訊或文字串流 (稱為「工作階段」)，並立即提供類似人類的口語回覆，實現低延遲的語音和文字互動。工作階段生命週期管理 (從初始信號交換到正常終止) 由開發人員控管。

本頁面說明如何使用 Gemini Live API，與 Gemini 模型展開對話。您可以透過 Vertex AI Studio、Google Gen AI SDK 或 WebSocket 啟動對話。

本頁面也會說明如何執行下列操作：

  • 延長工作階段的預設時間限制
  • 繼續執行前一個工作階段
  • 在工作階段期間更新系統指令
  • 設定工作階段的內容視窗
  • 為工作階段啟用轉錄功能

工作階段生命週期

如果沒有壓縮脈絡視窗，由於權杖限制，僅限音訊的對話時間最多為 15 分鐘，音訊和視訊對話時間最多為 2 分鐘。超過這些限制會導致工作階段終止，但您可以使用脈絡窗口壓縮功能，將工作階段延長至無限時間。

由於 WebSocket 連線限制，連線的生命週期約為 10 分鐘。連線終止時，工作階段也會終止。在這種情況下，您可以設定單一工作階段，透過工作階段續傳功能在多個連線中保持有效。連線結束前，您也會收到GoAway訊息，以便採取進一步行動。

並行工作階段數上限

在隨用隨付 (PayGo) 方案中，每個專案最多可有 1,000 個並行工作階段。使用佈建輸送量的客戶不受此限制。

開始練習

下列分頁說明如何使用 Vertex AI Studio、Google Gen AI SDK 或 WebSocket 啟動即時對話工作階段：

控制台

  1. 開啟 Vertex AI Studio > Stream realtime。
  2. 按一下「 開始工作階段」即可發起對話。

如要結束工作階段，請按一下「停止工作階段」。

Python

開始之前，請先使用 API 金鑰或應用程式預設憑證 (ADC) 向 Gemini Enterprise Agent Platform 進行驗證： 

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-native-audio"
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-native-audio"
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)。

如要延長工作階段，超過 10 分鐘的連線時間限制，您必須使用工作階段續傳功能重新連線。當您收到 goAway 通知，或連線因其他原因終止時，可以使用工作階段期間取得的工作階段控制代碼啟動新連線。這樣一來，工作階段就會在新連線上繼續執行，且內容不會遺失。您可以無限次數執行這項操作。如需續傳工作階段的範例，請參閱「續傳先前的工作階段」。

以下範例說明如何監聽 goAway 通知，偵測即將終止的會話：

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)

繼續執行前一個工作階段

Gemini Live API 支援工作階段續傳功能，可避免使用者在短暫中斷連線期間 (例如從 Wi-Fi 切換至 5G) 遺失對話內容。您可以在 24 小時內續傳先前的工作階段。系統會儲存快取資料 (包括文字、影片、音訊提示和模型輸出內容)，以實現工作階段續傳功能。系統會對這類快取資料強制執行專案層級的隱私權。

根據預設，工作階段續傳功能會停用。如要啟用工作階段續傳功能，請設定 BidiGenerateContentSetup 訊息的 sessionResumption 欄位。啟用後，伺服器會定期傳送包含 SessionResumptionUpdate 的訊息，以及繼續作業權杖。session_id如果 WebSocket 中斷連線，用戶端可以重新連線，並在新設定訊息中加入這些憑證。伺服器隨後會還原先前的脈絡，讓對話順暢接續。

繼續播放的時間有限 (通常約 10 分鐘)。如果用戶端未在這段時間內重新連線，系統就會捨棄工作階段狀態，以釋放伺服器資源。

以下範例會連線至服務、取得工作階段續傳控制代碼、模擬中斷連線，然後使用控制代碼重新連線，以續傳工作階段：

Python

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

# 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-native-audio"

async def resumable_session_example():
    """Demonstrates session resumption by connecting, disconnecting, and reconnecting."""
    session_handle = None

    print("Starting a new session...")
    try:
        async with client.aio.live.connect(
            model=MODEL,
            config=types.LiveConnectConfig(
                response_modalities=["audio"],
                session_resumption=types.SessionResumptionConfig(handle=None),
            ),
        ) as session:
            await session.send_content(
                content=types.Content(role="user", parts=[types.Part(text="Hello!")])
            )
            async for message in session.receive():
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        session_handle = update.new_handle
                        print(f"Received session handle: {session_handle}")
                        # For demonstration, we break to simulate a disconnect
                        # after receiving a handle.
                        break
                if message.server_content and message.server_content.turn_complete:
                    break
    except websockets.exceptions.WebSocketException as e:
        print(f"Initial connection failed: {e}")
        return

    if not session_handle:
        print("Did not receive a session handle. Cannot demonstrate resumption.")
        return

    print(f"\nSimulating disconnect and reconnecting with handle {session_handle}...")

    try:
        async with client.aio.live.connect(
            model=MODEL,
            config=types.LiveConnectConfig(
                response_modalities=["audio"],
                session_resumption=types.SessionResumptionConfig(handle=session_handle),
            ),
        ) as session:
            print("Successfully resumed session.")
            await session.send_content(
                content=types.Content(role="user", parts=[types.Part(text="I am back!")])
            )
            async for message in session.receive():
                if message.session_resumption_update:
                    update = message.session_resumption_update
                    if update.resumable and update.new_handle:
                        session_handle = update.new_handle
                        print(f"Received updated session handle: {session_handle}")
                if message.server_content:
                    print(f"Received server content: {message.server_content}")
                    if message.server_content.turn_complete:
                        break
            print("Resumed session finished.")
    except websockets.exceptions.WebSocketException as e:
        print(f"Failed to resume session: {e}")

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

啟用透明模式，順暢地繼續工作階段

啟用工作階段續傳功能時，您也可以啟用透明模式，讓使用者續傳工作階段時更順暢。啟用透明模式後，系統會明確傳回與情境快照對應的用戶端訊息索引，方便您在透過續傳控制代碼續傳工作階段時，判斷需要再次傳送的用戶端訊息。

如要啟用透明模式，請按照下列步驟操作：

Python

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

在工作階段期間更新系統指令

透過 Gemini Live API，您可以在進行中的工作階段更新系統指令，藉此調整模型的回覆內容，例如變更回覆語言或修改語氣。

如要在工作階段中更新系統指令，可以傳送具有 system 角色的文字內容。更新後的系統指令會在剩餘的工作階段中生效。

Python

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

設定工作階段的內容視窗

Gemini Live API 內容視窗用於儲存即時串流資料 (音訊每秒 25 個符記，影片每秒 258 個符記) 和其他內容，包括文字輸入內容和模型輸出內容。所有 Gemini Live API 模型都有 128,000 個符記的內容視窗限制。

在長時間的工作階段中，隨著對話進行，音訊和文字符記的記錄會不斷累積。如果這類記錄超過模型的限制，模型可能會產生幻覺、速度變慢，或工作階段遭到強制終止。如要啟用較長的工作階段，您可以設定工作階段設定中的 contextWindowCompression 欄位，啟用脈絡窗口壓縮功能。

啟用後，內容視窗壓縮功能會使用伺服器端滑動視窗，截斷最舊的輪次。當累積的權杖超過定義的最大長度 (在 Vertex AI Studio 中使用「內容大小上限」滑桿設定，或在 API 中使用 trigger_tokens)，伺服器會自動修剪最舊的輪次或摘要，以將內容維持在限制內。在 ContextWindowCompressionConfig 中，您可以設定滑動視窗機制，以及在 target_tokens 參數中定義的權杖數量，以觸發壓縮。

從使用者的角度來看，由於「記憶體」會持續管理，因此理論上工作階段時間無限。如果沒有壓縮，純音訊工作階段可能會在達到硬性限制前，限制在約 15 分鐘。

背景資訊長度和目標大小的長度下限和上限如下：

設定 (API 旗標) 最小值 最大值
背景資訊長度上限 (trigger_tokens) 5,000 128,000
脈絡長度目標 (target_tokens) 0 128,000

如何設定脈絡窗口：

控制台

  1. 開啟 Vertex AI Studio > Stream realtime。
  2. 按一下開啟「進階」選單。
  3. 在「工作階段內容」部分，使用「內容大小上限」滑桿，將內容大小設為介於 5,000 到 128,000 之間的值。
  4. (選用) 在同一區段中，使用「目標內容大小」滑桿，將目標大小設為介於 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 中。

如要提升多語言自動語音辨識 (ASR) 的轉錄品質，請使用 input_audio_transcription 或 output_audio_transcription 中的 language_codes 欄位提供語言提示。建議提供提示，以提升轉錄品質，因為這樣可降低語言偵測錯誤的風險，特別是簡短提示。language_codes 欄位接受 BCP-47 語言代碼清單 (例如「en-US」、「es-US」)。

config = {
    "response_modalities": ["audio", "text"],
    "input_audio_transcription": {
        "language_codes": ["en-US"]
    },
    "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.")

後續步驟