강화학습_6_0810_DQN

 

DQN의 2가지 특징은!

1. 시행을 저장하고 있다가 랜덤하게 배치로 꺼내서 학습한다.

        무작위로 꺼내 먹으면 더 대표성을 띈다.

                아마도 몬테 카를로 근사랑 관련이 있어보인다.

 

2. 최적의 값이라고 믿는 함수를 일정 이터레이션 마다 기존 학습된 함수로 업데이트 해준다.

        이것이 왜 성능을 올리는지는 아직 이해하려고 노력중이다.

                이것을 네트워크 분리라고 한다.

 

실행이 되는 코드를 누가 짰고 이걸 주석을 달면서 이해하려고 노력했다.

이것저것 실험하느라 원본과 조금 다르다.

밑에 링크를 남겨놓았다.

 

이 영상을 보면서 공부했다. 이 사람의 책도 샀다.

난 강화학습 고수가 된다! 되어가는 중이다! 주식, 스마트 팩토리, 블로그에 적용할 수 있기를 고대해본다.

 

유투브 영상

 

코드 출처

import gym
import collections
import random
 
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
 
#Hyperparameters
learning_rate = 0.0005
gamma         = 0.98 #discounted reward rate
buffer_limit  = 50000 #replay 저장 창고
batch_size    = 32 #한번에 몇개씩 털것이냐
 
class ReplayBuffer():
    def __init__(self):
        self.buffer = collections.deque(maxlen=buffer_limit) #maxlen을 지정해주어서 값이상 저장시 FIFO
    
    def put(self, transition): #transition --> (s,a,r/100.0,s_prime, done_mask) --> 이걸 버퍼에 저장
        self.buffer.append(transition)
    
    def sample(self, n): # 버퍼 중에 n개를 무작위 추출을 하겠다
        mini_batch = random.sample(self.buffer, n) 
        s_lst, a_lst, r_lst, s_prime_lst, done_mask_lst = [], [], [], [], []
        
        for transition in mini_batch: #32개를 텐서화해서 뱉기 위해 리스트로 저장
            s, a, r, s_prime, done_mask = transition
            s_lst.append(s)
            a_lst.append([a])
            r_lst.append([r])
            s_prime_lst.append(s_prime)
            done_mask_lst.append([done_mask])
 
        return torch.tensor(s_lst, dtype=torch.float), torch.tensor(a_lst), \
               torch.tensor(r_lst), torch.tensor(s_prime_lst, dtype=torch.float), \
               torch.tensor(done_mask_lst)
    
    def size(self): #버퍼 길이 나중에 초반 중복 추출이 겹치는걸 방지하기 위해서
        return len(self.buffer)
 
class Qnet(nn.Module):
    def __init__(self):
        super(Qnet, self).__init__()
        self.fc1 = nn.Linear(4, 16)
        self.fc2 = nn.Linear(16, 16)
        self.fc3 = nn.Linear(16, 2)
 
    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x
      
    def sample_action(self, obs, epsilon): #obs --> state(4개)
        out = self.forward(obs) #out --> tensor([-0.1974,  0.1102], grad_fn=<AddBackward0>)
        coin = random.random() #0~1 소수
        if coin < epsilon: #coin이 더 작으면
            return random.randint(0,1) #0과 1중 하나 랜덤하게
        else : 
            return out.argmax().item()
            
def train(q, q_target, memory, optimizer):
    for i in range(10):
        s,a,r,s_prime,done_mask = memory.sample(batch_size)
        q_out = q(s) #q_out --> (32,2)  
        q_a = q_out.gather(1,a) #q_a --> (32,1) q_out의 2개 중에 a가 선택한 col을 골라서 생성
        max_q_prime = q_target(s_prime).max(1)[0].unsqueeze(1) #unsqueeze 차원 추가
        target = r + gamma * max_q_prime * done_mask  #gamma -- discounted reward rate
        loss = F.smooth_l1_loss(q_a, target)
        
        optimizer.zero_grad() #기울기 쌓이는 것 방지
        loss.backward() #역전파
        optimizer.step() #웨이트 업데이트
 
def main():
    env = gym.make('CartPole-v1')
    q = Qnet()
    q_target = Qnet()
    q_target.load_state_dict(q.state_dict())
    memory = ReplayBuffer()
 
    print_interval = 20
    score = 0.0  
    optimizer = optim.Adam(q.parameters(), lr=learning_rate)
 
    for n_epi in range(100):
        epsilon = max(0.01, 0.15 - 0.01*(n_epi/50)) #Linear annealing from 8% to 1%
        s, _ = env.reset()
        done = False
 
        while not done:
            a = q.sample_action(torch.from_numpy(s).float(), epsilon)      
            s_prime, r, done, truncated, info = env.step(a)
            done_mask = 0.0 if done else 1.0
            memory.put((s,a,r/100.0,s_prime, done_mask)) #튜플을 넣음
            s = s_prime
 
            score += r
            if done:
                break
            
        if memory.size()>320:
            train(q, q_target, memory, optimizer)
 
        if n_epi%print_interval==0 and n_epi!=0:
            q_target.load_state_dict(q.state_dict())
            print("n_episode :{}, score : {:.1f}, n_buffer : {}, eps : {:.1f}%".format(
                                                            n_epi, score/print_interval, memory.size(), epsilon*100))
            score = 0.0
    env.close()
    
if __name__ == '__main__':
    main()

 

 

Be positive

Be rich
Live your life