Policy Gradient와 Actor–Critic · 학습
실행 조건
- 필요 패키지: gymnasium·stable-baselines3·PyTorch.
- SB3 정책:
MlpPolicy. - Gymnasium
step: 5개 반환값.
필요한 패키지가 없으면 Colab·Jupyter 코드 셀에서 %pip install gymnasium==1.3.0 stable-baselines3==2.9.2를 실행하세요.
이번에 배울 내용
확률적 정책의 직접 갱신을 확인한 뒤 Critic이 긴 Episode의 학습을 어떻게 바꾸는지 살펴봅니다.
정책을 직접 갱신하기
REINFORCE는 선택 행동의 Log Probability를 Return으로 가중합니다. Baseline을 빼도 Reward Function 자체가 바뀌는 것은 아닙니다.
Critic의 역할
A2C는 Critic으로 미래 가치를 추정하고 Advantage를 계산해 Actor를 갱신합니다. Rollout 길이는 실제 보상과 Bootstrap의 비중을 바꿉니다.
풀이 예제
Queue, Softmax 정책 1→16→3, , Adam .005, 30 Episode입니다. Episode별 Return 평균을 빼는 단순 Baseline을 씁니다.
"""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))
constant [-921.2 8.052 4. 3.6 ] threshold [-387.4 2.77 5.52 0. ] [-531.4 2.114 16. 0. ]