第 5 部分：如何使用音频、图像和视频¶

本节涵盖 ADK Live API 集成中的音频、图像和视频功能，包括支持的模型、音频模型架构、规范以及实现语音和视频功能的最佳实践。

如何使用音频¶

Live API 的音频功能通过双向音频流式传输实现亚秒级延迟的自然语音对话。本节涵盖如何向模型发送音频输入并接收音频响应，包括格式要求、流式传输最佳实践和客户端实现模式。

发送音频输入¶

音频格式要求：

在调用 send_realtime() 之前，确保你的音频数据已经是正确的格式：

格式：16 位 PCM（有符号整数）
采样率：16,000 Hz (16kHz)
声道：单声道（单通道）

ADK 不执行音频格式转换。以不正确的格式发送音频将导致质量差或错误。

演示实现：main.py:141-145

audio_blob = types.Blob(
    mime_type="audio/pcm;rate=16000",
    data=audio_data
)
live_request_queue.send_realtime(audio_blob)

发送音频输入的最佳实践¶

分块流式传输：以小块发送音频以实现低延迟。根据你的延迟要求选择块大小：
- 超低延迟（实时对话）：10-20ms 块（~320-640 字节 @ 16kHz）
- 平衡（推荐）：50-100ms 块（~1600-3200 字节 @ 16kHz）
- 较低开销：100-200ms 块（~3200-6400 字节 @ 16kHz）
在整个会话中使用一致的块大小以获得最佳性能。示例：100ms @ 16kHz = 16000 样本/秒 × 0.1 秒 × 2 字节/样本 = 3200 字节。
及时转发：ADK 的 LiveRequestQueue 及时转发每个块，而不进行合并或批处理。选择满足你的延迟和带宽要求的块大小。不要在发送下一个块之前等待模型响应。
连续处理：模型连续处理音频，而不是逐轮处理。启用自动 VAD（默认）后，只需连续流式传输并让 API 检测语音。
活动信号：仅当你明确禁用 VAD 以进行手动轮次控制时，才使用 send_activity_start() / send_activity_end()。VAD 默认启用，因此大多数应用程序不需要活动信号。

在客户端处理音频输入¶

在基于浏览器的应用程序中，捕获麦克风音频并将其发送到服务器需要使用带有 AudioWorklet 处理器的 Web Audio API。bidi-demo 演示了如何捕获麦克风输入，将其转换为所需的 16kHz 16 位 PCM 格式，并将其连续流式传输到 WebSocket 服务器。

架构：

音频捕获：使用 Web Audio API 以 16kHz 采样率访问麦克风
音频处理：AudioWorklet 处理器实时捕获音频帧
格式转换：将 Float32Array 样本转换为 16 位 PCM
WebSocket 流式传输：通过 WebSocket 发送 PCM 块到服务器

演示实现：audio-recorder.js:7-58

// 启动音频记录器 worklet
export async function startAudioRecorderWorklet(audioRecorderHandler) {
    // 创建具有 16kHz 采样率的 AudioContext
    // 这与 Live API 要求的输入格式（16 位 PCM @ 16kHz）相匹配
    const audioRecorderContext = new AudioContext({ sampleRate: 16000 });

    // 加载将实时处理音频的 AudioWorklet 模块
    // AudioWorklet 在单独的线程上运行以进行低延迟、无故障的音频处理
    const workletURL = new URL("./pcm-recorder-processor.js", import.meta.url);
    await audioRecorderContext.audioWorklet.addModule(workletURL);

    // 请求访问用户的麦克风
    // channelCount: 1 请求单声道音频（单通道），这是 Live API 所要求的
    micStream = await navigator.mediaDevices.getUserMedia({
        audio: { channelCount: 1 }
    });
    const source = audioRecorderContext.createMediaStreamSource(micStream);

    // 创建一个使用我们自定义 PCM 记录器处理器的 AudioWorkletNode
    // 此节点将捕获音频帧并将其发送到我们的处理程序
    const audioRecorderNode = new AudioWorkletNode(
        audioRecorderContext,
        "pcm-recorder-processor"
    );

    // 将麦克风源连接到 worklet 处理器
    // 处理器将接收音频帧并通过 port.postMessage 发布它们
    source.connect(audioRecorderNode);
    audioRecorderNode.port.onmessage = (event) => {
        // 将 Float32Array 转换为 Live API 要求的 16 位 PCM 格式
        const pcmData = convertFloat32ToPCM(event.data);

        // 将 PCM 数据发送到处理程序（它将转发到 WebSocket）
        audioRecorderHandler(pcmData);
    };
    return [audioRecorderNode, audioRecorderContext, micStream];
}

// 将 Float32 样本转换为 16 位 PCM
function convertFloat32ToPCM(inputData) {
    // 创建相同长度的 Int16Array
    const pcm16 = new Int16Array(inputData.length);
    for (let i = 0; i < inputData.length; i++) {
        // Web Audio API 提供范围 [-1.0, 1.0] 内的 Float32 样本
        // 乘以 0x7fff (32767) 以转换为 16 位有符号整数范围 [-32768, 32767]
        pcm16[i] = inputData[i] * 0x7fff;
    }
    // 返回底层 ArrayBuffer（二进制数据）以进行高效传输
    return pcm16.buffer;
}

演示实现：pcm-recorder-processor.js:1-19

// pcm-recorder-processor.js - 用于捕获音频的 AudioWorklet 处理器
class PCMProcessor extends AudioWorkletProcessor {
    constructor() {
        super();
    }

    process(inputs, outputs, parameters) {
        if (inputs.length > 0 && inputs[0].length > 0) {
            // 使用第一个通道（单声道）
            const inputChannel = inputs[0][0];
            // 复制缓冲区以避免回收内存的问题
            const inputCopy = new Float32Array(inputChannel);
            this.port.postMessage(inputCopy);
        }
        return true;
    }
}

registerProcessor("pcm-recorder-processor", PCMProcessor);

演示实现：app.js:865-874

// 音频记录器处理程序 - 为每个音频块调用
function audioRecorderHandler(pcmData) {
    if (websocket && websocket.readyState === WebSocket.OPEN && is_audio) {
        // 将音频作为二进制 WebSocket 帧发送（比 base64 JSON 更高效）
        websocket.send(pcmData);
        console.log("[CLIENT TO AGENT] Sent audio chunk: %s bytes", pcmData.byteLength);
    }
}

关键实现细节：

16kHz 采样率：必须使用 sampleRate: 16000 创建 AudioContext 以匹配 Live API 要求。现代浏览器支持此速率。
单声道音频：请求单通道音频 (channelCount: 1)，因为 Live API 期望单声道输入。这减少了带宽和处理开销。
AudioWorklet 处理：AudioWorklet 在与主 JavaScript 线程分开的线程上运行，确保低延迟、无故障的音频处理，而不会阻塞 UI。
Float32 到 PCM16 转换：Web Audio API 将音频作为范围 [-1.0, 1.0] 内的 Float32Array 值提供。乘以 32767 (0x7fff) 以转换为 16 位有符号整数 PCM。
二进制 WebSocket 帧：通过 WebSocket 二进制帧直接发送 PCM 数据作为 ArrayBuffer，而不是在 JSON 中进行 base64 编码。这将带宽减少了约 33%，并消除了编码/解码开销。
连续流式传输：AudioWorklet process() 方法以定期间隔自动调用（通常对于 16kHz 一次 128 个样本）。这为流式传输提供了一致的块大小。

此架构确保低延迟音频捕获和高效传输到服务器，然后服务器通过 LiveRequestQueue.send_realtime() 将其转发到 ADK Live API。

接收音频输出¶

当配置了 response_modalities=["AUDIO"] 时，模型在事件流中作为 inline_data 部分返回音频数据。

音频格式要求：

模型以以下格式输出音频：

格式：16 位 PCM（有符号整数）
采样率：24,000 Hz (24kHz) 用于原生音频模型
声道：单声道（单通道）
MIME 类型：audio/pcm;rate=24000

音频数据作为原始 PCM 字节到达，可用于播放或进一步处理。除非你需要不同的采样率或格式，否则不需要额外的转换。

接收音频输出：

from google.adk.agents.run_config import RunConfig, StreamingMode

# 配置音频输出
run_config = RunConfig(
    response_modalities=["AUDIO"],  # 音频响应所需
    streaming_mode=StreamingMode.BIDI
)

# 处理来自模型的音频输出
async for event in runner.run_live(
    user_id="user_123",
    session_id="session_456",
    live_request_queue=live_request_queue,
    run_config=run_config
):
    # 事件可能包含多个部分（文本、音频等）
    if event.content and event.content.parts:
        for part in event.content.parts:
            # 音频数据作为具有 audio/pcm MIME 类型的 inline_data 到达
            if part.inline_data and part.inline_data.mime_type.startswith("audio/pcm"):
                # 数据已经解码为原始字节（24kHz，16 位 PCM，单声道）
                audio_bytes = part.inline_data.data

                # 将音频流式传输到客户端的逻辑
                await stream_audio_to_client(audio_bytes)

                # 或保存到文件
                # with open("output.pcm", "ab") as f:
                #     f.write(audio_bytes)

自动 Base64 解码

Live API 传输协议将音频数据作为 base64 编码的字符串传输。google.genai 类型系统使用 Pydantic 的 base64 序列化功能 (val_json_bytes='base64') 在反序列化 API 响应时自动将 base64 字符串解码为字节。当你访问 part.inline_data.data 时，你收到的是即用型字节——无需手动 base64 解码。

在客户端处理音频事件¶

bidi-demo 使用不同的架构方法：它不处理服务器上的音频，而是将所有事件（包括音频数据）转发到 WebSocket 客户端，并在浏览器中处理音频播放。此模式分离了关注点——服务器专注于 ADK 事件流式传输，而客户端使用 Web Audio API 处理媒体播放。

演示实现：main.py:182-190

# bidi-demo 将所有事件（包括音频）转发到 WebSocket 客户端
async for event in runner.run_live(
    user_id=user_id,
    session_id=session_id,
    live_request_queue=live_request_queue,
    run_config=run_config
):
    event_json = event.model_dump_json(exclude_none=True, by_alias=True)
    await websocket.send_text(event_json)

演示实现（客户端 - JavaScript）：

客户端实现涉及三个组件：WebSocket 消息处理、使用 AudioWorklet 的音频播放器设置以及 AudioWorklet 处理器本身。

演示实现：app.js:544-673

// 1. WebSocket 消息处理程序
// 处理内容事件（文本或音频）
if (adkEvent.content && adkEvent.content.parts) {
    const parts = adkEvent.content.parts;

    for (const part of parts) {
        // 处理内联数据（音频）
        if (part.inlineData) {
            const mimeType = part.inlineData.mimeType;
            const data = part.inlineData.data;

            // 检查这是否是音频 PCM 数据且音频播放器已准备就绪
            if (mimeType && mimeType.startsWith("audio/pcm") && audioPlayerNode) {
                // 将 base64 解码为 ArrayBuffer 并发送到 AudioWorklet 进行播放
                audioPlayerNode.port.postMessage(base64ToArray(data));
            }
        }
    }
}

// 将 base64 音频数据解码为 ArrayBuffer
function base64ToArray(base64) {
    // 将 base64url 转换为标准 base64（符合 RFC 4648）
    // base64url 使用 '-' 和 '_' 代替 '+' 和 '/'，这对于 URL 是安全的
    let standardBase64 = base64.replace(/-/g, '+').replace(/_/g, '/');

    // 如果需要，添加填充 '=' 字符
    // Base64 字符串必须是 4 个字符的倍数
    while (standardBase64.length % 4) {
        standardBase64 += '=';
    }

    // 使用浏览器 API 将 base64 字符串解码为二进制字符串
    const binaryString = window.atob(standardBase64);
    const len = binaryString.length;
    const bytes = new Uint8Array(len);
    // 将每个字符代码 (0-255) 转换为字节
    for (let i = 0; i < len; i++) {
        bytes[i] = binaryString.charCodeAt(i);
    }
    // 返回底层 ArrayBuffer（二进制数据）
    return bytes.buffer;
}

演示实现：audio-player.js:5-24

// 2. 音频播放器设置
// 启动音频播放器 worklet
export async function startAudioPlayerWorklet() {
    // 创建具有 24kHz 采样率的 AudioContext
    // 这与 Live API 的输出音频格式（16 位 PCM @ 24kHz）相匹配
    // 注意：与输入速率 (16kHz) 不同 - Live API 以更高质量输出
    const audioContext = new AudioContext({
        sampleRate: 24000
    });

    // 加载将处理音频播放的 AudioWorklet 模块
    // AudioWorklet 在音频渲染线程上运行以实现流畅、低延迟的播放
    const workletURL = new URL('./pcm-player-processor.js', import.meta.url);
    await audioContext.audioWorklet.addModule(workletURL);

    // 使用我们自定义 PCM 播放器处理器创建一个 AudioWorkletNode
    // 此节点将通过 postMessage 接收音频数据并通过扬声器播放
    const audioPlayerNode = new AudioWorkletNode(audioContext, 'pcm-player-processor');

    // 将播放器节点连接到音频目的地（扬声器/耳机）
    // 这建立了音频图：AudioWorklet → AudioContext.destination
    audioPlayerNode.connect(audioContext.destination);

    return [audioPlayerNode, audioContext];
}

演示实现：pcm-player-processor.js:5-76

// 3. AudioWorklet 处理器（环形缓冲区）
// 缓冲和播放 PCM 音频的 AudioWorklet 处理器
class PCMPlayerProcessor extends AudioWorkletProcessor {
    constructor() {
        super();

        // 初始化环形缓冲区（24kHz x 180 秒 = ~430 万个样本）
        // 环形缓冲区吸收网络抖动并确保流畅播放
        this.bufferSize = 24000 * 180;
        this.buffer = new Float32Array(this.bufferSize);
        this.writeIndex = 0;  // 我们写入新音频数据的位置
        this.readIndex = 0;   // 我们读取以进行播放的位置

        // 处理来自主线程的传入消息
        this.port.onmessage = (event) => {
            // 在中断时重置缓冲区（例如，用户打断模型响应）
            if (event.data.command === 'endOfAudio') {
                this.readIndex = this.writeIndex; // 通过将读取跳转到写入位置来清除缓冲区
                return;
            }

            // 从传入的 ArrayBuffer 解码 Int16 数组
            // Live API 发送 16 位 PCM 音频数据
            const int16Samples = new Int16Array(event.data);

            // 将音频数据添加到环形缓冲区以进行播放
            this._enqueue(int16Samples);
        };
    }

    // 将传入的 Int16 数据推送到环形缓冲区
    _enqueue(int16Samples) {
        for (let i = 0; i < int16Samples.length; i++) {
            // 将 16 位整数转换为 Web Audio API 要求的 [-1.0, 1.0] 中的浮点数
            // 除以 32768（有符号 16 位整数的最大正值）
            const floatVal = int16Samples[i] / 32768;

            // 存储在当前写入位置的环形缓冲区中
            this.buffer[this.writeIndex] = floatVal;
            // 向前移动写入索引，在缓冲区末尾回绕（循环缓冲区）
            this.writeIndex = (this.writeIndex + 1) % this.bufferSize;

            // 溢出处理：如果写入赶上读取，则向前移动读取
            // 这会覆盖最旧的未播放样本（很少见，仅在极端网络延迟下）
            if (this.writeIndex === this.readIndex) {
                this.readIndex = (this.readIndex + 1) % this.bufferSize;
            }
        }
    }

    // 由 Web Audio 系统自动调用，每次约 128 个样本
    // 这在音频渲染线程上运行以实现精确计时
    process(inputs, outputs, parameters) {
        const output = outputs[0];
        const framesPerBlock = output[0].length;

        for (let frame = 0; frame < framesPerBlock; frame++) {
            // 将样本写入输出缓冲区（单声道到立体声）
            output[0][frame] = this.buffer[this.readIndex]; // 左声道
            if (output.length > 1) {
                output[1][frame] = this.buffer[this.readIndex]; // 右声道（立体声复制）
            }

            // 除非缓冲区为空，否则向前移动读取索引（下溢保护）
            if (this.readIndex != this.writeIndex) {
                this.readIndex = (this.readIndex + 1) % this.bufferSize;
            }
            // 如果 readIndex == writeIndex，我们没有数据了 - 输出静音 (0.0)
        }

        return true; // 保持处理器活动（返回 false 以终止）
    }
}

registerProcessor('pcm-player-processor', PCMPlayerProcessor);

关键实现模式：

Base64 解码：服务器以 JSON 中的 base64 编码字符串发送音频数据。客户端必须在传递给 AudioWorklet 之前解码为 ArrayBuffer。处理标准 base64 和 base64url 编码。
24kHz 采样率：必须使用 sampleRate: 24000 创建 AudioContext 以匹配 Live API 输出格式（与 16kHz 输入不同）。
环形缓冲区架构：使用循环缓冲区处理可变网络延迟并确保流畅播放。缓冲区存储 Float32 样本并通过覆盖最旧数据来处理溢出。
PCM16 到 Float32 转换：Live API 发送 16 位有符号整数。除以 32768 以转换为 Web Audio API 要求的范围 [-1.0, 1.0] 内的 Float32。
单声道到立体声：处理器将单声道音频复制到左声道和右声道以进行立体声输出，确保与所有音频设备的兼容性。
中断处理：在中断事件上，发送 endOfAudio 命令以通过设置 readIndex = writeIndex 清除缓冲区，防止播放陈旧音频。

此架构确保流畅、低延迟的音频播放，同时优雅地处理网络抖动和中断。

如何使用图像和视频¶

ADK 双向流式处理中的图像和视频都作为 JPEG 帧处理。ADK 不使用 HLS、mp4 或 H.264 等典型视频流式传输，而是使用简单的逐帧图像处理方法，其中静态图像和视频帧都作为单独的 JPEG 图像发送。

图像/视频规范：

格式：JPEG (image/jpeg)
帧率：建议最大 1 帧每秒 (1 FPS)
分辨率：768x768 像素（推荐）

演示实现：main.py:161-176

# 解码 base64 图像数据
image_data = base64.b64decode(json_message["data"])
mime_type = json_message.get("mimeType", "image/jpeg")

# 将图像作为 blob 发送
image_blob = types.Blob(
    mime_type=mime_type,
    data=image_data
)
live_request_queue.send_realtime(image_blob)

不适用于：

实时视频动作识别 - 1 FPS 太慢，无法捕获快速移动或动作
现场体育分析或运动跟踪 - 对于快速移动的主体来说，时间分辨率不足

图像处理的示例用例：

在 Shopper's Concierge 演示中，应用程序使用 send_realtime() 发送用户上传的图像。智能体从图像中识别上下文并在电子商务网站上搜索相关商品。

在客户端处理图像输入¶

在基于浏览器的应用程序中，从用户的网络摄像头捕获图像并将其发送到服务器需要使用 MediaDevices API 访问摄像头，将帧捕获到画布，并转换为 JPEG 格式。bidi-demo 演示了如何打开摄像头预览模态框，捕获单个帧，并将其作为 base64 编码的 JPEG 发送到 WebSocket 服务器。

架构：

摄像头访问：使用 navigator.mediaDevices.getUserMedia() 访问网络摄像头
视频预览：在 <video> 元素中显示实时摄像头源
帧捕获：将视频帧绘制到 <canvas> 并转换为 JPEG
Base64 编码：将画布转换为 base64 数据 URL 以进行传输
WebSocket 传输：作为 JSON 消息发送到服务器

演示实现：app.js:689-731

// 1. 打开摄像头预览
// 打开摄像头模态框并开始预览
async function openCameraPreview() {
    try {
        // 请求访问具有 768x768 分辨率的用户网络摄像头
        cameraStream = await navigator.mediaDevices.getUserMedia({
            video: {
                width: { ideal: 768 },
                height: { ideal: 768 },
                facingMode: 'user'
            }
        });

        // 将流设置为视频元素
        cameraPreview.srcObject = cameraStream;

        // 显示模态框
        cameraModal.classList.add('show');

    } catch (error) {
        console.error('Error accessing camera:', error);
        addSystemMessage(`Failed to access camera: ${error.message}`);
    }
}

// 关闭摄像头模态框并停止预览
function closeCameraPreview() {
    // 停止摄像头流
    if (cameraStream) {
        cameraStream.getTracks().forEach(track => track.stop());
        cameraStream = null;
    }

    // 清除视频源
    cameraPreview.srcObject = null;

    // 隐藏模态框
    cameraModal.classList.remove('show');
}

演示实现：app.js:734-802

// 2. 捕获并发送图像
// 从实时预览捕获图像
function captureImageFromPreview() {
    if (!cameraStream) {
        addSystemMessage('No camera stream available');
        return;
    }

    try {
        // 创建画布以捕获帧
        const canvas = document.createElement('canvas');
        canvas.width = cameraPreview.videoWidth;
        canvas.height = cameraPreview.videoHeight;
        const context = canvas.getContext('2d');

        // 将当前视频帧绘制到画布
        context.drawImage(cameraPreview, 0, 0, canvas.width, canvas.height);

        // 将画布转换为数据 URL 以进行显示
        const imageDataUrl = canvas.toDataURL('image/jpeg', 0.85);

        // 在聊天中显示捕获的图像
        const imageBubble = createImageBubble(imageDataUrl, true);
        messagesDiv.appendChild(imageBubble);

        // 将画布转换为 blob 以发送到服务器
        canvas.toBlob((blob) => {
            // 将 blob 转换为 base64 以发送到服务器
            const reader = new FileReader();
            reader.onloadend = () => {
                // 删除 data:image/jpeg;base64, 前缀
                const base64data = reader.result.split(',')[1];
                sendImage(base64data);
            };
            reader.readAsDataURL(blob);
        }, 'image/jpeg', 0.85);

        // 关闭摄像头模态框
        closeCameraPreview();

    } catch (error) {
        console.error('Error capturing image:', error);
        addSystemMessage(`Failed to capture image: ${error.message}`);
    }
}

// 将图像发送到服务器
function sendImage(base64Image) {
    if (websocket && websocket.readyState === WebSocket.OPEN) {
        const jsonMessage = JSON.stringify({
            type: "image",
            data: base64Image,
            mimeType: "image/jpeg"
        });
        websocket.send(jsonMessage);
        console.log("[CLIENT TO AGENT] Sent image");
    }
}

关键实现细节：

768x768 分辨率：请求 768x768 的理想分辨率以匹配推荐规范。浏览器将提供最接近的可用分辨率。
面向用户的摄像头：facingMode: 'user' 约束选择移动设备上的前置摄像头，适合自拍捕获。
画布帧捕获：使用 canvas.getContext('2d').drawImage() 从实时视频流捕获单个帧。这将创建当前视频帧的静态快照。
JPEG 压缩：toDataURL() 和 toBlob() 的第二个参数是质量（0.0 到 1.0）。使用 0.85 提供良好的质量，同时保持文件大小可控。
双重输出：代码创建用于立即 UI 显示的数据 URL 和用于高效 base64 编码的 blob，演示了响应式用户反馈的模式。
资源清理：关闭摄像头时始终调用 getTracks().forEach(track => track.stop()) 以释放硬件资源并关闭摄像头指示灯。
Base64 编码：FileReader 将 blob 转换为数据 URL (data:image/jpeg;base64,<data>)。在逗号处拆分并获取第二部分，以仅获取没有前缀的 base64 数据。

此实现提供了一个用户友好的摄像头界面，具有预览、单帧捕获和高效传输到服务器以供 Live API 处理的功能。

自定义视频流式传输工具支持¶

ADK 提供了特殊的工具支持，用于在流式传输会话期间处理视频帧。与同步执行的常规工具不同，流式传输工具可以在模型继续生成响应的同时异步生成视频帧。

流式传输工具生命周期：

开始：当模型调用它时，ADK 调用你的异步生成器函数
流式传输：你的函数通过 AsyncGenerator 连续生成结果
停止：ADK 在以下情况下取消生成器任务：
模型调用你提供的 stop_streaming() 函数
会话结束
发生错误

重要：你必须提供一个 stop_streaming(function_name: str) 函数作为工具，以允许模型显式停止流式传输操作。

要实现处理并向模型生成视频帧的自定义视频流式传输工具，请参阅流式传输工具文档。

了解音频模型架构¶

在使用 Live API 构建语音应用程序时，最重要的决定之一是选择正确的音频模型架构。Live API 支持两种根本不同类型的音频处理模型：原生音频和半级联。这些模型架构在处理音频输入和生成音频输出的方式上有所不同，这直接影响响应的自然度、工具执行的可靠性、延迟特性和整体用例的适用性。

了解这些架构有助于你根据应用程序的要求做出明智的模型选择决策——无论你是优先考虑自然对话 AI、生产可靠性还是特定功能的可用性。

原生音频模型¶

一个完全集成的端到端音频模型架构，其中模型直接处理音频输入并生成音频输出，而无需中间文本转换。这种方法可以实现更像人类的语音，具有自然的韵律。

音频模型架构	平台	模型	备注
原生音频	Gemini Live API	gemini-2.5-flash-native-audio-preview-09-2025	公开可用
原生音频	Vertex AI Live API	gemini-live-2.5-flash-preview-native-audio-09-2025	公开预览

关键特性：

端到端音频处理：直接处理音频输入并生成音频输出，无需中间转换为文本
自然韵律：产生更像人类的语音模式、语调和情感表达
扩展语音库：支持所有半级联语音加上来自文本转语音 (TTS) 服务的其他语音
自动语言检测：从对话上下文中确定语言，无需显式配置
高级对话功能：
情感对话：根据输入表达和语气调整响应风格，检测情感线索
主动音频：可以主动决定何时响应、提供建议或忽略不相关的输入
动态思维：支持思维总结和动态思维预算
仅音频响应模态：不支持带有 RunConfig 的 TEXT 响应模态，导致初始响应时间较慢

半级联模型¶

一种混合架构，结合了原生音频输入处理和文本转语音 (TTS) 输出生成。在某些文档中也称为“级联”模型。

音频输入被原生处理，但响应首先生成为文本，然后转换为语音。这种分离在生产环境中提供了更好的可靠性和更稳健的工具执行。

音频模型架构	平台	模型	备注
半级联	Gemini Live API	gemini-2.0-flash-live-001	将于 2025 年 12 月 09 日弃用
半级联	Vertex AI Live API	gemini-live-2.5-flash	私有 GA，不公开可用

关键特性：

混合架构：结合了原生音频输入处理和基于 TTS 的音频输出生成
TEXT 响应模态支持：除了 AUDIO 之外，还支持带有 RunConfig 的 TEXT 响应模态，从而为纯文本用例实现更快的响应
显式语言控制：支持通过 speech_config.language_code 进行手动语言代码配置
既定的 TTS 质量：利用经过验证的文本转语音技术实现一致的音频输出
支持的语音：Puck, Charon, Kore, Fenrir, Aoede, Leda, Orus, Zephyr（8 种预构建语音）

如何处理模型名称¶

在构建 ADK 应用程序时，你需要指定要使用的模型。推荐的方法是使用环境变量进行模型配置，这随着模型可用性和命名的变化提供了灵活性。

推荐模式：

import os
from google.adk.agents import Agent

# 使用环境变量，并回退到合理的默认值
agent = Agent(
    name="my_agent",
    model=os.getenv("DEMO_AGENT_MODEL", "gemini-2.5-flash-native-audio-preview-09-2025"),
    tools=[...],
    instruction="..."
)

为什么要使用环境变量：

模型可用性变化：模型会定期发布、更新和弃用（例如，gemini-2.0-flash-live-001 将于 2025 年 12 月 09 日弃用）
平台特定名称：Gemini Live API 和 Vertex AI Live API 对相同的功能使用不同的模型命名约定
轻松切换：通过更新 .env 文件更改模型而无需修改代码
环境特定配置：为开发、暂存和生产使用不同的模型

.env 文件中的配置：

# 用于 Gemini Live API（公开可用）
DEMO_AGENT_MODEL=gemini-2.5-flash-native-audio-preview-09-2025

# 用于 Vertex AI Live API（如果使用 Vertex AI）
# DEMO_AGENT_MODEL=gemini-live-2.5-flash-preview-native-audio-09-2025

选择正确的模型：

选择平台：在 Gemini Live API（公共）或 Vertex AI Live API（企业）之间做出决定
选择架构：
原生音频用于具有高级功能的自然对话 AI
半级联用于具有工具执行的生产可靠性
检查当前可用性：参考上面的模型表和官方文档
配置环境变量：在你的 .env 文件中设置 DEMO_AGENT_MODEL（参见 agent.py:11-16 和 main.py:83-96）

Live API 模型兼容性和可用性¶

有关 Live API 模型兼容性和可用性的最新信息：

Gemini Live API 模型：请参阅 Gemini 模型文档
Vertex AI Live API 模型：请参阅 Vertex AI 模型文档

在部署到生产环境之前，请务必在官方文档中验证模型可用性和功能支持。

音频转录¶

Live API 提供内置的音频转录功能，可自动将语音转换为文本，用于用户输入和模型输出。这消除了对外部转录服务的需求，并实现了实时字幕、对话记录和辅助功能。ADK 通过 RunConfig 公开了这些功能，允许你为任一或两个音频方向启用转录。

来源

Gemini Live API - 音频转录

配置：

from google.genai import types
from google.adk.agents.run_config import RunConfig

# 默认行为：音频转录默认启用
# 输入和输出转录均自动配置
run_config = RunConfig(
    response_modalities=["AUDIO"]
    # input_audio_transcription 默认为 AudioTranscriptionConfig()
    # output_audio_transcription 默认为 AudioTranscriptionConfig()
)

# 显式禁用转录：
run_config = RunConfig(
    response_modalities=["AUDIO"],
    input_audio_transcription=None,   # 显式禁用用户输入转录
    output_audio_transcription=None   # 显式禁用模型输出转录
)

# 仅启用输入转录（禁用输出）：
run_config = RunConfig(
    response_modalities=["AUDIO"],
    input_audio_transcription=types.AudioTranscriptionConfig(),  # 显式启用（与默认冗余）
    output_audio_transcription=None  # 显式禁用
)

# 仅启用输出转录（禁用输入）：
run_config = RunConfig(
    response_modalities=["AUDIO"],
    input_audio_transcription=None,  # 显式禁用
    output_audio_transcription=types.AudioTranscriptionConfig()  # 显式启用（与默认冗余）
)

事件结构：

转录作为 Event 对象上的 types.Transcription 对象传递：

from dataclasses import dataclass
from typing import Optional
from google.genai import types

@dataclass
class Event:
    content: Optional[Content]  # 音频/文本内容
    input_transcription: Optional[types.Transcription]  # 用户语音 → 文本
    output_transcription: Optional[types.Transcription]  # 模型语音 → 文本
    # ... 其他字段

了解更多

有关完整的事件结构，请参阅第 3 部分：Event 类。

每个 Transcription 对象有两个属性： - .text：转录文本（字符串） - .finished：指示转录是完成 (True) 还是部分 (False) 的布尔值

转录如何传递：

转录作为事件流中的单独字段到达，而不是作为内容部分。访问转录数据时始终使用防御性空值检查：

处理转录：

from google.adk.runners import Runner

# ... 运行器设置代码 ...

async for event in runner.run_live(...):
    # 用户语音转录（来自输入音频）
    if event.input_transcription:  # 第一次检查：转录对象存在
        # 访问转录文本和状态
        user_text = event.input_transcription.text
        is_finished = event.input_transcription.finished

        # 第二次检查：文本不是 None 或空
        # 这处理转录正在进行或为空的情况
        if user_text and user_text.strip():
            print(f"User said: {user_text} (finished: {is_finished})")

            # 你的字幕更新逻辑
            update_caption(user_text, is_user=True, is_final=is_finished)

    # 模型语音转录（来自输出音频）
    if event.output_transcription:  # 第一次检查：转录对象存在
        model_text = event.output_transcription.text
        is_finished = event.output_transcription.finished

        # 第二次检查：文本不是 None 或空
        # 这处理转录正在进行或为空的情况
        if model_text and model_text.strip():
            print(f"Model said: {model_text} (finished: {is_finished})")

            # 你的字幕更新逻辑
            update_caption(model_text, is_user=False, is_final=is_finished)

转录空值检查的最佳实践

始终对转录使用两级空值检查：

检查转录对象是否存在 (if event.input_transcription)
检查文本是否不为空 (if user_text and user_text.strip())

此模式可防止 None 值导致的错误，并处理可能为空的部分转录。

在客户端处理音频转录¶

在 Web 应用程序中，转录事件需要从服务器转发到浏览器并在 UI 中呈现。bidi-demo 演示了一种模式，其中服务器将所有 ADK 事件（包括转录事件）转发到 WebSocket 客户端，客户端处理将转录显示为带有部分与完成转录视觉指示器的语音气泡。

架构：

服务器端：通过 WebSocket 转发转录事件（已在上一节中显示）
客户端：处理来自 WebSocket 的 inputTranscription 和 outputTranscription 事件
UI 渲染：显示带有输入指示器的部分转录，当 finished: true 时完成

演示实现：app.js:438-525

// 处理输入转录（用户所说的话）
if (adkEvent.inputTranscription && adkEvent.inputTranscription.text) {
    const transcriptionText = adkEvent.inputTranscription.text;
    const isFinished = adkEvent.inputTranscription.finished;

    if (transcriptionText) {
        if (currentInputTranscriptionId == null) {
            // 创建新的转录气泡
            currentInputTranscriptionId = Math.random().toString(36).substring(7);
            currentInputTranscriptionElement = createMessageBubble(
                transcriptionText,
                true,  // isUser
                !isFinished  // isPartial
            );
            currentInputTranscriptionElement.id = currentInputTranscriptionId;
            currentInputTranscriptionElement.classList.add("transcription");
            messagesDiv.appendChild(currentInputTranscriptionElement);
        } else {
            // 更新现有的转录气泡
            if (currentOutputTranscriptionId == null && currentMessageId == null) {
                // 累积输入转录文本（Live API 发送增量片段）
                const existingText = currentInputTranscriptionElement
                    .querySelector(".bubble-text").textContent;
                const cleanText = existingText.replace(/\.\.\.$/, '');
                const accumulatedText = cleanText + transcriptionText;
                updateMessageBubble(
                    currentInputTranscriptionElement,
                    accumulatedText,
                    !isFinished
                );
            }
        }

        // 如果转录完成，重置状态
        if (isFinished) {
            currentInputTranscriptionId = null;
            currentInputTranscriptionElement = null;
        }
    }
}

// 处理输出转录（模型所说的话）
if (adkEvent.outputTranscription && adkEvent.outputTranscription.text) {
    const transcriptionText = adkEvent.outputTranscription.text;
    const isFinished = adkEvent.outputTranscription.finished;

    if (transcriptionText) {
        // 当模型开始响应时完成任何活动的输入转录
        if (currentInputTranscriptionId != null && currentOutputTranscriptionId == null) {
            const textElement = currentInputTranscriptionElement
                .querySelector(".bubble-text");
            const typingIndicator = textElement.querySelector(".typing-indicator");
            if (typingIndicator) {
                typingIndicator.remove();
            }
            currentInputTranscriptionId = null;
            currentInputTranscriptionElement = null;
        }

        if (currentOutputTranscriptionId == null) {
            // 为模型创建新的转录气泡
            currentOutputTranscriptionId = Math.random().toString(36).substring(7);
            currentOutputTranscriptionElement = createMessageBubble(
                transcriptionText,
                false,  // isUser
                !isFinished  // isPartial
            );
            currentOutputTranscriptionElement.id = currentOutputTranscriptionId;
            currentOutputTranscriptionElement.classList.add("transcription");
            messagesDiv.appendChild(currentOutputTranscriptionElement);
        } else {
            // 更新现有的转录气泡
            const existingText = currentOutputTranscriptionElement
                .querySelector(".bubble-text").textContent;
            const cleanText = existingText.replace(/\.\.\.$/, '');
            updateMessageBubble(
                currentOutputTranscriptionElement,
                cleanText + transcriptionText,
                !isFinished
            );
        }

        // 如果转录完成，重置状态
        if (isFinished) {
            currentOutputTranscriptionId = null;
            currentOutputTranscriptionElement = null;
        }
    }
}

关键实现模式：

增量文本累积：Live API 可能会分多个块发送转录。通过将新片段附加到现有内容来累积文本：
```
const accumulatedText = cleanText + transcriptionText;
```
部分与完成状态：使用 finished 标志来确定是否显示输入指示器：
finished: false → 显示输入指示器（例如，“...”）
finished: true → 移除输入指示器，完成气泡
气泡状态管理：使用 ID 分别跟踪输入和输出的当前转录气泡。仅在开始新的转录时创建新气泡：
```
if (currentInputTranscriptionId == null) {
    // 创建新气泡
} else {
    // 更新现有气泡
}
```
轮次协调：当模型开始响应（第一个输出转录到达）时，完成任何活动的输入转录以防止重叠更新。

此模式确保流畅的实时转录显示，并正确处理流式传输更新、轮次转换和用户视觉反馈。

多智能体转录要求¶

对于多智能体场景（具有 sub_agents 的智能体），无论你的 RunConfig 设置如何，ADK 都会自动启用音频转录。这种自动行为是智能体转移功能所必需的，其中文本转录用于在智能体之间传递对话上下文。

自动启用行为：

当智能体定义了 sub_agents 时，ADK 的 run_live() 方法会自动启用输入和输出音频转录，即使你显式将其设置为 None。这通过为下一个智能体提供文本上下文来确保智能体转移正常工作。

为什么这很重要：

无法禁用：你无法在多智能体场景中关闭转录
功能必需：没有文本上下文，智能体转移将中断
对开发者透明：转录事件自动可用
数据处理计划：你的应用程序将收到必须处理的转录事件

实现细节：

当满足以下两个条件时，自动启用发生在 Runner.run_live() 中： - 智能体定义了 sub_agents - 提供了 LiveRequestQueue（双向流式处理模式）

来源

runners.py:1236-1253

语音配置 (Speech Config)¶

Live API 提供语音配置功能，允许你自定义模型在生成音频响应时的声音。ADK 支持两个级别的语音配置：智能体级别（每个智能体的语音设置）和会话级别（通过 RunConfig 的全局语音设置）。这实现了复杂的多智能体场景，其中不同的智能体可以用不同的声音说话，以及具有一致语音特征的单智能体应用程序。

来源

Gemini Live API - 功能指南

智能体级别配置¶

你可以通过创建具有语音设置的自定义 Gemini LLM 实例，然后将该实例传递给 Agent，从而在每个智能体的基础上配置 speech_config。这在多智能体工作流中特别有用，其中不同的智能体代表不同的角色或人物。

配置：

from google.genai import types
from google.adk.agents import Agent
from google.adk.models.google_llm import Gemini
from google.adk.tools import google_search

# 创建具有自定义语音配置的 Gemini 实例
custom_llm = Gemini(
    model="gemini-2.5-flash-native-audio-preview-09-2025",
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Puck"
            )
        ),
        language_code="en-US"
    )
)

# 将 Gemini 实例传递给智能体
agent = Agent(
    model=custom_llm,
    tools=[google_search],
    instruction="You are a helpful assistant."
)

RunConfig 级别配置¶

你还可以在 RunConfig 中设置 speech_config，以便为会话中的所有智能体应用默认语音配置。这对于单智能体应用程序或当你希望所有智能体具有一致的声音时非常有用。

配置：

from google.genai import types
from google.adk.agents.run_config import RunConfig

run_config = RunConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Kore"
            )
        ),
        language_code="en-US"
    )
)

配置优先级¶

当同时提供智能体级别（通过 Gemini 实例）和会话级别（通过 RunConfig）的 speech_config 时，智能体级别配置优先。这允许你在 RunConfig 中设置默认语音，同时为特定智能体覆盖它。

优先级规则：

Gemini 实例具有 speech_config：使用 Gemini 的语音配置（最高优先级）
RunConfig 具有 speech_config：使用 RunConfig 的语音配置
均未指定：使用 Live API 默认语音（最低优先级）

示例：

from google.genai import types
from google.adk.agents import Agent
from google.adk.models.google_llm import Gemini
from google.adk.agents.run_config import RunConfig
from google.adk.tools import google_search

# 创建具有自定义语音的 Gemini 实例
custom_llm = Gemini(
    model="gemini-2.5-flash-native-audio-preview-09-2025",
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Puck"  # 智能体级别：最高优先级
            )
        )
    )
)

# 智能体使用具有自定义语音的 Gemini 实例
agent = Agent(
    model=custom_llm,
    tools=[google_search],
    instruction="You are a helpful assistant."
)

# 具有默认语音的 RunConfig（将被上述智能体的 Gemini 配置覆盖）
run_config = RunConfig(
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Kore"  # 这对于上面的智能体被覆盖
            )
        )
    )
)

多智能体语音配置¶

对于多智能体工作流，你可以通过创建具有不同 speech_config 值的单独 Gemini 实例，为不同的智能体分配不同的声音。这创造了更自然和可区分的对话，其中每个智能体都有自己的声音个性。

多智能体示例：

from google.genai import types
from google.adk.agents import Agent
from google.adk.models.google_llm import Gemini
from google.adk.agents.run_config import RunConfig

# 具有友好声音的客户服务智能体
customer_service_llm = Gemini(
    model="gemini-2.5-flash-native-audio-preview-09-2025",
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Aoede"  # 友好、温暖的声音
            )
        )
    )
)

customer_service_agent = Agent(
    name="customer_service",
    model=customer_service_llm,
    instruction="You are a friendly customer service representative."
)

# 具有专业声音的技术支持智能体
technical_support_llm = Gemini(
    model="gemini-2.5-flash-native-audio-preview-09-2025",
    speech_config=types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(
                voice_name="Charon"  # 专业、权威的声音
            )
        )
    )
)

technical_support_agent = Agent(
    name="technical_support",
    model=technical_support_llm,
    instruction="You are a technical support specialist."
)

# 协调工作流的根智能体
root_agent = Agent(
    name="root_agent",
    model="gemini-2.5-flash-native-audio-preview-09-2025",
    instruction="Coordinate customer service and technical support.",
    sub_agents=[customer_service_agent, technical_support_agent]
)

# 没有 speech_config 的 RunConfig - 每个智能体使用自己的声音
run_config = RunConfig(
    response_modalities=["AUDIO"]
)

在此示例中，当客户服务智能体说话时，用户会听到 "Aoede" 的声音。当技术支持智能体接管时，用户会听到 "Charon" 的声音。这创造了更具吸引力和自然的多智能体体验。

配置参数¶

voice_config：指定用于音频生成的预构建语音 - 通过嵌套的 VoiceConfig 和 PrebuiltVoiceConfig 对象进行配置 - voice_name：预构建语音的字符串标识符（例如，"Kore", "Puck", "Charon"）

language_code：用于语音合成的 ISO 639 语言代码（例如，"en-US", "ja-JP"） - 确定合成语音的语言和区域口音 - 模型特定行为： - 半级联模型：使用指定的 language_code 进行 TTS 输出 - 原生音频模型：可能会忽略 language_code 并从对话上下文中自动确定语言。请咨询模型特定文档以获取支持。

可用语音¶

可用语音因模型架构而异。要验证特定模型可用的语音： - 查看 Gemini Live API 文档以获取完整列表 - 在部署到生产环境之前在开发中测试语音配置 - 如果不支持某种语音，Live API 将返回错误

