Pythonを使用して独自の音声認識システムをトレーニングします。この操作の波は安定しています

近年、音声認識技術は急速に発展しており、携帯電話のSiri音声インテリジェントアシスタント、MicrosoftのCortana、さまざまなプラットフォームのスマートスピーカーなど、さまざまな音声認識プロジェクトが広く使用されています。

音声認識は知覚知能に属しますが、機械が単純な音声認識から音声理解へと移行できるようにすると、認知知能のレベルにまで上がります。機械が自然言語を理解する能力は、その機械が知的であるかどうかの指標にもなり、自然言語理解はまさに現在の難題です。

同時に、現在の音声認識プラットフォームのほとんどがインテリジェントクラウドに依存していることを考えると、音声認識のトレーニングはほとんどの人にとってまだ比較的謎めいています。そこで、今日は Python を使用して独自の音声認識システムを構築します。

最終モデルの認識効果は次のとおりです。

[[396315]]

実験前の準備

まず、使用する Python のバージョンは 3.6.5 で、使用するライブラリは画像処理用の cv2 ライブラリです。

Numpy ライブラリは行列演算に使用され、Keras フレームワークはモデルのトレーニングと読み込みに使用されます。 Librosa および python_speech_features ライブラリは、オーディオ機能の抽出に使用されます。 glob および pickle ライブラリは、ローカルデータセットを読み取るために使用されます。

[[396316]]

データセットの準備まず、データセットでは清華大学の thchs30 中国データを使用します。

これらの録音は、テキストの内容に応じて、A（文IDは1〜250）、B（文IDは251〜500）、C（501〜750）、D（751〜1000）の4つの部分に分かれています。グループ A、B、C にはトレーニング用に 30 人の発音文 10,893 個が含まれ、グループ D にはテスト用に 10 人の発音文 2,496 個が含まれます。

データフォルダーには、.wav ファイルと .trn ファイルが含まれます。trn ファイルには、.wav ファイルのテキスト記述が含まれます。最初の行は単語、2 行目はピンイン、3 行目は音素です。

データセットは次のとおりです。

[[396317]]

モデルトレーニング

1. 音声データセットのMFCC特徴を抽出する:
まず、人間の声は声道を通じて発せられ、声道の形状によってどのような音が発せられるかが決まります。この形状を正確に知ることができれば、生成された音素を正確に記述することができます。声道の形状は、音声の短時間パワースペクトルのエンベロープで明らかにされます。 MFCC は、このエンベロープを正確に記述する機能です。

抽出された MFCC 特徴を下図に示します。

したがって、データセットの読み取りに基づいて、その音声特徴を抽出して保存し、トレーニングのためにニューラルネットワークにロードしやすくする必要があります。

対応するコードは次のとおりです。

 #データセットファイルを読み取る  
 
 text_paths = glob.glob( 'data/*.trn' ) 
 
合計 = len(テキストパス) 
 
印刷（合計） 
 
 open(text_paths[ 0 ], 'r' , encoding= 'utf8' ) を fr として実行します: 
 
行 = fr.readlines 
 
印刷（行） 
 
 #データセットファイルtrnの内容を読み取り、配列に保存します  
 
テキスト = 
 
パス = 
 
 text_paths内のパスの場合: 
 
 open(path, 'r' , encoding= 'utf8' ) を fr として実行します: 
 
行 = fr.readlines 
 
行 = 行[ 0 ].strip( '\n' ).replace( ' ' , '' ) 
 
テキスト.append(行) 
 
パスを追加します(path.rstrip( '.trn' )) 
 
印刷(パス[ 0 ]、テキスト[ 0 ]) 
 
 #mfccの数を定義する  
 
 mfcc_dim = 13   
 
 #データセットのキャリブレーションに基づいて音素を読み取る  
 
 def load_and_trim(パス): 
 
オーディオ、sr = librosa.load(パス) 
 
エネルギー = librosa.feature.rmse(オーディオ) 
 
フレーム = np.nonzero(エネルギー >= np.max(エネルギー) / 5 ) 
 
インデックス = librosa.core.frames_to_samples(フレーム)[ 1 ] 
 
 audio = audio[indices[ 0 ]:indices[- 1 ]] indices.sizeの場合はaudio[ 0 : 0 ] 
 
