class_docstr: Observer for writing episodes to Reverb.
  
  This observer should be called at every step. It does not support batched
  trajectories. The steps are cached and written at the end of the episode.
  
  At the end of each episode, an item is written to Reverb. Each item is the
  trajectory containing an episode, including a boundary step in the end.
  Therefore, the sequence lengths of the items may vary. If you want a fixed
  sequence length, use `ReverbAddTrajectoryObserver` instead.
  
  Unfinished episodes remain in the cache and do not get written until
  `reset(write_cached_steps=True)` is called.
  
  TODO(b/176261664): Note that this observer only supports batch_size=1 from the
  consumer, if your episodes have variable lengths.
docstr: Cache the single step trajectory to be written into Reverb.
  
  Allows trajectory to be a flattened trajectory. No batch dimension allowed.
  
  Args:
    trajectory: The trajectory to be written which could be (possibly nested)
      trajectory object or a flattened version of a trajectory. It assumes
      there is *no* batch dimension.
  
  Raises:
    ValueError: If `bypass_partial_episodes` == False and episode length
      is > `max_sequence_length`.
code: def __call__(self, trajectory: trajectory_lib.Trajectory) -> None:
      """Cache the single step trajectory to be written into Reverb.
  
      Allows trajectory to be a flattened trajectory. No batch dimension allowed.
  
      Args:
        trajectory: The trajectory to be written which could be (possibly nested)
          trajectory object or a flattened version of a trajectory. It assumes
          there is *no* batch dimension.
  
      Raises:
        ValueError: If `bypass_partial_episodes` == False and episode length
          is > `max_sequence_length`.
      """
      # TODO(b/176494855): Raise an error if an invalid trajectory is passed in.
      # Currently, invalid `traj` value (mid->first, last->last) is not specially
      # handled and is treated as a normal mid->mid step.
      if (
          self._cached_steps >= self._max_sequence_length
          and not self._overflow_episode
      ):
          self._overflow_episode = True
          if self._bypass_partial_episodes:
              logging.error(
                  "The number of trajectories within the same episode exceeds "
                  "`max_sequence_length`. This episode is bypassed and will NOT "
                  "be written into the replay buffer. Consider increasing the "
                  "`max_sequence_length`."
              )
          else:
              raise ValueError(
                  "The number of trajectories within the same episode "
                  "exceeds `max_sequence_length`. Consider increasing the "
                  "`max_sequence_length` or set `bypass_partial_episodes` to true "
                  "to bypass the episodes with length more than "
                  "`max_sequence_length`."
              )
  
      # At the end of the overflowing episode, drop the cached incomplete episode
      # and reset the writer.
      if self._overflow_episode and trajectory.is_boundary():
          self.reset(write_cached_steps=False)
          return
  
      if not self._overflow_episode:
          self._get_writer().append(trajectory)
          self._writer_has_data = True
          self._cached_steps += 1
  
          # At the end of an episode, write the item to Reverb and clear the cache.
          if trajectory.is_boundary():
              self.reset(write_cached_steps=True)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from typing import Sequence, Text, Union
  from absl import logging
  import tensorflow
  from tf_agents.trajectories import trajectory
  from tf_agents.typing import types
  from tf_agents.utils import lazy_loader
label_desc: Use image node_id 6 for calling the ReverbAddEpisodeObserver obj's underlying member method code with example usage: obj.__call__(trajectory) without return types