本页面介绍了如何使用 Live API 开始与 Gemini 模型进行实时的互动式对话会话。您可以使用 Vertex AI Studio、Gen AI SDK 或 WebSockets 开始会话。
Live API 通过处理连续的音频或文本流，实现低延迟的语音和文本交互，从而提供即时、自然逼真的语音回答。
本页面还介绍了如何执行以下操作：
- 将会话延长到超出默认时限
- 恢复上一个会话
- 在会话期间更新系统指令
- 配置会话的上下文窗口
- 为会话启用转写功能
启动会话
以下标签页展示了如何使用 Vertex AI Studio、Gen AI SDK 或 WebSockets 开始实时对话会话：
控制台
- 依次打开 Vertex AI Studio > 实时流。
- 点击 启动会话以发起对话。
如需结束会话，请点击 停止会话。
Python
如需启用实时互动式对话，请在可访问麦克风和扬声器的本地计算机上运行以下示例（而非 Colab 笔记本）。此示例设置了与 API 的对话，让您可以发送音频提示并接收音频回答：
""" # Installation # on linux sudo apt-get install portaudio19-dev # on mac brew install portaudio python3 -m venv env source env/bin/activate pip install google-genai """ import asyncio import pyaudio from google import genai from google.genai import types CHUNK=4200 FORMAT=pyaudio.paInt16 CHANNELS=1 RECORD_SECONDS=5 MODEL = 'gemini-live-2.5-flash-preview-native-audio-09-2025' INPUT_RATE=16000 OUTPUT_RATE=24000 client = genai.Client( vertexai=True, project='PROJECT_ID', location='LOCATION', ) config = { "response_modalities": ["AUDIO"], "input_audio_transcription": {}, # Configure input transcription "output_audio_transcription": {}, # Configure output transcription } async def main(): print(MODEL) p = pyaudio.PyAudio() async with client.aio.live.connect(model=MODEL, config=config) as session: #exit() async def send(): stream = p.open( format=FORMAT, channels=CHANNELS, rate=INPUT_RATE, input=True, frames_per_buffer=CHUNK) while True: frame = stream.read(CHUNK) await session.send(input={"data": frame, "mime_type": "audio/pcm"}) async def receive(): output_stream = p.open( format=FORMAT, channels=CHANNELS, rate=OUTPUT_RATE, output=True, frames_per_buffer=CHUNK) async for message in session.receive(): if message.server_content.input_transcription: print(message.server_content.model_dump(mode="json", exclude_none=True)) if message.server_content.output_transcription: print(message.server_content.model_dump(mode="json", exclude_none=True)) if message.server_content.model_turn: for part in message.server_content.model_turn.parts: if part.inline_data.data: audio_data=part.inline_data.data output_stream.write(audio_data) send_task = asyncio.create_task(send()) receive_task = asyncio.create_task(receive()) await asyncio.gather(send_task, receive_task) asyncio.run(main())
Python
安装此示例所需的库：
pip install websockets numpy ipython
设置与 API 的对话，以便您可以发送文本提示并接收音频回答：
import asyncio import websockets import json import base64 import numpy as np from IPython.display import display, Markdown, Audio # Set model generation_config CONFIG = {"response_modalities": ["AUDIO"]} # Authenticate with Google Cloud. Get an access token by running: # gcloud auth application-default print-access-token bearer_token = ["ACCESS_TOKEN"] SERVICE_URL = "wss://{LOCATION}-aiplatform.googleapis.com/ws/google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } async def main() -> None: # Connect to the server async with websockets.connect(SERVICE_URL, additional_headers=headers) as ws: # Setup the session async def setup() -> None: await ws.send( json.dumps( { "setup": { "model": "gemini-live-2.5-flash-preview-native-audio-09-2025", "generation_config": CONFIG, } } ) ) # Receive setup response raw_response = await ws.recv(decode=False) setup_response = json.loads(raw_response.decode("ascii")) print(f"Connected: {setup_response}") return # Send text message async def send() -> bool: text_input = input("Input > ") if text_input.lower() in ("q", "quit", "exit"): return False msg = { "client_content": { "turns": [{"role": "user", "parts": [{"text": text_input}]}], "turn_complete": True, } } await ws.send(json.dumps(msg)) return True # Receive server response async def receive() -> None: responses = [] # Receive chucks of server response async for raw_response in ws: response = json.loads(raw_response.decode()) server_content = response.pop("serverContent", None) if server_content is None: break model_turn = server_content.pop("modelTurn", None) if model_turn is not None: parts = model_turn.pop("parts", None) if parts is not None: for part in parts: pcm_data = base64.b64decode(part["inlineData"]["data"]) responses.append(np.frombuffer(pcm_data, dtype=np.int16)) # End of turn turn_complete = server_content.pop("turnComplete", None) if turn_complete: break # Play the returned audio message display(Markdown("**Response >**")) display(Audio(np.concatenate(responses), rate=24000, autoplay=True)) return await setup() while True: if not await send(): break await receive()
开始对话，输入提示，或是输入 q、quit 或 exit 以退出。
await main()
并发会话数上限
对于随用随付 (PayGo) 方案，每个项目最多可以有 1,000 个并发会话。此限制不适用于使用预配吞吐量的客户。
延长会话
对话会话的默认时长上限为 10 分钟。系统会在会话结束前 60 秒向客户端发送 goAway 通知 (BidiGenerateContentServerMessage.goAway)。
您可以使用 Gen AI SDK 以 10 分钟为增量延长会话时长。您可以无限次延长会话。如需查看示例，请参阅恢复上一个会话。
恢复上一个会话
您可以在 24 小时内恢复之前的会话。会话恢复是通过存储缓存的数据（包括文本、视频、音频提示和模型输出）来实现的。系统会对这些缓存的数据强制执行项目级隐私设置。
默认情况下，会话恢复功能处于停用状态。 如需启用会话恢复功能，请设置 BidiGenerateContentSetup 消息的 sessionResumption 字段。如果启用，服务器会定期拍摄当前缓存的会话上下文的快照，并将其存储在内部存储空间中。
成功拍摄快照后，系统会返回一个 resumptionUpdate，其中包含您可以记录的句柄 ID，以便以后用于从快照恢复会话。
以下示例展示了如何启用会话恢复功能并检索句柄 ID：
Python
import asyncio from google import genai from google.genai import types client = genai.Client( vertexai=True, project='PROJECT_ID', location='LOCATION', ) model = "gemini-live-2.5-flash-preview-native-audio-09-2025" async def main(): previous_session_handle = None 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())
启用无缝会话恢复
启用会话恢复后，您还可以启用透明模式，以帮助用户更顺畅地完成恢复流程。
启用透明模式后，服务器定期发送的 SessionResumptionUpdate 消息会包含会话状态的句柄以及与该服务器端快照对应的客户端消息的索引。如果连接中断，此索引可帮助您确定客户端在生成快照之后发送了哪些消息（如果有）。通过在重新连接后仅重新发送这些消息，您可以确保更顺畅地恢复对话，而不会丢失最近的用户输入。
您可以按如下方式启用透明模式：
Python
types.LiveConnectConfig( response_modalities=["AUDIO"], session_resumption=types.SessionResumptionConfig( 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 个 token [TPS]，对于视频为每秒 258 个 token）和其他内容，包括文本输入和模型输出。
如果上下文窗口超过长度上限（在 Vertex AI Studio 中使用内容大小上限滑块设置，或在 API 中使用 trigger_tokens 设置），系统会使用上下文窗口压缩功能截断最早的对话轮次，以防止会话突然终止。当上下文窗口达到长度上限（在 Vertex AI Studio 中使用目标上下文大小滑块设置，或在 API 中使用 target_tokens 设置）时，系统会触发上下文窗口压缩功能，并删除对话中最早的部分，直到 token 总数回落到此目标大小。
例如，如果您的上下文长度上限设置为 32000 个 token，而目标大小设置为 16000 个 token：
- 第 1 轮：对话开始。在此示例中，请求使用 12,000 个 token。
- 总上下文大小：12,000 个 token
- 第 2 轮：您发出另一个请求。此请求使用另外 12,000 个 token。
- 总上下文大小：24,000 个 token
- 第 3 轮：您发出另一个请求。此请求使用 14,000 个 token。
- 总上下文大小：38,000 个 token
由于总上下文大小现在高于 32,000 个 token 的上限，因此系统现在会触发上下文窗口压缩功能。系统会返回到对话的开头，开始删除早期对话轮次，直到 token 总大小小于 16,000 个 token 的目标值：
- 它会删除第 1 轮（12,000 个令牌）。总数现在为 26,000 个 token，仍高于 16,000 个 token 的目标值。
- 它会删除第 2 轮（12,000 个令牌）。总数现在为 14,000 个 token。
最终结果是只有第 3 轮保留在活跃内存中，对话从该点继续进行。
上下文长度和目标大小的长度下限和上限如下：
| 设置（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 = types.LiveConnectConfig( temperature=0.7, response_modalities=['TEXT'], system_instruction=types.Content( parts=[types.Part(text='test instruction')], role='user' ), context_window_compression=types.ContextWindowCompressionConfig( trigger_tokens=1000, sliding_window=types.SlidingWindow(target_tokens=10), ), )
为会话启用音频转写功能
您可以为输入和输出音频启用转写功能。
如需接收转写内容，您必须更新会话配置。您需要添加 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.")