mirror of
https://github.com/huggingface/lerobot.git
synced 2026-08-08 17:39:44 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1834f819a6 |
@@ -215,4 +215,6 @@
|
|||||||
title: Environments
|
title: Environments
|
||||||
- local: api/configs
|
- local: api/configs
|
||||||
title: Configuration
|
title: Configuration
|
||||||
|
- local: api/rl
|
||||||
|
title: Reinforcement Learning
|
||||||
title: "API Reference"
|
title: "API Reference"
|
||||||
|
|||||||
@@ -1,93 +0,0 @@
|
|||||||
# Optimization
|
|
||||||
|
|
||||||
`OptimizerConfig` and `LRSchedulerConfig` are the base configuration classes for the optimizers and learning
|
|
||||||
rate schedulers used during training. `TrainPipelineConfig` composes one of each; see
|
|
||||||
[`~optim.factory.make_optimizer_and_scheduler`] for how they are built from a policy's parameters.
|
|
||||||
|
|
||||||
## make_optimizer_and_scheduler
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.factory.make_optimizer_and_scheduler
|
|
||||||
|
|
||||||
## OptimizerConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.OptimizerConfig
|
|
||||||
- type
|
|
||||||
- builds_multiple_optimizers
|
|
||||||
- default_choice_name
|
|
||||||
- build
|
|
||||||
|
|
||||||
## AdamConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.AdamConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## AdamWConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.AdamWConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## SGDConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.SGDConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## MultiAdamConfig
|
|
||||||
|
|
||||||
Builds a dictionary of Adam optimizers, one per parameter group — used when a policy needs separate
|
|
||||||
optimizers for different components (e.g. actor/critic/temperature in SAC).
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.MultiAdamConfig
|
|
||||||
- builds_multiple_optimizers
|
|
||||||
- build
|
|
||||||
|
|
||||||
## XVLAAdamWConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.XVLAAdamWConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## save_optimizer_state
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.save_optimizer_state
|
|
||||||
|
|
||||||
## load_optimizer_state
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.load_optimizer_state
|
|
||||||
|
|
||||||
## LRSchedulerConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.LRSchedulerConfig
|
|
||||||
- type
|
|
||||||
- build
|
|
||||||
|
|
||||||
## DiffuserSchedulerConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.DiffuserSchedulerConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## VQBeTSchedulerConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.VQBeTSchedulerConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## ConstantWithWarmupSchedulerConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.schedulers.ConstantWithWarmupSchedulerConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## CosineAnnealingWithWarmupSchedulerConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.schedulers.CosineAnnealingWithWarmupSchedulerConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## CosineDecayWithWarmupSchedulerConfig
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.CosineDecayWithWarmupSchedulerConfig
|
|
||||||
- build
|
|
||||||
|
|
||||||
## save_scheduler_state
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.save_scheduler_state
|
|
||||||
|
|
||||||
## load_scheduler_state
|
|
||||||
|
|
||||||
[[autodoc]] lerobot.optim.load_scheduler_state
|
|
||||||
@@ -0,0 +1,87 @@
|
|||||||
|
# Reinforcement Learning
|
||||||
|
|
||||||
|
`lerobot.rl` is the distributed actor/learner reinforcement-learning stack behind
|
||||||
|
[Train a Robot with RL](../hilserl) (HIL-SERL) and [Train RL in Simulation](../hilserl_sim). Algorithms,
|
||||||
|
the replay buffer, data sources, and the trainer are gRPC-free and usable standalone; the actor/learner
|
||||||
|
entry points (`actor`, `learner`, `learner_service`) additionally require `pip install 'lerobot[hilserl]'`.
|
||||||
|
|
||||||
|
## TrainRLServerPipelineConfig
|
||||||
|
|
||||||
|
Top-level configuration for both the `lerobot-actor` and `lerobot-learner` CLIs.
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.train_rl.TrainRLServerPipelineConfig
|
||||||
|
|
||||||
|
## RLAlgorithm
|
||||||
|
|
||||||
|
Abstract base every RL algorithm subclasses.
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.base.RLAlgorithm
|
||||||
|
|
||||||
|
## RLAlgorithmConfig
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.configs.RLAlgorithmConfig
|
||||||
|
|
||||||
|
## TrainingStats
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.configs.TrainingStats
|
||||||
|
|
||||||
|
## make_algorithm
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.factory.make_algorithm
|
||||||
|
|
||||||
|
## make_algorithm_config
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.factory.make_algorithm_config
|
||||||
|
|
||||||
|
## get_algorithm_class
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.factory.get_algorithm_class
|
||||||
|
|
||||||
|
## SAC
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.sac.sac_algorithm.SACAlgorithm
|
||||||
|
- all
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.algorithms.sac.configuration_sac.SACAlgorithmConfig
|
||||||
|
|
||||||
|
## ReplayBuffer
|
||||||
|
|
||||||
|
In-memory replay buffer of transitions, sampled in batches for off-policy training.
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.buffer.ReplayBuffer
|
||||||
|
- all
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.buffer.BatchTransition
|
||||||
|
|
||||||
|
## DataMixer
|
||||||
|
|
||||||
|
Abstract interface for combining online and offline data sources into training batches.
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.data_sources.DataMixer
|
||||||
|
- all
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.data_sources.OnlineOfflineMixer
|
||||||
|
- all
|
||||||
|
|
||||||
|
## RLTrainer
|
||||||
|
|
||||||
|
Unified training-step orchestrator: holds the algorithm, a `DataMixer`, and an optional preprocessor.
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.trainer.RLTrainer
|
||||||
|
- all
|
||||||
|
|
||||||
|
## Actor / learner CLIs
|
||||||
|
|
||||||
|
The distributed actor and learner processes communicate over gRPC; see [Train a Robot with
|
||||||
|
RL](../hilserl) for the full workflow.
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.actor.actor_cli
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.learner.train_cli
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.learner_service.LearnerService
|
||||||
|
- all
|
||||||
|
|
||||||
|
## eval_policy
|
||||||
|
|
||||||
|
[[autodoc]] lerobot.rl.eval_policy.eval_policy
|
||||||
+2
-2
@@ -447,10 +447,10 @@ ignore = [
|
|||||||
"src/lerobot/jobs/**" = ["D"]
|
"src/lerobot/jobs/**" = ["D"]
|
||||||
"src/lerobot/model/**" = ["D"]
|
"src/lerobot/model/**" = ["D"]
|
||||||
"src/lerobot/motors/**" = ["D"]
|
"src/lerobot/motors/**" = ["D"]
|
||||||
|
"src/lerobot/optim/**" = ["D"]
|
||||||
"src/lerobot/policies/**" = ["D"]
|
"src/lerobot/policies/**" = ["D"]
|
||||||
"src/lerobot/processor/**" = ["D"]
|
"src/lerobot/processor/**" = ["D"]
|
||||||
"src/lerobot/rewards/**" = ["D"]
|
"src/lerobot/rewards/**" = ["D"]
|
||||||
"src/lerobot/rl/**" = ["D"]
|
|
||||||
"src/lerobot/rollout/**" = ["D"]
|
"src/lerobot/rollout/**" = ["D"]
|
||||||
"src/lerobot/scripts/**" = ["D"]
|
"src/lerobot/scripts/**" = ["D"]
|
||||||
"src/lerobot/teleoperators/**" = ["D"]
|
"src/lerobot/teleoperators/**" = ["D"]
|
||||||
@@ -514,7 +514,7 @@ ignore-private = false
|
|||||||
ignore-property-decorators = false
|
ignore-property-decorators = false
|
||||||
ignore-module = false
|
ignore-module = false
|
||||||
ignore-setters = false
|
ignore-setters = false
|
||||||
fail-under = 55
|
fail-under = 55.5
|
||||||
output-format = "term-missing"
|
output-format = "term-missing"
|
||||||
color = true
|
color = true
|
||||||
paths = ["src/lerobot"]
|
paths = ["src/lerobot"]
|
||||||
|
|||||||
@@ -25,20 +25,14 @@ from lerobot.policies import PreTrainedPolicy
|
|||||||
def make_optimizer_and_scheduler(
|
def make_optimizer_and_scheduler(
|
||||||
cfg: TrainPipelineConfig, policy: PreTrainedPolicy
|
cfg: TrainPipelineConfig, policy: PreTrainedPolicy
|
||||||
) -> tuple[Optimizer, LRScheduler | None]:
|
) -> tuple[Optimizer, LRScheduler | None]:
|
||||||
"""Build the optimizer and, if configured, the learning rate scheduler for training a policy.
|
"""Generates the optimizer and scheduler based on configs.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg (`TrainPipelineConfig`):
|
cfg (TrainPipelineConfig): The training config that contains optimizer and scheduler configs
|
||||||
The training config, whose `optimizer` and `scheduler` fields are built.
|
policy (PreTrainedPolicy): The policy config from which parameters and presets must be taken from.
|
||||||
policy (`PreTrainedPolicy`):
|
|
||||||
The policy being trained; its parameters (or optimizer-preset groups, if
|
|
||||||
`cfg.use_policy_training_preset` is `True`) are passed to the optimizer.
|
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
`tuple[Optimizer, LRScheduler | None]`: The built optimizer, and scheduler if one was configured.
|
tuple[Optimizer, LRScheduler | None]: The couple (Optimizer, Scheduler). Scheduler can be `None`.
|
||||||
|
|
||||||
Raises:
|
|
||||||
ValueError: If `cfg.optimizer` is `None`.
|
|
||||||
"""
|
"""
|
||||||
params = policy.get_optim_params() if cfg.use_policy_training_preset else policy.parameters()
|
params = policy.get_optim_params() if cfg.use_policy_training_preset else policy.parameters()
|
||||||
if cfg.optimizer is None:
|
if cfg.optimizer is None:
|
||||||
|
|||||||
+13
-120
@@ -44,32 +44,12 @@ OptimizerParams = (
|
|||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class OptimizerConfig(draccus.ChoiceRegistry, abc.ABC):
|
class OptimizerConfig(draccus.ChoiceRegistry, abc.ABC):
|
||||||
"""Base configuration shared by every optimizer.
|
|
||||||
|
|
||||||
Concrete optimizers subclass this and register themselves with
|
|
||||||
`@OptimizerConfig.register_subclass("name")`, which is what makes `--optimizer.type=name` work on the
|
|
||||||
command line.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
lr (`float`):
|
|
||||||
Learning rate.
|
|
||||||
weight_decay (`float`):
|
|
||||||
Weight decay (L2 penalty) applied by the optimizer.
|
|
||||||
grad_clip_norm (`float`):
|
|
||||||
Maximum gradient norm; gradients are clipped to this value before each optimizer step.
|
|
||||||
"""
|
|
||||||
|
|
||||||
lr: float
|
lr: float
|
||||||
weight_decay: float
|
weight_decay: float
|
||||||
grad_clip_norm: float
|
grad_clip_norm: float
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def type(self) -> str:
|
def type(self) -> str:
|
||||||
"""Return the registered name this config was registered under.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`str`: The name passed to `@OptimizerConfig.register_subclass`, e.g. `"adam"`.
|
|
||||||
"""
|
|
||||||
return self.get_choice_name(self.__class__)
|
return self.get_choice_name(self.__class__)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@@ -79,16 +59,12 @@ class OptimizerConfig(draccus.ChoiceRegistry, abc.ABC):
|
|||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def default_choice_name(cls) -> str | None:
|
def default_choice_name(cls) -> str | None:
|
||||||
"""Return the registered name used when `--optimizer.type` is not specified.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`str | None`: `"adam"`.
|
|
||||||
"""
|
|
||||||
return "adam"
|
return "adam"
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def build(self, params: OptimizerParams) -> torch.optim.Optimizer | dict[str, torch.optim.Optimizer]:
|
def build(self, params: OptimizerParams) -> torch.optim.Optimizer | dict[str, torch.optim.Optimizer]:
|
||||||
"""Build the optimizer. It can be a single optimizer or a dictionary of optimizers.
|
"""
|
||||||
|
Build the optimizer. It can be a single optimizer or a dictionary of optimizers.
|
||||||
|
|
||||||
NOTE: Multiple optimizers are useful when you have different models to optimize.
|
NOTE: Multiple optimizers are useful when you have different models to optimize.
|
||||||
For example, you can have one optimizer for the policy and another one for the value function
|
For example, you can have one optimizer for the policy and another one for the value function
|
||||||
@@ -113,21 +89,6 @@ class OptimizerConfig(draccus.ChoiceRegistry, abc.ABC):
|
|||||||
@OptimizerConfig.register_subclass("adam")
|
@OptimizerConfig.register_subclass("adam")
|
||||||
@dataclass
|
@dataclass
|
||||||
class AdamConfig(OptimizerConfig):
|
class AdamConfig(OptimizerConfig):
|
||||||
"""Configuration for [`torch.optim.Adam`](https://docs.pytorch.org/docs/stable/generated/torch.optim.Adam.html).
|
|
||||||
|
|
||||||
Args:
|
|
||||||
lr (`float`, *optional*, defaults to 0.001):
|
|
||||||
Learning rate.
|
|
||||||
weight_decay (`float`, *optional*, defaults to 0.0):
|
|
||||||
Weight decay (L2 penalty).
|
|
||||||
grad_clip_norm (`float`, *optional*, defaults to 10.0):
|
|
||||||
Maximum gradient norm; gradients are clipped to this value before each optimizer step.
|
|
||||||
betas (`tuple[float, float]`, *optional*, defaults to `(0.9, 0.999)`):
|
|
||||||
Coefficients used for computing running averages of the gradient and its square.
|
|
||||||
eps (`float`, *optional*, defaults to 1e-08):
|
|
||||||
Term added to the denominator to improve numerical stability.
|
|
||||||
"""
|
|
||||||
|
|
||||||
lr: float = 1e-3
|
lr: float = 1e-3
|
||||||
betas: tuple[float, float] = (0.9, 0.999)
|
betas: tuple[float, float] = (0.9, 0.999)
|
||||||
eps: float = 1e-8
|
eps: float = 1e-8
|
||||||
@@ -135,15 +96,6 @@ class AdamConfig(OptimizerConfig):
|
|||||||
grad_clip_norm: float = 10.0
|
grad_clip_norm: float = 10.0
|
||||||
|
|
||||||
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
||||||
"""Build a [`torch.optim.Adam`](https://docs.pytorch.org/docs/stable/generated/torch.optim.Adam.html) instance from this config.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
params (`OptimizerParams`):
|
|
||||||
Parameters to optimize, as accepted by `torch.optim.Adam`.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`torch.optim.Optimizer`: The built optimizer.
|
|
||||||
"""
|
|
||||||
kwargs = asdict(self)
|
kwargs = asdict(self)
|
||||||
kwargs.pop("grad_clip_norm")
|
kwargs.pop("grad_clip_norm")
|
||||||
return torch.optim.Adam(params, **kwargs)
|
return torch.optim.Adam(params, **kwargs)
|
||||||
@@ -152,21 +104,6 @@ class AdamConfig(OptimizerConfig):
|
|||||||
@OptimizerConfig.register_subclass("adamw")
|
@OptimizerConfig.register_subclass("adamw")
|
||||||
@dataclass
|
@dataclass
|
||||||
class AdamWConfig(OptimizerConfig):
|
class AdamWConfig(OptimizerConfig):
|
||||||
"""Configuration for [`torch.optim.AdamW`](https://docs.pytorch.org/docs/stable/generated/torch.optim.AdamW.html).
|
|
||||||
|
|
||||||
Args:
|
|
||||||
lr (`float`, *optional*, defaults to 0.001):
|
|
||||||
Learning rate.
|
|
||||||
weight_decay (`float`, *optional*, defaults to 0.01):
|
|
||||||
Weight decay, applied decoupled from the gradient update as in the AdamW paper.
|
|
||||||
grad_clip_norm (`float`, *optional*, defaults to 10.0):
|
|
||||||
Maximum gradient norm; gradients are clipped to this value before each optimizer step.
|
|
||||||
betas (`tuple[float, float]`, *optional*, defaults to `(0.9, 0.999)`):
|
|
||||||
Coefficients used for computing running averages of the gradient and its square.
|
|
||||||
eps (`float`, *optional*, defaults to 1e-08):
|
|
||||||
Term added to the denominator to improve numerical stability.
|
|
||||||
"""
|
|
||||||
|
|
||||||
lr: float = 1e-3
|
lr: float = 1e-3
|
||||||
betas: tuple[float, float] = (0.9, 0.999)
|
betas: tuple[float, float] = (0.9, 0.999)
|
||||||
eps: float = 1e-8
|
eps: float = 1e-8
|
||||||
@@ -174,15 +111,6 @@ class AdamWConfig(OptimizerConfig):
|
|||||||
grad_clip_norm: float = 10.0
|
grad_clip_norm: float = 10.0
|
||||||
|
|
||||||
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
||||||
"""Build a [`torch.optim.AdamW`](https://docs.pytorch.org/docs/stable/generated/torch.optim.AdamW.html) instance from this config.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
params (`OptimizerParams`):
|
|
||||||
Parameters to optimize, as accepted by `torch.optim.AdamW`.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`torch.optim.Optimizer`: The built optimizer.
|
|
||||||
"""
|
|
||||||
kwargs = asdict(self)
|
kwargs = asdict(self)
|
||||||
kwargs.pop("grad_clip_norm")
|
kwargs.pop("grad_clip_norm")
|
||||||
return torch.optim.AdamW(params, **kwargs)
|
return torch.optim.AdamW(params, **kwargs)
|
||||||
@@ -191,23 +119,6 @@ class AdamWConfig(OptimizerConfig):
|
|||||||
@OptimizerConfig.register_subclass("sgd")
|
@OptimizerConfig.register_subclass("sgd")
|
||||||
@dataclass
|
@dataclass
|
||||||
class SGDConfig(OptimizerConfig):
|
class SGDConfig(OptimizerConfig):
|
||||||
"""Configuration for [`torch.optim.SGD`](https://docs.pytorch.org/docs/stable/generated/torch.optim.SGD.html).
|
|
||||||
|
|
||||||
Args:
|
|
||||||
lr (`float`, *optional*, defaults to 0.001):
|
|
||||||
Learning rate.
|
|
||||||
weight_decay (`float`, *optional*, defaults to 0.0):
|
|
||||||
Weight decay (L2 penalty).
|
|
||||||
grad_clip_norm (`float`, *optional*, defaults to 10.0):
|
|
||||||
Maximum gradient norm; gradients are clipped to this value before each optimizer step.
|
|
||||||
momentum (`float`, *optional*, defaults to 0.0):
|
|
||||||
Momentum factor.
|
|
||||||
dampening (`float`, *optional*, defaults to 0.0):
|
|
||||||
Dampening for momentum.
|
|
||||||
nesterov (`bool`, *optional*, defaults to `False`):
|
|
||||||
Whether to enable Nesterov momentum.
|
|
||||||
"""
|
|
||||||
|
|
||||||
lr: float = 1e-3
|
lr: float = 1e-3
|
||||||
momentum: float = 0.0
|
momentum: float = 0.0
|
||||||
dampening: float = 0.0
|
dampening: float = 0.0
|
||||||
@@ -216,15 +127,6 @@ class SGDConfig(OptimizerConfig):
|
|||||||
grad_clip_norm: float = 10.0
|
grad_clip_norm: float = 10.0
|
||||||
|
|
||||||
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
||||||
"""Build a [`torch.optim.SGD`](https://docs.pytorch.org/docs/stable/generated/torch.optim.SGD.html) instance from this config.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
params (`OptimizerParams`):
|
|
||||||
Parameters to optimize, as accepted by `torch.optim.SGD`.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`torch.optim.Optimizer`: The built optimizer.
|
|
||||||
"""
|
|
||||||
kwargs = asdict(self)
|
kwargs = asdict(self)
|
||||||
kwargs.pop("grad_clip_norm")
|
kwargs.pop("grad_clip_norm")
|
||||||
return torch.optim.SGD(params, **kwargs)
|
return torch.optim.SGD(params, **kwargs)
|
||||||
@@ -266,7 +168,8 @@ class XVLAAdamWConfig(OptimizerConfig):
|
|||||||
soft_prompt_warmup_lr_scale: float | None = None # If set, start soft-prompts at this scale (e.g., 0.01)
|
soft_prompt_warmup_lr_scale: float | None = None # If set, start soft-prompts at this scale (e.g., 0.01)
|
||||||
|
|
||||||
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
def build(self, params: OptimizerParams) -> torch.optim.Optimizer:
|
||||||
"""Build AdamW optimizer with differential learning rates.
|
"""
|
||||||
|
Build AdamW optimizer with differential learning rates.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
params: Must be a dict[str, Parameter] from dict(model.named_parameters())
|
params: Must be a dict[str, Parameter] from dict(model.named_parameters())
|
||||||
@@ -336,14 +239,10 @@ class MultiAdamConfig(OptimizerConfig):
|
|||||||
This creates a dictionary of Adam optimizers, each with its own hyperparameters.
|
This creates a dictionary of Adam optimizers, each with its own hyperparameters.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
lr (`float`, *optional*, defaults to 0.001):
|
lr: Default learning rate (used if not specified for a group)
|
||||||
Default learning rate, used for a group unless overridden in `optimizer_groups`.
|
weight_decay: Default weight decay (used if not specified for a group)
|
||||||
weight_decay (`float`, *optional*, defaults to 0.0):
|
optimizer_groups: Dictionary mapping parameter group names to their hyperparameters
|
||||||
Default weight decay, used for a group unless overridden in `optimizer_groups`.
|
grad_clip_norm: Gradient clipping norm
|
||||||
grad_clip_norm (`float`, *optional*, defaults to 10.0):
|
|
||||||
Maximum gradient norm; gradients are clipped to this value before each optimizer step.
|
|
||||||
optimizer_groups (`dict[str, dict[str, Any]]`, *optional*):
|
|
||||||
Per-group hyperparameter overrides (`lr`, `betas`, `eps`, `weight_decay`), keyed by group name.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
lr: float = 1e-3
|
lr: float = 1e-3
|
||||||
@@ -353,7 +252,6 @@ class MultiAdamConfig(OptimizerConfig):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def builds_multiple_optimizers(self) -> bool:
|
def builds_multiple_optimizers(self) -> bool:
|
||||||
"""`bool`: Always `True`; `build()` returns a dict of optimizers, one per parameter group."""
|
|
||||||
return True
|
return True
|
||||||
|
|
||||||
def build(self, params: OptimizerParams) -> dict[str, torch.optim.Optimizer]:
|
def build(self, params: OptimizerParams) -> dict[str, torch.optim.Optimizer]:
|
||||||
@@ -398,10 +296,8 @@ def save_optimizer_state(
|
|||||||
"""Save optimizer state to disk (non-sharded runs; sharded runs use the DCP channel).
|
"""Save optimizer state to disk (non-sharded runs; sharded runs use the DCP channel).
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
optimizer (`torch.optim.Optimizer | dict[str, torch.optim.Optimizer]`):
|
optimizer: Either a single optimizer or a dictionary of optimizers.
|
||||||
Either a single optimizer or a dictionary of optimizers.
|
save_dir: Directory to save the optimizer state.
|
||||||
save_dir (`Path`):
|
|
||||||
Directory to save the optimizer state.
|
|
||||||
"""
|
"""
|
||||||
if isinstance(optimizer, dict):
|
if isinstance(optimizer, dict):
|
||||||
# Handle dictionary of optimizers
|
# Handle dictionary of optimizers
|
||||||
@@ -429,14 +325,11 @@ def load_optimizer_state(
|
|||||||
"""Load optimizer state from disk.
|
"""Load optimizer state from disk.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
optimizer (`torch.optim.Optimizer | dict[str, torch.optim.Optimizer]`):
|
optimizer: Either a single optimizer or a dictionary of optimizers.
|
||||||
Either a single optimizer or a dictionary of optimizers.
|
save_dir: Directory to load the optimizer state from.
|
||||||
save_dir (`Path`):
|
|
||||||
Directory to load the optimizer state from.
|
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
`torch.optim.Optimizer | dict[str, torch.optim.Optimizer]`: The updated optimizer(s) with loaded
|
The updated optimizer(s) with loaded state.
|
||||||
state.
|
|
||||||
"""
|
"""
|
||||||
if isinstance(optimizer, dict):
|
if isinstance(optimizer, dict):
|
||||||
# Handle dictionary of optimizers
|
# Handle dictionary of optimizers
|
||||||
|
|||||||
@@ -36,64 +36,24 @@ else:
|
|||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class LRSchedulerConfig(draccus.ChoiceRegistry, abc.ABC):
|
class LRSchedulerConfig(draccus.ChoiceRegistry, abc.ABC):
|
||||||
"""Base configuration shared by every learning rate scheduler.
|
|
||||||
|
|
||||||
Concrete schedulers subclass this and register themselves with
|
|
||||||
`@LRSchedulerConfig.register_subclass("name")`, which is what makes `--scheduler.type=name` work on the
|
|
||||||
command line.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
num_warmup_steps (`int | None`):
|
|
||||||
Number of steps over which the learning rate ramps up from 0 before the scheduler's own
|
|
||||||
behavior takes over. `None` disables warmup.
|
|
||||||
"""
|
|
||||||
|
|
||||||
num_warmup_steps: int | None
|
num_warmup_steps: int | None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def type(self) -> str:
|
def type(self) -> str:
|
||||||
"""Return the registered name this config was registered under.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`str`: The name passed to `@LRSchedulerConfig.register_subclass`, e.g. `"diffuser"`.
|
|
||||||
"""
|
|
||||||
return self.get_choice_name(self.__class__)
|
return self.get_choice_name(self.__class__)
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def build(self, optimizer: Optimizer, num_training_steps: int) -> LRScheduler | None:
|
def build(self, optimizer: Optimizer, num_training_steps: int) -> LRScheduler | None:
|
||||||
"""Build the scheduler for a given optimizer and training length.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
optimizer (`Optimizer`):
|
|
||||||
The optimizer whose learning rate the scheduler will adjust.
|
|
||||||
num_training_steps (`int`):
|
|
||||||
Total number of training steps, used to compute decay/annealing schedules.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`LRScheduler | None`: The built scheduler.
|
|
||||||
"""
|
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
@LRSchedulerConfig.register_subclass("diffuser")
|
@LRSchedulerConfig.register_subclass("diffuser")
|
||||||
@dataclass
|
@dataclass
|
||||||
class DiffuserSchedulerConfig(LRSchedulerConfig):
|
class DiffuserSchedulerConfig(LRSchedulerConfig):
|
||||||
"""A [`diffusers`](https://huggingface.co/docs/diffusers) learning rate schedule.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
num_warmup_steps (`int`, *optional*):
|
|
||||||
Number of steps over which the learning rate ramps up from 0. `None` disables warmup.
|
|
||||||
name (`str`, *optional*, defaults to `"cosine"`):
|
|
||||||
Name of the `diffusers` schedule to build, e.g. `"cosine"`, `"linear"`, `"constant"`. See
|
|
||||||
[`diffusers.optimization.get_scheduler`](https://huggingface.co/docs/diffusers/api/schedulers/overview)
|
|
||||||
for the full list.
|
|
||||||
"""
|
|
||||||
|
|
||||||
name: str = "cosine"
|
name: str = "cosine"
|
||||||
num_warmup_steps: int | None = None
|
num_warmup_steps: int | None = None
|
||||||
|
|
||||||
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
||||||
"""See [`~optim.schedulers.LRSchedulerConfig.build`]. Delegates to `diffusers.optimization.get_scheduler`."""
|
|
||||||
require_package("diffusers", extra="diffusion")
|
require_package("diffusers", extra="diffusion")
|
||||||
|
|
||||||
kwargs = {**asdict(self), "num_training_steps": num_training_steps, "optimizer": optimizer}
|
kwargs = {**asdict(self), "num_training_steps": num_training_steps, "optimizer": optimizer}
|
||||||
@@ -103,31 +63,12 @@ class DiffuserSchedulerConfig(LRSchedulerConfig):
|
|||||||
@LRSchedulerConfig.register_subclass("vqbet")
|
@LRSchedulerConfig.register_subclass("vqbet")
|
||||||
@dataclass
|
@dataclass
|
||||||
class VQBeTSchedulerConfig(LRSchedulerConfig):
|
class VQBeTSchedulerConfig(LRSchedulerConfig):
|
||||||
"""Used to train VQ-BeT: constant LR during VQ-VAE pretraining, then warmup and cosine decay.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
num_warmup_steps (`int`):
|
|
||||||
Number of steps over which the learning rate ramps up from 0, counted from the end of VQ-VAE
|
|
||||||
pretraining.
|
|
||||||
num_vqvae_training_steps (`int`):
|
|
||||||
Number of initial steps spent pretraining the VQ-VAE, during which the LR stays at its peak.
|
|
||||||
num_cycles (`float`, *optional*, defaults to 0.5):
|
|
||||||
Number of cosine cycles in the decay phase; 0.5 decays smoothly to 0 by the end of training.
|
|
||||||
"""
|
|
||||||
|
|
||||||
num_warmup_steps: int
|
num_warmup_steps: int
|
||||||
num_vqvae_training_steps: int
|
num_vqvae_training_steps: int
|
||||||
num_cycles: float = 0.5
|
num_cycles: float = 0.5
|
||||||
|
|
||||||
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
||||||
"""See [`~optim.schedulers.LRSchedulerConfig.build`].
|
|
||||||
|
|
||||||
Holds the LR at its peak during VQ-VAE pretraining, then applies linear warmup followed by cosine
|
|
||||||
decay for the remaining steps.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def lr_lambda(current_step):
|
def lr_lambda(current_step):
|
||||||
"""Return the LR multiplier for `current_step`, per the VQ-BeT schedule."""
|
|
||||||
if current_step < self.num_vqvae_training_steps:
|
if current_step < self.num_vqvae_training_steps:
|
||||||
return float(1)
|
return float(1)
|
||||||
else:
|
else:
|
||||||
@@ -149,20 +90,14 @@ class ConstantWithWarmupSchedulerConfig(LRSchedulerConfig):
|
|||||||
|
|
||||||
Mirrors the ``warmup_constant_lambda`` used by LingBot-VA (upstream ``wan_va/train.py``):
|
Mirrors the ``warmup_constant_lambda`` used by LingBot-VA (upstream ``wan_va/train.py``):
|
||||||
the LR ramps linearly from 0 to the peak over ``num_warmup_steps`` steps, then stays flat.
|
the LR ramps linearly from 0 to the peak over ``num_warmup_steps`` steps, then stays flat.
|
||||||
|
|
||||||
Args:
|
|
||||||
num_warmup_steps (`int`, *optional*, defaults to 1000):
|
|
||||||
Number of steps over which the learning rate ramps up from 0 to its peak.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
num_warmup_steps: int = 1000
|
num_warmup_steps: int = 1000
|
||||||
|
|
||||||
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
||||||
"""See [`~optim.schedulers.LRSchedulerConfig.build`]."""
|
|
||||||
warmup_steps = self.num_warmup_steps or 0
|
warmup_steps = self.num_warmup_steps or 0
|
||||||
|
|
||||||
def lr_lambda(current_step):
|
def lr_lambda(current_step):
|
||||||
"""Return the LR multiplier for `current_step`: linear ramp, then constant `1.0`."""
|
|
||||||
if current_step < warmup_steps:
|
if current_step < warmup_steps:
|
||||||
return float(current_step) / float(max(1, warmup_steps))
|
return float(current_step) / float(max(1, warmup_steps))
|
||||||
return 1.0
|
return 1.0
|
||||||
@@ -176,19 +111,12 @@ class CosineAnnealingWithWarmupSchedulerConfig(LRSchedulerConfig):
|
|||||||
"""Linear warmup followed by cosine annealing from the peak LR to zero.
|
"""Linear warmup followed by cosine annealing from the peak LR to zero.
|
||||||
|
|
||||||
Used by EVO1; the annealing phase always spans the remaining training steps.
|
Used by EVO1; the annealing phase always spans the remaining training steps.
|
||||||
|
|
||||||
Args:
|
|
||||||
num_warmup_steps (`int`):
|
|
||||||
Number of steps over which the learning rate ramps up from 0 to its peak.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
num_warmup_steps: int
|
num_warmup_steps: int
|
||||||
|
|
||||||
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
||||||
"""See [`~optim.schedulers.LRSchedulerConfig.build`]."""
|
|
||||||
|
|
||||||
def lr_lambda(current_step: int) -> float:
|
def lr_lambda(current_step: int) -> float:
|
||||||
"""Return the LR multiplier for `current_step`: linear warmup, then cosine annealing to 0."""
|
|
||||||
if current_step < self.num_warmup_steps:
|
if current_step < self.num_warmup_steps:
|
||||||
return current_step / max(1, self.num_warmup_steps)
|
return current_step / max(1, self.num_warmup_steps)
|
||||||
progress = (current_step - self.num_warmup_steps) / max(
|
progress = (current_step - self.num_warmup_steps) / max(
|
||||||
@@ -206,18 +134,6 @@ class CosineDecayWithWarmupSchedulerConfig(LRSchedulerConfig):
|
|||||||
|
|
||||||
Automatically scales warmup and decay steps if num_training_steps < num_decay_steps.
|
Automatically scales warmup and decay steps if num_training_steps < num_decay_steps.
|
||||||
This ensures the learning rate schedule completes properly even with shorter training runs.
|
This ensures the learning rate schedule completes properly even with shorter training runs.
|
||||||
|
|
||||||
Args:
|
|
||||||
num_warmup_steps (`int`):
|
|
||||||
Number of steps over which the learning rate ramps up from `peak_lr / (num_warmup_steps + 1)`
|
|
||||||
to `peak_lr`.
|
|
||||||
num_decay_steps (`int`):
|
|
||||||
Number of steps over which the learning rate decays from `peak_lr` to `decay_lr`. Scaled down
|
|
||||||
automatically if `num_training_steps` is shorter than this.
|
|
||||||
peak_lr (`float`):
|
|
||||||
Learning rate reached at the end of warmup.
|
|
||||||
decay_lr (`float`):
|
|
||||||
Learning rate reached at the end of decay.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
num_warmup_steps: int
|
num_warmup_steps: int
|
||||||
@@ -226,11 +142,6 @@ class CosineDecayWithWarmupSchedulerConfig(LRSchedulerConfig):
|
|||||||
decay_lr: float
|
decay_lr: float
|
||||||
|
|
||||||
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
def build(self, optimizer: Optimizer, num_training_steps: int) -> LambdaLR:
|
||||||
"""See [`~optim.schedulers.LRSchedulerConfig.build`].
|
|
||||||
|
|
||||||
If `num_training_steps` is shorter than `num_decay_steps`, scales `num_warmup_steps` and
|
|
||||||
`num_decay_steps` down proportionally so the schedule still completes.
|
|
||||||
"""
|
|
||||||
# Auto-scale scheduler parameters if training steps are shorter than configured decay steps
|
# Auto-scale scheduler parameters if training steps are shorter than configured decay steps
|
||||||
actual_warmup_steps = self.num_warmup_steps
|
actual_warmup_steps = self.num_warmup_steps
|
||||||
actual_decay_steps = self.num_decay_steps
|
actual_decay_steps = self.num_decay_steps
|
||||||
@@ -250,17 +161,13 @@ class CosineDecayWithWarmupSchedulerConfig(LRSchedulerConfig):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def lr_lambda(current_step):
|
def lr_lambda(current_step):
|
||||||
"""Return the LR multiplier for `current_step`: linear warmup, then cosine decay."""
|
|
||||||
|
|
||||||
def linear_warmup_schedule(current_step):
|
def linear_warmup_schedule(current_step):
|
||||||
"""Return the LR multiplier during warmup, ramping from `1 / (warmup + 1)` to 1."""
|
|
||||||
if current_step <= 0:
|
if current_step <= 0:
|
||||||
return 1 / (actual_warmup_steps + 1)
|
return 1 / (actual_warmup_steps + 1)
|
||||||
frac = 1 - current_step / actual_warmup_steps
|
frac = 1 - current_step / actual_warmup_steps
|
||||||
return (1 / (actual_warmup_steps + 1) - 1) * frac + 1
|
return (1 / (actual_warmup_steps + 1) - 1) * frac + 1
|
||||||
|
|
||||||
def cosine_decay_schedule(current_step):
|
def cosine_decay_schedule(current_step):
|
||||||
"""Return the LR multiplier during decay, from 1 down to `decay_lr / peak_lr`."""
|
|
||||||
step = min(current_step, actual_decay_steps)
|
step = min(current_step, actual_decay_steps)
|
||||||
cosine_decay = 0.5 * (1 + math.cos(math.pi * step / actual_decay_steps))
|
cosine_decay = 0.5 * (1 + math.cos(math.pi * step / actual_decay_steps))
|
||||||
alpha = self.decay_lr / self.peak_lr
|
alpha = self.decay_lr / self.peak_lr
|
||||||
@@ -276,30 +183,11 @@ class CosineDecayWithWarmupSchedulerConfig(LRSchedulerConfig):
|
|||||||
|
|
||||||
|
|
||||||
def save_scheduler_state(scheduler: LRScheduler, save_dir: Path) -> None:
|
def save_scheduler_state(scheduler: LRScheduler, save_dir: Path) -> None:
|
||||||
"""Save a scheduler's state to disk.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
scheduler (`LRScheduler`):
|
|
||||||
The scheduler whose state to save.
|
|
||||||
save_dir (`Path`):
|
|
||||||
Directory to save the scheduler state.
|
|
||||||
"""
|
|
||||||
state_dict = scheduler.state_dict()
|
state_dict = scheduler.state_dict()
|
||||||
write_json(state_dict, save_dir / SCHEDULER_STATE)
|
write_json(state_dict, save_dir / SCHEDULER_STATE)
|
||||||
|
|
||||||
|
|
||||||
def load_scheduler_state(scheduler: LRScheduler, save_dir: Path) -> LRScheduler:
|
def load_scheduler_state(scheduler: LRScheduler, save_dir: Path) -> LRScheduler:
|
||||||
"""Load a scheduler's state from disk.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
scheduler (`LRScheduler`):
|
|
||||||
The scheduler to load state into.
|
|
||||||
save_dir (`Path`):
|
|
||||||
Directory to load the scheduler state from.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
`LRScheduler`: The same scheduler, with its state loaded.
|
|
||||||
"""
|
|
||||||
state_dict = deserialize_json_into_object(save_dir / SCHEDULER_STATE, scheduler.state_dict())
|
state_dict = deserialize_json_into_object(save_dir / SCHEDULER_STATE, scheduler.state_dict())
|
||||||
scheduler.load_state_dict(state_dict)
|
scheduler.load_state_dict(state_dict)
|
||||||
return scheduler
|
return scheduler
|
||||||
|
|||||||
+59
-23
@@ -13,8 +13,7 @@
|
|||||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
# See the License for the specific language governing permissions and
|
# See the License for the specific language governing permissions and
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
"""
|
"""Actor server runner for distributed HILSerl robot policy training.
|
||||||
Actor server runner for distributed HILSerl robot policy training.
|
|
||||||
|
|
||||||
This script implements the actor component of the distributed HILSerl architecture.
|
This script implements the actor component of the distributed HILSerl architecture.
|
||||||
It executes the policy in the robot environment, collects experience,
|
It executes the policy in the robot environment, collects experience,
|
||||||
@@ -119,6 +118,16 @@ from .train_rl import TrainRLServerPipelineConfig
|
|||||||
|
|
||||||
@parser.wrap()
|
@parser.wrap()
|
||||||
def actor_cli(cfg: TrainRLServerPipelineConfig):
|
def actor_cli(cfg: TrainRLServerPipelineConfig):
|
||||||
|
"""CLI entry point for the HILSerl actor server.
|
||||||
|
|
||||||
|
Connects to the learner server over gRPC, then launches (as threads or processes, depending on
|
||||||
|
`cfg.policy.concurrency.multiprocessing_context`) the background workers that receive updated
|
||||||
|
policy parameters and stream transitions/interactions back to the learner, while running the
|
||||||
|
policy-environment interaction loop (`act_with_policy`) on the main thread/process.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
cfg (`TrainRLServerPipelineConfig`): Parsed from the CLI.
|
||||||
|
"""
|
||||||
# Fail fast with a friendly error if the optional ``hilserl`` extra is missing.
|
# Fail fast with a friendly error if the optional ``hilserl`` extra is missing.
|
||||||
require_package("grpcio", extra="hilserl", import_name="grpc")
|
require_package("grpcio", extra="hilserl", import_name="grpc")
|
||||||
cfg.validate()
|
cfg.validate()
|
||||||
@@ -234,18 +243,19 @@ def act_with_policy(
|
|||||||
transitions_queue: Queue,
|
transitions_queue: Queue,
|
||||||
interactions_queue: Queue,
|
interactions_queue: Queue,
|
||||||
):
|
):
|
||||||
"""
|
"""Executes policy interaction within the environment.
|
||||||
Executes policy interaction within the environment.
|
|
||||||
|
|
||||||
This function rolls out the policy in the environment, collecting interaction data and pushing it to a queue for streaming to the learner.
|
This function rolls out the policy in the environment, collecting interaction data and pushing it to a queue for streaming to the learner.
|
||||||
Once an episode is completed, updated network parameters received from the learner are retrieved from a queue and loaded into the network.
|
Once an episode is completed, updated network parameters received from the learner are retrieved from a queue and loaded into the network.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg: Configuration settings for the interaction process.
|
cfg (`TrainRLServerPipelineConfig`): Training configuration.
|
||||||
shutdown_event: Event to check if the process should shutdown.
|
shutdown_event (`Event`): Set to stop the policy loop.
|
||||||
parameters_queue: Queue to receive updated network parameters from the learner.
|
parameters_queue (`Queue`): Queue of serialized learner weights, drained via
|
||||||
transitions_queue: Queue to send transitions to the learner.
|
`update_policy_parameters`.
|
||||||
interactions_queue: Queue to send interactions to the learner.
|
transitions_queue (`Queue`): Queue transitions are pushed to for streaming to the learner.
|
||||||
|
interactions_queue (`Queue`): Queue interaction messages are pushed to for streaming to the
|
||||||
|
learner.
|
||||||
"""
|
"""
|
||||||
# Initialize logging for multiprocessing
|
# Initialize logging for multiprocessing
|
||||||
if not use_threads(cfg):
|
if not use_threads(cfg):
|
||||||
@@ -440,7 +450,8 @@ def establish_learner_connection(
|
|||||||
Args:
|
Args:
|
||||||
stub (services_pb2_grpc.LearnerServiceStub): The stub to use for the connection.
|
stub (services_pb2_grpc.LearnerServiceStub): The stub to use for the connection.
|
||||||
shutdown_event (Event): The event to check if the connection should be established.
|
shutdown_event (Event): The event to check if the connection should be established.
|
||||||
attempts (int): The number of attempts to establish the connection.
|
attempts (int, *optional*, defaults to 30): The number of attempts to establish the connection.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: True if the connection is established, False otherwise.
|
bool: True if the connection is established, False otherwise.
|
||||||
"""
|
"""
|
||||||
@@ -473,7 +484,6 @@ def learner_service_client(
|
|||||||
Returns:
|
Returns:
|
||||||
tuple[services_pb2_grpc.LearnerServiceStub, grpc.Channel]: The stub and the channel.
|
tuple[services_pb2_grpc.LearnerServiceStub, grpc.Channel]: The stub and the channel.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
channel = grpc.insecure_channel(
|
channel = grpc.insecure_channel(
|
||||||
f"{host}:{port}",
|
f"{host}:{port}",
|
||||||
grpc_channel_options(),
|
grpc_channel_options(),
|
||||||
@@ -496,8 +506,8 @@ def receive_policy(
|
|||||||
cfg (TrainRLServerPipelineConfig): The configuration for the actor.
|
cfg (TrainRLServerPipelineConfig): The configuration for the actor.
|
||||||
parameters_queue (Queue): The queue to receive the parameters.
|
parameters_queue (Queue): The queue to receive the parameters.
|
||||||
shutdown_event (Event): The event to check if the process should shutdown.
|
shutdown_event (Event): The event to check if the process should shutdown.
|
||||||
learner_client (services_pb2_grpc.LearnerServiceStub | None): Optional pre-created stub.
|
learner_client (services_pb2_grpc.LearnerServiceStub | None, *optional*): Optional pre-created stub.
|
||||||
grpc_channel (grpc.Channel | None): Optional pre-created channel.
|
grpc_channel (grpc.Channel | None, *optional*): Optional pre-created channel.
|
||||||
"""
|
"""
|
||||||
logging.info("[ACTOR] Start receiving parameters from the Learner")
|
logging.info("[ACTOR] Start receiving parameters from the Learner")
|
||||||
if not use_threads(cfg):
|
if not use_threads(cfg):
|
||||||
@@ -557,10 +567,9 @@ def send_transitions(
|
|||||||
cfg (TrainRLServerPipelineConfig): The configuration for the actor.
|
cfg (TrainRLServerPipelineConfig): The configuration for the actor.
|
||||||
transitions_queue (Queue): The queue to receive the transitions.
|
transitions_queue (Queue): The queue to receive the transitions.
|
||||||
shutdown_event (Event): The event to check if the process should shutdown.
|
shutdown_event (Event): The event to check if the process should shutdown.
|
||||||
learner_client (services_pb2_grpc.LearnerServiceStub | None): Optional pre-created stub.
|
learner_client (services_pb2_grpc.LearnerServiceStub | None, *optional*): Optional pre-created stub.
|
||||||
grpc_channel (grpc.Channel | None): Optional pre-created channel.
|
grpc_channel (grpc.Channel | None, *optional*): Optional pre-created channel.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
if not use_threads(cfg):
|
if not use_threads(cfg):
|
||||||
# Create a process-specific log file
|
# Create a process-specific log file
|
||||||
log_dir = os.path.join(cfg.output_dir, "logs")
|
log_dir = os.path.join(cfg.output_dir, "logs")
|
||||||
@@ -612,10 +621,9 @@ def send_interactions(
|
|||||||
cfg (TrainRLServerPipelineConfig): The configuration for the actor.
|
cfg (TrainRLServerPipelineConfig): The configuration for the actor.
|
||||||
interactions_queue (Queue): The queue to receive the interactions.
|
interactions_queue (Queue): The queue to receive the interactions.
|
||||||
shutdown_event (Event): The event to check if the process should shutdown.
|
shutdown_event (Event): The event to check if the process should shutdown.
|
||||||
learner_client (services_pb2_grpc.LearnerServiceStub | None): Optional pre-created stub.
|
learner_client (services_pb2_grpc.LearnerServiceStub | None, *optional*): Optional pre-created stub.
|
||||||
grpc_channel (grpc.Channel | None): Optional pre-created channel.
|
grpc_channel (grpc.Channel | None, *optional*): Optional pre-created channel.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
if not use_threads(cfg):
|
if not use_threads(cfg):
|
||||||
# Create a process-specific log file
|
# Create a process-specific log file
|
||||||
log_dir = os.path.join(cfg.output_dir, "logs")
|
log_dir = os.path.join(cfg.output_dir, "logs")
|
||||||
@@ -657,6 +665,17 @@ def transitions_stream(
|
|||||||
transitions_queue: Queue,
|
transitions_queue: Queue,
|
||||||
timeout: float,
|
timeout: float,
|
||||||
) -> "Generator[Any, None, services_pb2.Empty]":
|
) -> "Generator[Any, None, services_pb2.Empty]":
|
||||||
|
"""GRPC client-streaming generator that forwards queued transitions to the learner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
shutdown_event (`Event`): Set to stop streaming and return.
|
||||||
|
transitions_queue (`Queue`): Queue of serialized transition batches, filled by
|
||||||
|
`push_transitions_to_transport_queue`.
|
||||||
|
timeout (`float`): Seconds to wait for a queue item before checking `shutdown_event` again.
|
||||||
|
|
||||||
|
Yields:
|
||||||
|
Chunks of a `services_pb2.Transition` message, produced by `send_bytes_in_chunks`.
|
||||||
|
"""
|
||||||
while not shutdown_event.is_set():
|
while not shutdown_event.is_set():
|
||||||
try:
|
try:
|
||||||
message = transitions_queue.get(block=True, timeout=timeout)
|
message = transitions_queue.get(block=True, timeout=timeout)
|
||||||
@@ -676,6 +695,16 @@ def interactions_stream(
|
|||||||
interactions_queue: Queue,
|
interactions_queue: Queue,
|
||||||
timeout: float,
|
timeout: float,
|
||||||
) -> "Generator[Any, None, services_pb2.Empty]":
|
) -> "Generator[Any, None, services_pb2.Empty]":
|
||||||
|
"""GRPC client-streaming generator that forwards queued interaction messages to the learner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
shutdown_event (`Event`): Set to stop streaming and return.
|
||||||
|
interactions_queue (`Queue`): Queue of serialized interaction messages.
|
||||||
|
timeout (`float`): Seconds to wait for a queue item before checking `shutdown_event` again.
|
||||||
|
|
||||||
|
Yields:
|
||||||
|
Chunks of a `services_pb2.InteractionMessage`, produced by `send_bytes_in_chunks`.
|
||||||
|
"""
|
||||||
while not shutdown_event.is_set():
|
while not shutdown_event.is_set():
|
||||||
try:
|
try:
|
||||||
message = interactions_queue.get(block=True, timeout=timeout)
|
message = interactions_queue.get(block=True, timeout=timeout)
|
||||||
@@ -718,12 +747,11 @@ def update_policy_parameters(algorithm: RLAlgorithm, parameters_queue: Queue, de
|
|||||||
|
|
||||||
|
|
||||||
def push_transitions_to_transport_queue(transitions: list, transitions_queue):
|
def push_transitions_to_transport_queue(transitions: list, transitions_queue):
|
||||||
"""Send transitions to learner in smaller chunks to avoid network issues.
|
"""Move `transitions` to CPU, check for NaNs, and enqueue them for the learner.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
transitions: List of transitions to send
|
transitions (`list`): Transitions to send, as produced by the actor's rollout loop.
|
||||||
message_queue: Queue to send messages to learner
|
transitions_queue (`Queue`): Queue drained by `transitions_stream`.
|
||||||
chunk_size: Size of each chunk to send
|
|
||||||
"""
|
"""
|
||||||
transition_to_send_to_learner = []
|
transition_to_send_to_learner = []
|
||||||
for transition in transitions:
|
for transition in transitions:
|
||||||
@@ -760,6 +788,13 @@ def get_frequency_stats(timer: TimerManager) -> dict[str, float]:
|
|||||||
|
|
||||||
|
|
||||||
def log_policy_frequency_issue(policy_fps: float, cfg: TrainRLServerPipelineConfig, interaction_step: int):
|
def log_policy_frequency_issue(policy_fps: float, cfg: TrainRLServerPipelineConfig, interaction_step: int):
|
||||||
|
"""Log a warning if `policy_fps` is below the environment's target `cfg.env.fps`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
policy_fps (`float`): Measured policy loop frequency.
|
||||||
|
cfg (`TrainRLServerPipelineConfig`): Provides the target `cfg.env.fps` to compare against.
|
||||||
|
interaction_step (`int`): Current interaction step, included in the warning message.
|
||||||
|
"""
|
||||||
if policy_fps < cfg.env.fps:
|
if policy_fps < cfg.env.fps:
|
||||||
logging.warning(
|
logging.warning(
|
||||||
f"[ACTOR] Policy FPS {policy_fps:.1f} below required {cfg.env.fps} at step {interaction_step}"
|
f"[ACTOR] Policy FPS {policy_fps:.1f} below required {cfg.env.fps} at step {interaction_step}"
|
||||||
@@ -767,6 +802,7 @@ def log_policy_frequency_issue(policy_fps: float, cfg: TrainRLServerPipelineConf
|
|||||||
|
|
||||||
|
|
||||||
def use_threads(cfg: TrainRLServerPipelineConfig) -> bool:
|
def use_threads(cfg: TrainRLServerPipelineConfig) -> bool:
|
||||||
|
"""Whether the actor's background workers should run as threads instead of processes."""
|
||||||
return cfg.policy.concurrency.actor == "threads"
|
return cfg.policy.concurrency.actor == "threads"
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -101,6 +101,7 @@ class RLAlgorithm(HubMixin, abc.ABC):
|
|||||||
|
|
||||||
@optimization_step.setter
|
@optimization_step.setter
|
||||||
def optimization_step(self, value: int) -> None:
|
def optimization_step(self, value: int) -> None:
|
||||||
|
"""Set the current learner optimization step."""
|
||||||
self._optimization_step = int(value)
|
self._optimization_step = int(value)
|
||||||
|
|
||||||
def get_weights(self) -> dict[str, Any]:
|
def get_weights(self) -> dict[str, Any]:
|
||||||
|
|||||||
@@ -45,7 +45,6 @@ class TrainingStats:
|
|||||||
|
|
||||||
def to_log_dict(self) -> dict[str, float]:
|
def to_log_dict(self) -> dict[str, float]:
|
||||||
"""Flatten all stats into a single dict for logging."""
|
"""Flatten all stats into a single dict for logging."""
|
||||||
|
|
||||||
d: dict[str, float] = {}
|
d: dict[str, float] = {}
|
||||||
for name, val in self.losses.items():
|
for name, val in self.losses.items():
|
||||||
d[name] = val
|
d[name] = val
|
||||||
@@ -98,6 +97,35 @@ class RLAlgorithmConfig(draccus.ChoiceRegistry, HubMixin, abc.ABC):
|
|||||||
revision: str | None = None,
|
revision: str | None = None,
|
||||||
**algo_kwargs: Any,
|
**algo_kwargs: Any,
|
||||||
) -> T:
|
) -> T:
|
||||||
|
"""Load an algorithm config from a local directory or the Hugging Face Hub.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
pretrained_name_or_path (`str | Path`):
|
||||||
|
Local directory containing `config.json`, or a Hub repo id.
|
||||||
|
force_download (`bool`, *optional*, defaults to `False`):
|
||||||
|
Whether to force re-download the config even if it's cached.
|
||||||
|
resume_download (`bool | None`, *optional*):
|
||||||
|
Whether to resume an interrupted download.
|
||||||
|
proxies (`dict[Any, Any] | None`, *optional*):
|
||||||
|
Proxies to use for the download request.
|
||||||
|
token (`str | bool | None`, *optional*):
|
||||||
|
Hugging Face Hub authentication token.
|
||||||
|
cache_dir (`str | Path | None`, *optional*):
|
||||||
|
Directory to cache the downloaded config in.
|
||||||
|
local_files_only (`bool`, *optional*, defaults to `False`):
|
||||||
|
Whether to only look for files locally, without querying the Hub.
|
||||||
|
revision (`str | None`, *optional*):
|
||||||
|
Hub revision (branch, tag, or commit hash) to load from.
|
||||||
|
**algo_kwargs: Attribute overrides applied to the loaded config instance.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
RLAlgorithmConfig: The loaded config, as the concrete registered subclass.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
FileNotFoundError: If no `config.json` is found locally or on the Hub.
|
||||||
|
TypeError: If loaded via a specific subclass but the config's registered type doesn't
|
||||||
|
match it.
|
||||||
|
"""
|
||||||
model_id = str(pretrained_name_or_path)
|
model_id = str(pretrained_name_or_path)
|
||||||
config_file: str | None = None
|
config_file: str | None = None
|
||||||
if Path(model_id).is_dir():
|
if Path(model_id).is_dir():
|
||||||
|
|||||||
@@ -24,8 +24,8 @@ def make_algorithm_config(algorithm_type: str, **kwargs) -> RLAlgorithmConfig:
|
|||||||
"""Instantiate an `RLAlgorithmConfig` from its registered type name.
|
"""Instantiate an `RLAlgorithmConfig` from its registered type name.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
algorithm_type: Registry key of the algorithm (e.g. ``"sac"``).
|
algorithm_type (`str`): Registry key of the algorithm (e.g. `"sac"`).
|
||||||
**kwargs: Keyword arguments forwarded to the config class constructor.
|
kwargs (`Any`, *optional*): Keyword arguments forwarded to the config class constructor.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
An instance of the matching ``RLAlgorithmConfig`` subclass.
|
An instance of the matching ``RLAlgorithmConfig`` subclass.
|
||||||
@@ -44,14 +44,13 @@ def make_algorithm_config(algorithm_type: str, **kwargs) -> RLAlgorithmConfig:
|
|||||||
|
|
||||||
|
|
||||||
def get_algorithm_class(name: str) -> type[RLAlgorithm]:
|
def get_algorithm_class(name: str) -> type[RLAlgorithm]:
|
||||||
"""
|
"""Retrieves an RL algorithm class by its registered name.
|
||||||
Retrieves an RL algorithm class by its registered name.
|
|
||||||
|
|
||||||
This function uses dynamic imports to avoid loading all algorithm classes into
|
This function uses dynamic imports to avoid loading all algorithm classes into
|
||||||
memory at once, improving startup time and reducing dependencies.
|
memory at once, improving startup time and reducing dependencies.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
name: The name of the algorithm. Supported names are "sac".
|
name (`str`): The name of the algorithm. Supported names are "sac".
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
The algorithm class corresponding to the given name.
|
The algorithm class corresponding to the given name.
|
||||||
@@ -70,8 +69,7 @@ def get_algorithm_class(name: str) -> type[RLAlgorithm]:
|
|||||||
|
|
||||||
|
|
||||||
def make_algorithm(cfg: RLAlgorithmConfig, policy: torch.nn.Module) -> RLAlgorithm:
|
def make_algorithm(cfg: RLAlgorithmConfig, policy: torch.nn.Module) -> RLAlgorithm:
|
||||||
"""
|
"""Instantiate an RL algorithm.
|
||||||
Instantiate an RL algorithm.
|
|
||||||
|
|
||||||
This factory function looks up the :class:`RLAlgorithm` subclass that matches
|
This factory function looks up the :class:`RLAlgorithm` subclass that matches
|
||||||
``cfg.type`` and instantiates it with the provided policy. It also enforces
|
``cfg.type`` and instantiates it with the provided policy. It also enforces
|
||||||
@@ -79,8 +77,8 @@ def make_algorithm(cfg: RLAlgorithmConfig, policy: torch.nn.Module) -> RLAlgorit
|
|||||||
normally handled by :meth:`TrainRLServerPipelineConfig.validate`).
|
normally handled by :meth:`TrainRLServerPipelineConfig.validate`).
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg: The algorithm configuration. Must have ``policy_config`` set.
|
cfg (`RLAlgorithmConfig`): The algorithm configuration. Must have `policy_config` set.
|
||||||
policy: The policy module the algorithm will train.
|
policy (`torch.nn.Module`): The policy module the algorithm will train.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
An instantiated :class:`RLAlgorithm`.
|
An instantiated :class:`RLAlgorithm`.
|
||||||
|
|||||||
@@ -39,52 +39,73 @@ class SACAlgorithmConfig(RLAlgorithmConfig):
|
|||||||
update loop. The policy-side (actor + observation encoder) lives in
|
update loop. The policy-side (actor + observation encoder) lives in
|
||||||
:class:`~lerobot.policies.gaussian_actor.GaussianActorConfig` and is
|
:class:`~lerobot.policies.gaussian_actor.GaussianActorConfig` and is
|
||||||
referenced via :attr:`policy_config`.
|
referenced via :attr:`policy_config`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
actor_lr (`float`, *optional*, defaults to 0.0003):
|
||||||
|
Learning rate for the actor network.
|
||||||
|
critic_lr (`float`, *optional*, defaults to 0.0003):
|
||||||
|
Learning rate for the critic network.
|
||||||
|
temperature_lr (`float`, *optional*, defaults to 0.0003):
|
||||||
|
Learning rate for the temperature parameter.
|
||||||
|
discount (`float`, *optional*, defaults to 0.99):
|
||||||
|
Discount factor for the Bellman update.
|
||||||
|
use_backup_entropy (`bool`, *optional*, defaults to `True`):
|
||||||
|
Whether to use backup entropy in the Bellman target.
|
||||||
|
critic_target_update_weight (`float`, *optional*, defaults to 0.005):
|
||||||
|
Polyak-averaging weight for the critic target update.
|
||||||
|
num_critics (`int`, *optional*, defaults to 2):
|
||||||
|
Number of critics in the ensemble.
|
||||||
|
num_subsample_critics (`int | None`, *optional*):
|
||||||
|
Number of critics to subsample from the ensemble for each Bellman target computation.
|
||||||
|
`None` uses the full ensemble.
|
||||||
|
critic_network_kwargs (`CriticNetworkConfig`, *optional*):
|
||||||
|
Configuration for the (continuous-action) critic network architecture.
|
||||||
|
discrete_critic_network_kwargs (`CriticNetworkConfig`, *optional*):
|
||||||
|
Configuration for the discrete-action critic network architecture.
|
||||||
|
temperature_init (`float`, *optional*, defaults to 1.0):
|
||||||
|
Initial value of the entropy temperature.
|
||||||
|
target_entropy (`float | None`, *optional*):
|
||||||
|
Target entropy for automatic temperature tuning. If `None`, defaults to `-|A|/2` where
|
||||||
|
`|A|` is the total action dimension (continuous + 1 if there is a discrete action head).
|
||||||
|
utd_ratio (`int`, *optional*, defaults to 1):
|
||||||
|
Update-to-data ratio. Set to `>1` to enable extra critic updates per env step.
|
||||||
|
policy_update_freq (`int`, *optional*, defaults to 1):
|
||||||
|
Frequency of policy updates, in units of critic updates.
|
||||||
|
grad_clip_norm (`float`, *optional*, defaults to 40.0):
|
||||||
|
Gradient-clipping norm applied during optimization.
|
||||||
|
use_torch_compile (`bool`, *optional*, defaults to `False`):
|
||||||
|
Whether to `torch.compile` the algorithm's forward passes. Currently disabled by default.
|
||||||
|
policy_config (`PreTrainedConfig | None`, *optional*):
|
||||||
|
The policy (actor) config this algorithm trains. Populated via `from_policy_config` or by
|
||||||
|
`TrainRLServerPipelineConfig.validate` before the algorithm is constructed.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# Optimizer learning rates
|
# Optimizer learning rates
|
||||||
# Learning rate for the actor network
|
|
||||||
actor_lr: float = 3e-4
|
actor_lr: float = 3e-4
|
||||||
# Learning rate for the critic network
|
|
||||||
critic_lr: float = 3e-4
|
critic_lr: float = 3e-4
|
||||||
# Learning rate for the temperature parameter
|
|
||||||
temperature_lr: float = 3e-4
|
temperature_lr: float = 3e-4
|
||||||
|
|
||||||
# Bellman update
|
# Bellman update
|
||||||
# Discount factor for the SAC algorithm
|
|
||||||
discount: float = 0.99
|
discount: float = 0.99
|
||||||
# Whether to use backup entropy for the SAC algorithm
|
|
||||||
use_backup_entropy: bool = True
|
use_backup_entropy: bool = True
|
||||||
# Weight for the critic target update
|
|
||||||
critic_target_update_weight: float = 0.005
|
critic_target_update_weight: float = 0.005
|
||||||
|
|
||||||
# Critic ensemble
|
# Critic ensemble
|
||||||
# Number of critics in the ensemble
|
|
||||||
num_critics: int = 2
|
num_critics: int = 2
|
||||||
# Number of subsampled critics for training
|
|
||||||
num_subsample_critics: int | None = None
|
num_subsample_critics: int | None = None
|
||||||
# Configuration for the critic network architecture
|
|
||||||
critic_network_kwargs: CriticNetworkConfig = field(default_factory=CriticNetworkConfig)
|
critic_network_kwargs: CriticNetworkConfig = field(default_factory=CriticNetworkConfig)
|
||||||
# Configuration for the discrete critic network
|
|
||||||
discrete_critic_network_kwargs: CriticNetworkConfig = field(default_factory=CriticNetworkConfig)
|
discrete_critic_network_kwargs: CriticNetworkConfig = field(default_factory=CriticNetworkConfig)
|
||||||
|
|
||||||
# Temperature / entropy
|
# Temperature / entropy
|
||||||
# Initial temperature value
|
|
||||||
temperature_init: float = 1.0
|
temperature_init: float = 1.0
|
||||||
# Target entropy for automatic temperature tuning. If ``None``, defaults to
|
|
||||||
# ``-|A|/2`` where ``|A|`` is the total action dimension (continuous + 1 if
|
|
||||||
# there is a discrete action head).
|
|
||||||
target_entropy: float | None = None
|
target_entropy: float | None = None
|
||||||
|
|
||||||
# Update loop
|
# Update loop
|
||||||
# Update-to-data ratio. Set to >1 to enable extra critic updates per env step.
|
|
||||||
utd_ratio: int = 1
|
utd_ratio: int = 1
|
||||||
# Frequency of policy updates
|
|
||||||
policy_update_freq: int = 1
|
policy_update_freq: int = 1
|
||||||
# Gradient clipping norm for the SAC algorithm
|
|
||||||
grad_clip_norm: float = 40.0
|
grad_clip_norm: float = 40.0
|
||||||
|
|
||||||
# Optimizations
|
# Optimizations
|
||||||
# torch.compile is currently disabled by default
|
|
||||||
use_torch_compile: bool = False
|
use_torch_compile: bool = False
|
||||||
|
|
||||||
# Policy config
|
# Policy config
|
||||||
|
|||||||
@@ -55,6 +55,15 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
policy: GaussianActorPolicy,
|
policy: GaussianActorPolicy,
|
||||||
config: SACAlgorithmConfig,
|
config: SACAlgorithmConfig,
|
||||||
):
|
):
|
||||||
|
"""Build the critic ensemble, target networks, and temperature from `config`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
policy (`GaussianActorPolicy`):
|
||||||
|
The actor policy this algorithm trains. Its observation encoder is shared with the
|
||||||
|
critics.
|
||||||
|
config (`SACAlgorithmConfig`):
|
||||||
|
Algorithm configuration.
|
||||||
|
"""
|
||||||
self.config = config
|
self.config = config
|
||||||
self.policy_config = config.policy_config
|
self.policy_config = config.policy_config
|
||||||
self.policy = policy
|
self.policy = policy
|
||||||
@@ -144,17 +153,18 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
use_target: bool = False,
|
use_target: bool = False,
|
||||||
observation_features: Tensor | None = None,
|
observation_features: Tensor | None = None,
|
||||||
) -> Tensor:
|
) -> Tensor:
|
||||||
"""Forward pass through a critic network ensemble
|
"""Forward pass through a critic network ensemble.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
observations: Dictionary of observations
|
observations: Dictionary of observations
|
||||||
actions: Action tensor
|
actions: Action tensor
|
||||||
use_target: If True, use target critics, otherwise use ensemble critics
|
use_target: If True, use target critics, otherwise use ensemble critics
|
||||||
|
observation_features: Optional pre-computed observation features to avoid recomputing
|
||||||
|
encoder output
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tensor of Q-values from all critics
|
Tensor of Q-values from all critics
|
||||||
"""
|
"""
|
||||||
|
|
||||||
critics = self.critic_target if use_target else self.critic_ensemble
|
critics = self.critic_target if use_target else self.critic_ensemble
|
||||||
q_values = critics(observations, actions, observation_features)
|
q_values = critics(observations, actions, observation_features)
|
||||||
return q_values
|
return q_values
|
||||||
@@ -162,7 +172,7 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
def _discrete_critic_forward(
|
def _discrete_critic_forward(
|
||||||
self, observations, use_target=False, observation_features=None
|
self, observations, use_target=False, observation_features=None
|
||||||
) -> torch.Tensor:
|
) -> torch.Tensor:
|
||||||
"""Forward pass through a discrete critic network
|
"""Forward pass through a discrete critic network.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
observations: Dictionary of observations
|
observations: Dictionary of observations
|
||||||
@@ -408,7 +418,7 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
return actor_loss
|
return actor_loss
|
||||||
|
|
||||||
def _compute_loss_temperature(self, batch: dict[str, Any]) -> Tensor:
|
def _compute_loss_temperature(self, batch: dict[str, Any]) -> Tensor:
|
||||||
"""Compute the temperature loss"""
|
"""Compute the temperature loss."""
|
||||||
observations = batch["state"]
|
observations = batch["state"]
|
||||||
observation_features = batch.get("observation_feature")
|
observation_features = batch.get("observation_feature")
|
||||||
|
|
||||||
@@ -420,7 +430,7 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
return temperature_loss
|
return temperature_loss
|
||||||
|
|
||||||
def _update_target_networks(self) -> None:
|
def _update_target_networks(self) -> None:
|
||||||
"""Update target networks with exponential moving average"""
|
"""Update target networks with exponential moving average."""
|
||||||
for target_p, p in zip(
|
for target_p, p in zip(
|
||||||
self.critic_target.parameters(), self.critic_ensemble.parameters(), strict=True
|
self.critic_target.parameters(), self.critic_ensemble.parameters(), strict=True
|
||||||
):
|
):
|
||||||
@@ -461,8 +471,7 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
return forward_batch
|
return forward_batch
|
||||||
|
|
||||||
def make_optimizers_and_scheduler(self) -> dict[str, Optimizer]:
|
def make_optimizers_and_scheduler(self) -> dict[str, Optimizer]:
|
||||||
"""
|
"""Creates and returns optimizers for the actor, critic, and temperature components of a reinforcement learning policy.
|
||||||
Creates and returns optimizers for the actor, critic, and temperature components of a reinforcement learning policy.
|
|
||||||
|
|
||||||
This function sets up Adam optimizers for:
|
This function sets up Adam optimizers for:
|
||||||
- The **actor network**, ensuring that only relevant parameters are optimized.
|
- The **actor network**, ensuring that only relevant parameters are optimized.
|
||||||
@@ -471,7 +480,7 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
|
|
||||||
It also initializes a learning rate scheduler, though currently, it is set to `None`.
|
It also initializes a learning rate scheduler, though currently, it is set to `None`.
|
||||||
|
|
||||||
NOTE:
|
Note:
|
||||||
- If the encoder is shared, its parameters are excluded from the actor's optimization process.
|
- If the encoder is shared, its parameters are excluded from the actor's optimization process.
|
||||||
- The policy's log temperature (`log_alpha`) is wrapped in a list to ensure proper optimization as a standalone tensor.
|
- The policy's log temperature (`log_alpha`) is wrapped in a list to ensure proper optimization as a standalone tensor.
|
||||||
|
|
||||||
@@ -496,6 +505,7 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
return self.optimizers
|
return self.optimizers
|
||||||
|
|
||||||
def get_optimizers(self) -> dict[str, Optimizer]:
|
def get_optimizers(self) -> dict[str, Optimizer]:
|
||||||
|
"""See [`~rl.algorithms.RLAlgorithm.get_optimizers`]."""
|
||||||
return self.optimizers
|
return self.optimizers
|
||||||
|
|
||||||
def get_weights(self) -> dict[str, Any]:
|
def get_weights(self) -> dict[str, Any]:
|
||||||
@@ -560,20 +570,18 @@ class SACAlgorithm(RLAlgorithm):
|
|||||||
def get_observation_features(
|
def get_observation_features(
|
||||||
self, observations: Tensor, next_observations: Tensor
|
self, observations: Tensor, next_observations: Tensor
|
||||||
) -> tuple[Tensor | None, Tensor | None]:
|
) -> tuple[Tensor | None, Tensor | None]:
|
||||||
"""
|
"""Get observation features from the policy encoder, acting as a cache.
|
||||||
Get observation features from the policy encoder. It act as cache for the observation features.
|
|
||||||
when the encoder is frozen, the observation features are not updated.
|
When the encoder is frozen, the observation features are not updated, so we can save compute
|
||||||
We can save compute by caching the observation features.
|
by caching them here instead of recomputing on every critic/actor forward pass.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
policy: The policy model
|
|
||||||
observations: The current observations
|
observations: The current observations
|
||||||
next_observations: The next observations
|
next_observations: The next observations
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
tuple: observation_features, next_observation_features
|
tuple: observation_features, next_observation_features
|
||||||
"""
|
"""
|
||||||
|
|
||||||
if self.policy.config.vision_encoder_name is None or not self.policy.config.freeze_vision_encoder:
|
if self.policy.config.vision_encoder_name is None or not self.policy.config.freeze_vision_encoder:
|
||||||
return None, None
|
return None, None
|
||||||
|
|
||||||
@@ -595,6 +603,8 @@ def _split_prefix(state: dict[str, torch.Tensor], prefix: str) -> dict[str, torc
|
|||||||
|
|
||||||
|
|
||||||
class CriticHead(nn.Module):
|
class CriticHead(nn.Module):
|
||||||
|
"""A single Q-value head: an MLP followed by a scalar linear output layer."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
input_dim: int,
|
input_dim: int,
|
||||||
@@ -605,6 +615,23 @@ class CriticHead(nn.Module):
|
|||||||
init_final: float | None = None,
|
init_final: float | None = None,
|
||||||
final_activation: Callable[[torch.Tensor], torch.Tensor] | str | None = None,
|
final_activation: Callable[[torch.Tensor], torch.Tensor] | str | None = None,
|
||||||
):
|
):
|
||||||
|
"""Build the MLP trunk and scalar output layer.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
input_dim (`int`): Dimension of the concatenated observation-encoding + action input.
|
||||||
|
hidden_dims (`list[int]`): Hidden layer widths of the MLP trunk.
|
||||||
|
activations (`Callable[[torch.Tensor], torch.Tensor] | str`, *optional*, defaults to `nn.SiLU()`):
|
||||||
|
Activation used between hidden layers.
|
||||||
|
activate_final (`bool`, *optional*, defaults to `False`): Whether to apply `activations`
|
||||||
|
after the last hidden layer.
|
||||||
|
dropout_rate (`float | None`, *optional*): Dropout probability applied between hidden
|
||||||
|
layers. `None` disables dropout.
|
||||||
|
init_final (`float | None`, *optional*): When set, the output layer's weight and bias are
|
||||||
|
initialized uniformly in `[-init_final, init_final]` instead of the default
|
||||||
|
orthogonal initialization.
|
||||||
|
final_activation (`Callable[[torch.Tensor], torch.Tensor] | str | None`, *optional*):
|
||||||
|
Activation applied after the MLP trunk's last hidden layer, before the output layer.
|
||||||
|
"""
|
||||||
super().__init__()
|
super().__init__()
|
||||||
self.net = MLP(
|
self.net = MLP(
|
||||||
input_dim=input_dim,
|
input_dim=input_dim,
|
||||||
@@ -622,17 +649,17 @@ class CriticHead(nn.Module):
|
|||||||
orthogonal_init()(self.output_layer.weight)
|
orthogonal_init()(self.output_layer.weight)
|
||||||
|
|
||||||
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
||||||
|
"""Compute the scalar Q-value for `x` (a concatenated observation-encoding + action tensor)."""
|
||||||
return self.output_layer(self.net(x))
|
return self.output_layer(self.net(x))
|
||||||
|
|
||||||
|
|
||||||
class CriticEnsemble(nn.Module):
|
class CriticEnsemble(nn.Module):
|
||||||
"""
|
"""CriticEnsemble wraps multiple CriticHead modules into an ensemble.
|
||||||
CriticEnsemble wraps multiple CriticHead modules into an ensemble.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
encoder (GaussianActorObservationEncoder): encoder for observations.
|
encoder (GaussianActorObservationEncoder): encoder for observations.
|
||||||
ensemble (List[CriticHead]): list of critic heads.
|
ensemble (List[CriticHead]): list of critic heads.
|
||||||
init_final (float | None): optional initializer scale for final layers.
|
init_final (float | None, *optional*): optional initializer scale for final layers.
|
||||||
|
|
||||||
Forward returns a tensor of shape (num_critics, batch_size) containing Q-values.
|
Forward returns a tensor of shape (num_critics, batch_size) containing Q-values.
|
||||||
"""
|
"""
|
||||||
@@ -643,6 +670,14 @@ class CriticEnsemble(nn.Module):
|
|||||||
ensemble: list[CriticHead],
|
ensemble: list[CriticHead],
|
||||||
init_final: float | None = None,
|
init_final: float | None = None,
|
||||||
):
|
):
|
||||||
|
"""Wrap `ensemble` behind the shared `encoder`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
encoder (`GaussianActorObservationEncoder`): Shared observation encoder for all critics.
|
||||||
|
ensemble (`list[CriticHead]`): The critic heads making up the ensemble.
|
||||||
|
init_final (`float | None`, *optional*): Stored for introspection; each `CriticHead` is
|
||||||
|
already initialized with it before being passed in here.
|
||||||
|
"""
|
||||||
super().__init__()
|
super().__init__()
|
||||||
self.encoder = encoder
|
self.encoder = encoder
|
||||||
self.init_final = init_final
|
self.init_final = init_final
|
||||||
@@ -654,6 +689,19 @@ class CriticEnsemble(nn.Module):
|
|||||||
actions: torch.Tensor,
|
actions: torch.Tensor,
|
||||||
observation_features: torch.Tensor | None = None,
|
observation_features: torch.Tensor | None = None,
|
||||||
) -> torch.Tensor:
|
) -> torch.Tensor:
|
||||||
|
"""Encode `observations` and return each ensemble member's Q-value for `actions`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
observations (`dict[str, torch.Tensor]`): Raw observation tensors, moved to the module's
|
||||||
|
device.
|
||||||
|
actions (`torch.Tensor`): Action tensor to evaluate.
|
||||||
|
observation_features (`torch.Tensor | None`, *optional*): Pre-computed encoder output,
|
||||||
|
e.g. from `SACAlgorithm.get_observation_features`. Bypasses re-encoding when the
|
||||||
|
vision encoder is frozen.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
torch.Tensor: Q-values of shape `(num_critics, batch_size)`.
|
||||||
|
"""
|
||||||
device = get_device_from_parameters(self)
|
device = get_device_from_parameters(self)
|
||||||
# Move each tensor in observations to device
|
# Move each tensor in observations to device
|
||||||
observations = {k: v.to(device) for k, v in observations.items()}
|
observations = {k: v.to(device) for k, v in observations.items()}
|
||||||
|
|||||||
+34
-27
@@ -30,6 +30,19 @@ from lerobot.utils.transition import Transition
|
|||||||
|
|
||||||
|
|
||||||
class BatchTransition(TypedDict):
|
class BatchTransition(TypedDict):
|
||||||
|
"""A batch of transitions sampled from a `ReplayBuffer`.
|
||||||
|
|
||||||
|
**Attributes**:
|
||||||
|
- **state** (`dict[str, torch.Tensor]`) -- Batched observation tensors at time `t`.
|
||||||
|
- **action** (`torch.Tensor`) -- Batched actions taken at time `t`.
|
||||||
|
- **reward** (`torch.Tensor`) -- Batched rewards received after `action`.
|
||||||
|
- **next_state** (`dict[str, torch.Tensor]`) -- Batched observation tensors at time `t+1`.
|
||||||
|
- **done** (`torch.Tensor`) -- Batched episode-termination flags.
|
||||||
|
- **truncated** (`torch.Tensor`) -- Batched episode-truncation flags.
|
||||||
|
- **complementary_info** (`dict[str, torch.Tensor | float | int] | None`) -- Optional extra
|
||||||
|
per-transition data (e.g. intervention flags), when present in the underlying dataset.
|
||||||
|
"""
|
||||||
|
|
||||||
state: dict[str, torch.Tensor]
|
state: dict[str, torch.Tensor]
|
||||||
action: torch.Tensor
|
action: torch.Tensor
|
||||||
reward: torch.Tensor
|
reward: torch.Tensor
|
||||||
@@ -40,10 +53,7 @@ class BatchTransition(TypedDict):
|
|||||||
|
|
||||||
|
|
||||||
def random_crop_vectorized(images: torch.Tensor, output_size: tuple) -> torch.Tensor:
|
def random_crop_vectorized(images: torch.Tensor, output_size: tuple) -> torch.Tensor:
|
||||||
"""
|
"""Perform a per-image random crop over a batch of images in a vectorized way."""
|
||||||
Perform a per-image random crop over a batch of images in a vectorized way.
|
|
||||||
(Same as shown previously.)
|
|
||||||
"""
|
|
||||||
B, C, H, W = images.shape # noqa: N806
|
B, C, H, W = images.shape # noqa: N806
|
||||||
crop_h, crop_w = output_size
|
crop_h, crop_w = output_size
|
||||||
|
|
||||||
@@ -72,13 +82,15 @@ def random_crop_vectorized(images: torch.Tensor, output_size: tuple) -> torch.Te
|
|||||||
|
|
||||||
|
|
||||||
def random_shift(images: torch.Tensor, pad: int = 4):
|
def random_shift(images: torch.Tensor, pad: int = 4):
|
||||||
"""Vectorized random shift, imgs: (B,C,H,W), pad: #pixels"""
|
"""Vectorized random shift. `images` has shape `(B, C, H, W)`; `pad` is the shift range in pixels."""
|
||||||
_, _, h, w = images.shape
|
_, _, h, w = images.shape
|
||||||
images = F.pad(input=images, pad=(pad, pad, pad, pad), mode="replicate")
|
images = F.pad(input=images, pad=(pad, pad, pad, pad), mode="replicate")
|
||||||
return random_crop_vectorized(images=images, output_size=(h, w))
|
return random_crop_vectorized(images=images, output_size=(h, w))
|
||||||
|
|
||||||
|
|
||||||
class ReplayBuffer:
|
class ReplayBuffer:
|
||||||
|
"""In-memory replay buffer of `Transition`s, sampled in batches for off-policy RL training."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
capacity: int,
|
capacity: int,
|
||||||
@@ -89,11 +101,12 @@ class ReplayBuffer:
|
|||||||
storage_device: str = "cpu",
|
storage_device: str = "cpu",
|
||||||
optimize_memory: bool = False,
|
optimize_memory: bool = False,
|
||||||
):
|
):
|
||||||
"""
|
"""Replay buffer for storing transitions.
|
||||||
Replay buffer for storing transitions.
|
|
||||||
It will allocate tensors on the specified device, when the first transition is added.
|
It will allocate tensors on the specified device, when the first transition is added.
|
||||||
NOTE: If you encounter memory issues, you can try to use the `optimize_memory` flag to save memory or
|
NOTE: If you encounter memory issues, you can try to use the `optimize_memory` flag to save memory or
|
||||||
and use the `storage_device` flag to store the buffer on a different device.
|
and use the `storage_device` flag to store the buffer on a different device.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
capacity (int): Maximum number of transitions to store in the buffer.
|
capacity (int): Maximum number of transitions to store in the buffer.
|
||||||
device (str): The device where the tensors will be moved when sampling ("cuda:0" or "cpu").
|
device (str): The device where the tensors will be moved when sampling ("cuda:0" or "cpu").
|
||||||
@@ -187,6 +200,7 @@ class ReplayBuffer:
|
|||||||
self.initialized = True
|
self.initialized = True
|
||||||
|
|
||||||
def __len__(self):
|
def __len__(self):
|
||||||
|
"""Number of transitions currently stored in the buffer."""
|
||||||
return self.size
|
return self.size
|
||||||
|
|
||||||
def add(
|
def add(
|
||||||
@@ -305,8 +319,8 @@ class ReplayBuffer:
|
|||||||
async_prefetch: bool = True,
|
async_prefetch: bool = True,
|
||||||
queue_size: int = 2,
|
queue_size: int = 2,
|
||||||
):
|
):
|
||||||
"""
|
"""Creates an infinite iterator that yields batches of transitions.
|
||||||
Creates an infinite iterator that yields batches of transitions.
|
|
||||||
Will automatically restart when internal iterator is exhausted.
|
Will automatically restart when internal iterator is exhausted.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
@@ -329,10 +343,9 @@ class ReplayBuffer:
|
|||||||
yield from iterator
|
yield from iterator
|
||||||
|
|
||||||
def _get_async_iterator(self, batch_size: int, queue_size: int = 2):
|
def _get_async_iterator(self, batch_size: int, queue_size: int = 2):
|
||||||
"""
|
"""Create an iterator that continuously yields prefetched batches in a background thread.
|
||||||
Create an iterator that continuously yields prefetched batches in a
|
|
||||||
background thread. The design is intentionally simple and avoids busy
|
The design is intentionally simple and avoids busy waiting / complex state management.
|
||||||
waiting / complex state management.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
batch_size (int): Size of batches to sample.
|
batch_size (int): Size of batches to sample.
|
||||||
@@ -383,8 +396,7 @@ class ReplayBuffer:
|
|||||||
producer_thread.join(timeout=1.0)
|
producer_thread.join(timeout=1.0)
|
||||||
|
|
||||||
def _get_naive_iterator(self, batch_size: int, queue_size: int = 2):
|
def _get_naive_iterator(self, batch_size: int, queue_size: int = 2):
|
||||||
"""
|
"""Creates a simple non-threaded iterator that yields batches.
|
||||||
Creates a simple non-threaded iterator that yields batches.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
batch_size (int): Size of batches to sample
|
batch_size (int): Size of batches to sample
|
||||||
@@ -398,6 +410,7 @@ class ReplayBuffer:
|
|||||||
queue = collections.deque()
|
queue = collections.deque()
|
||||||
|
|
||||||
def enqueue(n):
|
def enqueue(n):
|
||||||
|
"""Sample `n` more batches and append them to `queue`."""
|
||||||
for _ in range(n):
|
for _ in range(n):
|
||||||
data = self.sample(batch_size)
|
data = self.sample(batch_size)
|
||||||
queue.append(data)
|
queue.append(data)
|
||||||
@@ -419,8 +432,7 @@ class ReplayBuffer:
|
|||||||
storage_device: str = "cpu",
|
storage_device: str = "cpu",
|
||||||
optimize_memory: bool = False,
|
optimize_memory: bool = False,
|
||||||
) -> "ReplayBuffer":
|
) -> "ReplayBuffer":
|
||||||
"""
|
"""Convert a LeRobotDataset into a ReplayBuffer.
|
||||||
Convert a LeRobotDataset into a ReplayBuffer.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
lerobot_dataset (LeRobotDataset): The dataset to convert.
|
lerobot_dataset (LeRobotDataset): The dataset to convert.
|
||||||
@@ -509,9 +521,7 @@ class ReplayBuffer:
|
|||||||
root=None,
|
root=None,
|
||||||
task_name="from_replay_buffer",
|
task_name="from_replay_buffer",
|
||||||
) -> LeRobotDataset:
|
) -> LeRobotDataset:
|
||||||
"""
|
"""Converts all transitions in this ReplayBuffer into a single LeRobotDataset object."""
|
||||||
Converts all transitions in this ReplayBuffer into a single LeRobotDataset object.
|
|
||||||
"""
|
|
||||||
if self.size == 0:
|
if self.size == 0:
|
||||||
raise ValueError("The replay buffer is empty. Cannot convert to a dataset.")
|
raise ValueError("The replay buffer is empty. Cannot convert to a dataset.")
|
||||||
|
|
||||||
@@ -612,8 +622,7 @@ class ReplayBuffer:
|
|||||||
dataset: LeRobotDataset,
|
dataset: LeRobotDataset,
|
||||||
state_keys: Sequence[str] | None = None,
|
state_keys: Sequence[str] | None = None,
|
||||||
) -> list[Transition]:
|
) -> list[Transition]:
|
||||||
"""
|
"""Convert a LeRobotDataset into a list of RL (s, a, r, s', done) transitions.
|
||||||
Convert a LeRobotDataset into a list of RL (s, a, r, s', done) transitions.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
dataset (LeRobotDataset):
|
dataset (LeRobotDataset):
|
||||||
@@ -733,12 +742,11 @@ class ReplayBuffer:
|
|||||||
|
|
||||||
# Utility function to guess shapes/dtypes from a tensor
|
# Utility function to guess shapes/dtypes from a tensor
|
||||||
def guess_feature_info(t, name: str):
|
def guess_feature_info(t, name: str):
|
||||||
"""
|
"""Return a dictionary with the 'dtype' and 'shape' for a given tensor or scalar value.
|
||||||
Return a dictionary with the 'dtype' and 'shape' for a given tensor or scalar value.
|
|
||||||
If it looks like a 3D (C,H,W) shape, we might consider it an 'image'.
|
If it looks like a 3D (C,H,W) shape, we might consider it an 'image'.
|
||||||
Otherwise default to appropriate dtype for numeric.
|
Otherwise default to appropriate dtype for numeric.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
shape = tuple(t.shape)
|
shape = tuple(t.shape)
|
||||||
# Basic guess: if we have exactly 3 dims and shape[0] in {1, 3}, guess 'image'
|
# Basic guess: if we have exactly 3 dims and shape[0] in {1, 3}, guess 'image'
|
||||||
if len(shape) == 3 and shape[0] in [1, 3]:
|
if len(shape) == 3 and shape[0] in [1, 3]:
|
||||||
@@ -757,8 +765,7 @@ def guess_feature_info(t, name: str):
|
|||||||
def concatenate_batch_transitions(
|
def concatenate_batch_transitions(
|
||||||
left_batch_transitions: BatchTransition, right_batch_transition: BatchTransition
|
left_batch_transitions: BatchTransition, right_batch_transition: BatchTransition
|
||||||
) -> BatchTransition:
|
) -> BatchTransition:
|
||||||
"""
|
"""Concatenates two BatchTransition objects into one.
|
||||||
Concatenates two BatchTransition objects into one.
|
|
||||||
|
|
||||||
This function merges the right BatchTransition into the left one by concatenating
|
This function merges the right BatchTransition into the left one by concatenating
|
||||||
all corresponding tensors along dimension 0. The operation modifies the left_batch_transitions
|
all corresponding tensors along dimension 0. The operation modifies the left_batch_transitions
|
||||||
|
|||||||
@@ -29,8 +29,7 @@ from lerobot.utils.constants import DONE, REWARD
|
|||||||
|
|
||||||
|
|
||||||
def select_rect_roi(img):
|
def select_rect_roi(img):
|
||||||
"""
|
"""Allows the user to draw a rectangular ROI on the image.
|
||||||
Allows the user to draw a rectangular ROI on the image.
|
|
||||||
|
|
||||||
The user must click and drag to draw the rectangle.
|
The user must click and drag to draw the rectangle.
|
||||||
- While dragging, the rectangle is dynamically drawn.
|
- While dragging, the rectangle is dynamically drawn.
|
||||||
@@ -52,6 +51,7 @@ def select_rect_roi(img):
|
|||||||
index_x, index_y = -1, -1 # Initial click coordinates
|
index_x, index_y = -1, -1 # Initial click coordinates
|
||||||
|
|
||||||
def mouse_callback(event, x, y, flags, param):
|
def mouse_callback(event, x, y, flags, param):
|
||||||
|
"""`cv2.setMouseCallback` handler that drives the click-and-drag ROI selection."""
|
||||||
nonlocal index_x, index_y, drawing, roi, working_img
|
nonlocal index_x, index_y, drawing, roi, working_img
|
||||||
|
|
||||||
if event == cv2.EVENT_LBUTTONDOWN:
|
if event == cv2.EVENT_LBUTTONDOWN:
|
||||||
@@ -118,12 +118,11 @@ def select_rect_roi(img):
|
|||||||
|
|
||||||
|
|
||||||
def select_square_roi_for_images(images: dict) -> dict:
|
def select_square_roi_for_images(images: dict) -> dict:
|
||||||
"""
|
"""For each image in the provided dictionary, open a window to allow the user to select a ROI.
|
||||||
For each image in the provided dictionary, open a window to allow the user
|
|
||||||
to select a rectangular ROI. Returns a dictionary mapping each key to a tuple
|
|
||||||
(top, left, height, width) representing the ROI.
|
|
||||||
|
|
||||||
Parameters:
|
Returns a dictionary mapping each key to a tuple (top, left, height, width) representing the ROI.
|
||||||
|
|
||||||
|
Args:
|
||||||
images (dict): Dictionary where keys are identifiers and values are OpenCV images.
|
images (dict): Dictionary where keys are identifiers and values are OpenCV images.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
@@ -149,9 +148,7 @@ def select_square_roi_for_images(images: dict) -> dict:
|
|||||||
|
|
||||||
|
|
||||||
def get_image_from_lerobot_dataset(dataset: LeRobotDataset):
|
def get_image_from_lerobot_dataset(dataset: LeRobotDataset):
|
||||||
"""
|
"""Find the first row in the dataset and extract the image in order to be used for the crop."""
|
||||||
Find the first row in the dataset and extract the image in order to be used for the crop.
|
|
||||||
"""
|
|
||||||
row = dataset[0]
|
row = dataset[0]
|
||||||
image_dict = {}
|
image_dict = {}
|
||||||
for k in row:
|
for k in row:
|
||||||
@@ -169,19 +166,23 @@ def convert_lerobot_dataset_to_cropped_lerobot_dataset(
|
|||||||
push_to_hub: bool = False,
|
push_to_hub: bool = False,
|
||||||
task: str = "",
|
task: str = "",
|
||||||
) -> LeRobotDataset:
|
) -> LeRobotDataset:
|
||||||
"""
|
"""Converts an existing LeRobotDataset to a new one with cropped/resized image observations.
|
||||||
Converts an existing LeRobotDataset by iterating over its episodes and frames,
|
|
||||||
applying cropping and resizing to image observations, and saving a new dataset
|
Iterates over the source dataset's episodes and frames, applying cropping and resizing to image
|
||||||
with the transformed data.
|
observations, and saves a new dataset with the transformed data.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
original_dataset (LeRobotDataset): The source dataset.
|
original_dataset (`LeRobotDataset`): The source dataset.
|
||||||
crop_params_dict (dict[str, Tuple[int, int, int, int]]):
|
crop_params_dict (`dict[str, tuple[int, int, int, int]]`):
|
||||||
A dictionary mapping observation keys to crop parameters (top, left, height, width).
|
A dictionary mapping observation keys to crop parameters (top, left, height, width).
|
||||||
new_repo_id (str): Repository id for the new dataset.
|
new_repo_id (`str`): Repository id for the new dataset.
|
||||||
new_dataset_root (str): The root directory where the new dataset will be written.
|
new_dataset_root (`str`): The root directory where the new dataset will be written.
|
||||||
resize_size (tuple[int, int], optional): The target size (height, width) after cropping.
|
resize_size (`tuple[int, int]`, *optional*, defaults to `(128, 128)`): The target size
|
||||||
Defaults to (128, 128).
|
(height, width) after cropping.
|
||||||
|
push_to_hub (`bool`, *optional*, defaults to `False`): Whether to push the new dataset to the
|
||||||
|
Hugging Face Hub.
|
||||||
|
task (`str`, *optional*, defaults to `""`): Task description recorded on every frame of the
|
||||||
|
new dataset.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
LeRobotDataset: A new LeRobotDataset where the specified image observations have been cropped
|
LeRobotDataset: A new LeRobotDataset where the specified image observations have been cropped
|
||||||
|
|||||||
@@ -49,6 +49,18 @@ class OnlineOfflineMixer(DataMixer):
|
|||||||
offline_buffer: ReplayBuffer | None = None,
|
offline_buffer: ReplayBuffer | None = None,
|
||||||
online_ratio: float = 1.0,
|
online_ratio: float = 1.0,
|
||||||
):
|
):
|
||||||
|
"""Create the mixer.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
online_buffer (`ReplayBuffer`): Buffer of transitions collected online during training.
|
||||||
|
offline_buffer (`ReplayBuffer | None`, *optional*): Buffer of pre-collected offline
|
||||||
|
transitions. When `None`, every batch is drawn from `online_buffer` alone.
|
||||||
|
online_ratio (`float`, *optional*, defaults to 1.0): Fraction of each batch drawn from
|
||||||
|
`online_buffer`; the remainder comes from `offline_buffer`. Must be in `[0, 1]`.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: If `online_ratio` is not in `[0, 1]`.
|
||||||
|
"""
|
||||||
if not 0.0 <= online_ratio <= 1.0:
|
if not 0.0 <= online_ratio <= 1.0:
|
||||||
raise ValueError(f"online_ratio must be in [0, 1], got {online_ratio}")
|
raise ValueError(f"online_ratio must be in [0, 1], got {online_ratio}")
|
||||||
self.online_buffer = online_buffer
|
self.online_buffer = online_buffer
|
||||||
@@ -56,6 +68,7 @@ class OnlineOfflineMixer(DataMixer):
|
|||||||
self.online_ratio = online_ratio
|
self.online_ratio = online_ratio
|
||||||
|
|
||||||
def sample(self, batch_size: int) -> BatchType:
|
def sample(self, batch_size: int) -> BatchType:
|
||||||
|
"""See [`~rl.data_sources.DataMixer.sample`]."""
|
||||||
if self.offline_buffer is None:
|
if self.offline_buffer is None:
|
||||||
return self.online_buffer.sample(batch_size)
|
return self.online_buffer.sample(batch_size)
|
||||||
|
|
||||||
@@ -73,7 +86,6 @@ class OnlineOfflineMixer(DataMixer):
|
|||||||
queue_size: int = 2,
|
queue_size: int = 2,
|
||||||
):
|
):
|
||||||
"""Yield batches by composing buffer async iterators."""
|
"""Yield batches by composing buffer async iterators."""
|
||||||
|
|
||||||
n_online = max(1, int(batch_size * self.online_ratio))
|
n_online = max(1, int(batch_size * self.online_ratio))
|
||||||
|
|
||||||
online_iter = self.online_buffer.get_iterator(
|
online_iter = self.online_buffer.get_iterator(
|
||||||
|
|||||||
@@ -36,6 +36,13 @@ logging.basicConfig(level=logging.INFO)
|
|||||||
|
|
||||||
|
|
||||||
def eval_policy(env, policy, n_episodes):
|
def eval_policy(env, policy, n_episodes):
|
||||||
|
"""Roll out `policy` in `env` for `n_episodes` and log the per-episode and average reward.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
env (`gymnasium.Env`): A robot environment, built via `make_robot_env`.
|
||||||
|
policy (`PreTrainedPolicy`): A policy exposing `select_action(obs) -> action`.
|
||||||
|
n_episodes (`int`): Number of episodes to run.
|
||||||
|
"""
|
||||||
sum_reward_episode = []
|
sum_reward_episode = []
|
||||||
for _ in range(n_episodes):
|
for _ in range(n_episodes):
|
||||||
obs, _ = env.reset()
|
obs, _ = env.reset()
|
||||||
@@ -54,6 +61,12 @@ def eval_policy(env, policy, n_episodes):
|
|||||||
|
|
||||||
@parser.wrap()
|
@parser.wrap()
|
||||||
def main(cfg: TrainRLServerPipelineConfig):
|
def main(cfg: TrainRLServerPipelineConfig):
|
||||||
|
"""CLI entry point: load a pretrained policy and evaluate it for 10 episodes.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
cfg (`TrainRLServerPipelineConfig`): Parsed from the CLI. `cfg.env.pretrained_policy_name_or_path`
|
||||||
|
selects the checkpoint to load; `cfg.dataset.repo_id` provides normalization stats.
|
||||||
|
"""
|
||||||
env_cfg = cfg.env
|
env_cfg = cfg.env
|
||||||
env = make_robot_env(env_cfg)
|
env = make_robot_env(env_cfg)
|
||||||
dataset_cfg = cfg.dataset
|
dataset_cfg = cfg.dataset
|
||||||
|
|||||||
@@ -305,7 +305,9 @@ def make_robot_env(cfg: HILSerlRobotEnvConfig) -> tuple[gym.Env, Any]:
|
|||||||
"""Create robot environment from configuration.
|
"""Create robot environment from configuration.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg: Environment configuration.
|
cfg (`HILSerlRobotEnvConfig`): Environment configuration. `cfg.name == "gym_hil"` selects the
|
||||||
|
GymHIL simulation environment; otherwise a real-robot `RobotEnv` is built from
|
||||||
|
`cfg.robot`/`cfg.teleop`.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tuple of (gym environment, teleoperator device).
|
Tuple of (gym environment, teleoperator device).
|
||||||
@@ -363,10 +365,13 @@ def make_processors(
|
|||||||
"""Create environment and action processors.
|
"""Create environment and action processors.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
env: Robot environment instance.
|
env (`Env`): The environment returned by `make_robot_env`.
|
||||||
teleop_device: Teleoperator device for intervention.
|
teleop_device (`lerobot.teleoperators.teleoperator.Teleoperator | None`): The teleoperator
|
||||||
cfg: Processor configuration.
|
device returned by `make_robot_env`, used to configure intervention-related processor
|
||||||
device: Target device for computations.
|
steps. `None` for simulation environments.
|
||||||
|
cfg (`HILSerlRobotEnvConfig`): Environment configuration; provides the reward classifier,
|
||||||
|
gripper, and reset-behavior settings for the built processor steps.
|
||||||
|
device (`str`, *optional*, defaults to `"cpu"`): Torch device the processors run on.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tuple of (environment processor, action processor).
|
Tuple of (environment processor, action processor).
|
||||||
@@ -536,20 +541,21 @@ def step_env_and_process_transition(
|
|||||||
env_processor: DataProcessorPipeline[EnvTransition, EnvTransition],
|
env_processor: DataProcessorPipeline[EnvTransition, EnvTransition],
|
||||||
action_processor: DataProcessorPipeline[EnvTransition, EnvTransition],
|
action_processor: DataProcessorPipeline[EnvTransition, EnvTransition],
|
||||||
) -> EnvTransition:
|
) -> EnvTransition:
|
||||||
"""
|
"""Execute one step with processor pipeline.
|
||||||
Execute one step with processor pipeline.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
env: The robot environment
|
env (`Env`): The environment to step.
|
||||||
transition: Current transition state
|
transition (`EnvTransition`): The current transition; its observation is overwritten with the
|
||||||
action: Action to execute
|
action processor's input before dispatch, then discarded.
|
||||||
env_processor: Environment processor
|
action (`Tensor`): The raw action to process and send to `env`.
|
||||||
action_processor: Action processor
|
env_processor (`DataProcessorPipeline`): Post-processes the environment-produced transition
|
||||||
|
(e.g. reward shaping, termination overrides).
|
||||||
|
action_processor (`DataProcessorPipeline`): Pre-processes `action` before it reaches `env`
|
||||||
|
(e.g. intervention overrides, gripper handling).
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Processed transition with updated state.
|
Processed transition with updated state.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# Create action transition
|
# Create action transition
|
||||||
transition[TransitionKey.ACTION] = action
|
transition[TransitionKey.ACTION] = action
|
||||||
transition[TransitionKey.OBSERVATION] = (
|
transition[TransitionKey.OBSERVATION] = (
|
||||||
@@ -618,14 +624,16 @@ def control_loop(
|
|||||||
cfg: GymManipulatorConfig,
|
cfg: GymManipulatorConfig,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Main control loop for robot environment interaction.
|
"""Main control loop for robot environment interaction.
|
||||||
if cfg.mode == "record": then a dataset will be created and recorded
|
|
||||||
|
When `cfg.mode == "record"`, a dataset is created and recorded.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
env: The robot environment
|
env (`Env`): The environment to control, built via `make_robot_env`.
|
||||||
env_processor: Environment processor
|
env_processor (`DataProcessorPipeline`): Post-processes environment-produced transitions.
|
||||||
action_processor: Action processor
|
action_processor (`DataProcessorPipeline`): Pre-processes teleoperator actions before they
|
||||||
teleop_device: Teleoperator device
|
reach `env`.
|
||||||
cfg: gym_manipulator configuration
|
teleop_device (`Teleoperator`): Teleoperator device driving the robot.
|
||||||
|
cfg (`GymManipulatorConfig`): Control-loop configuration (mode, fps, episode/dataset settings).
|
||||||
"""
|
"""
|
||||||
dt = 1.0 / cfg.env.fps
|
dt = 1.0 / cfg.env.fps
|
||||||
|
|
||||||
|
|||||||
@@ -31,8 +31,7 @@ from lerobot.utils.constants import OBS_STATE
|
|||||||
@dataclass
|
@dataclass
|
||||||
@ProcessorStepRegistry.register("joint_velocity_processor")
|
@ProcessorStepRegistry.register("joint_velocity_processor")
|
||||||
class JointVelocityProcessorStep(ObservationProcessorStep):
|
class JointVelocityProcessorStep(ObservationProcessorStep):
|
||||||
"""
|
"""Calculates and appends joint velocity information to the observation state.
|
||||||
Calculates and appends joint velocity information to the observation state.
|
|
||||||
|
|
||||||
This step computes the velocity of each joint by calculating the finite
|
This step computes the velocity of each joint by calculating the finite
|
||||||
difference between the current and the last observed joint positions. The
|
difference between the current and the last observed joint positions. The
|
||||||
@@ -50,8 +49,7 @@ class JointVelocityProcessorStep(ObservationProcessorStep):
|
|||||||
last_joint_positions: torch.Tensor | None = None
|
last_joint_positions: torch.Tensor | None = None
|
||||||
|
|
||||||
def observation(self, observation: dict) -> dict:
|
def observation(self, observation: dict) -> dict:
|
||||||
"""
|
"""Computes joint velocities and adds them to the observation state.
|
||||||
Computes joint velocities and adds them to the observation state.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
observation: The input observation dictionary, expected to contain
|
observation: The input observation dictionary, expected to contain
|
||||||
@@ -89,8 +87,7 @@ class JointVelocityProcessorStep(ObservationProcessorStep):
|
|||||||
return new_observation
|
return new_observation
|
||||||
|
|
||||||
def get_config(self) -> dict[str, Any]:
|
def get_config(self) -> dict[str, Any]:
|
||||||
"""
|
"""Returns the configuration of the step for serialization.
|
||||||
Returns the configuration of the step for serialization.
|
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
A dictionary containing the time step `dt`.
|
A dictionary containing the time step `dt`.
|
||||||
@@ -106,8 +103,7 @@ class JointVelocityProcessorStep(ObservationProcessorStep):
|
|||||||
def transform_features(
|
def transform_features(
|
||||||
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
||||||
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
||||||
"""
|
"""Updates the `observation.state` feature to reflect the added velocities.
|
||||||
Updates the `observation.state` feature to reflect the added velocities.
|
|
||||||
|
|
||||||
This method doubles the size of the first dimension of the `observation.state`
|
This method doubles the size of the first dimension of the `observation.state`
|
||||||
shape to account for the concatenation of position and velocity vectors.
|
shape to account for the concatenation of position and velocity vectors.
|
||||||
@@ -132,8 +128,7 @@ class JointVelocityProcessorStep(ObservationProcessorStep):
|
|||||||
@dataclass
|
@dataclass
|
||||||
@ProcessorStepRegistry.register("current_processor")
|
@ProcessorStepRegistry.register("current_processor")
|
||||||
class MotorCurrentProcessorStep(ObservationProcessorStep):
|
class MotorCurrentProcessorStep(ObservationProcessorStep):
|
||||||
"""
|
"""Reads motor currents from a robot and appends them to the observation state.
|
||||||
Reads motor currents from a robot and appends them to the observation state.
|
|
||||||
|
|
||||||
This step queries the robot's hardware interface to get the present current
|
This step queries the robot's hardware interface to get the present current
|
||||||
for each motor and concatenates this information to the existing state vector.
|
for each motor and concatenates this information to the existing state vector.
|
||||||
@@ -146,8 +141,7 @@ class MotorCurrentProcessorStep(ObservationProcessorStep):
|
|||||||
robot: Robot | None = None
|
robot: Robot | None = None
|
||||||
|
|
||||||
def observation(self, observation: dict) -> dict:
|
def observation(self, observation: dict) -> dict:
|
||||||
"""
|
"""Fetches motor currents and adds them to the observation state.
|
||||||
Fetches motor currents and adds them to the observation state.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
observation: The input observation dictionary.
|
observation: The input observation dictionary.
|
||||||
@@ -184,8 +178,7 @@ class MotorCurrentProcessorStep(ObservationProcessorStep):
|
|||||||
def transform_features(
|
def transform_features(
|
||||||
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
||||||
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
||||||
"""
|
"""Updates the `observation.state` feature to reflect the added motor currents.
|
||||||
Updates the `observation.state` feature to reflect the added motor currents.
|
|
||||||
|
|
||||||
This method increases the size of the first dimension of the `observation.state`
|
This method increases the size of the first dimension of the `observation.state`
|
||||||
shape by the number of motors in the robot.
|
shape by the number of motors in the robot.
|
||||||
|
|||||||
+85
-70
@@ -14,8 +14,7 @@
|
|||||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
# See the License for the specific language governing permissions and
|
# See the License for the specific language governing permissions and
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
"""
|
"""Learner server runner for distributed HILSerl robot policy training.
|
||||||
Learner server runner for distributed HILSerl robot policy training.
|
|
||||||
|
|
||||||
This script implements the learner component of the distributed HILSerl architecture.
|
This script implements the learner component of the distributed HILSerl architecture.
|
||||||
It initializes the policy network, maintains replay buffers, and updates
|
It initializes the policy network, maintains replay buffers, and updates
|
||||||
@@ -121,6 +120,11 @@ from .trainer import RLTrainer
|
|||||||
|
|
||||||
@parser.wrap()
|
@parser.wrap()
|
||||||
def train_cli(cfg: TrainRLServerPipelineConfig):
|
def train_cli(cfg: TrainRLServerPipelineConfig):
|
||||||
|
"""CLI entry point for the HILSerl learner server.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
cfg (`TrainRLServerPipelineConfig`): Parsed from the CLI, forwarded to `train`.
|
||||||
|
"""
|
||||||
# Fail fast with a friendly error if the optional ``hilserl`` extra is missing.
|
# Fail fast with a friendly error if the optional ``hilserl`` extra is missing.
|
||||||
require_package("grpcio", extra="hilserl", import_name="grpc")
|
require_package("grpcio", extra="hilserl", import_name="grpc")
|
||||||
if not use_threads(cfg):
|
if not use_threads(cfg):
|
||||||
@@ -136,14 +140,13 @@ def train_cli(cfg: TrainRLServerPipelineConfig):
|
|||||||
|
|
||||||
|
|
||||||
def train(cfg: TrainRLServerPipelineConfig, job_name: str | None = None):
|
def train(cfg: TrainRLServerPipelineConfig, job_name: str | None = None):
|
||||||
"""
|
"""Main training function that initializes and runs the training process.
|
||||||
Main training function that initializes and runs the training process.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg (TrainRLServerPipelineConfig): The training configuration
|
cfg (`TrainRLServerPipelineConfig`): The training configuration.
|
||||||
job_name (str | None, optional): Job name for logging. Defaults to None.
|
job_name (`str | None`, *optional*): Job name for logging. Defaults to `cfg.job_name` when
|
||||||
|
unset.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
cfg.validate()
|
cfg.validate()
|
||||||
|
|
||||||
if job_name is None:
|
if job_name is None:
|
||||||
@@ -198,13 +201,12 @@ def start_learner_threads(
|
|||||||
wandb_logger: WandBLogger | None,
|
wandb_logger: WandBLogger | None,
|
||||||
shutdown_event: Any, # Event
|
shutdown_event: Any, # Event
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""Start the learner threads for training.
|
||||||
Start the learner threads for training.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg (TrainRLServerPipelineConfig): Training configuration
|
cfg (`TrainRLServerPipelineConfig`): Training configuration.
|
||||||
wandb_logger (WandBLogger | None): Logger for metrics
|
wandb_logger (`WandBLogger | None`): Logger for metrics.
|
||||||
shutdown_event: Event to signal shutdown
|
shutdown_event (`Event`): Event signaling the learner and its background workers to stop.
|
||||||
"""
|
"""
|
||||||
# Create multiprocessing queues
|
# Create multiprocessing queues
|
||||||
transition_queue = Queue()
|
transition_queue = Queue()
|
||||||
@@ -275,9 +277,7 @@ def add_actor_information_and_train(
|
|||||||
interaction_message_queue: Queue,
|
interaction_message_queue: Queue,
|
||||||
parameters_queue: Queue,
|
parameters_queue: Queue,
|
||||||
):
|
):
|
||||||
"""
|
"""Handles data transfer from the actor to the learner, manages training updates, and logs progress.
|
||||||
Handles data transfer from the actor to the learner, manages training updates,
|
|
||||||
and logs training progress in an online reinforcement learning setup.
|
|
||||||
|
|
||||||
This function continuously:
|
This function continuously:
|
||||||
- Transfers transitions from the actor to the replay buffer.
|
- Transfers transitions from the actor to the replay buffer.
|
||||||
@@ -482,17 +482,18 @@ def start_learner(
|
|||||||
shutdown_event: Any, # Event
|
shutdown_event: Any, # Event
|
||||||
cfg: TrainRLServerPipelineConfig,
|
cfg: TrainRLServerPipelineConfig,
|
||||||
):
|
):
|
||||||
"""
|
"""Start the learner server for training.
|
||||||
Start the learner server for training.
|
|
||||||
It will receive transitions and interaction messages from the actor server,
|
Receives transitions and interaction messages from the actor server, and sends policy parameters
|
||||||
and send policy parameters to the actor server.
|
to the actor server.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
parameters_queue: Queue for sending policy parameters to the actor
|
parameters_queue (`Queue`): Queue of serialized policy weights, drained and streamed to the
|
||||||
transition_queue: Queue for receiving transitions from the actor
|
actor by `LearnerService.StreamParameters`.
|
||||||
interaction_message_queue: Queue for receiving interaction messages from the actor
|
transition_queue (`Queue`): Queue filled by `LearnerService.SendTransitions`.
|
||||||
shutdown_event: Event to signal shutdown
|
interaction_message_queue (`Queue`): Queue filled by `LearnerService.SendInteractions`.
|
||||||
cfg: Training configuration
|
shutdown_event (`Event`): Event signaling this process/thread to stop.
|
||||||
|
cfg (`TrainRLServerPipelineConfig`): Training configuration.
|
||||||
"""
|
"""
|
||||||
if not use_threads(cfg):
|
if not use_threads(cfg):
|
||||||
# Create a process-specific log file
|
# Create a process-specific log file
|
||||||
@@ -560,8 +561,7 @@ def save_training_checkpoint(
|
|||||||
preprocessor=None,
|
preprocessor=None,
|
||||||
postprocessor=None,
|
postprocessor=None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""Save training checkpoint and associated data.
|
||||||
Save training checkpoint and associated data.
|
|
||||||
|
|
||||||
This function performs the following steps:
|
This function performs the following steps:
|
||||||
1. Creates a checkpoint directory with the current optimization step
|
1. Creates a checkpoint directory with the current optimization step
|
||||||
@@ -572,18 +572,26 @@ def save_training_checkpoint(
|
|||||||
6. If an offline replay buffer exists, saves it as a separate dataset
|
6. If an offline replay buffer exists, saves it as a separate dataset
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg: Training configuration
|
cfg (`TrainRLServerPipelineConfig`): Training configuration, saved alongside the checkpoint.
|
||||||
optimization_step: Current optimization step
|
optimization_step (`int`): Current optimization step; used to name the checkpoint directory.
|
||||||
online_steps: Total number of online steps
|
online_steps (`int`): Total number of online steps; used to size the checkpoint directory's
|
||||||
interaction_message: Dictionary containing interaction information
|
zero-padded step number.
|
||||||
policy: Policy model to save
|
interaction_message (`dict | None`): Latest interaction message; its `"Interaction step"`
|
||||||
optimizers: Dictionary of optimizers
|
entry is saved for resuming training.
|
||||||
replay_buffer: Replay buffer to save as dataset
|
policy (`Module`): Policy model to save.
|
||||||
offline_replay_buffer: Optional offline replay buffer to save
|
optimizers (`dict`): Dictionary of optimizers whose states are saved.
|
||||||
dataset_repo_id: Repository ID for dataset
|
replay_buffer (`ReplayBuffer`): Replay buffer to save as a dataset.
|
||||||
fps: Frames per second for dataset
|
algorithm (`lerobot.rl.algorithms.base.RLAlgorithm | None`, *optional*): Algorithm whose state
|
||||||
preprocessor: Optional preprocessor pipeline to save
|
dict (critic ensembles, temperature, etc.) should also be saved.
|
||||||
postprocessor: Optional postprocessor pipeline to save
|
offline_replay_buffer (`lerobot.rl.buffer.ReplayBuffer | None`, *optional*): Optional offline
|
||||||
|
replay buffer, saved as a separate dataset when provided.
|
||||||
|
dataset_repo_id (`str | None`, *optional*): Repository id used when converting the replay
|
||||||
|
buffer(s) to a dataset.
|
||||||
|
fps (`int`, *optional*, defaults to 30): Frames per second recorded on the saved dataset(s).
|
||||||
|
preprocessor (`PolicyProcessorPipeline | None`, *optional*): Optional preprocessor pipeline to
|
||||||
|
save alongside the policy.
|
||||||
|
postprocessor (`PolicyProcessorPipeline | None`, *optional*): Optional postprocessor pipeline
|
||||||
|
to save alongside the policy.
|
||||||
"""
|
"""
|
||||||
logging.info(f"Checkpoint policy after step {optimization_step}")
|
logging.info(f"Checkpoint policy after step {optimization_step}")
|
||||||
_num_digits = max(6, len(str(online_steps)))
|
_num_digits = max(6, len(str(online_steps)))
|
||||||
@@ -650,8 +658,7 @@ def save_training_checkpoint(
|
|||||||
|
|
||||||
|
|
||||||
def handle_resume_logic(cfg: TrainRLServerPipelineConfig) -> TrainRLServerPipelineConfig:
|
def handle_resume_logic(cfg: TrainRLServerPipelineConfig) -> TrainRLServerPipelineConfig:
|
||||||
"""
|
"""Handle the resume logic for training.
|
||||||
Handle the resume logic for training.
|
|
||||||
|
|
||||||
If resume is True:
|
If resume is True:
|
||||||
- Verifies that a checkpoint exists
|
- Verifies that a checkpoint exists
|
||||||
@@ -712,19 +719,19 @@ def load_training_state(
|
|||||||
algorithm: RLAlgorithm | None = None,
|
algorithm: RLAlgorithm | None = None,
|
||||||
device: str | torch.device = "cpu",
|
device: str | torch.device = "cpu",
|
||||||
):
|
):
|
||||||
"""
|
"""Loads the training state from the most recent checkpoint.
|
||||||
Loads the training state (optimizers, RNG, step + interaction step, and
|
|
||||||
algorithm-owned tensors) from the most recent checkpoint.
|
Restores optimizers, RNG state, the optimization/interaction step, and algorithm-owned tensors.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg (TrainRLServerPipelineConfig): Training configuration; `cfg.resume` gates the load and
|
cfg (`TrainRLServerPipelineConfig`): Training configuration; `cfg.resume` gates the load and
|
||||||
`cfg.output_dir` locates the last checkpoint.
|
`cfg.output_dir` locates the last checkpoint.
|
||||||
optimizers (Optimizer | dict[str, Optimizer]): Optimizers to load state into.
|
optimizers (`Optimizer | dict[str, Optimizer]`): Optimizers to load state into.
|
||||||
algorithm (RLAlgorithm | None, optional): Algorithm whose state dict should be restored.
|
algorithm (`RLAlgorithm | None`, *optional*): Algorithm whose state dict should be restored.
|
||||||
Required for full main-equivalent resume; the policy itself is restored separately via
|
Required for full main-equivalent resume; the policy itself is restored separately via
|
||||||
`make_policy`. Defaults to None.
|
`make_policy`.
|
||||||
device (str | torch.device, optional): Device on which to place loaded algorithm tensors.
|
device (`str | torch.device`, *optional*, defaults to `"cpu"`): Device on which to place
|
||||||
Defaults to "cpu".
|
loaded algorithm tensors.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
tuple[int | None, int | None]: `(optimization_step, interaction_step)`, or `(None, None)`
|
tuple[int | None, int | None]: `(optimization_step, interaction_step)`, or `(None, None)`
|
||||||
@@ -772,8 +779,7 @@ def load_training_state(
|
|||||||
|
|
||||||
|
|
||||||
def log_training_info(cfg: TrainRLServerPipelineConfig, policy: nn.Module) -> None:
|
def log_training_info(cfg: TrainRLServerPipelineConfig, policy: nn.Module) -> None:
|
||||||
"""
|
"""Log information about the training process.
|
||||||
Log information about the training process.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg (TrainRLServerPipelineConfig): Training configuration
|
cfg (TrainRLServerPipelineConfig): Training configuration
|
||||||
@@ -792,8 +798,7 @@ def log_training_info(cfg: TrainRLServerPipelineConfig, policy: nn.Module) -> No
|
|||||||
def initialize_replay_buffer(
|
def initialize_replay_buffer(
|
||||||
cfg: TrainRLServerPipelineConfig, device: str, storage_device: str
|
cfg: TrainRLServerPipelineConfig, device: str, storage_device: str
|
||||||
) -> ReplayBuffer:
|
) -> ReplayBuffer:
|
||||||
"""
|
"""Initialize a replay buffer, either empty or from a dataset if resuming.
|
||||||
Initialize a replay buffer, either empty or from a dataset if resuming.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg (TrainRLServerPipelineConfig): Training configuration
|
cfg (TrainRLServerPipelineConfig): Training configuration
|
||||||
@@ -837,8 +842,7 @@ def initialize_offline_replay_buffer(
|
|||||||
device: str,
|
device: str,
|
||||||
storage_device: str,
|
storage_device: str,
|
||||||
) -> ReplayBuffer:
|
) -> ReplayBuffer:
|
||||||
"""
|
"""Initialize an offline replay buffer from a dataset.
|
||||||
Initialize an offline replay buffer from a dataset.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cfg (TrainRLServerPipelineConfig): Training configuration
|
cfg (TrainRLServerPipelineConfig): Training configuration
|
||||||
@@ -875,6 +879,7 @@ def initialize_offline_replay_buffer(
|
|||||||
|
|
||||||
|
|
||||||
def use_threads(cfg: TrainRLServerPipelineConfig) -> bool:
|
def use_threads(cfg: TrainRLServerPipelineConfig) -> bool:
|
||||||
|
"""Whether the learner's background workers should run as threads instead of processes."""
|
||||||
return cfg.policy.concurrency.learner == "threads"
|
return cfg.policy.concurrency.learner == "threads"
|
||||||
|
|
||||||
|
|
||||||
@@ -884,14 +889,14 @@ def check_nan_in_transition(
|
|||||||
next_state: torch.Tensor,
|
next_state: torch.Tensor,
|
||||||
raise_error: bool = False,
|
raise_error: bool = False,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""
|
"""Check for NaN values in transition data.
|
||||||
Check for NaN values in transition data.
|
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
observations: Dictionary of observation tensors
|
observations (`Tensor`): Dictionary of observation tensors.
|
||||||
actions: Action tensor
|
actions (`Tensor`): Action tensor.
|
||||||
next_state: Dictionary of next state tensors
|
next_state (`Tensor`): Dictionary of next-observation tensors.
|
||||||
raise_error: If True, raises ValueError when NaN is detected
|
raise_error (`bool`, *optional*, defaults to `False`): Whether to raise a `ValueError` instead
|
||||||
|
of just logging when a NaN is found.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: True if NaN values were detected, False otherwise
|
bool: True if NaN values were detected, False otherwise
|
||||||
@@ -925,6 +930,12 @@ def check_nan_in_transition(
|
|||||||
|
|
||||||
|
|
||||||
def push_actor_policy_to_queue(parameters_queue: Queue, algorithm: RLAlgorithm) -> None:
|
def push_actor_policy_to_queue(parameters_queue: Queue, algorithm: RLAlgorithm) -> None:
|
||||||
|
"""Serialize `algorithm`'s current weights and enqueue them for the actor-facing gRPC stream.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
parameters_queue (`Queue`): Queue drained by `LearnerService.StreamParameters`.
|
||||||
|
algorithm (`RLAlgorithm`): Source of the weights, via `get_weights`.
|
||||||
|
"""
|
||||||
logging.debug("[LEARNER] Pushing actor policy to the queue")
|
logging.debug("[LEARNER] Pushing actor policy to the queue")
|
||||||
|
|
||||||
# Create a dictionary to hold all the state dicts
|
# Create a dictionary to hold all the state dicts
|
||||||
@@ -958,11 +969,13 @@ def process_transitions(
|
|||||||
"""Process all available transitions from the queue.
|
"""Process all available transitions from the queue.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
transition_queue: Queue for receiving transitions from the actor
|
transition_queue (`Queue`): Queue filled by `LearnerService.SendTransitions`.
|
||||||
replay_buffer: Replay buffer to add transitions to
|
replay_buffer (`ReplayBuffer`): Buffer every non-NaN transition is added to.
|
||||||
offline_replay_buffer: Offline replay buffer to add transitions to
|
offline_replay_buffer (`ReplayBuffer`): Buffer intervention transitions are additionally added
|
||||||
dataset_repo_id: Repository ID for dataset
|
to, when `dataset_repo_id` is set.
|
||||||
shutdown_event: Event to signal shutdown
|
dataset_repo_id (`str | None`): When set, transitions tagged as interventions are also added
|
||||||
|
to `offline_replay_buffer`.
|
||||||
|
shutdown_event (`Event`): Event that stops the loop when set.
|
||||||
"""
|
"""
|
||||||
while not transition_queue.empty() and not shutdown_event.is_set():
|
while not transition_queue.empty() and not shutdown_event.is_set():
|
||||||
transition_list = transition_queue.get()
|
transition_list = transition_queue.get()
|
||||||
@@ -996,10 +1009,12 @@ def process_interaction_messages(
|
|||||||
"""Process all available interaction messages from the queue.
|
"""Process all available interaction messages from the queue.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
interaction_message_queue: Queue for receiving interaction messages
|
interaction_message_queue (`Queue`): Queue filled by `LearnerService.SendInteractions`.
|
||||||
interaction_step_shift: Amount to shift interaction step by
|
interaction_step_shift (`int`): Offset added to each message's `"Interaction step"` so it
|
||||||
wandb_logger: Logger for tracking progress
|
stays consistent with checkpointed state after a resume.
|
||||||
shutdown_event: Event to signal shutdown
|
wandb_logger (`lerobot.common.wandb_utils.WandBLogger | None`): Logger the message is
|
||||||
|
forwarded to, when set.
|
||||||
|
shutdown_event (`Event`): Event that stops the loop when set.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
dict | None: The last interaction message processed, or None if none were processed
|
dict | None: The last interaction message processed, or None if none were processed
|
||||||
|
|||||||
@@ -44,10 +44,10 @@ SHUTDOWN_TIMEOUT = 10
|
|||||||
|
|
||||||
|
|
||||||
class LearnerService(_ServicerBase):
|
class LearnerService(_ServicerBase):
|
||||||
"""
|
"""Implementation of the LearnerService gRPC service.
|
||||||
Implementation of the LearnerService gRPC service
|
|
||||||
This service is used to send parameters to the Actor and receive transitions and interactions from the Actor
|
Sends policy parameters to the actor and receives transitions and interactions from it; see
|
||||||
check transport.proto for the gRPC service definition
|
`transport.proto` for the gRPC service definition.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
@@ -59,6 +59,19 @@ class LearnerService(_ServicerBase):
|
|||||||
interaction_message_queue: Queue,
|
interaction_message_queue: Queue,
|
||||||
queue_get_timeout: float = 0.001,
|
queue_get_timeout: float = 0.001,
|
||||||
):
|
):
|
||||||
|
"""Create the servicer.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
shutdown_event (`Event`): Set to stop `StreamParameters`'s push loop.
|
||||||
|
parameters_queue (`Queue`): Queue of serialized policy weights, drained and streamed to
|
||||||
|
the actor by `StreamParameters`.
|
||||||
|
seconds_between_pushes (`float`): Minimum interval between successive parameter pushes.
|
||||||
|
transition_queue (`Queue`): Queue filled by `SendTransitions` with received transitions.
|
||||||
|
interaction_message_queue (`Queue`): Queue filled by `SendInteractions` with received
|
||||||
|
interaction messages.
|
||||||
|
queue_get_timeout (`float`, *optional*, defaults to 0.001): Timeout used when polling
|
||||||
|
`parameters_queue`.
|
||||||
|
"""
|
||||||
self.shutdown_event = shutdown_event
|
self.shutdown_event = shutdown_event
|
||||||
self.parameters_queue = parameters_queue
|
self.parameters_queue = parameters_queue
|
||||||
self.seconds_between_pushes = seconds_between_pushes
|
self.seconds_between_pushes = seconds_between_pushes
|
||||||
@@ -69,6 +82,17 @@ class LearnerService(_ServicerBase):
|
|||||||
def StreamParameters( # noqa: N802
|
def StreamParameters( # noqa: N802
|
||||||
self, request: "services_pb2.Empty", context: "grpc.ServicerContext"
|
self, request: "services_pb2.Empty", context: "grpc.ServicerContext"
|
||||||
):
|
):
|
||||||
|
"""GRPC server-streaming RPC: push the latest policy parameters to the actor.
|
||||||
|
|
||||||
|
Runs until `shutdown_event` is set, pushing at most once every `seconds_between_pushes`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
request (`services_pb2.Empty`): Unused; required by the gRPC service signature.
|
||||||
|
context (`grpc.ServicerContext`): gRPC call context.
|
||||||
|
|
||||||
|
Yields:
|
||||||
|
Chunks of a `services_pb2.Parameters` message, produced by `send_bytes_in_chunks`.
|
||||||
|
"""
|
||||||
# TODO: authorize the request
|
# TODO: authorize the request
|
||||||
logging.info("[LEARNER] Received request to stream parameters from the Actor")
|
logging.info("[LEARNER] Received request to stream parameters from the Actor")
|
||||||
|
|
||||||
@@ -104,6 +128,16 @@ class LearnerService(_ServicerBase):
|
|||||||
return services_pb2.Empty()
|
return services_pb2.Empty()
|
||||||
|
|
||||||
def SendTransitions(self, request_iterator, _context: "grpc.ServicerContext"): # noqa: N802
|
def SendTransitions(self, request_iterator, _context: "grpc.ServicerContext"): # noqa: N802
|
||||||
|
"""GRPC client-streaming RPC: receive transition chunks from the actor into `transition_queue`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
request_iterator: Stream of `services_pb2.Transition` chunks sent by the actor's
|
||||||
|
`transitions_stream`.
|
||||||
|
_context (`grpc.ServicerContext`): gRPC call context.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
services_pb2.Empty: Acknowledgement sent once the actor closes the stream.
|
||||||
|
"""
|
||||||
# TODO: authorize the request
|
# TODO: authorize the request
|
||||||
logging.info("[LEARNER] Received request to receive transitions from the Actor")
|
logging.info("[LEARNER] Received request to receive transitions from the Actor")
|
||||||
|
|
||||||
@@ -118,6 +152,16 @@ class LearnerService(_ServicerBase):
|
|||||||
return services_pb2.Empty()
|
return services_pb2.Empty()
|
||||||
|
|
||||||
def SendInteractions(self, request_iterator, _context: "grpc.ServicerContext"): # noqa: N802
|
def SendInteractions(self, request_iterator, _context: "grpc.ServicerContext"): # noqa: N802
|
||||||
|
"""GRPC client-streaming RPC: receive interaction-message chunks into `interaction_message_queue`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
request_iterator: Stream of `services_pb2.InteractionMessage` chunks sent by the actor's
|
||||||
|
`interactions_stream`.
|
||||||
|
_context (`grpc.ServicerContext`): gRPC call context.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
services_pb2.Empty: Acknowledgement sent once the actor closes the stream.
|
||||||
|
"""
|
||||||
# TODO: authorize the request
|
# TODO: authorize the request
|
||||||
logging.info("[LEARNER] Received request to receive interactions from the Actor")
|
logging.info("[LEARNER] Received request to receive interactions from the Actor")
|
||||||
|
|
||||||
@@ -132,4 +176,5 @@ class LearnerService(_ServicerBase):
|
|||||||
return services_pb2.Empty()
|
return services_pb2.Empty()
|
||||||
|
|
||||||
def Ready(self, request: "services_pb2.Empty", context: "grpc.ServicerContext"): # noqa: N802
|
def Ready(self, request: "services_pb2.Empty", context: "grpc.ServicerContext"): # noqa: N802
|
||||||
|
"""GRPC health check: returns immediately, confirming the learner server is up."""
|
||||||
return services_pb2.Empty()
|
return services_pb2.Empty()
|
||||||
|
|||||||
@@ -23,6 +23,19 @@ from torch.multiprocessing import Queue
|
|||||||
|
|
||||||
|
|
||||||
def get_last_item_from_queue(queue: Queue, block=True, timeout: float = 0.1) -> Any:
|
def get_last_item_from_queue(queue: Queue, block=True, timeout: float = 0.1) -> Any:
|
||||||
|
"""Drain `queue` and return only the most recently enqueued item.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
queue (`Queue`): A `torch.multiprocessing.Queue` to drain.
|
||||||
|
block (`bool`, *optional*, defaults to `True`): Whether to block for up to `timeout` seconds
|
||||||
|
waiting for a first item before draining. When `False`, returns `None` if the queue is
|
||||||
|
currently empty.
|
||||||
|
timeout (`float`, *optional*, defaults to 0.1): Seconds to wait for a first item when `block`
|
||||||
|
is `True`.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Any: The most recent item, or `None` if the queue was (and stayed) empty.
|
||||||
|
"""
|
||||||
if block:
|
if block:
|
||||||
try:
|
try:
|
||||||
item = queue.get(timeout=timeout)
|
item = queue.get(timeout=timeout)
|
||||||
|
|||||||
@@ -28,6 +28,100 @@ from .algorithms.sac import SACAlgorithmConfig # noqa: F401
|
|||||||
|
|
||||||
@dataclass(kw_only=True)
|
@dataclass(kw_only=True)
|
||||||
class TrainRLServerPipelineConfig(TrainPipelineConfig):
|
class TrainRLServerPipelineConfig(TrainPipelineConfig):
|
||||||
|
"""Top-level config for the actor/learner distributed RL training server.
|
||||||
|
|
||||||
|
Extends [`~configs.train.TrainPipelineConfig`] with an optional (rather than required) offline
|
||||||
|
`dataset` and the RL-specific algorithm/data-mixing fields below.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
env (`lerobot.envs.configs.EnvConfig | None`, *optional*):
|
||||||
|
Simulation environment configuration, used for `env_eval_freq` evaluation rollouts.
|
||||||
|
policy (`lerobot.configs.policies.PreTrainedConfig | None`, *optional*):
|
||||||
|
The actor policy configuration.
|
||||||
|
reward_model (`lerobot.configs.rewards.RewardModelConfig | None`, *optional*):
|
||||||
|
Reward model configuration, when training a reward model instead of a policy.
|
||||||
|
output_dir (`pathlib.Path | None`, *optional*):
|
||||||
|
Directory to save run outputs to. Reusing the same value across runs overwrites its
|
||||||
|
contents unless `resume` is `True`.
|
||||||
|
job_name (`str | None`, *optional*):
|
||||||
|
Name used for logging and checkpoint directory naming.
|
||||||
|
resume (`bool`, *optional*, defaults to `False`):
|
||||||
|
Whether to resume a previous run from `--config_path`'s checkpoint.
|
||||||
|
seed (`int | None`, *optional*, defaults to 1000):
|
||||||
|
Random seed for model initialization, dataset shuffling, and evaluation environments.
|
||||||
|
cudnn_deterministic (`bool`, *optional*, defaults to `False`):
|
||||||
|
Whether to use deterministic cuDNN algorithms for reproducibility. Disables
|
||||||
|
`cudnn.benchmark`, which may reduce training speed.
|
||||||
|
num_workers (`int`, *optional*, defaults to 4):
|
||||||
|
Number of dataloader worker processes.
|
||||||
|
batch_size (`int`, *optional*, defaults to 8):
|
||||||
|
Offline-dataset dataloader batch size.
|
||||||
|
prefetch_factor (`int`, *optional*, defaults to 4):
|
||||||
|
Number of batches prefetched per dataloader worker.
|
||||||
|
persistent_workers (`bool`, *optional*, defaults to `True`):
|
||||||
|
Whether dataloader workers stay alive between epochs.
|
||||||
|
dataloader_multiprocessing_context (`str | None`, *optional*, defaults to `"spawn"`):
|
||||||
|
DataLoader worker start method. `None` uses Python's platform default.
|
||||||
|
steps (`int`, *optional*, defaults to 100000):
|
||||||
|
Total number of training steps.
|
||||||
|
env_eval_freq (`int`, *optional*, defaults to 20000):
|
||||||
|
Run the policy in the simulation environment every N steps to measure reward/success.
|
||||||
|
`0` disables environment evaluation.
|
||||||
|
log_freq (`int`, *optional*, defaults to 200):
|
||||||
|
Log training metrics every N steps.
|
||||||
|
eval_steps (`int`, *optional*, defaults to 0):
|
||||||
|
Compute eval loss on held-out episodes every N steps. `0` disables it.
|
||||||
|
max_eval_samples (`int`, *optional*, defaults to 0):
|
||||||
|
Cap on total eval samples, split uniformly across tasks. `0` uses all held-out data.
|
||||||
|
tolerance_s (`float`, *optional*, defaults to 0.0001):
|
||||||
|
Maximum timestamp tolerance, in seconds, when loading dataset frames.
|
||||||
|
save_checkpoint (`bool`, *optional*, defaults to `True`):
|
||||||
|
Whether to save training checkpoints at all.
|
||||||
|
save_freq (`int`, *optional*, defaults to 20000):
|
||||||
|
Save a checkpoint every N training steps, and after the last step. A non-positive value
|
||||||
|
disables periodic saving, keeping only the final checkpoint.
|
||||||
|
checkpoint_format (`CheckpointFormat`, *optional*, defaults to `CheckpointFormat.SAFETENSORS`):
|
||||||
|
Model-artifact format inside checkpoints.
|
||||||
|
use_policy_training_preset (`bool`, *optional*, defaults to `True`):
|
||||||
|
Whether to use the policy's own recommended optimizer/scheduler preset when `optimizer`/
|
||||||
|
`scheduler` are unset.
|
||||||
|
optimizer (`lerobot.optim.optimizers.OptimizerConfig | None`, *optional*):
|
||||||
|
Optimizer configuration override.
|
||||||
|
scheduler (`lerobot.optim.schedulers.LRSchedulerConfig | None`, *optional*):
|
||||||
|
Learning-rate scheduler configuration override.
|
||||||
|
parallelism (`ParallelismConfig`, *optional*):
|
||||||
|
Process topology: `dp_replicate`/`dp_shard` (HSDP) and context-parallel degree.
|
||||||
|
accelerator (`AcceleratorConfig`, *optional*):
|
||||||
|
Execution runtime handed to the Accelerator: mixed precision, gradient accumulation,
|
||||||
|
FSDP/DDP tuning knobs, compile & activation-checkpointing.
|
||||||
|
eval (`EvalConfig`, *optional*):
|
||||||
|
Simulation-environment evaluation configuration (number of episodes, batch size).
|
||||||
|
wandb (`WandBConfig`, *optional*):
|
||||||
|
Weights & Biases logging configuration.
|
||||||
|
peft (`lerobot.configs.default.PeftConfig | None`, *optional*):
|
||||||
|
PEFT (e.g. LoRA) adapter configuration for parameter-efficient fine-tuning.
|
||||||
|
job (`JobConfig`, *optional*):
|
||||||
|
Where to run training: local (default) or an HF Jobs flavor.
|
||||||
|
save_checkpoint_to_hub (`bool`, *optional*, defaults to `False`):
|
||||||
|
Whether to push each saved checkpoint to the Hub as it is written, not just the final
|
||||||
|
model.
|
||||||
|
sample_weighting (`lerobot.utils.sample_weighting.SampleWeightingConfig | None`, *optional*):
|
||||||
|
Sample weighting configuration (e.g. for RA-BC training).
|
||||||
|
rename_map (`dict`, *optional*):
|
||||||
|
Mapping to override observation image/state key names.
|
||||||
|
dataset (`DatasetConfig | None`, *optional*):
|
||||||
|
Optional offline dataset config. Unlike imitation-learning training, RL doesn't require an
|
||||||
|
offline dataset — data comes from the online replay buffer.
|
||||||
|
algorithm (`RLAlgorithmConfig | None`, *optional*):
|
||||||
|
RL algorithm configuration. Defaults to a SAC config (with `policy_config` populated from
|
||||||
|
`self.policy`) in `validate` when unset.
|
||||||
|
mixer (`str`, *optional*, defaults to `"online_offline"`):
|
||||||
|
Data mixer strategy name. Currently only `"online_offline"` is supported.
|
||||||
|
online_ratio (`float`, *optional*, defaults to 0.5):
|
||||||
|
Fraction of each training batch sampled from the online replay buffer when using
|
||||||
|
`OnlineOfflineMixer`; the remainder comes from the offline dataset.
|
||||||
|
"""
|
||||||
|
|
||||||
# NOTE: In RL, we don't need an offline dataset
|
# NOTE: In RL, we don't need an offline dataset
|
||||||
# TODO: Make `TrainPipelineConfig.dataset` optional
|
# TODO: Make `TrainPipelineConfig.dataset` optional
|
||||||
dataset: DatasetConfig | None = None # type: ignore[assignment] # because the parent class has made it's type non-optional
|
dataset: DatasetConfig | None = None # type: ignore[assignment] # because the parent class has made it's type non-optional
|
||||||
@@ -41,6 +135,11 @@ class TrainRLServerPipelineConfig(TrainPipelineConfig):
|
|||||||
online_ratio: float = 0.5
|
online_ratio: float = 0.5
|
||||||
|
|
||||||
def validate(self) -> None:
|
def validate(self) -> None:
|
||||||
|
"""See [`~configs.train.TrainPipelineConfig.validate`].
|
||||||
|
|
||||||
|
Additionally defaults `algorithm` to a SAC config and populates its `policy_config` from
|
||||||
|
`self.policy` when unset.
|
||||||
|
"""
|
||||||
super().validate()
|
super().validate()
|
||||||
|
|
||||||
if self.algorithm is None:
|
if self.algorithm is None:
|
||||||
|
|||||||
@@ -38,6 +38,16 @@ class RLTrainer:
|
|||||||
*,
|
*,
|
||||||
preprocessor: Any | None = None,
|
preprocessor: Any | None = None,
|
||||||
):
|
):
|
||||||
|
"""Build the trainer and its optimizers.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
algorithm (`RLAlgorithm`): The RL algorithm to train. `make_optimizers_and_scheduler` is
|
||||||
|
called on it immediately.
|
||||||
|
data_mixer (`DataMixer`): Data source the training-batch iterator is built from.
|
||||||
|
batch_size (`int`): Batch size requested from `data_mixer` on each training step.
|
||||||
|
preprocessor (`Any | None`, *optional*): When set, each sampled batch is passed through
|
||||||
|
`preprocess_rl_batch` before reaching the algorithm.
|
||||||
|
"""
|
||||||
self.algorithm = algorithm
|
self.algorithm = algorithm
|
||||||
self.data_mixer = data_mixer
|
self.data_mixer = data_mixer
|
||||||
self.batch_size = batch_size
|
self.batch_size = batch_size
|
||||||
@@ -90,12 +100,15 @@ class _PreprocessedIterator:
|
|||||||
__slots__ = ("_raw", "_preprocessor")
|
__slots__ = ("_raw", "_preprocessor")
|
||||||
|
|
||||||
def __init__(self, raw_iterator: Iterator[BatchType], preprocessor: Any) -> None:
|
def __init__(self, raw_iterator: Iterator[BatchType], preprocessor: Any) -> None:
|
||||||
|
"""Wrap `raw_iterator`, applying `preprocessor` to each yielded batch."""
|
||||||
self._raw = raw_iterator
|
self._raw = raw_iterator
|
||||||
self._preprocessor = preprocessor
|
self._preprocessor = preprocessor
|
||||||
|
|
||||||
def __iter__(self) -> _PreprocessedIterator:
|
def __iter__(self) -> _PreprocessedIterator:
|
||||||
|
"""Return `self` (this object is its own iterator)."""
|
||||||
return self
|
return self
|
||||||
|
|
||||||
def __next__(self) -> BatchType:
|
def __next__(self) -> BatchType:
|
||||||
|
"""Return the next preprocessed batch from the wrapped iterator."""
|
||||||
batch = next(self._raw)
|
batch = next(self._raw)
|
||||||
return preprocess_rl_batch(self._preprocessor, batch)
|
return preprocess_rl_batch(self._preprocessor, batch)
|
||||||
|
|||||||
@@ -60,7 +60,7 @@ PATH_TO_LEROBOT = PATH_TO_REPO / "src" / "lerobot"
|
|||||||
# Modules whose public objects are checked. Add a module here once its docstrings follow the standard.
|
# Modules whose public objects are checked. Add a module here once its docstrings follow the standard.
|
||||||
MODULES_TO_CHECK = [
|
MODULES_TO_CHECK = [
|
||||||
"lerobot.robots",
|
"lerobot.robots",
|
||||||
"lerobot.optim",
|
"lerobot.rl",
|
||||||
]
|
]
|
||||||
|
|
||||||
# Objects that do not yet follow the standard, so the check can be green from day one. Removing an entry
|
# Objects that do not yet follow the standard, so the check can be green from day one. Removing an entry
|
||||||
|
|||||||
Reference in New Issue
Block a user