中文语音识别与偏误检测系统开发

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家，觉得好请收藏。点击跳转到网站。

1. 系统概述

本系统旨在开发一个基于Paraformer模型的中文语音识别与偏误检测系统，能够将输入的音频转换为音素序列，并与标准音素序列进行比对，从而识别语音中的发音错误。该系统可应用于语言学习、语音质量评估等领域。

1.1 系统架构

系统主要包含以下模块：

音频预处理模块
Paraformer语音识别模块
音素转换模块
偏误检测模块
结果可视化模块

1.2 技术选型

核心模型：Paraformer (非自回归端到端语音识别模型)
编程语言：Python 3.8+
深度学习框架：PyTorch
音频处理：librosa, torchaudio
其他依赖：numpy, pandas, matplotlib

2. 环境配置与依赖安装

首先需要配置Python环境并安装必要的依赖包：

# 创建conda环境（可选）
conda create -n paraformer_asr python=3.8
conda activate paraformer_asr# 安装PyTorch (根据CUDA版本选择合适命令)
pip install torch torchaudio torchvision --extra-index-url https://download.pytorch.org/whl/cu113# 安装其他依赖
pip install librosa numpy pandas matplotlib scipy transformers sentencepiece

3. 音频预处理模块

音频预处理是语音识别的重要前置步骤，包括音频加载、重采样、降噪、分帧等操作。

import librosa
import numpy as np
import torchaudio
from scipy import signalclass AudioPreprocessor:def __init__(self, target_sr=16000, frame_length=25, frame_shift=10):"""初始化音频预处理器:param target_sr: 目标采样率:param frame_length: 帧长(ms):param frame_shift: 帧移(ms)"""self.target_sr = target_srself.frame_length = frame_lengthself.frame_shift = frame_shiftdef load_audio(self, audio_path):"""加载音频文件并重采样"""try:# 使用librosa加载音频waveform, sr = librosa.load(audio_path, sr=self.target_sr)return waveform, srexcept Exception as e:print(f"加载音频失败: {e}")return None, Nonedef preemphasis(self, waveform, coeff=0.97):"""预加重处理"""return np.append(waveform[0], waveform[1:] - coeff * waveform[:-1])def framing(self, waveform):"""分帧处理"""frame_size = int(self.frame_length * self.target_sr / 1000)frame_shift = int(self.frame_shift * self.target_sr / 1000)# 计算总帧数num_frames = 1 + (len(waveform) - frame_size) // frame_shiftframes = np.zeros((num_frames, frame_size))for i in range(num_frames):start = i * frame_shiftend = start + frame_sizeframes[i, :] = waveform[start:end]return framesdef add_noise(self, waveform, noise_level=0.005):"""添加随机噪声(数据增强)"""noise = np.random.randn(len(waveform))return waveform + noise_level * noisedef normalize(self, waveform):"""归一化处理"""return waveform / np.max(np.abs(waveform))def extract_features(self, audio_path, add_noise=False):"""提取音频特征"""waveform, sr = self.load_audio(audio_path)if waveform is None:return None# 预处理流程waveform = self.normalize(waveform)if add_noise:waveform = self.add_noise(waveform)waveform = self.preemphasis(waveform)frames = self.framing(waveform)# 计算MFCC特征mfcc_features = []for frame in frames:mfcc = librosa.feature.mfcc(y=frame, sr=sr, n_mfcc=13)mfcc_features.append(mfcc.T)return np.array(mfcc_features)

4. Paraformer模型加载与微调

Paraformer是一种非自回归端到端语音识别模型，具有高效、准确的特点。我们将基于预训练模型进行微调。

4.1 模型加载

