PPO와 공개 Benchmark · 학습
실행 조건
- 필요 패키지: gymnasium·stable-baselines3·PyTorch.
- SB3 정책:
MlpPolicy. - Gymnasium
step: 5개 반환값. - 공개 환경 시각화:
gymnasium[classic-control]추가.
필요한 패키지가 없으면 Colab·Jupyter 코드 셀에서 %pip install control==0.10.2 gymnasium==1.3.0 stable-baselines3==2.9.2를 실행하세요.
이번에 배울 내용
정책을 한 번에 크게 바꾸는 문제를 제한하는 방식과 실제 정책 성능을 연결합니다.
PPO의 제한
새 정책과 이전 정책이 같은 행동에 부여한 확률의 비를 사용합니다. Clipping은 목적함수에서 과도한 갱신의 영향을 제한하며 물리적 Action 한계를 대신하지 않습니다.
Benchmark의 조건
환경·학습량·seed·평가 방법이 맞아야 설정 차이를 비교할 수 있습니다. Zoo의 긴 학습 결과와 짧은 실습 실행을 동일한 결과로 보지 않습니다.
풀이 예제
Queue의 이산 행동
"""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))
constant [-921.2 8.052 4. 3.6 ] threshold [-387.4 2.77 5.52 0. ] [-531.4 2.114 16. 0. ]