オーディオを返す、SR 
 
 #オーディオの特徴を抽出して保存する  
 
特徴 = 
 
 iがtqdm(範囲(合計))の場合: 
 
パス = パス[i] 
 
オーディオ、sr = load_and_trim(パス) 
 
機能を追加します(mfcc(オーディオ、sr、numcep=mfcc_dim、nfft= 551 )) 
 
印刷(len(特徴), 特徴[ 0 ].shape)

2. ニューラルネットワークの前処理:
ニューラルネットワークをロードしてトレーニングする前に、読み込んだ MFCC 機能を正規化する必要があります。主な目的は、収束を高速化し、効果を向上させ、干渉を減らすことです。次に、データセットとラベルを処理して、入力と出力を定義します。

対応するコードは次のとおりです。

 #ランダムに100個のデータセットを選択  
 
サンプル = ランダム.サンプル(特徴、 100 ) 
 
サンプル = np.vstack(サンプル) 
 
 #正規化のための平均 MFCC 値  
 
 mfcc_mean = np.mean(サンプル、軸= 0 ) 
 
 # 正規化のための標準偏差を計算する  
 
 mfcc_std = np.std(サンプル、軸= 0 ) 
 
印刷(mfcc_mean) 
 
印刷(mfcc_std) 
 
 #機能を正規化する  
 
特徴 = [(特徴 - mfcc_mean) / (mfcc_std + 1e - 14 )特徴内の特徴の場合] 
 
 #データセットに読み込まれたラベルと対応するIDを保存します  
 
文字 = {} 
 
テキスト内のテキストの場合: 
 
テキスト内のcの場合: 
 
文字[c] = 文字.get(c, 0 ) + 1   
 
 chars = sorted(chars.items, key= lambda x: x[ 1 ], reverse= True ) です。 
 
 chars = [ chars内のchar [ 0 ] ] 
 
 print (len(chars), chars[: 100 ]) 
 
 char2id = {c: i for i, c in enumerate(chars)} 
 
 id2char = {i: c for i, c in enumerate(chars)} 
 
 data_index = np.arange(合計) 
 
 np.random.shuffle(データインデックス) 
 
 train_size = int( 0.9 * 合計) 
 
 test_size = 合計 - train_size 
 
トレーニングインデックス = データインデックス[:トレーニングサイズ] 
 
 test_index = data_index[train_size:] 
 
 #ニューラルネットワークの入力と出力X、Y読み取りデータセット機能  
 
 X_train = [ train_index内のiのfeatures[i]] 
 
 Y_train = [texts[i] for i in train_index] 
 
 X_test = [features[i] for i in test_index] 
 
 Y_test = [texts[i] for i in test_index]

3. ニューラルネットワーク関数の定義:
これらには、トレーニングバッチ、畳み込み層関数、正規化関数、活性化層関数などが含まれます。

最初の次元はセグメントの数です。元の音声が長いほど、最初の次元は大きくなります。2 番目の次元は MFCC 機能の次元です。元の音声の数値表現を取得した後、WaveNet を使用して実装できます。 MFCC 特徴は 1 次元シーケンスであるため、畳み込みには Conv1D が使用されます。因果関係とは、畳み込みの出力が現在の位置より前の入力にのみ関連していること、つまり将来の特徴は使用されないことを意味します。畳み込みの位置を前方にシフトすると理解できます。 WaveNet モデルの構造は次のとおりです。

詳細は以下の通りです。

バッチサイズ = 16   
 
 #トレーニングバッチの生成を定義し、一度に16をトレーニングします  
 
バッチジェネレータを定義します(x, y, batch_size=batch_size): 
 
オフセット = 0   
 
その間 真実： 
 
オフセット += バッチサイズ
 
 offset == batch_sizeまたはoffset >= len(x)の場合: 
 
データインデックス = np.arange(len(x)) 
 
 np.random.shuffle(データインデックス) 
 
 x = [x[i] (データインデックス内のiの場合)] 
 
 y = [y[i] (データインデックス内のiの場合)] 
 
