<jupyter_start><jupyter_text>IMDb 리뷰<jupyter_code>import torch
from torch import nn, optim
from torch.utils.data import (Dataset, DataLoader, TensorDataset)
import tqdm
import glob
import pathlib
import re

remove_marks_regex = re.compile('[,\.\(\)\[\]\*:;]|<.*?>')
shift_marks_regex = re.compile('([?!])')

def text2ids(text, vocab_dict):
    # !? 이외의 기호 삭제
    text = remove_marks_regex.sub('', text)
    # !? 와 단어 사이에 공백 삽입
    text = shift_marks_regex.sub(r' \1 ', text)
    tokens = text.split()
    return [vocab_dict.get(token, 0) for token in tokens]

def list2tensor(token_idxes, max_len=100, padding=True):
    if len(token_idxes) > max_len:
        token_idxes = token_idxes[:max_len]
    n_tokens = len(token_idxes)
    if padding:
        token_idxes = token_idxes + [0] * (max_len - len(token_idxes))
    return torch.tensor(token_idxes, dtype=torch.int64), n_tokens
class IMDBDataset(Dataset):
    def __init__(self, dir_path, train=True, max_len=100, padding=True):
        self.max_len = max_len
        self.padding = padding
        
        path = pathlib.Path(dir_path)
        vocab_path = path.joinpath('imdb.vocab')
        
        # 용어집 파일을 읽어서 행 단위로 분할
        self.vocab_array = vocab_path.open().read().strip().splitlines()
        # 단어가 키이고 값이 ID인 dict 만들기
        self.vocab_dict = dict((w, i+1) for (i, w) in enumerate(self.vocab_array))
        
        if train:
            target_path = path.joinpath('train')
        else:
            target_path = path.joinpath('test')
        pos_files = sorted(glob.glob(str(target_path.joinpath('pos/*.txt'))))
        neg_files = sorted(glob.glob(str(target_path.joinpath('neg/*.txt'))))
        
        # pos는 1, neg는 0인 label을 붙여서 (file_path, label)의 튜플 리스트 작성
        self.labeled_files = \
        list(zip([0]*len(neg_files), neg_files)) + \
        list(zip([1]*len(pos_files), pos_files))
    
    @property
    def vocab_size(self):
        return len(self.vocab_array)
    
    def __len__(self):
        return len(self.labeled_files)
    
    def __getitem__(self, idx):
        label, f = self.labeled_files[idx]
        # 파일의 텍스트 데이터를 읽어서 소문자로 변환
        data = open(f).read().lower()
        # 텍스트 데이터를 ID리스트로 변환
        data = text2ids(data, self.vocab_dict)
        # ID 리스트를 텐서로 변환
        data, n_tokens = list2tensor(data, self.max_len, self.padding)
        return data, label, n_tokens
        
train_data = IMDBDataset('/home/data/torch_test/aclImdb/')
test_data = IMDBDataset('/home/data/torch_test/aclImdb/', train=False)
train_loader = DataLoader(train_data, batch_size=32, shuffle=True, num_workers=4)
test_loader = DataLoader(test_data, batch_size=32, shuffle=False, num_workers=4)<jupyter_output><empty_output><jupyter_text>신경망 정의<jupyter_code>class SequenceTaggingNet(nn.Module):
    def __init__(self, num_embeddings, embedding_dim=50, hidden_size=50,
                 num_layers=1, dropout=0.2):
        super().__init__()
        self.emb = nn.Embedding(num_embeddings, embedding_dim, padding_idx=0)
        self.lstm = nn.LSTM(embedding_dim, hidden_size, num_layers,
                           batch_first=True, dropout=dropout)
        self.linear = nn.Linear(hidden_size, 1)
    
    def forward(self, x, h0=None, l=None):
        # ID를 Embedding으로 다차원 벡터로 변환
        # x는 (batch_size, step_size) -> (batch_size, step_size, embedding_dim)
        x = self.emb(x)
        # 초기 상태 h0와 함께 RNN에 x를 전달
        # x는 (batch_size, step_size, embedding_dim) -> (batch_size, step_size, hidden_dim)
        x, h = self.lstm(x, h0)
        # 마지막 단계만 추출
        # x 는 (batch_size, step_size, hidden_dim) -> (batch_size, 1)
        if l is not None:
            # 입력의 원래 길이가 있으면 그것을 이용
            x = x[list(range(len(x))), l-1, :]
        else:
            # 없으면 단순히 마지막 것을 이용
            x = x[:, -1, :]
        
        # 추출한 마지막 단계를 선형 계측에 넣는다
        x = self.linear(x)
        # 불필요한 차원을 삭제
        # (batch_size, 1) -> (batch_size, )
        x = x.squeeze()
        return x<jupyter_output><empty_output><jupyter_text>훈련 코드<jupyter_code>def eval_net(net, data_loader, device='cpu'):
    net.eval()
    ys = []
    ypreds = []
    for x, y, l in data_loader:
        x = x.to(device)
        y = y.to(device)
        l = l.to(device)
        with torch.no_grad():
            y_pred = net(x, l=l)
            y_pred = (y_pred > 0).long()
            ys.append(y)
            ypreds.append(y_pred)
    ys = torch.cat(ys)
    ypreds = torch.cat(ypreds)
    acc = (ys == ypreds).float().sum() / len(ys)
    return acc.item()<jupyter_output><empty_output><jupyter_text>평가 코드<jupyter_code>from statistics import mean

# num_embeddings 에는 0을 포함해서 train_data.vocab_size+1를 넣는다
net = SequenceTaggingNet(train_data.vocab_size+1, num_layers=2)
net.to('cuda:0')
opt = optim.Adam(net.parameters())
loss_f = nn.BCEWithLogitsLoss()

for epoch in range(10):
    losses = []
    net.train()
    for x, y, l in tqdm.tqdm(train_loader):
        x = x.to('cuda:0')
        y = y.to('cuda:0')
        l = l.to('cuda:0')
        y_pred = net(x, l=l)
        loss = loss_f(y_pred, y.float())
        net.zero_grad()
        loss.backward()
        opt.step()
        losses.append(loss.item())
    train_acc = eval_net(net, train_loader, 'cuda:0')
    val_acc = eval_net(net, test_loader, 'cuda:0')
    print(epoch, mean(losses), train_acc, val_acc)<jupyter_output>100%|██████████| 782/782 [00:21<00:00, 36.35it/s]
  0%|          | 0/782 [00:00<?, ?it/s]<jupyter_text>RNN을 사용하지 않는 모델 작성<jupyter_code>from sklearn.datasets import load_svmlight_file
from sklearn.linear_model import LogisticRegression

train_X, train_y = load_svmlight_file(
'/home/data/torch_test/aclImdb/train/labeledBow.feat')
test_X, test_y = load_svmlight_file(
'/home/data/torch_test/aclImdb/test/labeledBow.feat', n_features=train_X.shape[1])

model = LogisticRegression(C=0.1, max_iter=1000)
model.fit(train_X, train_y)
model.score(train_X, train_y), model.score(test_X, test_y)<jupyter_output>/opt/conda/lib/python3.7/site-packages/sklearn/linear_model/_logistic.py:940: ConvergenceWarning: lbfgs failed to converge (status=1):
STOP: TOTAL NO. of ITERATIONS REACHED LIMIT.

Increase the number of iterations (max_iter) or scale the data as shown in:
    https://scikit-learn.org/stable/modules/preprocessing.html
Please also refer to the documentation for alternative solver options:
    https://scikit-learn.org/stable/modules/linear_model.html#logistic-regression
  extra_warning_msg=_LOGISTIC_SOLVER_CONVERGENCE_MSG)<jupyter_text>PackedSequence 성질을 이용한 모델 작성<jupyter_code>class SequenceTaggingNet2(SequenceTaggingNet):
    def forward(self, x, h0=None, l=None):
        # ID를 Embedding으로 다차원 벡터로 변환
        x = self.emb(x)
        
        # 길이가 주어진 경우 PackedSequence 만들기
        if l is not None:
            x = nn.utils.rnn.pack_padded_sequence(x, l, batch_first=True)
            
        # RNN에 입력
        x, h = self.lstm(x, h0)
        
        # 마지막 단계를 추출해서 선형 계층에 넣는다
        if l is not None:
            # 길이 정보가 있으면 마지막 계층의 내부 상태 벡터를 직접 이용할 수 있다
            # LSTM은 보통 내부 상태 외에 블럭 셸 상태도 가지고 있으므로 
            # 내부 상태만 사용한다
            hidden_state, cell_state = h
            x = hidden_state[-1]
        else:
            x = x[:, -1, :]
            
        # 선형 계층에 넣는다
        x = self.linear(x).squeeze()
        return x<jupyter_output><empty_output><jupyter_text>훈련 코드<jupyter_code>for epoch in range(10):
    losses = []
    net.train()
    for x, y, l in tqdm.tqdm(train_loader):
        # 길이 배열을 길이 순으로 정렬
        l, sort_idx = torch.sort(l, descending=True)
        # 얻은 인덱스를 사용해서 x, y도 정렬
        x = x[sort_idx]
        y = y[sort_idx]
        
        x = x.to('cuda:0')
        y = y.to('cuda:0')
        
        y_pred = net(x, l=l)
        loss = loss_f(y_pred, y.float())
        net.zero_grad()
        loss.backward()
        opt.step()
        losses.append(loss.item())
    train_acc = eval_net(net, train_loader, 'cuda:0')
    val_acc = eval_net(net, test_loader, 'cuda:0')
    print(epoch, mean(losses), train_acc, val_acc)<jupyter_output>100%|██████████| 782/782 [00:23<00:00, 33.26it/s]
  0%|          | 0/782 [00:00<?, ?it/s]<jupyter_text>RNN을 사용한 문장 생성<jupyter_code># 모든 ascii 문자로 사전 만들기
import string
all_chars = string.printable
vocab_size = len(all_chars)
vocab_dict = dict((c, i) for (i, c) in enumerate(all_chars))

# 문자열을 수치 리스트로 변환하는 함수
def str2ints(s, vocab_dict):
    return [vocab_dict[c] for c in s]

# 수치 리스트를 문자열로 변환하는 함수
def ints2str(x, vocab_array):
    return "".join([vocab_array[i] for i in x])
import torch
from torch import nn, optim
from torch.utils.data import (Dataset, DataLoader, TensorDataset)
import tqdm
class ShakespeareDataset(Dataset):
    def __init__(self, path, chunk_size=200):
        # 파일을 읽어서 수치 리스트로 변환
        data = str2ints(open(path).read().strip(), vocab_dict)
        
        # 텐서로 변환해서 split 한다
        data = torch.tensor(data, dtype=torch.int64).split(chunk_size)
        
        # 마지막 덩어리(chunk)의 길이를 확인해서 부족한 경우 버린다
        if len(data[-1]) < chunk_size:
            data = data[:-1]
            
        self.data = data
        self.n_chunks = len(self.data)
    
    def __len__(self):
        return self.n_chunks
    
    def __getitem__(self, idx):
        return self.data[idx]
ds = ShakespeareDataset('/home/data/torch_test/input.txt', chunk_size=200)
loader = DataLoader(ds, batch_size=32, shuffle=True, num_workers=4)
class SequenceGenerationNet(nn.Module):
    def __init__(self, num_embeddings, embedding_dim=50,
                hidden_size=50, num_layers=1, dropout=0.2):
        super().__init__()
        self.emb = nn.Embedding(num_embeddings, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_size, num_layers, 
                           batch_first=True, dropout=dropout)
        # Linear의 output 크기는 첫 Embedding의 input 크기와 같은 num_embeddings
        self.linear = nn.Linear(hidden_size, num_embeddings)
        
    def forward(self, x, h0=None):
        x = self.emb(x)
        x, h = self.lstm(x, h0)
        x = self.linear(x)
        return x, h<jupyter_output><empty_output><jupyter_text>문장 생성 함수 작성<jupyter_code>def generate_seq(net, start_phrase='The King said ', length=200,
                temperature=0.8, device='cpu'):
    # 모델을 평가 모드로 설정
    net.eval()
    # 출력 수치를 저장할 리스트
    result = []
    
    # 시작 문자열을 텐서로 변환
    start_tensor = torch.tensor(str2ints(start_phrase, vocab_dict),
                               dtype=torch.int64).to(device)
    # 선두에 batch 차원을 붙인다
    x0 = start_tensor.unsqueeze(0)
    # RNN을 통해서 출력과 새로운 내부 상태를 얻는다
    o, h = net(x0)
    # 출력을 정규화 되어 있지 않은 확률로 변환
    out_dist = o[:, -1].view(-1).exp()
    # 확률로부터 실제 문자의 인덱스를 샘플링
    top_i = torch.multinomial(out_dist, 1)[0]
    # 결과 저장
    result.append(top_i)
    
    # 생성된 결과를 차례로 RNN에 넣는다
    for i in range(length):
        inp = torch.tensor([[top_i]], dtype=torch.int64)
        inp = inp.to(device)
        o, h = net(inp, h)
        out_dist = o.view(-1).exp()
        top_i = torch.multinomial(out_dist, 1)[0]
        result.append(top_i)
    
    return start_phrase + ints2str(result, all_chars)    
from statistics import mean

net = SequenceGenerationNet(vocab_size, 20, 50, num_layers=2, dropout=0.1)
net.to('cuda:0')
opt = optim.Adam(net.parameters())
# 다중 식별 문제이므로 SoftmaxCrossEntropyLoss 가 손실 함수가 된다
loss_f = nn.CrossEntropyLoss()

for epoch in range(50):
    net.train()
    losses = []
    for data in tqdm.tqdm(loader):
        # x는 처음부터 마지막의 하나 앞 문자까지 
        x = data[:, :-1]
        # y는 두번째부터 마지막 문자까지
        y = data[:, 1:]
        
        x = x.to('cuda:0')
        y = y.to('cuda:0')
        
        y_pred, _ = net(x)
        # batch와 step 축을 통합해서 CrossEntropyLoss에 전달
        loss = loss_f(y_pred.view(-1, vocab_size), y.view(-1))
        net.zero_grad()
        loss.backward()
        opt.step()
        losses.append(loss.item())
    # 현재 손실 함수와 생성된 문장 예 표시
    print(epoch, mean(losses))
    with torch.no_grad():
        print(generate_seq(net, device='cuda:0'))<jupyter_output>100%|██████████| 175/175 [00:07<00:00, 23.85it/s]
  0%|          | 0/175 [00:00<?, ?it/s]
{'fraction_non_alphanumeric': 0.1278655962923115, 'fraction_numerical': 0.019200529669783993, 'mean_word_length': 2.5953585242487356, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 40, 'https://': 2, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '577492', 'license': 'MIT', 'n_tokens_mistral': 5220, 'n_tokens_neox': 5502, 'n_words': 980, 'path': 'ml/deep_learning/pytorch/chapter 5.ipynb', 'repo_name': 'groovallstar/test2', 'size': '12083'}