<jupyter_start><jupyter_text>Chapter 3 - Deep Q-learning Deep Reinforcement Learning *in Action*  Listing 3.1<jupyter_code>from Gridworld import Gridworld
game = Gridworld(size=4, mode='static')
game.display()
game.makeMove('d')
game.makeMove('d')
game.makeMove('l')
game.display()
game.reward()
game.board.render_np()
game.board.render_np().shape<jupyter_output><empty_output><jupyter_text>Listing 3.2<jupyter_code>import numpy as np
import torch
from Gridworld import Gridworld
from IPython.display import clear_output
import random
from matplotlib import pylab as plt

l1 = 64
l2 = 150
l3 = 100
l4 = 4

model = torch.nn.Sequential(
    torch.nn.Linear(l1, l2),
    torch.nn.ReLU(),
    torch.nn.Linear(l2, l3),
    torch.nn.ReLU(),
    torch.nn.Linear(l3,l4)
)
loss_fn = torch.nn.MSELoss()
learning_rate = 1e-3
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

gamma = 0.9
epsilon = 1.0
learning_rate = 1e-3
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

gamma = 0.9
epsilon = 1.0<jupyter_output><empty_output><jupyter_text>Listing 3.3<jupyter_code>action_set = {
    0: 'u',
    1: 'd',
    2: 'l',
    3: 'r',
}
epochs = 1000
losses = [] # 손실값 담을 목록, 추세 그래프 그릴 시 사용
for i in range(epochs): # 주 훈련 루프
    game = Gridworld(size=4, mode='static') # 훈련 주기별로 새 게임 시작
    state_ = game.board.render_np().reshape(1,64) + np.random.rand(1,64)/10.0 # 게임 상태 정보에 잡음(무작위 요소)추가
    state1 = torch.from_numpy(state_).float() # numpy 배열을 PyTorch 텐서로 변환 후 새 변수에 설정
    status = 1 # 게임 진행 여부 (1 : 실행중, 0 : 종료)
    while(status == 1): # 훈련 주기별 반복 루프
        qval = model(state1) # 예측 시작
        qval_ = qval.data.numpy()
        if (random.random() < epsilon): # 엡실론 탐욕법 --> 동작 선택
            action_ = np.random.randint(0,4)
        else:
            action_ = np.argmax(qval_)
        
        action = action_set[action_] # 번호 -> 행동 으로 변환
        game.makeMove(action) # 동작 수행
        state2_ = game.board.render_np().reshape(1,64) + np.random.rand(1,64)/10.0
        state2 = torch.from_numpy(state2_).float() # 행동으로 인한 새 상태 획득
        reward = game.reward()
        with torch.no_grad():
            newQ = model(state2.reshape(1,64))
        maxQ = torch.max(newQ) # 새 상태로 최대 Q value 갱신
        if reward == -1: # 목표 Q value 계산
            Y = reward + (gamma * maxQ)
        else:
            Y = reward
        Y = torch.Tensor([Y]).detach()
        X = qval.squeeze()[action_] # 1인 차원 제거 함수 --> 현재 동작에 해당하는 성분만 갱신해 새 변수에 추가
        loss = loss_fn(X, Y)
        print(i, loss.item())
        clear_output(wait=True)
        optimizer.zero_grad()
        loss.backward()
        losses.append(loss.item())
        optimizer.step()
        state1 = state2
        if reward != -1: # 보상이 -1이 아니면 게임 종료(10, -10)
            status = 0
    if epsilon > 0.1: # 한 주기마다 엡실론 감소
        epsilon -= (1/epochs)
qval
action_
qval.squeeze()
qval.squeeze()[action_]
plt.figure(figsize=(10,7))
plt.plot(losses)
plt.xlabel("Epochs",fontsize=22)
plt.ylabel("Loss",fontsize=22)<jupyter_output><empty_output><jupyter_text>PyTorch Automatic Differentiation Review<jupyter_code>m = torch.Tensor([2.0])
m.requires_grad=True
b = torch.Tensor([1.0])
b.requires_grad=True
def linear_model(x,m,b):
    y = m + b#@ x + b
    return y
