Sherpa-ONNX 之关键词检测 KWS 入门实战

原创

buzzfrog

发布于 2025-12-16 22:31:06

301

文章被收录于专栏：云上修行云上修行

使用 sherpa-onnx 实现轻量级、高效的关键词检测（Keyword Spotting）

前言

在智能语音交互中，关键词检测（Keyword Spotting, KWS） 是唤醒语音助手的第一步。例如 "Hey Siri"、"小爱同学"、"你好小问" 等，都是通过 KWS 技术实现的。

本文将介绍如何使用 sherpa-onnx —— 一个由新一代 Kaldi 团队开发的开源语音识别工具包，来快速搭建一个实时关键词检测系统。

为什么选择 sherpa-onnx？

轻量高效：模型体积小（3.3M），适合边缘设备部署
跨平台：支持 Linux、macOS、Windows、Android、iOS、树莓派等
易于集成：提供 Python、C++、Java、Swift 等多种语言 API
中英文支持：提供预训练的中文和英文模型
自定义关键词：无需重新训练，即可添加自定义唤醒词

一、环境准备

1.1 安装依赖

pip3 install sherpa-onnx pyaudio numpy sentencepiece pypinyin

💡 提示：macOS 用户如果 pyaudio 安装失败，请先运行 brew install portaudio

1.2 下载预训练模型

这里我们使用官方提供的中文 Zipformer 模型，仅 3.3M 大小：

# 下载中文模型
wget https://github.com/k2-fsa/sherpa-onnx/releases/download/kws-models/sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01.tar.bz2

# 解压模型
tar xf sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01.tar.bz2

# 删除压缩包
rm sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01.tar.bz2

如果需要英文模型：

wget https://github.com/k2-fsa/sherpa-onnx/releases/download/kws-models/sherpa-onnx-kws-zipformer-gigaspeech-3.3M-2024-01-01.tar.bz2
tar xf sherpa-onnx-kws-zipformer-gigaspeech-3.3M-2024-01-01.tar.bz2

模型目录结构如下：

sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/
├── encoder-epoch-12-avg-2-chunk-16-left-64.onnx    # 编码器
├── decoder-epoch-12-avg-2-chunk-16-left-64.onnx    # 解码器
├── joiner-epoch-12-avg-2-chunk-16-left-64.onnx     # Joiner
├── encoder-epoch-12-avg-2-chunk-16-left-64.int8.onnx  # int8量化版（更快）
├── decoder-epoch-12-avg-2-chunk-16-left-64.int8.onnx
├── joiner-epoch-12-avg-2-chunk-16-left-64.int8.onnx
├── tokens.txt                                       # 词表
└── test_wavs/
    └── test_keywords.txt                           # 测试用关键词

二、代码实战

2.1 核心代码解析

下面我们逐步解析 Demo 代码的核心部分。

2.1.1 创建关键词检测器

import sherpa_onnx

def create_keyword_spotter(args):
    """创建关键词检测器"""
    kws = sherpa_onnx.KeywordSpotter(
        tokens=args.tokens,              # 词表文件
        encoder=args.encoder,            # 编码器模型
        decoder=args.decoder,            # 解码器模型
        joiner=args.joiner,              # Joiner 模型
        num_threads=args.num_threads,    # 推理线程数
        max_active_paths=args.max_active_paths,
        keywords_file=args.keywords_file,          # 关键词文件
        keywords_score=args.keywords_score,        # 关键词增强分数
        keywords_threshold=args.keywords_threshold, # 触发阈值
        num_trailing_blanks=args.num_trailing_blanks,
        provider=args.provider,          # 推理后端
    )
    return kws

参数说明：

参数	说明
`keywords_score`	关键词增强分数，越大越容易被检测到
`keywords_threshold`	触发阈值，越大需要更高的置信度才能触发
`num_trailing_blanks`	关键词后跟随的空白帧数，用于处理关键词重叠

2.1.2 音频采集与检测循环

import pyaudio
import numpy as np

# 配置音频参数
sample_rate = 16000
chunk_size = int(0.1 * sample_rate)  # 每次读取 100ms 音频

# 初始化 PyAudio
p = pyaudio.PyAudio()
audio_stream = p.open(
    format=pyaudio.paInt16,
    channels=1,
    rate=sample_rate,
    input=True,
    frames_per_buffer=chunk_size,
)

