開始及管理直播工作階段

Gemini Live API 會處理連續的音訊或文字串流 (稱為工作階段)，提供低延遲的語音和文字互動，並即時生成擬真的語音回覆。從初始交握到正常終止，工作階段生命週期管理都由開發人員控制。

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

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

  • 延長工作階段，超過預設時間限制
  • 繼續先前的課程
  • 在工作階段期間更新系統指令
  • 在 Gemini 3.8 Live 工作階段中更新系統指令和工具
  • 設定工作階段的脈絡窗口
  • 啟用會議語音轉錄功能
  • 使用客戶自行管理的加密金鑰 (CMEK) 加密工作階段資料

工作階段生命週期

如果沒有脈絡視窗壓縮功能，受限於詞元數量，純音訊工作階段最多只能持續 15 分鐘，影音工作階段則最多 2 分鐘。如果超過這些限制，工作階段就會終止，但您可以使用脈絡視窗壓縮功能，無限延長工作階段。

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

並行工作階段數量上限

系統會在工作階段開始時檢查並套用並行工作階段配額，無論您使用的是標準即付即用 (Standard PayGo) 還是佈建輸送量：

開始上課

下列分頁說明如何使用 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-3.8-live"
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-3.8-live"
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 分鐘，您可以使用 goAway 通知延長。系統會在工作階段結束前 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 欄位。啟用後，伺服器會定期傳送包含 session_id 和恢復權杖的 SessionResumptionUpdate 訊息。如果 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-3.8-live"

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,
   }
}

使用 CMEK 加密工作階段資料

工作階段恢復功能會將快取的工作階段資料儲存最多 24 小時。如要使用自己的金鑰 (而非Google-owned and Google-managed encryption key) 保護資料，請使用服務設定檔。服務設定檔會將 Cloud Key Management Service 金鑰連結至專案、位置和 API 範圍。只要有服務設定檔，Agent Platform 就會自動加密資料，工作階段設定和要求格式不會改變。

Gemini Live API 的服務設定檔 CMEK 支援 us 和 eu 多區域。不支援全域區域。

如要為 Gemini Live API 設定 CMEK，請按照下列步驟操作：

  1. 在專案中啟用 Cloud KMS API 和 Agent Platform API。請參閱事前準備一節。
  2. 在 us 或 eu 多區域中建立 Cloud KMS 金鑰。請參閱「建立金鑰環和金鑰」。
  3. 將金鑰的 roles/cloudkms.cryptoKeyEncrypterDecrypter 角色授予 Gemini Enterprise Agent Platform 服務代理。請參閱「授予 Agent Platform 權限」。
  4. 建立供應設定檔，將 scope 設為 GEMINI_LIVE，並將 cmekConfig.encryptionSpec.kmsKeyName 設為金鑰。請參閱「建立供應設定檔」。

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

Gemini Live API 可讓您在進行中的工作階段更新系統指令。這項功能可用於調整模型的回覆，例如變更回覆語言或語氣。

下列方法適用於 Gemini 2.5 Flash Live API Native Audio。如要瞭解 Gemini 3.8 Live，請參閱「使用 Gemini 3.8 Live 更新系統指令和工具」。

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

Python

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

使用 Gemini 3.8 Live 更新系統指令和工具

使用 Gemini 3.8 Live 時，您可以傳送 BidiGenerateContentContextUpdate 訊息，更新進行中工作階段的系統指令和工具。舉例來說，當對話轉移到新工作時，您可以變更回覆語言或語氣，或為模型提供一組不同的函式。使用 Google Gen AI SDK 呼叫 send_context_update。使用 WebSocket 時，請設定用戶端訊息的 context_update 欄位。請勿將具有 system 角色的用戶端內容傳送至 Gemini 3.8 Live。

設定要變更的欄位：

  • system_instruction：在工作階段的其餘部分，取代設定訊息中的系統指令。
  • tools：取代工作階段的整個工具清單。請加入模型仍應能使用的所有工具。如要移除所有工具，請傳送空白清單。

未設定的欄位會保留目前的值。系統會按照工作階段其餘輸入內容的順序處理更新。

Python

from google.genai import types

# Replace the system instruction.
await session.send_context_update(
    system_instruction="Respond only in Spanish."
)

# Replace the tool list.
get_current_weather = types.FunctionDeclaration(
    name="get_current_weather",
    description="Get the current weather in a given location",
    parameters=types.Schema(
        type=types.Type.OBJECT,
        properties={"location": types.Schema(type=types.Type.STRING)},
    ),
)
await session.send_context_update(
    tools=[types.Tool(function_declarations=[get_current_weather])]
)

# Remove all tools.
await session.send_context_update(tools=[])

伺服器端工具會在工作階段開始時設定，因此如果工具更新包含這些工具，則適用下列限制。函式宣告不受影響，可在任何工具更新中新增。

  • 如要在工具更新後使用「以 Google 搜尋強化事實基礎」功能，也請在設定訊息中宣告 google_search。
  • Enterprise 網頁基準建立、Agent Search 和 RAG Engine 工具只能在設定訊息中宣告。如果 context_update 包含這些內容，則會傳回 INVALID_ARGUMENT 錯誤。

設定工作階段的脈絡窗口

Gemini Live API 的脈絡窗口用於儲存即時串流資料 (音訊每秒 25 個詞元 (TPS)，影片每秒 258 個詞元) 和其他內容，包括文字輸入和模型輸出。所有 Gemini Live API 模型的脈絡窗口上限為 12.8 萬個詞元。

在長時間執行的工作階段中，隨著對話進行，音訊和文字詞元的記錄會累積。如果記錄超出模型限制，模型可能會產生幻覺、速度變慢，或強制終止工作階段。如要延長工作階段，請在工作階段設定中設定 contextWindowCompression 欄位，啟用脈絡視窗壓縮功能。

啟用後，脈絡視窗壓縮功能會使用伺服器端滑動視窗，截斷最舊的回合。當累積的詞元超過定義的最大長度 (使用 Vertex AI Studio 中的「Max content size」(內容大小上限) 滑桿或 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 物件，並確保 response_modalities 包含 text。

如要提升多語言自動語音辨識 (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.")

後續步驟