Gradio全解11——Streaming：流式传输的视频应用（8）——Gemini Live API：实时音视频连接

本章目录如下：

1. 《Gradio全解11——Streaming：流式传输的视频应用（1）——FastRTC：Python实时通信库》
2. 《Gradio全解11——Streaming：流式传输的视频应用（2）——Twilio：网络服务提供商》
3. 《Gradio全解11——Streaming：流式传输的视频应用（3）——YOLO系列模型技术架构与实战》
4. 《Gradio全解11——Streaming：流式传输的视频应用（4）——基于Gradio.WebRTC+YOLO的实时目标检测》
5. 《Gradio全解11——Streaming：流式传输的视频应用（5）——RT-DETR：实时端到端检测模型》
6. 《Gradio全解10——Streaming：流式传输的视频应用（6）——基于RT-DETR模型构建目标检测系统》
7. 《Gradio全解11——Streaming：流式传输的视频应用（7）——多模态Gemini模型及其思考模式》
8. 《Gradio全解11——Streaming：流式传输的视频应用（8）——Gemini Live API：实时音视频连接》
9. 《Gradio全解11——Streaming：流式传输的视频应用（9）——使用FastRTC+Gemini创建沉浸式音频+视频的艺术评论家》

11.8 Gemini Live API：实时音视频连接

本节介绍Gemini Live API实时音视频连接，内容包括Live API入门讲解和Live API - WebSockets API详解。

11.8.1 Live API——入门

本节介绍当前非常热门的实时功能，对应OpenAI的GPT-Realtime，Gemini也有自己的实时接口：Live API。本节内容包括Live API技术与功能介绍、选择音频生成架构和实施方案、异步发送并接受音频示例、第三方集成应用与下一步学习建议。

1. Live API技术与功能介绍

Live API支持与Gemini建立低延迟的实时语音和视频互动。它会处理连续的音频、视频或文本流，以提供即时、类似人类的语音回答，从而为用户打造自然的对话体验。Live API与App的交互如图11-9所示：

图11-9

Live API提供了一整套全面的功能，例如语音活动检测、原生音频功能、工具使用和函数调用、会话管理（用于管理长时间运行的对话）和临时令牌（用于安全的客户端身份验证），详细信息请参阅：Get started with Live API🖇️链接11-43。了解如何将Live API用于端到端的简单用例：Multimodal Live API🖇️链接11-44。

2. 选择音频生成架构和实施方案

在使用Live API开始构建之前，需要做出两项重要抉择：选择音频生成架构和选择实施方案。

选择音频生成架构：如果构建基于音频的用例，所选模型将决定用于生成音频响应的音频生成架构，有以下两种选择：

• 原生（Native）音频：可提供最自然逼真的语音效果和更优异的多语言性能，同时支持高级功能，如情感感知对话、主动音频（模型可自主决定忽略或响应特定输入）以及“思考”功能。 以下模型支持原生音频：
  • gemini-2.5-flash-preview-native-audio-dialog
  • gemini-2.5-flash-exp-native-audio-thinking-dialog
• 半级联（Half-cascade）音频：采用级联模型架构（原生音频输入+文字转语音输出），在生产环境中具有更优异的性能和可靠性，特别是在工具使用场景中。以下模型支持半级联音频架构：
  • gemini-live-2.5-flash-preview
  • gemini-2.0-flash-live-001

选择实施方案：在集成Live API时，有两种实施方案可选：

• Server-To-Server模式：后端服务器通过WebSocket连接Live API服务器。通常由客户端将流数据（音频、视频、文本）发送至后端服务器，再由服务器转发至Live API服务器。
• Client-To-Server模式：客户端代码直接通过WebSocket连接Live API服务器传输流数据，无需经过后端服务器中转。

Client-To-Server模式在流式传输音频和视频时通常性能更佳，设置更简便，因为无需实现从客户端到后端服务器再到Live API服务器的数据代理转发。注意：在生产环境中，为降低安全风险，我们建议使用临时令牌而非标准API密钥，临时令牌功能请参阅：Ephemeral tokens🖇️链接11-45。

