강의자료 · 전기공학을 위한 강화학습

공개 환경과 실험 기록

학습 목표

종료 처리 한 줄이 학습 Target을 바꿀 수 있습니다. 환경 경계와 공정한 정책 평가부터 확인합니다.

실행 조건

기준 Queue 예제는 NumPy로 실행합니다. 학습에 사용하지 않은 seed=100…104에서 정책을 비교합니다.

공개 환경 활동은 gymnasium·stable-baselines3를 사용합니다. 시각화할 때만 gymnasium[classic-control]이 필요합니다.

필요한 패키지가 없으면 Colab·Jupyter 코드 셀에서 %pip install control==0.10.2 gymnasium==1.3.0 stable-baselines3==2.9.2를 실행하세요.

적용 사례

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

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

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

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

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

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

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

설비 점검 시점 선택

매번 점검하는 비용과 고장을 놓치는 비용의 균형을 연구 문제로 구성합니다.

Observation·State: 최근 센서값·추세·마지막 점검 이후 시간. 실제 고장 상태와 관측의 차이 구분.

행동: 계속 운전·점검·교체.

Reward·Constraint: 운전 이익−점검·교체·고장 비용.

비교 기준: 주기 점검, 임계값 경보.

확인할 지표: 가동률, 고장 수, 총 비용. 고장 Label 누출 여부 확인.

한 Episode의 경계를 확인하기

시작 조건

Queue 예제로 reset→step→종료의 흐름을 실행합니다. seed=7, 행동 1(용량 2), 100 step입니다.

진행 방법

  1. 처음 5개 Transition의 (q,a,r,q_next,terminated,truncated)를 기록하세요.
  2. 시간 제한을 5 step으로 두고 마지막 Transition을 확인하세요.
  3. Terminal state의 종료와 데이터 수집 시간 제한을 구분해, Bootstrap target에 사용할 Mask를 적으세요.

남길 결과

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

env=Queue();q=env.reset(7)
for k in range(5):
    nq,r,te,tr,info=env.step(1);print(k,q,1,r,nq,te,tr);q=nq
# TODO: 시간 제한을 5로 바꾸고 경계를 확인하세요.

결과 기록·자가 점검

step:

q:

action:

reward:

next q:

terminated:

truncated:

확인할 점

Gymnasium 실행과 Transition 기록

시작 조건

Gymnasium CartPole-v1을 사용합니다. Random policy와 각도·Angular Velocity의 간단한 Rule을 seed=100,101,102에서 비교합니다.

진행 방법

  1. observation_space·action_space·Episode 제한을 출력하고, 공식 문서에서 관측 4개 성분과 두 행동의 의미를 확인하세요.
  2. 두 정책을 같은 seed에서 실행하고 Return·길이·종료 유형을 기록하세요.
  3. Transition 기록을 NPZ로 저장하고 다시 읽으세요. 첫 Transition과 Episode 경계를 찾아 관측 수·행동 수의 대응을 확인합니다.

남길 결과

"""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 name,policy in [('random',None),('rule',cartpole_rule)]:
    rows,data=collect(policy=policy);print(name,rows)
    save_trace(name+'_cartpole.npz','CartPole-v1',data,name,range(100,103))
# TODO: 첫 전이와 Episode 경계를 표에서 확인하세요.

결과 기록·자가 점검

정책:

seed:

Return:

Episode 길이:

terminated:

truncated:

확인할 점

본인의 연구 문제로 옮기기

시작 조건

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

진행 방법

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

남길 결과

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

결과 기록·자가 점검

항목:

구체적 정의·단위:

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

확인할 점

공식 예제·데이터