Policy Gradient and Actor–Critic

Learning objectives

Examine direct updates to a stochastic policy, then investigate how a critic changes learning over long episodes.

Execution conditions

• Required packages: gymnasium, stable-baselines3, and PyTorch.
• SB3 policy: MlpPolicy.
• Gymnasium step: 5 return values.

If the required package is missing, run %pip install gymnasium==1.3.0 stable-baselines3==2.9.2 in a Colab or Jupyter code cell.

Applications

Throughput and cost of equipment and communication resources

An average of 2 items arrive at the synthetic queue each time step. Increasing service capacity reduces queue length but increases energy cost.

Observation and state: Queue length $q=0$…20. Observation $q/20$.

Action: choose processing capacity 0, 2, or 4.

Reward and constraints: −(next queue length + 0.2×capacity² + 10×dropped-item count). Time limit at 100 steps.

Comparison baselines: Constant capacity of 2, and a threshold policy based on queue length.

Metrics to check: Evaluation return, mean queue length, mean capacity², and dropped items.

Mobile robot paths and safety

An application design case choosing either grid paths or continuous velocity control.

Observation and state: Position, target-relative position, and nearby obstacle observations.

Action: choose a grid direction or bounded linear and angular velocities.

Reward and constraints: Separate arrival, collision, and movement costs. Distinguish collisions from timeouts.

Comparison baseline: shortest path and rule-based avoidance.

Metrics to check: Success rate, collision rate, travel distance, and time. Evaluate on unseen maps.

Update a policy from one episode

Starting conditions

Queue, softmax policy 1→16→3, $\gamma=.95$, Adam .005, and 30 episodes. Use a simple baseline that subtracts the mean return for each episode.

Procedure

1. Calculate return-to-go by hand for the fixed rewards [-1,-2,-3], then compare with the loop output.
2. Run the provided REINFORCE implementation and record training returns separately from deterministic evaluation returns.
3. Explain the difference between stochastic actions during collection and argmax selection during evaluation.

Results to retain

• 3-step calculation table, learning curves, and an evaluation table comparing with the rule-based policy.

"""Synthetic service queue. An action chooses capacity, including its energy cost."""
import numpy as np

class Queue:
    def reset(self, seed=0):
        self.rng = np.random.default_rng(seed)
        self.q, self.t = 0, 0
        return self.q

    def step(self, action):
        capacity = (0, 2, 4)[int(action)]
        arriving = int(self.rng.poisson(2))
        raw = max(0, self.q - capacity) + arriving
        dropped = max(0, raw - 20)
        self.q = min(20, raw)
        self.t += 1
        reward = -(self.q + .2 * capacity**2 + 10 * dropped)
        return self.q, reward, False, self.t >= 100, {'queue': self.q, 'capacity': capacity, 'dropped': dropped}

def evaluate(policy, seeds=range(100, 105)):
    rows = []
    for seed in seeds:
        env = Queue(); q = env.reset(seed)
        total = waiting = energy = dropped = 0
        while True:
            q, r, terminated, truncated, info = env.step(policy(q))
            total += r; waiting += q; energy += info['capacity']**2; dropped += info['dropped']
            if terminated or truncated: break
        rows.append([total, waiting / env.t, energy / env.t, dropped])
    return np.array(rows)

policies = {'constant': lambda q: 1, 'threshold': lambda q: 0 if q == 0 else (1 if q < 5 else 2)}
for name, policy in policies.items():
    print(name, evaluate(policy).mean(axis=0))


import torch
from torch import nn
from torch.distributions import Categorical
def fit_reinforce(seed=0,baseline=True):
    torch.manual_seed(seed)
    model=nn.Sequential(nn.Linear(1,16),nn.Tanh(),nn.Linear(16,3))
    optimizer=torch.optim.Adam(model.parameters(),lr=.005);history=[]
    for episode in range(30):
        env=Queue();q=env.reset(seed*1000+episode);logs=[];rewards=[]
        for k in range(100):
            distribution=Categorical(logits=model(torch.tensor([[q/20]],dtype=torch.float32)))
            a=distribution.sample();q,r,te,tr,_=env.step(int(a.item()))
            logs.append(distribution.log_prob(a).squeeze());rewards.append(r/20)
            if te or tr:break
        returns=[];g=0
        for r in rewards[::-1]:g=r+.95*g;returns.append(g)
        returns=torch.tensor(returns[::-1],dtype=torch.float32)
        advantage=returns-returns.mean() if baseline else returns
        loss=-(torch.stack(logs)*advantage).mean()
        optimizer.zero_grad();loss.backward();optimizer.step();history.append(sum(rewards)*20)
    def policy(q):
        with torch.no_grad():return int(model(torch.tensor([[q/20]],dtype=torch.float32)).argmax())
    return policy,history
policy,history=fit_reinforce()
print(evaluate(policy).mean(0))

Record results and self-check

Time point:

reward:

Return-to-go:

Baseline:

Advantage:

Points to check

• Is the log probability that of the selected action?
• Can you explain how the probability of an action with negative advantage changes?

Rule-based control and A2C

Starting conditions

Discrete actions in Queue

Procedure

