PyTorch와 체육관을 이용한 근접 정책 최적화

1. 환경 설정하기 

PPO를 구현하기 전에 사전 필수 소프트웨어 라이브러리를 설치하고 정책을 적용할 적절한 환경을 선택해야 합니다. 

PyTorch와 필요한 라이브러리 설치하기

다음 소프트웨어를 설치해야 합니다: 

• PyTorch 및 수학 및 통계 함수에 사용되는 numpy와 그래프를 그리는 데 사용되는 matplotlib과 같은 소프트웨어 라이브러리.
• OpenAI의 오픈 소스 Gym 소프트웨어 패키지는 다양한 환경과 게임을 시뮬레이션하는 파이썬 라이브러리로, 강화 학습을 사용하여 해결할 수 있습니다. Gym API를 사용하여 알고리즘을 환경과 상호 작용하도록 만들 수 있습니다. 가끔 gym의 기능이 업그레이드 과정을 통해 변경되기도 하므로, 이 예시에서는 버전을 0.25.2로 고정합니다.

서버 또는 로컬 머신에 설치하려면 다음을 실행하십시오:

$ pip install torch numpy matplotlib gym==0.25.2

노트북을 사용하여 Google Colab이나 DataLab을 사용하여 설치하려면:

!pip install torch numpy matplotlib gym==0.25.2

CartPole 환경 생성하기

OpenAI Gym을 사용하여 CartPole 환경의 두 인스턴스(학습용 및 테스트용)를 생성합니다.

env_train = gym.make('CartPole-v1')
env_test = gym.make('CartPole-v1')

2. PyTorch에서 PPO 구현하기

이제 PyTorch를 사용하여 PPO를 구현해 봅시다.

정책 네트워크 정의하기

이전에 설명한대로, PPO는 액터-크리틱 모델로 구현됩니다. 액터는 정책을 구현하고 크리틱은 예상 값에 대한 예측을 합니다. 액터와 크리틱 신경망은 각 타임스텝의 상태를 동일하게 입력으로 취합니다. 따라서 액터와 크리틱 모델은 공통 신경망인 백본 아키텍처를 공유할 수 있습니다. 액터와 크리틱은 백본 아키텍처를 추가 레이어로 확장할 수 있습니다.

백본 네트워크 정의하기

다음 단계는 백본 네트워크를 설명합니다.

• 입력, 은닉, 출력 레이어가 있는 3개 레이어로 구성된 네트워크를 구현합니다.
• 입력 및 은닉 레이어 이후에는 활성화 함수를 사용합니다. 본 자습서에서는 계산 효율성을 위해 ReLU를 선택합니다.
• 우리는 강건한 네트워크를 얻기 위해 입력 및 은닉 레이어 뒤에 드롭아웃 함수를 적용합니다. 드롭아웃 함수는 일부 뉴런을 무작위로 0으로 만듭니다. 이는 특정 뉴런에 의존을 줄이고, 과적합을 방지하여 네트워크를 더 강건하게 만듭니다.

아래 코드는 주요 구현을 시연합니다:

class BackboneNetwork(nn.Module):
    def __init__(self, in_features, hidden_dimensions, out_features, dropout):
        super().__init__()
        self.layer1 = nn.Linear(in_features, hidden_dimensions)
        self.layer2 = nn.Linear(hidden_dimensions, hidden_dimensions)
        self.layer3 = nn.Linear(hidden_dimensions, out_features)
        self.dropout = nn.Dropout(dropout)
    def forward(self, x):
        x = self.layer1(x)
        x = f.relu(x)
        x = self.dropout(x)
        x = self.layer2(x)
        x = f.relu(x)
        x = self.dropout(x)
        x = self.layer3(x)
        return x

배우-평론가 네트워크를 정의하세요

이제 우리는 배우-평론가 클래스인 ActorCritic를 정의하는 데 이 네트워크를 사용할 수 있습니다. 배우는 정책을 모델링하고 행동을 예측합니다. 평론가는 가치 함수를 모델링하고 가치를 예측합니다. 둘 다 상태를 입력으로 사용합니다.

class ActorCritic(nn.Module):
    def __init__(self, actor, critic):
        super().__init__()
        self.actor = actor
        self.critic = critic
    def forward(self, state):
        action_pred = self.actor(state)
        value_pred = self.critic(state)
        return action_pred, value_pred

배우와 평론가 네트워크를 생성하세요

위에서 정의한 네트워크를 사용하여 배우와 평론가를 생성할 것입니다. 그런 다음, 배우와 평론가를 포함한 에이전트를 만들 것입니다.

에이전트를 만들기 전에 네트워크의 매개변수를 초기화하세요:

• 은닉 레이어의 차원, H는 구성 가능한 매개변수입니다. 숨겨진 레이어의 크기와 개수는 문제의 복잡성에 따라 다릅니다. 우리는 64 x 64 차원을 갖는 숨겨진 레이어를 사용할 것입니다.
• 입력 특성, N은 상태 배열의 크기입니다. 입력 레이어는 N x H 차원을 갖습니다. CartPole 환경에서 상태는 4 요소 배열입니다. 따라서 N은 4입니다.
• 배우 네트워크의 출력 특성, O는 환경에서의 행동 수입니다. 배우의 출력 레이어는 H x O 차원을 갖습니다. CartPole 환경에서는 2개의 행동이 있습니다.
• 평론가 네트워크의 출력 특성. 평론가 네트워크는 입력 상태에 대한 예상 값만을 예측하므로 출력 특성 수는 1입니다.
• 드롭아웃 비율.

