A3C (Asynchronous Advantage Actor‑Critic)¶

A3C combines the strengths of policy-based and value-based methods: multiple asynchronous workers explore the environment in parallel and update a shared (global) network using the advantage function. This PyTorch implementation uses multiprocessing with a shared global network and SharedAdam optimizer.

Components¶

Global Network: Shared parameters for both Actor (policy) and Critic (value) in a single Net module
Workers: Independent processes, each with its own environment and local network copy
SharedAdam: Optimizer with shared state across processes for consistent parameter updates
Advantage: TD-error used to weight policy gradients and update value function

Theory (based on the implementation)¶

Network Architecture¶

The Net module combines both Actor and Critic:

Actor branch: - Input → Linear(s_dim, 256) → ReLU6 - → mu: Linear(256, a_dim) → Tanh → scale by 2 (action range: [-2, 2]) - → sigma: Linear(256, a_dim) → Softplus + 0.001 (for numerical stability)

Critic branch: - Input → Linear(s_dim, 256) → ReLU6 - → value: Linear(256, 1)

Policy (Actor) — Gaussian Distribution¶

The actor outputs mean \(\mu(s)\) and standard deviation \(\sigma(s)\). Actions are sampled from:

\[ a \sim \mathcal{N}\big(\mu(s),\ \sigma^2(s)\big) \]

For multidimensional actions, an Independent distribution wraps the base Normal distribution.

Log-probability:

\[ \log \pi_\theta(a|s) = -\tfrac{1}{2}\,\frac{(a-\mu)^2}{\sigma^2} - \tfrac{1}{2}\,\log(2\pi\sigma^2) \]

Value Function (Critic)¶

The critic estimates state value \(V_\phi(s)\). The temporal difference error is:

\[ \text{TD} = R_t^{(n)} - V_\phi(s_t) \]

Value loss (mean squared error):

\[ \mathcal{L}_\text{value} = \mathbb{E}[\text{TD}^2] \]

N-Step Returns with Bootstrapping¶

The implementation uses proper n-step returns with bootstrapping:

\[ R_t^{(n)} = \sum_{k=0}^{n-1} \gamma^k r_{t+k} + \gamma^n V_\phi(s_{t+n}) \]

If the episode terminates, \(V_\phi(s_{t+n}) = 0\).

Loss Function¶

Policy loss (with entropy regularization):

\[ \mathcal{L}_\text{policy} = -\mathbb{E}\big[\log \pi_\theta(a_t|s_t) \cdot \text{TD} + 0.005 \cdot H[\pi]\big] \]

where \(H[\pi]\) is the entropy of the policy.

Total loss:

\[ \mathcal{L}_\text{total} = \mathbb{E}[\mathcal{L}_\text{policy} + \mathcal{L}_\text{value}] \]

The advantage (TD-error) is detached when computing policy loss to prevent backpropagation through the value function.

Asynchrony and Synchronization¶

The implementation uses torch.multiprocessing for parallel training:

Gradient computation: Each worker computes gradients on its local network
Push gradients: Local gradients are transferred to global network parameters (gp._grad = lp.grad)
Gradient clipping: Global gradients are clipped (max_norm=40.0) for stability
Optimizer step: SharedAdam updates global network parameters
Pull parameters: Local network loads updated global parameters (load_state_dict)

This push-and-pull happens every update_global_iter steps or when an episode ends.

Hyperparameters¶

lr: Learning rate for SharedAdam (default: 1e-4)
gamma: Discount factor (default: 0.99)
n_workers: Number of parallel workers (default: CPU count)
max_episodes: Total episodes to run (default: 10)
max_ep_step: Maximum steps per episode (default: 200)
update_global_iter: Frequency of global updates (default: 10)
Entropy coefficient: 0.005 (hardcoded in loss function)
Hidden layer size: 256 (hardcoded in Net architecture)

Training Algorithm (Pseudocode)¶

# Global setup
global_net = Net(s_dim, a_dim).share_memory()
optimizer = SharedAdam(global_net.parameters(), lr)

