RNN (Recurrent Neural Network) 자연어 처리

RNN (Recurrent Neural Network) 자연어 처리

231018 학습한 내용 정리

RNN 개요

정의

• Recurrent Neural Network (순환 신경망)
• 시퀀스 데이터 처리에 특화된 딥러닝 모델
• 이전 시간의 정보를 현재 시간에 활용

특징

• 시퀀스 처리: 순서가 있는 데이터를 효과적으로 처리
• 메모리: 이전 정보를 기억하는 능력
• 가중치 공유: 시간에 따라 동일한 가중치 사용
• 순환 구조: 출력이 다시 입력으로 사용되는 구조

장점

• 시퀀스 특화: 시퀀스 데이터에 최적화된 구조
• 메모리 효과: 이전 정보를 활용한 예측
• 유연성: 다양한 길이의 시퀀스 처리 가능
• 해석성: 시간에 따른 변화를 추적 가능

RNN 기본 구조

1. 기본 RNN

import tensorflow as tf
import numpy as np
import matplotlib.pyplot as plt
from tensorflow.keras import layers, models
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
from sklearn.model_selection import train_test_split

# 샘플 텍스트 데이터 생성
texts = [
    "I love this movie",
    "This movie is great",
    "I hate this film",
    "This film is terrible",
    "The acting is amazing",
    "The plot is boring",
    "Great cinematography",
    "Poor direction",
    "Excellent performance",
    "Disappointing ending"
]

# 간단한 감정 레이블 (0: 부정, 1: 긍정)
labels = [1, 1, 0, 0, 1, 0, 1, 0, 1, 0]

# 텍스트 토크나이징
tokenizer = Tokenizer(num_words=100, oov_token="<OOV>")
tokenizer.fit_on_texts(texts)
sequences = tokenizer.texts_to_sequences(texts)

# 시퀀스 패딩
max_length = 10
padded_sequences = pad_sequences(sequences, maxlen=max_length, padding='post')

# 데이터 분할
X_train, X_test, y_train, y_test = train_test_split(
    padded_sequences, labels, test_size=0.2, random_state=42
)

# 기본 RNN 모델
model = models.Sequential([
    layers.Embedding(100, 16, input_length=max_length),
    layers.SimpleRNN(32, return_sequences=False),
    layers.Dense(16, activation='relu'),
    layers.Dense(1, activation='sigmoid')
])

# 모델 컴파일
model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

# 모델 구조 확인
model.summary()

2. LSTM (Long Short-Term Memory)

# LSTM 모델
lstm_model = models.Sequential([
    layers.Embedding(100, 16, input_length=max_length),
    layers.LSTM(32, return_sequences=False),
    layers.Dense(16, activation='relu'),
    layers.Dense(1, activation='sigmoid')
])

lstm_model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

# 양방향 LSTM 모델
bidirectional_lstm_model = models.Sequential([
    layers.Embedding(100, 16, input_length=max_length),
    layers.Bidirectional(layers.LSTM(32, return_sequences=False)),
    layers.Dense(16, activation='relu'),
    layers.Dense(1, activation='sigmoid')
])

bidirectional_lstm_model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

# 다층 LSTM 모델
stacked_lstm_model = models.Sequential([
    layers.Embedding(100, 16, input_length=max_length),
    layers.LSTM(32, return_sequences=True),
    layers.LSTM(16, return_sequences=False),
    layers.Dense(16, activation='relu'),
    layers.Dense(1, activation='sigmoid')
])

stacked_lstm_model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

3. GRU (Gated Recurrent Unit)

# GRU 모델
gru_model = models.Sequential([
    layers.Embedding(100, 16, input_length=max_length),
    layers.GRU(32, return_sequences=False),
    layers.Dense(16, activation='relu'),
    layers.Dense(1, activation='sigmoid')
])

gru_model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

# 양방향 GRU 모델
bidirectional_gru_model = models.Sequential([
    layers.Embedding(100, 16, input_length=max_length),
    layers.Bidirectional(layers.GRU(32, return_sequences=False)),
    layers.Dense(16, activation='relu'),
    layers.Dense(1, activation='sigmoid')
])

bidirectional_gru_model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

RNN 고급 기법

1. 시퀀스-투-시퀀스 (Seq2Seq)

# 시퀀스-투-시퀀스 모델 구현
class Seq2SeqModel(models.Model):
    def __init__(self, vocab_size, embedding_dim, hidden_units):
        super(Seq2SeqModel, self).__init__()
        
        # 인코더
        self.encoder_embedding = layers.Embedding(vocab_size, embedding_dim)
        self.encoder_lstm = layers.LSTM(hidden_units, return_state=True)
        
        # 디코더
        self.decoder_embedding = layers.Embedding(vocab_size, embedding_dim)
        self.decoder_lstm = layers.LSTM(hidden_units, return_sequences=True, return_state=True)
        self.decoder_dense = layers.Dense(vocab_size, activation='softmax')
    
    def call(self, inputs):
        encoder_input, decoder_input = inputs
        
        # 인코더
        encoder_emb = self.encoder_embedding(encoder_input)
        encoder_output, state_h, state_c = self.encoder_lstm(encoder_emb)
        
        # 디코더
        decoder_emb = self.decoder_embedding(decoder_input)
        decoder_output, _, _ = self.decoder_lstm(decoder_emb, initial_state=[state_h, state_c])
        
        # 출력
        output = self.decoder_dense(decoder_output)
        return output

# Seq2Seq 모델 생성
seq2seq_model = Seq2SeqModel(vocab_size=100, embedding_dim=16, hidden_units=32)
seq2seq_model.compile(
    optimizer='adam',
    loss='sparse_categorical_crossentropy',
    metrics=['accuracy']
)

2. 어텐션 메커니즘

# 어텐션 메커니즘 구현
class AttentionLayer(layers.Layer):
    def __init__(self, units):
        super(AttentionLayer, self).__init__()
        self.W1 = layers.Dense(units)
        self.W2 = layers.Dense(units)
        self.V = layers.Dense(1)
    
    def call(self, query, values):
        # 어텐션 스코어 계산
        score = self.V(tf.nn.tanh(self.W1(query) + self.W2(values)))
        
        # 어텐션 가중치 계산
        attention_weights = tf.nn.softmax(score, axis=1)
        
        # 컨텍스트 벡터 계산
        context_vector = attention_weights * values
        context_vector = tf.reduce_sum(context_vector, axis=1)
        
        return context_vector, attention_weights

# 어텐션을 포함한 모델
class AttentionModel(models.Model):
    def __init__(self, vocab_size, embedding_dim, hidden_units):
        super(AttentionModel, self).__init__()
        
        self.embedding = layers.Embedding(vocab_size, embedding_dim)
        self.lstm = layers.LSTM(hidden_units, return_sequences=True)
        self.attention = AttentionLayer(hidden_units)
        self.dense = layers.Dense(1, activation='sigmoid')
    
    def call(self, inputs):
        x = self.embedding(inputs)
        x = self.lstm(x)
        
        # 마지막 히든 스테이트를 쿼리로 사용
        query = x[:, -1, :]
        context_vector, attention_weights = self.attention(query, x)
        
        output = self.dense(context_vector)
        return output

# 어텐션 모델 생성
attention_model = AttentionModel(vocab_size=100, embedding_dim=16, hidden_units=32)
attention_model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

3. 트랜스포머 기반 모델

# 트랜스포머 기반 모델 구현
class TransformerBlock(layers.Layer):
    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1):
        super(TransformerBlock, self).__init__()
        self.att = layers.MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        self.ffn = models.Sequential([
            layers.Dense(ff_dim, activation="relu"),
            layers.Dense(embed_dim),
        ])
        self.layernorm1 = layers.LayerNormalization(epsilon=1e-6)
        self.layernorm2 = layers.LayerNormalization(epsilon=1e-6)
        self.dropout1 = layers.Dropout(rate)
        self.dropout2 = layers.Dropout(rate)
    
    def call(self, inputs, training):
        attn_output = self.att(inputs, inputs)
        attn_output = self.dropout1(attn_output, training=training)
        out1 = self.layernorm1(inputs + attn_output)
        ffn_output = self.ffn(out1)
        ffn_output = self.dropout2(ffn_output, training=training)
        return self.layernorm2(out1 + ffn_output)

# 트랜스포머 모델
class TransformerModel(models.Model):
    def __init__(self, vocab_size, embed_dim, num_heads, ff_dim, num_blocks):
        super(TransformerModel, self).__init__()
        
        self.embedding = layers.Embedding(vocab_size, embed_dim)
        self.pos_encoding = layers.Embedding(max_length, embed_dim)
        self.transformer_blocks = [TransformerBlock(embed_dim, num_heads, ff_dim) for _ in range(num_blocks)]
        self.global_pool = layers.GlobalAveragePooling1D()
        self.dense = layers.Dense(1, activation='sigmoid')
    
    def call(self, inputs):
        x = self.embedding(inputs)
        positions = tf.range(start=0, limit=tf.shape(inputs)[1], delta=1)
        positions = self.pos_encoding(positions)
        x = x + positions
        
        for transformer_block in self.transformer_blocks:
            x = transformer_block(x)
        
        x = self.global_pool(x)
        return self.dense(x)

# 트랜스포머 모델 생성
transformer_model = TransformerModel(
    vocab_size=100, 
    embed_dim=32, 
    num_heads=4, 
    ff_dim=64, 
    num_blocks=2
)
transformer_model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy']
)

RNN 실무 적용 예시

1. 텍스트 분류

# 텍스트 분류 모델 학습
def train_text_classifier():
    # 더 큰 데이터셋 생성
    texts_large = [
        "I love this movie", "This movie is great", "I hate this film", "This film is terrible",
        "The acting is amazing", "The plot is boring", "Great cinematography", "Poor direction",
        "Excellent performance", "Disappointing ending", "Wonderful story", "Bad script",
        "Outstanding visuals", "Terrible editing", "Brilliant acting", "Awful dialogue",
        "Fantastic direction", "Horrible soundtrack", "Perfect casting", "Disappointing plot"
    ]
    
    labels_large = [1, 1, 0, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0]
    
    # 텍스트 토크나이징
    tokenizer = Tokenizer(num_words=100, oov_token="<OOV>")
    tokenizer.fit_on_texts(texts_large)
    sequences = tokenizer.texts_to_sequences(texts_large)
    
    # 시퀀스 패딩
    max_length = 10
    padded_sequences = pad_sequences(sequences, maxlen=max_length, padding='post')
    
    # 데이터 분할
    X_train, X_test, y_train, y_test = train_test_split(
        padded_sequences, labels_large, test_size=0.2, random_state=42
    )
    
    # LSTM 모델
    model = models.Sequential([
        layers.Embedding(100, 16, input_length=max_length),
        layers.LSTM(32, return_sequences=False),
        layers.Dropout(0.5),
        layers.Dense(16, activation='relu'),
        layers.Dense(1, activation='sigmoid')
    ])
    
    model.compile(
        optimizer='adam',
        loss='binary_crossentropy',
        metrics=['accuracy']
    )
    
    # 모델 학습
    history = model.fit(
        X_train, y_train,
        epochs=50,
        validation_data=(X_test, y_test),
        verbose=1
    )
    
    return model, history, tokenizer

# 텍스트 분류 모델 학습
text_model, text_history, text_tokenizer = train_text_classifier()

# 성능 평가
test_loss, test_accuracy = text_model.evaluate(X_test, y_test, verbose=0)
print(f"테스트 정확도: {test_accuracy:.4f}")

# 예측
def predict_sentiment(text, model, tokenizer):
    """텍스트 감정 예측"""
    sequence = tokenizer.texts_to_sequences([text])
    padded_sequence = pad_sequences(sequence, maxlen=10, padding='post')
    prediction = model.predict(padded_sequence)
    return "긍정" if prediction[0][0] > 0.5 else "부정"

# 예측 테스트
test_texts = ["I love this movie", "This film is terrible", "Great acting"]
for text in test_texts:
    sentiment = predict_sentiment(text, text_model, text_tokenizer)
    print(f"'{text}' -> {sentiment}")

2. 텍스트 생성

# 텍스트 생성 모델
def create_text_generation_model():
    # 샘플 텍스트 데이터
    text_data = """
    The quick brown fox jumps over the lazy dog.
    A journey of a thousand miles begins with a single step.
    The early bird catches the worm.
    Practice makes perfect.
    Knowledge is power.
    Time is money.
    Actions speak louder than words.
    Better late than never.
    """
    
    # 텍스트 전처리
    text_data = text_data.lower()
    chars = sorted(list(set(text_data)))
    char_to_idx = {char: idx for idx, char in enumerate(chars)}
    idx_to_char = {idx: char for char, idx in char_to_idx.items()}
    
    # 시퀀스 생성
    sequence_length = 20
    sequences = []
    next_chars = []
    
    for i in range(0, len(text_data) - sequence_length):
        sequences.append(text_data[i:i + sequence_length])
        next_chars.append(text_data[i + sequence_length])
    
    # 원-핫 인코딩
    X = np.zeros((len(sequences), sequence_length, len(chars)), dtype=bool)
    y = np.zeros((len(sequences), len(chars)), dtype=bool)
    
    for i, sequence in enumerate(sequences):
        for t, char in enumerate(sequence):
            X[i, t, char_to_idx[char]] = 1
        y[i, char_to_idx[next_chars[i]]] = 1
    
    # 텍스트 생성 모델
    model = models.Sequential([
        layers.LSTM(128, return_sequences=True, input_shape=(sequence_length, len(chars))),
        layers.Dropout(0.2),
        layers.LSTM(128, return_sequences=False),
        layers.Dropout(0.2),
        layers.Dense(len(chars), activation='softmax')
    ])
    
    model.compile(
        optimizer='adam',
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )
    
    return model, char_to_idx, idx_to_char, X, y

# 텍스트 생성 모델 생성
gen_model, char_to_idx, idx_to_char, X_gen, y_gen = create_text_generation_model()

# 모델 학습
gen_model.fit(X_gen, y_gen, epochs=100, verbose=0)

# 텍스트 생성 함수
def generate_text(model, char_to_idx, idx_to_char, seed_text, length=100):
    """텍스트 생성"""
    generated_text = seed_text
    
    for _ in range(length):
        # 시퀀스 준비
        sequence = np.zeros((1, 20, len(char_to_idx)))
        for i, char in enumerate(seed_text[-20:]):
            if char in char_to_idx:
                sequence[0, i, char_to_idx[char]] = 1
        
        # 다음 문자 예측
        prediction = model.predict(sequence, verbose=0)
        next_char_idx = np.argmax(prediction)
        next_char = idx_to_char[next_char_idx]
        
        generated_text += next_char
        seed_text = seed_text[1:] + next_char
    
    return generated_text

# 텍스트 생성 테스트
seed = "the quick brown fox"
generated = generate_text(gen_model, char_to_idx, idx_to_char, seed, 50)
print(f"생성된 텍스트: {generated}")

3. 시계열 예측

# 시계열 예측 모델
def create_time_series_model():
    # 시계열 데이터 생성
    time_steps = 100
    x = np.linspace(0, 4*np.pi, time_steps)
    y = np.sin(x) + 0.1 * np.random.randn(time_steps)
    
    # 시퀀스 데이터 생성
    sequence_length = 10
    X = []
    y_target = []
    
    for i in range(len(y) - sequence_length):
        X.append(y[i:i + sequence_length])
        y_target.append(y[i + sequence_length])
    
    X = np.array(X)
    y_target = np.array(y_target)
    
    # 데이터 분할
    split = int(0.8 * len(X))
    X_train, X_test = X[:split], X[split:]
    y_train, y_test = y_target[:split], y_target[split:]
    
    # 시계열 예측 모델
    model = models.Sequential([
        layers.LSTM(50, return_sequences=True, input_shape=(sequence_length, 1)),
        layers.LSTM(50, return_sequences=False),
        layers.Dense(25),
        layers.Dense(1)
    ])
    
    model.compile(
        optimizer='adam',
        loss='mse',
        metrics=['mae']
    )
    
    return model, X_train, X_test, y_train, y_test

# 시계열 모델 생성
ts_model, X_train_ts, X_test_ts, y_train_ts, y_test_ts = create_time_series_model()

# 모델 학습
ts_model.fit(
    X_train_ts, y_train_ts,
    epochs=50,
    validation_data=(X_test_ts, y_test_ts),
    verbose=0
)

# 예측
predictions = ts_model.predict(X_test_ts)

# 시각화
plt.figure(figsize=(15, 5))

# 훈련 데이터
plt.subplot(1, 3, 1)
plt.plot(y_train_ts, label='실제 값')
plt.title('훈련 데이터')
plt.legend()

# 테스트 데이터
plt.subplot(1, 3, 2)
plt.plot(y_test_ts, label='실제 값')
plt.plot(predictions, label='예측 값')
plt.title('테스트 데이터')
plt.legend()

# 예측 vs 실제
plt.subplot(1, 3, 3)
plt.scatter(y_test_ts, predictions, alpha=0.7)
plt.plot([y_test_ts.min(), y_test_ts.max()], [y_test_ts.min(), y_test_ts.max()], 'r--', lw=2)
plt.xlabel('실제 값')
plt.ylabel('예측 값')
plt.title('예측 vs 실제')

plt.tight_layout()
plt.show()

# 성능 평가
from sklearn.metrics import mean_squared_error, r2_score

mse = mean_squared_error(y_test_ts, predictions)
r2 = r2_score(y_test_ts, predictions)

print(f"평균제곱오차: {mse:.4f}")
print(f"결정계수: {r2:.4f}")

RNN 성능 최적화

1. 하이퍼파라미터 튜닝

import kerastuner as kt

def build_rnn_model(hp):
    """하이퍼파라미터 튜닝을 위한 RNN 모델"""
    model = models.Sequential()
    
    # 임베딩 레이어
    model.add(layers.Embedding(
        input_dim=100,
        output_dim=hp.Int('embedding_dim', min_value=16, max_value=64, step=16),
        input_length=10
    ))
    
    # RNN 레이어
    rnn_type = hp.Choice('rnn_type', values=['lstm', 'gru', 'simple_rnn'])
    rnn_units = hp.Int('rnn_units', min_value=16, max_value=128, step=16)
    
    if rnn_type == 'lstm':
        model.add(layers.LSTM(rnn_units, return_sequences=False))
    elif rnn_type == 'gru':
        model.add(layers.GRU(rnn_units, return_sequences=False))
    else:
        model.add(layers.SimpleRNN(rnn_units, return_sequences=False))
    
    # 드롭아웃
    model.add(layers.Dropout(
        rate=hp.Float('dropout_rate', min_value=0.2, max_value=0.5, step=0.1)
    ))
    
    # 완전 연결 레이어
    model.add(layers.Dense(
        units=hp.Int('dense_units', min_value=16, max_value=64, step=16),
        activation='relu'
    ))
    model.add(layers.Dense(1, activation='sigmoid'))
    
    # 컴파일
    model.compile(
        optimizer=hp.Choice('optimizer', values=['adam', 'rmsprop']),
        loss='binary_crossentropy',
        metrics=['accuracy']
    )
    
    return model

# 하이퍼파라미터 튜너 생성
tuner = kt.Hyperband(
    build_rnn_model,
    objective='val_accuracy',
    max_epochs=20,
    factor=3,
    directory='rnn_tuning',
    project_name='rnn_hyperparameter_tuning'
)

# 하이퍼파라미터 탐색
tuner.search(
    X_train, y_train,
    epochs=20,
    validation_data=(X_test, y_test),
    callbacks=[tf.keras.callbacks.EarlyStopping(patience=5)]
)

# 최적 모델 가져오기
best_model = tuner.get_best_models(num_models=1)[0]
best_hyperparameters = tuner.get_best_hyperparameters(num_trials=1)[0]

print("최적 하이퍼파라미터:")
print(best_hyperparameters.values)

2. 모델 앙상블

# RNN 모델 앙상블
def create_rnn_ensemble():
    """앙상블을 위한 여러 RNN 모델 생성"""
    models_list = []
    
    # 모델 1: LSTM
    model1 = models.Sequential([
        layers.Embedding(100, 16, input_length=10),
        layers.LSTM(32, return_sequences=False),
        layers.Dense(16, activation='relu'),
        layers.Dense(1, activation='sigmoid')
    ])
    models_list.append(model1)
    
    # 모델 2: GRU
    model2 = models.Sequential([
        layers.Embedding(100, 16, input_length=10),
        layers.GRU(32, return_sequences=False),
        layers.Dense(16, activation='relu'),
        layers.Dense(1, activation='sigmoid')
    ])
    models_list.append(model2)
    
    # 모델 3: 양방향 LSTM
    model3 = models.Sequential([
        layers.Embedding(100, 16, input_length=10),
        layers.Bidirectional(layers.LSTM(32, return_sequences=False)),
        layers.Dense(16, activation='relu'),
        layers.Dense(1, activation='sigmoid')
    ])
    models_list.append(model3)
    
    return models_list

# 앙상블 모델들 생성
ensemble_models = create_rnn_ensemble()

# 각 모델 컴파일 및 학습
for i, model in enumerate(ensemble_models):
    model.compile(
        optimizer='adam',
        loss='binary_crossentropy',
        metrics=['accuracy']
    )
    
    print(f"모델 {i+1} 학습 중...")
    model.fit(
        X_train, y_train,
        epochs=20,
        validation_data=(X_test, y_test),
        verbose=0
    )

# 앙상블 예측
def ensemble_predict(models, X):
    """앙상블 모델들의 예측 결과 평균"""
    predictions = []
    for model in models:
        pred = model.predict(X)
        predictions.append(pred)
    
    # 예측 결과 평균
    ensemble_pred = np.mean(predictions, axis=0)
    return ensemble_pred

# 앙상블 예측
ensemble_predictions = ensemble_predict(ensemble_models, X_test)
ensemble_predicted_classes = (ensemble_predictions > 0.5).astype(int)

# 앙상블 성능 평가
ensemble_accuracy = np.mean(ensemble_predicted_classes.flatten() == y_test)
print(f"앙상블 정확도: {ensemble_accuracy:.4f}")

# 개별 모델 성능과 비교
individual_accuracies = []
for model in ensemble_models:
    _, accuracy = model.evaluate(X_test, y_test, verbose=0)
    individual_accuracies.append(accuracy)

print("개별 모델 정확도:")
for i, acc in enumerate(individual_accuracies):
    print(f"모델 {i+1}: {acc:.4f}")
print(f"앙상블 정확도: {ensemble_accuracy:.4f}")

주의사항 및 모범 사례

1. 모델 설계

• 레이어 수: 과적합을 방지하기 위해 적절한 레이어 수 선택
• 히든 유닛: 데이터 복잡도에 맞는 히든 유닛 수 설정
• 드롭아웃: 과적합 방지를 위한 드롭아웃 적용

2. 학습 최적화

• 그래디언트 클리핑: 그래디언트 폭발 방지
• 배치 정규화: 학습 안정성 향상
• 조기 종료: 과적합 방지

3. 성능 모니터링

• 검증 데이터: 검증 데이터를 통한 성능 모니터링
• 콜백: 조기 종료와 학습률 감소 활용
• 시각화: 학습 곡선과 예측 결과 시각화

마무리

RNN은 시퀀스 데이터 처리에 특화된 딥러닝 모델로, 자연어 처리, 시계열 예측, 텍스트 생성 등 다양한 분야에서 활용됩니다. LSTM, GRU 등의 변형 모델을 통해 장기 의존성 문제를 해결하고, 어텐션 메커니즘과 트랜스포머를 통해 더욱 향상된 성능을 달성할 수 있습니다. 적절한 모델 설계와 학습 최적화를 통해 실무에서 요구되는 높은 성능의 시퀀스 모델을 구축할 수 있습니다.