다음 코드는 백본 네트워크를 기반으로 액터와 비평가 네트워크를 선언하는 방법을 보여줍니다:

def create_agent(hidden_dimensions, dropout):
    INPUT_FEATURES = env_train.observation_space.shape[0]
    HIDDEN_DIMENSIONS = hidden_dimensions
    ACTOR_OUTPUT_FEATURES = env_train.action_space.n
    CRITIC_OUTPUT_FEATURES = 1
    DROPOUT = dropout
    actor = BackboneNetwork(
            INPUT_FEATURES, HIDDEN_DIMENSIONS, ACTOR_OUTPUT_FEATURES, DROPOUT)
    critic = BackboneNetwork(
            INPUT_FEATURES, HIDDEN_DIMENSIONS, CRITIC_OUTPUT_FEATURES, DROPOUT)
    agent = ActorCritic(actor, critic)
    return agent

수익 계산하기

환경은 에이전트의 행동에 따라 각 단계에서 다음 단계로 넘어갈 때 보상을 제공합니다. 보상 R은 다음과 같이 표현됩니다:

수익은 예상되는 미래 보상의 누적 값을 정의합니다. 미래의 멀리 있는 타임스텝에서의 보상은 즉각적인 보상보다 가치가 낮습니다. 따라서 수익은 일반적으로 다음과 같이 정의되는 할인된 수익 G로 계산됩니다:

이 튜토리얼(및 다른 많은 참고 문헌)에서 수익은 할인된 수익을 의미합니다.

수익을 계산하려면:

• 모든 미래 상태에서 예상되는 보상으로 시작합니다.
• 각 미래 보상에 할인 인자의 지수로 곱합니다. 예를 들어, 2 타임스텝 후(현재로부터)의 예상 보상은 2로 곱해집니다.
• 모든 할인된 미래 보상을 합산하여 수익을 계산합니다.
• 수익 값을 정규화합니다.

함수 calculate_returns()는 아래와 같이 이러한 계산을 수행합니다:

def calculate_returns(rewards, discount_factor):
    returns = []
    cumulative_reward = 0
    for r in reversed(rewards):
        cumulative_reward = r + cumulative_reward * discount_factor
        returns.insert(0, cumulative_reward)
    returns = torch.tensor(returns)
    # 수익 정규화
    returns = (returns - returns.mean()) / returns.std()
    return returns

장점 함수 구현하기

장점은 비평가가 예측한 가치와 정책에 따라 액터가 선택한 행동의 예상 수익 간의 차이로 계산됩니다. 주어진 행동에 대해 장점은 특정 행동을 취하는 것의 이점을 임의의(평균) 행동과 비교하여 표현합니다.

원래 PPO 논문 (식 10)에서는 T 타임스탬프까지의 이점을 다음과 같이 표현합니다:

알고리즘을 코딩하는 동안, 설정된 타임스탬프 수까지 미리 보도록 하는 제약이 배치 크기를 통해 적용됩니다. 따라서 위의 식은 가치와 예상 수익의 차이로 단순화할 수 있습니다. 예상 수익은 상태-행동 가치 함수 Q에서 정량화됩니다.

따라서 아래의 단순화된 공식은 다음을 선택하는 이점을 표현합니다:

• 특정 행동
• 을 주어진 상태
• 에서 특정 정책
• 에 따라 특정 타임스탬프

에서 이 표현은 다음과 같습니다:

OpenAI는 또한 이 공식을 사용하여 RL을 구현합니다. calculate_advantages() 아래에 나타난 함수는 이점을 계산합니다:

def calculate_advantages(returns, values):
    advantages = returns - values
    # 이점을 정규화합니다
    advantages = (advantages - advantages.mean()) / advantages.std()
    return advantages

서로게이트 손실 및 클리핑 메커니즘

정책 손실은 PPO와 같은 특별한 기술 없이 표준 정책 그래디언트 손실이 될 것입니다. 표준 정책 그래디언트 손실은 다음의 곱으로 계산됩니다:

• 정책 행동 확률
• 이익 함수, 이익 함수는 다음의 차이로 계산됩니다:
• 정책 반환
• 기대값

표준 정책 그래디언트 손실은 급격한 정책 변경에 대한 보정을 할 수 없습니다. 대리 손실은 정책이 각 반복에서 변경될 수 있는 양을 제한하기 위해 표준 손실을 수정합니다. 이는 두 양 중 작은 값입니다:

• 제품:
• 정책 비율. 이 비율은 이전 행동 확률과 새로운 행동 확률의 차이를 표현합니다.
• 장점 함수
• 제품:
• 정책 비율의 클램프 값. 이 비율은 업데이트된 정책이 이전 정책의 일정 비율 이내에 있도록 클램프됩니다.
• 장점 함수

최적화 프로세스에서 대리 손실은 실제 손실의 대리로 사용됩니다.

클리핑 메커니즘

정책 비율, R, 새로운 정책과 이전 정책의 차이이며 새로운 매개변수 하에서 정책의 로그 확률과 이전 매개변수 하에서 정책의 로그 확률의 비율로 주어집니다:

클리핑된 정책 비율, R’, 다음과 같이 제약을 받습니다:

