<jupyter_start><jupyter_text>1. Imports & Setup 1.1 Imports<jupyter_code>import torch
from torch import nn
from torch.nn.functional import mse_loss
from torch.optim import Adam
import numpy as np
import matplotlib.pyplot as plt

from collections import deque
from typing import NamedTuple, Any
import random

from unityagents import UnityEnvironment
from dqn_agent import Agent, ReplayBuffer
from model import DQN<jupyter_output><empty_output><jupyter_text>1.2 Setup<jupyter_code># please do not modify the line below
env = UnityEnvironment(file_name="/Users/bothmena/Projects/ai/ReinforcementLearning/deep-reinforcement-learning/bin/Banana.app")<jupyter_output>INFO:unityagents:
'Academy' started successfully!
Unity Academy name: Academy
        Number of Brains: 1
        Number of External Brains : 1
        Lesson number : 0
        Reset Parameters :
		
Unity brain name: BananaBrain
        Number of Visual Observations (per agent): 0
        Vector Observation space type: continuous
        Vector Observation space size (per agent): 37
        Number of stacked Vector Observation: 1
        Vector Action space type: discrete
        Vector Action space size (per agent): 4
        Vector Action descriptions: , , ,<jupyter_text>1.3 Training HyperParameters<jupyter_code>BUFFER_SIZE = int(1e5)  # replay buffer size
BATCH_SIZE = 64  # mini-batch size
GAMMA = 0.99  # discount factor
TAU = 1e-3  # for soft update of target parameters
LR = 5e-4  # learning rate
UPDATE_EVERY = 4  # how often to update the network
DQN_HIDDEN_LAYERS = [64, 128, 128, 128]
model_id = '64_128_128_128'
SEED = 0<jupyter_output><empty_output><jupyter_text>2 Defining Experience<jupyter_code>class Experience(NamedTuple):
    state: Any
    action: Any
    reward: Any
    next_state: Any
    done: Any<jupyter_output><empty_output><jupyter_text>3 Defining DQN<jupyter_code>class DQN(nn.Module):
    def __init__(self, state_size, action_size, hidden_layers_units):
        super(DQN, self).__init__()
        layers = [
            nn.Linear(state_size, hidden_layers_units[0]),
        ]
        for i, units in enumerate(hidden_layers_units):
            next_units = hidden_layers_units[i+1] if i < len(hidden_layers_units) - 1 else action_size
            layers.append(nn.ReLU())
            layers.append(nn.Linear(units, next_units))

        self.fc_layers = nn.Sequential(*layers)

    def forward(self, state):
        return self.fc_layers(state)<jupyter_output><empty_output><jupyter_text>4. Defining ReplayBuffer  5. Defining Agent<jupyter_code>device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')

print(device)<jupyter_output>cpu<jupyter_text>6. Setting environment up<jupyter_code># get the default brain
brain_name = env.brain_names[0]
brain = env.brains[brain_name]
# reset the environment
env_info = env.reset(train_mode=True)[brain_name]
action_size = brain.vector_action_space_size
state_size = len(env_info.vector_observations[0])
env_info = env.reset(train_mode=True)[brain_name]
print(action_size, state_size)
# dir(env_info)
# dir(env)<jupyter_output><empty_output><jupyter_text>7. Train the agent with DQN<jupyter_code>class ReplayBuffer:
    def __init__(self, action_size, buffer_size, batch_size, seed):
        self.action_size = action_size
        self.batch_size = batch_size
        self.memory = deque(maxlen=buffer_size)
        self.seed = random.seed(seed)

    def add(self, experience: Experience):
        self.memory.append(experience)

    def sample(self):
        experiences = random.sample(self.memory, k=self.batch_size)

        states = torch.from_numpy(np.vstack([e.state for e in experiences if e is not None])).float().to(device)
        actions = torch.from_numpy(np.vstack([e.action for e in experiences if e is not None])).long().to(device)
        rewards = torch.from_numpy(np.vstack([e.reward for e in experiences if e is not None])).float().to(device)
        next_states = torch.from_numpy(np.vstack([e.next_state for e in experiences if e is not None])).float() \
            .to(device)
        dones = torch.from_numpy(np.vstack([e.done for e in experiences if e is not None]).astype(np.uint8)) \
            .float().to(device)
        
        return states, actions, rewards, next_states, dones

    def __len__(self):
        return len(self.memory)

