前言：理解了很久的CTC，每次都是點到即止，所以一直沒有很明確，現在重新整理。

定義

CTC (Connectionist Temporal Classification)是一種loss function

對比

傳統方法

 在傳統的語音識別的模型中，我們對語音模型進行訓練之前，往往都要將文字與語音進行嚴格的對齊操作。這樣就有兩點不太好：
 1. 嚴格對齊要花費人力、時間。
 2. 嚴格對齊之後，模型預測出的label只是區域性分類的結果，而無法給出整個序列的輸出結果，往往要對預測出的label做一些後處理才可以得到我們最終想要的結果。
  雖然現在已經有了一些比較成熟的開源對齊工具供大家使用，但是隨著deep learning越來越火，有人就會想，能不能讓我們的網路自己去學習對齊方式呢？因此CTC（Connectionist temporal classification）就應運而生啦。
  想一想，為什麼CTC就不需要去對齊語音和文字呢？因為CTC它允許我們的神經網路在任意一個時間段預測label，只有一個要求：就是輸出的序列順序只要是正確的就ok啦~這樣我們就不在需要讓文字和語音嚴格對齊了，而且CTC輸出的是整個序列標籤，因此也不需要我們再去做一些後處理操作。
  對一段音訊使用CTC和使用文字對齊的例子如下圖所示：
  

主要區別

訓練流程和傳統的神經網路類似，構建loss function，然後根據BP演算法進行訓練，不同之處在於傳統的神經網路的訓練準則是針對每幀資料，即每幀資料的訓練誤差最小，而CTC的訓練準則是基於序列（比如語音識別的一整句話）的，比如最大化p(z|x)$p(z|x)$ ，序列化的概率求解比較複雜，因為一個輸出序列可以對應很多的路徑，所有引入前後向演算法來簡化計算。

演算法細節

符號定義

概率計算

誤差反傳

參考文獻

實現程式碼

#coding=utf-8
import time

import
 
 tensorflow as tf
import scipy.io.wavfile as wav
import numpy as np

from six.moves import xrange as range

try:
    from python_speech_features import mfcc
except ImportError:
    print("Failed to import python_speech_features.\n Try pip install python_speech_features.")
    raise ImportError



# 常量
 

SPACE_TOKEN = '<space>'
SPACE_INDEX = 0
FIRST_INDEX = ord('a') - 1  # 0 is reserved to space

# mfcc預設提取出來的一幀13個特徵
num_features = 13
# 26個英文字母 + 1個空白 + 1個no label = 28 label個數
num_classes = ord('z') - ord('a') + 1 + 1 + 1

# 迭代次數
num_epochs = 200
# lstm隱藏單元數
num_hidden = 40
# 2層lstm網路
num_layers = 1
# batch_size設定為1
batch_size = 1
# 初始學習率
initial_learning_rate = 0.01

# 樣本個數
num_examples = 1
# 一個epoch有多少個batch
num_batches_per_epoch = int(num_examples/batch_size)


def sparse_tuple_from(sequences, dtype=np.int32):
    """得到一個list的稀疏表示，為了直接將資料賦值給tensorflow的tf.sparse_placeholder稀疏矩陣
    Args:
        sequences: 序列的列表
    Returns:
        一個三元組，和tensorflow的tf.sparse_placeholder同結構
    """
    indices = []
    values = []

    for n, seq in enumerate(sequences):
        indices.extend(zip([n]*len(seq), range(len(seq))))
        values.extend(seq)

    indices = np.asarray(indices, dtype=np.int64)
    values = np.asarray(values, dtype=dtype)
    shape = np.asarray([len(sequences), np.asarray(indices).max(0)[1]+1], dtype=np.int64)

    return indices, values, shape


def get_audio_feature():
  '''
  獲取wav檔案提取mfcc特徵之後的資料
  '''

  audio_filename = "audio.wav"

  #讀取wav檔案內容，fs為取樣率， audio為資料
  fs, audio = wav.read(audio_filename)

  #提取mfcc特徵
  inputs = mfcc(audio, samplerate=fs)
  # 對特徵資料進行歸一化，減去均值除以方差
  feature_inputs = np.asarray(inputs[np.newaxis, :])
  feature_inputs = (feature_inputs - np.mean(feature_inputs))/np.std(feature_inputs)

  #特徵資料的序列長度
  feature_seq_len = [feature_inputs.shape[1]]

  return feature_inputs, feature_seq_len

