大數(shù)據(jù)采集及預(yù)處理技術(shù)*

*語(yǔ)音直播標(biāo)注數(shù)據(jù)使用序號(hào)軟件配置要求1Pytorch/2sklearn/一、項(xiàng)目目標(biāo)：1、了解簡(jiǎn)單語(yǔ)音轉(zhuǎn)文本模型構(gòu)建方法。2、能夠使用標(biāo)注的直播數(shù)據(jù)訓(xùn)練語(yǔ)音轉(zhuǎn)義模型。3、熟悉自然語(yǔ)言處理語(yǔ)音轉(zhuǎn)錄模型的測(cè)試方法。二、環(huán)境要求：importosimporttorchimporttorch.nnasnnimporttorch.optimasoptimfromtorch.utils.dataimportDataset,DataLoaderimportnumpyasnpimportlibrosafromsklearn.model_selectionimporttrain_test_splitfromcollectionsimportCounterimporttorchaudioimporttorchaudio.transformsasTimportpandasaspd1、導(dǎo)入依賴庫(kù)classSpeechDataset(Dataset):def__init__(self,audio_paths,labels,char_to_idx,sample_rate=16000,max_audio_len=160000):"""audio_paths:音頻文件路徑列表，labels:對(duì)應(yīng)的文本標(biāo)簽列表，char_to_idx:字符到索引的映射字典，psample_rate:音頻采樣率，max_audio_len:音頻最大長(zhǎng)度（超過(guò)部分截?cái)啵蛔悴糠痔畛洌?""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、自定義數(shù)據(jù)集類def__getitem__(self,idx):"""獲取單個(gè)樣本，paramidx:樣本索引，return:音頻特征（MFCC）和對(duì)應(yīng)的標(biāo)簽索引"""audio_path=self.audio_paths[idx]waveform,sample_rate=torchaudio.load(audio_path)#轉(zhuǎn)換為單聲道(取均值)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)#標(biāo)準(zhǔn)化長(zhǎng)度

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)#處理標(biāo)簽

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):"""返回?cái)?shù)據(jù)集大小"""returnlen(self.audio_paths)classSpeechToTextModel(nn.Module):def__init__(self,input_dim,hidden_dim,output_dim,num_layers=2):"""初始化語(yǔ)音轉(zhuǎn)文本模型，input_dim:輸入特征維度（MFCC特征維度），hidden_dim:RNN隱藏層維度，output_dim:輸出維度（字符數(shù)量），num_layers:RNN層數(shù)"""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,時(shí)間步,特征維度)，return:模型輸出(batch_size,時(shí)間步,輸出維度)"""#x:(batch_size,time_steps,input_dim)，x_lengths:各序列實(shí)際長(zhǎng)度

#打包序列

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、模型定義#假設(shè)標(biāo)注文件包含音頻路徑和對(duì)應(yīng)文本標(biāo)簽，CSV文件data=pd.read_csv('custom_dataset.csv')char_counts=Counter() #構(gòu)建字符到索引的映射forlabelindata['label']:char_counts.update(label)char_to_idx={'<PAD>':0,'<UNK>':1}forchar,countinchar_counts.items():ifcount>=2: #只保留出現(xiàn)次數(shù)大于等于2的字符

char_to_idx[char]=len(char_to_idx)#分割數(shù)據(jù)集為訓(xùn)練集和驗(yàn)證集train_paths,val_paths,train_labels,val_labels=train_test_split(data['audio_path'],data['label'],test_size=0.2,random_state=42)#將PandasSeries轉(zhuǎn)換為列表train_paths=train_paths.tolist()val_paths=val_paths.tolist()train_labels=train_labels.tolist()val_labels=val_labels.tolist()4、數(shù)據(jù)預(yù)處理#創(chuàng)建數(shù)據(jù)集和數(shù)據(jù)加載器sample_rate=16000max_audio_len=160000 #10秒音頻（假設(shè)采樣率為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)#超參數(shù)input_dim=40 #MFCC特征維度hidden_dim=256 #RNN隱藏層維度output_dim=len(char_to_idx)#輸出維度（字符數(shù)量）num_layers=2 #RNN層數(shù)num_epochs=10 #訓(xùn)練輪數(shù)learning_rate=0.001 #學(xué)習(xí)率4、數(shù)據(jù)預(yù)處理#添加自定義collate_fn處理變長(zhǎng)序列defcollate_fn(batch):#按MFCC序列長(zhǎng)度排序（降序）

batch.sort(key=lambdax:x[0].shape[0],reverse=True)#分離MFCC和標(biāo)簽

mfccs,labels=zip(*batch)#獲取各序列長(zhǎng)度

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#處理標(biāo)簽

label_lengths=torch.tensor([len(label)forlabelinlabels])max_label_length=label_lengths.max().item()#初始化填充后的標(biāo)簽張量(batch_size,max_label_length)padded_labels=torch.zeros(len(labels),max_label_length,dtype=torch.long)#填充標(biāo)簽

fori,labelinenumerate(labels):length=label.shape[0]padded_labels[i,:length]=labelreturnpadded_mfccs,mfcc_lengths,padded_labels,label_lengths#初始化模型、損失函數(shù)和優(yōu)化器model=SpeechToTextModel(input_dim,hidden_dim,output_dim,num_layers)criterion=nn.CTCLoss(blank=0)#blank對(duì)應(yīng)<PAD>標(biāo)記optimizer=optim.Adam(model.parameters(),lr=learning_rate)#訓(xùn)練循環(huán)forepochinrange(num_epochs):model.train()total_loss=0formfccs,mfcc_lengths,labels,label_lengthsintrain_loader:optimizer.zero_grad()logits=model(mfccs,mfcc_lengths)#準(zhǔn)備CTC輸入(time_steps,batch_size,num_classes)log_probs=logits.transpose(0,1).log_softmax(2)#計(jì)算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、模型訓(xùn)練#驗(yàn)證

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}')#保存模型

"""保存模型和所有必要參數(shù)"""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'):#加載保存的狀態(tài)

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

char_to_idx=state['char_to_idx']idx_to_char={v:kfork,vinchar_to_idx.items()}#初始化模型結(jié)構(gòu)

model=SpeechToTextModel(input_dim=input_dim,hidden_dim=hidden_dim,output_dim=len(char_to_idx),num_layers=2).to(device)#加載模型參數(shù)

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'):#預(yù)處理音頻

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貪婪解碼"""