<jupyter_start><jupyter_text>Language modelling Обучим две различные символьные модели для генерации динозавров:* модель на символьных биграмах* ***RNN***-модель.  Bigram model<jupyter_code>!cat dinos.txt | tail
names = ['<' + name.strip().lower() + '>' for name in open('dinos.txt').readlines()]
print(names[:10])
import nltk<jupyter_output><empty_output><jupyter_text>Вычислим частоту каждого символа в корпусе имен динозавров<jupyter_code>chars = [char for name in names for char in name]
freq = nltk.FreqDist(chars)
freq
print(list(freq.keys()))
freq.most_common(10)<jupyter_output><empty_output><jupyter_text>Define a function to estimate probabilty of character<jupyter_code>l = sum([freq[char] for char in freq])
def unigram_prob(char):
    return freq[char] / l
print('p(a) = %1.4f' %unigram_prob('a'))<jupyter_output>p(a) = 0.1160<jupyter_text>Вычислим условную вероятность каждого символа в зависимости от того, какой символ стоял на предыдущей позиции.<jupyter_code>cfreq = nltk.ConditionalFreqDist(nltk.bigrams(chars))
cfreq['a']<jupyter_output><empty_output><jupyter_text>Оценим условные вероятности с помощью MLE.<jupyter_code>cprob = nltk.ConditionalProbDist(cfreq, nltk.MLEProbDist)
print('p(a a) = %1.4f' %cprob['a'].prob('a'))
print('p(a b) = %1.4f' %cprob['a'].prob('b'))
print('p(a u) = %1.4f' %cprob['a'].prob('u'))
cprob['a'].generate()
import numpy an np<jupyter_output><empty_output><jupyter_text>Задание 1.1) Напишите функцию, которая генерирует имя динозавра **фиксированной** длины. Используйте '<' как начальный символ.2) Напишите функцию, которая генерирует имя динозавра любой дины.<jupyter_code>def generate_n_chars(cprob, n):
    name = '<'
    for i in range(n):
        name += cprob[name[-1]].generate()
        
        if name[-1] == '>':
            break
    
    if name[-1] != '>':
        name += '>'
        return name<jupyter_output><empty_output><jupyter_text>Реккурентные нейронные сети (RNN)Исходная последовательность:$x_{1:n} = x_1, x_2, \ldots, x_n$, $x_i \in \mathbb{R}^{d_{in}}$Для каждого входного значения $x_{1:i}$ получаем на выходе $y_i$:$y_i = RNN(x_{1:i})$, $y_i \in \mathbb{R}^{d_{out}}$Для всей последовательности $x_{1:n}$:$y_{1:n} = RNN^{*}(x_{1:n})$, $y_i \in \mathbb{R}^{d_{out}}$ $R$ - рекурсивная функция активации, зависящая от двух параметров: $x_i$ и $s_{i-1}$ (вектор предыдущего состояния)$RNN^{*}(x_{1:n}, s_0) = y_{1:n}$$y_i = O(s_i) = g(W^{out}[s_{i} ,x_i] +b)$$s_i = R(s_{i-1}, x_i)$$s_i = R(s_{i-1}, x_i) = g(W^{hid}[s_{i-1} ,x_i] +b)$  -- конкатенация $[s_{i-1}, x]$$x_i \in \mathbb{R}^{d_{in}}$, $y_i \in \mathbb{R}^{ d_{out}}$, $s_i \in \mathbb{R}^{d_{hid}}$$W^{hid} \in \mathbb{R}^{(d_{in}+d_{out}) \times d_{hid}}$, $W^{out} \in \mathbb{R}^{d_{hid} \times d_{out}}$ Построим языковую модель на основе RNN с помощью pytorch<jupyter_code>import numpy as np
import random
import torch
import torch.nn as nn
import torch.optim as optim
import pdb
from torch.utils.data import Dataset, DataLoader

%load_ext autoreload
%autoreload 2

torch.set_printoptions(linewidth=200)
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
hidden_size = 50<jupyter_output><empty_output><jupyter_text>Подготовим данные<jupyter_code>class DinosDataset(Dataset):
    def __init__(self):
        super().__init__()
        with open('dinos.txt') as f:
            content = f.read().lower()
            self.vocab = sorted(set(content)) + ['<', '>']
            self.vocab_size = len(self.vocab)
            self.lines = content.splitlines()
        self.ch_to_idx = {c:i for i, c in enumerate(self.vocab)}
        self.idx_to_ch = {i:c for i, c in enumerate(self.vocab)}
    
    def __getitem__(self, index):
        line = self.lines[index]
        #teacher forcing
        x_str = '<' + line 
        y_str = line + '>' 
        x = torch.zeros([len(x_str), self.vocab_size], dtype=torch.float)
        y = torch.empty(len(x_str), dtype=torch.long)
#         y = torch.empty(len(x_str), dtype=torch.int) # не будет работать т.к. pytorch ожидает .long
#         print('x', x)
#         print('y', y)
        for i, (x_ch, y_ch) in enumerate(zip(x_str, y_str)):
            x[i][self.ch_to_idx[x_ch]] = 1
            y[i] = self.ch_to_idx[y_ch]