# 创建 stream
stream = kws.create_stream()

while True:
    # 1. 读取麦克风音频
    audio_data = audio_stream.read(chunk_size, exception_on_overflow=False)
    
    # 2. 转换为 float32 格式（范围 -1.0 到 1.0）
    samples_int16 = np.frombuffer(audio_data, dtype=np.int16)
    samples_float32 = samples_int16.astype(np.float32) / 32768.0
    
    # 3. 送入检测器
    stream.accept_waveform(sample_rate, samples_float32)
    
    # 4. 执行解码
    while kws.is_ready(stream):
        kws.decode_stream(stream)
        result = kws.get_result(stream)
        
        if result:
            print(f"🎯 检测到关键词: {result}")
            # 重要：检测到后必须重置 stream
            kws.reset_stream(stream)

关键要点：

音频格式转换：PyAudio 返回的是 int16 格式，需要转换为 float32（范围 -1.0 到 1.0）
流式检测：使用 accept_waveform() 不断输入音频片段
重置 Stream：检测到关键词后必须调用 reset_stream()，否则会持续触发

2.2 完整 Demo 运行

假设模型已下载到当前目录，运行以下命令：

python3 kws_demo.py \
  --tokens ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/tokens.txt \
  --encoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/encoder-epoch-12-avg-2-chunk-16-left-64.onnx \
  --decoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/decoder-epoch-12-avg-2-chunk-16-left-64.onnx \
  --joiner ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/joiner-epoch-12-avg-2-chunk-16-left-64.onnx \
  --keywords-file ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/test_wavs/test_keywords.txt

运行效果：

2025-12-16 14:12:25 - INFO - 可用的音频输入设备:
2025-12-16 14:12:25 - INFO -   默认输入设备 ID: 11, 名称: default
2025-12-16 14:12:25 - INFO -   设备 ID: 1, 名称: rockchip-es8388: dailink-multicodecs ES8323 HiFi-0 (hw:1,0), 输入通道数: 2
2025-12-16 14:12:25 - INFO -   设备 ID: 3, 名称: MCP01: USB Audio (hw:3,0), 输入通道数: 1
2025-12-16 14:12:25 - INFO -   设备 ID: 7, 名称: pulse, 输入通道数: 32
2025-12-16 14:12:25 - INFO -   设备 ID: 11, 名称: default, 输入通道数: 32
2025-12-16 14:12:25 - INFO - 正在初始化关键词检测器...
2025-12-16 14:12:27 - INFO - 关键词检测器初始化完成！

2025-12-16 14:12:27 - INFO - ============================================================
2025-12-16 14:12:27 - INFO - 关键词检测已启动！请对着麦克风说出关键词...
2025-12-16 14:12:27 - INFO - 关键词文件: ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/test_wavs/test_keywords.txt
2025-12-16 14:12:27 - INFO - 按 Ctrl+C 停止程序
2025-12-16 14:12:27 - INFO - ============================================================
2025-12-16 14:12:29 - INFO - ========================================
2025-12-16 14:12:29 - INFO - 🎯 检测到关键词！第 1 次
2025-12-16 14:12:29 - INFO -    关键词: 小傅
2025-12-16 14:12:29 - INFO -    时间: 2025-12-16 14:12:29.641
2025-12-16 14:12:29 - INFO - ========================================

如果想使用 int8 量化模型 加速推理：

python3 kws_demo.py \
  --tokens ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/tokens.txt \
  --encoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/encoder-epoch-12-avg-2-chunk-16-left-64.int8.onnx \
  --decoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/decoder-epoch-12-avg-2-chunk-16-left-64.int8.onnx \
  --joiner ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/joiner-epoch-12-avg-2-chunk-16-left-64.int8.onnx \
  --keywords-file ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/test_wavs/test_keywords.txt

三、自定义关键词

sherpa-onnx 的一大优势是可以在不重新训练模型的情况下添加自定义关键词。

3.1 准备原始关键词文件

创建 keywords_raw.txt，每行一个关键词，格式为：关键词 @显示名称

你好军哥 @你好军哥
你好问问 @你好问问
小爱同学 @小爱同学
嘿小度 @嘿小度

3.2 使用 text2token 工具转换

sherpa-onnx 使用拼音 token 来表示关键词，需要使用官方工具进行转换：

sherpa-onnx-cli text2token \
  --tokens sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/tokens.txt \
  --tokens-type ppinyin \
  keywords_raw.txt keywords.txt

参数说明：

--tokens：模型的词表文件
--tokens-type ppinyin：中文模型使用 ppinyin（带声调拼音）
输入文件：keywords_raw.txt
输出文件：keywords.txt

转换后的 keywords.txt 内容如下：

n ǐ h ǎo j ūn g ē @你好军哥
n ǐ h ǎo w èn w èn @你好问问
x iǎo ài t óng x ué @小爱同学
h ēi x iǎo d ù @嘿小度

3.3 使用自定义关键词运行

python3 kws_demo.py \
  --tokens ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/tokens.txt \
  --encoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/encoder-epoch-12-avg-2-chunk-16-left-64.onnx \
  --decoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/decoder-epoch-12-avg-2-chunk-16-left-64.onnx \
  --joiner ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/joiner-epoch-12-avg-2-chunk-16-left-64.onnx \
  --keywords-file ./keywords.txt

四、参数调优

在实际应用中，可能需要根据场景调整一些参数：

4.1 核心参数

参数	默认值	作用	调优建议
`--keywords-score`	1.0	关键词的增强分数	如果漏检多，适当增大；如果误检多，适当减小
`--keywords-threshold`	0.25	触发阈值	越大越难触发，适合减少误唤醒
`--num-trailing-blanks`	1	关键词后的空白帧数	如果关键词有重叠 token，设为较大值（如 8）

4.2 性能参数

参数	默认值	说明
`--num-threads`	2	推理线程数，根据 CPU 核心数调整
`--provider`	cpu	推理后端：`cpu`、`cuda`（GPU）、`coreml`（Apple 芯片）
`--max-active-paths`	4	解码活跃路径数，增大可提高准确率但会变慢

4.3 音频参数

参数	默认值	说明
`--sample-rate`	16000	采样率，模型固定 16kHz
`--chunk-duration`	0.1	每次读取音频时长，影响实时性和 CPU 占用

五、架构解析

5.1 KWS 系统流程

┌──────────────┐    ┌───────────────┐    ┌───────────────┐
│   麦克风     │ -> │  音频预处理    │ -> │  特征提取     │
│   PyAudio    │    │  int16->float │    │  Encoder      │
└──────────────┘    └───────────────┘    └───────────────┘
                                                 │
                                                 v
┌──────────────┐    ┌───────────────┐    ┌───────────────┐
│   输出结果   │ <- │  关键词匹配    │ <- │  解码搜索     │
│   Callback   │    │  Keywords     │    │  Decoder+Join │
└──────────────┘    └───────────────┘    └───────────────┘

5.2 Zipformer 模型架构

sherpa-onnx 使用的是 Zipformer 模型，这是一种基于 RNN-Transducer 架构的高效端到端语音模型：

Encoder：将音频特征编码为高维向量
Decoder：预测下一个 token
Joiner：结合 Encoder 和 Decoder 的输出，计算最终概率

相比传统的 Conformer，Zipformer 具有更少的参数和更快的推理速度。

六、常见问题

Q1: 误唤醒太多怎么办？

增大 --keywords-threshold（如 0.4、0.5）
减小 --keywords-score

Q2: 漏检太多怎么办？

减小 --keywords-threshold（如 0.15、0.20）
增大 --keywords-score

Q3: 如何在安静环境下表现更好？

确保麦克风增益适中
可以考虑添加 VAD（语音活动检测）前处理

Q4: 支持多关键词吗？

支持！在 keywords.txt 中每行写一个关键词即可，检测到后会返回对应的显示名称。

Q5: 如何部署到嵌入式设备？

sherpa-onnx 支持多种平台，可以使用：

C++ API 直接集成
交叉编译到 ARM/RISC-V
使用 int8 量化模型减少内存占用

七、总结

本文介绍了如何使用 sherpa-onnx 快速搭建一个实时关键词检测系统：

安装依赖并下载预训练模型
理解核心 API：KeywordSpotter、create_stream()、accept_waveform()
掌握自定义关键词的方法
学会参数调优以适应不同场景

