循环神经网络

import numpy as np
import csv
import tensorflow as tf

from tensorflow.keras.layers import SimpleRNN, GRU, LSTM, Dense, Embedding
from tensorflow.keras.models import Sequential
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.preprocessing.text import Tokenizer
from sklearn.model_selection import train_test_split
from tensorflow.keras.models import load_model

print(tf.__version__)


def read_data():
    FILE_NAME = 'sentiment_analysis.csv'
    sentence = []
    sentiment = []
    with open(FILE_NAME, 'r', encoding='UTF-8') as csvfile:
        csv_reader = csv.reader(csvfile)  # 使用csv.reader来读取csv文件
        for row in csv_reader:
            sentence.append(row)
    print('the first sentence is : ', sentence[0])
    for i in range(len(sentence)):
        s = ''
        for item in sentence[i]:
            s += item
        sentence[i] = s.split('\t')[1]
        sentiment.append(s.split('\t')[0])
    print('there are ', len(sentence), ' sentences in total')  # 字符串列表
    print('there are ', len(sentiment), 'sentimental labels in total (0/1)')  # 0/1的列表

    return sentence, sentiment


# 模型配置
VOCAB_SIZE = 5000  # 选取前5000个最频繁出现的单词
EMBED_SIZE = 200  # 词嵌入大小
NUM_HIDDEN = 128  # RNN中隐含层神经元个数
BATCH_SIZE = 64
NUM_EPOCH = 3


# 需要将数据导入当前目录
def process_data(sentence, sentiment):
    # word segmentation, padding and obtaining vocabulary
    tokenizer = Tokenizer(VOCAB_SIZE, oov_token='<UNK>')  # 1. define tokenizer
    tokenizer.fit_on_texts(sentence)  # 2. fit on texts
    sequence = tokenizer.texts_to_sequences(sentence)  # 3. texts to sequences
    X = pad_sequences(sequence, padding='post')  # 4. pad sequences
    X = X[:, :120]  # length cut
    Y = np.array(sentiment)

    # for Tokenizer, although when the vocabulary size has been set, for example, 5000, 
    # the vocabulary will contain all words in texts. However, when transfering texts to 
    # sequences, only words with top 5000 frequency will be converted to corresponding 
    # index, other words will be converted to <UNK>
    vocab = tokenizer.word_index
    print(vocab)
    print(len(vocab))

    vocab_size = VOCAB_SIZE + 1

    print('X: ', X)
    print("Y: ", Y)
    print('X size: ', X.shape)
    print('Y size: ', Y.shape)

    Xtrain, Xtest, Ytrain, Ytest = train_test_split(X, Y)
    print('vocabulary size: ', vocab_size)  # 5001，'0'用于填充

    return Xtrain, Xtest, Ytrain, Ytest, vocab_size


def run(MODE='TRAIN'):
    sentence, sentiment = read_data()  # 加载数据
    Xtrain, Xtest, Ytrain, Ytest, vocab_size = process_data(sentence, sentiment)

    if MODE == 'TRAIN':
        model_rnn = Sequential()
        model_rnn.add(Embedding(input_dim=vocab_size, output_dim=EMBED_SIZE, input_length=120, mask_zero=True))
        model_rnn.add(SimpleRNN(NUM_HIDDEN, activation='relu'))  # LSTM/GRU
        model_rnn.add(Dense(NUM_HIDDEN, activation='relu'))
        model_rnn.add(Dense(2, activation='softmax'))
        model_rnn.summary()

        model_rnn.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])
        history = model_rnn.fit(Xtrain, Ytrain, batch_size=BATCH_SIZE, epochs=NUM_EPOCH, validation_data=(Xtest, Ytest))
        model_rnn.save('rnn_classifier.h5')  # 保存模型为h5格式

    elif MODE == 'TEST':
        model_rnn = load_model('rnn_classifier.h5')
        result = model_rnn.evaluate(Xtest, Ytest)
        print('RNN sentiment analysis accuracy: ', result[1])


run('TRAIN')
run('TEST')

import tensorflow as tf

import numpy as np
import os
import time

# 读取数据，需要将数据导入当前目录
text = open('shakespeare.txt', 'rb').read().decode(encoding='utf-8')
print('Length of text: {} characters'.format(len(text)))

# 查看文章中的前250个字符
print(text[:250])

# 查看文件中不同字符的种类
vocab = sorted(set(text))
print('{} unique characters'.format(len(vocab)))

# 将文本向量化
char2idx = {u: i for i, u in enumerate(vocab)}  # 词典
idx2char = np.array(vocab)  # 反向词典

text_as_int = np.array([char2idx[c] for c in text])
print(text_as_int.shape)  # (1115394,)

# 查看文章中前13个字符是如何被映射到13个整数的
print('{} ---- characters mapped to int ---- > {}'.format(repr(text[:13]), text_as_int[:13]))

# 一条字符序列的最大长度
seq_length = 100
examples_per_epoch = len(text) // (seq_length + 1)

# 将数据转为tf.data.Dataset格式
char_dataset = tf.data.Dataset.from_tensor_slices(text_as_int)

sequences = char_dataset.batch(seq_length + 1, drop_remainder=True)

for item in sequences.take(5):
    print(repr(''.join(idx2char[item.numpy()])))


# 利用右移的方式构造target
def split_input_target(chunk):
    input_text = chunk[:-1]
    target_text = chunk[1:]
    return input_text, target_text


dataset = sequences.map(split_input_target)

print(dataset)
for input_example, target_example in dataset.take(1):
    print('Input data: ', repr(''.join(idx2char[input_example.numpy()])))
    print('Target data:', repr(''.join(idx2char[target_example.numpy()])))

# Batch size
BATCH_SIZE = 64

# Buffer size to shuffle the dataset
# (TF data is designed to work with possibly infinite sequences,
# so it doesn't attempt to shuffle the entire sequence in memory. Instead,
# it maintains a buffer in which it shuffles elements).
BUFFER_SIZE = 10000
dataset = dataset.shuffle(BUFFER_SIZE).batch(BATCH_SIZE, drop_remainder=True)
print(dataset)

# 词典大小
vocab_size = len(vocab)
# 词嵌入大小
embedding_dim = 300
# RNN隐含层单元数
rnn_units = 512


# 构造模型
def build_model(vocab_size, embedding_dim, rnn_units, batch_size):
    model = tf.keras.Sequential([
        tf.keras.layers.Embedding(vocab_size, embedding_dim, batch_input_shape=[batch_size, None]),
        tf.keras.layers.GRU(rnn_units, return_sequences=True, stateful=True, recurrent_initializer='glorot_uniform'),
        tf.keras.layers.Dense(vocab_size)
    ])
    return model


model = build_model(vocab_size=len(vocab), embedding_dim=embedding_dim, rnn_units=rnn_units, batch_size=BATCH_SIZE)

# 尝试模型的输出
for input_example_batch, target_example_batch in dataset.take(1):
    example_batch_predictions = model(input_example_batch)
    print(example_batch_predictions.shape, "# (batch_size, sequence_length, vocab_size)")

model.summary()


def loss(labels, logits):
    return tf.keras.losses.sparse_categorical_crossentropy(labels, logits, from_logits=True)


example_batch_loss = loss(target_example_batch, example_batch_predictions)
print("Prediction shape: ", example_batch_predictions.shape, " # (batch_size, sequence_length, vocab_size)")
print("scalar_loss:      ", example_batch_loss.numpy().mean())

# 配置检查点
checkpoint_dir = './training_checkpoints'
checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt_{epoch}")

checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(filepath=checkpoint_prefix, save_weights_only=True)

model.compile(optimizer='adam', loss=loss)

EPOCHS = 10

model.fit(dataset, epochs=EPOCHS, callbacks=[checkpoint_callback])

# 从检查点中恢复模型，并设置batch_size为1
model = build_model(vocab_size, embedding_dim, rnn_units, batch_size=1)
model.load_weights(tf.train.latest_checkpoint(checkpoint_dir))
model.build(tf.TensorShape([1, None]))


def generate_text(model, start_string):
    # Evaluation step (generating text using the learned model)

    # Number of characters to generate
    num_generate = 1000

    # Converting our start string to numbers (vectorizing)
    input_eval = [char2idx[s] for s in start_string]
    input_eval = tf.expand_dims(input_eval, 0)

    # Empty string to store our results
    text_generated = []

    # Low temperatures results in more predictable text.
    # Higher temperatures results in more surprising text.
    # Experiment to find the best setting.
    temperature = 1.0

    # Here batch size == 1
    model.reset_states()
    for i in range(num_generate):
        predictions = model(input_eval)
        # remove the batch dimension
        predictions = tf.squeeze(predictions, 0)

        # using a categorical distribution to predict the character returned by the model
        predictions = predictions / temperature
        predicted_id = tf.random.categorical(predictions, num_samples=1)[-1, 0].numpy()

        # We pass the predicted character as the next input to the model
        # along with the previous hidden state
        input_eval = tf.expand_dims([predicted_id], 0)

        text_generated.append(idx2char[predicted_id])

    return start_string + ''.join(text_generated)


print(generate_text(model, start_string=u"ROMEO: "))