주어진 이점At은 이전 섹션에서 보여진 대로, 그리고 위에서 보여준 정책 비율에 따라, 대리 손실은 다음과 같이 계산됩니다:

아래 코드는 클리핑 메커니즘과 대리 손실을 구현하는 방법을 보여줍니다.

def calculate_surrogate_loss(
        actions_log_probability_old,
        actions_log_probability_new,
        epsilon,
        advantages):
    advantages = advantages.detach()
    policy_ratio = (
            actions_log_probability_new - actions_log_probability_old
            ).exp()
    surrogate_loss_1 = policy_ratio * advantages
    surrogate_loss_2 = torch.clamp(
            policy_ratio, min=1.0-epsilon, max=1.0+epsilon
            ) * advantages
    surrogate_loss = torch.min(surrogate_loss_1, surrogate_loss_2)
    return surrogate_loss

3. 에이전트 훈련

이제 에이전트를 훈련해 봅시다.

정책 및 가치 손실 계산

이제 정책과 가치 손실을 계산할 준비가 되었습니다:

• 정책 손실은 대리 손실과 엔트로피 보너스의 합입니다.
• 가치 손실은 비평가가 예측한 가치와 정책에 의해 생성된 수익(누적 보상) 간의 차이를 기반으로 합니다. 가치 손실 계산은 스무스 L1 손실 함수를 사용합니다. 이것은 손실 함수를 부드럽게 하고 이상치에 덜 민감하게 만듭니다.

위에서 계산된 두 손실 모두 텐서입니다. 경량하강법은 스칼라 값에 기반합니다. 손실을 나타내는 단일 스칼라 값을 얻으려면 .sum() 함수를 사용하여 텐서 요소를 합칩니다. 아래 함수에서 이를 수행하는 방법을 보여줍니다:

def calculate_losses(
        surrogate_loss, entropy, entropy_coefficient, returns, value_pred):
    entropy_bonus = entropy_coefficient * entropy
    policy_loss = -(surrogate_loss + entropy_bonus).sum()
    value_loss = f.smooth_l1_loss(returns, value_pred).sum()
    return policy_loss, value_loss

훈련 루프 정의

훈련 프로세스를 시작하기 전에 빈 배열로 구성된 버퍼 세트를 생성합니다. 훈련 알고리즘은 이 버퍼를 사용하여 에이전트의 행동, 환경의 상태 및 각 시간 단계에서의 보상을 저장합니다. 아래 함수는 이러한 버퍼를 초기화합니다:

def init_training():
    states = []
    actions = []
    actions_log_probability = []
    values = []
    rewards = []
    done = False
    episode_reward = 0
    return states, actions, actions_log_probability, values, rewards, done, episode_reward

각 훈련 반복에서 에이전트는 해당 반복의 정책 매개변수로 실행됩니다. 에이전트는 종료 조건에 도달할 때까지 루프에서 시간 단계에 따라 환경과 상호 작용합니다.

각 시간 단계 후, 에이전트의 행동, 보상 및 값이 각각의 버퍼에 추가됩니다. 에피소드가 종료되면 함수는 에피소드 결과를 요약한 업데이트된 버퍼 세트를 반환합니다.

훈련 루프를 실행하기 전에:

• 모델을 훈련 모드로 설정하려면 agent.train()을 사용하세요.
• 환경을 임의의 상태로 재설정하려면 env.reset()을 사용하세요. 이는 이번 훈련 반복의 시작 상태입니다.

다음 단계에서는 훈련 루프의 각 타임스텝에서 발생하는 일을 설명합니다:

• 상태를 에이전트에 전달합니다.
• 에이전트가 반환합니다:
• 정책(액터)을 기반으로 주어진 상태에 대한 예측된 행동. 이 예측된 행동 텐서를 소프트맥스 함수에 통과시켜 행동 확률 집합을 얻습니다.
• 비평가를 기반으로 한 상태의 예측 값.
• 에이전트가 취할 행동을 선택합니다:
• 행동 확률을 사용하여 확률 분포를 추정합니다.
• 이 분포에서 샘플을 뽑아 무작위로 행동을 선택합니다. dist.sample() 함수가 이를 수행합니다.
• env.step() 함수를 사용하여 이 행동을 환경에 전달하여 이 시간 단계에 대한 환경의 반응을 시뮬레이션합니다. 에이전트의 행동에 따라 환경은 다음을 생성합니다:
• 새로운 상태
• 보상
• 부울 반환 값 done (환경이 종단 상태에 도달했는지를 나타냄)
• 에이전트의 행동, 보상, 예측값 및 새 상태의 값을 해당 버퍼에 추가합니다.

env.step() 함수가 done의 부울 반환 값에 대해 true를 반환할 때 교육 에피소드가 종료됩니다.

에피소드가 종료된 후 각 타임스텝의 누적 값 사용하여 해당 에피소드의 누적 수익을 계산하고 각 타임스텝의 보상을 더하여 계산합니다. 이를 수행하기 위해 앞서 설명한 calculate_returns() 함수를 사용합니다. 이 함수의 입력은 할인 계수 및 각 타임스텝에서의 보상이 포함된 버퍼입니다. 이러한 반환 값 및 각 타임스텝의 누적 값을 사용하여 calculate_advantages() 함수를 사용하여 이득을 계산합니다.