class Agent:
    def __init__(self, state_size, action_size, buffer_size, seed):
        """Initialize an Agent object.

        Params
        ======
            state_size (int): dimension of each state
            action_size (int): dimension of each action
            seed (int): random seed
        """
        self.state_size = state_size
        self.action_size = action_size
        self.seed = random.seed(seed)

        self.dqn_local = DQN(state_size, action_size, DQN_HIDDEN_LAYERS)
        self.dqn_target = DQN(state_size, action_size, DQN_HIDDEN_LAYERS)
        self.optimizer = Adam(self.dqn_local.parameters(), lr=LR)

        self.buffer = ReplayBuffer(action_size, buffer_size, batch_size=BATCH_SIZE, seed=seed)
        self.t_step = 0

    def step(self, experience: Experience):
        # Save experience in replay memory
        self.buffer.add(experience)

        # Learn every UPDATE_EVERY time steps.
        self.t_step = (self.t_step + 1) % UPDATE_EVERY
        if self.t_step == 0 and len(self.buffer) >= BATCH_SIZE:
            experiences = self.buffer.sample()
            self.learn(experiences, GAMMA)

    def act(self, state, eps=0.):
        """Returns actions for given state as per current policy.

        Params
        ======
            state (array_like): current state
            eps (float): epsilon, for epsilon-greedy action selection
        """
        random_val = np.random.random()

        if random_val < eps:
            return np.random.choice(self.action_size)

        state = torch.from_numpy(state).float().unsqueeze(0).to(device)
        self.dqn_local.eval()
        with torch.no_grad():
            action = np.argmax(self.dqn_local(state)).item()
        self.dqn_local.train()

        return action

    def learn(self, experiences: tuple, gamma):
        states, actions, rewards, next_states, dones = experiences
        # Get max predicted Q values (for next states) from target model
        q_target_next = self.dqn_target(next_states).detach().max(1)[0].unsqueeze(1)
        # Compute Q targets for current states
        q_targets = rewards + (gamma * q_target_next * (1 - dones))

        # Get expected Q values from local model
        q_predicted = self.dqn_local(states)
        q_expected = q_predicted.gather(1, actions)

        # Compute loss
        loss = mse_loss(q_expected, q_targets)

        # Minimize the loss
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        # ------------------- update target network ------------------- #
        self.soft_update()

    def soft_update(self):
        """Soft update model parameters.
        θ_target = τ*θ_local + (1 - τ)*θ_target

        Params
        ======
            local_model (PyTorch model): weights will be copied from
            target_model (PyTorch model): weights will be copied to
            tau (float): interpolation parameter
        """
        for target_param, local_param in zip(self.dqn_target.parameters(), self.dqn_local.parameters()):
            target_param.data.copy_(TAU * local_param.data + (1.0 - TAU) * target_param.data)

agent = Agent(state_size=state_size, action_size=action_size, buffer_size=BUFFER_SIZE, seed=SEED)
# first attempt: [64, 64]
scores = train_dqn(n_episodes=2000, max_t=1000, eps_start=1.0, eps_end=0.01, eps_decay=0.995, save_every=2000)
plt.figure(figsize=(20, 10))
plt.plot(scores)
plt.savefig('saved_plots/results_{}.png'.format(model_id))
plt.figure(figsize=(20, 10))
plt.plot(scores)
plt.savefig('saved_plots/results_{}.png'.format(model_id))
plt.figure(figsize=(20, 10))
plt.plot(scores)
plt.savefig('saved_plots/results_{}.png'.format(model_id))
scores_64_128_128_128 = get_avg_first_10()
print(scores_64_128_128_128)
scores_64_128 = get_avg_first_10()
print(scores_64_128)
# scores_64_64_64 = get_avg_first_10()
print(scores_64_64_64)
scores_64_64 = get_avg_first_10()
print(scores_64_64)
step = 20
def get_avg_first_10():
    avg_scores = []
    for i in range(step, len(scores), step):
        avg = 0
        for j in range(i - step, i):
            avg += scores[j]/step
        avg_scores.append(round(avg, 2))
    
    return avg_scores[:10]

scores_64_128_64 = get_avg_first_10()
print(scores_64_128_64)<jupyter_output>[-0.35, -0.05, -0.35, 0.75, 0.45, 0.25, 0.95, 2.2, 2.55]
{'fraction_non_alphanumeric': 0.10913500404203719, 'fraction_numerical': 0.025753551218385493, 'mean_word_length': 3.2534381139489192, 'pattern_counts': {'":': 0, '<': 32, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '30676', 'license': 'MIT', 'n_tokens_mistral': 3042, 'n_tokens_neox': 2779, 'n_words': 657, 'path': 'p1_navigation/Navigation.ipynb', 'repo_name': 'bothmena/drl-nanodegree', 'size': '8659'}