def get_audio_label():
  '''
  將label文字轉換成整數序列，然後再換成稀疏三元組
  '''
  target_filename = 'label.txt'

  with open(target_filename, 'r') as f:
    #原始文字為“she had your dark suit in greasy wash water all year”
    line = f.readlines()[0].strip()
    targets = line.replace(' ', '  ')
    # 放入list中，空格用''代替
    #['she', '', 'had', '', 'your', '', 'dark', '', 'suit', '', 'in', '', 'greasy', '', 'wash', '', 'water', '', 'all', '', 'year']
    targets = targets.split(' ')

    # 每個字母作為一個label,轉換成如下：
    #['s' 'h' 'e' '<space>' 'h' 'a' 'd' '<space>' 'y' 'o' 'u' 'r' '<space>' 'd'
    # 'a' 'r' 'k' '<space>' 's' 'u' 'i' 't' '<space>' 'i' 'n' '<space>' 'g' 'r'
    # 'e' 'a' 's' 'y' '<space>' 'w' 'a' 's' 'h' '<space>' 'w' 'a' 't' 'e' 'r'
    #'<space>' 'a' 'l' 'l' '<space>' 'y' 'e' 'a' 'r']
    targets = np.hstack([SPACE_TOKEN if x == '' else list(x) for x in targets])

    # 將label轉換成整數序列表示:
    # [19  8  5  0  8  1  4  0 25 15 21 18  0  4  1 18 11  0 19 21  9 20  0  9 14
    # 0  7 18  5  1 19 25  0 23  1 19  8  0 23  1 20  5 18  0  1 12 12  0 25  5
    # 1 18]
    targets = np.asarray([SPACE_INDEX if x == SPACE_TOKEN else ord(x) - FIRST_INDEX
                      for x in targets])

    # 將列表轉換成稀疏三元組
    train_targets = sparse_tuple_from([targets])
  return train_targets



def inference(inputs, seq_len):
  '''
  2層雙向LSTM的網路結構定義

  Args：
  inputs： 輸入資料，形狀是[batch_size, 序列最大長度，一幀特徵的個數13]
        序列最大長度是指，一個樣本在轉成特徵矩陣之後儲存在一個矩陣中，
        在n個樣本組成的batch中，因為不同的樣本的序列長度不一樣，在組成的3維資料中，
        第2維的長度要足夠容納下所有的樣本的特徵序列長度。
  seq_len: batch裡每個樣本的有效的序列長度
  '''

  #定義一個向前計算的LSTM單元，40個隱藏單元
  cell_fw = tf.contrib.rnn.LSTMCell(num_hidden, 
                        initializer=tf.random_normal_initializer(
                                        mean=0.0, stddev=0.1),
                        state_is_tuple=True)

  # 組成一個有2個cell的list
  cells_fw = [cell_fw] * num_layers
  # 定義一個向後計算的LSTM單元，40個隱藏單元
  cell_bw = tf.contrib.rnn.LSTMCell(num_hidden, 
                        initializer=tf.random_normal_initializer(
                                        mean=0.0, stddev=0.1),
                        state_is_tuple=True)
  # 組成一個有2個cell的list
  cells_bw = [cell_bw] * num_layers

  # 將前面定義向前計算和向後計算的2個cell的list組成雙向lstm網路
  # sequence_length為實際有效的長度，大小為batch_size，
  # 相當於表示batch中每個樣本的實際有用的序列長度有多長。
  # 輸出的outputs寬度是隱藏單元的個數，即num_hidden的大小
  outputs, _, _ = tf.contrib.rnn.stack_bidirectional_dynamic_rnn(cells_fw,
                                                                 cells_bw,
                                                                 inputs,
                                                               dtype=tf.float32,
                                                        sequence_length=seq_len)

  #獲得輸入資料的形狀
  shape = tf.shape(inputs)
  batch_s, max_timesteps = shape[0], shape[1]

  # 將2層LSTM的輸出轉換成寬度為40的矩陣
  # 後面進行全連線計算
  outputs = tf.reshape(outputs, [-1, num_hidden])

  W = tf.Variable(tf.truncated_normal([num_hidden,
                                         num_classes],
                                        stddev=0.1))

  b = tf.Variable(tf.constant(0., shape=[num_classes]))

  # 進行全連線線性計算
  logits = tf.matmul(outputs, W) + b

  # 將全連線計算的結果，由寬度40變成寬度80，
  # 即最後的輸入給CTC的資料寬度必須是26+2的寬度
  logits = tf.reshape(logits, [batch_s, -1, num_classes])

  # 轉置，將第一維和第二維交換。
  # 變成序列的長度放第一維，batch_size放第二維。
  # 也是為了適應Tensorflow的CTC的輸入格式
  logits = tf.transpose(logits, (1, 0, 2))

  return logits

