<jupyter_start><jupyter_text>构建数据集<jupyter_code>def preprocess_sentence(line):    
    line = re.split('([: ,.(){}\[\]=])',line)        
    line = list(filter(lambda x: x!=' 'and x!='',line))
    
    new_line = '<start> ' + ' '.join(line) + ' <end>'
    return new_line
def create_dataset(path,num_examples,rand_max=15,duplicate=3):
    input_data = []
    output_data = []
    
    lines = io.open(path,encoding='utf-8').read().strip().split('\n')
    if num_examples == -1:
        num_examples = len(lines)
    
    for i in range(1,num_examples):
        
        rand_nums = set(random.randint(1,rand_max) for _ in range(duplicate))
        for rand_num in rand_nums:
            data = ''
            for j in range(i - rand_num,i):
                line = preprocess_sentence(lines[j].strip()) + ' '
                data += line
            input_data.append(data.strip())
            output_data.append(preprocess_sentence(lines[i].strip()))
        
    return input_data,output_data
def max_length(tensor):
    return max(len(t) for t in tensor)

def get_num_words(lang,scale):
    #获取词典大小
    lang_tokenizer = tf.keras.preprocessing.text.Tokenizer(filters='')
    lang_tokenizer.fit_on_texts(lang)
    num_words = int(len(lang_tokenizer.word_index) * scale)
    return num_words
    
def tokenize(lang,scale):
    num_words = get_num_words(lang,scale)
    lang_tokenizer = tf.keras.preprocessing.text.Tokenizer(num_words=num_words,
                                                           oov_token='<unk>',filters='')
    lang_tokenizer.fit_on_texts(lang)
    tensor = lang_tokenizer.texts_to_sequences(lang)
    tensor = tf.keras.preprocessing.sequence.pad_sequences(tensor,padding='post')
    
    return tensor, lang_tokenizer

def load_dataset(path,num_examples=None,scale=0.9):
    inp_lang,targ_lang = create_dataset(path, num_examples)

    input_tensor, inp_lang_tokenizer = tokenize(inp_lang,scale)
    target_tensor, targ_lang_tokenizer = tokenize(targ_lang,scale)

    return input_tensor, target_tensor, inp_lang_tokenizer, targ_lang_tokenizer<jupyter_output><empty_output><jupyter_text>限制数据集的大小以加快实验速度（可选）<jupyter_code># 尝试实验不同大小的数据集
num_examples = -1
scale = 0.95
path = "../00-data/tf_data.txt"
input_tensor, target_tensor, inp_lang, targ_lang = load_dataset(path, num_examples,scale)

# 计算目标张量的最大长度 （max_length）
max_length_targ, max_length_inp = max_length(target_tensor), max_length(input_tensor)
len(input_tensor)
len(target_tensor)
"""目前先不采用"""
# 采用 90 - 10 的比例切分训练集和验证集
input_tensor_train, input_tensor_val, target_tensor_train, target_tensor_val = train_test_split(input_tensor, target_tensor, test_size=0.1)

# 显示长度
print(len(input_tensor_train), len(target_tensor_train), len(input_tensor_val), len(target_tensor_val))<jupyter_output>2501 2501 278 278<jupyter_text>创建一个 tf.data 数据集<jupyter_code>BUFFER_SIZE = len(input_tensor)
BATCH_SIZE = 64
steps_per_epoch = len(input_tensor)//BATCH_SIZE
embedding_dim = 256
units = 256
vocab_inp_size = len(inp_lang.word_index)+1
vocab_tar_size = len(targ_lang.word_index)+1

dataset = tf.data.Dataset.from_tensor_slices((input_tensor, target_tensor)).shuffle(BUFFER_SIZE)
dataset = dataset.batch(BATCH_SIZE, drop_remainder=True)
example_input_batch, example_target_batch = next(iter(dataset))
example_input_batch.shape, example_target_batch.shape<jupyter_output><empty_output><jupyter_text>编写编码器encoder和解码器decoder模型<jupyter_code>class Encoder(tf.keras.Model):
    def __init__(self, vocab_size, embedding_dim, enc_units, batch_size):
        super(Encoder, self).__init__()
        self.batch_size = batch_size
        self.enc_units = enc_units
        self.embedding = tf.keras.layers.Embedding(vocab_size, embedding_dim,batch_input_shape=[batch_size,None])
        self.lstm = tf.keras.layers.LSTM(self.enc_units,
                                        return_sequences=True,
                                        return_state=True,
                                        dropout=0.1,
                                        recurrent_dropout=0.1)

    def call(self, x):
        x = self.embedding(x)
        output_l1,state_l1,_ = self.lstm(x)
        #output_l2,state_l2,_ = self.lstm(output_l1)
        return output_l1,state_l1

    def initialize_hidden_state(self):
        return tf.zeros((self.batch_size, self.enc_units))
