当前位置：首页 > news >正文

嵌入式设备轻量级语音识别实战：从STM32到树莓派的智能语音控制

news 2025/10/14 23:14:22

前言

随着AIoT（人工智能物联网）的快速发展，将AI能力下沉到边缘设备已成为趋势。本文将详细介绍如何在资源受限的嵌入式设备上部署轻量级语音识别模型，实现简单的语音控制功能，如"开灯"、“关灯”、"设置闹钟"等命令识别。

一、技术架构概述

1.1 系统架构

在嵌入式设备上实现语音识别，主要包含以下几个模块：

[麦克风] → [音频采集] → [预处理] → [特征提取] → [模型推理] → [命令执行]

1.2 技术选型

硬件平台对比

平台	主频	RAM	Flash	适用场景	优缺点
STM32F407	168MHz	192KB	1MB	简单命令识别	功耗低，资源极限
STM32H7	480MHz	1MB	2MB	中等复杂度	性能较强，成本适中
树莓派4B	1.5GHz	2-8GB	SD卡	复杂应用	性能强，功耗较高

模型选择

关键词识别（KWS）: 适用于STM32等MCU
TinyML模型: 经过量化优化的神经网络
Edge Impulse: 端到端的嵌入式ML平台

二、环境搭建

2.1 树莓派环境配置

# 更新系统
sudo apt-get update && sudo apt-get upgrade# 安装音频处理库
sudo apt-get install portaudio19-dev python3-pyaudio
sudo apt-get install libatlas-base-dev# 安装Python依赖
pip3 install numpy scipy
pip3 install tensorflow-lite
pip3 install sounddevice
pip3 install librosa

2.2 STM32开发环境

STM32CubeMX: 配置硬件资源
STM32CubeIDE: 集成开发环境
X-CUBE-AI: STM32的AI扩展包

三、轻量级语音识别模型实现

3.1 数据采集与预处理

音频采集模块（Python - 树莓派版本）

import sounddevice as sd
import numpy as np
import queue
import threadingclass AudioCapture:def __init__(self, sample_rate=16000, channels=1, chunk_size=512):self.sample_rate = sample_rateself.channels = channelsself.chunk_size = chunk_sizeself.audio_queue = queue.Queue()self.is_recording = Falsedef audio_callback(self, indata, frames, time, status):"""音频流回调函数"""if status:print(f"音频状态: {status}")self.audio_queue.put(indata.copy())def start_recording(self):"""开始录音"""self.is_recording = Trueself.stream = sd.InputStream(callback=self.audio_callback,channels=self.channels,samplerate=self.sample_rate,blocksize=self.chunk_size)self.stream.start()def stop_recording(self):"""停止录音"""self.is_recording = Falseself.stream.stop()self.stream.close()def get_audio_data(self, duration=1.0):"""获取指定时长的音频数据"""frames_needed = int(self.sample_rate * duration)audio_data = []while len(audio_data) < frames_needed:if not self.audio_queue.empty():chunk = self.audio_queue.get()audio_data.extend(chunk[:, 0])return np.array(audio_data[:frames_needed])

3.2 特征提取 - MFCC

import librosa
import numpy as npclass FeatureExtractor:def __init__(self, n_mfcc=13, n_fft=512, hop_length=256):self.n_mfcc = n_mfccself.n_fft = n_fftself.hop_length = hop_lengthdef extract_mfcc(self, audio_data, sample_rate=16000):"""提取MFCC特征"""# 预加重pre_emphasis = 0.97emphasized_signal = np.append(audio_data[0], audio_data[1:] - pre_emphasis * audio_data[:-1])# 提取MFCCmfcc = librosa.feature.mfcc(y=emphasized_signal,sr=sample_rate,n_mfcc=self.n_mfcc,n_fft=self.n_fft,hop_length=self.hop_length)# 归一化mfcc_normalized = (mfcc - np.mean(mfcc)) / np.std(mfcc)return mfcc_normalized.Tdef prepare_input(self, mfcc_features, max_length=100):"""准备模型输入"""# 填充或截断到固定长度if len(mfcc_features) < max_length:pad_width = ((0, max_length - len(mfcc_features)), (0, 0))mfcc_padded = np.pad(mfcc_features, pad_width, mode='constant')else:mfcc_padded = mfcc_features[:max_length]return mfcc_padded

