Perform random-sample one-step tabular Q-planning with prioritized sweeping. Notes ----- This approach uses a priority queue to retrieve the state-action pairs from the agent's history with largest change to their Q-values if backed up. When
(self)
| 1676 | return priority |
| 1677 | |
| 1678 | def _simulate_behavior(self): |
| 1679 | """ |
| 1680 | Perform random-sample one-step tabular Q-planning with prioritized |
| 1681 | sweeping. |
| 1682 | |
| 1683 | Notes |
| 1684 | ----- |
| 1685 | This approach uses a priority queue to retrieve the state-action pairs |
| 1686 | from the agent's history with largest change to their Q-values if |
| 1687 | backed up. When the first pair in the queue is backed up, the effect on |
| 1688 | each of its predecessor pairs is computed. If the predecessor's |
| 1689 | priority is greater than a small threshold the pair is added to the |
| 1690 | queue and the process is repeated until either the queue is empty or we |
| 1691 | have exceeded a `n_simulated_actions` updates. |
| 1692 | """ |
| 1693 | env_model = self.parameters["model"] |
| 1694 | sweep_queue = self.derived_variables["sweep_queue"] |
| 1695 | for _ in range(self.n_simulated_actions): |
| 1696 | if len(sweep_queue) == 0: |
| 1697 | break |
| 1698 | |
| 1699 | # select (s, a) pair with the largest update (priority) |
| 1700 | sq_items = list(sweep_queue.items()) |
| 1701 | (s_sim, a_sim), _ = sorted(sq_items, key=lambda x: x[1], reverse=True)[0] |
| 1702 | |
| 1703 | # remove entry from queue |
| 1704 | del sweep_queue[(s_sim, a_sim)] |
| 1705 | |
| 1706 | # update Q function for (s_sim, a_sim) using the full-backup |
| 1707 | # version of the TD(0) Q-learning update |
| 1708 | self._update(s_sim, a_sim) |
| 1709 | |
| 1710 | # get all (_s, _a) pairs that lead to s_sim (ie., s_sim's predecessors) |
| 1711 | pairs = env_model.state_action_pairs_leading_to_outcome(s_sim) |
| 1712 | |
| 1713 | # add predecessors to queue if their priority exceeds thresh |
| 1714 | for (_s, _a) in pairs: |
| 1715 | self._update_queue(_s, _a) |
| 1716 | |
| 1717 | def _update(self, s, a): |
| 1718 | """ |
no test coverage detected