torch_mixins.py 8.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225
  1. from ray.rllib.policy.policy import Policy, PolicyState
  2. from ray.rllib.policy.sample_batch import SampleBatch
  3. from ray.rllib.policy.torch_policy import TorchPolicy
  4. from ray.rllib.utils.annotations import DeveloperAPI, override
  5. from ray.rllib.utils.framework import try_import_torch
  6. from ray.rllib.utils.schedules import PiecewiseSchedule
  7. torch, nn = try_import_torch()
  8. @DeveloperAPI
  9. class LearningRateSchedule:
  10. """Mixin for TorchPolicy that adds a learning rate schedule."""
  11. def __init__(self, lr, lr_schedule):
  12. self._lr_schedule = None
  13. # Disable any scheduling behavior related to learning if Learner API is active.
  14. # Schedules are handled by Learner class.
  15. if lr_schedule is None:
  16. self.cur_lr = lr
  17. else:
  18. self._lr_schedule = PiecewiseSchedule(
  19. lr_schedule, outside_value=lr_schedule[-1][-1], framework=None
  20. )
  21. self.cur_lr = self._lr_schedule.value(0)
  22. @override(Policy)
  23. def on_global_var_update(self, global_vars):
  24. super().on_global_var_update(global_vars)
  25. if self._lr_schedule and not self.config.get("_enable_learner_api", False):
  26. self.cur_lr = self._lr_schedule.value(global_vars["timestep"])
  27. for opt in self._optimizers:
  28. for p in opt.param_groups:
  29. p["lr"] = self.cur_lr
  30. @DeveloperAPI
  31. class EntropyCoeffSchedule:
  32. """Mixin for TorchPolicy that adds entropy coeff decay."""
  33. def __init__(self, entropy_coeff, entropy_coeff_schedule):
  34. self._entropy_coeff_schedule = None
  35. # Disable any scheduling behavior related to learning if Learner API is active.
  36. # Schedules are handled by Learner class.
  37. if entropy_coeff_schedule is None or (
  38. self.config.get("_enable_learner_api", False)
  39. ):
  40. self.entropy_coeff = entropy_coeff
  41. else:
  42. # Allows for custom schedule similar to lr_schedule format
  43. if isinstance(entropy_coeff_schedule, list):
  44. self._entropy_coeff_schedule = PiecewiseSchedule(
  45. entropy_coeff_schedule,
  46. outside_value=entropy_coeff_schedule[-1][-1],
  47. framework=None,
  48. )
  49. else:
  50. # Implements previous version but enforces outside_value
  51. self._entropy_coeff_schedule = PiecewiseSchedule(
  52. [[0, entropy_coeff], [entropy_coeff_schedule, 0.0]],
  53. outside_value=0.0,
  54. framework=None,
  55. )
  56. self.entropy_coeff = self._entropy_coeff_schedule.value(0)
  57. @override(Policy)
  58. def on_global_var_update(self, global_vars):
  59. super(EntropyCoeffSchedule, self).on_global_var_update(global_vars)
  60. if self._entropy_coeff_schedule is not None:
  61. self.entropy_coeff = self._entropy_coeff_schedule.value(
  62. global_vars["timestep"]
  63. )
  64. @DeveloperAPI
  65. class KLCoeffMixin:
  66. """Assigns the `update_kl()` method to a TorchPolicy.
  67. This is used by Algorithms to update the KL coefficient
  68. after each learning step based on `config.kl_target` and
  69. the measured KL value (from the train_batch).
  70. """
  71. def __init__(self, config):
  72. # The current KL value (as python float).
  73. self.kl_coeff = config["kl_coeff"]
  74. # Constant target value.
  75. self.kl_target = config["kl_target"]
  76. def update_kl(self, sampled_kl):
  77. # Update the current KL value based on the recently measured value.
  78. if sampled_kl > 2.0 * self.kl_target:
  79. self.kl_coeff *= 1.5
  80. elif sampled_kl < 0.5 * self.kl_target:
  81. self.kl_coeff *= 0.5
  82. # Return the current KL value.
  83. return self.kl_coeff
  84. @override(TorchPolicy)
  85. def get_state(self) -> PolicyState:
  86. state = super().get_state()
  87. # Add current kl-coeff value.
  88. state["current_kl_coeff"] = self.kl_coeff
  89. return state
  90. @override(TorchPolicy)
  91. def set_state(self, state: PolicyState) -> None:
  92. # Set current kl-coeff value first.
  93. self.kl_coeff = state.pop("current_kl_coeff", self.config["kl_coeff"])
  94. # Call super's set_state with rest of the state dict.
  95. super().set_state(state)
  96. @DeveloperAPI
  97. class ValueNetworkMixin:
  98. """Assigns the `_value()` method to a TorchPolicy.
  99. This way, Policy can call `_value()` to get the current VF estimate on a
  100. single(!) observation (as done in `postprocess_trajectory_fn`).
  101. Note: When doing this, an actual forward pass is being performed.
  102. This is different from only calling `model.value_function()`, where
  103. the result of the most recent forward pass is being used to return an
  104. already calculated tensor.
  105. """
  106. def __init__(self, config):
  107. # When doing GAE, we need the value function estimate on the
  108. # observation.
  109. if config.get("use_gae") or config.get("vtrace"):
  110. # Input dict is provided to us automatically via the Model's
  111. # requirements. It's a single-timestep (last one in trajectory)
  112. # input_dict.
  113. def value(**input_dict):
  114. input_dict = SampleBatch(input_dict)
  115. input_dict = self._lazy_tensor_dict(input_dict)
  116. model_out, _ = self.model(input_dict)
  117. # [0] = remove the batch dim.
  118. return self.model.value_function()[0].item()
  119. # When not doing GAE, we do not require the value function's output.
  120. else:
  121. def value(*args, **kwargs):
  122. return 0.0
  123. self._value = value
  124. def extra_action_out(self, input_dict, state_batches, model, action_dist):
  125. """Defines extra fetches per action computation.
  126. Args:
  127. input_dict (Dict[str, TensorType]): The input dict used for the action
  128. computing forward pass.
  129. state_batches (List[TensorType]): List of state tensors (empty for
  130. non-RNNs).
  131. model (ModelV2): The Model object of the Policy.
  132. action_dist: The instantiated distribution
  133. object, resulting from the model's outputs and the given
  134. distribution class.
  135. Returns:
  136. Dict[str, TensorType]: Dict with extra tf fetches to perform per
  137. action computation.
  138. """
  139. # Return value function outputs. VF estimates will hence be added to
  140. # the SampleBatches produced by the sampler(s) to generate the train
  141. # batches going into the loss function.
  142. return {
  143. SampleBatch.VF_PREDS: model.value_function(),
  144. }
  145. @DeveloperAPI
  146. class TargetNetworkMixin:
  147. """Mixin class adding a method for (soft) target net(s) synchronizations.
  148. - Adds the `update_target` method to the policy.
  149. Calling `update_target` updates all target Q-networks' weights from their
  150. respective "main" Q-networks, based on tau (smooth, partial updating).
  151. """
  152. def __init__(self):
  153. # Hard initial update from Q-net(s) to target Q-net(s).
  154. tau = self.config.get("tau", 1.0)
  155. self.update_target(tau=tau)
  156. def update_target(self, tau=None):
  157. # Update_target_fn will be called periodically to copy Q network to
  158. # target Q network, using (soft) tau-synching.
  159. tau = tau or self.config.get("tau", 1.0)
  160. model_state_dict = self.model.state_dict()
  161. # Support partial (soft) synching.
  162. # If tau == 1.0: Full sync from Q-model to target Q-model.
  163. if self.config.get("_enable_rl_module_api", False):
  164. target_current_network_pairs = self.model.get_target_network_pairs()
  165. for target_network, current_network in target_current_network_pairs:
  166. current_state_dict = current_network.state_dict()
  167. new_state_dict = {
  168. k: tau * current_state_dict[k] + (1 - tau) * v
  169. for k, v in target_network.state_dict().items()
  170. }
  171. target_network.load_state_dict(new_state_dict)
  172. else:
  173. # Support partial (soft) synching.
  174. # If tau == 1.0: Full sync from Q-model to target Q-model.
  175. target_state_dict = next(iter(self.target_models.values())).state_dict()
  176. model_state_dict = {
  177. k: tau * model_state_dict[k] + (1 - tau) * v
  178. for k, v in target_state_dict.items()
  179. }
  180. for target in self.target_models.values():
  181. target.load_state_dict(model_state_dict)
  182. @override(TorchPolicy)
  183. def set_weights(self, weights):
  184. # Makes sure that whenever we restore weights for this policy's
  185. # model, we sync the target network (from the main model)
  186. # at the same time.
  187. TorchPolicy.set_weights(self, weights)
  188. self.update_target()