1. Pass the two preceding rule-based policies policies to evaluate and record results for the common evaluation seeds. Find where QueueGym transforms queue length to $q/20$ and reward to $1/20$ of its original value.
2. Run fit_sb3("A2C",seed=0,steps=2048). Select the learned policy with model.predict(...,deterministic=True) and evaluate it on the same seeds using evaluate in the notebook.
3. Record episode return, mean queue length, mean squared capacity, and total dropped items for the rule-based and learned policies in a table with the same columns. Identify improvements and regressions, and state that your judgment is based on this training budget and evaluation seed.

Results to retain

• Evaluation table for baseline and learned policies. Plot comparing queue length and cost.

"""Synthetic service queue. An action chooses capacity, including its energy cost."""
import numpy as np

class Queue:
    def reset(self, seed=0):
        self.rng = np.random.default_rng(seed)
        self.q, self.t = 0, 0
        return self.q

    def step(self, action):
        capacity = (0, 2, 4)[int(action)]
        arriving = int(self.rng.poisson(2))
        raw = max(0, self.q - capacity) + arriving
        dropped = max(0, raw - 20)
        self.q = min(20, raw)
        self.t += 1
        reward = -(self.q + .2 * capacity**2 + 10 * dropped)
        return self.q, reward, False, self.t >= 100, {'queue': self.q, 'capacity': capacity, 'dropped': dropped}

def evaluate(policy, seeds=range(100, 105)):
    rows = []
    for seed in seeds:
        env = Queue(); q = env.reset(seed)
        total = waiting = energy = dropped = 0
        while True:
            q, r, terminated, truncated, info = env.step(policy(q))
            total += r; waiting += q; energy += info['capacity']**2; dropped += info['dropped']
            if terminated or truncated: break
        rows.append([total, waiting / env.t, energy / env.t, dropped])
    return np.array(rows)

policies = {'constant': lambda q: 1, 'threshold': lambda q: 0 if q == 0 else (1 if q < 5 else 2)}
for name, policy in policies.items():
    print(name, evaluate(policy).mean(axis=0))

# Run the Queue or Motor definition first. Install gymnasium and stable-baselines3 if absent.
import gymnasium as gym
from gymnasium import spaces
from stable_baselines3 import A2C, PPO, SAC
class QueueGym(gym.Env):
    def __init__(self):
        self.model=Queue();self.action_space=spaces.Discrete(3)
        self.observation_space=spaces.Box(0.,1.,(1,),dtype=np.float32)
    def reset(self,seed=None,options=None):
        super().reset(seed=seed)
        return np.array([self.model.reset(seed)/20],dtype=np.float32),{}
    def step(self,a):
        q,r,te,tr,info=self.model.step(a)
        return np.array([q/20],dtype=np.float32),r/20,te,tr,info
class MotorGym(gym.Env):
    def __init__(self):
        self.model=Motor();self.action_space=spaces.Box(-1.,1.,(1,),dtype=np.float32)
        self.observation_space=spaces.Box(-np.inf,np.inf,(2,),dtype=np.float32)
    def reset(self,seed=None,options=None):
        super().reset(seed=seed)
        return self.model.reset(seed),{}
    def step(self,a):return self.model.step(a)

def fit_sb3(algorithm='PPO',seed=0,steps=2048):
    cls={'A2C':A2C,'PPO':PPO,'SAC':SAC}[algorithm]
    env=MotorGym() if algorithm=='SAC' else QueueGym()
    options={'policy_kwargs':dict(net_arch=[32,32]),'seed':seed,'verbose':0,'device':'cpu'}
    if algorithm=='PPO':options.update(n_steps=128,batch_size=64,n_epochs=3)
    if algorithm=='A2C':options.update(n_steps=32)
    if algorithm=='SAC':options.update(learning_starts=100,buffer_size=5000,batch_size=64)
    model=cls('MlpPolicy',env,**options);model.learn(total_timesteps=steps)
    return model

model=fit_sb3('A2C')
print(evaluate(lambda q:int(model.predict(np.array([q/20],dtype=np.float32),deterministic=True)[0])).mean(0))

Record results and self-check

Policy:

Training budget:

Evaluation conditions:

Main error and delay:

Input cost:

Constraint violations:

Points to check

• Did you distinguish training return from evaluation metrics in their original units?
• Did you first validate with a synthetic model instead of real equipment?

Apply this to your research problem

Starting conditions

Based on the preceding experiments, choose Queue, Motor, or one of your research problems and design an experiment for real application.

Procedure

1. Define state, observation, action, reward, termination, and time limit in one table. Record each variable's units and the information you can actually obtain.
2. Write 3 expected transitions numerically and check whether rewards for good and bad actions match your intent.
3. Choose one rule-based comparison baseline, one evaluation condition, and one failure case of greatest concern. Also list the required measurements or simulator.

Results to retain

• MDP design table, 3 transition examples, and comparative experiment plan.
• If real research data are unavailable, design using synthetic conditions from an application case.

You may continue with the research problem and design table selected earlier. Revisit existing entries and transition examples using the concepts learned in this activity, and update changed entries or reasons for retaining them.

Record results and self-check

Item:

Concrete definition and units:

Observability and verification method:

Points to check

• Did you exclude future information and evaluation labels from observations?
• Did you define domain-specific metrics rather than judging the research contribution solely by training return?

Official examples and data

• SB3 · A2C