開始及管理直播工作階段

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 啟動即時對話工作階段:

控制台

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

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

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 欄位。啟用後,伺服器會定期傳送包含 SessionResumptionUpdate 的訊息,以及繼續作業權杖。session_id如果 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 內容視窗用於儲存即時串流資料 (音訊每秒 25 個符記,影片每秒 258 個符記) 和其他內容,包括文字輸入內容和模型輸出內容。工作階段的脈絡窗口限制為:

  • 原生音訊模型支援 12.8 萬個權杖
  • 其他 Live API 模型支援 3.2 萬個權杖

在長時間的對話中,隨著對話進行,音訊和文字符記的記錄會不斷累積。如果這類記錄超過模型上限,模型可能會產生幻覺、速度變慢,或強制終止工作階段。如要延長工作階段時間,您可以設定工作階段設定中的 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_tokenscontext_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_transcriptionoutput_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.")

後續步驟