3. 异步发送并接受音频示例

以下示例演示如何读取WAV文件，以正确格式发送音频，并将接收数据保存为WAV文件。代码如下所示：

# Test file: https://storage.googleapis.com/generativeai-downloads/data/16000.wav
# Install helpers for converting files: pip install librosa soundfile
import asyncio
import io
from pathlib import Path
import wave
from google import genai
from google.genai import types
import soundfile as sf
import librosa
client = genai.Client()
# Half cascade model:
# model = "gemini-live-2.5-flash-preview"
# Native audio output model:
model = "gemini-2.5-flash-preview-native-audio-dialog"
config = {"response_modalities": ["AUDIO"],"system_instruction": "You are a helpful assistant and answer in a friendly tone.",
}
async def main():async with client.aio.live.connect(model=model, config=config) as session:buffer = io.BytesIO()y, sr = librosa.load("sample.wav", sr=16000)sf.write(buffer, y, sr, format='RAW', subtype='PCM_16')buffer.seek(0)audio_bytes = buffer.read()# If already in correct format, you can use this:# audio_bytes = Path("sample.pcm").read_bytes()await session.send_realtime_input(audio=types.Blob(data=audio_bytes, mime_type="audio/pcm;rate=16000"))wf = wave.open("audio.wav", "wb")wf.setnchannels(1) # monowf.setsampwidth(2) # 16 bit samplingwf.setframerate(24000)  # Output is 24kHzasync for response in session.receive():if response.data is not None:wf.writeframes(response.data)# Un-comment this code to print audio data info# if response.server_content.model_turn is not None:#      print(response.server_content.model_turn.parts[0].inline_data.mime_type)wf.close()
if __name__ == "__main__":asyncio.run(main())

这段代码实现了一个异步音频处理流程，主要用于实时音频流的发送和接收处理。以下是详细解析：

• ‌异步上下文管理‌。使用异步上下文管理器建立与Live API服务器的实时直连，model和config指定模型和配置参数。config将响应模态设置为AUDIO来接收音频，关于客户端的WebSocket连接将在11.8.4节详细讲解。
• ‌音频输入处理。首先通过io.BytesIO()初始化二进制内存缓冲区，用于临时存储音频数据。然后使用librosa.load‌加载本地WAV文件，并强制重采样到16kHz（语音处理的常用采样率）。最后使用sf.write‌将音频数据转为RAW格式（PCM_16编码）字节流，存入内存缓冲区buffer。
• 实时音频发送。首先重置缓冲区指针到起始位置，读取缓冲区中的全部音频字节数据。然后将音频数据封装为Blob对象，明确声明MIME类型为16kHz的PCM格式。最后通过异步调用发送到服务端。
• 输出文件配置。创建输出WAV文件，配置为单声道、16位深度、24kHz采样率（常见于高质量语音输出）。为防止命名冲突，可使用UUID产生随机文件名。
• 异步接收响应。持续接收服务端返回的音频数据流，直接写入WAV文件。注释部分显示可扩展为解析元数据（如语音识别结果）。

掌握这个发送和接受音频的复杂示例后，接下来了解下可以更简单的使用Gemini Live API的第三方集成应用及下一步学习建议。

4. 第三方集成应用与下一步学习建议

如果希望开发过程更简单，可以使用以下第三方集成应用：

• Daily and Pipecat🖇️链接11-46：Daily团队是Pipecat的幕后推手，自2016年以来提供全球WebRTC基础设施和企业级可靠性服务。Pipecat是一个开源的对话式AI框架，用于构建能够实时观察、聆听和说话的语音及能够整合音频、视频、图像和文本等多种媒介的多模态AI助手。该框架通过协调AI服务、网络传输和音频处理，实现超低延迟的自然流畅对话。
• LiveKit🖇️链接11-47：它是开源框架和实时云平台，使用Google AI和LiveKit Agents构建世界级的用于语音、视频和机器人的多模态AI应用程序。LiveKit提供SIP协议支持以实现电话系统集成，并提供多语言全功能前端SDK。框架采用WebRTC传输协议连接终端用户设备，能够实现高质量、低延迟的实时体验。
• Voximplant🖇️链接11-48：它是Google Gemini Live API客户端，提供市场上最灵活、最强大的Gemini Live API集成方案，通过闪电般快速的接口，将呼入/呼出电话连接至Gemini驱动的智能体。作为平台的原生集成功能，可帮助开发者轻松为客户打造极其自然且引人入胜的对话体验，与能理解需求、处理复杂任务并解决问题的AI进行自然轻松的交流。

