Created
November 6, 2019 05:26
-
-
Save jadechip/8c147de912ebd868ff68f7310ad29491 to your computer and use it in GitHub Desktop.
Reinforcement learning - collaboration and competition companion code
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class Agent(): | |
"""Interacts with and learns from the environment.""" | |
def __init__(self, state_size, action_size, random_seed, memory): | |
"""Initialize an Agent object. | |
Params | |
====== | |
state_size (int): dimension of each state | |
action_size (int): dimension of each action | |
random_seed (int): random seed | |
""" | |
self.state_size = state_size | |
self.action_size = action_size | |
self.seed = random.seed(random_seed) | |
# Common replay buffer for shared experiences | |
self.memory = memory | |
# Actor Network (w/ Target Network) | |
self.actor_local = Actor(state_size, action_size, random_seed).to(device) | |
self.actor_target = Actor(state_size, action_size, random_seed).to(device) | |
self.actor_optimizer = optim.Adam(self.actor_local.parameters(), lr=LR_ACTOR) | |
self.actor_loss = [] | |
# Critic Network (w/ Target Network) | |
self.critic_local = Critic(state_size, action_size, random_seed).to(device) | |
self.critic_target = Critic(state_size, action_size, random_seed).to(device) | |
self.critic_optimizer = optim.Adam(self.critic_local.parameters(), lr=LR_CRITIC, weight_decay=WEIGHT_DECAY) | |
self.critic_loss = [] | |
# Noise process | |
self.noise = OUNoise(action_size, random_seed) | |
def step(self, state, action, reward, next_state, done): | |
"""Save experience in replay memory, and use random sample from buffer to learn.""" | |
# Save experience / reward | |
self.memory.add(state, action, reward, next_state, done) | |
# Learn, if enough samples are available in memory | |
if len(self.memory) > BATCH_SIZE: | |
# return self.memory.sample() | |
experiences = self.memory.sample() | |
# experiences = [self.memory.sample() for _ in range(self.num_agents)] | |
self.learn(experiences, GAMMA) | |
def act(self, state, add_noise=True): | |
"""Returns actions for given state as per current policy.""" | |
state = torch.from_numpy(state).float().to(device) | |
# Set module to eval mode | |
self.actor_local.eval() | |
with torch.no_grad(): | |
action = self.actor_local(state).cpu().data.numpy() | |
# Set module to training mode | |
self.actor_local.train() | |
if add_noise: | |
action += self.noise.sample() | |
return np.clip(action, -1, 1) | |
def reset(self): | |
self.noise.reset() | |
def learn(self, experiences, gamma): | |
"""Update policy and value parameters using given batch of experience tuples. | |
Q_targets = r + γ * critic_target(next_state, actor_target(next_state)) | |
where: | |
actor_target(state) -> action | |
critic_target(state, action) -> Q-value | |
Params | |
====== | |
experiences (Tuple[torch.Tensor]): tuple of (s, a, r, s', done) tuples | |
gamma (float): discount factor | |
""" | |
states, actions, rewards, next_states, dones = experiences | |
# ---------------------------- update critic ---------------------------- # | |
# Get predicted next-state actions and Q values from target models | |
actions_next = self.actor_target(next_states) | |
Q_targets_next = self.critic_target(next_states, actions_next) | |
# Compute Q targets for current states (y_i) | |
# What we actually got, in terms of rewards | |
Q_targets = rewards + (gamma * Q_targets_next * (1 - dones)) | |
# Compute critic loss | |
# what the local critic produced | |
Q_expected = self.critic_local(states, actions) | |
critic_loss = F.mse_loss(Q_expected, Q_targets) | |
self.critic_loss.append(critic_loss) | |
# Minimize the loss | |
self.critic_optimizer.zero_grad() | |
critic_loss.backward() | |
self.critic_optimizer.step() | |
# ---------------------------- update actor ---------------------------- # | |
# Compute actor loss | |
actions_pred = self.actor_local(states) | |
actor_loss = -self.critic_local(states, actions_pred).mean() | |
self.actor_loss.append(actor_loss) | |
# Minimize the loss | |
self.actor_optimizer.zero_grad() | |
actor_loss.backward() | |
self.actor_optimizer.step() | |
# ----------------------- update target networks ----------------------- # | |
self.soft_update(self.critic_local, self.critic_target, TAU) | |
self.soft_update(self.actor_local, self.actor_target, TAU) | |
def soft_update(self, local_model, target_model, tau): | |
"""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(target_model.parameters(), local_model.parameters()): | |
target_param.data.copy_(tau*local_param.data + (1.0-tau)*target_param.data) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment