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