TIL| 강화학습...

🌞 시작하는 글

코딩이 재미없다는 기분이 들면 그건 내가 잘못된게 아니라 일이 진짜 구린거라서다.
이것저것 하는 요즘은 몸은 죽을꺼 같지만 재밌다. (일 제외)

🎯 오늘의 TODO LIST

⚗️ 오늘의 실험

Reinforce

보상을 많이 받은 행동일 수록 더 학습되게 만드는 방식. 좋은 행동을 하면 보상을 주고 나쁜 행동을 하면 반대로...

총보상계산: G = r + gamma * G

할인율 gamma = 0.99 라고 하면,
마지막 보상: G3 = 1
그 전 보상: G2 = 1 + 0.99 1 = 1.99
그 전 보상은: G1 = 1 + 0.99 1.99 = 2.97

Policy Loss 계산

-loss = -log_prob Gt => -확률의 로그 보상
0.8인 확률에 대한 보상이 3일때
log(0.8)= -0.22
loss = -(-0.22) * 3 = 0.66
policy loss 를 계산할때 사용되는 보상은 미래 보상도 계산에 염두해 두어야 하기 때문에 총보상계산이 필요한것...

역전파:

신경망이 틀린 이유를 찾아서, 각 가중치(weight)를 얼마나 바꿔야 할지 계산하는 방법... 예를들어.. 아 이거 잘 안나오네.. 몇번 레이어까지 거슬러올라가서 다시 해보자... 라고 하는것.
마찬가지로 역전파를 하기 위해 policy loss가 필요함

# 랜덤 시드 설정 - 실험 재현성을 위해 필요
seed = 42  # 42는 관례적으로 많이 사용되는 시드값
torch.manual_seed(seed)  # PyTorch의 랜덤 시드
np.random.seed(seed)  # NumPy의 랜덤 시드


"""
            # - 상태(state)는 [카트위치, 카트속도, 막대각도, 막대각속도] 4차원 벡터
            # - 행동(action)은 [왼쪽(0), 오른쪽(1)] 2가지 중 하나
            # 예시: 상태 [-0.5, 0.2, 0.1, -0.3]이 입력되면
            # 정책 신경망은 [0.3, 0.7] 같은 두 행동의 확률을 출력
            # (0.3 = 왼쪽으로 움직일 확률, 0.7 = 오른쪽으로 움직일 확률)
"""

class PolicyNetwork(nn.Module): # nn.Module은 신경망 모델을 정의하기 위한 클래스, 파이토치에서 신경망을 만들때 많이 사용
    def __init__(self, input_dim, output_dim): # 입력 차원과 출력 차원을 받아서 신경망 모델을 만들어줌
        super(PolicyNetwork, self).__init__()
        self.fc1 = nn.Linear(input_dim, 128)  # input_dim 차원의 입력을 받아 128차원의 은닉층으로 변환
        self.fc2 = nn.Linear(128, output_dim) # 128차원의 은닉층을 output_dim 차원의 출력으로 변환

    def forward(self, x): # 실제 계산하는 함수
        x = F.relu(self.fc1(x))             # ReLU 활성화 함수 사용( "음수는 버리고 양수만 통과시키는 함수")
        x = F.softmax(self.fc2(x), dim=1)   # 확률 출력 (소프트맥스: 각각의 비율를 리턴)
        return x