encoder = Encoder(vocab_inp_size, embedding_dim, units, BATCH_SIZE)

# 样本输入
sample_hidden = encoder.initialize_hidden_state()
sample_output,sample_hidden = encoder(example_input_batch)
print ('Encoder output shape: (batch size, sequence length, units) {}'.format(sample_output.shape))
print ('Encoder Hidden state shape: (batch size, units) {}'.format(sample_hidden.shape))
class BahdanauAttention(tf.keras.layers.Layer):
    def __init__(self, units):
        super(BahdanauAttention, self).__init__()
        self.W1 = tf.keras.layers.Dense(units)
        self.W2 = tf.keras.layers.Dense(units)
        self.V = tf.keras.layers.Dense(1)

    def call(self, query, values):
        # 隐藏层的形状 == （批大小，隐藏层大小）
        # hidden_with_time_axis 的形状 == （批大小，1，隐藏层大小）
        # 这样做是为了执行加法以计算分数  
        hidden_with_time_axis = tf.expand_dims(query, 1)

        # 分数的形状 == （批大小，最大长度，1）
        # 我们在最后一个轴上得到 1， 因为我们把分数应用于 self.V
        # 在应用 self.V 之前，张量的形状是（批大小，最大长度，单位）
        score = self.V(tf.nn.tanh(
            self.W1(values) + self.W2(hidden_with_time_axis)))

        # 注意力权重 （attention_weights） 的形状 == （批大小，最大长度，1）
        attention_weights = tf.nn.softmax(score, axis=1)

        # 上下文向量 （context_vector） 求和之后的形状 == （批大小，隐藏层大小）
        context_vector = attention_weights * values
        context_vector = tf.reduce_sum(context_vector, axis=1)

        return context_vector, attention_weights
attention_layer = BahdanauAttention(10)
attention_result, attention_weights = attention_layer(sample_hidden, sample_output)

print("Attention result shape: (batch size, units) {}".format(attention_result.shape))
print("Attention weights shape: (batch_size, sequence_length, 1) {}".format(attention_weights.shape))
class Decoder(tf.keras.Model):
    def __init__(self, vocab_size, embedding_dim, dec_units, batch_sz):
        super(Decoder, self).__init__()
        self.batch_sz = batch_sz
        self.dec_units = dec_units
        self.embedding = tf.keras.layers.Embedding(vocab_size, embedding_dim)
        self.train_LSTM = tf.keras.layers.LSTM(self.dec_units,
                                       return_sequences=True,
                                       return_state=True,
                                       dropout=0.1,
                                       recurrent_dropout=0.1)
        self.infer_LSTM = tf.keras.layers.LSTM(self.dec_units,
                                       return_sequences=True,
                                       return_state=True)
        
        self.fc = tf.keras.layers.Dense(vocab_size)

        # 用于注意力
        self.attention = BahdanauAttention(self.dec_units)

    def call(self, x, hidden, enc_output,is_train=True):
        # 编码器输出 （enc_output） 的形状 == （批大小，最大长度，隐藏层大小）
        context_vector, attention_weights = self.attention(hidden, enc_output)

        # x 在通过嵌入层后的形状 == （批大小，1，嵌入维度）
        x = self.embedding(x)

        # x 在拼接 （concatenation） 后的形状 == （批大小，1，嵌入维度 + 隐藏层大小）
        x = tf.concat([tf.expand_dims(context_vector, 1), x], axis=-1)

        # 将合并后的向量传送到 LSTM
        if is_train:
            output, state,_ = self.train_LSTM(x)
        else:
            output, state,_ = self.infer_LSTM(x)

        # 输出的形状 == （批大小 * 1，隐藏层大小）
        output = tf.reshape(output, (-1, output.shape[2]))

        # 输出的形状 == （批大小，vocab）
        x = self.fc(output)

        return x, state, attention_weights
decoder = Decoder(vocab_tar_size, embedding_dim, units, BATCH_SIZE)

sample_decoder_output, _, _ = decoder(tf.random.uniform((BATCH_SIZE, 1)),
                                      sample_hidden, sample_output,is_train=True)

print ('Decoder output shape: (batch_size, vocab size) {}'.format(sample_decoder_output.shape))
optimizer = tf.keras.optimizers.Adam()
loss_object = tf.keras.losses.SparseCategoricalCrossentropy(
    from_logits=True, reduction='none')

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_)
checkpoint_dir = '../02-checkpoints/'
checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt")
checkpoint = tf.train.Checkpoint(optimizer=optimizer,
                                 encoder=encoder,
                                 decoder=decoder)
