DQN
학습 목표
같은 Queue 문제에서 Q-table과 Neural Network의 상태 표현을 비교하고, 학습한 정책을 규칙 정책과 평가합니다.
실행 조건
기준 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를 실행하세요.
적용 사례
장비·통신 자원의 처리량과 비용
합성 대기열에 매 시점 평균 2건이 도착합니다. 서비스 용량을 높이면 대기는 줄지만 에너지 비용이 늘어납니다.
Observation·State: 대기 수 …20. 관측 .
행동: 처리 용량 0·2·4 중 선택.
Reward·Constraint: −(다음 대기 수 + 0.2×용량² + 10×유실 수). 100 step에서 시간 제한.
비교 기준: 항상 용량 2, 대기 수에 따른 Threshold policy.
확인할 지표: 평가 Return, 평균 대기 수, 평균 용량², 유실 수.
에너지 저장장치 운전 설계
실측 자료를 연결하기 전, 가격·부하·저장량의 역할을 구분하는 연구 설계 사례입니다.
Observation·State: SOC, 부하, 가격, 시간. 미래 정보를 포함하지 않는 관측.
행동: 충·방전 전력. 이산 단계 또는 연속 범위를 먼저 결정.
Reward·Constraint: 전력 비용과 열화 비용. SOC·전력 한계의 처리 규칙 명시.
비교 기준: 시간대 규칙, 가능하면 동일 정보의 최적화.
확인할 지표: 총 비용, 제약 위반, 최대 수전 전력. 날짜 단위 Train/Test 분리.
전체 재학습을 늘리기보다 핵심 변경 한 가지와 정책의 실제 지표를 연결해 봅니다.
동일한 작은 문제의 DQN
시작 조건
- 환경: Queue.
- 네트워크: .
- Optimizer: Adam,
lr=.001. - Discount Factor: .
- Replay: 3000. Batch: 32.
- 학습량: 1800 Transition.
Q-table과 Neural Network이 상태별 행동을 표현하는 방식을 비교하고, 학습한 정책의 성능을 평가합니다.
진행 방법
- 먼저 Queue의 규칙 정책 성능을 확인한 뒤 제공된 학습 함수를 실행하세요.
- 학습 Loss와 …20의 선택 행동을 그리세요.
- 학습에 쓰지 않은
seed=100…104에서 Return·대기·비용·유실을 평가하세요.
남길 결과
- Loss·상태별 행동 그림, 규칙 정책과의 평가 표.
"""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))
결과 기록·자가 점검
학습 seed:
학습 Transition 수:
Return 평균·표준편차:
평균 대기:
용량²:
유실:
확인할 점
- Loss 감소를 곧바로 정책 성능 향상이라고 해석하지 않았는가?
- 짧은 학습에서 기준보다 나쁘게 나온 결과도 그대로 기록했는가?
공개 DQN 구현으로 옮기기
시작 조건
SB3 DQN, CartPole-v1, 2048 Transition입니다. 앞의 Queue 결과와 점수를 직접 비교하지 않고, 같은 환경의 Random·Rule과 비교합니다.
진행 방법
- 공개 DQN의 Policy·Replay 크기·Target 갱신 주기·학습 시작 시점을 앞의 구현과 대응시키세요.
- 제공된 짧은 학습을 실행하고
seed=100…102에서 결정적으로 평가하세요. - 학습 정책과 Random·Rule의 Episode 길이·Return을 비교하고, 학습량이 충분한지 판단할 근거를 적으세요.
남길 결과
- 구현 설정 대응 표와 공통 평가 표.
"""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.
model=public_model('DQN')
for name,policy in [('random',None),('rule',cartpole_rule),('dqn',lambda o:int(model.predict(o,deterministic=True)[0]))]:
rows,data=collect(policy=policy);print(name,rows[:,1:3].mean(0),rows[:,1:3].std(0))
# TODO: 공개 구현과 앞의 코드에서 Replay·Target의 위치를 찾아 연결하세요.
결과 기록·자가 점검
정책:
학습 Transition 수:
평가 Return 평균·표준편차:
평균 길이:
확인할 점
- SB3 DQN을 Double·Dueling·PER가 포함된 구현이라고 오해하지 않았는가?
- 환경이 다른 Queue 점수와 CartPole 점수로 우열을 주장하지 않았는가?
본인의 연구 문제로 옮기기
시작 조건
앞의 실험을 바탕으로 Queue·Motor 또는 본인의 연구 문제 하나를 선택해, 실제 적용을 위한 실험을 설계합니다.
진행 방법
- State·Observation·Action·Reward·종료·시간 제한을 표 한 장으로 정의하세요. 각 변수의 단위와 실제로 얻을 수 있는 정보를 적습니다.
- 예상 Transition 3개를 수치로 적어, 좋은 행동과 나쁜 행동의 보상이 의도에 맞는지 확인하세요.
- 규칙 기반 비교 기준 한 개, 평가 조건 한 개, 가장 우려되는 실패 사례 한 개를 정하세요. 필요한 실측 자료나 Simulator도 적습니다.
남길 결과
- MDP 설계 표, Transition 예 3개, 비교 실험 계획.
- 실제 연구 자료가 없으면 적용 사례의 합성 조건으로 설계합니다.
앞서 선택한 연구 문제와 설계 표를 이어서 사용해도 됩니다. 이번 활동에서 배운 개념으로 기존 항목과 Transition 예를 다시 검토하고, 변경한 항목이나 그대로 둔 이유를 갱신하세요.
결과 기록·자가 점검
항목:
구체적 정의·단위:
관측 가능 여부·확인 방법:
확인할 점
- 미래 정보나 평가 Label을 관측에 넣지 않았는가?
- 학습 Return만으로 연구 기여를 판단하지 않고, 해당 도메인의 지표를 정했는가?