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를 실행하세요.

적용 사례

장비·통신 자원의 처리량과 비용

합성 대기열에 매 시점 평균 2건이 도착합니다. 서비스 용량을 높이면 대기는 줄지만 에너지 비용이 늘어납니다.

Observation·State: 대기 수 $q=0$…20. 관측 $q/20$.

행동: 처리 용량 0·2·4 중 선택.

Reward·Constraint: −(다음 대기 수 + 0.2×용량² + 10×유실 수). 100 step에서 시간 제한.

비교 기준: 항상 용량 2, 대기 수에 따른 Threshold policy.

확인할 지표: 평가 Return, 평균 대기 수, 평균 용량², 유실 수.

에너지 저장장치 운전 설계

실측 자료를 연결하기 전, 가격·부하·저장량의 역할을 구분하는 연구 설계 사례입니다.

Observation·State: SOC, 부하, 가격, 시간. 미래 정보를 포함하지 않는 관측.

행동: 충·방전 전력. 이산 단계 또는 연속 범위를 먼저 결정.

Reward·Constraint: 전력 비용과 열화 비용. SOC·전력 한계의 처리 규칙 명시.

비교 기준: 시간대 규칙, 가능하면 동일 정보의 최적화.

확인할 지표: 총 비용, 제약 위반, 최대 수전 전력. 날짜 단위 Train/Test 분리.

이산·연속 행동 선택이 알고리즘과 맞는지 먼저 확인합니다. 자신의 도메인에서는 규칙 제어를 같은 정보·입력 제한으로 비교합니다.

규칙 제어와 PPO

시작 조건

Queue의 이산 행동

진행 방법

1. 앞의 policies 두 규칙 정책을 evaluate에 넣어 공통 평가 Seed의 결과를 기록하세요. QueueGym에서 대기 수를 $q/20$으로, 보상을 원래 값의 $1/20$로 변환하는 위치를 찾으세요.
2. fit_sb3("PPO",seed=0,steps=2048)를 실행하세요. 학습한 정책을 model.predict(...,deterministic=True)로 선택하고, 노트북의 evaluate로 같은 Seed에서 평가하세요.
3. 규칙 정책과 학습 정책의 Episode Return·평균 대기 수·평균 용량 제곱·총 유실 수를 같은 열의 표로 기록하세요. 개선된 값과 악화된 값을 짚어 설명하고, 판단의 근거가 이번 학습량과 평가 Seed의 결과임을 밝히세요.

남길 결과

• 기준·학습 정책의 평가 표. 대기와 비용 비교 그림.

"""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))

결과 기록·자가 점검

정책:

학습량:

평가 조건:

주요 오차·대기:

입력 비용:

제약 위반:

확인할 점

• 학습 Return과 원래 단위의 평가 지표를 구분했는가?
• 실제 장비 대신 합성 모델에서 먼저 검증했는가?

PPO를 공개 Benchmark에서 비교하기

시작 조건

CartPole-v1에서 clip_range=.1과 .2, 각각 2048 Transition입니다. 같은 seed와 평가군을 사용합니다.

진행 방법

1. 공개 구현의 clip_range·n_steps·batch_size·n_epochs가 무엇을 제어하는지 코드와 공식 설명을 연결하세요.
2. Clipping 설정만 바꾼 두 모델을 같은 학습량으로 실행하세요.
3. 공통 평가군의 Return·길이를 비교하세요. RL Baselines3 Zoo의 CartPole PPO 설정과 한 가지 차이를 찾아 그 영향을 가설로 적습니다.

남길 결과

• 두 설정 평가 표와 설정 차이에 관한 해석.

"""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: Zoo 설정에서 목적이 다른 항목 한 개를 고르세요.

결과 기록·자가 점검

clip_range:

학습량:

Return 평균·표준편차:

Episode 길이:

Zoo와 다른 설정:

확인할 점

• Zoo의 긴 학습 결과와 짧은 실행의 점수를 동등한 Benchmark로 비교하지 않았는가?
• 환경·학습량·seed를 함께 기록했는가?

본인의 연구 문제로 옮기기

시작 조건

앞의 실험을 바탕으로 Queue·Motor 또는 본인의 연구 문제 하나를 선택해, 실제 적용을 위한 실험을 설계합니다.

진행 방법

1. State·Observation·Action·Reward·종료·시간 제한을 표 한 장으로 정의하세요. 각 변수의 단위와 실제로 얻을 수 있는 정보를 적습니다.
2. 예상 Transition 3개를 수치로 적어, 좋은 행동과 나쁜 행동의 보상이 의도에 맞는지 확인하세요.
3. 규칙 기반 비교 기준 한 개, 평가 조건 한 개, 가장 우려되는 실패 사례 한 개를 정하세요. 필요한 실측 자료나 Simulator도 적습니다.

남길 결과

• MDP 설계 표, Transition 예 3개, 비교 실험 계획.
• 실제 연구 자료가 없으면 적용 사례의 합성 조건으로 설계합니다.

앞서 선택한 연구 문제와 설계 표를 이어서 사용해도 됩니다. 이번 활동에서 배운 개념으로 기존 항목과 Transition 예를 다시 검토하고, 변경한 항목이나 그대로 둔 이유를 갱신하세요.

결과 기록·자가 점검

항목:

구체적 정의·단위:

관측 가능 여부·확인 방법:

확인할 점

• 미래 정보나 평가 Label을 관측에 넣지 않았는가?
• 학습 Return만으로 연구 기여를 판단하지 않고, 해당 도메인의 지표를 정했는가?

공식 예제·데이터

• SB3 · PPO
• RL Baselines3 Zoo · PPO 설정