@tf.function
def train_step(inp,targ,enc_hidden):
    loss = 0
    
    with tf.GradientTape() as tape:
        enc_output,enc_hidden = encoder(inp)
        dec_hidden = enc_hidden
        
        dec_hidden = enc_hidden
        
        dec_input = tf.expand_dims([targ_lang.word_index['<start>']] * BATCH_SIZE, 1)
        
        for t in range(1,targ.shape[1]):
            predictions,dec_hidden,_ = decoder(dec_input,dec_hidden,enc_output,is_train=True)
            loss += loss_function(targ[:,t],predictions)
            
            dec_input = tf.expand_dims(targ[:,t],1)
            
        batch_loss = (loss / int(targ.shape[1]))
        variables = encoder.trainable_variables + decoder.trainable_variables
        gradients = tape.gradient(loss,variables)
        optimizer.apply_gradients(zip(gradients,variables))
        
        return batch_loss
EPOCHS = 30

for epoch in range(EPOCHS):
    start = time.time()

    enc_hidden = encoder.initialize_hidden_state()
    total_loss = 0

    for (batch, (inp, targ)) in enumerate(dataset.take(steps_per_epoch)):
        batch_loss = train_step(inp, targ, enc_hidden)
        total_loss += batch_loss

        #if batch % 5 == 0:
        print('Epoch {} Batch {} Loss {:.4f}'.format(epoch + 1,
                                                         batch,
                                                         batch_loss.numpy()))
    # 每 5 个周期（epoch），保存（检查点）一次模型
    if (epoch + 1) % 5 == 0:
        checkpoint.save(file_prefix=checkpoint_prefix)    

    print('Epoch {} Loss {:.4f}'.format(epoch + 1,
                                      total_loss / steps_per_epoch))
    print('Time taken for 1 epoch {} sec\n'.format(time.time() - start))<jupyter_output><empty_output><jupyter_text>预测<jupyter_code>def create_inference_data(lines):
    data = ""
    for line in lines:
        line = preprocess_sentence(line.strip()) + ' '
        data += line
    return data.strip()
def evaluate(sentence):
    attention_plot = np.zeros((max_length_targ, max_length_inp))

    sentence = create_inference_data(sentence)

    inputs = [inp_lang.word_index[i] for i in sentence.split(' ')]
    inputs = tf.keras.preprocessing.sequence.pad_sequences([inputs],
                                                           maxlen=max_length_inp,
                                                           padding='post')
    inputs = tf.convert_to_tensor(inputs)

    result = ''

    hidden = [tf.zeros((1, units))]
    enc_out, enc_hidden = encoder(inputs)

    dec_hidden = enc_hidden
    dec_input = tf.expand_dims([targ_lang.word_index['<start>']], 0)

    for t in range(max_length_targ):
        predictions, dec_hidden, attention_weights = decoder(dec_input,
                                                             dec_hidden,
                                                             enc_out,
                                                            is_train=False)

        # 存储注意力权重以便后面制图
        attention_weights = tf.reshape(attention_weights, (-1, ))
        attention_plot[t] = attention_weights.numpy()

        predicted_id = tf.argmax(predictions[0]).numpy()

        result += targ_lang.index_word[predicted_id] + ' '

        if targ_lang.index_word[predicted_id] == '<end>':
            return result, sentence, attention_plot

        # 预测的 ID 被输送回模型
        dec_input = tf.expand_dims([predicted_id], 0)

    return result, sentence, attention_plot
# 注意力权重制图函数
def plot_attention(attention, sentence, predicted_sentence):
    fig = plt.figure(figsize=(20,20))
    ax = fig.add_subplot(1, 1, 1)
    ax.matshow(attention, cmap='viridis')

    fontdict = {'fontsize': 10}

    ax.set_xticklabels([''] + sentence, fontdict=fontdict)
    ax.set_yticklabels([''] + predicted_sentence, fontdict=fontdict, rotation=90)

    ax.xaxis.set_major_locator(ticker.MultipleLocator(1))
    ax.yaxis.set_major_locator(ticker.MultipleLocator(1))

    plt.show()
def translate(sentence):
    result,sentence,attention_plot = evaluate(sentence)
    
    print('Input: %s' % (sentence))
    print('Predicted translation: {}'.format(result))
    
    attention_plot = attention_plot[:len(result.split(' ')),:len(sentence.split(' '))]
    plot_attention(attention_plot,sentence.split(' '),result.split(' '))    
    print(result.split(' '))
lines = io.open(path,encoding='utf-8').read().strip().split('\n')[12:13]
lines = create_inference_data(lines)
lines<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11604957160342717, 'fraction_numerical': 0.009868421052631578, 'mean_word_length': 2.7652649769585254, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '290089', 'license': 'Apache-2.0', 'n_tokens_mistral': 4605, 'n_tokens_neox': 4564, 'n_words': 679, 'path': 'model/02-Seq2Seq/.~Seq2Seq.ipynb', 'repo_name': 'KyelYang/AI-Coder-1', 'size': '13072'}