#         print('x', x)
#         print('y', y)        
        return x, y
    
    def __len__(self):
        return len(self.lines)
trn_ds = DinosDataset()
# trn_dl = DataLoader(trn_ds, shuffle=True)
trn_ds.lines[1]
print(trn_ds.idx_to_ch)
trn_ds.vocab_size
x, y = trn_ds[1]
x.shape
y.shape
[trn_ds.idx_to_ch[i.item()] for i in y]<jupyter_output><empty_output><jupyter_text>Опишем модель, функцию потерь и алгоритм оптимизации<jupyter_code>class RNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__()
        self.i2h = nn.Linear(input_size + hidden_size, hidden_size)
        self.dropout = nn.Dropout(0.3)
        self.i2o = nn.Linear(hidden_size, output_size)
    
    def forward(self, h_prev, x):
        combined = torch.cat([h_prev, x], dim = 1) # concatenate x and h
        h = torch.tanh(self.dropout(self.i2h(combined)))
        y = self.i2o(h)
        return h, y
model = RNN(trn_ds.vocab_size, hidden_size, trn_ds.vocab_size).to(device)
loss_fn = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=1e-2)<jupyter_output><empty_output><jupyter_text>![rnn](images/dinos3.png)<jupyter_code>def sample(model): # функция пораждает список индексов символов
    model.eval()
    word_size=0
    end_char_idx = trn_ds.ch_to_idx['>'] # чтобы понимать, когда слово кончилось (индекс последнего символа)
    start_char_idx = trn_ds.ch_to_idx['<'] # seed для начала слова (индекс начального символа)
    with torch.no_grad():
        h_prev = torch.zeros([1, hidden_size], dtype=torch.float, device=device) # первый вектор скрытого состояния
        x = h_prev.new_zeros([1, trn_ds.vocab_size])
        
        # "сгенерирвали" начало слова
        indices = [start_char_idx]
        x[0, start_char_idx] = 1 # one-hot вектор для начаьного символа
        predicted_char_idx = start_char_idx
        
        while predicted_char_idx != end_char_idx and word_size != 50:
            h_prev, y_pred = model(h_prev, x)
            y_softmax_scores = torch.softmax(y_pred, dim=1)
            
            np.random.seed(np.random.randint(1, 5000)) # для увеливичения случайностей
            # .cpu().numpy() перевод на процессор потому-что np работает с числами на процессоре | .ravel() убрать измерение
            idx = np.random.choice(np.arange(trn_ds.vocab_size), p=y_softmax_scores.cpu().numpy().ravel()) # список символов и их вероятности
            indices.append(idx)
            
            # нули все кроме самого вероятного. Подаем самое лучшее, чтобы избежать шаблонов
            x = (y_pred == y_pred.max(1)[0]).float() # нетривиальное решение: подаём не всегда то же, что выдали
            
            predicted_char_idx = idx
            
            word_size += 1
        
        if word_size == 50:
            indices.append(newline_idx) # если 50 то добавляем символ конца строки
    return indices
def print_sample(sample_idxs):
#     [print(trn_ds.idx_to_ch[x], end ='') for x in sample_idxs]
#     print()
    print(''.join([idx_to_ch[x] for x in sample_idxs])) # принимаем индексы(словарь) и в цикле возвращием символы в одной строке<jupyter_output><empty_output><jupyter_text>Обучим получившуюся модель<jupyter_code>def train_one_epoch(model, loss_fn, optimizer):
    model.train()
    for line_num, (x, y) in enumerate(trn_dl):
        loss = 0
        optimizer.zero_grad()
        h_prev = torch.zeros([1, hidden_size], dtype=torch.float, device=device)
        x, y = x.to(device), y.to(device)
        for i in range(x.shape[1]):
            h_prev, y_pred = model(h_prev, x[:, i])
            loss += loss_fn(y_pred, y[:, i])
            
        if (line_num+1) % 100 == 0: # можно печтать реже заменить на 200
#         if (line_num) % 100 == 0:
#             print('loss', loss.item() / y.size()[0])
            print_sample(sample(model))
#         optimizer.zero_grad()           
        loss.backward()
        optimizer.step()
def train(model, loss_fn, optimizer, dataset='dinos', epochs=1):
    for e in range(1, epochs+1):
        print('Epoch:{}'.format(e))
        train_one_epoch(model, loss_fn, optimizer)
        print()
train(model, loss_fn, optimizer, epochs = 10)<jupyter_output>Epoch:1
<ig>
<qzorep>
<drrui>
<saurus>
<iaqnosaisus>
<namrus>
<haurhscunur>
<yrsasturus>
<kyrippsaliamobtesaftoa>
<dinrasausus>
<turklour>
<saurucaurus>
<slsah>
<mannupalros>
<oudutsneuss>

Epoch:2
<shrotapasrus>
<euaasabrus>
<iaucmtaurus>
<klrur>
<sprusaures>
<aurucauras>
<telaoaaurus>
<tnnnuseurus>
<tanras>
<aulucauras>
<smlataitaurus>
<llronuonoosaurus>
<aulucaurds>
<sanaucaurus>
<trndsausus>

