大数据采集与预处理技术（微课版）课件 9.4 语音直播数据标注-语音文本转换模型训练（拓展）

大数据采集及预处理技术*

*语音直播标注数据使用序号软件配置要求1Pytorch/2sklearn/一、项目目标：1、了解简单语音转文本模型构建方法。2、能够使用标注的直播数据训练语音转义模型。3、熟悉自然语言处理语音转录模型的测试方法。二、环境要求：importosimporttorchimporttorch.nnasnnimporttorch.optimasoptimfromtorch.utils.dataimportDataset,DataLoaderimportnumpyasnpimportlibrosafromsklearn.model_selectionimporttrain_test_splitfromcollectionsimportCounterimporttorchaudioimporttorchaudio.transformsasTimportpandasaspd1、导入依赖库classSpeechDataset(Dataset):def__init__(self,audio_paths,labels,char_to_idx,sample_rate=16000,max_audio_len=160000):"""audio_paths:音频文件路径列表，labels:对应的文本标签列表，char_to_idx:字符到索引的映射字典，psample_rate:音频采样率，max_audio_len:音频最大长度（超过部分截断，不足部分填充）"""self.audio_paths=audio_pathsself.labels=labelsself.char_to_idx=char_to_idxself.sample_rate=sample_rateself.max_audio_len=max_audio_len#定义MFCC变换

self.mfcc_transform=T.MFCC(sample_rate=self.sample_rate,n_mfcc=40,melkwargs={'n_fft':512,'win_length':400,'hop_length':160,'n_mels':128,'center':False#避免边缘填充

})2、自定义数据集类def__getitem__(self,idx):"""获取单个样本，paramidx:样本索引，return:音频特征（MFCC）和对应的标签索引"""audio_path=self.audio_paths[idx]waveform,sample_rate=torchaudio.load(audio_path)#转换为单声道(取均值)ifwaveform.shape[0]>1:#如果是立体声

waveform=torch.mean(waveform,dim=0,keepdim=True)

#重采样（如果需要）

ifsample_rate!=self.sample_rate:resampler=T.Resample(sample_rate,self.sample_rate)waveform=resampler(waveform)#标准化长度

ifwaveform.shape[1]>self.max_audio_len:waveform=waveform[:,:self.max_audio_len]else:padding_length=self.max_audio_len-waveform.shape[1]waveform=torch.nn.functional.pad(waveform,(0,padding_length))#提取MFCC特征-确保输出是2D(time_steps,n_mfcc)mfcc=self.mfcc_transform(waveform)#(1,n_mfcc,time_steps)mfcc=mfcc.squeeze(0).transpose(0,1)#(time_steps,n_mfcc)#处理标签

label=self.labels[idx]label_indices=[self.char_to_idx.get(char,1)forcharinlabel]#1是<UNK>的索引

label_indices=torch.tensor(label_indices,dtype=torch.long)returnmfcc,label_indices

def__len__(self):"""返回数据集大小"""returnlen(self.audio_paths)classSpeechToTextModel(nn.Module):def__init__(self,input_dim,hidden_dim,output_dim,num_layers=2):"""初始化语音转文本模型，input_dim:输入特征维度（MFCC特征维度），hidden_dim:RNN隐藏层维度，output_dim:输出维度（字符数量），num_layers:RNN层数"""super(SpeechToTextModel,self).__init__()self.rnn=nn.LSTM(input_dim,hidden_dim,num_layers,batch_first=True,bidirectional=True,dropout=0.3ifnum_layers>1else0)self.fc=nn.Linear(hidden_dim*2,output_dim)#双向LSTM，输出维度为hidden_dim*2defforward(self,x,x_lengths):"""前向传播，paramx:输入特征(batch_size,时间步,特征维度)，return:模型输出(batch_size,时间步,输出维度)"""#x:(batch_size,time_steps,input_dim)，x_lengths:各序列实际长度

#打包序列

packed_input=nn.utils.rnn.pack_padded_sequence(x,x_lengths.cpu(),batch_first=True,enforce_sorted=True)packed_output,_=self.rnn(packed_input)#解包序列

output,_=nn.utils.rnn.pad_packed_sequence(packed_output,batch_first=True)logits=self.fc(output)returnlogits3、模型定义#假设标注文件包含音频路径和对应文本标签，CSV文件data=pd.read_csv('custom_dataset.csv')char_counts=Counter() #构建字符到索引的映射forlabelindata['label']:char_counts.update(label)char_to_idx={'<PAD>':0,'<UNK>':1}forchar,countinchar_counts.items():ifcount>=2: #只保留出现次数大于等于2的字符

