Python 人工智能实战：机器翻译

    conda create -n tf_env python=3.7 tensorflow numpy scipy scikit-learn matplotlib ipykernel notebook

    activate tf_env # activate conda environment
    
         
    代码解读

    import os

    
    
         
    代码解读

    ## 4.3 文件格式转换
    ```python
    from PIL import Image
    import numpy as np
    
    images = []
    for file in os.listdir("./opensubtitles"):
    with open('./opensubtitles/' + file, encoding='utf-8') as f:
        text = ''
        for line in f:
            text += line.strip('\n').lower().replace('-',' ')
    
        w, h = img.size
    
        pixels = list(img.getdata())
        pixel_matrix = [pixels[i*w:(i+1)*w] for i in range(h)]
        images.append((text,np.array(pixel_matrix)))
    
    with open('train.txt', 'w', encoding='utf-8') as fw:
    for image in images:
        fw.write(' '.join([str(x) for x in image[1].flatten()])+'\t'+image[0]+'\n')
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import string
    
    all_lines = []
    with open('train.txt', encoding='utf-8') as fr:
    lines = fr.readlines()[::3]
    for line in lines:
        all_lines.extend([line[:64], line[64:]])
    
    new_lines = []        
    for line in all_lines:
    new_line = ''.join([char for char in line if char not in string.punctuation and 
                        char!='']).encode('ascii', errors='ignore').decode('ascii').lower()
    
    if len(new_line)>0:
        new_lines.append(new_line)
    
    unique_words = set(['<unk>'] + sorted(list(set(' '.join(new_lines).split()))))    
    
    vocab_size = min(len(unique_words), 30000)
    
    word2idx = {'<pad>':0}
    idx2word = {0:'<pad>'}
    for word in unique_words:
    idx = len(idx2word)
    if vocab_size==None or idx < vocab_size:
        word2idx[word] = idx
        idx2word[idx] = word
    
    with open('vocab.txt', 'w', encoding='utf-8') as fw:
    fw.write('<pad>\t0\n<unk>\t1\n'+'\n'.join([str(k)+ '\t'+ str(v) for k, v in word2idx.items()]))
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import random
    
    random.seed(0)
    train_lines = random.sample(range(len(new_lines)), int(len(new_lines)*0.7))
    dev_lines   = random.sample(list(set(range(len(new_lines)))-set(train_lines)),int(len(new_lines)*0.1))
    test_lines  = list(set(range(len(new_lines)))-set(train_lines)-set(dev_lines))
    
    print(len(train_lines))
    print(len(dev_lines))
    print(len(test_lines))
    
    train_file = './train.txt'
    dev_file   = './dev.txt'
    test_file  = './test.txt'
    
    with open(train_file, 'w', encoding='utf-8') as fw:
    for idx in train_lines:
        fw.write(new_lines[idx]+'\t'+new_lines[idx+1][:-1]+'\n')
    
    with open(dev_file, 'w', encoding='utf-8') as fw:
    for idx in dev_lines:
        fw.write(new_lines[idx]+'\t'+new_lines[idx+1][:-1]+'\n')
    
    with open(test_file, 'w', encoding='utf-8') as fw:
    for idx in test_lines:
        fw.write(new_lines[idx]+'\t'+new_lines[idx+1][:-1]+'\n')
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import tensorflow as tf
    
    
    class TextDataGenerator(object):
    
    def __init__(self, data_path, batch_size):
    
        self._batch_size = batch_size
        input_files = ['./train.txt']
        target_files = ['./dev.txt', './test.txt']
    
        dataset = tf.data.Dataset.from_tensor_slices((input_files,target_files))
        dataset = dataset.repeat().shuffle(buffer_size=len(input_files)).interleave(lambda x: tf.data.TextLineDataset(x), cycle_length=len(input_files), block_length=1)
    
        # preprocess function
        def _preprocess(src, trg):
    
            src = tf.strings.split(src).to_tensor()
            src = tf.keras.preprocessing.sequence.pad_sequences(src, maxlen=64, padding='post')
            src = tf.one_hot(src, depth=len(word2idx), dtype=tf.float32)[..., :-1]
    
            trg = tf.strings.split(trg).to_tensor()
            trg = tf.keras.preprocessing.sequence.pad_sequences(trg, maxlen=64, padding='post')
            trg = tf.concat(([word2idx['<start>']] * (trg.shape[0]-1)), axis=-1)
            trg = tf.concat(([[word2idx.get(token.numpy(), word2idx['<unk>'])] for token in trg[:-1]], [[word2idx['<end>']]]), axis=-1)
    
            return {"inputs": src}, {"outputs": trg}
    
        # apply preprocessing to dataset
        dataset = dataset.map(_preprocess)
        dataset = dataset.batch(batch_size, drop_remainder=True)
    
        # prefetch data into buffer
        dataset = dataset.prefetch(buffer_size=tf.data.AUTOTUNE)
    
        self._dataset = dataset
    
    
    @property
    def steps_per_epoch(self):
        num_samples = sum([sum(1 for _ in open(f,'r'))//2 for f in ['./train.txt']])
        return num_samples // self._batch_size
    
    
    @property
    def total_steps(self):
        num_samples = sum([sum(1 for _ in open(f,'r'))//2 for f in ['./train.txt', './dev.txt', './test.txt']])
        return num_samples // self._batch_size
    
    
    def get_generator(self):
        return iter(self._dataset)
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import tensorflow as tf
    import numpy as np
    
    max_len = 64
    num_examples = None
    
    
    def preprocess_sentence(sent):
    """
    sent : sentence containing space separated words
    """
    sent = '<start>'+sent+'<end>'
    return sent
    
    
    def load_dataset(path, num_examples=None):
      """Load sentences and labels from path."""
    
      sentences = []
      labels = []
      lang_labels=[]
    
      with open(path, encoding="utf-8") as f:
      lines = f.read().strip().split("\n")
    
      count = 0
      for line in lines:
          source, target = line.split("\t")
    
          source = preprocess_sentence(source)
          target = preprocess_sentence(target)
    
          if len(source.split()) > max_len or len(target.split()) > max_len:
              continue
    
          sentences.append(source)
          labels.append(target)
          lang_labels.append(count % 4)
    
          if num_examples is not None:
              count += 1
              if count >= num_examples:
                  break
    
      print(sentences[0])
      print(labels[0])
      return sentences, labels,lang_labels
    
    sentences_train, labels_train,lang_labels_train=load_dataset('./train.txt', num_examples)
    sentences_val, labels_val,lang_labels_val=load_dataset('./dev.txt', num_examples)
    sentences_test, labels_test,lang_labels_test=load_dataset('./test.txt', num_examples)
    print(len(sentences_train))
    print(len(sentences_val))
    print(len(sentences_test))
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    from collections import Counter
    import re
    
    vocab_size = 30000
    tokenizer = Tokenizer(num_words=vocab_size, oov_token="<unk>", lower=False)
    tokenizer.fit_on_texts(sentences_train)
    
    train_seq = tokenizer.texts_to_sequences(sentences_train)
    train_seq = pad_sequences(train_seq, maxlen=max_len)
    train_labels = tokenizer.texts_to_sequences(labels_train)
    train_labels = pad_sequences(train_labels, maxlen=max_len)
    
    word_index = tokenizer.word_index
    
    embedding_dim = 256
    """Creating an embedding matrix"""
    num_tokens = len(word_index) + 1
    hits = 0
    misses = 0
    
    embedding_matrix = np.zeros((num_tokens, embedding_dim))
    
    for word, i in word_index.items():
    embedding_vector = embeddings_index.get(word)
    if embedding_vector is not None:
        # Words not found in embedding index will be all-zeros.
        # This includes the representation for "padding" and "OOV"
        embedding_matrix[i] = embedding_vector
        hits += 1
    else:
        misses += 1
    
    print("Converted {} words ({:.2f}% of tokens)".format(hits, 100.0 * hits / num_tokens))
    print("Missed {}".format(misses))
    
    del embeddings_index
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    from keras_preprocessing.text import Tokenizer
    from keras_preprocessing.sequence import pad_sequences
    
    tokenizer = Tokenizer(oov_token='<unk>', filters='')
    tokenizer.fit_on_texts(labels_train)
    label_word_index = tokenizer.word_index
    
    train_label_seq = tokenizer.texts_to_sequences(labels_train)
    train_label_seq = pad_sequences(train_label_seq, padding='post', maxlen=max_len)
    
    reverse_word_index = dict([(value, key) for (key, value) in label_word_index.items()]) 
    
    def decode_review(text):
    return''.join([reverse_word_index.get(i, '?') for i in text])
    
    print(decode_review(train_label_seq[0]))
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    BATCH_SIZE = 32
    BUFFER_SIZE = 10000
    
    train_dataset = tf.data.Dataset.from_tensor_slices((train_seq, train_label_seq))
    train_dataset = train_dataset.cache()
    train_dataset = train_dataset.shuffle(BUFFER_SIZE)
    train_dataset = train_dataset.padded_batch(BATCH_SIZE, padded_shapes=([-1],[max_len]))
    train_dataset = train_dataset.prefetch(tf.data.experimental.AUTOTUNE)
    
      
      
      
      
      
      
      
    
    代码解读

    from tensorflow.keras.layers import Input, Embedding, LSTM, Dense, Concatenate
    from tensorflow.keras.models import Model
    from tensorflow.keras.optimizers import Adam
    from sklearn.model_selection import KFold
    
    num_layers = 2
    units = 256
    
    def build_model():
    
    inputs = Input(shape=(None,))
    embedding_layer = Embedding(vocab_size, embedding_dim, mask_zero=True)(inputs)
    
    encoder = LSTM(units, return_state=True, name='encoder')
    enc_output, enc_hidden, c_state = encoder(embedding_layer)
    enc_states = [enc_hidden, c_state]
    
    decoder_inputs = Input(shape=(None,), name='decoder_inputs')
    dec_embedding = Embedding(vocab_size, embedding_dim, mask_zero=True)(decoder_inputs)
    
    decoder_lstm = LSTM(units, return_sequences=True, return_state=True, name='decoder_lstm')
    
    dec_outputs, _, _ = decoder_lstm(dec_embedding, initial_state=enc_states)
    
    outputs = Dense(vocab_size, activation='softmax')(dec_outputs)
    model = Model([inputs, decoder_inputs], outputs)
    
    optimizer = Adam(lr=0.001)
    loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
    
    model.compile(optimizer=optimizer, loss=[loss])
    
    return model
    
    model = build_model()
    print(model.summary())
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    loss_object = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
    
    def loss_function(real, pred):
    mask = tf.math.logical_not(tf.math.equal(real, 0))
    loss_ = loss_object(real, pred)
    
    mask = tf.cast(mask, dtype=loss_.dtype)
    loss_ *= mask
    
    return tf.reduce_mean(loss_)
    
      
      
      
      
      
      
      
      
      
    
    代码解读

    optimizer = tf.keras.optimizers.Adam()
    
    
    代码解读

    history = model.fit(train_dataset, epochs=NUM_EPOCHS, validation_data=val_dataset, verbose=1)
    
    
    代码解读

    loss = history.history['loss']
    val_loss = history.history['val_loss']
    
    plt.plot(loss, label='Training Loss')
    plt.plot(val_loss, label='Validation Loss')
    plt.legend(loc='upper right')
    plt.title('Training and Validation Loss')
    plt.show()
    
      
      
      
      
      
      
      
    
    代码解读

    def predict_sentence(sentence):
    sentence = preprocess_sentence(sentence)
    sequence = tokenizer.texts_to_sequences([sentence])[0][:max_len]
    sequence = pad_sequences([sequence], maxlen=max_len)
    
    translation = translate_model.predict([sequence])
    
    predicted_sentence = tokenizer.sequences_to_texts(translation)[0]
    final_sentence = re.sub("<.*?>","",predicted_sentence)
    return final_sentence
    
    translate_sentence = predict_sentence(sentences[0])
    print(translate_sentence)
    
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读