import torch
from transformers import AutoModelForSpeechSeq2Seq, AutoProcessorclass ParaformerASR:def __init__(self, model_path="paraformer-zh"):"""初始化Paraformer模型:param model_path: 预训练模型路径或名称"""self.device = "cuda" if torch.cuda.is_available() else "cpu"self.model = Noneself.processor = Noneself.model_path = model_pathself.load_model()def load_model(self):"""加载预训练模型和处理器"""try:self.processor = AutoProcessor.from_pretrained(self.model_path)self.model = AutoModelForSpeechSeq2Seq.from_pretrained(self.model_path, torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32)self.model.to(self.device)print("模型加载成功")except Exception as e:print(f"模型加载失败: {e}")def transcribe(self, audio_path):"""语音识别"""if self.model is None or self.processor is None:print("模型未加载")return Nonetry:# 加载音频文件waveform, sr = torchaudio.load(audio_path)# 重采样到16kHzif sr != 16000:resampler = torchaudio.transforms.Resample(sr, 16000)waveform = resampler(waveform)# 处理音频输入inputs = self.processor(waveform.squeeze().numpy(), sampling_rate=16000, return_tensors="pt", padding=True)inputs = inputs.to(self.device)# 生成识别结果with torch.no_grad():outputs = self.model.generate(**inputs)# 解码结果transcription = self.processor.batch_decode(outputs, skip_special_tokens=True)[0]return transcriptionexcept Exception as e:print(f"语音识别失败: {e}")return Nonedef fine_tune(self, train_dataset, eval_dataset=None, epochs=3, batch_size=8, learning_rate=5e-5):"""微调模型:param train_dataset: 训练数据集:param eval_dataset: 验证数据集(可选):param epochs: 训练轮数:param batch_size: 批次大小:param learning_rate: 学习率"""if self.model is None:print("模型未加载")returnfrom transformers import TrainingArguments, Trainertraining_args = TrainingArguments(output_dir="./results",num_train_epochs=epochs,per_device_train_batch_size=batch_size,per_device_eval_batch_size=batch_size,warmup_steps=500,weight_decay=0.01,logging_dir="./logs",logging_steps=10,evaluation_strategy="epoch" if eval_dataset else "no",save_strategy="epoch",load_best_model_at_end=True if eval_dataset else False,fp16=torch.cuda.is_available(),learning_rate=learning_rate,)trainer = Trainer(model=self.model,args=training_args,train_dataset=train_dataset,eval_dataset=eval_dataset,tokenizer=self.processor.tokenizer,)print("开始微调模型...")trainer.train()print("模型微调完成")

4.2 数据准备与微调

from datasets import Dataset, Audio
import pandas as pddef prepare_dataset(csv_path, audio_dir):"""准备训练数据集:param csv_path: 包含音频路径和转录文本的CSV文件:param audio_dir: 音频文件目录"""# 读取CSV文件df = pd.read_csv(csv_path)# 构建完整音频路径df["audio_path"] = df["audio_file"].apply(lambda x: f"{audio_dir}/{x}")# 验证音频文件是否存在df = df[df["audio_path"].apply(lambda x: os.path.exists(x))]# 创建HuggingFace数据集dataset = Dataset.from_pandas(df)# 加载音频列dataset = dataset.cast_column("audio_path", Audio())return datasetdef preprocess_dataset(dataset, processor):"""预处理数据集:param dataset: 原始数据集:param processor: Paraformer处理器"""def prepare_example(batch):audio = batch["audio_path"]# 处理音频inputs = processor(audio["array"], sampling_rate=audio["sampling_rate"], text=batch["transcription"],return_tensors="pt",padding=True,truncation=True)return inputs# 映射预处理函数dataset = dataset.map(prepare_example,remove_columns=dataset.column_names,batched=True,batch_size=4)return dataset

5. 音素转换模块

将识别出的文本转换为音素序列，便于与标准音素序列进行比对。

import pypinyin
from pypinyin import Styleclass PhonemeConverter:def __init__(self):"""初始化音素转换器"""# 定义音素映射表self.phoneme_map = {'a': 'a', 'ai': 'ai', 'an': 'an', 'ang': 'ang', 'ao': 'ao','ba': 'b a', 'bai': 'b ai', 'ban': 'b an', 'bang': 'b ang',# 完整的音素映射表...}def text_to_phonemes(self, text):"""将中文文本转换为音素序列"""# 获取拼音pinyin_list = pypinyin.lazy_pinyin(text, style=Style.TONE3)# 转换为音素phoneme_sequence = []for pinyin in pinyin_list:# 去除声调数字base_pinyin = ''.join([c for c in pinyin if not c.isdigit()])# 查找音素映射if base_pinyin in self.phoneme_map:phonemes = self.phoneme_map[base_pinyin].split()phoneme_sequence.extend(phonemes)return ' '.join(phoneme_sequence)def compare_phonemes(self, reference, hypothesis):"""比较参考音素序列和假设音素序列:param reference: 标准音素序列:param hypothesis: 识别出的音素序列:return: 错误列表，包含错误类型和位置"""ref_phonemes = reference.split()hyp_phonemes = hypothesis.split()errors = []min_len = min(len(ref_phonemes), len(hyp_phonemes))# 比对音素for i in range(min_len):if ref_phonemes[i] != hyp_phonemes[i]:errors.append({'position': i,'reference': ref_phonemes[i],'hypothesis': hyp_phonemes[i],'type': 'substitution'})# 处理插入或删除错误if len(ref_phonemes) > len(hyp_phonemes):for i in range(min_len, len(ref_phonemes)):errors.append({'position': i,'reference': ref_phonemes[i],'hypothesis': None,'type': 'deletion'})elif len(hyp_phonemes) > len(ref_phonemes):for i in range(min_len, len(hyp_phonemes)):errors.append({'position': i,'reference': None,'hypothesis': hyp_phonemes[i],'type': 'insertion'})return errors

