Chapter 5: Actor-Critic Methods

WIA-AI-025 | Best of Both Worlds 🎭

5.1 Introduction to Actor-Critic

Actor-Critic methods combine the best of policy gradient and value-based approaches. The "actor" updates the policy while the "critic" evaluates actions by estimating the value function.

Key Insight:

Actor: Policy π_θ(a|s) - decides which action to take
Critic: Value function V_w(s) or Q_w(s,a) - evaluates how good the action is

Advantages of Actor-Critic

5.2 Basic Actor-Critic Algorithm

The simplest actor-critic uses a learned value function as the baseline:

Algorithm: One-Step Actor-Critic

Initialize actor π_θ and critic V_w

For each episode:
    s = env.reset()

    while not done:
        a ~ π_θ(·|s)  // Actor chooses action
        s', r, done = env.step(a)

        δ = r + γ V_w(s') - V_w(s)  // TD error (Critic)

        w ← w + α_w δ ∇_w V_w(s)       // Update Critic
        θ ← θ + α_θ δ ∇_θ log π_θ(a|s) // Update Actor

        s ← s'

Implementation

import torch
import torch.nn as nn
import torch.optim as optim

class Actor(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super(Actor, self).__init__()
        self.network = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )

    def forward(self, state):
        return self.network(state)

class Critic(nn.Module):
    def __init__(self, state_dim, hidden_dim=128):
        super(Critic, self).__init__()
        self.network = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

    def forward(self, state):
        return self.network(state)

class ActorCritic:
    def __init__(self, state_dim, action_dim, lr_actor=1e-3, lr_critic=1e-3, gamma=0.99):
        self.actor = Actor(state_dim, action_dim)
        self.critic = Critic(state_dim)

        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=lr_actor)
        self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=lr_critic)

        self.gamma = gamma

    def select_action(self, state):
        state = torch.FloatTensor(state).unsqueeze(0)
        probs = self.actor(state)
        dist = torch.distributions.Categorical(probs)
        action = dist.sample()
        return action.item(), dist.log_prob(action)

    def update(self, state, action_log_prob, reward, next_state, done):
        state = torch.FloatTensor(state).unsqueeze(0)
        next_state = torch.FloatTensor(next_state).unsqueeze(0)

        # Compute TD error
        value = self.critic(state)
        next_value = self.critic(next_state)
        td_target = reward + self.gamma * next_value * (1 - done)
        td_error = td_target - value

        # Update Critic
        critic_loss = td_error.pow(2)
        self.critic_optimizer.zero_grad()
        critic_loss.backward()
        self.critic_optimizer.step()

        # Update Actor
        actor_loss = -action_log_prob * td_error.detach()
        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        self.actor_optimizer.step()

5.3 Advantage Actor-Critic (A2C)

A2C uses the advantage function instead of Q-values:

A(s,a) = Q(s,a) - V(s)
       = r + γV(s') - V(s)  // Using TD(0)

Advantage tells us how much better action a is compared to the average action in state s.

Why Advantages?

Benefits of Advantage Functions:
  • Reduces variance by subtracting baseline V(s)
  • Centers values around zero (positive = good, negative = bad)
  • More stable gradients for policy updates
  • Only need to learn V(s), not Q(s,a)

A2C with Multiple Workers

def train_a2c(env, num_workers=4, steps_per_update=5):
    """
    A2C with parallel environment workers.
    """
    agent = ActorCritic(state_dim, action_dim)

    # Create multiple environment instances
    envs = [make_env() for _ in range(num_workers)]

    for update in range(num_updates):
        states_list = []
        actions_list = []
        rewards_list = []
        next_states_list = []
        dones_list = []

        # Collect experience from all workers
        for env in envs:
            state = env.reset()

            for step in range(steps_per_update):
                action, log_prob = agent.select_action(state)
                next_state, reward, done, _ = env.step(action)

                states_list.append(state)
                actions_list.append(log_prob)
                rewards_list.append(reward)
                next_states_list.append(next_state)
                dones_list.append(done)

                state = next_state if not done else env.reset()

        # Batch update
        states = torch.FloatTensor(states_list)
        actions = torch.stack(actions_list)
        rewards = torch.FloatTensor(rewards_list)
        next_states = torch.FloatTensor(next_states_list)
        dones = torch.FloatTensor(dones_list)

        # Compute advantages
        values = agent.critic(states)
        next_values = agent.critic(next_states)
        advantages = rewards + agent.gamma * next_values * (1 - dones) - values

        # Update networks
        critic_loss = advantages.pow(2).mean()
        actor_loss = -(actions * advantages.detach()).mean()

        total_loss = actor_loss + 0.5 * critic_loss

        agent.actor_optimizer.zero_grad()
        agent.critic_optimizer.zero_grad()
        total_loss.backward()
        agent.actor_optimizer.step()
        agent.critic_optimizer.step()

    return agent

5.4 Asynchronous Advantage Actor-Critic (A3C)

A3C parallelizes learning across multiple CPU threads, each running its own environment:

Key Innovation:

Each worker independently updates a shared global network asynchronously. This provides diverse experience and stabilizes learning without needing experience replay.

A3C Architecture

import threading
import torch.multiprocessing as mp

class A3C:
    def __init__(self, state_dim, action_dim, num_workers=4):
        # Shared global network
        self.global_actor = Actor(state_dim, action_dim).share_memory()
        self.global_critic = Critic(state_dim).share_memory()

        self.num_workers = num_workers

    def train(self, env_fn, max_episodes=1000):
        # Shared optimizer
        optimizer = torch.optim.Adam([
            {'params': self.global_actor.parameters()},
            {'params': self.global_critic.parameters()}
        ])

        # Create worker processes
        processes = []
        for rank in range(self.num_workers):
            p = mp.Process(target=self.worker_process,
                          args=(rank, env_fn, optimizer, max_episodes))
            p.start()
            processes.append(p)

        for p in processes:
            p.join()

    def worker_process(self, rank, env_fn, optimizer, max_episodes):
        env = env_fn()

        # Local networks
        local_actor = Actor(state_dim, action_dim)
        local_critic = Critic(state_dim)

        for episode in range(max_episodes):
            # Sync with global
            local_actor.load_state_dict(self.global_actor.state_dict())
            local_critic.load_state_dict(self.global_critic.state_dict())

            # Collect trajectory
            states, actions, rewards = [], [], []
            state = env.reset()
            done = False

            while not done:
                action, log_prob = self.select_action(state, local_actor)
                next_state, reward, done, _ = env.step(action)

                states.append(state)
                actions.append(log_prob)
                rewards.append(reward)

                state = next_state

            # Compute returns
            returns = self.compute_returns(rewards)

            # Compute gradients
            actor_loss = 0
            critic_loss = 0

            for s, a, R in zip(states, actions, returns):
                s = torch.FloatTensor(s).unsqueeze(0)
                value = local_critic(s)
                advantage = R - value.item()

                actor_loss += -a * advantage
                critic_loss += (R - value).pow(2)

            total_loss = actor_loss + 0.5 * critic_loss

            # Update global networks
            optimizer.zero_grad()
            total_loss.backward()

            # Copy gradients to global
            for local_param, global_param in zip(local_actor.parameters(),
                                                 self.global_actor.parameters()):
                global_param._grad = local_param.grad

            for local_param, global_param in zip(local_critic.parameters(),
                                                 self.global_critic.parameters()):
                global_param._grad = local_param.grad

            optimizer.step()

5.5 Soft Actor-Critic (SAC)

SAC is an off-policy actor-critic algorithm that maximizes both reward and entropy:

J(π) = E[Σ γ^t (r_t + α H(π(·|s_t)))]

where:
- H(π) = -E[log π(a|s)] is the entropy
- α is the temperature parameter
- Higher entropy → more exploration

SAC Key Features

SAC Implementation

class SAC:
    def __init__(self, state_dim, action_dim, hidden_dim=256):
        # Actor (stochastic policy)
        self.actor = GaussianPolicy(state_dim, action_dim, hidden_dim)

        # Two Q-functions (critics) to reduce overestimation
        self.critic1 = QNetwork(state_dim, action_dim, hidden_dim)
        self.critic2 = QNetwork(state_dim, action_dim, hidden_dim)

        # Target critics
        self.critic1_target = QNetwork(state_dim, action_dim, hidden_dim)
        self.critic2_target = QNetwork(state_dim, action_dim, hidden_dim)

        self.critic1_target.load_state_dict(self.critic1.state_dict())
        self.critic2_target.load_state_dict(self.critic2.state_dict())

        # Automatic entropy tuning
        self.target_entropy = -action_dim
        self.log_alpha = torch.zeros(1, requires_grad=True)
        self.alpha = self.log_alpha.exp()

        # Optimizers
        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=3e-4)
        self.critic1_optimizer = optim.Adam(self.critic1.parameters(), lr=3e-4)
        self.critic2_optimizer = optim.Adam(self.critic2.parameters(), lr=3e-4)
        self.alpha_optimizer = optim.Adam([self.log_alpha], lr=3e-4)

        self.replay_buffer = ReplayBuffer(capacity=1000000)

    def select_action(self, state, evaluate=False):
        state = torch.FloatTensor(state).unsqueeze(0)
        if evaluate:
            _, _, action = self.actor.sample(state)
        else:
            action, _, _ = self.actor.sample(state)
        return action.detach().cpu().numpy()[0]

    def update(self, batch_size=256):
        # Sample from replay buffer
        state, action, reward, next_state, done = \
            self.replay_buffer.sample(batch_size)

        with torch.no_grad():
            # Sample next action from current policy
            next_action, next_log_prob, _ = self.actor.sample(next_state)

            # Compute target Q-value
            target_q1 = self.critic1_target(next_state, next_action)
            target_q2 = self.critic2_target(next_state, next_action)
            target_q = torch.min(target_q1, target_q2) - self.alpha * next_log_prob
            target_q = reward + (1 - done) * 0.99 * target_q

        # Update Q-functions
        current_q1 = self.critic1(state, action)
        current_q2 = self.critic2(state, action)

        critic1_loss = F.mse_loss(current_q1, target_q)
        critic2_loss = F.mse_loss(current_q2, target_q)

        self.critic1_optimizer.zero_grad()
        critic1_loss.backward()
        self.critic1_optimizer.step()

        self.critic2_optimizer.zero_grad()
        critic2_loss.backward()
        self.critic2_optimizer.step()

        # Update policy
        new_action, log_prob, _ = self.actor.sample(state)
        q1_new = self.critic1(state, new_action)
        q2_new = self.critic2(state, new_action)
        q_new = torch.min(q1_new, q2_new)

        actor_loss = (self.alpha * log_prob - q_new).mean()

        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        self.actor_optimizer.step()

        # Update temperature (alpha)
        alpha_loss = -(self.log_alpha * (log_prob + self.target_entropy).detach()).mean()

        self.alpha_optimizer.zero_grad()
        alpha_loss.backward()
        self.alpha_optimizer.step()

        self.alpha = self.log_alpha.exp()

        # Soft update target networks
        self.soft_update(self.critic1_target, self.critic1, tau=0.005)
        self.soft_update(self.critic2_target, self.critic2, tau=0.005)

    def soft_update(self, target, source, tau):
        for target_param, param in zip(target.parameters(), source.parameters()):
            target_param.data.copy_(tau * param.data + (1.0 - tau) * target_param.data)

