open-xiaoai 接入 Gemini Live API 实战：让小爱音箱实现自动 VAD 连续语音对话

• 人工智能
• 大模型
• AI 应用
• AI Agent
• 语音
• 智能硬件
• 固件

本文基于 open-xiaoai 仓库中 examples/gemini 演示程序，讲解如何让已刷机的小爱音箱通过麦克风与扬声器，直接接入 Google 的 Gemini Live API，实现「自动 VAD + 连续对话」的纯语音交互。全文沿演示程序的架构、音频双向数据链路、Gemini Live 会话关键参数与回音抑制设计逐层展开，读完后你既能按步骤部署该演示，也能理解 Rust 扩展与 Python 业务层之间「谁负责什么」的边界，为自行扩展更多 AI 生态功能打下基础。

一、演示定位与前提条件

1.1 这个演示做什么

examples/gemini 是小爱音箱接入 Gemini Live API 的演示程序，核心能力有两点：

• 自动 VAD：程序端不跑任何本地 VAD 模型，用户说完话的判定由 Live API 服务端完成（后文源码层面会印证这一点）；
• 连续对话：一次live.connect建连后，会话循环持续收发音频，支持多轮对话而无需反复重连。

1.2 API 密钥（必读）

运行前必须先到 Google AI Studio 注册账号并创建 API 密钥，然后通过以下两种方式之一注入程序：

1. 设置环境变量GEMINI_API_KEY；
2. 或直接修改 examples/gemini/gemini/gemini.py 中的密钥常量。

对应源码为：

GEMINI_MODEL = "gemini-2.0-flash-live-001" GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") or "你的 API KEY"

注意：环境变量优先；仓库中默认模型常量是gemini-2.0-flash-live-001，实际可用模型请以 Google 官方文档的实时列表为准。

1.3 硬件与固件前提

演示程序本身只跑在 PC / NAS 等 x86 机器上，真正的「耳朵」和「嘴巴」在小爱音箱侧。因此 README 中特别强调：

请先确认你已经将小爱音箱刷机成功，并安装运行了 Rust 补丁程序（packages/client-rust/README.md），否则该项目启动后收不到音频输入，将无法正常工作。

具体链条是：

1. 按照 docs/flash.md 的教程刷机小爱音箱（仓库声明仅适用于小爱音箱 Pro（LX06）与Xiaomi 智能音箱 Pro（OH2P）两款机型）；
2. 在小爱音箱上安装并运行 packages/client-rust 补丁程序，它负责转发麦克风音频流、响应 Server 端指令（播放音频流等）；
3. Client 端启动时通过/data/open-xiaoai/server.txt中的地址（形如ws://<你的服务器IP>:4399）连回 Server 端。

安全提醒同样来自仓库文档：Server 地址必须由你本人指定为可信地址，不要连接来路不明的 server；该协议默认提供「执行任意脚本」的能力，公网运行时务必提高警惕。

1.4 软件环境

根据 examples/gemini/README.md，编译运行需要安装：

• uv（Python 包与项目管理器）
• Rust 工具链（用于编译本目录下的 pyo3 扩展）

为什么要 Rust？看 examples/gemini/pyproject.toml 的构建配置就明白了：

[build-system] requires = ["maturin>=1.0,<2.0"] build-backend = "maturin" [tool.maturin] features = ["pyo3/extension-module"]

examples/gemini/src/下的 Rust 代码会经 maturin 编译成名为open_xiaoai_server的 Python 扩展模块（见 examples/gemini/Cargo.toml 中crate-type = ["cdylib"]），Python 端import open_xiaoai_server导入的就是它。

二、项目架构：Rust 通信底座 + Python 业务层

2.1 整体分工

open-xiaoai 是「Client 端（音箱上）+ Server 端（PC 上）」的双端结构，packages/client-rust/README.md 对 Client 端职责的定义是：

• 共享通信协议（Server 端和 Client 端实时双向通信）；
• 转发小爱音箱的麦克风输入音频流到 Server 端；
• 转发小爱音箱上的事件到 Server 端（语音识别结果、播放状态等）；
• 响应来自 Server 端的指令调用（执行脚本、播放音频流、系统升级等）。

也就是说，Client 端只负责转发和被动响应，不实现业务逻辑——因为小爱音箱的内存与算力极为有限，语音识别这类任务不适合放在端侧；而业务逻辑放在 Python / Node.js 这类生态丰富的语言里开发效率更高。Gemini 演示正是这一设计哲学的直接产物：Server 端 =open_xiaoai_serverRust 扩展（网络与音频搬运）+ Python（Gemini 会话与回音抑制策略）。

2.2 Rust 扩展模块 open_xiaoai_server

扩展模块的入口在 examples/gemini/src/lib.rs，它向 Python 暴露了三个函数：

其中 examples/gemini/src/python.rs 的PythonManager是一个「函数名 → Python 可调用对象」的注册表（HashMap<String, PyObject>），Rust 侧收到音频流时通过call_fn反查并调用 Python 函数；日志方面则用 examples/gemini/src/macros.rs 中的pylog!宏把 Rust 日志转发为 Python 端的print输出，这样整个演示进程的日志统一出现在 Python 控制台。

2.3 WebSocket Server：监听 4399 端口

examples/gemini/src/server.rs 中AppServer::run()的行为：

pub async fn run() { let addr = "0.0.0.0:4399"; let listener = TcpListener::bind(&addr).await ...; crate::pylog!("✅ 已启动: {:?}", addr); while let Ok((stream, addr)) = listener.accept().await { // 同一时刻只处理一个连接 AppServer::handle_connection(stream, addr).await; } }

两个值得注意的实现细节（从源码结构看）：

• 端口固定为 4399，与 Client 端文档中ws://<IP>:4399的约定一致；
• 循环内对handle_connection是顺序await的，即同一时刻只处理一个连接，这与「单一音箱对接单一演示」的定位相符，不适合直接当作多设备服务使用。

连接建立后，init()会注册Event/Stream两类消息处理器、注册get_versionRPC，并在 1 秒后执行一次自检任务 examples/gemini/src/server.rs：

async fn test() -> Result<(), AppError> { SpeakerManager::play_text("已连接").await?; // RPC 调用 Client 端 start_recording：16kHz / 16bit / 单声道 // period_size: 1440 / 4, buffer_size: 1440 // RPC 调用 Client 端 start_play：24kHz / 16bit / 单声道 ... }

即：音箱连上 Server 后会先播报「已连接」，随后 Server 通过 RPC 命令音箱端启动录音（16 kHz）与播放（24 kHz）通道。这也是验证整条链路是否打通的最直观信号。

2.4 Python 主流程编排

examples/gemini/main.py 只有十几行，但把两端任务缝合成一个异步整体：

async def main(): xiaoai_task = asyncio.create_task(XiaoAi.start()) gemini_task = asyncio.create_task( Gemini.start( on_audio=XiaoAi.output_audio, set_is_speaking=XiaoAi.set_is_speaking, ) ) await asyncio.gather(xiaoai_task, gemini_task)

• XiaoAi.start()：注册on_input_data回调并启动 Rust 侧的 WebSocket Server（examples/gemini/gemini/xiaoai.py）；
• Gemini.start(on_audio, set_is_speaking)：建立 Gemini Live 会话，并把「下行音频」与「AI 是否在说话」两个钩子交给XiaoAi。

依赖版本约束见 examples/gemini/pyproject.toml：Python>=3.12、google-genai>=1.12.1、numpy等；Rust 侧pyo3启用了abi3-py38并依赖本地路径的 packages/client-rust（examples/gemini/Cargo.toml），因此 Gemini 演示与 client-rust 共用同一套通信协议实现。

三、编译运行：从克隆到听到「已连接」

3.1 完整操作步骤

继承 examples/gemini/README.md 的原始命令，克隆地址使用本仓库地址：

# 克隆代码 git clone https://gitcode.com/gh_mirrors/op/open-xiaoai.git # 进入当前项目根目录 cd examples/gemini # 安装 Python 依赖 uv sync --locked # 编译运行 uv run main.py

几点补充说明：

• uv sync --locked会依据uv.lock锁定版本安装 Python 依赖，同时触发本地包open-xiaoai-gemini的构建——uv的构建钩子会调用 maturin 编译src/下的 Rust 扩展，这就是 README 中「编译运行」只需一条uv run main.py的原因；
• examples/gemini/pyproject.toml 中[tool.uv]的cache-keys配置了pyproject.toml、Cargo.toml和src/**/*.rs，任一 Rust 文件变化都会触发扩展重新构建；
• 启动成功的控制台信号依次为：✅ 已启动: "0.0.0.0:4399"（Server 就绪）→ 音箱侧 Client 连入后✅ 已连接: ...→ 音箱播报「已连接」。

3.2 音箱侧的配套动作

在 PC 端程序跑起来后，需要在小爱音箱上把 Server 地址写入配置文件并运行 Client 补丁（详见 packages/client-rust/README.md）：

# 设置 server 地址（注意替换成自己的 server 地址） echo 'ws://192.168.31.227:4399' > /data/open-xiaoai/server.txt # 运行启动脚本 init.sh 完成安装 curl -sSfL <init.sh 下载地址> | sh

如需开机自启动，README 给出的做法是把boot.sh下载到/data/init.sh后reboot。

四、音频双向数据链路：谁把声音搬到哪

这是理解整个演示的关键。数据流在「音箱侧 client-rust 二进制 → WebSocket → Rust 扩展 Server → Python → Gemini Live API」之间往返，下面按上行、下行拆解。

4.1 上行：麦克风 → Gemini Live API

1. Server 端建连后通过 RPC 命令 Client 执行start_recording，Client 端 packages/client-rust/src/bin/client.rs 中start_recording的实现在录音回调里不断调用MessageManager::send_stream("record", bytes, None)，把麦克风 PCM 分片打上record标签回传；
2. Server 端的 examples/gemini/src/server.rson_stream收到tag == "record"的二进制流后，转成PyBytes并调用 Python 侧注册的on_input_data：

async fn on_stream(stream: Stream) -> Result<(), AppError> { let Stream { tag, bytes, .. } = stream; match tag.as_str() { "record" => { let data = Python::with_gil(|py| PyBytes::new(py, &bytes).into()); PythonManager::instance().call_fn("on_input_data", Some(data))?; } _ => {} } Ok(()) }

3. Python 侧 XiaoAi.input_audio 做回音抑制判断后，将音频提交给 Gemini：

def input_audio(cls, data: bytes): if cls.is_ai_speaking: # 如果 AI 正在回答问题，则不发送音频，防止把 AI 的声音被当做输入录制进来 # 暂不支持中断 AI 的回复，需要等待 AI 回答完成后才能重新响应用户的语音输入 return async def send_audio_task(): audio_array = np.frombuffer(data, dtype=np.uint16) await Gemini.send_audio(audio_array.tobytes()) asyncio.run_coroutine_threadsafe(send_audio_task(), cls.loop)

这里asyncio.run_coroutine_threadsafe的作用是：on_input_data由 Rust 线程（pyo3 GIL 上下文）触发，需要把协程调度回main()所在的事件循环执行。np.frombuffer(..., dtype=np.uint16).tobytes()本质上是把 16-bit PCM 原样透传——采样率等信息并不在 payload 里，而是在发送时由 MIME 声明（见 4.3 节）。

4. Gemini.send_audio 将音频以实时输入形式送入 Live 会话：

await cls.session.send_realtime_input( audio=types.Blob(data=data, mime_type="audio/pcm;rate=16000") )

audio/pcm;rate=16000与 Server 端给 Client 的录音参数（16 kHz / 16 bit / 单声道，见 examples/gemini/src/server.rs）完全一致，两端采样率对齐是链路正常的前提。

4.2 下行：Gemini Live API → 扬声器

反向链路在 Gemini.start 的接收循环中：

async for response in session.receive(): if response.server_content is None: continue if response.server_content.interrupted is True: continue if set_is_speaking: await set_is_speaking(True) if response.data is not None: if on_audio: await on_audio(response.data) if response.text is not None: if on_text: await on_text(response.text)

• response.data（二进制音频）经on_audio回调进入 XiaoAi.output_audio，调用 Rust 暴露的open_xiaoai_server.on_output_data(data)；
• examples/gemini/src/lib.rs 中on_output_data将其封装为send_stream("play", bytes, None)发往 Client；
• Client 端 packages/client-rust/src/bin/client.rs 的on_stream收到tag == "play"的流后交给AudioPlayer::instance().play(bytes)输出到扬声器。

4.3 音频参数对照表

综合 Server 端test()下发给 Client 的 RPC 参数与 packages/client-rust/src/services/audio/config.rs 的AudioConfig结构（pcm/channels/bits_per_sample/sample_rate/period_size/buffer_size六个字段），本演示的实际音频参数如下：

补充两点：config.rs中还有一个全局默认AUDIO_CONFIG（16 kHz、period 160、buffer 480），但本演示在start_recording/start_play的 RPC 请求里显式携带了 1440 系列的配置，实际以 RPC 下发值为准。另外pcm: "noop"表示不做端侧 PCM 编解码转换，原始 PCM 直接透传。

4.4 「自动 VAD」到底发生在哪一层

README 宣称「支持自动 VAD」，但整个仓库中并没有本地 VAD 模型。从源码结构看可以印证：上行音频全部通过send_realtime_input无截断地灌入 Live 会话，语句结束检测由 Gemini Live API 的服务端 VAD 完成——API 判定一句话讲完才会产出回答。这与 examples/xiaozhi 小智演示中在本地跑 Silero VAD 的做法形成对照：Gemini 演示把 VAD 职责完全外包给了云端 Live 接口，换来的是更少的本地代码与端侧算力占用。

连续对话则由Gemini.start的while True接收循环保证：会话长连接保持期间，send_audio可持续上行、session.receive()可持续下行，无需每轮重新live.connect。文本通道同样可用：Gemini.send_text 通过send_client_content(..., turn_complete=True)发送文本并显式结束本轮。

五、Gemini Live 会话关键参数详解

Gemini 类中的会话配置是「把 Live API 调成小爱语音助手」的核心，逐项拆解：

config = types.LiveConnectConfig( response_modalities=[types.Modality.AUDIO], system_instruction=types.Content( parts=[types.Part(text="你是小爱音箱，请用中文回答用户的问题。")] ), speech_config=types.SpeechConfig( language_code="cmn-CN", voice_config=types.VoiceConfig( prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Leda") ), ), context_window_compression=( types.ContextWindowCompressionConfig( sliding_window=types.SlidingWindow(), ) ), )

客户端构造也很简单：genai.Client(api_key=GEMINI_API_KEY)，建连使用async with cls.client.aio.live.connect(model=GEMINI_MODEL, config=cls.config) as session（examples/gemini/gemini/gemini.py），异步会话类型即genai.live.AsyncSession。

六、回音抑制设计：AI 如何避免「听到自己」

演示没有硬件全双工回声消除，靠的是一段精巧的软开关逻辑，这也是理解「为什么不能打断 AI」的钥匙。

6.1 状态机

XiaoAi 用is_ai_speaking+speaking_count两个类变量实现带延迟确认的说话状态：

@classmethod async def set_is_speaking(cls, is_speaking: bool): if is_speaking: cls.speaking_count += 1 cls.is_ai_speaking = True return # 延迟 1 秒，如果 AI 还在说话，则设置为不说话 async def set_is_speaking_task(): speaking_count = cls.speaking_count await asyncio.sleep(1) if cls.speaking_count == speaking_count: cls.is_ai_speaking = False asyncio.run_coroutine_threadsafe(set_is_speaking_task(), cls.loop)

• Gemini.start每收到一帧非打断的服务端内容就set_is_speaking(True)，speaking_count累加并立即进入「AI 说话中」状态；
• 在等待下一轮响应前会先set_is_speaking(False)（gemini.py 的waiting for response处），但 False 不是立即生效：挂起一个 1 秒的延迟任务，1 秒内若没有新的True到来（speaking_count不再变化），才真正把is_ai_speaking置回False。

这套「计数器 + 1 秒宽限」的设计用于吸收音频帧与文本帧之间、以及多帧响应之间的状态抖动：只要还在持续收到 AI 输出，开关就一直是关着的。

6.2 为什么由此「不支持打断」

回到 XiaoAi.input_audio：

if cls.is_ai_speaking: # 如果 AI 正在回答问题，则不发送音频，防止把 AI 的声音被当做输入录制进来 # 暂不支持中断 AI 的回复，需要等待 AI 回答完成后才能重新响应用户的语音输入 return

音箱的麦克风会同时录到扬声器播放的 AI 声音。若不加拦截，AI 的输出会被当作用户输入再次送进 Live API，形成「自问自答」的正反馈。演示的选择是：AI 说话期间直接丢弃所有上行音频——副作用就是此时用户说什么都听不到，即 README「注意事项」里声明的「暂不支持中断 AI 的回复」。若要支持打断（barge-in），通常需要在 Client 端做 AEC 回声消除、或在 Server 端跑本地 VAD 区分人声与播放声，这正是仓库提示「可以借助 Python 丰富的 AI 生态自行实现」的扩展方向。

七、注意事项与扩展方向

继承 examples/gemini/README.md 的原始声明并补充源码层面的依据：

1. 不支持中断：AI 回答完毕前用户的语音会被整体丢弃（见 6.2 节），这是回音抑制策略的直接代价；
2. 单连接模型：Server 端同一时刻只处理一个 Client 连接（server.rs），且协议没有鉴权与加密，仅适合可信局域网环境；
3. 演示定位：仓库明确该项目「只是一个简单的演示程序，抛砖引玉」，想要更多功能可以借助 Python 丰富的 AI 生态自行实现——比如替换system_instruction换人格、接入on_text回调驱动文字显示、把send_text暴露给遥控器按键实现「语音 + 文本」混合交互等；
4. 适用机型限制：整个 open-xiaoai 项目（含本演示）仅支持小爱音箱 Pro（LX06）与 Xiaomi 智能音箱 Pro（OH2P）；
5. 版本时效：GEMINI_MODEL常量当前指向gemini-2.0-flash-live-001，且主仓库 README 已声明项目停止维护，模型可用性、google-genaiSDK 行为请以 Google 官方文档的当前版本为准。

八、关键文件索引

按本文顺序走完「刷机 → 部署 client → 配置 API Key →uv run main.py」后，听到「已连接」播报即代表链路打通：此后对小爱音箱说话，你的语音会经 WebSocket 上行到 Gemini Live API，回答以 PCM 流形式经同一条链路播回音箱，整个过程无需任何文本输入。

• 人工智能
• 大模型
• AI 应用
• AI Agent
• 语音
• 智能硬件
• 固件