Epoch:3
<srbrom>
<scohsanruk>
<cunibilrus>
<cpronhsaurus>
<sparhansdaurus>
<laresgpaurus>
<ugsstlkaurus>
<pttbgkcaurus>
<ahonaoneurus>
<hvrugrsiurus>
<yonotrcaurus>
<slsactocaurus>
<samtssurus>
<doriyrlsaurus>
<duasmkorusnur>

Epoch:4
<hacsas>
<puasstsaurus>
<ltrucaurus>
<sltalpurus>
<gacrasuurus>
<snerop>
<spostarmdoon>
<duaatnaoltos>
<tciuosaurus>
<qrhxsosaurus>
<fuashrdauras>
<slgasaptptius>
<mcurizturus>
<ihcgtcsaurus>
<aelrcnaurus>

Epoch:5
<uisus>
<diurysaurus>
<laltbsg>
<aaurapaurus>
<antusaurus>
<sortusaures>
<amsarnsaurur>
<crmgcourus>
<cotopaurus>
<lrrtasarrus>
<gtcroraurus>
<asapunt[...]<jupyter_text>**Перепешите функцию сэмплирования так, чтобы она порождала панграмы (слова, в которых каждая буква встречается только 1-ин раз)**<jupyter_code>def get_mask(indices):
    mask = []
    for i in np.arange(trn_ds.vocab_size):
        if i in indices and i != trn_ds.ch_to_idx['u']:
            mask.append(0)
        else:
            mask.append(1)
    return np.array(mask)
def sample(model):
    model.eval()
    word_size=0
    end_char_idx = trn_ds.ch_to_idx['>']
    start_char_idx = trn_ds.ch_to_idx['<']
    with torch.no_grad():
        h_prev = torch.zeros([1, hidden_size], dtype=torch.float, device=device)
        x = h_prev.new_zeros([1, trn_ds.vocab_size])
        
        indices = [start_char_idx]
        x[0, start_char_idx] = 1
        predicted_char_idx = start_char_idx
        
        while predicted_char_idx != end_char_idx and word_size != 50:
            h_prev, y_pred = model(h_prev, x)
            y_softmax_scores = torch.softmax(y_pred, dim=1)
            
            np.random.seed(np.random.randint(1, 5000))
            probas = y_softmax_scores.cpu().numpy().ravel() #1
            probas *= get_mask(indices) #2
            sum_probas = sum(probas) #3
            idx = np.random.choice(np.arange(trn_ds.vocab_size), p=[el / sum_probas for el in probas]) #4
            indices.append(idx)
            
            x = (y_pred == y_pred.max(1)[0]).float()
            
            predicted_char_idx = idx
            
            word_size += 1
        
        if word_size == 50:
            indices.append(newline_idx)
    return indices
def print_sample(sample_idxs):
    print(''.join([idx_to_ch[x] for x in sample_idxs]))
for i in range(10):
    print_sample(sample(model))<jupyter_output><empty_output><jupyter_text>**Перепешите функцию сэмплирования так, чтобы было возможно менять температуру сэмплирования**<jupyter_code>def sample(model):
    model.eval()
    word_size=0
    end_char_idx = trn_ds.ch_to_idx['>']
    start_char_idx = trn_ds.ch_to_idx['<']
    
    with torch.no_grad():
        h_prev = torch.zeros([1, hidden_size], dtype=torch.float, device=device)
        x = h_prev.new_zeros([1, trn_ds.vocab_size])
        
        indices = [start_char_idx]
        x[0, start_char_idx] = 1
        predicted_char_idx = start_char_idx
        
        while predicted_char_idx != end_char_idx and word_size != 50:
            h_prev, y_pred = model(h_prev, x)
            y_softmax_scores = torch.softmax(y_pred, dim=1)
            
            np.random.seed(np.random.randint(1, 5000))
            probas = y_softmax_scores.cpu().numpy().ravel() #1
            probas = probas ** (1/T) #2 меняем вероятности в соответствии с температурой
            sum_probas = sum(probas) #3
            idx = np.random.choice(np.arange(trn_ds.vocab_size), p=[el / sum_probas for el in probas]) #4
            indices.append(idx)
            
            x = (y_pred == y_pred.max(1)[0]).float()
            predicted_char_idx = idx
            word_size += 1
        
        if word_size == 50:
            indices.append(newline_idx)
    return indices
# температура = 1 значит без изменений, но чем выше температура тем больше изменений
for i in range(10):
    print_sample(sample(model, 1))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13816151608083668, 'fraction_numerical': 0.012502948808681292, 'mean_word_length': 3.3376534788540244, 'pattern_counts': {'":': 0, '<': 131, '<?xml version=': 0, '>': 131, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '660716', 'license': 'MIT', 'n_tokens_mistral': 5179, 'n_tokens_neox': 5303, 'n_words': 963, 'path': 'Lectures notebooks/(Lectures notebooks) netology Machine learning/25. Language models. Countable language models and probabilistic language models. Recurrent neural networks/language_models.ipynb', 'repo_name': 'Alex110117/data_analysis', 'size': '12717'}