char_to_idx[char]=len(char_to_idx)#分割数据集为训练集和验证集train_paths,val_paths,train_labels,val_labels=train_test_split(data['audio_path'],data['label'],test_size=0.2,random_state=42)#将PandasSeries转换为列表train_paths=train_paths.tolist()val_paths=val_paths.tolist()train_labels=train_labels.tolist()val_labels=val_labels.tolist()4、数据预处理#创建数据集和数据加载器sample_rate=16000max_audio_len=160000 #10秒音频（假设采样率为16kHz）train_dataset=SpeechDataset(train_paths,train_labels,char_to_idx,sample_rate,max_audio_len)val_dataset=SpeechDataset(val_paths,val_labels,char_to_idx,sample_rate,max_audio_len)train_loader=DataLoader(train_dataset,batch_size=16,shuffle=True,collate_fn=collate_fn)val_loader=DataLoader(val_dataset,batch_size=16,shuffle=False,collate_fn=collate_fn)#超参数input_dim=40 #MFCC特征维度hidden_dim=256 #RNN隐藏层维度output_dim=len(char_to_idx)#输出维度（字符数量）num_layers=2 #RNN层数num_epochs=10 #训练轮数learning_rate=0.001 #学习率4、数据预处理#添加自定义collate_fn处理变长序列defcollate_fn(batch):#按MFCC序列长度排序（降序）

batch.sort(key=lambdax:x[0].shape[0],reverse=True)#分离MFCC和标签

mfccs,labels=zip(*batch)#获取各序列长度

mfcc_lengths=torch.tensor([mfcc.shape[0]formfccinmfccs])max_mfcc_length=mfcc_lengths.max().item()#获取特征维度

feature_dim=mfccs[0].shape[1]#初始化填充后的MFCC张量(batch_size,max_length,feature_dim)padded_mfccs=torch.zeros(len(mfccs),max_mfcc_length,feature_dim)#填充MFCC序列

fori,mfccinenumerate(mfccs):length=mfcc.shape[0]padded_mfccs[i,:length,:]=mfcc#处理标签

label_lengths=torch.tensor([len(label)forlabelinlabels])max_label_length=label_lengths.max().item()#初始化填充后的标签张量(batch_size,max_label_length)padded_labels=torch.zeros(len(labels),max_label_length,dtype=torch.long)#填充标签

fori,labelinenumerate(labels):length=label.shape[0]padded_labels[i,:length]=labelreturnpadded_mfccs,mfcc_lengths,padded_labels,label_lengths#初始化模型、损失函数和优化器model=SpeechToTextModel(input_dim,hidden_dim,output_dim,num_layers)criterion=nn.CTCLoss(blank=0)#blank对应<PAD>标记optimizer=optim.Adam(model.parameters(),lr=learning_rate)#训练循环forepochinrange(num_epochs):model.train()total_loss=0formfccs,mfcc_lengths,labels,label_lengthsintrain_loader:optimizer.zero_grad()logits=model(mfccs,mfcc_lengths)#准备CTC输入(time_steps,batch_size,num_classes)log_probs=logits.transpose(0,1).log_softmax(2)#计算CTC损失

loss=criterion(log_probs,labels,mfcc_lengths//2,#考虑可能的卷积下采样

label_lengths)loss.backward()optimizer.step()total_loss+=loss.item()avg_train_loss=total_loss/len(train_loader)print(f'Epoch[{epoch+1}/{num_epochs}],TrainLoss:{avg_train_loss:.4f}')5、模型训练#验证

model.eval()val_loss=0withtorch.no_grad():formfccs,mfcc_lengths,labels,label_lengthsinval_loader:logits=model(mfccs,mfcc_lengths)log_probs=logits.transpose(0,1).log_softmax(2)loss=criterion(log_probs,labels,mfcc_lengths//2,label_lengths)val_loss+=loss.item()avg_val_loss=val_loss/len(val_loader)print(f'Epoch[{epoch+1}/{num_epochs}],ValLoss:{avg_val_loss:.4f}')#保存模型

"""保存模型和所有必要参数"""state={'model_state_dict':model.state_dict(),'optimizer_state_dict':optimizer.state_dict()ifoptimizerelseNone,'epoch':epoch,'loss':loss,'char_to_idx':char_to_idx,#必须保存字符映射

}torch.save(state,f'speech_to_text_model_epoch{epoch+1}.pth')importosimporttorchfrommodelAudioToTextimportSpeechToTextModel,SpeechDatasetdefload_model(model_path,input_dim=40,hidden_dim=256,device='cpu'):#加载保存的状态

state=torch.load(model_path,map_location=device)#重建字符映射

char_to_idx=state['char_to_idx']idx_to_char={v:kfork,vinchar_to_idx.items()}#初始化模型结构

model=SpeechToTextModel(input_dim=input_dim,hidden_dim=hidden_dim,output_dim=len(char_to_idx),num_layers=2).to(device)#加载模型参数

model.load_state_dict(state['model_state_dict'])model.eval()print(f"模型从{model_path}加载成功")returnmodel,char_to_idx,idx_to_char6、模型加载和推理defpredict_audio(model,audio_path,char_to_idx,idx_to_char,device='cpu'):#预处理音频

dataset=SpeechDataset([audio_path],[''],char_to_idx,sample_rate=16000,max_audio_len=160000)mfcc,_=dataset[0]mfcc=mfcc.unsqueeze(0).to(device)#添加batch维度

mfcc_length=torch.tensor([mfcc.shape[1]]).to(device)#推理

withtorch.no_grad():logits=model(mfcc,mfcc_length)log_probs=torch.log_softmax(logits,dim=2)#贪婪解码

decoded_indices=greedy_decode(log_probs.squeeze(0))decoded_text=''.join([idx_to_char[idx]foridxindecoded_indices])returndecoded_textdefgreedy_decode(log_probs,blank_idx=0):"""CTC贪婪解码"""