6. 偏误检测与分析模块

基于音素序列比对结果，检测发音错误并进行统计分析。

class ErrorAnalyzer:def __init__(self):"""初始化错误分析器"""self.error_types = {'substitution': '替换错误','insertion': '插入错误','deletion': '删除错误'}def analyze_errors(self, errors):"""分析发音错误:param errors: 错误列表:return: 分析结果字典"""if not errors:return {'total_errors': 0,'error_distribution': {},'common_errors': [],'error_rate': 0.0}# 统计错误类型分布error_dist = {et: 0 for et in self.error_types.values()}for error in errors:error_dist[self.error_types[error['type']]] += 1# 统计常见错误对error_pairs = {}for error in errors:if error['type'] == 'substitution':pair = (error['reference'], error['hypothesis'])error_pairs[pair] = error_pairs.get(pair, 0) + 1# 排序常见错误common_errors = sorted(error_pairs.items(), key=lambda x: x[1], reverse=True)[:5]# 计算错误率total_phonemes = len(errors) + sum(1 for e in errors if e['type'] == 'deletion')error_rate = len(errors) / total_phonemes if total_phonemes > 0 else 0return {'total_errors': len(errors),'error_distribution': error_dist,'common_errors': common_errors,'error_rate': error_rate}def generate_report(self, analysis_result):"""生成错误分析报告"""report = []report.append(f"总错误数: {analysis_result['total_errors']}")report.append(f"错误率: {analysis_result['error_rate']:.2%}")report.append("\n错误类型分布:")for et, count in analysis_result['error_distribution'].items():report.append(f"  {et}: {count}")report.append("\n常见错误替换:")for (ref, hyp), count in analysis_result['common_errors']:report.append(f"  {ref} → {hyp}: {count}次")return "\n".join(report)

7. 系统集成与用户界面

将各模块集成到一个完整的系统中，并提供简单的用户界面。

