借助 Live API,您可以与 Gemini 进行低延迟的双向语音和视频互动。
本指南介绍了如何设置双向互动式对话、调整音频设置、管理会话等。
支持的模型
您可以将 Live API 与以下模型搭配使用:
模型版本 | 可用性级别 |
---|---|
gemini-live-2.5-flash |
非公开正式版* |
gemini-live-2.5-flash-preview-native-audio |
公开预览版 |
*请与您的 Google 客户支持团队代表联系,以请求访问权限。
发起对话
控制台
- 依次打开 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' INPUT_RATE=16000 OUTPUT_RATE=24000 client = genai.Client( vertexai=True, project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_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"}) await asyncio.sleep(10**-12) 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) await asyncio.sleep(10**-12) send_task = asyncio.create_task(send()) receive_task = asyncio.create_task(receive()) await asyncio.gather(send_task, receive_task) asyncio.run(main())
Python
设置与 API 的对话,以便您发送文本提示并接收音频回答:
# Set model generation_config CONFIG = {"response_modalities": ["AUDIO"]} headers = { "Content-Type": "application/json", "Authorization": f"Bearer {bearer_token[0]}", } async def main() -> None: # Connect to the server async with 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", "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()
更改语言和语音设置
Live API 使用 Chirp 3 来支持各种高清语音和语言的合成语音回答。如需查看每种语音的完整列表和演示,请参阅 Chirp 3:高清语音。
如需设置回答语音和语言,请执行以下操作:
控制台
- 依次打开 Vertex AI Studio > 流式传输实时数据。
- 在输出展开器中,从语音下拉菜单中选择一种语音。
- 在同一展开器中,从语言下拉菜单中选择一种语言。
- 点击 开始会话以开始会话。
Python
config = LiveConnectConfig( response_modalities=["AUDIO"], speech_config=SpeechConfig( voice_config=VoiceConfig( prebuilt_voice_config=PrebuiltVoiceConfig( voice_name=voice_name, ) ), language_code="en-US", ), )
更改语音活动检测设置
语音活动检测 (VAD) 可让模型识别用户何时在说话。这对于创建自然对话至关重要,因为用户可以随时中断模型。
模型会自动对连续的音频输入流执行语音活动检测 (VAD)。您可以使用设置消息的 realtimeInputConfig.automaticActivityDetection
字段配置 VAD 设置。当 VAD 检测到中断时,系统会取消并舍弃正在进行的生成操作。会话历史记录中仅保留已发送给客户端的信息。然后,服务器会发送一条消息来报告中断情况。
如果音频流暂停超过 1 秒(例如,用户关闭麦克风),请发送 audioStreamEnd
事件以刷新所有缓存的音频。客户端可以随时恢复发送音频数据。
或者,在设置消息中将 realtimeInputConfig.automaticActivityDetection.disabled
设置为 true
,以停用自动 VAD。在此配置中,客户端会检测用户语音,并在适当的时间发送 activityStart
和 activityEnd
消息。audioStreamEnd
未发送。中断以 activityEnd
标记。
Python
config = LiveConnectConfig( response_modalities=["TEXT"], realtime_input_config=RealtimeInputConfig( automatic_activity_detection=AutomaticActivityDetection( disabled=False, # default start_of_speech_sensitivity=StartSensitivity.START_SENSITIVITY_LOW, # Either START_SENSITIVITY_LOW or START_SENSITIVITY_HIGH end_of_speech_sensitivity=EndSensitivity.END_SENSITIVITY_LOW, # Either END_SENSITIVITY_LOW or END_SENSITIVITY_HIGH prefix_padding_ms=20, silence_duration_ms=100, ) ), ) async with client.aio.live.connect( model=MODEL_ID, config=config, ) as session: audio_bytes = Path("sample.pcm").read_bytes() await session.send_realtime_input( media=Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000") ) # if stream gets paused, send: # await session.send_realtime_input(audio_stream_end=True) response = [] async for message in session.receive(): if message.server_content.interrupted is True: # The model generation was interrupted response.append("The session was interrupted") if message.text: response.append(message.text) display(Markdown(f"**Response >** {''.join(response)}"))
延长会话
对话会话的默认最长时长为 10 分钟。系统会在会话结束前 60 秒向客户端发送 goAway
通知 (BidiGenerateContentServerMessage.goAway
)。
您可以使用 Gen AI SDK 以 10 分钟为增量延长会话时长。您可以延长会话的次数没有限制。如需查看示例,请参阅启用和停用会话恢复。
上下文窗口
实时 API 上下文窗口用于存储实时流式传输的数据(音频为每秒 25 个令牌 [TPS],视频为 258 TPS)和其他内容,包括文本输入和模型输出。
如果上下文窗口超过最大长度(在 Vertex AI Studio 中使用内容大小上限滑块设置,或在 API 中使用 trigger_tokens
设置),系统会使用上下文窗口压缩功能截断最旧的对话轮次,以防止会话突然终止。当上下文窗口达到最大长度(在 Vertex AI Studio 中使用目标上下文大小滑块设置,或在 API 中使用 target_tokens
设置)时,系统会触发上下文窗口压缩,并删除对话中最旧的部分,直到令牌总数降回此目标大小。
例如,如果您的上下文长度上限设置为 32000
个 token,而目标大小设置为 16000
个 token,则:
- 第 1 回合:对话开始。在此示例中,请求使用了 12,000 个令牌。
- 总上下文大小:12,000 个 token
- 第 2 回合:您再次提出要求。此请求使用了另外 12,000 个令牌。
- 总上下文大小:24,000 个 token
- 第 3 回合:您再次提出要求。此请求使用了 14,000 个令牌。
- 总上下文大小:38,000 个 token
由于总上下文大小现在超过了 32,000 个 token 的上限,因此系统会触发上下文窗口压缩。系统会返回到对话的开头,并开始删除旧对话轮次,直到令牌总大小小于 16,000 个令牌的目标值:
- 它会删除 Turn 1(12,000 个令牌)。现在的总数为 26,000 个 token,仍高于 16,000 个 token 的目标值。
- 它会删除 Turn 2(12,000 个令牌)。总数现在为 14,000 个令牌。
最终结果是,只有第 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), ), )
并发会话
每个项目最多可以有 5,000 个并发会话。
在对话期间更新系统指令
借助 Live API,您可以在活跃会话期间更新系统指令。使用此参数可调整模型的回答,例如更改回答语言或修改语气。
如需在会话中更新系统指令,您可以发送具有 system
角色的文本内容。更新后的系统指令将在剩余会话期间保持有效。
Python
session.send_client_content( turns=types.Content( role="system", parts=[types.Part(text="new system instruction")] ), turn_complete=False )
启用和停用会话恢复
会话恢复功能可让您在 24 小时内重新连接到之前的会话。这是通过存储缓存数据来实现的,包括文本、视频、音频提示和模型输出。系统会对缓存的这些数据强制执行项目级层隐私设置。
默认情况下,会话恢复处于停用状态。
如需启用会话恢复功能,请设置 BidiGenerateContentSetup
消息的 sessionResumption
字段。如果启用,服务器将定期拍摄当前缓存的会话上下文的快照,并将其存储在内部存储空间中。
成功拍摄快照后,系统会返回一个 resumptionUpdate
,其中包含您可以记录的句柄 ID,以便日后从快照恢复会话。
以下示例展示了如何启用会话恢复并检索句柄 ID:
Python
import asyncio from google import genai from google.genai import types client = genai.Client( vertexai=True, project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_LOCATION, ) model = "gemini-live-2.5-flash" 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
types.LiveConnectConfig( response_modalities=["AUDIO"], session_resumption=types.SessionResumptionConfig( transparent=True, ), )
启用透明模式后,系统会明确返回与上下文快照对应的客户端消息的索引。这有助于确定在从恢复句柄恢复会话时需要重新发送哪个客户端消息。
更多信息
如需详细了解如何使用 Live API,请参阅: