[딥러닝] 지도학습 후 강화학습 (DeepSeek 알고리즘)

DeepSeek 알고리즘을 보던 중 지도학습 후 강화학습을 했다는 것을 보고

DeepSeek에게 해당 코드를 만들어 달라고 함.

 

1) 지도학습

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, LSTM, Embedding
import psutil
import os

p = psutil.Process(os.getpid())  # 현재 프로세스 객체 얻기
p.cpu_affinity([0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15])  # CPU 0과 1에만 프로세스 실행 제한
# p.cpu_affinity([15,16,17,18])  # CPU 0과 1에만 프로세스 실행 제한

# 데이터 준비 (예: 간단한 텍스트 데이터)
texts = [
    "토끼와 거북이는 친구입니다",
    "토끼는 빠르고 거북이는 느립니다",
    "어느 날 토끼와 거북이는 경주를 하기로 했습니다",
    "토끼는 자신만만하게 출발했습니다",
    "거북이는 천천히 하지만 꾸준히 걸었습니다",
    "토끼는 거북이가 느린 것을 보고 나무 아래에서 잠들었습니다",
    "거북이는 꾸준히 걸어 결승선에 도착했습니다",
    "토끼는 늦게 깨어나 거북이가 이긴 것을 알았습니다",
    "토끼는 자신의 잘못을 깨달았습니다",
    "거북이는 꾸준함의 중요성을 일깨워 주었습니다",
    "토끼는 거북이를 비웃으며 달렸습니다",
    "거북이는 포기하지 않고 꾸준히 걸었습니다",
    "토끼는 자신의 실수를 후회했습니다",
    "거북이의 승리는 모두를 놀라게 했습니다",
    "이 이야기는 꾸준함의 중요성을 가르쳐 줍니다"
]
tokenizer = tf.keras.preprocessing.text.Tokenizer()
tokenizer.fit_on_texts(texts)
sequences = tokenizer.texts_to_sequences(texts)
vocab_size = len(tokenizer.word_index) + 1

# 입력과 출력 데이터 생성
X = []
y = []
for seq in sequences:
    for i in range(1, len(seq)):
        X.append(seq[:i])
        y.append(seq[i])
X = tf.keras.preprocessing.sequence.pad_sequences(X, maxlen=10, padding='pre')
y = tf.keras.utils.to_categorical(y, num_classes=vocab_size)

# 모델 정의
def build_model(vocab_size):
    model = Sequential([
        Embedding(input_dim=vocab_size, output_dim=10, input_length=10),
        LSTM(100, return_sequences=True),  # LSTM 층 추가
        tf.keras.layers.Dropout(0.2),  # 드롭아웃 추가
        LSTM(100, return_sequences=True),  # LSTM 층 추가
        tf.keras.layers.Dropout(0.2),  # 드롭아웃 추가
        LSTM(50),
        tf.keras.layers.Dropout(0.2),  # 드롭아웃 추가
        Dense(vocab_size, activation='softmax')
    ])
    model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
    return model

# 지도학습으로 모델 학습
model = build_model(vocab_size)

# 모델 컴파일
optimizer = tf.keras.optimizers.Adam(learning_rate=0.001)  # 학습률 조정
model.compile(optimizer=optimizer, loss='categorical_crossentropy', metrics=['accuracy'])

# 지도학습으로 모델 학습
model.fit(X, y, epochs=5000, batch_size=32)

 

학습결과

...
Epoch 1995/2000
2/2 [==============================] - 0s 11ms/step - loss: 0.3236 - accuracy: 0.8333
Epoch 1996/2000
2/2 [==============================] - 0s 11ms/step - loss: 0.3128 - accuracy: 0.9000
Epoch 1997/2000
2/2 [==============================] - 0s 11ms/step - loss: 0.3264 - accuracy: 0.8333
Epoch 1998/2000
2/2 [==============================] - 0s 11ms/step - loss: 0.3206 - accuracy: 0.8667
Epoch 1999/2000
2/2 [==============================] - 0s 10ms/step - loss: 0.3201 - accuracy: 0.8500
Epoch 2000/2000
2/2 [==============================] - 0s 11ms/step - loss: 0.3168 - accuracy: 0.8667

 

 

 