import os
import json
from datetime import datetimeclass SpeechErrorDetectionSystem:def __init__(self, model_path="paraformer-zh"):"""初始化语音偏误检测系统:param model_path: Paraformer模型路径"""self.audio_preprocessor = AudioPreprocessor()self.asr_model = ParaformerASR(model_path)self.phoneme_converter = PhonemeConverter()self.error_analyzer = ErrorAnalyzer()self.results_dir = "./results"# 创建结果目录os.makedirs(self.results_dir, exist_ok=True)def process_audio(self, audio_path, reference_text):"""处理音频文件并检测发音错误:param audio_path: 音频文件路径:param reference_text: 参考文本:return: 处理结果字典"""# 1. 语音识别recognized_text = self.asr_model.transcribe(audio_path)if recognized_text is None:return None# 2. 转换为音素序列reference_phonemes = self.phoneme_converter.text_to_phonemes(reference_text)hypothesis_phonemes = self.phoneme_converter.text_to_phonemes(recognized_text)# 3. 比对音素序列errors = self.phoneme_converter.compare_phonemes(reference_phonemes, hypothesis_phonemes)# 4. 分析错误analysis_result = self.error_analyzer.analyze_errors(errors)# 5. 保存结果result = {'audio_file': os.path.basename(audio_path),'reference_text': reference_text,'recognized_text': recognized_text,'reference_phonemes': reference_phonemes,'hypothesis_phonemes': hypothesis_phonemes,'errors': errors,'analysis': analysis_result,'timestamp': datetime.now().isoformat()}# 保存为JSON文件result_file = os.path.join(self.results_dir,f"result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")with open(result_file, 'w', encoding='utf-8') as f:json.dump(result, f, ensure_ascii=False, indent=2)return resultdef visualize_results(self, result):"""可视化分析结果"""import matplotlib.pyplot as plt# 错误类型分布饼图error_dist = result['analysis']['error_distribution']labels = [et for et in error_dist.keys() if error_dist[et] > 0]sizes = [error_dist[et] for et in labels]plt.figure(figsize=(12, 5))plt.subplot(1, 2, 1)plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90)plt.title('错误类型分布')# 常见错误条形图common_errors = result['analysis']['common_errors']if common_errors:plt.subplot(1, 2, 2)pairs = [f"{ref}→{hyp}" for (ref, hyp), _ in common_errors]counts = [count for _, count in common_errors]plt.bar(pairs, counts)plt.title('常见替换错误')plt.xticks(rotation=45)plt.tight_layout()plt.show()def run_interactive(self):"""交互式运行系统"""print("=== 中文语音偏误检测系统 ===")while True:print("\n选项:")print("1. 检测音频文件")print("2. 批量处理目录")print("3. 退出")choice = input("请选择操作: ")if choice == "1":audio_path = input("输入音频文件路径: ")if not os.path.exists(audio_path):print("文件不存在")continuereference_text = input("输入参考文本: ")result = self.process_audio(audio_path, reference_text)if result:print("\n识别结果:")print(f"参考文本: {result['reference_text']}")print(f"识别结果: {result['recognized_text']}")print("\n音素比对:")print(f"参考音素: {result['reference_phonemes']}")print(f"识别音素: {result['hypothesis_phonemes']}")print("\n错误分析:")print(self.error_analyzer.generate_report(result['analysis']))self.visualize_results(result)elif choice == "2":dir_path = input("输入音频目录路径: ")if not os.path.isdir(dir_path):print("目录不存在")continue# 假设目录中有对应的参考文本文件for audio_file in os.listdir(dir_path):if audio_file.endswith(('.wav', '.mp3')):audio_path = os.path.join(dir_path, audio_file)text_file = os.path.splitext(audio_file)[0] + ".txt"text_path = os.path.join(dir_path, text_file)if os.path.exists(text_path):with open(text_path, 'r', encoding='utf-8') as f:reference_text = f.read().strip()print(f"\n处理文件: {audio_file}")result = self.process_audio(audio_path, reference_text)if result:print(f"错误数: {result['analysis']['total_errors']}")print(f"错误率: {result['analysis']['error_rate']:.2%}")elif choice == "3":print("退出系统")breakelse:print("无效选择")

8. 模型评估与优化

8.1 评估指标

def evaluate_model(model, test_dataset, processor):"""评估模型性能"""from transformers import EvalPredictionimport numpy as npdef compute_metrics(p: EvalPrediction):pred_ids = p.predictionslabel_ids = p.label_ids# 解码预测和标签pred_str = processor.batch_decode(pred_ids, skip_special_tokens=True)label_str = processor.batch_decode(label_ids, skip_special_tokens=True)# 计算WER(词错误率)wer = calculate_wer(pred_str, label_str)# 计算CER(字错误率)cer = calculate_cer(pred_str, label_str)return {"wer": wer, "cer": cer}trainer = Trainer(model=model,eval_dataset=test_dataset,tokenizer=processor.tokenizer,compute_metrics=compute_metrics)return trainer.evaluate()def calculate_wer(predictions, references):"""计算词错误率"""from jiwer import werreturn wer(references, predictions)def calculate_cer(predictions, references):"""计算字错误率"""from jiwer import cerreturn cer(references, predictions)

8.2 模型优化技术

class ModelOptimizer:def __init__(self, model, processor):self.model = modelself.processor = processordef apply_quantization(self):"""应用动态量化减小模型大小"""from torch.quantization import quantize_dynamicself.model = quantize_dynamic(self.model,{torch.nn.Linear},dtype=torch.qint8)return self.modeldef apply_pruning(self, amount=0.2):"""应用权重剪枝"""from torch.nn.utils import prune# 对模型中的所有线性层进行剪枝for name, module in self.model.named_modules():if isinstance(module, torch.nn.Linear):prune.l1_unstructured(module, name='weight', amount=amount)prune.remove(module, 'weight')return self.modeldef optimize_for_inference(self):"""优化模型用于推理"""self.model.eval()# 应用脚本化if hasattr(torch.jit, 'script'):self.model = torch.jit.script(self.model)return self.model