オフセット = バッチサイズ
 
 X_data = x[オフセット - バッチサイズ: オフセット] 
 
 Y_data = y[オフセット - バッチサイズ: オフセット] 
 
 X_maxlen = max([X_data[i].shape[ 0 ] iが範囲内(batch_size)]) 
 
 Y_maxlen = max([len(Y_data[i]) for i in range(batch_size)]) 
 
 X_batch = np.zeros([batch_size, X_maxlen, mfcc_dim]) 
 
 Y_batch = np.ones([batch_size, Y_maxlen]) * len(char2id) 
 
 X_length = np.zeros([batch_size, 1 ], dtype= 'int32' ) 
 
 Y_length = np.zeros([batch_size, 1 ], dtype= 'int32' ) 
 
 i が範囲内(batch_size)の場合: 
 
 X_length[i, 0 ] = X_data[i].shape[ 0 ] 
 
 X_バッチ[i, :X_長さ[i, 0 ], :] = X_データ[i] 
 
 Y_長さ[i, 0 ] = len(Y_データ[i]) 
 
 Y_batch[i, :Y_length[i, 0 ]] = [char2id[c] 、 cはY_data[i]内] 
 
入力 = { 'X' : X_バッチ、 'Y' : Y_バッチ、 'X_長さ' : X_長さ、 'Y_長さ' : Y_長さ} 
 
出力 = { 'ctc' : np.zeros([batch_size])} 
 
エポック = 50   
 
ブロック数 = 3   
 
フィルター = 128   
 
 X = 入力(形状=( None 、 mfcc_dim,)、 dtype= 'float32' 、 名前= 'X' ) 
 
 Y = 入力(形状=(なし,), dtype= 'float32' , 名前= 'Y' ) 
 
 X_length = 入力(形状=( 1 ,), dtype= 'int32' , 名前= 'X_length' ) 
 
 Y_length = 入力(shape=( 1 ,), dtype= 'int32' , name= 'Y_length' ) 
 
 #畳み込み1層  
 
 def conv1d(入力、フィルター、カーネルサイズ、膨張率): 
 
 return Conv1D(filters=filters, kernel_size=kernel_size, strides= 1 , padding= 'causal' , activation= None , 
 
 dilation_rate=dilation_rate)(入力) 
 
 #標準化機能  
 
 def batchnorm(入力): 
 
 BatchNormalization(入力)を返す
 
 #活性化層機能  
 
 def activation(入力, activation): 
 
アクティベーション(アクティベーション)(入力)を返す
 
 #全結合層関数  
 
 def res_block(入力、フィルター、カーネルサイズ、膨張率): 
 
 hf = activation(batchnorm(conv1d(入力、フィルター、カーネルサイズ、膨張率))、 'tanh' ) 
 
 hg = activation(batchnorm(conv1d(入力、フィルター、カーネルサイズ、膨張率))、 'シグモイド' ) 
 
 h0 = 乗算([hf, hg]) 
 
 ha = activation(batchnorm(conv1d(h0, フィルター, 1 , 1 )), 'tanh' ) 
 
 hs = activation(batchnorm(conv1d(h0, フィルター, 1 , 1 )), 'tanh' ) 
 
 Add([ha, inputs]), hsを返す
 
 h0 = activation(batchnorm(conv1d(X, フィルター, 1 , 1 )), 'tanh' ) 
 
ショートカット = 
 
 i が範囲内(num_blocks)の場合: 
 
 rが[ 1 , 2 , 4 , 8 , 16 ]の場合: 
 
 h0, s = res_block(h0, フィルター, 7 , r) 
 
ショートカット.append(s) 
 
 h1 = activation(Add(shortcut), 'relu' ) 
 
 h1 = activation(batchnorm(conv1d(h1, フィルター, 1 , 1 )), 'relu' ) 
 
 #softmax損失関数の出力結果  
 
 Y_pred = activation(batchnorm(conv1d(h1, len(char2id) + 1 , 1 , 1 )), 'softmax' ) 
 
 sub_model = モデル(入力=X、出力=Y_pred) 
 
 #損失関数を計算する  
 
 calc_ctc_loss (引数): 
 
 y、yp、ypl、yl = 引数
 
 K.ctc_batch_cost(y, yp, ypl, yl)を返す

