def __init__(self, training_file='../res/trump_tweets.txt', model_file='../res/model.pt', n_epochs=1000000,
hidden_size=256, n_layers=2, learning_rate=0.001, chunk_len=140):
self.training_file = training_file
self.model_file = model_file
self.n_epochs = n_epochs
self.hidden_size = hidden_size
self.n_layers = n_layers
self.learning_rate = learning_rate
self.chunk_len = chunk_len
self.file, self.file_len = read_file(training_file)
if os.path.isfile(model_file):
self.decoder = torch.load(model_file)
print('Loaded old model!')
else:
self.decoder = RNN(n_characters, hidden_size, n_characters, n_layers)
print('Constructed new model!')
self.decoder_optimizer = torch.optim.Adam(self.decoder.parameters(), learning_rate)
self.criterion = nn.CrossEntropyLoss()
self.generator = Generator(self.decoder)
python类RNN的实例源码
def __init__(self, input_size, hidden_size, output_size, n_layers=1, gpu=-1):
self.decoder = RNN(input_size, hidden_size, output_size, n_layers, gpu)
if gpu >= 0:
print("Use GPU %d" % torch.cuda.current_device())
self.decoder.cuda()
self.optimizer = torch.optim.Adam(self.decoder.parameters(), lr=0.01)
self.criterion = nn.CrossEntropyLoss()
def main(_):
# Save default params and set scope
saved_params = FLAGS.__flags
if saved_params['ensemble']:
model_name = 'ensemble'
elif saved_params['ngram'] == 1:
model_name = 'unigram'
elif saved_params['ngram'] == 2:
model_name = 'bigram'
elif saved_params['ngram'] == 3:
model_name = 'trigram'
else:
assert True, 'Not supported ngram %d'% saved_params['ngram']
model_name += '_embedding' if saved_params['embed'] else '_no_embedding'
saved_params['model_name'] = '%s' % model_name
saved_params['checkpoint_dir'] += model_name
pprint.PrettyPrinter().pprint(saved_params)
saved_dataset = get_data(saved_params)
validation_writer = open(saved_params['valid_result_path'], 'a')
validation_writer.write(model_name + "\n")
validation_writer.write("[dim_hidden, dim_rnn_cell, learning_rate, lstm_dropout, lstm_layer, hidden_dropout, dim_embed]\n")
validation_writer.write("combination\ttop1\ttop5\tepoch\n")
# Run the model
for _ in range(saved_params['valid_iteration']):
# Sample parameter sets
params, combination = sample_parameters(saved_params.copy())
dataset = saved_dataset[:]
# Initialize embeddings
uni_init = get_char2vec(dataset[0][0][:], params['dim_embed_unigram'], dataset[3][0])
bi_init = get_char2vec(dataset[0][1][:], params['dim_embed_bigram'], dataset[3][4])
tri_init = get_char2vec(dataset[0][2][:], params['dim_embed_trigram'], dataset[3][5])
print(model_name, 'Parameter sets: ', end='')
pprint.PrettyPrinter().pprint(combination)
rnn_model = RNN(params, [uni_init, bi_init, tri_init])
top1, top5, ep = experiment(rnn_model, dataset, params)
validation_writer.write(str(combination) + '\t')
validation_writer.write(str(top1) + '\t' + str(top5) + '\tEp:' + str(ep) + '\n')
validation_writer.close()
def train_and_test(challenge, rnn_cell):
'''
????
:return:
'''
train, test = helper.extract_file(challenge)
vocab, word_idx, story_maxlen, query_maxlen = helper.get_vocab(train, test)
vocab_size = len(vocab) + 1 # Reserve 0 for masking via pad_sequences
x, xq, y = helper.vectorize_stories(train, word_idx, story_maxlen, query_maxlen)
tx, txq, ty = helper.vectorize_stories(test, word_idx, story_maxlen, query_maxlen)
with tf.Graph().as_default() as graph:
story_pl, question_pl, answer_pl, dropout_pl = get_placeholder(vocab_size, story_maxlen, query_maxlen)
rnn = model.RNN(rnn_cell, FLAGS.embed_dim, FLAGS.rnn_size, vocab_size)
logits = rnn.inference(story_pl, question_pl, dropout_pl)
loss = rnn.loss(logits, answer_pl)
train_op = rnn.train(loss, FLAGS.init_learning_rate)
correct = rnn.eval(logits, answer_pl)
init = tf.global_variables_initializer()
gpu_options = tf.GPUOptions(per_process_gpu_memory_fraction=FLAGS.gpu_fraction)
with tf.Session(config=tf.ConfigProto(gpu_options=gpu_options), graph=graph) as sess:
# ???????
sess.run(init)
max_test_acc = 0
for i in range(FLAGS.num_epochs):
batch_id = 1
train_gen = helper.generate_data(FLAGS.batch_size, x, xq, y)
for x_batch, xq_batch, y_batch in train_gen:
feed_dict = {story_pl: x_batch, question_pl: xq_batch, answer_pl: y_batch,
dropout_pl: FLAGS.dropout}
cost, _ = sess.run([loss, train_op], feed_dict=feed_dict)
# ?????
# if batch_id % FLAGS.show_every_n_batches == 0:
# print ('Epoch {:>3} Batch {:>4} train_loss = {:.3f}'.format(i, batch_id, cost))
batch_id += 1
# ??epoch??????
test_gen = helper.generate_data(FLAGS.batch_size, tx, txq, ty)
total_correct = 0
total = len(tx)
for tx_batch, txq_batch, ty_batch in test_gen:
feed_dict = {story_pl: tx_batch, question_pl: txq_batch, answer_pl: ty_batch,
dropout_pl: 1.0}
cor = sess.run(correct, feed_dict=feed_dict)
total_correct += int(cor)
acc = total_correct * 1.0 / total
# ??max test accuary
if acc > max_test_acc:
max_test_acc = acc
print (
'Epoch{:>3} train_loss = {:.3f} accuary = {:.3f} max_text_acc = {:.3f}'.format(i, cost, acc,
max_test_acc))
return max_test_acc