2) 강화학습

import matplotlib.pyplot as plt

# 모델 복사
def copy_model(model):
    """모델의 구조와 가중치를 복사하여 새로운 모델을 생성합니다."""
    new_model = build_model(vocab_size)  # 동일한 구조의 새 모델 생성
    new_model.set_weights(model.get_weights())  # 가중치 복사
    return new_model

# 강화학습용 모델 생성
rl_model = copy_model(model)

# 강화학습을 위한 PPO Trainer 클래스
class PPOTrainer:
    def __init__(self, model, reward_model, gamma=0.99, clip_ratio=0.2):
        self.model = model
        self.reward_model = reward_model
        self.gamma = gamma
        self.clip_ratio = clip_ratio
        self.episode_rewards = []  # 에피소드별 보상을 저장할 리스트
        self.episode_losses = []   # 에피소드별 손실을 저장할 리스트

    def get_action(self, state):
        state = np.array([state])
        probs = self.model.predict(state, verbose=0)[0]
        action = np.random.choice(len(probs), p=probs)
        return action, probs

    def train_step(self, states, actions, rewards, old_probs):
        states = np.array(states)
        actions = np.array(actions)
        rewards = np.array(rewards)
        old_probs = np.array(old_probs)

        with tf.GradientTape() as tape:
            new_probs = self.model(states, training=True)
            new_probs = tf.reduce_max(new_probs * tf.one_hot(actions, depth=vocab_size), axis=1)
            ratio = new_probs / old_probs
            clipped_ratio = tf.clip_by_value(ratio, 1 - self.clip_ratio, 1 + self.clip_ratio)
            loss = -tf.reduce_mean(tf.minimum(ratio * rewards, clipped_ratio * rewards))

        grads = tape.gradient(loss, self.model.trainable_variables)
        optimizer = tf.keras.optimizers.Adam()
        optimizer.apply_gradients(zip(grads, self.model.trainable_variables))
        return loss.numpy()

# 보상 모델 (간단한 예시)
def reward_model(state, action):
    # 여기서는 단순히 정답과 일치하면 높은 보상을 주는 것으로 가정
    correct_actions = [2, 3, 1]  # 예시 정답
    if action in correct_actions:
        return 1.0  # 정답일 때 보상
    else:
        return -0.1  # 오답일 때 패널티


# 강화학습 실행
ppo_trainer = PPOTrainer(rl_model, reward_model)

def R_fun(E):
    # 진행 상황 시각화를 위한 리스트
    episode_rewards_history = []
    episode_losses_history = []
    
    for episode in range(E):  # 50번의 에피소드
        state = [0] * 10  # 초기 상태 (패딩된 입력)
        states, actions, rewards, old_probs = [], [], [], []
    
        for step in range(10):  # 각 에피소드에서 10번의 스텝
            action, probs = ppo_trainer.get_action(state)
            
            epsilon = 0.1  # 탐험 비율
            if np.random.rand() < epsilon:
                action = np.random.choice(len(probs))  # 무작위 선택
            else:
                action = np.argmax(probs)  # 최적 선택
        
            reward = reward_model(state, action)
            states.append(state)
            actions.append(action)
            rewards.append(reward)
            old_probs.append(probs[action])
    
            # 다음 상태로 이동 (예시)
            state = state[1:] + [action]
    
        # 보상 할인 적용
        discounted_rewards = []
        cumulative_reward = 0
        for r in reversed(rewards):
            cumulative_reward = r + ppo_trainer.gamma * cumulative_reward
            discounted_rewards.insert(0, cumulative_reward)
        discounted_rewards = np.array(discounted_rewards)
    
        # PPO 학습 및 손실 계산
        loss = ppo_trainer.train_step(states, actions, discounted_rewards, old_probs)
    
        # 에피소드별 보상 및 손실 저장
        total_reward = sum(rewards)
        ppo_trainer.episode_rewards.append(total_reward)
        ppo_trainer.episode_losses.append(loss)
    
        # 진행 상황 출력
        print(f"Episode {episode + 1}, Total Reward: {total_reward}, Loss: {loss}")
    
        # 진행 상황 기록
        episode_rewards_history.append(total_reward)
        episode_losses_history.append(loss)
    
    # 학습 결과 시각화
    plt.figure(figsize=(12, 5))
    
    # 보상 그래프
    plt.subplot(1, 2, 1)
    plt.plot(episode_rewards_history, label='Episode Reward')
    plt.xlabel('Episode')
    plt.ylabel('Total Reward')
    plt.title('Episode Rewards')
    plt.legend()
    
    # 손실 그래프
    plt.subplot(1, 2, 2)
    plt.plot(episode_losses_history, label='Episode Loss', color='orange')
    plt.xlabel('Episode')
    plt.ylabel('Loss')
    plt.title('Episode Losses')
    plt.legend()
    
    plt.show()

