Source code for ding.worker.replay_buffer.episode_buffer
from typing import List
from ding.worker.replay_buffer import NaiveReplayBuffer
from ding.utils import BUFFER_REGISTRY
[docs]@BUFFER_REGISTRY.register('episode')
class EpisodeReplayBuffer(NaiveReplayBuffer):
r"""
Overview:
Episode replay buffer is a buffer to store complete episodes, i.e. Each element in episode buffer is an episode.
Some algorithms do not want to sample `batch_size` complete episodes, however, they want some transitions with
some fixed length. As a result, ``sample`` should be overwritten for those requirements.
Interface:
start, close, push, update, sample, clear, count, state_dict, load_state_dict, default_config
"""
@property
def episode_len(self) -> List[int]:
return [len(episode) for episode in self._data]