RNN/LSTM 텍스트 분류 & Doc2Vec+LSTM 결합 정리

멀티캠퍼스 데이터분석 6월 15일 수업 내용 — RNN 감성 분류, LSTM 자연어 모델 문제(head_type 비교), Doc2Vec 모델 저장/로드 + LSTM 결합

---

📌 목차

1. RNN 감성 분류 실습
2. LSTM 자연어 모델 문제 — head_type 비교
3. Doc2Vec 모델 저장/로드 + LSTM 결합

---

1. RNN 감성 분류 실습

데이터 로드 & 토큰화

import pandas as pd
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader, random_split
from torch.nn.utils.rnn import pad_sequence
from konlpy.tag import Komoran
from collections import Counter
from tqdm import tqdm   # 진행 상태 로그 표시

df = pd.read_csv('../data/ratings_train.txt', sep='\t')
df.dropna(inplace=True)
df.drop_duplicates('document', inplace=True)
df = df[:5000]

komoran = Komoran()

def tokenize(text):
    allow_pos = ['NNP', 'NNG', 'VV', 'VA', 'MAG', 'SL']
    result = []
    for word, pos in komoran.pos(text):
        if pos in allow_pos:
            result.append(word)
    return result

tokenized_sentence = [tokenize(text) for text in df['document']]

단어 사전 생성

vocab = {'<PAD>': 0, '<UNK>': 1}

all_tokens   = [token for tokens in tokenized_sentence for token in tokens]
token_counts = Counter(all_tokens)

# 최소 등장 3회 이상인 단어만 사전에 등록
for token, count in token_counts.items():
    if count >= 3:
        vocab[token] = len(vocab)

💡 dict.get(key, default)
키가 존재하면 해당 값을, 존재하지 않으면 두 번째 인자(default)를 반환합니다.
vocab.get('마케팅', vocab['<UNK>']) → 사전에 없으면 <UNK> 인덱스 반환

Dataset & collate_fn

class RNNDataset(Dataset):
    def __init__(self, tokenized_texts, labels, vocab):
        self.labels = labels.values
        self.data = [
            [vocab.get(token, vocab['<UNK>']) for token in tokens]
            for tokens in tokenized_texts
        ]
    def __len__(self):
        return len(self.data)
    def __getitem__(self, idx):
        return (torch.tensor(self.data[idx], dtype=torch.long),
                torch.tensor(self.labels[idx], dtype=torch.long))

def collate_fn(batch):
    text_list  = [item[0] for item in batch]
    label_list = [item[1] for item in batch]

    # 배치 내 최대 길이에 맞춰 패딩
    padded_texts = pad_sequence(text_list, batch_first=True, padding_value=vocab['<PAD>'])
    labels       = torch.tensor(label_list, dtype=torch.long)
    return padded_texts, labels

train/val 분할 & DataLoader

dataset = RNNDataset(tokenized_sentence, df['label'], vocab)

train_size = int(len(dataset) * 0.8)   # int() — 정수 인덱스/크기로 사용하기 위해 변환
test_size  = len(dataset) - train_size

train_dataset, val_dataset = random_split(dataset, [train_size, test_size])

train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True, collate_fn=collate_fn)
val_loader   = DataLoader(val_dataset,   batch_size=64, shuffle=True, collate_fn=collate_fn)

💡 random_split() — Dataset을 지정한 크기 비율로 무작위 분할합니다.
train_test_split() 과 달리 PyTorch Dataset 객체에 바로 사용할 수 있습니다.

RNNCLF 모델 정의

class RNNCLF(nn.Module):
    def __init__(self, vocab_size, emb_dim, hidden_size, num_classes):
        super().__init__()
        self.emb = nn.Embedding(vocab_size, emb_dim, padding_idx=vocab['<PAD>'])
        self.rnn = nn.RNN(emb_dim, hidden_size, batch_first=True)
        self.fc  = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        embedding = self.emb(x)   # [batch, seq_len, emb_dim]

        # rnn_out : 모든 시점의 출력 [batch, seq_len, hidden_size]
        # hidden  : 마지막 시점의 은닉 상태 [1, batch, hidden_size]
        rnn_out, hidden = self.rnn(embedding)

        last_hidden = hidden.squeeze(0)   # [batch, hidden_size]
        return self.fc(last_hidden)

학습 루프 (tqdm 진행 표시)

model     = RNNCLF(len(vocab), emb_dim=64, hidden_size=128, num_classes=2)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

epochs = 50

for epoch in range(epochs):
    model.train()
    train_loss, correct_train, total_train = 0, 0, 0

    for inputs, labels in tqdm(train_loader, desc=f'Epoch {epoch+1}/{epochs} Train'):
        optimizer.zero_grad()
        output = model(inputs)
        loss   = criterion(output, labels)
        loss.backward()
        optimizer.step()

        train_loss     += loss.item()
        pred            = torch.argmax(output, dim=1)
        correct_train  += (pred == labels).sum().item()
        total_train    += labels.size(0)

    train_acc      = (correct_train / total_train) * 100
    avg_train_loss = train_loss / len(train_loader)

    # 검증
    model.eval()
    val_loss, correct_val, total_val = 0, 0, 0

    with torch.no_grad():
        for inputs, labels in val_loader:
            output = model(inputs)
            loss   = criterion(output, labels)

            val_loss      += loss.item()
            pred            = torch.argmax(output, dim=1)
            correct_val    += (pred == labels).sum().item()
            total_val      += labels.size(0)

    val_acc      = (correct_val / total_val) * 100
    avg_val_loss = val_loss / len(val_loader)

    if (epoch + 1) % 10 == 0:
        print(f'Train Loss: {round(avg_train_loss, 4)} Train Acc: {train_acc}')
        print(f'Val Loss: {round(avg_val_loss, 4)} Val Acc: {val_acc}')

---

2. LSTM 자연어 모델 문제 — head_type 비교

텍스트 정규화 & 전처리

import re
import pandas as pd
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader, random_split
from torch.nn.utils.rnn import pad_sequence
from konlpy.tag import Komoran
from collections import Counter
from tqdm import tqdm

df = pd.read_csv('../data/ratings_train.txt', sep='\t')

def normalize(text):
    text = re.sub(r'[^가-힣0-9a-zA-Z\s\.]', ' ', str(text))
    text = re.sub(r'\s+', ' ', text).strip()
    return text

df['document'] = df['document'].map(normalize)
df = df.loc[~(df['document'] == '')]   # 공백 텍스트 제거

df.dropna(inplace=True)
df.drop_duplicates('document', inplace=True)

토큰화 & 단어 사전 (최소 출현 2회)

komoran = Komoran()

def tokenize(text):
    allow_pos = ['NNP', 'NNG', 'VV', 'VA', 'MAG', 'SL']
    return [word for word, pos in komoran.pos(text) if pos in allow_pos]

df = df[:5000]
tokenized_sentences = [tokenize(text) for text in df['document']]

vocab = {'<PAD>': 0, '<UNK>': 1}
all_tokens  = [token for tokens in tokenized_sentences for token in tokens]
token_count = Counter(all_tokens)

for token, count in token_count.items():
    if count >= 2:
        vocab[token] = len(vocab)

Dataset & collate_fn (zip 언패킹 방식)

class LSTMDataset(Dataset):
    def __init__(self, tokenized_texts, labels, vocab):
        self.labels = labels.values
        self.data = [
            [vocab.get(token, vocab['<UNK>']) for token in tokens]
            for tokens in tokenized_texts
        ]
    def __len__(self):
        return len(self.data)
    def __getitem__(self, idx):
        return (torch.tensor(self.data[idx], dtype=torch.long),
                torch.tensor(self.labels[idx], dtype=torch.long))

dataset = LSTMDataset(tokenized_sentences, df['label'], vocab)

train_size = int(len(dataset) * 0.8)
val_size   = len(dataset) - train_size
train_dataset, val_dataset = random_split(dataset, [train_size, val_size])

def collate_fn(batch):
    # zip(*batch)로 (text들, label들) 언패킹
    text_list, labels = zip(*batch)
    padded_texts = pad_sequence(text_list, batch_first=True, padding_value=vocab['<PAD>'])
    labels       = torch.tensor(labels, dtype=torch.long)
    return padded_texts, labels

train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True, collate_fn=collate_fn)
val_loader   = DataLoader(val_dataset,   batch_size=64, shuffle=True, collate_fn=collate_fn)

LSTMCLF 모델 — head_type 선택 가능

class LSTMCLF(nn.Module):
    def __init__(self, vocab_size, emb_dim, hidden_size, num_classes,
                 dropout=0.5, head_type='last'):
        super().__init__()
        self.head_type = head_type

        self.emb = nn.Embedding(vocab_size, emb_dim, padding_idx=vocab['<PAD>'])
        torch.nn.init.xavier_uniform_(self.emb.weight)   # Xavier 초기화

        self.lstm    = nn.LSTM(emb_dim, hidden_size, batch_first=True)
        self.dropout = nn.Dropout(dropout)
        self.fc      = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        embedding = self.emb(x)

        # LSTM 출력: (out, (hidden, cell))
        lstm_out, (hidden, cell) = self.lstm(embedding)

        if self.head_type == 'last':
            # 마지막 시점의 은닉 상태만 사용
            last_hidden = hidden.squeeze(0)
        elif self.head_type == 'mean':
            # 모든 시점의 평균
            last_hidden = torch.mean(lstm_out, dim=1)
        elif self.head_type == 'max':
            # 모든 시점 중 최댓값 (Max Pooling)
            last_hidden, _ = torch.max(lstm_out, dim=1)

        dropout_hidden = self.dropout(last_hidden)
        return self.fc(dropout_hidden)

head_type 옵션 비교

head_type	사용 데이터	특징
'last'	마지막 시점 은닉 상태	문장 전체 맥락이 압축된 최종 상태
'mean'	모든 시점의 평균	문장 전체에서 고른 정보 반영
'max'	모든 시점 중 최댓값	가장 강하게 활성화된 특징 강조

model = LSTMCLF(len(vocab), emb_dim=64, hidden_size=128,
                num_classes=2, head_type='max')

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

epochs = 50

for epoch in range(epochs):
    model.train()
    train_loss, correct_train, total_train = 0, 0, 0

    for inputs, labels in tqdm(train_loader, desc=f'Epoch {epoch+1}/{epochs}'):
        optimizer.zero_grad()
        output = model(inputs)
        loss   = criterion(output, labels)
        loss.backward()
        optimizer.step()

        train_loss    += loss.item()
        pred           = torch.argmax(output, dim=1)
        correct_train += (pred == labels).sum().item()
        total_train   += labels.size(0)

    train_acc      = (correct_train / total_train) * 100
    avg_train_loss = train_loss / len(train_loader)

    model.eval()
    val_loss, correct_val, total_val = 0, 0, 0

    with torch.no_grad():
        for inputs, labels in val_loader:
            output = model(inputs)
            loss   = criterion(output, labels)

            val_loss     += loss.item()
            pred          = torch.argmax(output, dim=1)
            correct_val  += (pred == labels).sum().item()
            total_val    += labels.size(0)

    val_acc      = (correct_val / total_val) * 100
    avg_val_loss = val_loss / len(val_loader)

    if (epoch + 1) % 10 == 0:
        print(f'LSTM Train Loss: {round(avg_train_loss, 4)} Train Acc: {train_acc}')
        print(f'LSTM Val Loss: {round(avg_val_loss, 4)} Val Acc: {val_acc}')

---

3. Doc2Vec 모델 저장/로드 + LSTM 결합

Doc2Vec으로 문서를 벡터화한 뒤, 그 벡터를 LSTM의 입력으로 사용하는 하이브리드 구조입니다.

Doc2Vec 학습 & 모델 저장

import pandas as pd
from konlpy.tag import Komoran
from gensim.models.doc2vec import Doc2Vec, TaggedDocument
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader, random_split
from tqdm import tqdm

df = pd.read_csv('../data/ratings_train.txt', sep='\t')
df = df[:100]

komoran = Komoran()

def tokenize(text):
    allow_pos = ['NNP', 'NNG', 'VV', 'VA', 'MAG', 'SL']
    return [word for word, pos in komoran.pos(text) if pos in allow_pos]

tokenized_sentence = [tokenize(text) for text in df['document']]

tagged_data = [
    TaggedDocument(words=doc, tags=[str(i)])
    for i, doc in enumerate(tokenized_sentence)
]

d2v = Doc2Vec(
    tagged_data, vector_size=64, window=5,
    min_count=1, workers=2, epochs=20
)

# 모델 저장
d2v.save('my_model.model')

💡 모델 저장/로드 시 주의
save() 한 파일명과 load() 할 파일명이 정확히 일치해야 합니다.
경로가 다르면 FileNotFoundError 가 발생합니다.

저장된 모델 로드 & Test 데이터 벡터화

loaded_doc2vec = Doc2Vec.load('my_doc2vec.model')

df = pd.read_csv('../data/ratings_test.txt', sep='\t')
df.dropna(inplace=True)
df.drop_duplicates('document', inplace=True)
df = df[:5000]

# 학습 시 사용한 것과 동일한 토큰화 함수 사용 (필수)
tokenized_sentences = [tokenize(text) for text in df['document']]

X_vector = [loaded_doc2vec.infer_vector(sentence) for sentence in tokenized_sentences]

X = np.array(X_vector)
y = df['label'].values
print(X.shape)   # (5000, 64)

⚠️ 토큰화 함수 일관성
학습 시 사용한 토큰화 함수와 추론(예측) 시 사용하는 토큰화 함수가 반드시 동일해야 합니다.
다른 전처리를 사용하면 벡터 공간이 어긋나 예측 성능이 저하됩니다.

LSTMDataset — 벡터 입력형

class LSTMDataset(Dataset):
    def __init__(self, vectors, labels):
        self.labels = labels
        self.data   = vectors
    def __len__(self):
        return len(self.data)
    def __getitem__(self, idx):
        return (torch.tensor(self.data[idx], dtype=torch.float32),
                torch.tensor(self.labels[idx], dtype=torch.long))

dataset = LSTMDataset(X, y)

train_size = int(len(dataset) * 0.8)
val_size   = len(dataset) - train_size
train_dataset, val_dataset = random_split(dataset, [train_size, val_size])

# 이미 고정 크기 벡터이므로 collate_fn(패딩) 불필요
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)
val_loader   = DataLoader(val_dataset,   batch_size=64, shuffle=True)

💡 이전 실습과의 차이 — collate_fn이 필요 없는 이유
토큰 시퀀스는 길이가 제각각이라 패딩이 필요했지만,
Doc2Vec으로 벡터화한 데이터는 이미 고정 크기(vector_size)이므로 패딩이 불필요합니다.

LSTMCLF — Embedding 없이 벡터 직접 입력

class LSTMCLF(nn.Module):
    def __init__(self, input_dim, hidden_size, num_classes,
                 dropout=0.5, head_type='last'):
        super().__init__()
        self.head_type = head_type
        # input_dim : Doc2Vec 벡터 차원 수 (Embedding 레이어 불필요)
        self.lstm    = nn.LSTM(input_dim, hidden_size, batch_first=True)
        self.dropout = nn.Dropout(dropout)
        self.fc      = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        # LSTM은 시퀀스 입력이 필요하므로 차원 확장
        x = x.unsqueeze(1)   # [batch, input_dim] → [batch, 1, input_dim]

        lstm_out, (hidden, cell) = self.lstm(x)

        if self.head_type == 'last':
            last_hidden = hidden.squeeze(0)
        elif self.head_type == 'mean':
            last_hidden = torch.mean(lstm_out, dim=1)
        elif self.head_type == 'max':
            last_hidden, _ = torch.max(lstm_out, dim=1)

        dropout_hidden = self.dropout(last_hidden)
        return self.fc(dropout_hidden)

💡 x.unsqueeze(1) 이 필요한 이유
Doc2Vec 벡터는 시퀀스가 아닌 단일 벡터([batch, input_dim])입니다.
LSTM은 [batch, seq_len, input_dim] 형태의 시퀀스 입력이 필요하므로
seq_len=1인 차원을 추가해줍니다.

학습 실행

model = LSTMCLF(input_dim=64, hidden_size=128, num_classes=2,
                head_type='last', dropout=0.3)

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

epochs = 50

for epoch in range(epochs):
    model.train()
    train_loss, correct_train, total_train = 0, 0, 0

    for inputs, labels in tqdm(train_loader, desc=f'Epoch {epoch+1}/{epochs}'):
        optimizer.zero_grad()
        output = model(inputs)
        loss   = criterion(output, labels)
        loss.backward()
        optimizer.step()

        train_loss    += loss.item()
        pred           = torch.argmax(output, dim=1)
        correct_train += (pred == labels).sum().item()
        total_train   += labels.size(0)

    train_acc      = (correct_train / total_train) * 100
    avg_train_loss = train_loss / len(train_loader)

    model.eval()
    val_loss, correct_val, total_val = 0, 0, 0

    with torch.no_grad():
        for inputs, labels in val_loader:
            output = model(inputs)
            loss   = criterion(output, labels)

            val_loss     += loss.item()
            pred          = torch.argmax(output, dim=1)
            correct_val  += (pred == labels).sum().item()
            total_val    += labels.size(0)

    val_acc      = (correct_val / total_val) * 100
    avg_val_loss = val_loss / len(val_loader)

    if (epoch + 1) % 10 == 0:
        print(f'LSTM Train Loss: {round(avg_train_loss, 4)} Train Acc: {train_acc}')
        print(f'LSTM Val Loss: {round(avg_val_loss, 4)} Val Acc: {val_acc}')

3가지 접근법 비교

	RNN (토큰 시퀀스)	LSTM (토큰 시퀀스)	LSTM (Doc2Vec 벡터)
입력 형태	가변 길이 토큰	가변 길이 토큰	고정 크기 벡터
Embedding 레이어	필요	필요	불필요
collate_fn (패딩)	필요	필요	불필요
시퀀스 길이	문장 길이만큼	문장 길이만큼	1 (단일 벡터)
사전 학습	없음	없음	Doc2Vec 별도 학습 필요

---

📎 핵심 개념 요약

개념	설명
dict.get(key, default)	키 없으면 기본값 반환
random_split(dataset, [a, b])	Dataset을 비율대로 무작위 분할
tqdm(loader, desc=...)	반복문 진행 상태를 로그로 표시
pad_sequence()	가변 길이 시퀀스를 최대 길이로 패딩
hidden.squeeze(0)	RNN/LSTM 배치 차원(1) 제거
RNN forward 반환	(out, hidden)
LSTM forward 반환	(out, (hidden, cell))
head_type='last'	마지막 시점 은닉 상태 사용
head_type='mean'	모든 시점 평균 사용
head_type='max'	모든 시점 중 최댓값 사용
nn.init.xavier_uniform_()	가중치 균등 분포 초기화
Doc2Vec.save() / .load()	모델 저장 및 불러오기
infer_vector()	학습된 Doc2Vec으로 새 문서 벡터화
토큰화 함수 일관성	학습/추론 시 동일한 전처리 필수
x.unsqueeze(1)	단일 벡터를 시퀀스 형태로 차원 확장
torch.argmax(output, dim=1)	가장 높은 확률의 클래스 인덱스
zip(*batch)	배치 데이터를 (텍스트들, 라벨들)로 언패킹