PPO and public benchmarks

Learning objectives

Connect the method for limiting large policy changes in a single update to actual policy performance.

Execution conditions

• Required packages: gymnasium, stable-baselines3, and PyTorch.
• SB3 policy: MlpPolicy.
• Gymnasium step: 5 return values.
• Public environment visualization: Add gymnasium[classic-control].

If the required package is missing, run %pip install control==0.10.2 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.

Design energy storage operation

A research design case distinguishing the roles of price, load, and stored energy before connecting measured data.

Observation and state: SOC, load, price, and time. Observations must exclude future information.

Action: charging/discharging power. First choose discrete levels or a continuous range.

Reward and constraints: electricity and degradation costs. Specify rules for handling SOC and power limits.

Comparison baseline: time-of-day rules and, if possible, optimization using the same information.

Metrics to check: total cost, constraint violations, and maximum grid power draw. Split train/test data by date.

First check whether the discrete or continuous action choice matches the algorithm. In your domain, compare rule-based control using the same information and input limits.

Rule-based control and PPO

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("PPO",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('PPO')
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?

Compare PPO on a public benchmark

Starting conditions

CartPole-v1 with clip_range=.1 and .2, each for 2048 transitions. Use the same seed and evaluation group.

Procedure

1. Connect clip_range, n_steps, batch_size, and n_epochs in the public implementation to their roles using the code and official explanations.
2. Run two models that differ only in clipping settings with the same training budget.
3. Compare returns and lengths on the common evaluation set. Identify one difference from the CartPole PPO configuration in RL Baselines3 Zoo and state a hypothesis about its effect.

Results to retain

• Evaluation table for both configurations and interpretation of their differences.

"""Small public-environment benchmark; evaluation traces are ordinary numeric arrays."""
import json
from importlib.metadata import version
from pathlib import Path
import numpy as np
import gymnasium as gym

def collect(env_id='CartPole-v1', policy=None, seeds=range(100,103)):
    env=gym.make(env_id); transitions=[]; summary=[]
    for seed in seeds:
        obs,info=env.reset(seed=seed);env.action_space.seed(seed);total=0.;length=0
        while True:
            action=env.action_space.sample() if policy is None else policy(obs)
            nxt,reward,terminated,truncated,info=env.step(action)
            transitions.append((obs.copy(),np.asarray(action).copy(),reward,nxt.copy(),terminated,truncated,seed))
            obs=nxt;total+=reward;length+=1
            if terminated or truncated:break
        summary.append([seed,total,length,int(terminated),int(truncated)])
    env.close()
    columns=list(zip(*transitions))
    data={k:np.asarray(v) for k,v in zip(['observations','actions','rewards','next_observations','terminated','truncated','episode_seed'],columns)}
    return np.asarray(summary),data

def cartpole_rule(obs):return int(obs[2]+.25*obs[3]>0)
def zero_torque(obs):return np.zeros(1,dtype=np.float32)

def save_trace(path, env_id, data, policy_name, seeds):
    # Contains numeric arrays and JSON text only; never requires pickle.
    metadata=dict(env_id=env_id,policy=policy_name,seeds=list(seeds),gymnasium=version('gymnasium'))
    np.savez_compressed(path,**data,metadata_json=np.array(json.dumps(metadata)))
    with np.load(path,allow_pickle=False) as saved:
        assert len(saved['observations'])==len(saved['rewards'])
        print(json.loads(str(saved['metadata_json'])))

def public_model(algorithm='DQN',env_id='CartPole-v1',seed=7,steps=2048,**changes):
    from stable_baselines3 import DQN,PPO,SAC
    options=dict(policy_kwargs=dict(net_arch=[32,32]),seed=seed,verbose=0,device='cpu')
    if algorithm=='DQN':options.update(buffer_size=5000,learning_starts=100,target_update_interval=100,batch_size=32)
    if algorithm=='PPO':options.update(n_steps=128,batch_size=64,n_epochs=3)
    if algorithm=='SAC':options.update(buffer_size=5000,learning_starts=100,batch_size=64)
    options.update(changes)
    env=gym.make(env_id)
    model={'DQN':DQN,'PPO':PPO,'SAC':SAC}[algorithm]('MlpPolicy',env,**options)
    model.learn(total_timesteps=steps)
    env.close()
    return model

print('Gymnasium',version('gymnasium'))
env=gym.make('CartPole-v1')
print('Observation:',env.observation_space,'Action:',env.action_space,'Limit:',env.spec.max_episode_steps)
env.close()
# For visual inspection, create a separate env with render_mode='rgb_array'.
# Rendering requires gymnasium[classic-control]; do not render every training step.

for clipping in [.1,.2]:
    model=public_model('PPO',clip_range=clipping)
    rows,data=collect(policy=lambda o:int(model.predict(o,deterministic=True)[0]))
    print(clipping,rows[:,1:3].mean(0),rows[:,1:3].std(0))
# TODO: Choose one Zoo configuration item whose purpose differs.

Record results and self-check

clip_range:

Training budget:

Return mean and standard deviation:

Episode length:

Settings different from Zoo:

Points to check

• Did you avoid comparing long Zoo training results and short-run scores as equivalent benchmarks?
• Did you record the environment, training budget, and seed together?

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 · PPO
• RL Baselines3 Zoo · PPO configuration