1. 构建自定义关键词唤醒系统基础框架
任务目标
构建一个最小可运行的本地关键词唤醒系统,实现从麦克风采集音频、提取特征、检测预设唤醒词、触发响应动作的完整闭环。系统支持自定义唤醒词,通过模板匹配方式实现检测,不依赖云端服务。
环境依赖
- Python 3.8+
- 依赖库:
pyaudio(音频采集)、numpy(数值计算)、scipy(信号处理) - 安装命令:
pip install pyaudio numpy scipy
核心实现
系统采用能量检测进行语音活动检测(VAD),对激活后的音频段提取MFCC特征,与预注册的唤醒词模板计算余弦相似度,超过阈值则触发响应。唤醒词模板通过录制3次样本并取平均特征生成。
import numpy as np
import pyaudio
import wave
import time
from scipy.signal import get_window
from scipy.fftpack import dct
# ---------- 音频参数 ----------
SAMPLE_RATE = 16000
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
VAD_THRESHOLD = 500 # 能量阈值,区分静音/语音
SILENCE_DURATION = 0.6 # 语音结束判定静音时长(秒)
DETECT_THRESHOLD = 0.82 # 唤醒词匹配相似度阈值
# ---------- MFCC特征提取 ----------
def extract_mfcc(audio, n_mfcc=13, n_fft=512, n_filt=26):
"""从音频信号中提取MFCC特征"""
# 预加重
audio = np.append(audio[0], audio[1:] - 0.97 * audio[:-1])
# 分帧
frame_len = int(0.025 * SAMPLE_RATE)
frame_step = int(0.010 * SAMPLE_RATE)
frames = []
for start in range(0, len(audio) - frame_len, frame_step):
frames.append(audio[start:start+frame_len])
if not frames:
return np.array([])
frames = np.array(frames)
# 加窗
frames *= get_window('hamming', frame_len)
# FFT与功率谱
mag = np.abs(np.fft.rfft(frames, n_fft))
power = mag ** 2
# Mel滤波器组
mel_points = np.linspace(0, 2595*np.log10(1+SAMPLE_RATE/2/700), n_filt+2)
hz_points = 700 * (10**(mel_points/2595) - 1)
bin_points = np.floor((n_fft+1) * hz_points / SAMPLE_RATE).astype(int)
filt_bank = np.zeros((n_filt, n_fft//2+1))
for i in range(1, n_filt+1):
left, center, right = bin_points[i-1], bin_points[i], bin_points[i+1]
filt_bank[i-1, left:center] = np.linspace(0, 1, center-left)
filt_bank[i-1, center:right] = np.linspace(1, 0, right-center)
filter_banks = np.dot(power, filt_bank.T)
filter_banks = np.where(filter_banks == 0, np.finfo(float).eps, filter_banks)
filter_banks = 20 * np.log10(filter_banks)
# DCT取MFCC
mfcc = dct(filter_banks, type=2, axis=1, norm='ortho')[:, :n_mfcc]
# 加能量特征
energy = np.sum(power, axis=1, keepdims=True)
energy = np.log10(np.maximum(energy, np.finfo(float).eps))
return np.hstack([mfcc, energy])
def audio_to_feature(audio_data):
"""将原始音频字节转为MFCC特征(时间平均)"""
audio = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
mfcc = extract_mfcc(audio)
if len(mfcc) == 0:
return None
return np.mean(mfcc, axis=0)
# ---------- 唤醒词模板注册 ----------
def register_wake_word(record_seconds=1.2, samples=3):
"""录制并生成唤醒词模板(录制samples次取平均)"""
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT, channels=CHANNELS, rate=SAMPLE_RATE,
input=True, frames_per_buffer=CHUNK)
print(f"请在提示后说唤醒词,共需录制{samples}次...")
templates = []
for i in range(samples):
input("按回车开始录制第%d次..." % (i+1))
frames = []
for _ in range(int(SAMPLE_RATE / CHUNK * record_seconds)):
data = stream.read(CHUNK)
frames.append(data)
audio_data = b''.join(frames)
feat = audio_to_feature(audio_data)
if feat is not None:
templates.append(feat)
print(f"第{i+1}次录制完成")
stream.stop_stream()
stream.close()
p.terminate()
if len(templates) == 0:
raise RuntimeError("模板录制失败")
return np.mean(templates, axis=0)
# ---------- 实时唤醒检测 ----------
def wake_word_detection(wake_template):
"""实时监听麦克风,检测到唤醒词后触发响应"""
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT, channels=CHANNELS, rate=SAMPLE_RATE,
input=True, frames_per_buffer=CHUNK)
print("唤醒检测中... 说唤醒词试试(Ctrl+C退出)")
audio_buffer = []
speech_active = False
silence_frames = 0
try:
while True:
data = stream.read(CHUNK)
audio_array = np.frombuffer(data, dtype=np.int16)
energy = np.sqrt(np.mean(audio_array.astype(np.float32)**2))
if energy > VAD_THRESHOLD and not speech_active:
# 语音开始
speech_active = True
audio_buffer = [data]
silence_frames = 0
elif speech_active:
audio_buffer.append(data)
if energy < VAD_THRESHOLD:
silence_frames += 1
# 静音持续足够长,判定语音结束
if silence_frames > int(SAMPLE_RATE / CHUNK * SILENCE_DURATION):
speech_active = False
full_audio = b''.join(audio_buffer)
# 检查是否为唤醒词
feat = audio_to_feature(full_audio)
if feat is not None:
similarity = np.dot(feat, wake_template) / (
np.linalg.norm(feat) * np.linalg.norm(wake_template) + 1e-8)
if similarity > DETECT_THRESHOLD:
print(f"[唤醒] 检测到唤醒词!相似度={similarity:.3f}")
trigger_response()
audio_buffer = []
silence_frames = 0
else:
silence_frames = 0
except KeyboardInterrupt:
pass
finally:
stream.stop_stream()
stream.close()
p.terminate()
def trigger_response():
"""唤醒后的响应动作(示例:打印提示)"""
print(">>> 设备已唤醒,等待指令...")
# ---------- 主程序入口 ----------
if __name__ == "__main__":
import sys
if len(sys.argv) > 1 and sys.argv[1] == "--register":
# 注册唤醒词模板并保存
template = register_wake_word()
np.save("wake_template.npy", template)
print("唤醒词模板已保存至 wake_template.npy")
else:
# 加载模板并开始检测
try:
template = np.load("wake_template.npy")
except FileNotFoundError:
print("未找到模板文件,请先运行: python wake_system.py --register")
sys.exit(1)
wake_word_detection(template)
运行与验证
- 注册唤醒词:运行
python wake_system.py --register,按提示连续3次说出你的唤醒词(如"小助手"),程序将生成wake_template.npy模板文件。 - 启动检测:运行
python wake_system.py,程序开始监听麦克风。 - 验证通过条件:说出已注册的唤醒词,控制台输出
[唤醒] 检测到唤醒词!即表示系统工作正常。说其他词语不应触发唤醒。
关键参数调优
- VAD_THRESHOLD:环境噪音大时调高,安静环境可调低,避免误触发或漏触发。
- DETECT_THRESHOLD:值越高误唤醒越少但可能漏报;值越低越灵敏但误报增多。建议在0.75~0.90之间调整。
- SILENCE_DURATION:唤醒词说完后停顿超过该时长即判定语音结束,可根据说话习惯调整。
常见问题
- 录音无声音:检查系统麦克风权限,确认
pyaudio能访问设备。 - 频繁误唤醒:调高
DETECT_THRESHOLD或重新录制更清晰的模板。 - 唤醒词检测不到:调低
DETECT_THRESHOLD,或重新录制模板(保持环境安静、语速均匀)。