| 264 | |
| 265 | |
| 266 | class DQNAgent(object): |
| 267 | def __init__(self, state_size, action_size): |
| 268 | self.state_size = state_size |
| 269 | self.action_size = action_size |
| 270 | self.memory = ReplayBuffer(state_size, action_size, size=500) |
| 271 | self.gamma = 0.95 # discount rate |
| 272 | self.epsilon = 1.0 # exploration rate |
| 273 | self.epsilon_min = 0.01 |
| 274 | self.epsilon_decay = 0.995 |
| 275 | self.model = mlp(state_size, action_size) |
| 276 | |
| 277 | |
| 278 | def update_replay_memory(self, state, action, reward, next_state, done): |
| 279 | self.memory.store(state, action, reward, next_state, done) |
| 280 | |
| 281 | |
| 282 | def act(self, state): |
| 283 | if np.random.rand() <= self.epsilon: |
| 284 | return np.random.choice(self.action_size) |
| 285 | act_values = self.model.predict(state, verbose=0) |
| 286 | return np.argmax(act_values[0]) # returns action |
| 287 | |
| 288 | |
| 289 | def replay(self, batch_size=32): |
| 290 | # first check if replay buffer contains enough data |
| 291 | if self.memory.size < batch_size: |
| 292 | return |
| 293 | |
| 294 | # sample a batch of data from the replay memory |
| 295 | minibatch = self.memory.sample_batch(batch_size) |
| 296 | states = minibatch['s'] |
| 297 | actions = minibatch['a'] |
| 298 | rewards = minibatch['r'] |
| 299 | next_states = minibatch['s2'] |
| 300 | done = minibatch['d'] |
| 301 | |
| 302 | # Calculate the tentative target: Q(s',a) |
| 303 | target = rewards + (1 - done) * self.gamma * np.amax(self.model.predict(next_states, verbose=0), axis=1) |
| 304 | |
| 305 | # With the Keras API, the target (usually) must have the same |
| 306 | # shape as the predictions. |
| 307 | # However, we only need to update the network for the actions |
| 308 | # which were actually taken. |
| 309 | # We can accomplish this by setting the target to be equal to |
| 310 | # the prediction for all values. |
| 311 | # Then, only change the targets for the actions taken. |
| 312 | # Q(s,a) |
| 313 | target_full = self.model.predict(states, verbose=0) |
| 314 | target_full[np.arange(batch_size), actions] = target |
| 315 | |
| 316 | # Run one training step |
| 317 | self.model.train_on_batch(states, target_full) |
| 318 | |
| 319 | if self.epsilon > self.epsilon_min: |
| 320 | self.epsilon *= self.epsilon_decay |
| 321 | |
| 322 | |
| 323 | def load(self, name): |