这些是第三方合作伙伴平台，已通过WebRTC协议集成Gemini Live API，可简化实时音频和视频应用的开发。

下一步学习建议：

• 阅读完整的Live API Capabilities guide🖇️链接11-49：了解关键功能与配置，包括语音活动检测和原生音频特性。
• 阅读Tool use🖇️链接11-50：学习如何将Live API与工具及函数调用进行集成。
• 阅读Session management🖇️链接11-51：掌握长对话管理的具体方法。
• 阅读Ephemeral tokens🖇️链接11-52 ，了解Client-To-Terver应用中的安全认证机制。

下面讲解Live API的底层原理：WebSockets API。

11.8.2 Live API——WebSockets API参考

Live API底层使用的是WebSockets进行连接。本节将介绍WebSockets API概念及功能、Live API连接时的会话配置、发送与接收操作及消息类型。

1. WebSockets API概念及功能

Live API是一种使用WebSockets的有状态API，WebSocket连接会在客户端和Gemini服务器之间建立会话。客户端发起新连接后，会话可以与服务器交换消息，以执行以下操作：

• 向Gemini服务器发送文本、音频或视频。
• 接收来自Gemini服务器的音频、文本或函数调用请求。

WebSocket（🖇️链接11-53 ）是一种计算机通信协议，通过简单TCP（Transmission Control Protocol，传输控制协议）连接提供全双工通信通道，于2011年被IETF（Internet Engineering Task Force，互联网工程任务组）标准化为RFC 6455（🖇️链接11-54）。当前允许Web应用程序使用此协议的规范称为WebSockets，这是由WHATWG（Web Hypertext Application Technology Working Group，Web超文本应用技术工作组）维护的现行标准，也是W3C（World Wide Web Consortium，万维网联盟）的WebSocket API的继任者。

WebSocket协议不同于被大多数网页使用的HTTP协议。RFC 6455规定WebSocket"设计工作在HTTP 443和80端口，并支持HTTP代理及中介设备"，因此尽管存在差异，但WebSocket协议与HTTP保持兼容。为实现兼容性，WebSocket握手过程中，使用HTTP Upgrade报头将HTTP协议切换至WebSocket协议。

当Google客户端开启会话时，请连接到此WebSocket端点： wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent。注意： 此网址适用于版本v1beta。有关底层WebSocket API和消息类型的详细信息，请参阅Live API - WebSockets API reference🖇️链接11-55。

2. 连接时的会话配置

连接后的初始消息中会包含会话配置，比如代码client.aio.live.connect(model='...', config=config)，其中aio代表AsyncClient，函数connect的构造参数包含的会话配置有模型名称、生成配置、系统说明和工具等，用户可以在会话期间更改除模型名称以外的配置参数，关于genai.Client的详细配置请参阅：Google Gen AI SDK🖇️链接11-56。connect构造参数详情如下所示：

{"model": string,"generationConfig": {"candidateCount": integer,"maxOutputTokens": integer,"temperature": number,"topP": number,"topK": integer,"presencePenalty": number,"frequencyPenalty": number,"responseModalities": [string],"speechConfig": object,"mediaResolution": object},"systemInstruction": string,"tools": [object]
}

其中模型名称、系统说明和工具容易理解和使用，下面着重讲解生成配置generationConfig。generationConfig用于模型生成与输出，详情请参阅：GenerationConfig🖇️链接11-57。以本示例的十个参数为例，解读如下：