def main():
  # 輸入特徵資料，形狀為：[batch_size, 序列長度，一幀特徵數]
  inputs = tf.placeholder(tf.float32, [None, None, num_features])

  # 輸入資料的label，定義成稀疏sparse_placeholder會生成稀疏的tensor：SparseTensor
  # 這個結構可以直接輸入給ctc求loss
  targets = tf.sparse_placeholder(tf.int32)

  # 序列的長度，大小是[batch_size]大小
  # 表示的是batch中每個樣本的有效序列長度是多少
  seq_len = tf.placeholder(tf.int32, [None])

  # 向前計算網路，定義網路結構，輸入是特徵資料，輸出提供給ctc計算損失值。
  logits = inference(inputs, seq_len)

  # ctc計算損失
  # 引數targets必須是一個值為int32的稀疏tensor的結構：tf.SparseTensor
  # 引數logits是前面lstm網路的輸出
  # 引數seq_len是這個batch的樣本中，每個樣本的序列長度。
  loss = tf.nn.ctc_loss(targets, logits, seq_len)

  # 計算損失的平均值
  cost = tf.reduce_mean(loss)

  # 採用衝量優化方法
  optimizer = tf.train.MomentumOptimizer(initial_learning_rate, 0.9).minimize(cost)

  # 還有另外一個ctc的函式：tf.contrib.ctc.ctc_beam_search_decoder
  # 本函式會得到更好的結果，但是效果比ctc_beam_search_decoder低
  # 返回的結果中，decode是ctc解碼的結果，即輸入的資料解碼出結果序列是什麼
  decoded, _ = tf.nn.ctc_greedy_decoder(logits, seq_len)

  # 採用計算編輯距離的方式計算，計算decode後結果的錯誤率。
  ler = tf.reduce_mean(tf.edit_distance(tf.cast(decoded[0], tf.int32),
                                          targets))
  config = tf.ConfigProto()
  config.gpu_options.allow_growth = True

  with tf.Session(config=config) as session:
    # 初始化變數
    tf.global_variables_initializer().run()

    for curr_epoch in range(num_epochs):
      train_cost = train_ler = 0
      start = time.time()

      for batch in range(num_batches_per_epoch):
        #獲取訓練資料，本例中只去一個樣本的訓練資料
        train_inputs, train_seq_len = get_audio_feature()
        # 獲取這個樣本的label
        train_targets = get_audio_label()
        feed = {inputs: train_inputs,
                  targets: train_targets,
                  seq_len: train_seq_len}

        # 一次訓練，更新引數
        batch_cost, _ = session.run([cost, optimizer], feed)
        # 計算累加的訓練的損失值
        train_cost += batch_cost * batch_size
        # 計算訓練集的錯誤率
        train_ler += session.run(ler, feed_dict=feed)*batch_size

      train_cost /= num_examples
      train_ler /= num_examples

      # 列印每一輪迭代的損失值，錯誤率
      log = "Epoch {}/{}, train_cost = {:.3f}, train_ler = {:.3f}, time = {:.3f}"
      print(log.format(curr_epoch+1, num_epochs, train_cost, train_ler,
                          time.time() - start))
    # 在進行了1200次訓練之後，計算一次實際的測試，並且輸出
    # 讀取測試資料，這裡讀取的和訓練資料的同一個樣本
    test_inputs, test_seq_len = get_audio_feature()
    test_targets = get_audio_label()
    test_feed = {inputs: test_inputs,
                  targets: test_targets,
                  seq_len: test_seq_len}
    d = session.run(decoded[0], feed_dict=test_feed)
    # 將得到的測試語音經過ctc解碼後的整數序列轉換成字母
    str_decoded = ''.join([chr(x) for x in np.asarray(d[1]) + FIRST_INDEX])
    # 將no label轉換成空
    str_decoded = str_decoded.replace(chr(ord('z') + 1), '')
    # 將空白轉換成空格
    str_decoded = str_decoded.replace(chr(ord('a') - 1), ' ')
    # 列印最後的結果
    print('Decoded:\n%s' % str_decoded)

if __name__ == "__main__":
  main()