半级联模型支持这些语音： - Puck - Charon - Kore - Fenrir - Aoede - Leda - Orus - Zephyr

原生音频模型支持扩展的语音列表，其中包括所有半级联语音加上来自文本转语音 (TTS) 服务的其他语音。有关原生音频模型支持的完整语音列表： - 请参阅 Gemini Live API 文档 - 或查看原生音频模型也支持的 Text-to-Speech 语音列表

与半级联模型相比，扩展的语音列表提供了更多语音特征、口音和语言选项。

平台可用性¶

两个平台都支持语音配置，但语音可用性可能有所不同：

Gemini Live API:

✅ 完全支持，具有记录的语音选项
✅ 半级联模型：8 种语音 (Puck, Charon, Kore, Fenrir, Aoede, Leda, Orus, Zephyr)
✅ 原生音频模型：扩展语音列表（参见文档）

Vertex AI Live API:

✅ 支持语音配置
⚠️ 平台特定差异：语音可用性可能与 Gemini Live API 不同
⚠️ 需要验证：查看 Vertex AI 文档以获取当前支持的语音列表

最佳实践：在开发期间始终在目标平台上测试你选择的语音配置。如果你的平台/模型组合不支持某种语音，Live API 将在连接时返回错误。

重要说明¶

模型兼容性：语音配置仅适用于具有音频输出功能的 Live API 模型
配置级别：你可以在智能体级别（通过 Gemini(speech_config=...)）或会话级别（RunConfig(speech_config=...)）设置 speech_config。智能体级别配置优先。
智能体级别用法：要为每个智能体配置语音，请创建一个具有 speech_config 的 Gemini 实例并将其传递给 Agent(model=gemini_instance)
默认行为：如果在任一级别均未指定 speech_config，Live API 将使用默认语音
原生音频模型：根据对话上下文自动确定语言；可能不支持显式 language_code
语音可用性：具体语音名称可能因模型而异；请参阅当前 Live API 文档以获取所选模型支持的语音

了解更多

有关完整的 RunConfig 参考，请参阅第 4 部分：了解 RunConfig。

语音活动检测 (VAD)¶

语音活动检测 (VAD) 是一项 Live API 功能，可自动检测用户何时开始和停止说话，从而实现自然的轮流对话而无需手动控制。VAD 在所有 Live API 模型上默认启用，允许模型根据检测到的语音活动自动管理对话轮次。

来源

Gemini Live API - 语音活动检测

VAD 如何工作¶

当启用 VAD（默认）时，Live API 会自动：

检测语音开始：识别用户何时开始说话
检测语音结束：识别用户何时停止说话（自然停顿）
管理轮流对话：允许模型在用户说完后做出响应
处理打断：通过来回交流实现自然的对话流

这创造了一种免提、自然的对话体验，用户无需手动发出正在说话或说完的信号。

何时禁用 VAD¶

你应该在以下场景中禁用自动 VAD：

一键通实现：你的应用程序手动控制何时发送音频（例如，嘈杂环境中的音频交互应用程序或具有串扰的房间）
客户端语音检测：你的应用程序使用客户端 VAD 向服务器发送活动信号，以减少连续音频流式传输带来的 CPU 和网络开销
特定的 UX 模式：你的设计要求用户手动指示何时说完

当你禁用 VAD（默认启用）时，你必须使用手动活动信号 (ActivityStart/ActivityEnd) 来控制对话轮次。有关手动轮次控制的详细信息，请参阅第 2 部分：活动信号。

VAD 配置¶

默认行为（启用 VAD，无需配置）：

from google.adk.agents.run_config import RunConfig

# VAD 默认启用 - 无需显式配置
run_config = RunConfig(
    response_modalities=["AUDIO"]
)

禁用自动 VAD（启用手动轮次控制）：

from google.genai import types
from google.adk.agents.run_config import RunConfig

run_config = RunConfig(
    response_modalities=["AUDIO"],
    realtime_input_config=types.RealtimeInputConfig(
        automatic_activity_detection=types.AutomaticActivityDetection(
            disabled=True  # 禁用自动 VAD
        )
    )
)

客户端 VAD 示例¶

在构建支持语音的应用程序时，你可能希望实现客户端语音活动检测 (VAD) 以减少 CPU 和网络开销。此模式将基于浏览器的 VAD 与手动活动信号相结合，以控制何时将音频发送到服务器。

架构：

客户端：浏览器使用 Web Audio API（具有基于 RMS 的 VAD 的 AudioWorklet）检测语音活动
信号协调：检测到语音时发送 activity_start，语音停止时发送 activity_end
音频流式传输：仅在活动语音期间发送音频块
服务器配置：禁用自动 VAD，因为客户端处理检测

服务器端配置¶

配置：

from fastapi import FastAPI, WebSocket
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.adk.agents.live_request_queue import LiveRequestQueue
from google.genai import types

# 配置 RunConfig 以禁用自动 VAD
run_config = RunConfig(
    streaming_mode=StreamingMode.BIDI,
    response_modalities=["AUDIO"],
    realtime_input_config=types.RealtimeInputConfig(
        automatic_activity_detection=types.AutomaticActivityDetection(
            disabled=True  # 客户端处理 VAD
        )
    )
)

WebSocket 上游任务¶

实现：

async def upstream_task(websocket: WebSocket, live_request_queue: LiveRequestQueue):
    """接收来自客户端的音频和活动信号。"""
    try:
        while True:
            # 从 WebSocket 接收 JSON 消息
            message = await websocket.receive_json()

            if message.get("type") == "activity_start":
                # 客户端检测到语音 - 向模型发送信号
                live_request_queue.send_activity_start()

            elif message.get("type") == "activity_end":
                # 客户端检测到静音 - 向模型发送信号
                live_request_queue.send_activity_end()

            elif message.get("type") == "audio":
                # 将音频块流式传输到模型
                import base64
                audio_data = base64.b64decode(message["data"])
                audio_blob = types.Blob(
                    mime_type="audio/pcm;rate=16000",
                    data=audio_data
                )
                live_request_queue.send_realtime(audio_blob)

    except WebSocketDisconnect:
        live_request_queue.close()

客户端 VAD 实现¶

实现：

// vad-processor.js - 用于语音检测的 AudioWorklet 处理器
class VADProcessor extends AudioWorkletProcessor {
    constructor() {
        super();
        this.threshold = 0.05;  // 根据环境调整
    }

    process(inputs, outputs, parameters) {
        const input = inputs[0];
        if (input && input.length > 0) {
            const channelData = input[0];
            let sum = 0;

            // 计算 RMS (均方根)
            for (let i = 0; i < channelData.length; i++) {
                sum += channelData[i] ** 2;
            }
            const rms = Math.sqrt(sum / channelData.length);

            // 发送语音检测状态信号
            this.port.postMessage({
                voice: rms > this.threshold,
                rms: rms
            });
        }
        return true;
    }
}
registerProcessor('vad-processor', VADProcessor);

客户端协调¶

协调 VAD 信号：

// 主应用程序逻辑
let isSilence = true;
let lastVoiceTime = 0;
const SILENCE_TIMEOUT = 2000;  // 发送 activity_end 之前的 2 秒静音

