Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

开始和管理实时会话

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

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

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

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

会话生命周期

如果没有上下文窗口压缩，由于 token 限制，仅音频会话的时长上限为 15 分钟，音频-视频会话的时长上限为 2 分钟。超出这些限制将会终止会话，但您可以使用上下文窗口压缩将会话延长到无限时长。

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

并发会话数上限

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

启动会话

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

控制台

依次打开 Vertex AI Studio > 实时流。
点击 启动会话以发起对话。

如需结束会话，请点击 停止会话。

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

使用 WebSockets 时，连接是通过标准 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 字段。如果启用，服务器会定期发送包含 session_id 和恢复 token 的 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-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 个 token [TPS]，对于视频为每秒 258 个 token）和其他内容，包括文本输入和模型输出。所有 Gemini Live API 模型都有 128k token 的上下文窗口限制。

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

启用后，上下文窗口压缩会使用服务器端滑动窗口来截断最早的对话轮次。当累积的 token 超出定义的最大长度（在 Vertex AI Studio 中使用内容大小上限 滑块设置，或在 API 中使用 trigger_tokens 设置）时，服务器会自动修剪最早的对话轮次或对其进行总结，以将上下文保持在限制范围内。在 ContextWindowCompressionConfig 中，您可以配置滑动窗口机制以及在 target_tokens 参数中定义的触发压缩的 token 数量。

这样一来，从用户的角度来看，会话时长理论上是无限的，因为“内存”会不断得到管理。如果没有压缩，仅音频会话的时长可能会限制在约 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 中。

如需提高多语言自动语音识别 (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.")

开始和管理实时会话 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

会话生命周期

并发会话数上限

启动会话

控制台

Python

Python

延长会话

Python

恢复上一个会话

Python

使用透明模式启用无缝会话恢复

Python

在会话期间更新系统指令

Python

配置会话的上下文窗口

控制台

Python

为会话启用音频转写功能

处理回答

后续步骤

开始和管理实时会话