<jupyter_start><jupyter_text>파이토치 기반 객체지향 방식의 정책경사 예제(Polocy Gradient Implemented by OOP using PyTorch)- Sungjin (<NAME>, 2019 정책 경사 (Polocy Gradient) 방법은 강화학습의 방법들 중에 정책 기반 방법 중의 기본적인 한 가지이다.    예제 10-5 파이토치 - Cartpole을 다루는 PG 전체 코드 1. 라이브러리 불러들이기<jupyter_code>%matplotlib inline
# Using ipython
from IPython.display import clear_output

import random
from itertools import count
import numpy as np
import matplotlib.pyplot as plt<jupyter_output><empty_output><jupyter_text>2. 강화학습 환경 관리 라이브러리 불러들이기<jupyter_code>import gym<jupyter_output><empty_output><jupyter_text>3. 파이토치 라이브러리 불러들이기<jupyter_code># Using torch
import torch as TC
import torch.nn as nn
import torch.nn.functional as F
from torch.autograd import Variable
from torch.distributions import Bernoulli<jupyter_output><empty_output><jupyter_text>II. 기반 기능 구현  강화학습에 필요한 공통 기능 구현<jupyter_code>def list_norm_inplace(buff):
    r_mean = np.mean(buff)
    r_std = np.std(buff)
    for ii in range(len(buff)):
        buff[ii] = (buff[ii] - r_mean) / r_std

        
def plot_durations(episode_durations):
    plt.figure(2)
    plt.clf()
    durations_t = TC.FloatTensor(episode_durations)
    plt.title('Training...')
    plt.xlabel('Episode')
    plt.ylabel('Duration')
    plt.plot(durations_t.numpy())
    # Take 100 episode averages and plot them too
    if len(durations_t) >= 100:
        means = durations_t.unfold(0, 100, 1).mean(1).view(-1)
        means = TC.cat((TC.zeros(99), means))
        plt.plot(means.numpy())
    plt.show()
    
        
def plot_durations_ii(ii, episode_durations, ee, ee_duration=100):
    episode_durations.append(ii+1)
    if (ee + 1) % ee_duration == 0:
        clear_output()
        plot_durations(episode_durations)<jupyter_output><empty_output><jupyter_text>정책경사를 위한 기반 기능 구현상태 정보($s_t$)를 입력해 행동에 대한 정책을 결정하는 요소인 정책 확률을 출력하는 함수형 클래스를 구현한다. $$p_t (a_t=1|s_t) = f_{\mathrm{PGNET}}(s_t)$$이 정책 확률은 Bernoulli 확률 분포함수를 나타내는 torch.distributions.Bernoulli라는 함수형 클래스에 의해 정책으로 변환된다.$$\pi_t (a_t=1|s_t) = P_{\mathrm{Bernoulli}}(p_t (a_t=1|s_t))$$<jupyter_code>class PGNET(nn.Module):
    # 순수하게 Policy gradient로 구성
    def __init__(self, num_state):
        super(PGNET, self).__init__()
    
        self.fc_in = nn.Linear(num_state, 24)
        self.fc_hidden = nn.Linear(24, 36)
        self.fc_out = nn.Linear(36, 1)
        
    def forward(self, x):
        x = F.relu(self.fc_in(x))
        x = F.relu(self.fc_hidden(x))
        x = TC.sigmoid(self.fc_out(x))
        return x<jupyter_output><empty_output><jupyter_text>III. 메인 코드  정책 경사 방법의 강화학습을 위한 수행 객체<jupyter_code>class PGNET_AGENT(PGNET):
    # Policy gradient를 실제 사용하기 위해 필요한 변환을 고려함
    def __init__(self, num_state, render_flag=False):
        self.forget_factor = 0.99
        self.learning_rate = 0.01
        self.num_episode = 5000            
        self.num_batch = 5
        self.render_flag = render_flag
        self.steps_in_batch = 0
        self.episode_durations = []
        
        super().__init__(num_state)                
        self.optimizer = TC.optim.RMSprop(self.parameters(), lr=self.learning_rate)        
        self.init_buff()        
        
    def forward(self, state):
        state_var = Variable(TC.from_numpy(state).float())
        prob = super().forward(state_var)
        # check this function to know what the meaing of log_prob. Is it log(prob)?
        return Bernoulli(prob)
    
    def push_buff_done(self, reward, state, action, done_flag=False):
        if done_flag: # devide between episodes
            self.reward_buff.append(0)
        else:
            self.reward_buff.append(reward)
        self.state_buff.append(state)
        self.action_buff.append(action)
        
    def pull_buff(self, ii):
        return self.reward_buff[ii], self.state_buff[ii], self.action_buff[ii]
    
    def init_buff(self):
        self.reward_buff = []
        self.state_buff = []
        self.action_buff = []
        
    def transform_discount_reward(self, steps):
        future_reward = 0
        for ii in reversed(range(steps)):
            if self.reward_buff[ii] == 0:
                future_reward = 0
            else:
                future_reward = future_reward * self.forget_factor + self.reward_buff[ii]
                self.reward_buff[ii] = future_reward
        list_norm_inplace(self.reward_buff)
    
    def train(self, steps):
        self.transform_discount_reward(steps)
        self.optimizer.zero_grad()
        for ii in range(steps):
            reward, state, action = self.pull_buff(ii)
            action_var = Variable(TC.FloatTensor([float(action)]))
            policy = self.forward(state)
            loss = -policy.log_prob(action_var) * reward            
            loss.backward()
        self.optimizer.step()        
        self.init_buff()

    def step(self, env, state, ee, ii, ee_duration=100):        
        policy = self.forward(state)
        action = policy.sample().data.numpy().astype(int)[0]

        next_state, reward, done_flag, _ = env.step(action)
        if self.render_flag: 
            env.render()        
        self.push_buff_done(reward, state, action, done_flag)

        self.steps_in_batch += 1
        state = next_state
        
        return state, done_flag
        
    def run_episode(self, env, ee):
        state = env.reset()        
        for ii in count():            
            state, done_flag = self.step(env, state, ee, ii, ee_duration=100)
            if done_flag:
                plot_durations_ii(ii, self.episode_durations, ee, ee_duration=100)
                break

    def train_episode(self, ee):
        if ee > 0 and ee % self.num_batch == 0:
            self.train(self.steps_in_batch)  
            self.steps_in_batch = 0            
    
    def run(self, env):
        for ee in range(self.num_episode):
            self.run_episode(env, ee)
            self.train_episode(ee)<jupyter_output><empty_output><jupyter_text>수행 객체 호출 함수 및 호출 명령 수행먼저 환경을 만든다. 환경은 GYM을 사용한다. GYM은 공개 소프트웨어로 강화학습을 위한 다양한 환경을 제공한다. 여기서는 CartPole이라는 환경을 사용한다. - GYM을 구할 수 있는 위치: https://gym.openai.com다음은 환경을 통해 자신을 학습시키는 에이전트의 인스턴스를 학습한다. 그리고 생성한 에이전트를 수행한다. 여기서의 수행은 인스턴스의 학습을 의미한다. 추후로는 학습을 하고 그 후에 태스트를 하는 것까지를 포함할 예정이다.최종적으로는 만든 환경을 닫는다.<jupyter_code>env = gym.make('CartPole-v0')                
a_agent = PGNET_AGENT(env.observation_space.shape[0], render_flag=False)
a_agent.run(env)        
env.close()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10432213930348258, 'fraction_numerical': 0.01181592039800995, 'mean_word_length': 2.5287986834887546, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 32, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '274028', 'license': 'MIT', 'n_tokens_mistral': 2711, 'n_tokens_neox': 2946, 'n_words': 498, 'path': 'RL/PyTorch_PG.ipynb', 'repo_name': 'kimwooglae/keraspp', 'size': '6436'}