다음 파이썬 함수는 이러한 단계를 구현하는 방법을 보여줍니다:

def forward_pass(env, agent, optimizer, discount_factor):
    states, actions, actions_log_probability, values, rewards, done, episode_reward = init_training()
    state = env.reset()
    agent.train()
    while not done:
        state = torch.FloatTensor(state).unsqueeze(0)
        states.append(state)
        action_pred, value_pred = agent(state)
        action_prob = f.softmax(action_pred, dim=-1)
        dist = distributions.Categorical(action_prob)
        action = dist.sample()
        log_prob_action = dist.log_prob(action)
        state, reward, done, _ = env.step(action.item())
        actions.append(action)
        actions_log_probability.append(log_prob_action)
        values.append(value_pred)
        rewards.append(reward)
        episode_reward += reward
    states = torch.cat(states)
    actions = torch.cat(actions)
    actions_log_probability = torch.cat(actions_log_probability)
    values = torch.cat(values).squeeze(-1)
    returns = calculate_returns(rewards, discount_factor)
    advantages = calculate_advantages(returns, values)
    return episode_reward, states, actions, actions_log_probability, advantages, returns

모델 매개변수 업데이트

각 훈련 반복은 많은 타임스텝(종료 조건에 도달할 때까지)으로 구성된 완전한 에피소드를 모델을 실행합니다. 각 타임스텝에서 정책 매개변수, 에이전트의 행동, 반환 값 및 이점을 저장합니다. 각 반복 후, 해당 반복의 모든 타임스텝을 통해 정책의 성능에 따라 모델을 업데이트합니다.

CartPole 환경에서의 최대 타임스텝 수는 500입니다. 보다 복잡한 환경에서는 더 많은 타임스텝이 필요할 수 있으며, 수백만 개가 될 수도 있습니다. 이러한 경우에는 훈련 결과 데이터셋을 배치로 분할해야 합니다. 각 배치의 타임스텝 수를 최적화 배치 크기라고 합니다.

따라서 모델 매개변수를 업데이트하는 단계는 다음과 같습니다:

• 훈련 결과 데이터셋을 배치로 분할합니다.
• 각 배치에 대해:
• 에이전트의 행동과 각 상태에 대한 예측된 값 가져오기.
• 이러한 예측된 행동을 사용하여 새로운 행동 확률 분포를 추정합니다.
• 이 분포를 사용하여 엔트로피를 계산하십시오.
• 이 분포를 사용하여 학습 결과 데이터 집합의 동작의 로그 확률을 얻으십시오. 이것은 학습 결과 데이터 집합의 동일한 동작들의 이전 로그 확률 세트입니다. 이러한 동작들의 이전 로그 확률은 이전 섹션에서 설명된 학습 루프에서 계산되었습니다.
• 동작의 이전 및 새로운 확률 분포를 사용하여 대리 손실을 계산하십시오.
• 대리 손실, 엔트로피 및 장점을 사용하여 정책 손실 및 가치 손실을 계산하십시오.
• 정책 및 가치 손실에 별도로 .backward()를 실행하십시오. 이렇게 하면 손실 함수의 그래디언트가 업데이트됩니다.
• 옵티마이저에 .step()을 실행하여 정책 매개변수를 업데이트합니다. 이 경우, 우리는 속도와 견고성을 균형있게 유지하기 위해 Adam 옵티마이저를 사용합니다.
• 정책과 가치 손실을 누적합니다.
• 각 배치에서 역전파 (위의 작업)를 PPO_STEPS 매개변수의 값에 따라 여러 번 반복합니다. 각 배치에서 역전파를 반복하는 것은 계산 효율적입니다. 왜냐하면 추가적인 순방향 패스를 실행할 필요 없이 훈련 데이터셋의 크기를 효과적으로 증가시킬 수 있기 때문입니다. 샘플링과 최적화 사이의 각 교대에서 환경 단계의 수를 반복 배치 크기라고 합니다.
• 평균 정책 손실과 가치 손실을 반환합니다.

아래 코드는 이러한 단계를 구현합니다:

def update_policy(
        agent,
        states,
        actions,
        actions_log_probability_old,
        advantages,
        returns,
        optimizer,
        ppo_steps,
        epsilon,
        entropy_coefficient):
    BATCH_SIZE = 128
    total_policy_loss = 0
    total_value_loss = 0
    actions_log_probability_old = actions_log_probability_old.detach()
    actions = actions.detach()
    training_results_dataset = TensorDataset(
            states,
            actions,
            actions_log_probability_old,
            advantages,
            returns)
    batch_dataset = DataLoader(
            training_results_dataset,
            batch_size=BATCH_SIZE,
            shuffle=False)
    for _ in range(ppo_steps):
        for batch_idx, (states, actions, actions_log_probability_old, advantages, returns) in enumerate(batch_dataset):
            # 모든 입력 상태에 대한 동작의 새로운 로그 확률 가져오기
            action_pred, value_pred = agent(states)
            value_pred = value_pred.squeeze(-1)
            action_prob = f.softmax(action_pred, dim=-1)
            probability_distribution_new = distributions.Categorical(
                    action_prob)
            entropy = probability_distribution_new.entropy()
            # 이전 동작을 사용하여 새로운 로그 확률 추정
            actions_log_probability_new = probability_distribution_new.log_prob(actions)
            surrogate_loss = calculate_surrogate_loss(
                    actions_log_probability_old,
                    actions_log_probability_new,
                    epsilon,
                    advantages)
            policy_loss, value_loss = calculate_losses(
                    surrogate_loss,
                    entropy,
                    entropy_coefficient,
                    returns,
                    value_pred)
            optimizer.zero_grad()
            policy_loss.backward()
            value_loss.backward()
            optimizer.step()
            total_policy_loss += policy_loss.item()
            total_value_loss += value_loss.item()
    return total_policy_loss / ppo_steps, total_value_loss / ppo_steps

