Policy Gradient와 Actor–Critic

학습 목표

확률적 정책의 직접 갱신을 확인한 뒤 Critic이 긴 Episode의 학습을 어떻게 바꾸는지 살펴봅니다.

실행 조건

• 필요 패키지: gymnasium·stable-baselines3·PyTorch.
• SB3 정책: MlpPolicy.
• Gymnasium step: 5개 반환값.

필요한 패키지가 없으면 Colab·Jupyter 코드 셀에서 %pip install 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: 위치·목표 상대 위치·근접 장애물 관측.

행동: 격자 방향 선택 또는 제한된 Linear Velocity·Angular Velocity.

Reward·Constraint: 도착·충돌·이동 비용을 분리. 충돌과 시간 초과를 구분.

비교 기준: 최단 경로, 규칙 기반 회피.

확인할 지표: 성공률, 충돌률, 이동 거리, 시간. 미관측 지도에서 평가.

한 Episode로 정책을 갱신하기

시작 조건

Queue, Softmax 정책 1→16→3, $\gamma=.95$, Adam .005, 30 Episode입니다. Episode별 Return 평균을 빼는 단순 Baseline을 씁니다.

진행 방법

1. 고정 보상 [-1,-2,-3]의 Return-to-go를 직접 계산한 뒤 루프 결과와 대조하세요.
2. 제공된 REINFORCE를 실행하고 학습 Return과 결정적 평가 Return을 구분해 기록하세요.
3. 확률적 행동 수집과 평가의 argmax 선택이 다름을 설명하세요.

남길 결과

• 3-step 계산 표와 학습곡선, 규칙 정책과의 평가 표.

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

결과 기록·자가 점검

시점:

reward:

Return-to-go:

Baseline:

Advantage:

확인할 점

• 로그 확률이 선택 행동의 확률인가?
• Advantage가 음수인 행동의 확률이 어떻게 변하는지 설명할 수 있는가?

규칙 제어와 A2C

시작 조건

Queue의 이산 행동

진행 방법

1. 앞의 policies 두 규칙 정책을 evaluate에 넣어 공통 평가 Seed의 결과를 기록하세요. QueueGym에서 대기 수를 $q/20$으로, 보상을 원래 값의 $1/20$로 변환하는 위치를 찾으세요.
2. fit_sb3("A2C",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('A2C')
print(evaluate(lambda q:int(model.predict(np.array([q/20],dtype=np.float32),deterministic=True)[0])).mean(0))

결과 기록·자가 점검

정책:

학습량:

평가 조건:

주요 오차·대기:

입력 비용:

제약 위반:

확인할 점

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

본인의 연구 문제로 옮기기

시작 조건

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

진행 방법

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

남길 결과

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

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

결과 기록·자가 점검

항목:

구체적 정의·단위:

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

확인할 점

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

공식 예제·데이터

• SB3 · A2C