// 设置 VAD 处理器
const vadNode = new AudioWorkletNode(audioContext, 'vad-processor');
vadNode.port.onmessage = (event) => {
    const { voice, rms } = event.data;

    if (voice) {
        // 检测到语音
        if (isSilence) {
            // 从静音转换为语音 - 发送 activity_start
            websocket.send(JSON.stringify({ type: "activity_start" }));
            isSilence = false;
        }
        lastVoiceTime = Date.now();
    } else {
        // 未检测到语音 - 检查是否超过静音超时
        if (!isSilence && Date.now() - lastVoiceTime > SILENCE_TIMEOUT) {
            // 持续静音 - 发送 activity_end
            websocket.send(JSON.stringify({ type: "activity_end" }));
            isSilence = true;
        }
    }
};

// 设置音频记录器以流式传输块
audioRecorderNode.port.onmessage = (event) => {
    const audioData = event.data;  // Float32Array

    // 仅在检测到语音时发送音频
    if (!isSilence) {
        // 转换为 PCM16 并发送到服务器
        const pcm16 = convertFloat32ToPCM(audioData);
        const base64Audio = arrayBufferToBase64(pcm16);

        websocket.send(JSON.stringify({
            type: "audio",
            mime_type: "audio/pcm;rate=16000",
            data: base64Audio
        }));
    }
};

关键实现细节：

基于 RMS 的语音检测：AudioWorklet 处理器计算音频样本的均方根 (RMS) 以检测语音活动。RMS 提供了一种简单但有效的音频能量度量，可以区分语音和静音。
可调阈值：threshold 值（示例中为 0.05）可以根据环境进行调整。较低的阈值更敏感（检测较小的语音但可能会因背景噪音而触发），较高的阈值需要更大的语音。
静音超时：在发送 activity_end 之前使用超时（例如 2000 毫秒），以避免在语音自然停顿期间过早结束轮次。这创造了更自然的对话流。
状态管理：跟踪 isSilence 状态以检测静音和语音之间的转换。仅在静音→语音转换时发送 activity_start，并仅在持续静音后发送 activity_end。
条件音频流式传输：仅在 !isSilence 时发送音频块以减少带宽。根据对话的语音静音比，这可以节省约 50-90% 的网络流量。
AudioWorklet 线程分离：VAD 处理器在音频渲染线程上运行，确保实时性能不受主线程 JavaScript 执行或网络延迟的影响。

客户端 VAD 的好处¶

此模式提供了几个优点：

减少 CPU 和网络开销：仅在活动语音期间发送音频，而不是连续静音
更快的响应：无需服务器往返即可立即进行本地检测
更好的控制：根据客户端环境微调 VAD 灵敏度

活动信号时序

使用带有客户端 VAD 的手动活动信号时：

始终在发送第一个音频块之前发送 activity_start
始终在发送最后一个音频块之后发送 activity_end
模型将仅处理 activity_start 和 activity_end 信号之间的音频
不正确的时序可能会导致模型忽略音频或产生意外行为

主动性和情感对话¶

Live API 提供了高级对话功能，可实现更自然和上下文感知的交互。主动音频允许模型智能地决定何时响应、在没有显式提示的情况下提供建议或忽略不相关的输入。情感对话使模型能够检测并适应语音语调和内容中的情感线索，调整其响应风格以进行更具同理心的互动。这些功能目前仅在原生音频模型上受支持。

来源

Gemini Live API - 主动音频 | 情感对话

配置：

from google.genai import types
from google.adk.agents.run_config import RunConfig

run_config = RunConfig(
    # 模型可以在没有显式提示的情况下发起响应
    proactivity=types.ProactivityConfig(proactive_audio=True),

    # 模型适应用户情绪
    enable_affective_dialog=True
)

主动性：

启用后，模型可以：

在未被询问的情况下提供建议
主动提供后续信息
忽略不相关或离题的输入
根据上下文预测用户需求

情感对话：

模型分析语音语调和内容中的情感线索以：

检测用户情绪（沮丧、高兴、困惑等）
相应地调整响应风格和语气
在客户服务场景中提供同理心响应
根据检测到的情绪调整正式程度

实际示例 - 客户服务机器人：

from google.genai import types
from google.adk.agents.run_config import RunConfig, StreamingMode

# 配置同理心客户服务
run_config = RunConfig(
    response_modalities=["AUDIO"],
    streaming_mode=StreamingMode.BIDI,

    # 模型可以主动提供帮助
    proactivity=types.ProactivityConfig(proactive_audio=True),

    # 模型适应客户情绪
    enable_affective_dialog=True
)

# 示例交互（说明性 - 实际模型行为可能有所不同）：
# Customer: "I've been waiting for my order for three weeks..."
# [模型可能会检测到语气中的沮丧并调整响应]
# Model: "I'm really sorry to hear about this delay. Let me check your order
#        status right away. Can you provide your order number?"
#
# [主动性在行动]
# Model: "I see you previously asked about shipping updates. Would you like
#        me to set up notifications for future orders?"
#
# 注意：主动和情感行为是概率性的。模型的情感意识和主动建议将根据上下文、
# 对话历史记录和固有的模型变异性而有所不同。

平台兼容性¶

这些功能是模型特定的，并具有平台含义：

Gemini Live API:

✅ 在 gemini-2.5-flash-native-audio-preview-09-2025（原生音频模型）上受支持
❌ 在 gemini-live-2.5-flash-preview（半级联模型）上不受支持

Vertex AI Live API:

❌ 目前在 gemini-live-2.5-flash（半级联模型）上不受支持
⚠️ 平台特定差异：主动性和情感对话需要原生音频模型，目前仅在 Gemini Live API 上可用

关键见解：如果你的应用程序需要主动音频或情感对话功能，则必须使用带有原生音频模型的 Gemini Live API。任一平台上的半级联模型均不支持这些功能。

测试主动性：

要验证主动行为是否正常工作：

创建开放式上下文：提供信息而不提问

User: "I'm planning a trip to Japan next month."
Expected: Model offers suggestions, asks follow-up questions

测试情感响应：

User: [frustrated tone] "This isn't working at all!"
Expected: Model acknowledges emotion, adjusts response style

监控非提示响应：
- 模型应偶尔提供相关信息
- 应忽略真正不相关的输入
- 应根据上下文预测用户需求

何时禁用：

考虑在以下情况下禁用主动性/情感对话： - 正式/专业环境，其中情感适应不合适 - 高精度任务，其中可预测性至关重要 - 辅助功能应用程序，其中期望一致的行为 - 测试/调试，其中需要确定性行为

总结¶

在这一部分中，你学习了如何在 ADK 双向流式处理应用程序中实现多模态功能，重点关注音频、图像和视频功能。我们涵盖了音频规格和格式要求，探讨了原生音频和半级联架构之间的差异，检查了如何通过 LiveRequestQueue 和 Events 发送和接收音频流，并了解了音频转录、语音活动检测和主动/情感对话等高级功能。你现在了解了如何构建具有适当音频处理的自然语音启用 AI 体验，实现用于视觉上下文的视频流式传输，并根据平台功能配置模型特定功能。凭借对 ADK 多模态流式传输功能的全面了解，你已准备好构建处理文本、音频、图像和视频的生产就绪应用程序——在各种用例中创建丰富、交互式的 AI 体验。

恭喜！ 你已完成 ADK 双向流式处理开发人员指南。你现在对如何使用 Google 的 Agent Development Kit 构建生产就绪的实时流式传输 AI 应用程序有了全面的了解。

← 上一篇：第 4 部分 - 了解 RunConfig