4. PPO 에이전트 실행하기

PPO 에이전트를 마칠 때입니다.

성능 평가하기

에이전트의 성능을 평가하려면 새로운 환경을 만들고 이 새 환경에서 에이전트를 실행하여 누적 보상을 계산해야 합니다. .eval() 함수를 사용하여 에이전트를 평가 모드로 설정해야 합니다. 이는 훈련 루프와 동일합니다. 아래 코드 스니펫은 평가 함수를 구현한 것입니다:

def evaluate(env, agent):
    agent.eval()
    rewards = []
    done = False
    episode_reward = 0
    state = env.reset()
    while not done:
        state = torch.FloatTensor(state).unsqueeze(0)
        with torch.no_grad():
            action_pred, _ = agent(state)
            action_prob = f.softmax(action_pred, dim=-1)
        action = torch.argmax(action_prob, dim=-1)
        state, reward, done, _ = env.step(action.item())
        episode_reward += reward
    return episode_reward

훈련 결과 시각화하기

훈련 과정의 진행 상황을 시각화하기 위해 Matplotlib 라이브러리를 사용할 것입니다. 아래 함수는 훈련 및 테스트 루프에서의 보상을 플롯하는 방법을 보여줍니다:

def plot_train_rewards(train_rewards, reward_threshold):
    plt.figure(figsize=(12, 8))
    plt.plot(train_rewards, label='Training Reward')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Training Reward', fontsize=20)
    plt.hlines(reward_threshold, 0, len(train_rewards), color='y')
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

def plot_test_rewards(test_rewards, reward_threshold):
    plt.figure(figsize=(12, 8))
    plt.plot(test_rewards, label='Testing Reward')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Testing Reward', fontsize=20)
    plt.hlines(reward_threshold, 0, len(test_rewards), color='y')
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

아래 예시 플롯에서는 각각 훈련 및 테스트 환경에서 정책을 적용하여 얻은 훈련 및 테스트 보상을 보여줍니다. 이 플롯의 형태는 코드를 실행할 때마다 다를 수 있습니다. 이는 훈련 과정에 내재된 무작위성 때문입니다.

훈련 보상 (훈련 환경에서 정책을 적용하여 얻음). 저자 제공 이미지.

테스트 보상 (테스트 환경에서 정책을 적용하여 얻음). 저자 제공 이미지.

위의 출력 그래프에서 훈련 과정의 진행을 관찰하십시오:

• 보상은 낮은 값부터 시작합니다. 훈련이 진행됨에 따라 보상이 증가합니다.
• 보상은 증가하면서 무작위로 변동합니다. 에이전트가 정책 공간을 탐험하기 때문입니다.
• 훈련이 종료되고, 테스트 보상은 많은 반복에서 임계값(475) 주변에 안정화되었습니다.
• 보상은 500으로 제한됩니다. 이는 환경(Gym CartPole v1)에서 부과된 제약입니다.

마찬가지로 반복을 통해 값과 정책 손실을 그래프로 나타낼 수 있습니다:

def plot_losses(policy_losses, value_losses):
    plt.figure(figsize=(12, 8))
    plt.plot(value_losses, label='Value Losses')
    plt.plot(policy_losses, label='Policy Losses')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Loss', fontsize=20)
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

아래 예시 그래프는 훈련 에피소드를 통해 추적된 손실을 보여줍니다:

훈련 과정을 통해 발생하는 가치와 정책 손실. 저자 이미지

그래프를 관찰하고 주목하세요:

• 손실은 무작위로 분포되어 있으며 어떠한 패턴도 따르지 않습니다.
• 이는 보상을 극대화하는 것이 목표인 RL 훈련의 전형적인 특징입니다.

PPO 알고리즘을 실행하세요

PPO를 사용하여 에이전트를 훈련시키는 모든 구성 요소를 갖추셨습니다. 이를 모두 결합하려면 다음을 해야 합니다:

• 할인 요소, 배치 크기, 학습률 등과 같은 하이퍼파라미터 선언
• 각 반복에서 보상과 손실을 저장할 널 배열로 버퍼를 인스턴스화합니다.
• create_agent() 함수를 사용하여 에이전트 인스턴스를 생성합니다.
• forward_pass() 및 update_policy() 함수를 사용하여 순방향 및 역방향 패스를 반복적으로 실행합니다.
• evaluate() 함수를 사용하여 정책의 성능을 테스트합니다.
• 훈련 및 평가 함수에서 정책, 가치 손실 및 보상을 각각의 버퍼에 추가합니다.
• 지난 몇 개의 타임 스텝 동안의 보상과 손실의 평균을 계산합니다. 아래 예시는 지난 40개의 타임 스텝에 대한 보상과 손실을 평균합니다.
• 평가 결과를 일정한 간격으로 출력합니다. 아래 예시는 매 10단계마다 출력합니다.
• 평균 보상이 특정 임계값을 넘을 때 프로세스를 종료합니다.

