DQN Variants와 Replay Buffer Sampling
학습 목표
네트워크 개선을 모두 적용하기보다 어떤 문제를 해결하는 변화인지 구분하고, 필요한 한 가지를 검증합니다.
실행 조건
NumPy·PyTorch·Gymnasium을 사용합니다. 마지막 활동은 외부 데이터 카드를 읽고 직접 수집한 전이를 분석합니다.
적용 사례
장비·통신 자원의 처리량과 비용
합성 대기열에 매 시점 평균 2건이 도착합니다. 서비스 용량을 높이면 대기는 줄지만 에너지 비용이 늘어납니다.
Observation·State: 대기 수 …20. 관측 .
행동: 처리 용량 0·2·4 중 선택.
Reward·Constraint: −(다음 대기 수 + 0.2×용량² + 10×유실 수). 100 step에서 시간 제한.
비교 기준: 항상 용량 2, 대기 수에 따른 Threshold policy.
확인할 지표: 평가 Return, 평균 대기 수, 평균 용량², 유실 수.
설비 점검 시점 선택
매번 점검하는 비용과 고장을 놓치는 비용의 균형을 연구 문제로 구성합니다.
Observation·State: 최근 센서값·추세·마지막 점검 이후 시간. 실제 고장 상태와 관측의 차이 구분.
행동: 계속 운전·점검·교체.
Reward·Constraint: 운전 이익−점검·교체·고장 비용.
비교 기준: 주기 점검, 임계값 경보.
확인할 지표: 가동률, 고장 수, 총 비용. 고장 Label 누출 여부 확인.
선택·평가·표현·샘플링 구분
시작 조건
- , .
- , .
- Dueling: , .
- TD Error: .
진행 방법
- DQN·Double DQN의 Target을 손으로 구하고 식의 다른 부분을 표시하세요.
- Dueling의 평균 보정과 PER의 /.6 샘플 확률을 작은 배열로 확인하세요.
- 세 기법이 각각 어떤 실패 가능성을 겨냥하는지 한 줄씩 연결하세요.
남길 결과
- Target·표현·샘플 확률을 구분한 계산 표.
import numpy as np
online=np.array([2.,4.,3.]);target=np.array([5.,1.,2.])
print('DQN',1+.95*target.max(),'Double',1+.95*target[online.argmax()])
advantage=np.array([1.,2.,6.]);print('Dueling Q',2+advantage-advantage.mean())
error=np.array([.1,1.,5.])
for alpha in [0,.6]:
probability=(error+1e-5)**alpha;probability/=probability.sum();print(alpha,probability)
결과 기록·자가 점검
기법:
바꾸는 부분:
계산 예:
겨냥하는 문제:
확인할 점
- 행동 선택과 평가의 네트워크를 구분했는가?
- 샘플 확률과 Loss 가중치를 같은 값으로 혼동하지 않았는가?
필요한 개선 하나만 검증
시작 조건
Queue에서 dqn을 기준으로 double·dueling·per 중 한 가지만 선택해 비교합니다. 각각 1800 Transition, 같은 seed=0과 평가군을 사용합니다.
진행 방법
- 코드에서 선택한 기법이 바꾸는 위치를 찾고, 결과가 달라질 이유를 먼저 정하세요.
- dqn과 선택 기법을 같은 학습량·평가군으로 실행하세요.
- Return·대기·비용 중 실제로 바뀐 지표를 적고, 이 결과에서 그 기법이 필요한지 판단하세요.
남길 결과
- 선택 이유와 두 설정 평가 표. 필요하면
seed=1재실행은 확장으로 둡니다.
"""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
for variant in ['dqn','double']:
policy,loss=train_dqn(0,variant=variant);print(variant,evaluate(policy).mean(0))
# TODO: double 대신 검증할 기법 한 가지를 선택하세요.
결과 기록·자가 점검
학습 seed:
학습 Transition 수:
Return 평균·표준편차:
평균 대기:
용량²:
유실:
확인할 점
- Loss 감소를 곧바로 정책 성능 향상이라고 해석하지 않았는가?
- 짧은 학습에서 기준보다 나쁘게 나온 결과도 그대로 기록했는가?
Transition 데이터의 분포 살펴보기
시작 조건
공개 CartPole 환경에서 Random·Rule이 수집한 Transition을 비교합니다. Replay는 수집 정책의 방문 범위를 벗어난 경험을 만들 수 없습니다.
진행 방법
- 동일한 seed의 두 정책에서 각도·Angular Velocity·행동의 분포와 Episode 길이를 비교하세요.
- NPZ로 저장한 Transition을 다시 읽고, 한 Transition을 여러 번 뽑는 것과 새로운 상태를 경험하는 것의 차이를 설명하세요.
- Minari Fourrooms-Random 데이터 카드에서 수집 정책·관측 형식·환경 ID·라이선스를 찾아, 내 연구 데이터에 필요한 항목을 표로 정리하세요. 데이터 다운로드와 학습은 확장으로 둡니다.
남길 결과
- 방문 상태 분포 그림 1개와 데이터 카드 요약 표.
"""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.
import matplotlib.pyplot as plt
for name,policy in [('random',None),('rule',cartpole_rule)]:
rows,data=collect(policy=policy)
plt.hist(data['observations'][:,2],bins=20,alpha=.5,density=True,label=name)
save_trace(name+'_replay.npz','CartPole-v1',data,name,range(100,103))
plt.xlabel('Pole angle [rad]');plt.ylabel('Density');plt.legend()
# TODO: 행동 분포와 데이터 카드 항목을 비교하세요.
결과 기록·자가 점검
자료:
수집 정책:
환경·버전:
관측·행동 형식:
분포·이용 조건:
확인할 점
- Transition을 임의로 섞은 뒤 Episode가 연속이라고 해석하지 않았는가?
- 공개 환경에서 직접 수집한 자료와 외부 공개 데이터셋을 구분했는가?