rlightning.buffer.utils.utils

Default preprocessing utilities for buffer transitions.

rlightning.buffer.utils.utils.default_compute_gae(rewards: MockModule('torch.Tensor'), values: MockModule('torch.Tensor'), next_values: MockModule('torch.Tensor'), dones: MockModule('torch.Tensor'), gamma: float, lam: float, normalize_adv: bool = True) → Tuple[MockModule('torch.Tensor'), MockModule('torch.Tensor')][source]

Compute Generalized Advantage Estimation (GAE).

This version uses an explicit loop over the batch dimension.

Parameters:

  • rewards (torch.Tensor) – Rewards at each timestep

  • values (torch.Tensor) – Value function estimates at each timestep

  • next_values (torch.Tensor) – Value function estimates at the next timestep

  • dones (torch.Tensor) – Done flags at each timestep

  • gamma (float) – Discount factor

  • lam (float) – GAE lambda parameter

  • normalize_adv (bool) – Whether to normalize the advantages

Returns:

Computed advantages and returns

Return type:

Tuple[torch.Tensor, torch.Tensor]

rlightning.buffer.utils.utils.default_env_ret_preprocess_fn(transition_dict: Dict[str, Any], env_ret: EnvRet, obs_preprocessor: Preprocessor, reward_preprocessor: Preprocessor) → Dict[str, Any][source]

Populate a transition dict from an EnvRet.

rlightning.buffer.utils.utils.default_gae_no_loop(rewards: MockModule('torch.Tensor'), values: MockModule('torch.Tensor'), next_values: MockModule('torch.Tensor'), dones: MockModule('torch.Tensor'), gamma: float, lam: float, normalize_adv: bool = True) → Tuple[MockModule('torch.Tensor'), MockModule('torch.Tensor')][source]

Compute Generalized Advantage Estimation (GAE).

This version uses matrix operations to eliminate explicit loops for efficiency.

Parameters:

  • rewards (torch.Tensor) – Rewards at each timestep

  • values (torch.Tensor) – Value function estimates at each timestep

  • next_values (torch.Tensor) – Value function estimates at the next timestep

  • dones (torch.Tensor) – Done flags at each timestep

  • gamma (float) – Discount factor

  • lam (float) – GAE lambda parameter

  • normalize_adv (bool) – Whether to normalize the advantages

Returns:

Computed advantages and returns

Return type:

Tuple[torch.Tensor, torch.Tensor]

rlightning.buffer.utils.utils.default_policy_resp_preprocess_fn(transition_dict: Dict[str, Any], policy_resp: PolicyResponse) → Dict[str, Any][source]

Populate a transition dict from a PolicyResponse.

rlightning.buffer.utils.utils.default_postprocess_fn(episode_buffer: Dict[str, List[Any]]) → Dict[str, Any][source]

Convert an episode buffer into a flat training-ready dict.

rlightning.buffer.utils.utils.default_preprocess_fn(transition_dict: ~typing.Dict[str, ~typing.Any], env_ret: ~rlightning.types.env_rets.EnvRet | None = None, policy_resp: ~rlightning.types.policy_response.PolicyResponse | None = None, obs_preprocessor: ~rlightning.buffer.utils.preprocessors.Preprocessor | None = <function default_obs_preprocessor>, reward_preprocessor: ~rlightning.buffer.utils.preprocessors.Preprocessor | None = <function default_reward_preprocessor>, env_ret_preprocess_fn: ~typing.Callable | None = <function default_env_ret_preprocess_fn>, policy_resp_preprocess_fn: ~typing.Callable | None = <function default_policy_resp_preprocess_fn>) → Dict[str, Any][source]

Default transition preprocess function. It will use the given obs_preprocessor and reward_preprocessor to preprocess both env_ret and policy_resp, or either one of them. When adding transition in a sync manner (env_ret and policy_resp are paired in one step of rollout), both env_ret and policy_resp should be provided. When adding transition in an async manner, only one of them should be provided.

Parameters:

  • transition_dict – The dict to aggregate transition data from env_ret and policy_resp.

  • env_ret (Optional[EnvRet]) – The environment return to be preprocessed.

  • policy_resp (Optional[PolicyResponse]) – The policy response to be preprocessed.

  • obs_preprocessor (Optional[Preprocessor]) – The preprocessor for observations.

  • reward_preprocessor (Optional[Preprocessor]) – The preprocessor for rewards.

  • env_ret_preprocess_fn (Optional[Callable]) – Function to preprocess env_ret.

  • policy_resp_preprocess_fn (Optional[Callable]) – Function to preprocess policy_resp.

Returns:

The preprocessed transition dict.