☰
open-xiaoai:小爱音箱接入小智 AI 的 Python Server 端详解(examples/xiaozhi 完整实践)
2026/10/14 23:48:07 网站建设 项目流程
  • 人工智能
  • 大模型
  • AI 应用
  • AI Agent
  • 语音
  • 智能硬件
  • 固件

【免费下载链接】open-xiaoai

🔊 让小爱音箱「听见你的声音」,解锁无限可能。

项目地址:https://gitcode.com/gh_mirrors/op/open-xiaoai
点击查看免费下载

本篇以仓库中 examples/xiaozhi/README.md 为核心,完整讲解如何用 open-xiaoai 的 Python 演示 Server 将小爱音箱接入小智 AI(xiaozhi-esp32 生态):从 Docker 一键运行到源码级的 WebSocket 协议、唤醒词(KWS)、VAD 断句与打断机制,读完你可以独立部署、调参并自定义唤醒逻辑。

一、这个演示程序解决什么问题

open-xiaoai 的整体思路是「接管小爱音箱的耳朵和嘴巴」:刷机后的音箱上运行一个 Rust Client 补丁程序(负责音频采集、事件转发和命令执行),局域网内的 Server 端程序则通过 WebSocket 与音箱互通,把音频流接到任意 AI 后端。examples/xiaozhi 是其中的 Python 版 Server 演示,目标是让小爱音箱接入小智 AI,具备以下能力(见 examples/xiaozhi/README.md):

  • 小爱音箱接入小智 AI,走 xiaozhi-esp32-server 的 WebSocket 协议;
  • 支持连续对话和中途打断;
  • 自定义唤醒词(中英文)和提示语;
  • 支持自定义消息处理(before_wakeup/after_wakeup回调),方便个人定制。

同时原文档有明确的适用边界(建议部署前阅读):

本项目只是一个简单的演示程序,抛砖引玉。诸如一些音频压缩、加密传输、多账号管理等功能并未提供,建议只在局域网内测试运行,不推荐部署在公网服务器上(消耗流量 100kb/s),请自行评估相关风险,合理使用。

此外,根据仓库根目录 README 的说明,整套方案目前仅适用于小爱音箱 Pro(LX06)和Xiaomi 智能音箱 Pro(OH2P)两款机型,其他型号请勿直接使用。

运行前提:需要先在音箱上完成刷机,并启动运行 Rust 补丁程序,教程见 packages/client-rust/README.md。

二、快速开始

2.1 克隆仓库并修改配置

# 克隆 open-xiaoai 仓库,然后进入示例目录 cd examples/xiaozhi

然后把 config.py 里的配置改成你自己的。原文档给出的最小配置如下:

APP_CONFIG = { "wakeup": { # 自定义唤醒词 "keywords": [ "豆包豆包", "你好小智", "hi siri", ], }, "xiaozhi": { "OTA_URL": "https://api.tenclass.net/xiaozhi/ota/", "WEBSOCKET_URL": "wss://api.tenclass.net/xiaozhi/v1/", }, }

OTA_URL与WEBSOCKET_URL指向小智 AI 的官方云服务;如果你想用自己部署的 xiaozhi-esp32-server,把这两个地址改成自己的即可(详见第六节常见问题)。

2.2 Docker 一键运行

推荐方式,直接拉取镜像运行,并把本地的config.py挂载进容器覆盖默认配置:

docker run -it --rm -p 4399:4399 -v $(pwd)/config.py:/app/config.py idootop/open-xiaoai-xiaozhi:latest

4399是 Server 与音箱上 Rust Client 通信的 WebSocket 端口。查看 examples/xiaozhi/Dockerfile 可知镜像内部做了什么:基于python:3.12-slim,安装libopus与 Rust 工具链,用uv sync --locked安装 Python 依赖(依赖清单在 pyproject.toml),再用 maturin 编译 Rust 扩展;运行阶段按TARGETARCH自动下载对应架构(x64/aarch64)的onnxruntime-linux动态库,并固定CLI=true环境变量,启动命令会先执行python xiaozhi/services/audio/kws/keywords.py(重新生成唤醒词 token 文件)再运行main.py。

2.3 源码编译运行

自行编译运行需要安装三类依赖:uv(Python 包管理)、Rust 工具链、Opus 动态链接库。关键命令:

# 安装 Python 依赖 uv sync --locked # 编译运行(GUI 模式,不支持唤醒词唤醒) uv run main.py # 或者设置环境变量 CLI=true,开启 CLI 模式(支持自定义唤醒词) CLI=true uv run main.py

两个模式的差异直接体现在源码里:

  • xiaozhi.py 的_initialize_display()根据get_env("CLI")决定加载NoDisplay还是GuiDisplay;
  • KWS 与 VAD 的启动入口 KWS.start() / VAD.start() 都有if not get_env("CLI"): return判断——因此只有 CLI 模式才启用唤醒词唤醒;
  • pyproject.toml 声明了requires-python = ">=3.12",核心依赖包括sherpa-onnx==1.12.32、onnxruntime==1.23.2、opuslib-next、websockets等。

如果只是想先体验小智 AI 本身(不连小爱音箱),可以运行:

uv run main.py --mode xiaozhi

--mode参数在 xiaoai.py 的XiaoAI.setup_mode()中解析:xiaoai(默认)使用小爱音箱的音频输入输出,xiaozhi则改用电脑本机的麦克风和扬声器。从源码看,这一切换由 stream.py 的MyAudio.create()实现——xiaozhi模式下返回真正的PyAudio实例,否则返回模拟 PyAudio 接口的MyAudio,其数据流实际走与音箱 Rust Client 的 WebSocket。

三、config.py 完整配置详解

仓库中 config.py 的完整结构与参数含义如下(结合源码补充了默认值与取值说明):

APP_CONFIG = { "wakeup": { # 自定义唤醒词列表(英文字母要全小写) "keywords": [ "天猫精灵", "小度小度", "豆包豆包", "你好小智", "你好小爱", "hi siri", "hey siri", ], # 静音多久后自动退出唤醒(秒) "timeout": 20, # 语音识别结果回调 "before_wakeup": before_wakeup, # 退出唤醒时的提示语(设置为空可关闭) "after_wakeup": after_wakeup, }, "vad": { # 语音检测阈值(0-1,越小越灵敏) "threshold": 0.10, # 最小语音时长(ms) "min_speech_duration": 250, # 最小静默时长(ms) "min_silence_duration": 500, }, "xiaozhi": { "OTA_URL": "https://api.tenclass.net/xiaozhi/ota/", "WEBSOCKET_URL": "wss://api.tenclass.net/xiaozhi/v1/", "WEBSOCKET_ACCESS_TOKEN": "", #(可选)一般用不到这个值 "VERIFICATION_CODE": "", # 首次对话时,验证码会在这里更新 "DEVICE_ID": "", # 如果没有提示绑定设备,则将 DEVICE_ID 清空后,重启应用再次尝试 }, }

各参数在源码中的作用:

参数源码位置说明
wakeup.keywordskeywords.py唤醒词文本,启动时被text2token编码为 sherpa-onnx 可用的keywords.txt
wakeup.timeoutevent.py唤醒后等待用户开口的超时时间,超时则播放after_wakeup提示并回到待命
wakeup.before_wakeupevent.py收到唤醒源(kws或xiaoai)后的回调,返回True才真正进入小智会话
wakeup.after_wakeupevent.py超时未说话或退出唤醒时的提示语回调
vad.thresholdvad/init.pySilero VAD 语音概率阈值,默认 0.01,越小越灵敏
vad.min_speech_durationvad/init.py累计语音超过该时长(ms)才判定「开始说话」,默认 250
vad.min_silence_durationvad/init.py累计静默超过该时长(ms)才判定「说话结束」,默认 500
xiaozhi.OTA_URLconfig.py设备激活/OTA 查询接口,用于获取服务端返回的mqtt等元信息
xiaozhi.WEBSOCKET_URLwebsocket_protocol.py小智 AI 的音频会话通道
xiaozhi.VERIFICATION_CODExiaozhi.py首次绑定设备时,服务端会把验证码写入此字段
xiaozhi.DEVICE_IDutils/config.py设备 ID,空则自动生成一个 MAC 风格的哈希并回写到 config.py

其中DEVICE_ID的生成逻辑值得注意:ConfigManager 会校验已有DEVICE_ID是否为合法 MAC 格式,不合法或为空时基于uuid.getnode()生成新的 MAC 地址并同步写回config.py文件——这也是为什么绑定失败后需要「将DEVICE_ID清空再重启」来重新进入绑定流程。

before_wakeup/after_wakeup两个回调的默认实现在 config.py 中:kws来源直接播放「你好主人,请问有什么吩咐?」并返回True;xiaoai来源(对小爱说「召唤小智」)会先speaker.abort_xiaoai()打断原有小爱服务、等待 2 秒让 TTS 恢复后再播报提示语。abort_xiaoai()在 speaker.py 中通过重启mico_aivs_lab服务实现,注释也明确说明重启约需 1-2s,期间音箱自带 TTS 不可用。

四、工作原理:两个进程的音频与事件链路

从源码结构看,整个系统由两部分组成:

  1. Rust 扩展模块open_xiaoai_server(src/lib.rs、src/server.rs):用 PyO3/maturin 编译进 Python 进程,在0.0.0.0:4399上监听 WebSocket;
  2. Python 客户端(xiaozhi/ 目录):负责唤醒、VAD、编解码,以及与远端小智 AI 的 WebSocket 会话。

4.1 Server 端(音箱 ↔ 本机)

src/server.rs 的AppServer::run()绑定0.0.0.0:4399,同一时刻只处理一个连接(即音箱上 Rust Client 发起的连接)。连接建立后注册两类处理器:

  • on_stream:收到record标签的音频流时,回调 Python 的on_input_data(server.rs);
  • on_event:把音箱上的事件(语音识别结果、播放状态等)以 JSON 转给 Python 的on_event。

连接建立 1 秒后还会执行一次自检(server.rs):播放「已连接」,并通过 RPC 调用音箱上的start_recording(16kHz/16bit/单声道)与start_play(24kHz)确认双向音频通路可用。

Python 侧的注册入口在 xiaoai.py 的init_xiaoai():把on_input_data、on_event注册给 Rust 模块,并启动后台 asyncio 事件循环;XiaoAI.on_output_data()则把要播放的 PCM 交给 Rust 模块的on_output_data(src/lib.rs),后者通过MessageManager::send_stream("play", ...)下发到音箱。

4.2 事件解析:如何「听」到小爱的话

xiaoai.py 的on_event()解析音箱转发的事件流,只关心两类:

  • instruction且NewLine内容里header.namespace == "SpeechRecognizer"、name == "RecognizeResult"的指令文本:
    • 若结果为空且is_vad_begin为真,说明用户喊了「小爱同学」但还没说完话,触发EventManager.on_interrupt()(打断当前小智回答);
    • 若拿到is_final的最终识别文本,则调用EventManager.wakeup(text, "xiaoai"),交由before_wakeup回调决定是否切换小智;
  • playing事件:同步音箱播放状态到speaker.status。

4.3 音频编解码管线

codec.py 中的AudioCodec是小智协议的编解码核心:

  • 上行:read_audio()从输入流(MyStream环形缓冲)读取 16kHz PCM,经opuslib按 60ms 帧长编码为 Opus 帧,由 xiaozhi.py 的主循环在LISTENING状态下经protocol.send_audio()发送;
  • 下行:write_audio()把小智 TTS 的 Opus 帧解码为 24kHz PCM 写入输出流,GlobalStream.on_output_data再把 PCM 送回报给 Rust 的play流(xiaoai.py)。

采样率约定来自 websocket_protocol.py:server_sample_rate = 24000、server_frame_duration = 60ms,每帧 1440 个采样点;源码注释里也留了一条现实提示——xiaozhi-esp32-server 返回的 hello 参数写的是 16k,但实际 TTS 音频是 24k,因此客户端固定按 24k 解码(websocket_protocol.py)。

4.4 小智 AI 的 WebSocket 协议

WebsocketProtocol 连接WEBSOCKET_URL时携带请求头Authorization(可选 Token)、Protocol-Version: 1、Device-Id、Client-Id,随后发送 hello 协商消息(format: opus、sample_rate: 16000、frame_duration: 60),并在 10 秒内等待服务端 hello 应答才算通道打开。会话期间由 heartbeat() 每秒发送一次ping,发送失败则自动重连。

收到的 JSON 消息按type分发(xiaozhi.py):

  • tts:start/stop控制SPEAKING状态与输入输出流切换;sentence_start携带增量文本打印并更新界面,同时用正则验证码.*?(\d+)/控制面板.*?(\d+)从回复文本中提取绑定验证码,自动写回config.py;
  • stt:打印用户语音识别结果(💬 我说:...);
  • llm:携带情绪标签,更新界面表情。

设备状态机为IDLE / CONNECTING / LISTENING / SPEAKING(set_device_state()):进入LISTENING时关闭输出流、打开输入流;进入SPEAKING时反之,从而保证说话与收听互不干扰。

五、自定义唤醒词的实现(KWS)

原文档强调本示例「支持自定义唤醒词(中英文)和提示语」,其实现链路由三个文件构成:

  1. 词表编码:keywords.py 读取APP_CONFIG["wakeup"]["keywords"],用 sherpa-onnx 的text2token(cjkchar+bpe分词类型 +bpe.model)把每个唤醒词转成 token 序列,写入 xiaozhi/models/keywords.txt。中文直接以字符 token 表示,英文需要 BPE 切分——这正是「英文唤醒词建议用空格分开发音」(如openai写成open ai)的原因;
  2. 识别引擎:sherpa.py 加载encoder.onnx/decoder.onnx/joiner.onnx构建KeywordSpotter,参数为keywords_score=2.0、keywords_threshold=0.2;检测循环以 16kHz 输入音频流(kws/init.py),命中后把小写唤醒词传给EventManager.wakeup(text, "kws");
  3. 会话调度:event.py 的__EventManager是打断/唤醒/连续对话的状态中枢,详见下一节。

注意 kws/init.py 中的一处细节:当设备处于LISTENING或SPEAKING状态时会跳过唤醒检测,避免对话过程中唤醒词误触发。

VAD 则基于 Silero 模型(silero.py):OnnxWrapper维护 2×128 的 GRU 状态和 64 采样点上下文窗口,每次送入 512 个采样(16kHz 下约 32ms),输出语音概率;vad/init.py 的检测循环按probability >= threshold切分语音帧/静音帧,并在累计时长超过min_speech_duration/min_silence_duration时分别回调on_speech/on_silence。

六、唤醒后的会话流程:连续对话与打断

event.py 的__start_session()是连续对话的核心状态机(仅在 CLI 模式生效),步骤如下:

  1. 先把设备置回IDLE,并send_abort_speaking取消之前的 TTS 输出;
  2. 若触发源是on_interrupt(用户喊「小爱同学」),直接返回——这就是「打断小智回答」的实现:中断 TTS、结束当前会话,用户后续可继续对小爱或小智说话;
  3. 否则(唤醒成功)先用 VAD 等待「余音结束」,再进入wait_next_step(timeout=wakeup.timeout)等待用户开口:
    • 超时未说话 → 回到IDLE,播放after_wakeup提示语(默认「主人再见,拜拜」);
    • 检测到语音 → 把 VAD 已缓存的语音片段前置(set_speech_frames),打开录音流,send_start_listening(MANUAL)进入LISTENING,把音频持续推给小智服务器;
  4. 说完后(VAD 判静音)调用send_stop_listening(),回到IDLE。而小智的 TTS 一结束(on_tts_end),会话再次start_session()等待下一句话——这就实现了免唤醒连续对话。

wakeup()入口处还会短暂pause()KWS,等before_wakeup回调播完提示语再resume(),避免提示语自身触发唤醒(event.py)。

七、常见问题(FAQ)

以下内容继承自 examples/xiaozhi/README.md 常见问题部分,并结合源码补充了机制说明:

Q:第一次运行提示输入验证码绑定设备,如何操作?

第一次启动对话时,语音会提示使用验证码绑定设备。打开小智 AI 管理后台,根据提示创建 Agent 并绑定设备即可。验证码会打印在终端,或者自动写进config.py:

APP_CONFIG = { "xiaozhi": { "VERIFICATION_CODE": "", # 首次对话时,验证码会在这里更新 "DEVICE_ID": "", # 如果没有提示绑定设备,则将 DEVICE_ID 清空后,重启应用再次尝试 }, # ... 其他配置 }

从源码看,验证码的提取发生在 _handle_tts_message():对每条sentence_start文本跑正则验证码.*?(\d+)(或控制面板.*?(\d+)),命中后由ConfigManager.update_config_file()正则替换写回 config.py。如果对话正常但始终没有验证码提示,说明随机生成的DEVICE_ID可能已被其他设备绑定——把config.py中的DEVICE_ID清空后重启即可重新进入绑定流程。绑定成功后可能需要重启应用才生效。

Q:回答太长了,如何打断小智 AI 的回答?

直接召唤「小爱同学」即可。从实现看,小爱的RecognizeResult中一旦检测到空文本 +is_vad_begin,就触发EventManager.on_interrupt()(xiaoai.py),递增会话 ID 使旧的wait_next_step立即返回interrupted,当前 TTS 被abort。

Q:有时候话还没说完 AI 就开始回答了,如何优化?

调大config.py中的vad.min_silence_duration(最小静默时长,ms),然后重启应用 / Docker:

APP_CONFIG = { "vad": { # 最小静默时长(ms) "min_silence_duration": 1000, }, # ... 其他配置 }

该值在 vad/init.py 中控制「静音累计超过多少才判定说话结束」,调大后停顿更久才提交给 ASR。

Q:对话时文字识别不准?

文字识别结果取决于小智 AI 服务器端的语音识别(ASR)方案,本端只负责上传 16kHz Opus 音频流,无法在客户端优化。

Q:唤醒词一直没有反应?

先调低vad.threshold试试:

APP_CONFIG = { "vad": { # 语音检测阈值(0-1,越小越灵敏) "threshold": 0.05, }, # ... 其他配置 }

另外两点:应用刚启动时需要加载模型文件(Silero VAD 与 sherpa-onnx KWS 均运行在 CPU 上),比较耗时,建议等 30s 再试;英文唤醒词可以把最小发音用空格分开(如openai改为open ai),提高text2token的切分匹配率。实在不行就换更易识别的唤醒词,比如「天猫精灵」。

Q:怎样使用自己部署的 xiaozhi-esp32-server 服务?

把config.py里的两个接口地址替换成自己的服务,重启应用:

APP_CONFIG = { "xiaozhi": { "OTA_URL": "https://your-domain/xiaozhi/ota/", "WEBSOCKET_URL": "wss://your-domain/xiaozhi/v1/", }, # ... 其他配置 }

注意OTA_URL需要返回mqtt字段(utils/config.py 校验),即标准 xiaozhi-esp32-server 的 OTA 接口格式。

Q:自己编译运行时,模型文件在哪里下载?

ASR/VAD/KWS 相关模型体积较大,未直接提交在 git 仓库中。从仓库的 release(vad-kws-models标签)下载 VAD + KWS 模型压缩包,解压到 examples/xiaozhi/xiaozhi/models/ 路径下即可,源码中通过 file.py 的get_model_file_path()定位这些文件(如silero_vad.onnx、encoder.onnx、tokens.txt、bpe.model)。

八、小结

examples/xiaozhi 用不到千行 Python 代码串起了一条完整链路:音箱麦克风 → Rust Clientrecord流 → 本机 4399 端口 WebSocket → KWS/Silero VAD 本地决策 → Opus 编码 → 小智 AI WebSocket 会话 → TTS Opus 解码 →play流回到音箱扬声器,并通过EventManager的会话状态机实现了自定义唤醒、免唤醒连续对话与「小爱同学」随时打断。作为演示程序,它刻意省略了音频压缩传输、加密、多账号等生产级特性(原文档明确建议仅局域网使用);如果你想做更深度定制,可以参考原文档提到的相关项目 oxa-server(提供更完善的 config 配置方式),或基于 speaker.py 暴露的run_shell/ask_xiaoai等接口自行扩展业务逻辑。

延伸阅读(仓库内路径):

  • examples/xiaozhi/main.py:入口,模式解析与优雅退出(SIGINT)
  • examples/xiaozhi/xiaozhi/xiaozhi.py:主应用类、状态机与协议回调
  • examples/xiaozhi/xiaozhi/services/protocols/websocket_protocol.py:小智 WebSocket 协议实现
  • packages/client-rust/README.md:音箱端 Rust Client 补丁教程
  • docs/flash.md:小爱音箱刷机教程
  • 人工智能
  • 大模型
  • AI 应用
  • AI Agent
  • 语音
  • 智能硬件
  • 固件

【免费下载链接】open-xiaoai

🔊 让小爱音箱「听见你的声音」,解锁无限可能。

项目地址:https://gitcode.com/gh_mirrors/op/open-xiaoai
点击查看免费下载

相关推荐

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询