5.6 Twin Delayed DDPG (TD3)

TD3 improves DDPG with three key modifications:

  1. Twin Critics: Two Q-networks, use minimum for updates
  2. Delayed Policy Updates: Update policy less frequently than critics
  3. Target Policy Smoothing: Add noise to target actions

TD3 Algorithm

class TD3:
    def __init__(self, state_dim, action_dim, max_action):
        # Actor
        self.actor = DeterministicActor(state_dim, action_dim, max_action)
        self.actor_target = DeterministicActor(state_dim, action_dim, max_action)
        self.actor_target.load_state_dict(self.actor.state_dict())

        # Twin critics
        self.critic1 = Critic(state_dim, action_dim)
        self.critic2 = Critic(state_dim, action_dim)
        self.critic1_target = Critic(state_dim, action_dim)
        self.critic2_target = Critic(state_dim, action_dim)

        self.critic1_target.load_state_dict(self.critic1.state_dict())
        self.critic2_target.load_state_dict(self.critic2.state_dict())

        self.max_action = max_action
        self.policy_delay = 2
        self.total_it = 0

    def update(self, replay_buffer, batch_size=256):
        self.total_it += 1

        state, action, reward, next_state, done = replay_buffer.sample(batch_size)

        with torch.no_grad():
            # Target policy smoothing
            noise = (torch.randn_like(action) * 0.2).clamp(-0.5, 0.5)
            next_action = (self.actor_target(next_state) + noise).clamp(-self.max_action, self.max_action)

            # Compute target Q
            target_q1 = self.critic1_target(next_state, next_action)
            target_q2 = self.critic2_target(next_state, next_action)
            target_q = torch.min(target_q1, target_q2)
            target_q = reward + (1 - done) * 0.99 * target_q

        # Update critics
        current_q1 = self.critic1(state, action)
        current_q2 = self.critic2(state, action)

        critic1_loss = F.mse_loss(current_q1, target_q)
        critic2_loss = F.mse_loss(current_q2, target_q)

        self.critic1_optimizer.zero_grad()
        critic1_loss.backward()
        self.critic1_optimizer.step()

        self.critic2_optimizer.zero_grad()
        critic2_loss.backward()
        self.critic2_optimizer.step()

        # Delayed policy updates
        if self.total_it % self.policy_delay == 0:
            # Update actor
            actor_loss = -self.critic1(state, self.actor(state)).mean()

            self.actor_optimizer.zero_grad()
            actor_loss.backward()
            self.actor_optimizer.step()

            # Soft update targets
            for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()):
                target_param.data.copy_(0.005 * param.data + 0.995 * target_param.data)

            for param, target_param in zip(self.critic1.parameters(), self.critic1_target.parameters()):
                target_param.data.copy_(0.005 * param.data + 0.995 * target_param.data)

            for param, target_param in zip(self.critic2.parameters(), self.critic2_target.parameters()):
                target_param.data.copy_(0.005 * param.data + 0.995 * target_param.data)