아래 코드는 Python에서 이 작업을 수행하는 함수를 선언하는 방법을 보여줍니다.

def run_ppo():
    MAX_EPISODES = 500
    DISCOUNT_FACTOR = 0.99
    REWARD_THRESHOLD = 475
    PRINT_INTERVAL = 10
    PPO_STEPS = 8
    N_TRIALS = 100
    EPSILON = 0.2
    ENTROPY_COEFFICIENT = 0.01
    HIDDEN_DIMENSIONS = 64
    DROPOUT = 0.2
    LEARNING_RATE = 0.001
    train_rewards = []
    test_rewards = []
    policy_losses = []
    value_losses = []
    agent = create_agent(HIDDEN_DIMENSIONS, DROPOUT)
    optimizer = optim.Adam(agent.parameters(), lr=LEARNING_RATE)
    for episode in range(1, MAX_EPISODES+1):
        train_reward, states, actions, actions_log_probability, advantages, returns = forward_pass(
                env_train,
                agent,
                optimizer,
                DISCOUNT_FACTOR)
        policy_loss, value_loss = update_policy(
                agent,
                states,
                actions,
                actions_log_probability,
                advantages,
                returns,
                optimizer,
                PPO_STEPS,
                EPSILON,
                ENTROPY_COEFFICIENT)
        test_reward = evaluate(env_test, agent)
        policy_losses.append(policy_loss)
        value_losses.append(value_loss)
        train_rewards.append(train_reward)
        test_rewards.append(test_reward)
        mean_train_rewards = np.mean(train_rewards[-N_TRIALS:])
        mean_test_rewards = np.mean(test_rewards[-N_TRIALS:])
        mean_abs_policy_loss = np.mean(np.abs(policy_losses[-N_TRIALS:]))
        mean_abs_value_loss = np.mean(np.abs(value_losses[-N_TRIALS:]))
        if episode % PRINT_INTERVAL == 0:
            print(f'Episode: {episode:3} | \
                  Mean Train Rewards: {mean_train_rewards:3.1f} \
                  | Mean Test Rewards: {mean_test_rewards:3.1f} \
                  | Mean Abs Policy Loss: {mean_abs_policy_loss:2.2f} \
                  | Mean Abs Value Loss: {mean_abs_value_loss:2.2f}')
        if mean_test_rewards >= REWARD_THRESHOLD:
            print(f'Reached reward threshold in {episode} episodes')
            break
    plot_train_rewards(train_rewards, REWARD_THRESHOLD)
    plot_test_rewards(test_rewards, REWARD_THRESHOLD)
    plot_losses(policy_losses, value_losses)

프로그램을 실행합니다:

run_ppo()

출력은 아래 예시와 유사해야 합니다:

Episode:  10 | Mean Train Rewards: 22.3 | Mean Test Rewards: 30.4 | Mean Abs Policy Loss: 0.37 | Mean Abs Value Loss: 0.39 
Episode:  20 | Mean Train Rewards: 38.6 | Mean Test Rewards: 69.8 | Mean Abs Policy Loss: 0.46 | Mean Abs Value Loss: 0.37 
.
.
.
Episode: 100 | Mean Train Rewards: 289.5 | Mean Test Rewards: 427.3 | Mean Abs Policy Loss: 1.73 | Mean Abs Value Loss: 0.21 
Episode: 110 | Mean Train Rewards: 357.7 | Mean Test Rewards: 461.4 | Mean Abs Policy Loss: 1.86 | Mean Abs Value Loss: 0.22 
Reached reward threshold in 116 episodes

당신은 이 DataLab 노트북에서 작업 중인 프로그램을 볼 수 있고 실행할 수 있습니다!

5. 하이퍼파라미터 튜닝 및 최적화

머신 러닝에서 하이퍼파라미터는 훈련 프로세스를 제어합니다. 아래에서 PPO에서 사용되는 중요한 하이퍼파라미터 중 일부를 설명합니다: 