• candidateCount：返回生成响应的数量。若未设置，默认值为1。请注意此参数不适用于旧版模型。
• maxOutputTokens：响应候选内容中包含的最大token数。注意：默认值因模型而异，请参阅getModel函数返回的Model.output_token_limit属性。
• temperature：控制输出结果的随机性，取值范围为[0.0, 2.0]。注意：默认值因模型而异，请参阅getModel函数返回的Model.temperature属性。
• topP：采样时的token最大累积概率。该模型采用Top-k和Top-p（核采样）组合采样方式。注意：默认值因模型而异，由getModel函数返回的Model.top_p属性指定。
• topK：采样时的最大token数量，只考虑前topK个最可能token。Gemini模型采用Top-p（核采样）或Top-k与核采样组合方式。
• presencePenalty：若token已在响应中出现，则对其后续token的对数概率施加存在性惩罚presencePenalty。该惩罚为二元开关机制，不依赖于token的使用次数（首次出现后即触发）。如需随使用次数增加的惩罚机制，请使用frequencyPenalty参数。
• frequencyPenalty：基于token在响应中出现次数的频率惩罚机制，它乘以当前token出现次数后应用于后续token的对数概率。正惩罚值将按使用次数比例抑制已使用token的重复出现，从而有效增加响应词汇多样性。反之，负惩罚值将鼓励已使用token的重复出现。
• responseModalities：返回响应中应设的模态类型，且服务端的模型能够返回的模态集合，即该设置必须与响应模态完全匹配。模型可能支持多种模态组合，若请求的模态与任何支持组合不匹配，将返回错误，空列表等效于仅请求文本模态。模态类型Modality包括：MODALITY_UNSPECIFIED、TEXT、IMAGE、AUDIO、VIDEO、DOCUMENT。
• speechConfig：语音生成配置参数，详情请参阅：SpeechConfig。
• mediaResolution：若指定该参数，输入将使用指定的媒体解决方案设置，详情请参阅：MediaResolution。

请注意，SDK中的名称及大小写可能有所不同，比如responseModalities可能为response_modalities，详情请参阅：python-genai SDK配置🖇️链接11-58 。另外，请注意并非所有参数都适用于每个模型。

3. 发送与接收操作及消息类型

本节将解析客户端发送到服务器的消息类型，以及从服务器接收的消息类型。
发送消息：客户端要通过WebSocket连接交换消息，必须通过已建立的WebSocket连接发送JSON对象。该JSON对象必须只包含以下对象集中某一个字段：

{"setup": BidiGenerateContentSetup,"clientContent": BidiGenerateContentClientContent,"realtimeInput": BidiGenerateContentRealtimeInput,"toolResponse": BidiGenerateContentToolResponse
}

客户端消息类型说明如表11-5所示：

表11-5

消息类型	描述
BidiGenerateContentSetup	在首条消息中发送的会话配置
BidiGenerateContentClientContent	客户端发送的当前会话增量内容更新
BidiGenerateContentRealtimeInput	实时音频、视频或文本输入
BidiGenerateContentToolResponse	对从服务器接收的ToolCallMessage的响应

接收消息：要从Gemini Live API接收消息，需要监听WebSocket的’message’事件，然后根据支持的服务器消息定义解析结果。发送和接受消息的样例代码如下所示：

from google import genai
client = genai.Client()
async with client.aio.live.connect(model='...', config=config) as session:await session.send(input='Hello world!', end_of_turn=True)async for message in session.receive():print(message)

服务器返回消息类型为BidiGenerateContentServerMessage，它作为BidiGenerateContent调用的响应消息，可能包含usageMetadata字段，但除此之外，联合字段messageType只包含以下消息类型之一：setupComplete、serverContent、toolCall、toolCallCancellation、goAway、sessionResumptionUpdate。详情请参阅：BidiGenerateContentServerMessage🖇️链接11-59 。

如需详细了解常用的API资源类型，如Blob、Content、FunctionCall、FunctionResponse、GenerationConfig、GroundingMetadata、ModalityTokenCount和Tool，请参阅Generating content🖇️链接11-60 。