PyTorch로 처음부터 PPO 코딩하기 (2/4 부)

4 부작 시리즈의 로드맵.

시리즈의 Part 2에 오신 것을 환영합니다. 여기서 PyTorch를 사용하여 처음부터 PPO (Proximal Policy Optimization) 코딩을 시작합니다. Part 1을 읽지 않았다면 먼저 읽으십시오 .

앞으로는 GitHub 요점 대신 코드 스크린 샷을 게시 할 것입니다. 코드를 복사하여 붙여 넣는 것을 원하지 않기 때문입니다 ( 기본 저장소 로 이동하면됩니다 ). 대신 다른 창에서 수동으로 코딩하는 동안이 자습서를 따르는 것이 좋습니다.

우리는 OpenAI의 Spinning Up 문서Actor-Critic Framework에 있는 의사 코드로 PPO-clip 변형을 따를 것 입니다. 다음은 의사 코드의 그림입니다.

OpenAI의 Spinning Up 문서에있는 PPO의 의사 코드.

초기 생각 : 단 8 단계? 좋은. 이것은 학습 알고리즘에 대한 의사 코드이므로 먼저 코드 흐름 방식을 설계하는 것이 현명 할 수 있습니다. 이 의사 코드는 하나의 함수에 모두 들어갈 수있는 것처럼 보입니다. 우리는 그것을 부를 것 learn입니다. 여러 단계에 대한 서브 루틴을 작성해야 할 것 같습니다 (예 : 3 단계에서는 기본적으로 시뮬레이션을 여러 개 롤아웃하기를 원합니다.이 경우 rollout나중에 같은 것을 정의 할 수 있습니다 ). 따라서 모든 것을 클래스로 캡슐화하는 것이 가장 좋습니다 PPO. 이런 식으로 환경에서 훈련하기 위해 먼저 PPO객체를 만든 다음 간단히 learn.

먼저 ppo.py 파일에 PPO 클래스를 설정해 보겠습니다 .

시원하고 뒷면을 두 드리십시오. 1 단계를 살펴 보겠습니다.

1 단계

여기에서 우리의 행위자와 비평가 네트워크를 초기화 할 것입니다. 즉, 신경망 모듈을 가져 오거나 직접 작성해야합니다. 후자를합시다. .NET 을 사용하여 신경망을 만드는 방법에 대한 PyTorch의 자습서 와 유사한 작업을 수행합니다 torch.nn. 매우 기본적인 피드 포워드 신경망을 만들 것입니다. 신경망에 익숙하지 않다면 이 시리즈를 시청 하세요 .

새 파일 network.py 에 신경망 모듈을 설정해 보겠습니다 .

import torch
from torch import nn
import torch.nn.functional as F
import numpy as np
class FeedForwardNN(nn.Module):
  def __init__(self):
    super(FeedForwardNN, self).__init__()

def __init__(self, in_dim, out_dim):
    super(FeedForwardNN, self).__init__()
self.layer1 = nn.Linear(in_dim, 64)
    self.layer2 = nn.Linear(64, 64)
    self.layer3 = nn.Linear(64, out_dim)

def forward(self, obs):
  # Convert observation to tensor if it's a numpy array
  if isinstance(obs, np.ndarray):
    obs = torch.tensor(obs, dtype=torch.float)
  
  activation1 = F.relu(self.layer1(obs))
  activation2 = F.relu(self.layer2(activation1))
  output = self.layer3(activation2)
return output
network.py 코드를 완성하십시오.

ppo.py로 돌아 가기 ; 이제 1 단계를 정말 쉽게 수행하고 초기 정책 또는 행위자, 매개 변수 및 가치 함수 또는 비평가 매개 변수를 정의 할 준비가되어 있어야합니다.

from network import FeedForwardNN
self.actor = FeedForwardNN(

def __init__(self, env):
  # Extract environment information
  self.env = env
  self.obs_dim = env.observation_space.shape[0]
  self.act_dim = env.action_space.shape[0]

# ALG STEP 1
# Initialize actor and critic networks
self.actor = FeedForwardNN(self.obs_dim, self.act_dim)
self.critic = FeedForwardNN(self.obs_dim, 1)

이제 2 단계로 넘어갑니다.

2 단계

쉬운. 그들은 우리가 for몇 번의 반복에 대해 배울 루프 를 정의하기를 원합니다 . 이제 반복으로 반복 할 수 있지만 Stable Baselines PPO2를 사용하면을 호출 할 때 총 학습 시간 단계를 지정할 수 있다는 것도 알고 learn있습니다. 그 디자인을 따라 갑시다. 이런 식으로 무한 반복을 계산하는 대신 중지하기 전에 훈련 할 시간 단계를 지정할 수 있습니다.

def learn(self, total_timesteps):
  t_so_far = 0 # Timesteps simulated so far
while t_so_far < total_timesteps:              # ALG STEP 2
    # Increment t_so_far somewhere below

3 단계 :

3 단계

첫 번째 미니 챌린지. 현재 배우 정책을 실행하여 일련의 에피소드에서 데이터를 수집해야합니다. 물론 rollout입니다. 우리는 각각 rollout에서 수집 된 데이터를 batch. 이제 어떤 데이터가 필요합니까? 의사 코드를 조금 살펴 보겠습니다.

OpenAI의 Spinning Up 문서에있는 PPO의 의사 코드.

6 단계와 7 단계에서 sₜ를 볼 때 타임 스텝 당 관찰이 필요한 것 같습니다. 또한 6 단계와 7 단계에서 aₜ가있는 타임 스텝 당 작업이 필요합니다. 6 단계에서 π_θ (aₜ | sₜ)가있는 작업 확률과 보상이 필요합니다. 4 단계와 7 아에 Rₜ와 -to-이동하고 증가하기 위하여 것을 잊지 마세요 t_so_farlearn, 우리는 배치 당 시뮬레이션 얼마나 많은 시간 단계 알아야합니다; 배치에서 실행 된 각 에피소드의 길이를 반환 해 보겠습니다 (나중에 평균 에피소드 길이를 로깅하는 데 사용할 수 있기 때문에 아직 합산하지 않습니다. 반환하기 전에 에피소드 길이를 합산 할 수도 있지만 실제로는 중요하지 않습니다).

또한 배치 당 실행할 시간 단계 수를 파악해야합니다. 나에게 하이퍼 파라미터처럼 들립니다. 먼저 _init_hyperparameters기본 하이퍼 파라미터를 정의 하는 함수 를 만들고 __init__.

def __init__(self, env):
  ...
  self._init_hyperparameters()
def _init_hyperparameters(self):
  # Default values for hyperparameters, will need to change later.
  self.timesteps_per_batch = 4800            # timesteps per batch
  self.max_timesteps_per_episode = 1600      # timesteps per episode

def rollout(self):
  # Batch data
  batch_obs = []             # batch observations
  batch_acts = []            # batch actions
  batch_log_probs = []       # log probs of each action
  batch_rews = []            # batch rewards
  batch_rtgs = []            # batch rewards-to-go
  batch_lens = []            # episodic lengths in batch
  • 관찰 : (배치 당 시간 단계 수, 관찰 차원)
  • 작업 : (배치 당 시간 단계 수, 작업 차원)
  • 로그 확률 : (배치 당 시간 단계 수)
  • 보상 : (에피소드 수, 에피소드 당 타임 스텝 수)
  • 이동 보상 : (배치 당 타임 스텝 수)
  • 배치 길이 : (에피소드 수)

obs = self.env.reset()
done = False
for ep_t in range(self.max_timesteps_per_episode):
  
  action = self.env.action_space.sample()
  obs, rew, done, _ = self.env.step(action)
if done:
    break

# Number of timesteps run so far this batch
t = 0
while t < self.timesteps_per_batch:
# Rewards this episode
  ep_rews = []
obs = self.env.reset()
  done = False
for ep_t in range(self.max_timesteps_per_episode):
    # Increment timesteps ran this batch so far
    t += 1
# Collect observation
    batch_obs.append(obs)
action, log_prob = self.get_action(obs)
    obs, rew, done, _ = self.env.step(action)
  
    # Collect reward, action, and log prob
    ep_rews.append(rew)
    batch_acts.append(action)
    batch_log_probs.append(log_prob)
if done:
      break
# Collect episodic length and rewards
  batch_lens.append(ep_t + 1) # plus 1 because timestep starts at 0
  batch_rews.append(ep_rews)

참고 : 작업은 테스트시 결정적입니다. 즉, "평균"작업은 테스트 중 실제 작업이됩니다. 그러나 훈련 중에는이 분포가 도움이 될 수있는 탐색 적 요소가 필요합니다.

from torch.distributions import MultivariateNormal
def __init(self, env):
  ...
  # Create our variable for the matrix.
  # Note that I chose 0.5 for stdev arbitrarily.
  self.cov_var = torch.full(size=(self.act_dim,), fill_value=0.5)
  
  # Create the covariance matrix
  self.cov_mat = torch.diag(self.cov_var)
def get_action(self, obs):
# Query the actor network for a mean action.
  # Same thing as calling self.actor.forward(obs)
  mean = self.actor(obs)
# Create our Multivariate Normal Distribution
  dist = MultivariateNormal(mean, self.cov_mat)
# Sample an action from the distribution and get its log prob
  action = dist.sample()
  log_prob = dist.log_prob(action)
  
  # Return the sampled action and the log prob of that action
  # Note that I'm calling detach() since the action and log_prob  
  # are tensors with computation graphs, so I want to get rid
  # of the graph and just convert the action to numpy array.
  # log prob as tensor is fine. Our computation graph will
  # start later down the line.
  return action.detach().numpy(), log_prob.detach()
4 단계

# Reshape data as tensors in the shape specified before returning
batch_obs = torch.tensor(batch_obs, dtype=torch.float)
batch_acts = torch.tensor(batch_acts, dtype=torch.float)
batch_log_probs = torch.tensor(batch_log_probs, dtype=torch.float)
# ALG STEP #4
batch_rtgs = self.compute_rtgs(batch_rews)
# Return the batch data
return batch_obs, batch_acts, batch_log_probs, batch_rtgs, batch_lens
Reward-to-go 공식.

여기서 G는 이동 보상 함수이고, sₖ는 타임 스텝 k에서의 관찰, T는 에피소드 당 타임 스텝, γ는 할인 계수, R (sᵢ)은 일부 관찰 sᵢ가 주어진 보상입니다.

여러 에피소드를 제외하고는 정확히 동일한 워크 플로를 적용 할 것입니다 (일관된 순서를 유지하기 위해 에피소드도 뒤로 반복해야 함).

def compute_rtgs(self, batch_rews):
  # The rewards-to-go (rtg) per episode per batch to return.
  # The shape will be (num timesteps per episode)
  batch_rtgs = []
# Iterate through each episode backwards to maintain same order
  # in batch_rtgs
  for ep_rews in reversed(batch_rews):
discounted_reward = 0 # The discounted reward so far
for rew in reversed(ep_rews):
      discounted_reward = rew + discounted_reward * self.gamma
      batch_rtgs.insert(0, discounted_reward)
# Convert the rewards-to-go into a tensor
  batch_rtgs = torch.tensor(batch_rtgs, dtype=torch.float)
return batch_rtgs
def _init_hyperparameters(self):
  ...
  self.gamma = 0.95

def learn(self, total_timesteps):
  ...
  while t_so_far < total_timesteps:
    # ALG STEP 3
    batch_obs, batch_acts, batch_log_probs, batch_rtgs, batch_lens = self.rollout()
__init__, 배우다
롤아웃
get_action, compute_rtgs, _init_hyperparameters

축하합니다! 우리는 이미 베어 본 PPO 구현을 반쯤 완료했으며 대부분의 코드를 완료했습니다. 3 부에서는 PPO 구현을 완료합니다.

이 시점까지 질문이 있으시면 주저하지 말고 의견을 남기거나 eyyu@ucsd.edu로 저에게 연락하십시오. 그렇지 않으면 Part 3 에서 만나요 !

Suggested posts

기계 학습 워크 플로를 실행하여 데이터를 변환하고 txtai로 AI 기반 텍스트 인덱스를 구축합니다.

데이터 흐름에 NLP 변환 적용

기계 학습 워크 플로를 실행하여 데이터를 변환하고 txtai로 AI 기반 텍스트 인덱스를 구축합니다.

txtai는 기계 학습 워크 플로우를 실행하여 데이터를 변환하고 AI 기반 텍스트 인덱스를 구축하여 유사성 검색을 수행합니다. txtai는 텍스트 스 니펫, 문서, 오디오 및 이미지 인덱싱을 지원합니다.

EBM의 성능 및 설명 가능성

설명 가능한 부스팅 머신이 두 가지를 모두 얻는 방법

EBM의 성능 및 설명 가능성

그 이후로 효율적인 모델과 설명 가능한 모델 중에서 선택해야했습니다. 한편으로 로지스틱 회귀와 같은 단순한 모델은 해석 가능하지만 성능이 지연됩니다.

Related posts

"실용적인 프로그래머"의 5 가지 필수 사항

역대 베스트셀러 코딩 북의 요점

"실용적인 프로그래머"의 5 가지 필수 사항

Pragmatic Programmer는 1999 년에 처음 출판되었으며 이후 역대 최고의 프로그래밍 책으로 선정되었습니다. 저자 Andy Hunt와 David Thomas는 Agile Manifesto의 원저자 중 하나였으며 몇 가지 심각한 자격을 가지고 있습니다.

대규모 GraphQL 쿼리 공격으로부터 보호

공격자가 공개적으로 사용 가능한 GraphQL 인터페이스를 사용하여 사이트를 스크랩하거나 서비스 거부 공격을 실행하는 방법에 대해 알아보십시오. 이들은 4 가지 방법 중 하나로이를 수행 할 수 있습니다. 단일 대형 쿼리를 신중하게 구성하여 실행하고, 관련 데이터를 가져올 수있는 병렬 쿼리를 많이 작성하고, 일괄 요청을 사용하여 많은 쿼리를 연속적으로 실행하고, 마지막으로 많은 요청을 보냅니다.

기술 인터뷰의 사회적 구성 요소

코딩 문제는 스트레스가 많지만 스트레스에 대한 당신의 반응은 당신의 기술적 능력보다 더 크게 말합니다.

기술 인터뷰의 사회적 구성 요소

기술 업계의 직책을 위해 인터뷰 할 때 일반적으로 제안을 고려하기 전에 최소한 3 차례의 인터뷰를 거치게됩니다. 라운드는 일반적으로 다음과 같습니다. 그렇게 생각하면 잘못된 것입니다.

훌륭한 개발자의 3 가지 행동 특성

훌륭한 개발자의 3 가지 행동 특성

훌륭한 개발자를 만드는 비 기술적 인 것들 나는이 기사를 작성하는 것을 한동안 미루고 있습니다. 나는 그것을 작성할 자격이 있다고 생각하지 못했습니다. 오늘은 쓸 때라고 생각했습니다.