开始和管理实时会话

Gemini Live API 通过处理连续的音频或文本流（称为会话），实现低延迟的语音和文本交互，从而提供即时、自然逼真的语音回答。会话生命周期管理（从初始握手到正常终止）由开发者控制。

本页面介绍了如何使用 Gemini Live API 开始与 Gemini 模型进行对话会话。您可以使用 Vertex AI Studio、Gen AI SDK 或 WebSockets 开始会话。

本页面还介绍了如何执行以下操作：

• 将会话延长到超出默认时限
• 恢复上一个会话
• 在会话期间更新系统指令
• 配置会话的上下文窗口
• 为会话启用转写功能

会话生命周期

如果不进行压缩，纯音频会话的时长上限为 15 分钟，音频-视频会话的时长上限为 2 分钟。超出这些限制会导致会话终止，但您可以使用上下文窗口压缩功能将对话时长延长至无限。

连接的生命周期限制为大约 10 分钟。当连接终止时，会话也会终止。在这种情况下，您可以使用会话恢复功能，将单个会话配置为在多个连接中保持活跃状态。您还会在连接结束前收到一条 GoAway 消息，以便您采取进一步的措施。

并发会话数上限

对于随用随付 (PayGo) 方案，每个项目最多可以有 1,000 个并发会话。此限制不适用于使用预配吞吐量的客户。

启动会话

以下标签页展示了如何使用 Vertex AI Studio、Gen AI SDK 或 WebSockets 开始实时对话会话：

控制台

1. 依次打开 Vertex AI Studio > 实时流。
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())
      

注意： 虽然某些参数（包括系统指令）可以在会话中进行调整，但模型和其他参数通常在设置消息处理完毕后就无法更改。

延长会话

注意：仅当使用 Gen AI SDK（而不是 Vertex AI Studio）时，会话延长功能才可用。

对话会话的默认时长上限为 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 字段。如果启用，服务器会定期发送包含 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-live-2.5-flash-preview-native-audio-09-2025"

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_config=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_config=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_content(
      content=types.Content(
          role="system", parts=[types.Part(text="new system instruction")]
      ),
      turn_complete=False
  )
      

配置会话的上下文窗口

Gemini Live API 上下文窗口用于存储实时流式数据（对于音频为每秒 25 个 token [TPS]，对于视频为每秒 258 个 token）和其他内容，包括文本输入和模型输出。会话的上下文窗口限制为：

• 12.8 万个 token，适用于原生音频模型
• 其他 Live API 模型的令牌数量为 32,000

在长时间运行的会话中，随着对话的进行，音频和文本令牌的历史记录会不断累积。如果此历史记录超出模型的限制，模型可能会产生幻觉、运行速度变慢，或者会话可能会被强制终止。如需启用更长的会话，您可以在会话配置中设置 contextWindowCompression 字段，以启用上下文窗口压缩。

启用上下文窗口压缩后，系统会使用服务器端滑动窗口截断最旧的对话轮次。当累积的 token 超过定义的最大长度（在 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 > 实时流。
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 中。

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.")

后续步骤

• 发送音频和视频流
• Gemini Live API 的最佳实践
• 设计多模态提示
• 函数调用简介