树莓派 4 全离线语音助手实战：唤醒词 + 关键词识别 + ASR 三级分流
原文：https://dev.to/voxrtio/building-a-fully-offline-voice-assistant-on-raspberry-pi-4-3d1m（作者 @voxrtio）
我们要构建什么
一个完全运行在 Raspberry Pi 4 上的语音助手。没有云端、没有 API key、没有按分钟计费。用户说“Hey Assistant”，设备唤醒，然后处理之后说出的内容。
两条交互路径：
- 用户说出 14 个固定命令之一（
turn on lights、play music、stop等）。设备立刻识别并执行动作。 - 用户说出固定集合之外的内容（
read me the news、what time does the store open）。设备将语音转写为文本，并处理这个开放式请求。
三个模型协同工作。唤醒词检测器 全天候 24/7 监听麦克风，成本低、常驻运行。关键词识别器 接管接下来的几秒，尝试匹配固定词表。流式 ASR 则在关键词识别器没有匹配到时，转写开放式语音。它更重，但只在真正需要时运行。
三个模型都通过三条 pip install 命令安装，可在无互联网连接的 35 美元 Raspberry Pi 4 上运行。本文会完整讲解 Python 代码，给出等价的 Node.js、Go 和 C 片段，并解释为什么在同一个运行时里拆成三个 SDK，可以在不放弃开放式语音能力的同时保持很低的常驻成本。
为什么采用分流模式（先 WW，再 KWS，最后 ASR）
核心思想是：在真正需要之前，不运行最重的模型。
唤醒词检测器是整个链路里最轻的一环。我们的模型约 100 KB，在一台 15 美元的 Raspberry Pi Zero 2 W 上大约占用一个 A53 核心的 5.3%（全天候持续运行，数据来自我们此前的基准测试，链接见下文）。在 Pi 4 上这一成本还会更低，因为 Cortex-A72 核心更快，且运行在 64 位模式下。这意味着唤醒词可以持续监听环境，而不会从设备上的其他任务抢走有意义的 CPU。
但唤醒词只知道一件事：触发短语（本例中的 "Hey Assistant"）是否被说出。要理解用户真正想要什么，还需要另一个模型。
一个显然的选择是：唤醒词一触发就启动 ASR。这确实可行，但对大多数真实请求来说属于杀鸡用牛刀。在语音助手里，用户通常只会说一小组命令：开灯、设定时器、放音乐。对这些命令跑完整的开放词表 ASR，就像本来 strcmp 就能搞定的事，却去用正则表达式。
KWS（关键词识别器） 用很低的成本解决这个问题。我们的模型以 32 ms 帧处理 16 kHz 单声道音频，并将每一帧归入 14 个固定命令类之一。它不转写任意语音，但能快速回答“是的，这是命令 X”或“没有匹配”。
分流逻辑如下：
- WW 监听状态。 唤醒词持续运行，等待
"Hey Assistant"。 - WW 触发。 切换到 KWS，给用户大约 2 秒说出命令。
- KWS 匹配。 执行命令动作（开灯、停止等），回到状态 1。
- KWS 未匹配。 回退到 ASR，转写开放式语音，直到静音或超时。
- ASR 结束。 处理文本（交给本地 LLM、解析器或发送到你的后端），回到状态 1。
从计算角度看，这就像一个 switch：已知情况走对应分支，其他一切走 default。WW 在任意时刻都几乎不耗资源。KWS 只在唤醒词之后运行（通常 1 到 2 秒）。ASR 只在命令未命中 KWS 词表时运行。实践中，大多数语音助手请求都落在一个小命令词表内，具体比例则高度依赖你的产品。
硬件
最低配置：
- Raspberry Pi 4（4GB）。 4GB 内存为 ASR 模型（权重常驻内存）提供了充裕余量。2GB 版本能启动整个链路，但留给设备上其他程序的余地很小。Pi 5 和 Pi 3B+ 也可以，但我们在 Pi 4 上测试，因为它是目前开发者最容易拿到的 Pi 型号。
- USB 麦克风（10 到 15 美元，任何免驱设备），或 I2S 麦克风 HAT（如 ReSpeaker 2-Mics Pi HAT，约 10 美元）以获得更干净的音频。
- 32GB microSD 卡，安装 64 位 Raspberry Pi OS（aarch64）。必须使用 64 位系统。我们的 Linux SDK 只发布 aarch64 wheel，不支持 32 位 armhf。
- 可选： 如果你想要 TTS 语音回复，需要音箱或耳机。TTS 正在开发中，本教程不涉及。
只用唤醒词（不跑其余链路）时，即使 Pi Zero 2 W 也能运行。我们在之前的文章中测试过：在 15 美元的 Pi Zero 2 W 上跑唤醒词。加入 KWS 和 ASR 会提高算力需求，所以本教程使用 Pi 4。
安装三个 SDK
三个模型分别以独立包的形式发布。
pip install voxrt-wake-word voxrt-kws voxrt-asr
三个库底层共用同一个 VoxRT 运行时。该运行时是一个带 C ABI 的 Rust 推理引擎。我们没有打包成一个 voxrt-all 包，因为很多集成方只需要管线中的一部分（比如自助终端只需要唤醒词，听写应用只需要 ASR），让用户下载永远不会加载的权重文件没有意义。
模型权重以文件形式单独下载。
# 唤醒词 "Hey Assistant" curl -LO https://github.com/VoxRT/voxrt-wake-word-models/releases/download/v0.1.0/voxrt_wake_word.vxrt # 关键词识别（14 条命令） curl -LO https://github.com/VoxRT/voxrt-kws-models/releases/download/v0.1.0/voxrt_kws.vxrt # 流式 ASR（PC 中等规模模型） curl -LO https://github.com/VoxRT/voxrt-asr-models/releases/download/v0.1.2/streaming_medium_pc.vxrt
（具体版本号请以各仓库 README 为准，它们会定期更新。）
Node.js、Go 和 C 的安装方式见下文"在其他语言中使用同一管线"部分。
代码：Python 中的分流管道
完整示例分为三部分。
- 麦克风采集。 三个 SDK 共用。
- 状态机。 控制任一时刻哪个引擎接收音频。
- 装配。 将一切串起来的事件循环。
音频采集
我们使用 sounddevice（一个 PortAudio 封装库），这是在 Linux 上用 Python 读取麦克风的标准方式。它提供基于回调的 API。PortAudio 音频线程会用每一块新采样数据调用我们的函数。我们将数据推入队列，主线程再从队列中取出。
import sounddevice as sd
import numpy as np
from queue import Queue
SAMPLE_RATE = 16000 # 三个模型都期望 16 kHz 单声道
CHUNK_FRAMES = 512 # 32 ms，与 WW 和 KWS 内部块大小一致
audio_queue: "Queue[np.ndarray]" = Queue(maxsize=64)
def on_audio(indata, frames, time_info, status):
if status:
print(f"audio status: {status}")
# indata 是 (frames, 1) 的 int16。由于 sounddevice 在回调之间
# 复用同一块底层存储，所以需要复制缓冲区。
audio_queue.put(indata.copy())
stream = sd.InputStream(
samplerate=SAMPLE_RATE,
channels=1,
dtype="int16",
blocksize=CHUNK_FRAMES,
callback=on_audio,
)
stream.start()
调用之后，32 ms 的 int16 采样块会稳定地进入 audio_queue。下一个问题是：任一时刻由哪个引擎处理它们。
状态机
我们维护一个状态枚举和一个主循环，主循环从队列取出数据块并路由到正确的引擎。
from enum import Enum
class State(Enum):
IDLE = 1 # 只有 WW 在监听
KWS_CHECK = 2 # WW 触发，KWS 检查 2 秒
ASR_TRANSCRIBE = 3 # KWS 未命中，ASR 开始转写
state = State.IDLE
初始化三个引擎
from voxrt_wake_word import WakeWordEngine
from voxrt_kws import KwsEngine
from voxrt_asr import AsrStreamingEngine, DecodeMode
ww = WakeWordEngine.from_path("voxrt_wake_word.vxrt")
ww.threshold = 0.9
ww.cooldown_frames = 100 # 检测到唤醒词后忽略 100 帧（约 3.2 秒）
kws = KwsEngine.from_path(
"voxrt_kws.vxrt",
threshold=0.9,
consecutive_frames_required=3,
cooldown_frames=25,
)
print(f"KWS commands: {kws.class_names}")
asr = AsrStreamingEngine.from_path(
"streaming_medium_pc.vxrt",
mode=DecodeMode.RNNT,
)
主事件循环
分流就在这里发生。按状态拆分如下：
import time
from queue import Empty
KWS_WINDOW_SEC = 2.0 # 唤醒词之后留给 KWS 的窗口
ASR_SILENCE_SEC = 1.5 # ASR 语句末尾的静音判定时长
def handle_command(name):
print(f"running fixed command: {name}")
# 按需接到 GPIO、智能家居 API 等
def handle_open_query(text):
print(f"open-ended query: {text}")
# 按需发给本地 LLM、解析器等
kws_check_deadline = 0.0
asr_buffer_f32 = []
asr_last_speech_time = 0.0
try:
while stream.active:
try:
chunk_i16 = audio_queue.get(timeout=0.1)
except Empty:
continue
samples_i16 = chunk_i16.flatten().tolist()
if state == State.IDLE:
# WW 始终在监听。
for det in ww.push_pcm_i16(samples_i16):
print(f"[WW] wake detected, score={det.score:.3f}")
state = State.KWS_CHECK
kws_check_deadline = time.time() + KWS_WINDOW_SEC
elif state == State.KWS_CHECK:
# 在接下来 2 秒内尝试匹配固定命令。
matched = False
for det in kws.push_pcm_i16(samples_i16):
print(f"[KWS] command: {det.class_name} score={det.score:.3f}")
handle_command(det.class_name)
state = State.IDLE
matched = True
break
if not matched and time.time() > kws_check_deadline:
print("[KWS] no match, falling back to ASR")
state = State.ASR_TRANSCRIBE
asr_buffer_f32 = []
asr_last_speech_time = time.time()
elif state == State.ASR_TRANSCRIBE:
# ASR 期望 f32 采样。从 int16 转换。
f32 = [s / 32768.0 for s in samples_i16]
asr_buffer_f32.extend(f32)
# ASR 需要 200 ms 块（3200 个采样）。
while len(asr_buffer_f32) >= 3200:
chunk_200ms = asr_buffer_f32[:3200]
asr_buffer_f32 = asr_buffer_f32[3200:]
partial = asr.push_audio(chunk_200ms)
if partial:
print(f"[ASR partial] {partial}")
asr_last_speech_time = time.time()
# 如果静音超过 1.5 秒，结束当前语句。
if time.time() - asr_last_speech_time > ASR_SILENCE_SEC:
final = asr.stop()
print(f"[ASR final] {final}")
handle_open_query(final)
# asr.reset() 保留已加载的权重并重置流式状态，
# 比通过 from_path() 重新加载约 150 MB 的模型更划算。
# 如果你的 voxrt-asr 版本没有暴露 reset()，可以在这里回退
# 到 AsrStreamingEngine.from_path(...)，但要预期权重重新
# 加载时会停顿数秒。
asr.reset()
state = State.IDLE
finally:
stream.stop()
stream.close()
一个关键细节：WW 和 KWS 需要 int16 样本，ASR 需要归一化到 [-1.0, 1.0] 的 float32。转换只有一行（s / 32768.0），但很容易忘记。块大小也不同：WW 和 KWS 是 32 ms，ASR 是 200 ms（与 ASR 模型内部的注意力窗口匹配）。
这并非“一个 API 搞定一切”，而是“统一的集成模式（from_path 后跟 push_something 再读取结果），输入格式按各模型的需求匹配”。唤醒词和 KWS 在内部形状上一致（都是对短块进行事件检测），因此它们共享同一套签名。ASR 的工作方式不同（它是流式解码器），所以签名也不同。
其他语言中的同一套流水线
Node.js、Go 和 C 的完整快速入门示例都在仓库里：
github.com/VoxRT/voxrt-wake-word-linux/tree/main/examples/{nodejs,go,c}/github.com/VoxRT/voxrt-kws-linux/tree/main/examples/{nodejs,go,c}/github.com/VoxRT/voxrt-asr-linux/tree/main/examples/{nodejs,go,c}/
下面是关键代码片段，展示 push_pcm_i16 和 push_audio 在每种语言中的写法。完整的可运行示例请点击仓库链接查看。
Node.js
const { WakeWordEngine } = require("@voxrt/wake-word");
const { KwsEngine } = require("@voxrt/kws");
const { AsrStreamingEngine, DecodeMode } = require("@voxrt/asr");
const ww = WakeWordEngine.fromPath("voxrt_wake_word.vxrt");
ww.threshold = 0.9;
// 通过树莓派上可用的任意 ALSA/PortAudio Node 绑定采集麦克风。
mic.on("data", (buffer) => {
// Buffer 是更大 ArrayBuffer 的一个切片。使用 byteOffset 和
// byteLength 构建一个只覆盖所收到样本的视图。
const samples = new Int16Array(
buffer.buffer,
buffer.byteOffset,
buffer.length / 2,
);
for (const det of ww.pushPcmI16(samples)) {
console.log(`[WW] score=${det.score}`);
// 在此处将状态切换为 KWS_CHECK
}
});
Go
import (
"log"
wakeword "github.com/VoxRT/voxrt-wake-word-linux/go"
// kws "github.com/VoxRT/voxrt-kws-linux/go"
// asr "github.com/VoxRT/voxrt-asr-linux/go"
)
engine, err := wakeword.OpenFromPath("voxrt_wake_word.vxrt")
if err != nil {
log.Fatal(err)
}
defer engine.Close()
_ = engine.SetThreshold(0.9)
_ = engine.SetCooldownFrames(100)
// 通过树莓派上可用的任意 ALSA/PortAudio Go 绑定采集麦克风。
for chunk := range micChan { // chunk 是长度为 512 的 []int16
for _, d := range engine.PushPcmI16(chunk) {
log.Printf("[WW] score=%.3f", d.Score)
// 在此处切换状态
}
}
C
#include <stdint.h>
#include <stddef.h>
#include <stdio.h>
#include <voxrt_wake_word.h>
// load_vxrt_model() 对模型文件执行 mmap 并返回 (bytes, len)。
// 具体实现见仓库中 examples/c/alsa-mic-quickstart/main.c，
// 约 20 行，围绕 open() + fstat() + mmap() 展开。
// alsa_read_chunk() 是对 snd_pcm_readi() 的轻量封装。
const uint8_t *model_bytes = NULL;
size_t model_len = 0;
if (load_vxrt_model("voxrt_wake_word.vxrt", &model_bytes, &model_len) != 0) {
fprintf(stderr, "failed to mmap the wake-word model file\n");
return 1;
}
voxrt_wake_word_t *engine = NULL;
voxrt_status_t rc = voxrt_wake_word_create(model_bytes, model_len, &engine);
if (rc != VOXRT_OK) {
fprintf(stderr, "voxrt_wake_word_create failed: %d\n", (int)rc);
return 2;
}
voxrt_wake_word_set_threshold(engine, 0.9f);
voxrt_wake_word_set_cooldown_frames(engine, 100);
// ALSA 采集，16 kHz 单声道 int16，每块 512 个样本。
int16_t chunk[512];
voxrt_wake_word_detection_t dets[8];
while (alsa_read_chunk(pcm_handle, chunk, 512) == 0) {
size_t written = 0;
rc = voxrt_wake_word_push_pcm_i16(engine, chunk, 512, dets, 8, &written);
// VOXRT_ERR_BUFFER_TOO_SMALL 只是表示还能容纳更多检测结果；
// 已返回的结果仍然有效。
if (rc != VOXRT_OK && rc != VOXRT_ERR_BUFFER_TOO_SMALL) {
fprintf(stderr, "push failed: %d\n", (int)rc);
break;
}
for (size_t i = 0; i < written; i++) {
printf("[WW] score=%.3f\n", dets[i].score);
// 在此处切换状态
}
}
voxrt_wake_word_destroy(engine);
// 记得 munmap(model_bytes, model_len)
每个 SDK 的 quickstart 文件都已经完成了这些工作：打开麦克风、把音频喂给引擎、打印检测结果。我们的分级检测循环就是在此基础上构建的状态机。
树莓派 4 上的预期性能
重要声明。 以下数字是从其他硬件上的基准测试外推得出的。我们还没有在树莓派 4 上跑过完整的分级检测循环。你的数值应该会落在这个区间内，但只有在自己板子上实测后才能确定。
- IDLE 状态（仅唤醒词引擎）。 约占一个 CPU 核心的 1% 到 2%。从树莓派 Zero 2 W 上的 5.3% 外推而来（1 GHz 的 A53，运行默认的 32 位 Raspberry Pi OS）。树莓派 4 使用 1.5 GHz 的 64 位 A72，IPC 更高，且在操作系统运行于 aarch64 模式时 NEON 更高效。
- KWS_CHECK 状态（唤醒后 2 秒）。 突发期间约占 15% 到 25% CPU。从骁龙 662 上的 16% RTF 外推而来，但树莓派 4 的 A72 比 SD662 中的 Kryo 260 Gold 核心早一代，因此 RTF 预计会略高一些。KWS 只在短时间窗口内运行。
- ASR_TRANSCRIBE 状态。 活跃转写期间约占 40% 到 70% CPU。ASR 是三者中最重的，为移动端骁龙调优的 NEON 专用 int8 内核在 A72 上发挥不出同样的效率。在树莓派 4 aarch64 上应该仍能实现实时流式转写（RTF 低于 1.0），但余量是三者中最小的。先在自己设备上实测，再下结论。
内存占用：三个引擎全部初始化后大约 200 到 300 MB。ASR 占大头，其权重约 150 MB。这就是我们推荐 4 GB 内存版树莓派 4 的原因。2 GB 版也能运行，但留给其他软件的空间很小。
发布后如果我们拿到树莓派 4 的真实数据，会写一篇后续文章给出实际测量结果。
本文未涉及的内容
- TTS（语音应答）：开发中。独立的端侧语音克隆模型。发布时会单独写一篇文章。
- 自定义唤醒词短语：目前我们只提供预训练好的
"Hey Assistant"模型。训练自己的触发短语需要单独的训练流水线，该流水线尚未公开发布。 - 说话人验证 / 声纹生物识别：未包含。这些模型不会告诉你说话人是谁。
- 多语言支持：唤醒词和 KWS 目前仅支持英文。ASR 模型大体上是多语言的，但我们只在英文上做过测试。
以上所有内容都在 VoxRT 的路线图上。如果其中任何一项对你的用例至关重要，请在评论区告诉我们，以便我们安排优先级。
仓库与资源
- 唤醒词 Linux SDK：github.com/VoxRT/voxrt-wake-word-linux
- KWS Linux SDK：github.com/VoxRT/voxrt-kws-linux
- ASR Linux SDK：github.com/VoxRT/voxrt-asr-linux
- 全部 SDK：github.com/VoxRT
- 我们之前关于端侧语音的文章：dev.to/voxrtio
如果你正在构建任何语音驱动应用，并希望音频保留在设备端，这三个 SDK 都已放在 GitHub 上，文档在 README 中。每个仓库都包含 Python、Node.js、Go 和 C 的完整快速入门示例。
原文：https://dev.to/voxrtio/building-a-fully-offline-voice-assistant-on-raspberry-pi-4-3d1m（作者 @voxrtio）