• 학습률: 학습률은 각 반복에서 정책 매개변수가 얼마나 변할 수 있는지를 결정합니다. 확률적 경사 하강법에서 정책 매개변수가 각 반복에서 업데이트되는 양은 학습률과 기울기의 곱에 의해 결정됩니다. 
• 클리핑 매개변수: 이것은 엡실론(ε)으로도 불립니다. 정책 비율이 클리핑되는 정도를 결정합니다. 새로운 정책과 이전 정책의 비율은 범위 [1-ε, 1+ε] 내에서 변동할 수 있습니다. 이 범위를 벗어나면 인위적으로 해당 범위 내에 위치하도록 클리핑됩니다.
• 배치 크기: 이것은 각 기울기 업데이트에 고려해야 할 단계 수를 나타냅니다. PPO에서 배치 크기는 정책을 적용하고 정책 매개변수를 업데이트하기 위해 필요한 시간 단계의 수입니다. 이 기사에서는 배치 크기를 64로 사용했습니다. 
• 반복 단계: 이것은 각 배치를 역전파 실행에 재사용하는 횟수를 의미합니다. 이 기사의 코드에서는 이를 PPO_STEPS라고 참조합니다. 복잡한 환경에서 많은 횟수로 전방향 패스를 실행하는 것은 계산적으로 비용이 많이 듭니다. 더 효율적인 대안은 각 배치를 몇 번 다시 실행하는 것입니다. 일반적으로 5에서 10 사이의 값을 사용하는 것이 권장됩니다. 
• 할인 요소: 이것은 감마(gamma)로도 불립니다.즉시 보상이 미래의 보상보다 더 가치 있는 정도를 표현합니다. 이는 시간가치를 계산할 때 이자율 개념과 유사합니다. 가 0에 가까울수록 미래의 보상이 덜 가치 있고, 에이전트는 즉각적인 보상을 우선시해야 합니다. 1에 가까울수록 미래의 보상이 중요하다는 것을 의미합니다.
• 엔트로피 계수: 엔트로피 계수는 엔트로피 보너스를 결정하며, 이는 엔트로피 계수와 분포의 엔트로피의 곱으로 계산됩니다. 엔트로피 보너스의 역할은 정책에 더 많은 무작위성을 도입하는 것입니다. 이는 에이전트가 정책 공간을 탐색하도록 장려합니다. 그러나 이러한 무작위성이 너무 높을 때 훈련이 최적의 정책으로 수렴하지 않습니다.
• 훈련의 성공 기준: 훈련이 성공했는지 결정하는 기준을 설정해야 합니다. 이를 하는 일반적인 방법은 최근 N 번의 평균 보상이 특정 임계값을 초과하는 경우로 조건을 설정하는 것입니다. 위의 예시 코드에서는 N_TRIALS 변수로 표현됩니다. 이 값을 더 높게 설정할수록, 훈련에 더 많은 에피소드가 필요하여 훈련 시간이 더 오래 걸리게 됩니다. 또한 계산적으로 더 비싼 더 견고한 정책으로 이어집니다. PPO는 확률적 정책이므로 에이전트가 임계값을 넘지 못하는 에피소드가 발생할 수 있습니다. 따라서 N_TRIALS의 값이 너무 높으면 훈련이 종료되지 않을 수 있습니다.

PPO 성능 최적화 전략

훈련 PPO 알고리즘의 성능을 최적화하는 데에는 시행착오와 다양한 하이퍼파라미터 값을 실험하는 것이 필요합니다. 그러나 몇 가지 넓은 지침이 있습니다:

• 할인 계수: CartPole 환경과 같이 장기적 보상이 중요한 경우에는 폴이 오랫동안 안정적으로 유지되어야 하는 경우, 0.99와 같은 중간 gamma 값을 시작으로 하십시오.
• 엔트로피 보너스: 복잡한 환경에서는 에이전트가 최적 정책을 찾기 위해 행동 공간을 탐색해야 합니다. 엔트로피 보너스는 탐색을 촉진합니다. 엔트로피 보너스는 대리 손실에 추가됩니다. 엔트로피 계수를 결정하기 전에 대리 손실과 분포의 엔트로피의 크기를 확인하십시오. 이 글에서는 엔트로피 계수로 0.01을 사용했습니다.
• 클리핑 매개변수: 클리핑 매개변수는 업데이트된 정책이 현재 정책과 얼마나 다를 수 있는지를 결정합니다. 클리핑 매개변수의 값이 클수록 환경 탐색이 더 잘 이루어지지만 훈련이 불안정해질 위험이 있습니다. 불안정한 업데이트를 방지하면서 서서히 탐색을 허용하는 클리핑 매개변수가 이상적입니다. 이 문서에서는 클리핑 매개변수를 0.2로 사용했습니다.
• 학습률: 학습률이 너무 높으면 정책이 큰 단계로 업데이트되어 매 반복마다 훈련 과정이 불안정해질 수 있습니다. 너무 낮으면 훈련이 너무 오래 걸립니다. 이 튜토리얼에서는 환경에 잘 맞는 0.001의 학습률을 사용했습니다. 많은 경우에는 1e-5의 학습률을 사용하는 것이 권장됩니다.

PPO에서의 도전과 모범 사례

PPO의 개념과 구현 세부 사항을 설명한 후에는 도전 과제와 모범 사례에 대해 논의해 보겠습니다.

훈련 중 PPO에서 일반적으로 발생하는 도전 과제

PPO는 널리 사용되지만, 이 기술을 성공적으로 사용하여 실제 문제를 해결하기 위해 잠재적인 도전 과제를 인식해야 합니다. 이러한 도전 과제 중 일부는 다음과 같습니다:

• 수렴 속도 느림:복잡한 환경에서 PPO는 샘플 비효율적이며 최적 정책에 수렴하기 위해 환경과의 많은 상호 작용이 필요합니다. 이로 인해 훈련이 느리고 비용이 많이 듭니다.
• 하이퍼파라미터에 대한 민감도: PPO는 정책 공간을 효율적으로 탐색하는 데 의존합니다. 훈련 과정의 안정성과 수렴 속도는 하이퍼파라미터의 값에 민감합니다. 이러한 하이퍼파라미터의 최적 값은 종종 시행착오를 통해 결정될 수 있습니다.
• 과적합: RL 환경은 일반적으로 무작위 매개변수로 초기화됩니다. PPO 훈련은 에이전트 환경을 기반으로 최적의 정책을 찾는 데 기초합니다. 때로는 훈련 과정이 특정 환경에 대해 최적의 매개변수 집합으로 수렴하지만 임의의 환경에 대해서는 그렇지 않을 수 있습니다. 이는 일반적으로 많은 반복을 통해 해결됩니다. 각각은 서로 다른 무작위 훈련 환경을 가지고 있습니다.
• 동적 환경: 간단한 RL 환경인 CartPole 환경과 같은 환경은 정적인데, 규칙이 시간이 흐름에 따라 동일합니다. 로봇이 불안정한 움직이는 표면에서 걷기를 배우는 등 다른 많은 환경은 동적입니다. 이러한 환경에서 잘 수행하기 위해서는 PPO가 종종 추가적인 세밀한 조정이 필요합니다.
• 탐험 vs 이용: PPO의 클리핑 메커니즘은 정책 업데이트가 신뢰할 수 있는 영역 내에서 이루어지도록 보장합니다. 그러나 이는 또한 에이전트가 행동 공간을 탐험하는 것을 방해합니다. 이는 특히 복잡한 환경에서 지역 최적값으로 수렴할 수 있습니다. 반면, 에이전트가 너무 많이 탐험하도록 허용하는 것은 어떤 최적 정책에도 수렴하지 못하게 할 수 있습니다.

PPO 모델 학습을 위한 모범 사례

PPO를 사용하여 좋은 결과를 얻으려면 다음과 같은 몇 가지 모범 사례를 권장합니다:

• 입력 특성 정규화: 반환 및 이점 값의 정규화는 데이터의 변동성을 줄이고 안정적인 그래디언트 업데이트로 이어집니다. 데이터를 정규화하면 모든 값이 일관된 숫자 범위로 변환됩니다. 특잇값 및 극단적인 값의 영향을 줄이고, 그렇지 않으면 그래디언트 업데이트를 왜곡시키고 수렴 속도를 늦출 수 있습니다.
• 적절히 큰 배치 크기 사용: 작은 배치는 빠른 업데이트와 훈련을 가능하게 하지만 지역 최적해로 수렴하거나 훈련 과정에서 불안정성을 초래할 수 있습니다. 더 큰 배치 크기는 에이전트가 견고한 정책을 학습하게 도와 안정적인 훈련 과정을 이끌어냅니다. 그러나 너무 큰 배치 크기도 비효율적입니다. 계산 비용을 증가시키는 것 외에도, 큰 배치에서 추정된 평균에 기반한 경사 업데이트로 인해 정책 업데이트가 가치 함수에 미치는 영향이 덜 반응적일 수 있습니다. 게다가 해당 특정 배치에 오버피팅될 수 있습니다. 
• 반복 단계: 각 배치를 5-10번 반복해서 재사용하는 것이 일반적으로 바람직합니다. 이렇게 하면 훈련 과정이 더 효율적으로 이루어집니다. 동일한 배치를 너무 많이 재사용하면 오버피팅될 수 있습니다. 코드에서는 이 하이퍼파라미터를 PPO_STEPS라고 참조합니다. 
• 정기적인 평가 수행: 과적합을 감지하려면 정책의 효과를 주기적으로 모니터링하는 것이 중요합니다. 정책이 특정 시나리오에서 효과적이지 않은 경우, 추가 훈련이나 세부 조정이 필요할 수 있습니다.
• 하이퍼파라미터 튜닝: 앞에서 설명한 대로, PPO 훈련은 하이퍼파라미터 값에 민감합니다. 특정 문제에 대한 올바른 값 세트를 결정하기 위해 다양한 하이퍼파라미터 값을 실험해보세요.
• 공유 백본 네트워크: 이 기사에 설명된 대로, 공유 백본을 사용하면 배우 및 비평가 네트워크 간의 불균형을 방지할 수 있습니다. 배우와 비평가 간의 백본 네트워크를 공유하면 공통된 특징 추출과 환경에 대한 공통된 이해를 돕습니다. 이는 학습 프로세스를 더 효율적이고 안정적으로 만듭니다. 또한 알고리즘의 계산 공간과 시간 복잡성을 줄이는 데 도움이 됩니다.
• 은닉층의 수와 크기: 보다 복잡한 환경을 위해 은닉층의 수와 차원을 늘립니다. CartPole과 같은 간단한 문제는 단일 은닉층으로 해결할 수 있습니다. 이 기사에서 사용된 은닉층은 64차원을 갖습니다. 필요 이상으로 네트워크를 훨씬 크게 만드는 것은 계산상 낭비가 되고 불안정하게 만들 수 있습니다.
• 얼리 스톱핑: 평가 지표가 충족되면 훈련을 중지하여 과적합을 방지하고 자원 낭비를 피할 수 있습니다. 일반적인 평가 지표는 에이전트가 지난 N개의 이벤트에서 임계 보상을 초과했을 때입니다.

결론

이 기사에서는 RL 문제를 해결하는 방법으로 PPO를 논의했습니다. 그런 다음 PyTorch를 사용하여 PPO를 구현하는 단계를 상세히 설명했습니다. 마지막으로 PPO에 대한 성능 팁과 모범 사례를 제시했습니다.

가장 좋은 방법은 코드를 직접 구현하는 것입니다. 또한 코드를 수정하여 Gym의 다른 클래식 제어 환경과 작동하도록 할 수 있습니다. Python 및 OpenAI의 Gymnasium을 사용하여 RL 에이전트를 구현하는 방법을 알아보려면 파이썬으로 Gymnasium에서 강화 학습 코스를 따르세요!