#with torch.no_grad():
y = linear_model(torch.Tensor([4]),m,b)
y.grad_fn
y.backward()
m.grad<jupyter_output><empty_output><jupyter_text>Listing 3.4<jupyter_code>def test_model(model, mode='static', display=True):
    i = 0
    test_game = Gridworld(mode=mode)
    state_ = test_game.board.render_np().reshape(1,64) + np.random.rand(1,64)/10.0
    state = torch.from_numpy(state_).float()
    if display:
        print("Initial State:")
        print(test_game.display())
    status = 1
    while(status == 1):
        qval = model(state)
        qval_ = qval.data.numpy()
        action_ = np.argmax(qval_) # 최대 Q value에 해당하는 행동 선택
        action = action_set[action_]
        if display:
            print('Move #: %s; Taking action: %s' % (i, action))
        test_game.makeMove(action)
        state_ = test_game.board.render_np().reshape(1,64) + np.random.rand(1,64)/10.0
        state = torch.from_numpy(state_).float()
        if display:
            print(test_game.display())
        reward = test_game.reward()
        if reward != -1:
            if reward > 0:
                status = 2
                if display:
                    print("Game won! Reward: %s" % (reward,))
            else:
                status = 0
                if display:
                    print("Game LOST. Reward: %s" % (reward,))
        i += 1
        if (i > 15):
            if display:
                print("Game lost; too many moves.")
            break
    
    win = True if status == 2 else False
    return win
test_model(model)
test_model(model,"random")<jupyter_output>Initial State:
[[' ' ' ' ' ' ' ']
 [' ' 'P' '+' ' ']
 ['-' ' ' ' ' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 0; Taking action: d
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' 'P' ' ' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 1; Taking action: r
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' ' ' 'P' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 2; Taking action: l
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' 'P' ' ' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 3; Taking action: d
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' 'P' ' ' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 4; Taking action: d
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' 'P' ' ' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 5; Taking action: r
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' ' ' 'P' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 6; Taking action: l
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' 'P' ' ' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 7; Taking action: d
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 ['-' 'P' ' ' ' ']
 [' ' 'W' ' ' ' ']]
Move #: 8; Taking action: d
[[' ' ' ' ' ' ' ']
 [' ' ' ' '+' ' ']
 [[...]<jupyter_text>Listing 3.5<jupyter_code>l1 = 64
l2 = 150
l3 = 100
l4 = 4

model = torch.nn.Sequential(
    torch.nn.Linear(l1, l2),
    torch.nn.ReLU(),
    torch.nn.Linear(l2, l3),
    torch.nn.ReLU(),
    torch.nn.Linear(l3,l4)
)
loss_fn = torch.nn.MSELoss()
learning_rate = 1e-3
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

learning_rate = 1e-3
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

gamma = 0.9
epsilon = 0.3
from collections import deque
epochs = 5000
losses = []
mem_size = 1000 #A
batch_size = 200 #B
replay = deque(maxlen=mem_size) #C
max_moves = 50 #D
h = 0
for i in range(epochs):
    game = Gridworld(size=4, mode='random')
    state1_ = game.board.render_np().reshape(1,64) + np.random.rand(1,64)/100.0
    state1 = torch.from_numpy(state1_).float()
    status = 1
    mov = 0
    while(status == 1): 
        mov += 1
        qval = model(state1) #E
        qval_ = qval.data.numpy()
        if (random.random() < epsilon): #F
            action_ = np.random.randint(0,4)
        else:
            action_ = np.argmax(qval_)
        
        action = action_set[action_]
        game.makeMove(action)
        state2_ = game.board.render_np().reshape(1,64) + np.random.rand(1,64)/100.0
        state2 = torch.from_numpy(state2_).float()
        reward = game.reward()
        done = True if reward > 0 else False
        exp =  (state1, action_, reward, state2, done) #G
        replay.append(exp) #H
        state1 = state2
        
        if len(replay) > batch_size: #I
            minibatch = random.sample(replay, batch_size) #J
            state1_batch = torch.cat([s1 for (s1,a,r,s2,d) in minibatch]) #K
            action_batch = torch.Tensor([a for (s1,a,r,s2,d) in minibatch])
            reward_batch = torch.Tensor([r for (s1,a,r,s2,d) in minibatch])
            state2_batch = torch.cat([s2 for (s1,a,r,s2,d) in minibatch])
            done_batch = torch.Tensor([d for (s1,a,r,s2,d) in minibatch])
            
            Q1 = model(state1_batch) #L
            with torch.no_grad():
                Q2 = model(state2_batch) #M
            
            Y = reward_batch + gamma * ((1 - done_batch) * torch.max(Q2,dim=1)[0]) #N
            X = Q1.gather(dim=1,index=action_batch.long().unsqueeze(dim=1)).squeeze()
            loss = loss_fn(X, Y.detach())
            print(i, loss.item())
            clear_output(wait=True)
            optimizer.zero_grad()
            loss.backward()
            losses.append(loss.item())
            optimizer.step()

        if reward != -1 or mov > max_moves: #O
            status = 0
            mov = 0
losses = np.array(losses)

#A Set the total size of the experience replay memory
#B Set the minibatch size
#C Create the memory replay as a deque list
#D Maximum number of moves before game is over
#E Compute Q-values from input state in order to select action
#F Select action using epsilon-greedy strategy
#G Create experience of state, reward, action and next state as a tuple
#H Add experience to experience replay list
#I If replay list is at least as long as minibatch size, begin minibatch training
#J Randomly sample a subset of the replay list
#K Separate out the components of each experience into separate minibatch tensors
#L Re-compute Q-values for minibatch of states to get gradients
#M Compute Q-values for minibatch of next states but don't compute gradients
#N Compute the target Q-values we want the DQN to learn
#O If game is over, reset status and mov number
def running_mean(x,N=50):
    c = x.shape[0] - N
    y = np.zeros(c)
    conv = np.ones(N)
    for i in range(c):
        y[i] = (x[i:i+N] @ conv)/N
    return y
plt.figure(figsize=(10,7))
plt.plot(losses)
plt.xlabel("Epochs",fontsize=22)
plt.ylabel("Loss",fontsize=22)
max_games = 1000
wins = 0
for i in range(max_games):
    win = test_model(model, mode='random', display=False)
    if win:
        wins += 1
win_perc = float(wins) / float(max_games)
print("Games played: {0}, # of wins: {1}".format(max_games,wins))
print("Win percentage: {}%".format(100.0*win_perc))
test_model(model, mode='random')<jupyter_output>Initial State:
[[' ' ' ' '+' 'W']
 [' ' ' ' ' ' ' ']
 [' ' ' ' ' ' ' ']
 [' ' ' ' 'P' '-']]
Move #: 0; Taking action: u
[[' ' ' ' '+' 'W']
 [' ' ' ' ' ' ' ']
 [' ' ' ' 'P' ' ']
 [' ' ' ' ' ' '-']]
Move #: 1; Taking action: u
[[' ' ' ' '+' 'W']
 [' ' ' ' 'P' ' ']
 [' ' ' ' ' ' ' ']
 [' ' ' ' ' ' '-']]
Move #: 2; Taking action: u
[[' ' ' ' '+' 'W']
 [' ' ' ' ' ' ' ']
 [' ' ' ' ' ' ' ']
 [' ' ' ' ' ' '-']]
Game won! Reward: 10<jupyter_text>Listing 3.7<jupyter_code>import copy

l1 = 64
l2 = 150
l3 = 100
l4 = 4


model = torch.nn.Sequential(
    torch.nn.Linear(l1, l2),
    torch.nn.ReLU(),
    torch.nn.Linear(l2, l3),
    torch.nn.ReLU(),
    torch.nn.Linear(l3,l4)
)

model2 = copy.deepcopy(model) #A
model2.load_state_dict(model.state_dict()) #B

loss_fn = torch.nn.MSELoss()
learning_rate = 1e-3
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

gamma = 0.9
epsilon = 0.3
from collections import deque
epochs = 5000
losses = []
mem_size = 1000
batch_size = 200
replay = deque(maxlen=mem_size)
max_moves = 50
h = 0
sync_freq = 500 #A
j=0
for i in range(epochs):
    game = Gridworld(size=4, mode='random')
    state1_ = game.board.render_np().reshape(1,64) + np.random.rand(1,64)/100.0
    state1 = torch.from_numpy(state1_).float()
    status = 1
    mov = 0
    while(status == 1): 
        j+=1
        mov += 1
        qval = model(state1)
        qval_ = qval.data.numpy()
        if (random.random() < epsilon):
            action_ = np.random.randint(0,4)
        else:
            action_ = np.argmax(qval_)
        
        action = action_set[action_]
        game.makeMove(action)
        state2_ = game.board.render_np().reshape(1,64) + np.random.rand(1,64)/100.0
        state2 = torch.from_numpy(state2_).float()
        reward = game.reward()
        done = True if reward > 0 else False
        exp =  (state1, action_, reward, state2, done)
        replay.append(exp) #H
        state1 = state2
        
        if len(replay) > batch_size:
            minibatch = random.sample(replay, batch_size)
            state1_batch = torch.cat([s1 for (s1,a,r,s2,d) in minibatch])
            action_batch = torch.Tensor([a for (s1,a,r,s2,d) in minibatch])
            reward_batch = torch.Tensor([r for (s1,a,r,s2,d) in minibatch])
            state2_batch = torch.cat([s2 for (s1,a,r,s2,d) in minibatch])
            done_batch = torch.Tensor([d for (s1,a,r,s2,d) in minibatch])
            Q1 = model(state1_batch) 
            with torch.no_grad():
                Q2 = model2(state2_batch) #B
            
            Y = reward_batch + gamma * ((1-done_batch) * torch.max(Q2,dim=1)[0])
            X = Q1.gather(dim=1,index=action_batch.long().unsqueeze(dim=1)).squeeze()
            loss = loss_fn(X, Y.detach())
            print(i, loss.item())
            clear_output(wait=True)
            optimizer.zero_grad()
            loss.backward()
            losses.append(loss.item())
            optimizer.step()
            
            if j % sync_freq == 0: #C
                model2.load_state_dict(model.state_dict())
        if reward != -1 or mov > max_moves:
            status = 0
            mov = 0
        
losses = np.array(losses)

#A Set the update frequency for synchronizing the target model parameters to the main DQN
#B Use the target network to get the maiximum Q-value for the next state
#C Copy the main model parameters to the target network
plt.figure(figsize=(10,7))
plt.plot(losses)
plt.xlabel("Epochs",fontsize=22)
plt.ylabel("Loss",fontsize=22)
max_games = 1000
wins = 0
for i in range(max_games):
    win = test_model(model, mode='random', display=False)
    if win:
        wins += 1
win_perc = float(wins) / float(max_games)
print("Games played: {0}, # of wins: {1}".format(max_games,wins))
print("Win percentage: {}%".format(100.0*win_perc))<jupyter_output>Games played: 1000, # of wins: 911
Win percentage: 91.10000000000001%
{'fraction_non_alphanumeric': 0.16594422771275827, 'fraction_numerical': 0.03207628955353273, 'mean_word_length': 2.4939424533064107, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 38, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '43203', 'license': 'MIT', 'n_tokens_mistral': 5638, 'n_tokens_neox': 5529, 'n_words': 1164, 'path': 'Chapter 3/Ch3_book.ipynb', 'repo_name': 'az8602/DeepReinforcementLearningInAction', 'size': '13842'}