中文語音識別與偏誤檢測系統開發

前些天發現了一個巨牛的人工智能學習網站，通俗易懂，風趣幽默，忍不住分享一下給大家，覺得好請收藏。點擊跳轉到網站。

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}