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