Implementation Prioritized Experience Replay

README.md

@@ -5,6 +5,7 @@ Reinforcement Learning Library PyTorch backend.
 - [ ] DDPG
 - [ ] HER
 - [x] DQN
+- [ ] Prioritized Experience Replay
 - [ ] A2C
 - [ ] Apex
 

insomnia/replay_buffers/__init__.py

@@ -1,3 +1,4 @@
 from insomnia.replay_buffers.replay_buffer import ReplayBuffer
 from . import replay_buffer
-
+from insomnia.replay_buffers.buffer_core import BaseBuffer
+from insomnia.replay_buffers.prioritized_buffer import PrioritizedBuffer

insomnia/replay_buffers/buffer_core.py

@@ -0,0 +1,22 @@
+from collections import namedtuple
+
+
+class BaseBuffer:
+    """The Base of Experience Replay.
+    """
+    Transition = namedtuple('Transition', ('state', 'action', 'next_state', 'reward', 'terminal'))
+
+    def store_transition(self):
+        """To store state transition
+        """
+        raise NotImplementedError
+
+    def sample_buffer(self):
+        """Sampling data
+        """
+        raise NotImplementedError
+
+    def __len__(self):
+        """Return data length
+        """
+        raise NotImplementedError

insomnia/replay_buffers/replay_buffer.py

@@ -1,8 +1,9 @@
 import numpy as np
 import torch
+from buffer_core import BaseBuffer
 
 
-class ReplayBuffer:
+class ReplayBuffer(BaseBuffer):
     """Experience Replay.
 
     Experience Replay is just sampling randomly from buffer.
@@ -16,7 +17,7 @@ class ReplayBuffer:
         reward_memory (torch.Tensor): the buffer of reward.
         terminal_memory(torch.Tensor): the buffer for done or not.
     """
-    def __init__(self, state_dim, act_dim, cuda, max_size=10000):
+    def __init__(self, state_dim, act_dim, cuda=True, max_size=10000):
         """Initial of ReplayBuffer
 
         Args:
@@ -82,3 +83,8 @@ def __len__(self):
 
         """
         return self.mem_control
+
+
+if __name__ == '__main__':
+    buffer = ReplayBuffer([3], 2, False)
+    print(len(buffer))