R_fun(100)

 

학습결과

- 아래는 Episode 1000으로 되어 있으나 1000번을 5번 정도 총 5000번 정도 진행 함.

...
Episode 995, Total Reward: 6.699999999999999, Loss: -4.0389814376831055
Episode 996, Total Reward: 6.699999999999999, Loss: -4.038989543914795
Episode 997, Total Reward: 6.699999999999999, Loss: -4.03898286819458
Episode 998, Total Reward: 7.8, Loss: -4.458654880523682
Episode 999, Total Reward: 2.2999999999999994, Loss: -0.7512975335121155
Episode 1000, Total Reward: 5.6, Loss: -3.1054749488830566

 

 

3) 예측

# 강화학습까지 완료한 후 예측을 수행하는 함수
def generate_text(model, tokenizer, seed_text, max_length=10):
    """
    학습된 모델을 사용하여 텍스트를 생성합니다.
    
    :param model: 학습된 모델
    :param tokenizer: 텍스트를 토큰화하는 데 사용된 tokenizer
    :param seed_text: 초기 입력 텍스트
    :param max_length: 생성할 텍스트의 최대 길이
    :return: 생성된 텍스트
    """
    # 초기 입력 텍스트를 토큰화하고 패딩
    token_list = tokenizer.texts_to_sequences([seed_text])[0]
    token_list = tf.keras.preprocessing.sequence.pad_sequences([token_list], maxlen=max_length-1, padding='pre')
    
    generated_text = seed_text  # 생성된 텍스트를 저장할 변수

    for _ in range(max_length):  # 최대 길이까지 텍스트 생성
        # 다음 단어 예측
        predicted_probs = model.predict(token_list, verbose=0)[0]
        predicted_id = np.argmax(predicted_probs)  # 가장 높은 확률의 단어 선택
        output_word = tokenizer.index_word.get(predicted_id, "")  # 단어로 변환

        # 생성된 단어를 텍스트에 추가
        generated_text += " " + output_word

        # 다음 입력을 위해 토큰 리스트 업데이트
        token_list = np.append(token_list[0][1:], predicted_id)
        token_list = tf.keras.preprocessing.sequence.pad_sequences([token_list], maxlen=max_length-1, padding='pre')

    return generated_text

# 예측 실행
seed_text = "토끼와 거북이는"  # 초기 입력 텍스트
generated_text = generate_text(rl_model, tokenizer, seed_text, max_length=10)
print(f"Generated Text: {generated_text}")

 

예측결과

Generated Text: 토끼와 거북이는 꾸준히 모두를 중요성을 꾸준히 꾸준히 꾸준히 꾸준히 꾸준히 꾸준히 거북이는

 

다음 단어를 정확히 찾지는 못 함. 중요한 것은 지도학습 후 해당 모델을 강화학습으로 사용하는 구조를 파악 함.