5.7 Comparison of Actor-Critic Methods

Algorithm On/Off-Policy Action Space Key Feature
A2C On-policy Discrete/Continuous Synchronous parallel workers
A3C On-policy Discrete/Continuous Asynchronous workers
DDPG Off-policy Continuous Deterministic policy + replay
TD3 Off-policy Continuous Twin critics + delayed updates
SAC Off-policy Continuous Maximum entropy + auto-tuning

5.8 Practical Tips

Choosing an Algorithm

Common Pitfalls

Summary

  • Actor-Critic combines policy gradient (actor) and value function (critic)
  • Advantages reduce variance compared to pure policy gradient
  • A2C uses synchronous parallel workers for data collection
  • A3C uses asynchronous updates without experience replay
  • SAC maximizes entropy for exploration and is highly sample efficient
  • TD3 improves DDPG with twin critics and delayed updates
  • Off-policy methods (SAC, TD3) are more sample efficient
  • On-policy methods (A2C, A3C) are simpler but need more samples
  • Choice depends on action space, sample efficiency needs, and stability requirements
  • Actor-critic is the foundation for many modern RL successes

Review Questions

1. What are the roles of the actor and critic?

Answer: The actor learns the policy π(a|s) and decides actions. The critic learns the value function V(s) or Q(s,a) and evaluates how good actions are.

2. Why use advantage functions instead of Q-values?

Answer: Advantages reduce variance by subtracting a baseline V(s), provide centered values, and enable learning only V(s) instead of Q(s,a) for all actions.

3. What makes SAC different from other actor-critic methods?

Answer: SAC maximizes both reward and entropy, is off-policy with replay buffer, uses twin critics, and automatically tunes the temperature parameter.