sherpa-onnx 的轻量设计使其非常适合在边缘设备上部署，无论是树莓派、Android 手机还是智能音箱，都可以轻松运行。

#附录

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Sherpa-ONNX 关键词检测 Demo

使用 PyAudio 从麦克风实时读取音频，结合 sherpa-onnx 进行关键词检测 (KWS)。
当检测到预定义的关键词时，打印日志信息。

参考: https://github.com/k2-fsa/sherpa-onnx
模型下载: https://k2-fsa.github.io/sherpa/onnx/kws/pretrained_models/index.html

用法示例（中文模型）:
    python3 kws_demo.py \
        --tokens ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/tokens.txt \
        --encoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/encoder-epoch-12-avg-2-chunk-16-left-64.onnx \
        --decoder ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/decoder-epoch-12-avg-2-chunk-16-left-64.onnx \
        --joiner ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/joiner-epoch-12-avg-2-chunk-16-left-64.onnx \
        --keywords-file ./sherpa-onnx-kws-zipformer-wenetspeech-3.3M-2024-01-01/test_wavs/test_keywords.txt
"""

import argparse
import sys
import time
import logging
from pathlib import Path
from datetime import datetime

import numpy as np

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

# 检查并导入 pyaudio
try:
    import pyaudio
except ImportError:
    logger.error("请先安装 pyaudio: pip3 install pyaudio")
    sys.exit(1)

# 检查并导入 sherpa_onnx
try:
    import sherpa_onnx
except ImportError:
    logger.error("请先安装 sherpa-onnx: pip3 install sherpa-onnx")
    sys.exit(1)


def check_file_exists(filepath: str, description: str = "文件") -> bool:
    """检查文件是否存在"""
    if not Path(filepath).is_file():
        logger.error(f"{description}不存在: {filepath}")
        return False
    return True


def get_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(
        description="Sherpa-ONNX 关键词检测 Demo - 使用 PyAudio 从麦克风实时检测关键词",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )

    # 模型文件参数
    parser.add_argument(
        "--tokens",
        type=str,
        required=True,
        help="tokens.txt 文件路径"
    )
    parser.add_argument(
        "--encoder",
        type=str,
        required=True,
        help="编码器 ONNX 模型路径"
    )
    parser.add_argument(
        "--decoder",
        type=str,
        required=True,
        help="解码器 ONNX 模型路径"
    )
    parser.add_argument(
        "--joiner",
        type=str,
        required=True,
        help="joiner ONNX 模型路径"
    )
    parser.add_argument(
        "--keywords-file",
        type=str,
        required=True,
        help="关键词文件路径，每行一个关键词（需要先用 text2token 工具处理）"
    )

    # 推理参数
    parser.add_argument(
        "--num-threads",
        type=int,
        default=2,
        help="神经网络推理使用的线程数"
    )
    parser.add_argument(
        "--provider",
        type=str,
        default="cpu",
        choices=["cpu", "cuda", "coreml"],
        help="推理后端: cpu, cuda, coreml"
    )
    parser.add_argument(
        "--max-active-paths",
        type=int,
        default=4,
        help="解码时保留的最大活跃路径数"
    )
    parser.add_argument(
        "--num-trailing-blanks",
        type=int,
        default=1,
        help="关键词后跟随的空白帧数（如果关键词之间有重叠token，可设置为较大值如8）"
    )
    parser.add_argument(
        "--keywords-score",
        type=float,
        default=1.0,
        help="关键词 token 的增强分数，越大越容易被检测到"
    )
    parser.add_argument(
        "--keywords-threshold",
        type=float,
        default=0.25,
        help="关键词触发阈值（概率），越大越难触发"
    )

    # 音频参数
    parser.add_argument(
        "--sample-rate",
        type=int,
        default=16000,
        help="音频采样率（Hz）"
    )
    parser.add_argument(
        "--chunk-duration",
        type=float,
        default=0.1,
        help="每次读取的音频时长（秒）"
    )

    return parser.parse_args()


def list_audio_devices():
    """列出所有可用的音频输入设备"""
    p = pyaudio.PyAudio()
    logger.info("可用的音频输入设备:")
    
    default_input_device = p.get_default_input_device_info()
    logger.info(f"  默认输入设备 ID: {default_input_device['index']}, 名称: {default_input_device['name']}")
    
    for i in range(p.get_device_count()):
        dev_info = p.get_device_info_by_index(i)
        if dev_info['maxInputChannels'] > 0:  # 只显示有输入通道的设备
            logger.info(f"  设备 ID: {i}, 名称: {dev_info['name']}, 输入通道数: {dev_info['maxInputChannels']}")
    
    p.terminate()
    return default_input_device['index']


def create_keyword_spotter(args) -> sherpa_onnx.KeywordSpotter:
    """创建关键词检测器"""
    logger.info("正在初始化关键词检测器...")
    
    kws = sherpa_onnx.KeywordSpotter(
        tokens=args.tokens,
        encoder=args.encoder,
        decoder=args.decoder,
        joiner=args.joiner,
        num_threads=args.num_threads,
        max_active_paths=args.max_active_paths,
        keywords_file=args.keywords_file,
        keywords_score=args.keywords_score,
        keywords_threshold=args.keywords_threshold,
        num_trailing_blanks=args.num_trailing_blanks,
        provider=args.provider,
    )
    
    logger.info("关键词检测器初始化完成！")
    return kws


def main():
    """主函数"""
    args = get_args()
    
    # 检查所有必需文件是否存在
    files_ok = True
    files_ok &= check_file_exists(args.tokens, "tokens 文件")
    files_ok &= check_file_exists(args.encoder, "encoder 模型")
    files_ok &= check_file_exists(args.decoder, "decoder 模型")
    files_ok &= check_file_exists(args.joiner, "joiner 模型")
    files_ok &= check_file_exists(args.keywords_file, "关键词文件")
    
    if not files_ok:
        logger.error("请检查模型文件路径是否正确！")
        logger.error("模型下载地址: https://k2-fsa.github.io/sherpa/onnx/kws/pretrained_models/index.html")
        sys.exit(1)
    
    # 列出音频设备
    default_device_id = list_audio_devices()
    
    # 创建关键词检测器
    kws = create_keyword_spotter(args)
    
    # 创建 stream
    stream = kws.create_stream()
    
    # 配置 PyAudio
    sample_rate = args.sample_rate
    chunk_size = int(args.chunk_duration * sample_rate)  # 每次读取的采样点数
    
    p = pyaudio.PyAudio()
    
    try:
        # 打开麦克风输入流
        audio_stream = p.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=sample_rate,
            input=True,
            frames_per_buffer=chunk_size,
        )
        
        logger.info("=" * 60)
        logger.info("关键词检测已启动！请对着麦克风说出关键词...")
        logger.info(f"关键词文件: {args.keywords_file}")
        logger.info("按 Ctrl+C 停止程序")
        logger.info("=" * 60)
        
        detection_count = 0
        
        while True:
            # 从麦克风读取音频数据
            audio_data = audio_stream.read(chunk_size, exception_on_overflow=False)
            
            # 将 bytes 转换为 numpy array
            samples_int16 = np.frombuffer(audio_data, dtype=np.int16)
            samples_float32 = samples_int16.astype(np.float32) / 32768.0
            
            # 将音频数据送入关键词检测器
            stream.accept_waveform(sample_rate, samples_float32)
            
            # 执行检测
            while kws.is_ready(stream):
                kws.decode_stream(stream)
                result = kws.get_result(stream)
                
                if result:
                    detection_count += 1
                    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
                    
                    # 打印检测日志
                    logger.info("=" * 40)
                    logger.info(f"🎯 检测到关键词！第 {detection_count} 次")
                    logger.info(f"   关键词: {result}")
                    logger.info(f"   时间: {timestamp}")
                    logger.info("=" * 40)
                    
                    # 重要：检测到关键词后必须重置 stream
                    kws.reset_stream(stream)
                    
    except KeyboardInterrupt:
        logger.info("\n程序已停止（用户中断）")
    finally:
        # 清理资源
        audio_stream.stop_stream()
        audio_stream.close()
        p.terminate()
        logger.info(f"总共检测到 {detection_count} 次关键词")


if __name__ == "__main__":
    main()

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

人工智能

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

登录后参与评论

0 条评论

热度