9. 系统部署与应用

9.1 Flask Web API

from flask import Flask, request, jsonify
import uuidapp = Flask(__name__)
system = SpeechErrorDetectionSystem()@app.route('/api/analyze', methods=['POST'])
def analyze_audio():if 'audio' not in request.files or 'text' not in request.form:return jsonify({'error': 'Missing audio file or reference text'}), 400audio_file = request.files['audio']reference_text = request.form['text']# 保存临时文件temp_dir = "temp_uploads"os.makedirs(temp_dir, exist_ok=True)audio_path = os.path.join(temp_dir, f"{uuid.uuid4()}.wav")audio_file.save(audio_path)# 处理音频result = system.process_audio(audio_path, reference_text)# 删除临时文件os.remove(audio_path)if result is None:return jsonify({'error': 'Audio processing failed'}), 500return jsonify(result)if __name__ == '__main__':app.run(host='0.0.0.0', port=5000, debug=True)

9.2 Gradio交互界面

import gradio as grdef create_gradio_interface():system = SpeechErrorDetectionSystem()def analyze_audio(audio_file, reference_text):result = system.process_audio(audio_file, reference_text)if result is None:return "处理失败，请重试"report = [f"参考文本: {result['reference_text']}",f"识别结果: {result['recognized_text']}","\n错误分析:",system.error_analyzer.generate_report(result['analysis'])]return "\n".join(report)interface = gr.Interface(fn=analyze_audio,inputs=[gr.Audio(source="upload", type="filepath", label="上传音频"),gr.Textbox(lines=2, label="参考文本")],outputs=gr.Textbox(lines=10, label="分析结果"),title="中文语音偏误检测系统",description="上传音频文件和参考文本，系统将检测发音错误")return interfaceif __name__ == "__main__":interface = create_gradio_interface()interface.launch()

10. 系统测试与验证

10.1 单元测试

import unittest
import tempfileclass TestSpeechErrorDetectionSystem(unittest.TestCase):@classmethoddef setUpClass(cls):cls.system = SpeechErrorDetectionSystem()cls.test_audio = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)# 创建一个简单的测试音频文件import soundfile as sfimport numpy as npsf.write(cls.test_audio.name, np.random.randn(16000), 16000)cls.test_audio.close()def test_audio_preprocessing(self):preprocessor = AudioPreprocessor()features = preprocessor.extract_features(self.test_audio.name)self.assertIsNotNone(features)self.assertGreater(features.shape[0], 0)def test_phoneme_conversion(self):text = "你好世界"phonemes = self.system.phoneme_converter.text_to_phonemes(text)self.assertIsInstance(phonemes, str)self.assertGreater(len(phonemes), 0)def test_error_analysis(self):ref_phonemes = "n i h a o"hyp_phonemes = "n i h o a"errors = self.system.phoneme_converter.compare_phonemes(ref_phonemes, hyp_phonemes)self.assertEqual(len(errors), 2)self.assertEqual(errors[0]['type'], 'substitution')def test_full_pipeline(self):result = self.system.process_audio(self.test_audio.name, "测试文本")self.assertIsNotNone(result)self.assertIn('analysis', result)@classmethoddef tearDownClass(cls):os.unlink(cls.test_audio.name)if __name__ == '__main__':unittest.main()

10.2 性能测试

import time
import statisticsdef performance_test(system, audio_files, reference_texts, iterations=10):"""系统性能测试"""times = []memory_usage = []for i in range(iterations):start_time = time.time()# 测试内存使用(近似)import psutilprocess = psutil.Process()start_mem = process.memory_info().rss / 1024 / 1024  # MBfor audio, text in zip(audio_files, reference_texts):system.process_audio(audio, text)end_time = time.time()end_mem = process.memory_info().rss / 1024 / 1024times.append(end_time - start_time)memory_usage.append(end_mem - start_mem)return {'avg_time': statistics.mean(times),'std_time': statistics.stdev(times),'avg_memory': statistics.mean(memory_usage),'std_memory': statistics.stdev(memory_usage),'iterations': iterations}