# REINFORCE 알고리즘 구현 함수
# REINFORCE 알고리즘 구현 함수
def reinforce(device, env, policy, optimizer, num_episodes=100, gamma=0.99):
    rewards_per_episode = []  # 각 에피소드별 총 보상을 저장할 리스트 추후에 보상 리스트를 리턴하고 그래프 그리기 위함
    
    for episode in range(num_episodes):  # 지정된 에피소드 수만큼 반복
        saved_log_probs = []  # 각 스텝에서의 행동 로그 확률 저장
        rewards = []  # 각 스텝에서 받은 보상 저장
        state, _ = env.reset(seed=seed)  # 환경 초기화, 초기 상태 받기
        done = False  # 에피소드 종료 여부

        ###############################1) 에피소드 실행 파트################################
        while not done:
            state = torch.FloatTensor(state).unsqueeze(0).to(device)  # 상태를 PyTorch 텐서로 변환하고 배치 차원 추가
            """
            # 텐서라는건... 파이토치에서 사용하는 데이터 형식 텐서는 데이터 배열을 표현하는 파이토치의 기본 데이터 구조 
            # 예시) 1차원 텐서: [1, 2, 3] 2차원 텐서: [[1, 2, 3], [4, 5, 6]] 3차원 텐서: [[[1, 2, 3], [4, 5, 6]], [[7, 8, 9], [10, 11, 12]]]
            # unsqueeze 함수는 텐서의 차원을 증가시키는 함수 예를 들어 1차원 텐서 [1, 2, 3]을 unsqueeze(0)하면 2차원 텐서 [[1, 2, 3]]이 됨 
            """

            probs = policy(state)  # 정책 신경망으로 각 행동의 확률 계산
            # 정책 신경망은 상태를 입력으로 받아 각 행동의 확률을 출력하는 신경망
            # 예를 들어 상태가 [-0.5, 0.2, 0.1, -0.3]이면 각 행동의 확률이 [0.3, 0.7]일 것임 왼쪽으로 갈 확률 0.3 오른쪽으로 갈 확률 0.7
            m = torch.distributions.Categorical(probs)  # 확률 분포 생성

            action = m.sample()  # 확률 분포에서 행동 샘플링
            saved_log_probs.append(m.log_prob(action))  # 선택된 행동의 로그 확률 저장 => 이건 나중에 손실 확률 계산용으로 사용됨
            state, reward, truncated, terminated, _ = env.step(action.item())  # 환경에서 행동 실행
            done = np.logical_or(truncated, terminated)  # 에피소드 종료 조건 확인
            rewards.append(reward)  # 받은 보상 저장

        rewards_per_episode.append(sum(rewards))  # 에피소드의 총 보상 기록


        ############################2) 총보상계산파트################################

        # Discounted return 계산 - 미래 보상의 현재 가치 계산
        # 강화학습에서는 미래의 보상도 고려해야 하지만, 먼 미래의 보상은 현재 시점에서 불확실성이 크기 때문에
        # 할인율(gamma)를 적용하여 먼 미래의 보상을 할인하여 계산.
        discounted_returns = []
        G = 0  # G는 각 시점에서의 할인된 누적 보상
        for r in reversed(rewards):  # 마지막 스텝부터 역순으로 계산하는 이유는 미래의 보상부터 현재로 거슬러 올라가며 계산하기 위함
            G = r + gamma * G  # 현재 보상(r)과 이전에 계산된 미래 보상(G)에 할인율을 곱한 값을 더한다
            discounted_returns.insert(0, G)  # 계산된 할인 보상을 리스트의 맨 앞에 추가 (시간 순서대로 정렬하기 위함)
        # 계산된 할인 보상들을 PyTorch 텐서로 변환
        discounted_returns = torch.tensor(discounted_returns).to(device)

        # 학습 안정화를 위한 정규화
        # 보상값들의 스케일이 너무 크거나 작으면 학습이 불안정할 수 있어서 평균을 0, 표준편차를 1의 정규분포로 만든다
        # 1e-8을 더하는 것은 표준편차가 0이 되는 것을 방지하기 위함
        discounted_returns = (discounted_returns - discounted_returns.mean()) / (discounted_returns.std() + 1e-8)





        #################################3) POLICY LOSS 계산파트################################

        # REINFORCE 알고리즘의 핵심인 정책 경사(Policy Gradient) 손실 함수를 계산합니다
        policy_loss = []
        for log_prob, Gt in zip(saved_log_probs, discounted_returns):
            # -log(prob) * G_t: 정책의 로그 확률과 할인 보상을 곱한다
            policy_loss.append(-log_prob * Gt) # 음의 부호는 0~1 사이의 수로 만들기 좋으니까ㅎ
        # 모든 시간 스텝의 손실을 하나의 텐서로 결합하고 총합을 계산
        policy_loss = torch.stack(policy_loss).sum()

        #################################4) 역전파################################


        optimizer.zero_grad()  # 이전 스텝의 그래디언트를 초기화 (그래디언트 누적 방지)
        policy_loss.backward()  # 손실함수에 대한 그래디언트를 계산 (역전파)
        optimizer.step()  # 계산된 그래디언트를 사용하여 신경망의 가중치를 업데이트

        # 학습 진행상황 모니터링 (5 에피소드마다 현재 보상을 출력)
        if episode % 5 == 0:
            print(f'Episode {episode}, Reward: {sum(rewards)}')

    return rewards_per_episode

# 학습 환경 설정
device = "cuda" if torch.cuda.is_available() else "cpu"  # GPU 사용 가능시 GPU 사용, 아니면 CPU 사용
print(f"current device: {device}")
policy = PolicyNetwork(env.observation_space.shape[0], env.action_space.n).to(device)  # 정책 신경망 생성
optimizer = optim.Adam(policy.parameters(), lr=1e-3)  # Adam 옵티마이저 설정, 학습률 0.001

# 에이전트 훈련 실행
rewards = reinforce(device, env, policy, optimizer, num_episodes=1000)

# 학습 결과 시각화
plt.plot(rewards)  # 보상 그래프 그리기
plt.xlabel('Episode')  # x축 레이블
plt.ylabel('Total Reward')  # y축 레이블
plt.title('Training Progress')  # 그래프 제목
plt.show()  # 그래프 표시