# Each worker runs in parallel:
def worker_process(worker_id):
    local_net = Net(s_dim, a_dim)
    local_net.load_state_dict(global_net.state_dict())  # Initial sync
    env = env_function(worker_id)

    while global_episodes < max_episodes:
        s = env.reset()
        buffer_s, buffer_a, buffer_r = [], [], []
        episode_reward = 0

        for t in range(max_ep_step):
            # Select action
            a = local_net.choose_action(s)
            s', r, done = env.step(clip(a, action_space))

            # Store transition
            buffer_s.append(s)
            buffer_a.append(a)
            buffer_r.append(r)
            episode_reward += r

            # Update condition
            if t % update_global_iter == 0 or done:
                # Compute n-step returns with bootstrapping
                if done:
                    v_s_ = 0
                else:
                    v_s_ = local_net.forward(s')[2]  # value estimate

                # Reverse accumulation
                returns = []
                for r in reversed(buffer_r):
                    v_s_ = r + gamma * v_s_
                    returns.insert(0, v_s_)

                # Compute loss
                loss = local_net.loss_func(buffer_s, buffer_a, returns)

                # Push gradients to global, pull updated parameters
                optimizer.zero_grad()
                loss.backward()
                transfer_gradients(local_net, global_net)
                clip_grad_norm(global_net.parameters(), max_norm=40.0)
                optimizer.step()
                local_net.load_state_dict(global_net.state_dict())

                # Clear buffers
                buffer_s, buffer_a, buffer_r = [], [], []

                if done:
                    record_episode(episode_reward)
                    break

            s = s'

Quick Start¶

Here's a complete example training A3C on the B747 environment to track a sinusoidal pitch angle:

import numpy as np
import torch
from tensoraerospace.envs.b747 import ImprovedB747Env
from tensoraerospace.signals.standard import sinusoid_vertical_shift
from tensoraerospace.utils import convert_tp_to_sec_tp, generate_time_period
from tensoraerospace.agent.a3c import Agent, setup_global_params

# Set random seed
SEED = 42
np.random.seed(SEED)
torch.manual_seed(SEED)

# Create time base and reference signal
dt = 0.1
_tp = generate_time_period(tn=20, dt=dt)
tps = convert_tp_to_sec_tp(_tp, dt=dt)
number_time_steps = len(_tp)

reference_signals = np.reshape(
    sinusoid_vertical_shift(
        tp=np.asarray(tps),
        frequency=0.05,
        amplitude=np.deg2rad(1.0),
        vertical_shift=0.0,
    ),
    [1, -1],
)

# Initial state: [u, w, q, theta]
init_state = np.array([0.0, 0.0, 0.0, 0.0], dtype=np.float32)

# Configure hyperparameters
setup_global_params(
    max_episodes=3000,
    max_ep_step=number_time_steps,
    gamma=0.99,
    update_global_iter=10,
    lr=1e-4,
)

# Environment factory
def make_env(worker_id: int):
    return ImprovedB747Env(
        initial_state=init_state,
        reference_signal=reference_signals,
        number_time_steps=number_time_steps,
        dt=dt,
        initial_elevator_deg=0.0,
    )

# Create and train agent
agent = Agent(
    env_function=make_env,
    gamma=0.99,
    n_workers=4,
    lr=1e-4,
    max_episodes=3000,
    max_ep_step=number_time_steps,
    update_global_iter=10,
    render=False,
    run_in_main=True,  # Set False for multiprocessing
    log_dir="runs/a3c_b747",
)

# Train
agent.train()

# Evaluate
eval_env = make_env(0)
obs, _ = eval_env.reset()
agent.gnet.eval()
episode_reward = 0.0

with torch.no_grad():
    terminated = truncated = False
    while not (terminated or truncated):
        obs_tensor = torch.from_numpy(np.array(obs).reshape(1, -1).astype(np.float32))
        mu, _, _ = agent.gnet.forward(obs_tensor)
        action = mu.cpu().numpy().reshape(-1)
        obs, reward, terminated, truncated, _ = eval_env.step(action)
        episode_reward += reward

print(f"Evaluation reward: {episode_reward:.4f}")
eval_env.close()
agent.close()

Monitoring with TensorBoard¶

tensorboard --logdir=runs/a3c_b747

Metrics include: - Loss/w*/total: Total loss per worker - Loss/w*/value: Value loss (TD error) - Loss/w*/policy: Policy loss - Loss/w*/entropy: Policy entropy - Performance/w*/episode_reward: Episode rewards - Performance/w*/moving_avg_reward: Moving average

Best Practices

Use run_in_main=True for notebooks/debugging
Set run_in_main=False and n_workers=8 for production training
Actions are automatically clipped to env.action_space.low/high
Sigma has minimum value 0.001 for numerical stability
Monitor TensorBoard for entropy collapse or value loss divergence

Advanced Example: Training on B747 Environment¶

This complete example demonstrates training an A3C agent on the ImprovedB747Env to track a sinusoidal pitch angle reference.

Setup and Environment Creation¶

import numpy as np
import torch
import matplotlib.pyplot as plt
from queue import Empty

from tensoraerospace.envs.b747 import ImprovedB747Env
from tensoraerospace.signals.standard import sinusoid_vertical_shift
from tensoraerospace.utils import convert_tp_to_sec_tp, generate_time_period
from tensoraerospace.agent.a3c import Agent, setup_global_params

# Set random seed for reproducibility
SEED = 42
np.random.seed(SEED)
torch.manual_seed(SEED)

# Create time base
dt = 0.1  # seconds
_tp = generate_time_period(tn=20, dt=dt)
tps = convert_tp_to_sec_tp(_tp, dt=dt)
number_time_steps = len(_tp)

print(f"Episode length: {number_time_steps} steps ({number_time_steps * dt:.1f} seconds)")

# Generate sinusoidal reference signal for pitch angle (theta)
reference_signals = np.reshape(
    sinusoid_vertical_shift(
        tp=np.asarray(tps),
        frequency=0.05,             # Hz
        amplitude=np.deg2rad(1.0),  # 1 degree amplitude
        vertical_shift=0.0,
    ),
    [1, -1],
)

# Define initial state: [u, w, q, theta]
init_state = np.array([0.0, 0.0, 0.0, 0.0], dtype=np.float32)

# Create environment
env = ImprovedB747Env(
    initial_state=init_state,
    reference_signal=reference_signals,
    number_time_steps=number_time_steps,
    dt=dt,
    initial_elevator_deg=0.0,
)

print(f"Observation space: {env.observation_space}")
print(f"Action space: {env.action_space}")

Configure and Create Agent¶

# Configure hyperparameters
setup_global_params(
    max_episodes=3000,
    max_ep_step=number_time_steps,
    gamma=0.99,
    update_global_iter=10,
    lr=1e-4,
)

# Environment factory function
def make_env(worker_id: int):
    """Create environment for each worker."""
    return ImprovedB747Env(
        initial_state=init_state,
        reference_signal=reference_signals,
        number_time_steps=number_time_steps,
        dt=dt,
        initial_elevator_deg=0.0,
    )

# Create A3C agent
agent = Agent(
    env_function=make_env,
    gamma=0.99,
    n_workers=4,              # Use 4 parallel workers
    lr=1e-4,
    max_episodes=3000,
    max_ep_step=number_time_steps,
    update_global_iter=10,
    render=False,
    run_in_main=True,         # Set to False for true multiprocessing
    log_dir="runs/a3c_b747",
)

print("A3C Agent created successfully!")

Train the Agent¶

import time

print("Starting A3C training...\n")

episode_rewards = []
start_time = time.time()

# Start training (synchronous if run_in_main=True)
agent.train()

# Collect rewards from queue
while True:
    try:
        r = agent.res_queue.get_nowait()
    except Empty:
        break
    if r is None:
        break
    episode_rewards.append(float(r))

training_time = time.time() - start_time
print(f"\nTraining completed in {training_time:.2f} seconds")
print(f"Total episodes: {len(episode_rewards)}")
print(f"Final reward (moving avg): {episode_rewards[-1]:.4f}")

Plot Training Progress¶

plt.figure(figsize=(12, 5))
plt.plot(episode_rewards, label='Moving avg reward', alpha=0.7)

# Add smoothed trend
window = 50
if len(episode_rewards) >= window:
    smoothed = np.convolve(episode_rewards, np.ones(window)/window, mode='valid')
    plt.plot(range(window-1, len(episode_rewards)), smoothed, 
             'r-', linewidth=2, label=f'Smoothed (MA{window})')

plt.grid(True, alpha=0.3)
plt.xlabel('Episode')
plt.ylabel('Reward (moving average)')
plt.title('A3C Training Progress on B747 Environment')
plt.legend()
plt.tight_layout()
plt.show()

Evaluate Trained Policy¶

# Deterministic evaluation using policy mean
eval_env = make_env(0)
obs, info = eval_env.reset()

agent.gnet.eval()
episode_reward = 0.0
terminated = False
truncated = False

with torch.no_grad():
    while not (terminated or truncated):
        obs_tensor = torch.from_numpy(np.array(obs).reshape(1, -1).astype(np.float32))
        mu, sigma, value = agent.gnet.forward(obs_tensor)

        # Use mean for deterministic policy
        action = mu.cpu().numpy().reshape(-1)

        obs, reward, terminated, truncated, info = eval_env.step(action)
        episode_reward += float(reward)

print(f"Deterministic evaluation reward: {episode_reward:.4f}")

# Plot pitch angle tracking
eval_env.unwrapped.model.plot_transient_process(
    'theta',
    tps,
    reference_signals[0],
    to_deg=True,
    figsize=(15, 4)
)

eval_env.close()
agent.close()

Monitor with TensorBoard¶

tensorboard --logdir=runs/a3c_b747

Available metrics: - Loss/w*/total: Total loss per worker - Loss/w*/value: Value function loss (TD error squared) - Loss/w*/policy: Policy loss (negative expected advantage) - Loss/w*/entropy: Policy entropy (exploration measure) - Performance/w*/episode_reward: Raw episode rewards - Performance/w*/moving_avg_reward: Exponentially weighted moving average

Expected Results¶

After 3000 episodes of training: - Agent learns to track sinusoidal pitch reference with ~1° amplitude - Final moving average reward: approximately -1.6 to -2.0 - Pitch tracking error: < 0.5° RMS

Tips for Better Performance¶

Increase training duration: 10000+ episodes for better convergence
Tune hyperparameters:
Lower lr (5e-5) for more stable learning
Increase update_global_iter (20-30) for smoother gradients
Use multiple workers: Set run_in_main=False and n_workers=8 for faster training
Adjust reference signal: Try different frequencies and amplitudes
Monitor TensorBoard: Watch for entropy collapse or value loss divergence

Unified training interface¶

A3C follows the shared unified train() signature from BaseRLModel:

agent.train(
    num_episodes=500,   # optional: overrides self.max_episodes
    max_steps=200,      # optional: overrides self.max_ep_step
)

Calling agent.train() with no arguments still works and preserves the values passed to the Agent constructor (max_episodes, max_ep_step). The method returns a dict with global_ep, global_step and global_ep_r.

API Reference¶

Agent¶

`Agent(env_function, gamma=DEFAULT_GAMMA, n_workers=None, lr=DEFAULT_LR, max_episodes=DEFAULT_MAX_EP, max_ep_step=DEFAULT_MAX_EP_STEP, update_global_iter=DEFAULT_UPDATE_GLOBAL_ITER, render=False, run_in_main=False, log_dir='runs/a3c', wandb_project=None, wandb_entity=None, wandb_run_name=None, wandb_tags=None, wandb_config=None)` ¶

Simple A3C Agent wrapper around multiprocessing Workers.

Parameters:

Name	Type	Description	Default
`env_function`	`Callable[[int], Env]`	callable that returns a new env for a given worker id.	required
`gamma`	`float`	discount factor.	`DEFAULT_GAMMA`
`n_workers`	`Optional[int]`	number of worker processes.	`None`
`lr`	`float`	learning rate for SharedAdam.	`DEFAULT_LR`
`max_episodes`	`int`	total episodes to run per global counter.	`DEFAULT_MAX_EP`
`max_ep_step`	`int`	max steps per episode.	`DEFAULT_MAX_EP_STEP`
`update_global_iter`	`int`	frequency to push/pull.	`DEFAULT_UPDATE_GLOBAL_ITER`
`render`	`bool`	render from worker w0 (optional).	`False`

Note: For unit tests or debugging, set run_in_main=True to avoid spawning processes. The single worker will run in the main process.

Configure A3C agent wrapper.

Parameters:

Name	Type	Description	Default
`env_function`	`Callable[[int], Env]`	Factory returning an environment per worker id.	required
`gamma`	`float`	Discount factor.	`DEFAULT_GAMMA`
`n_workers`	`Optional[int]`	Number of worker processes; defaults to CPU count.	`None`
`lr`	`float`	Learning rate for optimizer.	`DEFAULT_LR`
`max_episodes`	`int`	Total episodes to run.	`DEFAULT_MAX_EP`
`max_ep_step`	`int`	Max steps per episode.	`DEFAULT_MAX_EP_STEP`
`update_global_iter`	`int`	Sync frequency for global net updates.	`DEFAULT_UPDATE_GLOBAL_ITER`
`render`	`bool`	Whether to render from worker 0.	`False`
`run_in_main`	`bool`	If True, run worker inline for debugging/tests.	`False`
`log_dir`	`str`	TensorBoard log directory.	`'runs/a3c'`

`train(num_episodes=None, *, max_steps=None, save_best=False, save_path=None, verbose=True, **kwargs)` ¶

Launch training across worker processes (unified interface).

Parameters:

Name	Type	Description	Default
`num_episodes`	`Optional[int]`	Override for `self.max_episodes`; when `None` the value supplied at construction is used, preserving the original no-argument call style.	`None`
`max_steps`	`Optional[int]`	Override for `self.max_ep_step`.	`None`
`save_best`	`bool`	Reserved for API consistency.	`False`
`save_path`	`Optional[str]`	Reserved for API consistency.	`None`
`verbose`	`bool`	Reserved for symmetry.	`True`
`**kwargs`	`Any`	Currently unused.	`{}`

Returns:

Name	Type	Description
`dict`	`dict`	Summary dictionary (`global_ep`, `global_step`,
	`dict`	`global_ep_r`).

`close()` ¶

Close TensorBoard writer and cleanup resources.

`get_param_env()` ¶

Return serializable configuration of the agent.

`save(path=None, save_gradients=False)` ¶

Save A3C agent to the specified directory.

Saves the global actor-critic network and configuration. Optionally saves the shared optimizer state for resuming training.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path \| None`	Base save directory. If None, saves to the current working directory.	`None`
`save_gradients`	`bool`	If True, also save optimizer state dict.	`False`

Returns:

Name	Type	Description
`Path`	`Path`	The directory where the model was saved.

`load(path, env_function=None, load_gradients=False)` `classmethod` ¶

Load an A3C agent from a checkpoint directory.

Parameters:

Name	Type	Description	Default
`path`	`Union[str, Path]`	Directory containing saved model files.	required
`env_function`	`Optional[Callable[[int], Env]]`	Factory returning an environment per worker id. Required because A3C needs environments for its workers.	`None`
`load_gradients`	`bool`	If True, restore optimizer state.	`False`

Returns:

Name	Type	Description
`Agent`	`'Agent'`	Reconstructed agent.

`from_pretrained(repo_name, env_function=None, access_token=None, version=None, load_gradients=False)` `classmethod` ¶

Load pretrained model from a local directory or Hugging Face Hub.

Parameters:

Name	Type	Description	Default
`repo_name`	`str`	Path to a local folder or a Hugging Face repo id (e.g. `"namespace/repo_name"`).	required
`env_function`	`Optional[Callable[[int], Env]]`	Factory returning an environment per worker id (required).	`None`
`access_token`	`Optional[str]`	Hugging Face access token for private repos.	`None`
`version`	`Optional[str]`	Revision / branch / tag on Hugging Face.	`None`
`load_gradients`	`bool`	Restore optimizer state for continued training.	`False`

Returns:

Name	Type	Description
`Agent`	`'Agent'`	Initialized agent.

`publish_to_hub(repo_name, folder_path, access_token=None)` ¶

Upload a saved model folder to Hugging Face Hub.

Parameters:

Name	Type	Description	Default
`repo_name`	`str`	Repository id on Hugging Face (e.g. `"user/my-a3c"`).	required
`folder_path`	`Union[str, Path]`	Local folder produced by :meth:`save`.	required
`access_token`	`Optional[str]`	Hugging Face access token.	`None`

Worker¶

`Worker(gnet, opt, global_ep, global_ep_r, res_queue, name, num_actions, num_observations, MAX_EP, MAX_EP_STEP, GAMMA, update_global_iter, env_function=None, env=None, render=False, writer=None, global_step=None, global_env_step=None)` ¶

Bases: Process

Worker process class for asynchronous agent training.

Parameters:

Name	Type	Description	Default
`env`	`Env`	Environment for agent training.	`None`
`gnet`	`Net`	Global model for shared training.	required
`opt`	`SharedAdam`	Optimizer for global network.	required
`global_ep`	`Value`	Global episode counter.	required
`global_ep_r`	`Value`	Global total reward counter across all processes.	required
`res_queue`	`Queue`	Queue for storing results.	required
`name`	`int`	Process number.	required
`num_actions`	`int`	Number of possible actions in the environment.	required
`num_observations`	`int`	Number of observations (state variables) in the environment.	required
`MAX_EP`	`int`	Maximum number of episodes.	required
`MAX_EP_STEP`	`int`	Maximum number of steps per episode.	required
`GAMMA`	`float`	Discount factor for future rewards.	required
`update_global_iter`	`int`	Frequency of global model updates.	required
`render`	`bool`	Whether to render the environment. Defaults to False.	`False`
`writer`	`Optional[SummaryWriter]`	TensorBoard writer. Defaults to None.	`None`
`global_step`	`Optional[Value]`	Global step counter. Defaults to None.	`None`

Attributes:

Name	Type	Description
`name`	`str`	Unique process name.
`g_ep`	`Value`	Global episode counter.
`g_ep_r`	`Value`	Global total reward counter.
`res_queue`	`Queue`	Results queue.
`gnet`	`Net`	Global neural network.
`opt`	`SharedAdam`	Optimizer for updating global network.
`lnet`	`Net`	Local neural network.
`env`	`Env`	OpenAI Gym environment.
`gamma`	`float`	Discount factor.
`max_ep`	`int`	Maximum number of episodes.
`max_ep_step`	`int`	Maximum number of steps per episode.
`update_global_iter`	`int`	Frequency of global network updates.
`render`	`bool`	Whether to render the environment.
`writer`	`Optional[SummaryWriter]`	TensorBoard writer.
`global_step`	`Optional[Value]`	Global step counter.

Initialize worker process.

Parameters:

Name	Type	Description	Default
`env_function`	`Optional[Callable[[int], Env]]`	Factory callable taking worker id and returning a fresh env. The env is created inside `run()` so that under fork multiprocessing each worker gets its own env (avoids sharing a single env object across processes). This is the preferred way to supply an env.	`None`
`env`	`Optional[Env]`	Legacy/direct env instance. Provided for backward compatibility (and for single-process tests); DO NOT use this path with real multi-process training because a single env would be shared across forked workers.	`None`
`gnet`	`Net`	Global shared network.	required
`opt`	`SharedAdam`	Shared optimizer.	required
`global_ep`	`Any`	Shared episode counter.	required
`global_ep_r`	`Any`	Shared reward accumulator.	required
`res_queue`	`Queue`	Queue for results.	required
`name`	`int`	Worker id.	required
`num_actions`	`int`	Action dimension.	required
`num_observations`	`int`	Observation dimension.	required
`MAX_EP`	`int`	Max episodes to run.	required
`MAX_EP_STEP`	`int`	Max steps per episode.	required
`GAMMA`	`float`	Discount factor.	required
`update_global_iter`	`int`	Steps between syncs with global net.	required
`render`	`bool`	Whether to render (only worker 0 typically).	`False`
`writer`	`Optional[MetricWriter]`	TensorBoard writer for metrics.	`None`
`global_step`	`Any \| None`	Shared global update counter (push_and_pull count).	`None`
`global_env_step`	`Any \| None`	Shared global env-step counter (incremented once per `env.step`); used as the canonical TensorBoard X-axis for all scalar writes.	`None`

`run()` ¶

Execute worker process containing agent training.

Network¶

`Net(s_dim, a_dim)` ¶

Bases: Module

Neural network for policy and value function approximation in RL.

Parameters:

Name	Type	Description	Default
`s_dim`	`int`	State space dimension.	required
`a_dim`	`int`	Action space dimension.	required

Attributes:

Name	Type	Description
`s_dim`	`int`	State space dimension.
`a_dim`	`int`	Action space dimension.
`a1`	`Linear`	First policy layer.
`mu`	`Linear`	Mean layer of policy distribution.
`sigma`	`Linear`	Standard deviation layer of policy distribution.
`c1`	`Linear`	First value function layer.
`v`	`Linear`	Value function output layer.
`distribution`	`Distribution`	Distribution for modeling agent actions.

Create network layers.

Parameters:

Name	Type	Description	Default
`s_dim`	`int`	State dimension.	required
`a_dim`	`int`	Action dimension.	required

`forward(x)` ¶

Perform one forward pass.

Parameters:

Name	Type	Description	Default
`x`	`Tensor`	Input data, environment state.	required

Returns:

Type	Description
`Tensor`	Tuple[torch.Tensor, torch.Tensor, torch.Tensor]: Predicted mu, sigma,
`Tensor`	and value for the given state.

`choose_action(s)` ¶

Select agent action based on current state.

Parameters:

Name	Type	Description	Default
`s`	`Tensor`	Current environment state.	required

Returns:

Type	Description
`ndarray`	np.ndarray: Selected action.

`loss_func(s, a, v_t)` ¶

Compute loss function for network training.

Parameters:

Name	Type	Description	Default
`s`	`Tensor`	States.	required
`a`	`Tensor`	Actions.	required
`v_t`	`Tensor`	Target state value function values.	required

Returns:

Type	Description
`Tensor`	torch.Tensor: Loss function value.

Optimizer¶

`SharedAdam(params, lr=0.001, betas=(0.9, 0.99), eps=1e-08, weight_decay=0)` ¶

Bases: Adam

Adam optimizer with shared state for multi-process training.

This optimizer stores its internal state tensors (step counter, exp_avg, exp_avg_sq) in shared memory so multiple worker processes can update a single set of parameters consistently.

Parameters:

Name	Type	Description	Default
`params`	`Iterable[Parameter]`	Parameters to optimize.	required
`lr`	`float`	Learning rate. Defaults to `1e-3`.	`0.001`
`betas`	`Tuple[float, float]`	Coefficients used for computing running averages of gradient and its square. Defaults to `(0.9, 0.99)`.	`(0.9, 0.99)`
`eps`	`float`	Term added to the denominator for numerical stability. Defaults to `1e-8`.	`1e-08`
`weight_decay`	`float`	Weight decay (L2 penalty). Defaults to `0`.	`0`

Initialize shared Adam optimizer.

Parameters:

Name	Type	Description	Default
`params`	`Iterable[Parameter]`	Iterable of parameters to optimize.	required
`lr`	`float`	Learning rate.	`0.001`
`betas`	`Tuple[float, float]`	Beta coefficients for Adam moments.	`(0.9, 0.99)`
`eps`	`float`	Numerical stability term.	`1e-08`
`weight_decay`	`float`	L2 weight decay.	`0`

Utilities¶

`setup_global_params(*, max_episodes=DEFAULT_MAX_EP, max_ep_step=DEFAULT_MAX_EP_STEP, gamma=DEFAULT_GAMMA, update_global_iter=DEFAULT_UPDATE_GLOBAL_ITER, lr=DEFAULT_LR)` ¶

Update defaults used by Agent.

This matches the previous TF API name to ease migration.

Implementation Details¶

Key Features¶

Unified Network: Single Net module with shared layers, reducing memory overhead
ReLU6 Activation: More stable gradients compared to standard ReLU
Gradient Clipping: Max norm of 40.0 prevents exploding gradients
Entropy Regularization: Coefficient of 0.005 encourages exploration
SharedAdam: Optimizer state shared across processes for consistent updates
Proper Bootstrapping: N-step returns include terminal state value when episode continues

Advantages over Synchronous Methods¶

Parallel Experience Collection: Multiple workers explore simultaneously
Decorrelated Samples: Different workers in different states reduce correlation
No Replay Buffer: Online learning reduces memory requirements
Natural Exploration: Asynchrony provides diversity without ε-greedy

Debugging Tips¶

Use run_in_main=True to run single worker without multiprocessing
Check TensorBoard for loss divergence or entropy collapse
Reduce lr if training is unstable
Increase update_global_iter for more stable gradients
Ensure environment is properly seeded for reproducibility

References¶

Asynchronous Methods for Deep Reinforcement Learning (Mnih et al., 2016)
PyTorch Multiprocessing Best Practices

Tested Environments¶

Unity ML-Agents environments
Gymnasium continuous control tasks
TensorAeroSpace LinearLongitudinal* environments
Custom aerospace control environments

A3C (Asynchronous Advantage Actor‑Critic)¶

Components¶

Theory (based on the implementation)¶

Network Architecture¶

Policy (Actor) — Gaussian Distribution¶

Value Function (Critic)¶

N-Step Returns with Bootstrapping¶

Loss Function¶

Asynchrony and Synchronization¶

Hyperparameters¶

Training Algorithm (Pseudocode)¶

Quick Start¶

Monitoring with TensorBoard¶

Advanced Example: Training on B747 Environment¶

Setup and Environment Creation¶

Configure and Create Agent¶

Train the Agent¶

Plot Training Progress¶

Evaluate Trained Policy¶

Monitor with TensorBoard¶

Expected Results¶

Tips for Better Performance¶

Unified training interface¶

API Reference¶

Agent¶

train(num_episodes=None, *, max_steps=None, save_best=False, save_path=None, verbose=True, **kwargs) ¶

close() ¶

get_param_env() ¶

save(path=None, save_gradients=False) ¶

load(path, env_function=None, load_gradients=False) classmethod ¶

from_pretrained(repo_name, env_function=None, access_token=None, version=None, load_gradients=False) classmethod ¶

publish_to_hub(repo_name, folder_path, access_token=None) ¶

Worker¶

Worker(gnet, opt, global_ep, global_ep_r, res_queue, name, num_actions, num_observations, MAX_EP, MAX_EP_STEP, GAMMA, update_global_iter, env_function=None, env=None, render=False, writer=None, global_step=None, global_env_step=None) ¶

run() ¶

Network¶

Net(s_dim, a_dim) ¶

forward(x) ¶

choose_action(s) ¶

loss_func(s, a, v_t) ¶

Optimizer¶

SharedAdam(params, lr=0.001, betas=(0.9, 0.99), eps=1e-08, weight_decay=0) ¶

Utilities¶

setup_global_params(*, max_episodes=DEFAULT_MAX_EP, max_ep_step=DEFAULT_MAX_EP_STEP, gamma=DEFAULT_GAMMA, update_global_iter=DEFAULT_UPDATE_GLOBAL_ITER, lr=DEFAULT_LR) ¶

Implementation Details¶

Key Features¶

Advantages over Synchronous Methods¶

Debugging Tips¶

References¶

Tested Environments¶

`train(num_episodes=None, *, max_steps=None, save_best=False, save_path=None, verbose=True, **kwargs)` ¶

`close()` ¶

`get_param_env()` ¶

`save(path=None, save_gradients=False)` ¶

`load(path, env_function=None, load_gradients=False)` `classmethod` ¶

`from_pretrained(repo_name, env_function=None, access_token=None, version=None, load_gradients=False)` `classmethod` ¶

`publish_to_hub(repo_name, folder_path, access_token=None)` ¶

`Worker(gnet, opt, global_ep, global_ep_r, res_queue, name, num_actions, num_observations, MAX_EP, MAX_EP_STEP, GAMMA, update_global_iter, env_function=None, env=None, render=False, writer=None, global_step=None, global_env_step=None)` ¶

`run()` ¶

`Net(s_dim, a_dim)` ¶

`forward(x)` ¶

`choose_action(s)` ¶

`loss_func(s, a, v_t)` ¶

`SharedAdam(params, lr=0.001, betas=(0.9, 0.99), eps=1e-08, weight_decay=0)` ¶

`setup_global_params(*, max_episodes=DEFAULT_MAX_EP, max_ep_step=DEFAULT_MAX_EP_STEP, gamma=DEFAULT_GAMMA, update_global_iter=DEFAULT_UPDATE_GLOBAL_ITER, lr=DEFAULT_LR)` ¶