import logging import math from typing import Any, Dict, List, Optional import numpy as np from ray.data import Dataset from ray.rllib.offline.estimators.fqe_torch_model import FQETorchModel from ray.rllib.offline.estimators.off_policy_estimator import OffPolicyEstimator from ray.rllib.offline.offline_evaluation_utils import compute_q_and_v_values from ray.rllib.offline.offline_evaluator import OfflineEvaluator from ray.rllib.policy import Policy from ray.rllib.policy.sample_batch import SampleBatch, convert_ma_batch_to_sample_batch from ray.rllib.utils.annotations import DeveloperAPI, override from ray.rllib.utils.numpy import convert_to_numpy from ray.rllib.utils.typing import SampleBatchType logger = logging.getLogger() @DeveloperAPI class DirectMethod(OffPolicyEstimator): r"""The Direct Method estimator. Let s_t, a_t, and r_t be the state, action, and reward at timestep t. This method trains a Q-model for the evaluation policy \pi_e on behavior data generated by \pi_b. Currently, RLlib implements this using Fitted-Q Evaluation (FQE). You can also implement your own model and pass it in as `q_model_config = {"type": your_model_class, **your_kwargs}`. This estimator computes the expected return for \pi_e for an episode as: V^{\pi_e}(s_0) = \sum_{a \in A} \pi_e(a | s_0) Q(s_0, a) and returns the mean and standard deviation over episodes. For more information refer to https://arxiv.org/pdf/1911.06854.pdf""" @override(OffPolicyEstimator) def __init__( self, policy: Policy, gamma: float, epsilon_greedy: float = 0.0, q_model_config: Optional[Dict] = None, ): """Initializes a Direct Method OPE Estimator. Args: policy: Policy to evaluate. gamma: Discount factor of the environment. epsilon_greedy: The probability by which we act acording to a fully random policy during deployment. With 1-epsilon_greedy we act according the target policy. q_model_config: Arguments to specify the Q-model. Must specify a `type` key pointing to the Q-model class. This Q-model is trained in the train() method and is used to compute the state-value estimates for the DirectMethod estimator. It must implement `train` and `estimate_v`. TODO (Rohan138): Unify this with RLModule API. """ super().__init__(policy, gamma, epsilon_greedy) # Some dummy policies and ones that are not based on a tensor framework # backend can come without a config or without a framework key. if hasattr(policy, "config"): assert ( policy.config.get("framework", "torch") == "torch" ), "Framework must be torch to use DirectMethod." q_model_config = q_model_config or {} model_cls = q_model_config.pop("type", FQETorchModel) self.model = model_cls( policy=policy, gamma=gamma, **q_model_config, ) assert hasattr( self.model, "estimate_v" ), "self.model must implement `estimate_v`!" @override(OffPolicyEstimator) def estimate_on_single_episode(self, episode: SampleBatch) -> Dict[str, Any]: estimates_per_epsiode = {} rewards = episode["rewards"] v_behavior = 0.0 for t in range(episode.count): v_behavior += rewards[t] * self.gamma**t v_target = self._compute_v_target(episode[:1]) estimates_per_epsiode["v_behavior"] = v_behavior estimates_per_epsiode["v_target"] = v_target return estimates_per_epsiode @override(OffPolicyEstimator) def estimate_on_single_step_samples( self, batch: SampleBatch ) -> Dict[str, List[float]]: estimates_per_epsiode = {} rewards = batch["rewards"] v_behavior = rewards v_target = self._compute_v_target(batch) estimates_per_epsiode["v_behavior"] = v_behavior estimates_per_epsiode["v_target"] = v_target return estimates_per_epsiode def _compute_v_target(self, init_step): v_target = self.model.estimate_v(init_step) v_target = convert_to_numpy(v_target) return v_target @override(OffPolicyEstimator) def train(self, batch: SampleBatchType) -> Dict[str, Any]: """Trains self.model on the given batch. Args: batch: A SampleBatchType to train on Returns: A dict with key "loss" and value as the mean training loss. """ batch = convert_ma_batch_to_sample_batch(batch) losses = self.model.train(batch) return {"loss": np.mean(losses)} @override(OfflineEvaluator) def estimate_on_dataset( self, dataset: Dataset, *, n_parallelism: int = ... ) -> Dict[str, Any]: """Calculates the Direct Method estimate on the given dataset. Note: This estimate works for only discrete action spaces for now. Args: dataset: Dataset to compute the estimate on. Each record in dataset should include the following columns: `obs`, `actions`, `action_prob` and `rewards`. The `obs` on each row shoud be a vector of D dimensions. n_parallelism: The number of parallel workers to use. Returns: Dictionary with the following keys: v_target: The estimated value of the target policy. v_behavior: The estimated value of the behavior policy. v_gain: The estimated gain of the target policy over the behavior policy. v_std: The standard deviation of the estimated value of the target. """ # compute v_values batch_size = max(dataset.count() // n_parallelism, 1) updated_ds = dataset.map_batches( compute_q_and_v_values, batch_size=batch_size, batch_format="pandas", fn_kwargs={ "model_class": self.model.__class__, "model_state": self.model.get_state(), "compute_q_values": False, }, ) v_behavior = updated_ds.mean("rewards") v_target = updated_ds.mean("v_values") v_gain_mean = v_target / v_behavior v_gain_ste = ( updated_ds.std("v_values") / v_behavior / math.sqrt(dataset.count()) ) return { "v_behavior": v_behavior, "v_target": v_target, "v_gain_mean": v_gain_mean, "v_gain_ste": v_gain_ste, }