3.3 轻量级CNN模型设计

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layersdef create_lightweight_model(input_shape=(100, 13), num_classes=5):"""创建轻量级CNN模型输入: MFCC特征 (时间步, 特征维度)输出: 命令类别 (开灯, 关灯, 设置闹钟, 停止, 其他)"""model = keras.Sequential([# 输入层layers.Input(shape=input_shape),# 特征提取层 - 使用深度可分离卷积减少参数layers.Reshape((input_shape[0], input_shape[1], 1)),# 第一个卷积块layers.SeparableConv2D(8, (3, 3), padding='same', activation='relu'),layers.BatchNormalization(),layers.MaxPooling2D((2, 2)),# 第二个卷积块layers.SeparableConv2D(16, (3, 3), padding='same', activation='relu'),layers.BatchNormalization(),layers.MaxPooling2D((2, 2)),# 全局平均池化（减少参数）layers.GlobalAveragePooling2D(),# 分类层layers.Dense(32, activation='relu'),layers.Dropout(0.3),layers.Dense(num_classes, activation='softmax')])return model# 模型量化函数
def quantize_model(model, representative_dataset):"""将模型量化为TFLite格式"""converter = tf.lite.TFLiteConverter.from_keras_model(model)converter.optimizations = [tf.lite.Optimize.DEFAULT]converter.representative_dataset = representative_dataset# INT8量化converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]converter.inference_input_type = tf.int8converter.inference_output_type = tf.int8tflite_model = converter.convert()return tflite_model

3.4 模型训练代码

import os
import numpy as np
from sklearn.model_selection import train_test_splitclass VoiceCommandTrainer:def __init__(self):self.commands = ['turn_on_light', 'turn_off_light', 'set_alarm', 'stop', 'unknown']self.model = Noneself.feature_extractor = FeatureExtractor()def prepare_dataset(self, data_dir):"""准备训练数据集"""X, y = [], []for idx, command in enumerate(self.commands):command_dir = os.path.join(data_dir, command)for audio_file in os.listdir(command_dir):if audio_file.endswith('.wav'):file_path = os.path.join(command_dir, audio_file)# 加载音频audio, sr = librosa.load(file_path, sr=16000)# 提取特征mfcc = self.feature_extractor.extract_mfcc(audio)mfcc_input = self.feature_extractor.prepare_input(mfcc)X.append(mfcc_input)y.append(idx)return np.array(X), np.array(y)def train(self, X, y, epochs=50, batch_size=32):"""训练模型"""# 划分数据集X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)# 创建模型self.model = create_lightweight_model()# 编译模型self.model.compile(optimizer='adam',loss='sparse_categorical_crossentropy',metrics=['accuracy'])# 训练history = self.model.fit(X_train, y_train,validation_data=(X_test, y_test),epochs=epochs,batch_size=batch_size,verbose=1)return historydef export_tflite(self, save_path='voice_command.tflite'):"""导出TFLite模型"""def representative_dataset():for _ in range(100):data = np.random.rand(1, 100, 13).astype(np.float32)yield [data]tflite_model = quantize_model(self.model, representative_dataset)with open(save_path, 'wb') as f:f.write(tflite_model)print(f"模型已保存到: {save_path}")print(f"模型大小: {len(tflite_model) / 1024:.2f} KB")

3.5 实时推理引擎

import tensorflow as tf
import numpy as np
import threading
import timeclass VoiceCommandInference:def __init__(self, model_path='voice_command.tflite'):# 加载TFLite模型self.interpreter = tf.lite.Interpreter(model_path=model_path)self.interpreter.allocate_tensors()# 获取输入输出细节self.input_details = self.interpreter.get_input_details()self.output_details = self.interpreter.get_output_details()# 命令映射self.commands = {0: 'turn_on_light',1: 'turn_off_light',2: 'set_alarm',3: 'stop',4: 'unknown'}# 初始化组件self.audio_capture = AudioCapture()self.feature_extractor = FeatureExtractor()# 控制标志self.is_running = Falsedef preprocess_audio(self, audio_data):"""预处理音频数据"""# 提取MFCC特征mfcc = self.feature_extractor.extract_mfcc(audio_data)mfcc_input = self.feature_extractor.prepare_input(mfcc)# 转换为模型输入格式input_data = np.expand_dims(mfcc_input, axis=0).astype(np.float32)# 量化（如果需要）input_scale = self.input_details[0]['quantization'][0]input_zero_point = self.input_details[0]['quantization'][1]if input_scale != 0:input_data = (input_data / input_scale + input_zero_point).astype(np.int8)return input_datadef predict(self, audio_data):"""执行推理"""# 预处理input_data = self.preprocess_audio(audio_data)# 设置输入self.interpreter.set_tensor(self.input_details[0]['index'], input_data)# 推理start_time = time.time()self.interpreter.invoke()inference_time = (time.time() - start_time) * 1000# 获取输出output_data = self.interpreter.get_tensor(self.output_details[0]['index'])# 反量化（如果需要）output_scale = self.output_details[0]['quantization'][0]output_zero_point = self.output_details[0]['quantization'][1]if output_scale != 0:output_data = (output_data.astype(np.float32) - output_zero_point) * output_scale# 获取预测结果prediction = np.argmax(output_data[0])confidence = np.max(output_data[0])return {'command': self.commands[prediction],'confidence': float(confidence),'inference_time_ms': inference_time}def execute_command(self, command, confidence):"""执行识别到的命令"""if confidence < 0.7:print("置信度过低，忽略命令")returnif command == 'turn_on_light':print("执行: 开灯")# GPIO.output(LED_PIN, GPIO.HIGH)  # 树莓派GPIO控制elif command == 'turn_off_light':print("执行: 关灯")# GPIO.output(LED_PIN, GPIO.LOW)elif command == 'set_alarm':print("执行: 设置闹钟")# 调用闹钟设置函数elif command == 'stop':print("执行: 停止")self.is_running = Falsedef run(self):"""主运行循环"""print("语音识别系统启动...")self.is_running = Trueself.audio_capture.start_recording()try:while self.is_running:# 获取1秒音频数据audio_data = self.audio_capture.get_audio_data(duration=1.0)# 简单的VAD（语音活动检测）if np.max(np.abs(audio_data)) > 0.01:# 执行推理result = self.predict(audio_data)print(f"识别结果: {result['command']}, "f"置信度: {result['confidence']:.2f}, "f"推理时间: {result['inference_time_ms']:.2f}ms")# 执行命令self.execute_command(result['command'], result['confidence'])time.sleep(0.1)  # 短暂延迟except KeyboardInterrupt:print("\n系统停止")finally:self.audio_capture.stop_recording()

四、STM32部署方案

4.1 STM32 C代码实现

// voice_recognition.h
#ifndef VOICE_RECOGNITION_H
#define VOICE_RECOGNITION_H#include "stm32f4xx_hal.h"
#include "arm_math.h"#define SAMPLE_RATE     16000
#define FRAME_SIZE      512
#define MFCC_COEFFS     13
#define MAX_FRAMES      100typedef struct {float32_t buffer[FRAME_SIZE];float32_t mfcc_features[MAX_FRAMES][MFCC_COEFFS];uint32_t frame_count;
} AudioProcessor;typedef enum {CMD_TURN_ON_LIGHT = 0,CMD_TURN_OFF_LIGHT,CMD_SET_ALARM,CMD_STOP,CMD_UNKNOWN
} VoiceCommand;// 函数声明
void AudioProcessor_Init(AudioProcessor* processor);
void AudioProcessor_ProcessFrame(AudioProcessor* processor, int16_t* samples);
VoiceCommand AudioProcessor_RunInference(AudioProcessor* processor);#endif// voice_recognition.c
#include "voice_recognition.h"
#include "ai_platform.h"
#include "network.h"  // STM32CubeMX生成的网络头文件// 全局变量
static ai_handle network = AI_HANDLE_NULL;
static ai_network_report network_info;// AI缓冲区
AI_ALIGNED(4)
static ai_u8 activations[AI_NETWORK_DATA_ACTIVATIONS_SIZE];AI_ALIGNED(4)
static ai_float in_data[AI_NETWORK_IN_1_SIZE];AI_ALIGNED(4)
static ai_float out_data[AI_NETWORK_OUT_1_SIZE];// 初始化音频处理器
void AudioProcessor_Init(AudioProcessor* processor) {memset(processor, 0, sizeof(AudioProcessor));// 初始化AI网络ai_error err;const ai_handle acts[] = {activations};err = ai_network_create_and_init(&network, acts, NULL);if (err.type != AI_ERROR_NONE) {Error_Handler();}
}// MFCC特征提取（简化版）
static void Extract_MFCC(float32_t* audio_frame, float32_t* mfcc_out) {static float32_t fft_buffer[FRAME_SIZE * 2];static float32_t power_spectrum[FRAME_SIZE];// 1. 预加重float32_t pre_emphasis = 0.97f;for (int i = FRAME_SIZE - 1; i > 0; i--) {audio_frame[i] = audio_frame[i] - pre_emphasis * audio_frame[i-1];}// 2. 加窗（汉明窗）for (int i = 0; i < FRAME_SIZE; i++) {float32_t window = 0.54f - 0.46f * cosf(2 * PI * i / (FRAME_SIZE - 1));audio_frame[i] *= window;}// 3. FFTarm_rfft_fast_instance_f32 fft_inst;arm_rfft_fast_init_f32(&fft_inst, FRAME_SIZE);arm_rfft_fast_f32(&fft_inst, audio_frame, fft_buffer, 0);// 4. 功率谱arm_cmplx_mag_squared_f32(fft_buffer, power_spectrum, FRAME_SIZE/2);// 5. Mel滤波器组（简化实现）// 这里需要预先计算好Mel滤波器系数static const int num_mel_filters = 26;float32_t mel_energies[num_mel_filters];// 应用Mel滤波器（简化）for (int i = 0; i < num_mel_filters; i++) {mel_energies[i] = 0;for (int j = 0; j < FRAME_SIZE/2; j++) {// mel_energies[i] += power_spectrum[j] * mel_filter_bank[i][j];}mel_energies[i] = log10f(mel_energies[i] + 1e-10f);}// 6. DCT得到MFCCarm_dct4_instance_f32 dct_inst;arm_dct4_init_f32(&dct_inst, &arm_rfft_fast_instance_f32, &arm_cfft_radix4_instance_f32, num_mel_filters, num_mel_filters/2, 0.125f);// 执行DCT（取前13个系数）for (int i = 0; i < MFCC_COEFFS; i++) {mfcc_out[i] = mel_energies[i];  // 简化处理}
}// 处理音频帧
void AudioProcessor_ProcessFrame(AudioProcessor* processor, int16_t* samples) {// 转换为floatfor (int i = 0; i < FRAME_SIZE; i++) {processor->buffer[i] = (float32_t)samples[i] / 32768.0f;}// 提取MFCC特征if (processor->frame_count < MAX_FRAMES) {Extract_MFCC(processor->buffer, processor->mfcc_features[processor->frame_count]);processor->frame_count++;}
}// 运行推理
VoiceCommand AudioProcessor_RunInference(AudioProcessor* processor) {// 准备输入数据memcpy(in_data, processor->mfcc_features, MAX_FRAMES * MFCC_COEFFS * sizeof(float32_t));// 归一化float32_t mean = 0, std_dev = 0;arm_mean_f32(in_data, MAX_FRAMES * MFCC_COEFFS, &mean);arm_std_f32(in_data, MAX_FRAMES * MFCC_COEFFS, &std_dev);for (int i = 0; i < MAX_FRAMES * MFCC_COEFFS; i++) {in_data[i] = (in_data[i] - mean) / (std_dev + 1e-8f);}// 设置输入输出缓冲区ai_input[0].data = AI_HANDLE_PTR(in_data);ai_output[0].data = AI_HANDLE_PTR(out_data);// 运行推理ai_i32 n_batch = ai_network_run(network, &ai_input[0], &ai_output[0]);if (n_batch != 1) {return CMD_UNKNOWN;}// 找到最大概率的类别int max_idx = 0;float32_t max_prob = out_data[0];for (int i = 1; i < 5; i++) {if (out_data[i] > max_prob) {max_prob = out_data[i];max_idx = i;}}// 置信度阈值if (max_prob < 0.7f) {return CMD_UNKNOWN;}return (VoiceCommand)max_idx;
}// main.c 主程序示例
int main(void) {HAL_Init();SystemClock_Config();// 初始化外设MX_GPIO_Init();MX_I2S_Init();MX_DMA_Init();// 初始化音频处理器AudioProcessor processor;AudioProcessor_Init(&processor);// 音频缓冲区int16_t audio_buffer[FRAME_SIZE];while (1) {// 从I2S/PDM麦克风读取音频HAL_I2S_Receive(&hi2s2, (uint16_t*)audio_buffer, FRAME_SIZE, HAL_MAX_DELAY);// 处理音频帧AudioProcessor_ProcessFrame(&processor, audio_buffer);// 每秒执行一次推理if (processor.frame_count >= MAX_FRAMES) {VoiceCommand cmd = AudioProcessor_RunInference(&processor);// 执行命令switch(cmd) {case CMD_TURN_ON_LIGHT:HAL_GPIO_WritePin(LED_GPIO_Port, LED_Pin, GPIO_PIN_SET);break;case CMD_TURN_OFF_LIGHT:HAL_GPIO_WritePin(LED_GPIO_Port, LED_Pin, GPIO_PIN_RESET);break;case CMD_SET_ALARM:// 设置RTC闹钟Set_Alarm();break;case CMD_STOP:// 停止某个操作break;default:break;}// 重置处理器processor.frame_count = 0;}HAL_Delay(10);}
}

五、优化技巧

5.1 模型优化策略

量化技术
- INT8量化：将FP32模型转换为INT8，减少75%的模型大小
- 动态范围量化：保持激活函数动态量化
- 全整数量化：输入输出都是整数，适合MCU

模型剪枝

# TensorFlow模型剪枝示例
import tensorflow_model_optimization as tfmotprune_low_magnitude = tfmot.sparsity.keras.prune_low_magnitude# 定义剪枝参数
pruning_params = {'pruning_schedule': tfmot.sparsity.keras.PolynomialDecay(initial_sparsity=0.30,final_sparsity=0.70,begin_step=0,end_step=end_step)
}# 应用剪枝
model_for_pruning = prune_low_magnitude(model, **pruning_params)