4. モデルのトレーニング:
トレーニングのプロセスは次のようになります。

 ctc_loss = Lambda(calc_ctc_loss, output_shape=( 1 ,), name= 'ctc' )([Y, Y_pred, X_length, Y_length]) 
 
 # モデルのトレーニングをロードする  
 
モデル = モデル(入力=[X, Y, X_length, Y_length], 出力=ctc_loss) 
 
 #オプティマイザーを作成する  
 
オプティマイザー = SGD(lr= 0.02 、モメンタム= 0.9 、ネステロフ= True 、クリップノルム= 5 ) 
 
 #モデルをアクティブにして計算を開始する  
 
 model.compile(loss={ 'ctc' : lambda ctc_true, ctc_pred: ctc_pred}, optimizer=optimizer) 
 
チェックポインター = ModelCheckpoint(ファイルパス = 'asr.h5' 、詳細 = 0 ) 
 
 lr_decay = ReduceLROnPlateau(モニター= '損失' 、係数= 0.2 、忍耐= 1 、min_lr= 0.000 ) 
 
 #トレーニング開始  
 
履歴 = model.fit_generator( 
 
ジェネレータ = batch_generator(X_train, Y_train)、 
 
 steps_per_epoch=len(X_train) // バッチサイズ、 
 
エポック=時代、 
 
検証データ = バッチジェネレータ(X_test, Y_test)、 
 
 validation_steps=len(X_test) // バッチサイズ、 
 
コールバック=[チェックポインタ、lr_decay]) 
 
 #モデルを保存  
 
サブモデルを保存します( 'asr.h5' ) 
 
 #pl=pkl に単語を保存します  
 
 open( 'dictionary.pkl' , 'wb' ) を fw として実行します: 
 
 pickle.dump([char2id, id2char, mfcc_mean, mfcc_std], fw) 
 
 train_loss = history.history[ '損失' ] 
 
有効な損失 = history.history[ 'val_loss' ] 
 
 plt.plot(np.linspace( 1 , エポック, エポック), train_loss, ラベル= 'train' ) 
 
 plt.plot(np.linspace( 1 , エポック, エポック), valid_loss, ラベル= 'valid' ) 
 
 plt.legend(loc= '右上' ) 
 
 plt.xlabel( 'エポック' ) 
 
 plt.ylabel( '損失' ) 
 
 plt.show

[[396319]]

モデルのテスト
音声データセットによって生成された辞書を読み取り、モデルを呼び出して音声の特徴を認識します。

コードは次のとおりです。

 wavs = glob.glob( 'A2_103.wav' ) 
 
印刷（wavs） 
 
 open( 'dictionary.pkl' , 'rb' ) を fr として実行します: 
 
 [char2id、id2char、mfcc_mean、mfcc_std] = pickle.load(fr) 
 
 mfcc_dim = 13   
 
モデル = load_model( 'asr.h5' ) 
 
インデックス = np.random.randint(len(wavs)) 
 
印刷(wavs[インデックス]) 
 
オーディオ、sr = librosa.load(wavs[index]) 
 
エネルギー = librosa.feature.rmse(オーディオ) 
 
フレーム = np.nonzero(エネルギー >= np.max(エネルギー) / 5 ) 
 
インデックス = librosa.core.frames_to_samples(フレーム)[ 1 ] 
 
 audio = audio[indices[ 0 ]:indices[- 1 ]] indices.sizeの場合はaudio[ 0 : 0 ] 
 
 X_data = mfcc(オーディオ、sr、numcep=mfcc_dim、nfft= 551 ) 
 
 X_データ = (X_データ - mfcc_mean) / (mfcc_std + 1e - 14 ) 
 
印刷(X_data.shape) 
 
 pred = model.predict(np.expand_dims(X_data, axis= 0 )) 
 
 pred_ids = K.eval(K.ctc_decode(pred, [X_data.shape[ 0 ]], greedy= False , beam_width= 10 , top_paths= 1 )[ 0 ][ 0 ]) 
 
 pred_ids = pred_ids.flatten.tolist 
 
 print ( '' .join([id2char[i] for i in pred_ids])) 
 
収量（入力、出力）