DQN · 학습
실행 조건
기준 Queue 예제는 NumPy로 실행합니다. 학습에 사용하지 않은 seed=100…104에서 정책을 비교합니다.
PyTorch가 필요합니다. 짧은 합성 실험이며 학습 성능을 보장하는 설정은 아닙니다.
공개 환경 활동은 gymnasium·stable-baselines3를 사용합니다. 시각화할 때만 gymnasium[classic-control]이 필요합니다.
필요한 패키지가 없으면 Colab·Jupyter 코드 셀에서 %pip install control==0.10.2 gymnasium==1.3.0 stable-baselines3==2.9.2를 실행하세요.
이번에 배울 내용
상태가 늘어날 때 Q-table을 Neural Network으로 바꿔도 정책이 실제로 좋아지는지 검증합니다.
함수 근사
DQN은 Q-table을 Neural Network으로 바꿉니다. Replay는 저장된 경험을 다시 사용하고 Target Network는 Target을 만드는 Parameter의 갱신을 분리합니다.
공개 구현 읽기
Policy 이름만 맞추지 말고 Buffer·학습 시작·Target 갱신·입력 Normalization·탐색 일정을 대응시킵니다. 다른 환경의 Return은 같은 성능 단위가 아닙니다.
풀이 예제
- 환경: Queue.
- 네트워크: .
- Optimizer: Adam,
lr=.001. - Discount Factor: .
- Replay: 3000. Batch: 32.
- 학습량: 1800 Transition.
같은 Queue 문제에서 Q-table과 Neural Network의 상태 표현을 비교하고, 학습한 정책을 규칙 정책과 평가합니다.
"""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))
# Requires the Queue and evaluate definitions in the preceding cell.
import random
from collections import deque
import torch
from torch import nn
def train_dqn(seed=0, variant='dqn', steps=1800):
torch.manual_seed(seed); rng = random.Random(seed)
class Network(nn.Module):
def __init__(self):
super().__init__()
self.shared = nn.Sequential(nn.Linear(1, 32), nn.ReLU())
self.values = nn.Linear(32, 1 if variant == 'dueling' else 3)
self.advantage = nn.Linear(32, 3)
def forward(self, x):
h = self.shared(x); v = self.values(h)
if variant != 'dueling': return v
a = self.advantage(h)
return v + a - a.mean(dim=-1, keepdim=True)
online = Network(); target = Network(); target.load_state_dict(online.state_dict())
optimizer = torch.optim.Adam(online.parameters(), lr=.001)
buffer = deque(maxlen=3000); priorities = deque(maxlen=3000)
env = Queue(); q = env.reset(seed); losses = []; sample_counts = [0, 0]
def tensor(states): return torch.tensor(states, dtype=torch.float32).reshape(-1, 1) / 20
for step in range(steps):
epsilon = max(.1, 1 - step / 1000)
action = rng.randrange(3) if rng.random() < epsilon else int(online(tensor([q])).argmax())
nq, reward, terminated, truncated, _ = env.step(action)
buffer.append((q, action, reward / 20, nq, terminated))
priorities.append(max(priorities, default=1.0))
q = env.reset(seed + step + 1) if terminated or truncated else nq
if len(buffer) < 64: continue
probabilities = np.array(priorities, dtype=float)**.6
probabilities /= probabilities.sum()
if variant == 'per':
# Sampling with replacement; importance weighting uses the same probabilities.
ids = rng.choices(range(len(buffer)), weights=probabilities, k=32)
weights = (len(buffer) * probabilities[ids])**(-.4)
weights /= weights.max()
else:
ids = rng.sample(range(len(buffer)), 32); weights = np.ones(32)
bq, ba, br, bn, bt = zip(*(buffer[i] for i in ids))
predicted = online(tensor(bq)).gather(1, torch.tensor(ba).reshape(-1, 1)).flatten()
with torch.no_grad():
if variant == 'double':
selected = online(tensor(bn)).argmax(dim=1, keepdim=True)
next_value = target(tensor(bn)).gather(1, selected).flatten()
else: next_value = target(tensor(bn)).max(dim=1).values
expected = torch.tensor(br, dtype=torch.float32) + .95 * (~torch.tensor(bt)) * next_value
td = expected - predicted
loss = (torch.tensor(weights, dtype=torch.float32) * td.square()).mean()
optimizer.zero_grad(); loss.backward(); nn.utils.clip_grad_norm_(online.parameters(), 5); optimizer.step()
for i, delta in zip(ids, td.detach().numpy()): priorities[i] = abs(float(delta)) + 1e-5
losses.append(float(loss.detach()))
if step % 100 == 0: target.load_state_dict(online.state_dict())
def policy(q):
with torch.no_grad(): return int(online(tensor([q])).argmax())
return policy, losses
policy,losses=train_dqn(0,variant='dqn')
print(evaluate(policy).mean(0))
constant [-921.2 8.052 4. 3.6 ] threshold [-387.4 2.77 5.52 0. ] [-437.32 2.114 11.296 0. ]