Algorithm Specifications¶

Base classes and the algorithm registry that maps names (e.g. "DQN") to their concrete AlgorithmSpec subclass.

Registry¶

class agilerl.models.algo.AlgorithmRegistry¶

Central registry mapping algorithm names to their spec classes.

Populated at import time by the register() decorator applied to each concrete AlgorithmSpec subclass.

add(name: str, spec_cls: type[AlgorithmSpec]) → None¶

Parameters:

name (str) – Algorithm name (e.g. "DQN").
spec_cls (type[AlgorithmSpec]) – The spec class to register.

get(name: str) → RegistryEntry¶

Look up an entry by algorithm name.

Parameters:: name (str) – Algorithm name.
Returns:: The registry entry.
Return type:: RegistryEntry
Raises:: KeyError – If name is not registered.

agilerl.models.algo.ALGO_REGISTRY = <agilerl.models.algo.AlgorithmRegistry object>¶

Central registry mapping algorithm names to their spec classes.

Populated at import time by the register() decorator applied to each concrete AlgorithmSpec subclass.

Base Specs¶

class agilerl.models.algo.AlgorithmSpec(*, batch_size: Annotated[int, Ge(ge=1)] = 128, hp_config: Any | None = None)¶

Base specification for all algorithms.

Defines common fields and behavior for algorithm specifications, including batch size and hyperparameter configuration. Concrete subclasses must set the agent_type class variable and override get_training_fn().

The algorithm class is resolved lazily from agilerl.algorithms using the naming convention <Name>Spec -> <Name> (e.g. PPOSpec -> PPO). This avoids importing heavy dependencies at spec-import time.

classmethod algo_class() → type[RLAlgorithm | MultiAgentRLAlgorithm | LLMAlgorithm]¶: Lazily resolve the algorithm class from agilerl.algorithms.

build_algorithm() → AlgoT¶: Build the algorithm instance using spec fields + runtime args.

static get_training_fn() → Callable[..., tuple[PopulationT, list[float]]]¶

Return the training function for this algorithm.

Concrete specs must override this to return their training function (e.g. train_off_policy).

Returns:: Training function
Return type:: Callable[…, tuple[PopulationT, list[float]]]
Raises:: NotImplementedError – If the training function is not implemented.

get_training_kwargs(*, training: TrainingSpec, env_spec: EnvSpecT, memory: ReplayBufferT = None, n_step_memory: ReplayBufferT = None) → dict[str, Any]¶

Return additional kwargs for the training loop.

Parameters:

training (TrainingSpec) – Training specification.
env_spec (EnvSpecT) – Environment specification.
memory (ReplayBufferT | None) – Replay buffer instance.
n_step_memory (ReplayBufferT | None) – N-step replay buffer for combined PER + n-step setups.

Returns:

Extra keyword arguments for the training function.

Return type:

dict[str, Any]

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property name: str¶: Return the name of the algorithm.

class agilerl.models.algo.RLAlgorithmSpec(*, batch_size: Annotated[int, Ge(ge=1)] = 128, hp_config: Any | None = None, learn_step: Annotated[int, Ge(ge=1)] = 5, gamma: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.99)¶

Specification for single-agent reinforcement learning algorithms.

Extends AlgorithmSpec with single-agent specific fields like network configuration, learning step frequency, and discount factor.

Build a single-agent algorithm instance from spec fields.

Parameters:

observation_space (SupportedObservationSpace | None) – Observation space.
action_space (SupportedActionSpace | None) – Action space.
index (int | None) – Index of the algorithm in the population.
resume_from_checkpoint (str | None) – Path to resume from checkpoint.
device (str | torch.device) – Torch device. Defaults to “cpu”.
accelerator (Accelerator | None) – Accelerator object for distributed computing.

Returns:

Single-agent algorithm instance.

Return type:

RLAlgorithm

Raises:

ValueError – If observation_space, action_space, or index is None.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agilerl.models.algo.MultiAgentRLAlgorithmSpec(*, batch_size: Annotated[int, Ge(ge=1)] = 128, hp_config: Any | None = None, learn_step: Annotated[int, Ge(ge=1)] = 2048, gamma: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.99, torch_compiler: str | None = None)¶

Specification for multi-agent reinforcement learning algorithms.

Extends AlgorithmSpec with multi-agent specific fields and support for multiple observation/action spaces and agent IDs.

build_algorithm(observation_spaces: dict[str, SupportedObservationSpace] | None = None, action_spaces: dict[str, SupportedActionSpace] | None = None, index: int | None = None, resume_from_checkpoint: str | None = None, device: str | torch.device = 'cpu', accelerator: Accelerator | None = None) → MultiAgentRLAlgorithm¶

Build a multi-agent algorithm from spec fields.

Parameters:

observation_spaces (dict[str, SupportedObservationSpace] | None) – Per-agent observation spaces.
action_spaces (dict[str, SupportedActionSpace] | None) – Per-agent action spaces.
index (int | None) – Index of the algorithm in the population.
resume_from_checkpoint (str | None) – Path to resume from checkpoint.
device (str | torch.device) – Torch device. Defaults to “cpu”.
accelerator (Accelerator | None) – Accelerator object for distributed computing.

Returns:

Multi-agent algorithm instance.

Return type:

MultiAgentRLAlgorithm

Raises:

ValueError – If observation_spaces, action_spaces, or index is None.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agilerl.models.algo.LLMAlgorithmSpec(*, batch_size: Annotated[int, Ge(ge=1)] = 128, hp_config: Any | None = None, beta: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.001, max_grad_norm: Annotated[float, Ge(ge=0.0)] = 0.1, update_epochs: Annotated[int, Ge(ge=1)] = 1, reduce_memory_peak: bool = False, use_separate_reference_adapter: bool = False, calc_position_embeddings: bool = True, gradient_checkpointing: bool = True, use_liger_loss: bool = False, seed: int = 42, quantization: str | dict[str, Any] | None = None, activation_offload: bool = False, use_sequence_packing: bool = False, lora_target_scope: str | None = None, fused_logprobs_chunk_rows: int | None = None, fused_loss_chunk_rows: int | None = None, vllm_importance_sampling_correction: bool = True, vllm_importance_sampling_cap: Annotated[float, Ge(ge=0.0)] = 2.0, attn_implementation: str | None = None, pretrained_model_name_or_path: Annotated[str | None, MinLen(min_length=1)] = None, max_model_len: Annotated[int, Ge(ge=1)] = 1024, lora_config: Any | None = None)¶

Specification for LLM fine-tuning algorithms.

Extends AlgorithmSpec with LLM-specific fields including LoRA configuration, model parameters, and training hyperparameters.

Subclasses must set the env_type class variable to indicate which LLM gym type the algorithm requires ("reasoning" for ReasoningGym or "preference" for PreferenceGym).

Build an LLM algorithm instance from spec fields.

Parameters:

tokenizer (Any | None) – A HuggingFace AutoTokenizer instance.
index (int) – Index of the algorithm in the population.
resume_from_checkpoint (str | None) – Path to resume from checkpoint.
accelerator (Accelerator | None) – HuggingFace Accelerator instance.
device (str | torch.device) – Torch device. Defaults to “cpu”.
actor_network (Any | None) – Pre-built or cloned actor network. When provided, this is passed directly to the algorithm constructor instead of loading the model from pretrained_model_name_or_path.

Returns:

LLM algorithm instance.

Return type:

LLMAlgorithm

Raises:

ValueError – If tokenizer is None.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].