文本预处理
import collections
import re
# 读取文件
def read_time_machine():
with open('a.txt', 'r')as f:
lines = [re.sub('[^a-z]+', ' ', line.strip().lower())for linein f]
return lines
lines = read_time_machine()
print('# sentences %d' %len(lines))
# 分词
def tokenize(sentences, token='word'):
"""Split sentences into word or char tokens"""
if token =='word':
return [sentence.split(' ')for sentencein sentences]
elif token =='char':
return [list(sentence)for sentencein sentences]
else:
print('ERROR: unkown token type '+token)
tokens = tokenize(lines)
class Vocab(object):
def __init__(self, tokens, min_freq=0, use_special_tokens=False):
counter = count_corpus(tokens)# :
self.token_freqs =list(counter.items())
self.idx_to_token = []
if use_special_tokens:
# padding, begin of sentence, end of sentence, unknown
self.pad, self.bos, self.eos, self.unk = (0, 1, 2, 3)
self.idx_to_token += ['', '', '', '']
else:
self.unk =0
self.idx_to_token += ['']
self.idx_to_token += [tokenfor token, freqin self.token_freqs
if freq >= min_freqand tokennot in self.idx_to_token]
self.token_to_idx =dict()
for idx, tokenin enumerate(self.idx_to_token):
self.token_to_idx[token] = idx
def __len__(self):
return len(self.idx_to_token)
def __getitem__(self, tokens):
if not isinstance(tokens, (list, tuple)):
return self.token_to_idx.get(tokens, self.unk)
return [self.__getitem__(token)for tokenin tokens]
def to_tokens(self, indices):
if not isinstance(indices, (list, tuple)):
return self.idx_to_token[indices]
return [self.idx_to_token[index]for indexin indices]
def count_corpus(sentences):
tokens = [tkfor stin sentencesfor tkin st]
return collections.Counter(tokens)# 返回一个字典，记录每个词的出现次数
vocab = Vocab(tokens,use_special_tokens=True)
print(vocab[['i','love','shuai'],['shuai',vocab.bos]])
Sequence Sample
import torch
import random
def load_data_jay_lyrics():
with open('../data/jaychou_lyrics.txt')as f:
corpus_chars = f.read()
corpus_chars = corpus_chars.replace('\n', ' ').replace('\r', ' ')
corpus_chars = corpus_chars[0:18]
idx_to_char =list(set(corpus_chars))
char_to_idx =dict([(char, i)for i, charin enumerate(idx_to_char)])
vocab_size =len(char_to_idx)
corpus_indices = [char_to_idx[char]for charin corpus_chars]
return corpus_indices, char_to_idx, idx_to_char, vocab_size
corpus_indices, char_to_idx, idx_to_char, vocab_size = load_data_jay_lyrics()
batch_size =3
num_steps =5
def data_iter_consecutive(corpus_indices, batch_size, num_steps, device=None):
if deviceis None:
device = torch.device('cuda' if torch.cuda.is_available()else 'cpu')
corpus_len =len(corpus_indices) // batch_size * batch_size# 保留下来的序列的长度
corpus_indices = corpus_indices[: corpus_len]# 仅保留前corpus_len个字符
indices = torch.tensor(corpus_indices, device=device)
indices = indices.view(batch_size, -1)# resize成(batch_size, )
batch_num = (indices.shape[1] -1) // num_steps
for iin range(batch_num):
i = i * num_steps
X = indices[:, i: i + num_steps]
Y = indices[:, i +1: i + num_steps +1]
yield X, Y
for x,yin data_iter_consecutive(corpus_indices,batch_size,num_steps):
print(x)
print(y)
num_examples = (len(corpus_indices) -1) // num_steps# 下取整，得到不重叠情况下的样本个数
print(num_examples)
example_indices = [i * num_stepsfor iin range(num_examples)]# 每个样本的第一个字符在corpus_indices中的下标
print(example_indices)
random.shuffle(example_indices)
print(example_indices)
def data_iter_random(corpus_indices, batch_size, num_steps, device=None):
# 减1是因为对于长度为n的序列，X最多只有包含其中的前n - 1个字符
num_examples = (len(corpus_indices) -1) // num_steps# 下取整，得到不重叠情况下的样本个数
example_indices = [i * num_stepsfor iin range(num_examples)]# 每个样本的第一个字符在corpus_indices中的下标
random.shuffle(example_indices)
num_examples = num_examples // batch_size * batch_size
def _data(i):
# 返回从i开始的长为num_steps的序列
return corpus_indices[i: i + num_steps]
if deviceis None:
device = torch.device('cuda' if torch.cuda.is_available()else 'cpu')
for iin range(0, num_examples, batch_size):
# 每次选出batch_size个随机样本
batch_indices = example_indices[i: i + batch_size]# 当前batch的各个样本的首字符的下标
X = [_data(j)for jin batch_indices]
Y = [_data(j +1)for jin batch_indices]
yield torch.tensor(X, device=device), torch.tensor(Y, device=device)
x_list = []
num =0
for x,yin data_iter_random(corpus_indices,batch_size,num_steps):
num +=1
print(num)
a = [1,2,3,4,5,6,7,8,9,10,11,12]
a = torch.tensor(a)
print(a.reshape((3,-1)))
