[RL] 가치 기반 강화학습 (NFQ, DQN, 이중 DQN, DQNPolicy, 듀얼링)

NFQ(Neural Fitted Q iteration)

인공신경망을 강화학습의 함수 근사에 적용한 초기 알고리즘

Fitted Value Iteration → Fitted Q Iteration  → Neural FQI 로 발전

배치(batch) 강화학습 : 데이터를 수집 후에 일괄적으로 학습하는 방식

기본 아이디어 :

환경과 상호작용을 통해 데이터를 만듦

만들어진 데이터로 행동 가치 함수 Q를 신경망으로 학습

위 과정의 반복

 

NFQ와 역전파 알고리즘

지도학습에서는 데이터의 레이블을 예측

강화학습에서 TD 목표(=보상+다음 상태의 수익)는 모델 자체에서 나옴

다음 상태의 수익에 대한 추정치는 오차 역전파의 대상에서 제외해야함.

 

함수 근사의 문제

1) 훈련의 불안정성

2) iid 가정 위배

 

함수 근사의 문제에 대한 NFQ의 해결책

동일한 배치를 신경망에 여러번 학습시킨다. 

배치의 크기를 키우고, 하나의 배치에 다양한 사례를 포함시킨다. 

Deep Q Network

구글 딥마인드가 발표한 최초의 딥러닝 + 강화학습 모형

목표를 고정시켜 학습이 불안정해지는 것을 방지

목표망은 고정 시키고, 온라인 망을 학습

상관이 강한 상태들 사이에서 작은 차이를 반영할 수 있도록 신경망의 크기도 키움

 

DQN에서 게임과 관련된 기법들

DQN 논문은 게임을 대상으로 연구 → 관련된 기법들 적용

1) 보상 깎기(Clipping Rewards)

2) 프레임 건너뛰기(Skipping Frames)

 

QNetwork

QNetwork 클래스는 BasePolicy를 상속받아 구현

관측 공간, 행동 공간, 특징 추출기 등을 입력으로 받아 초기화

네트워크 구조(net_arch)를 지정할 수 있으며, 기본값은 [64,64]

forward 매서드에서 관측값을 입력받아 Q-값을 예측

 

DQNPolicy

메인 Q-네트워크(q-net) 와 목표 Q-네트워크(q_net_target)를 가지고 있음

네트워크 구조, 활성화함수, 특징 추출기 등을 사용자가 지정할 수 있음

_build 메서드에서 네트워크와 최적화 알고리즘을 초기화

make_q_net 메서드를 통해 Q-네트워크를 생성

forward와 _predict 메서드로 행동을 예측

set_training_mode 메서드로 훈련 모드와 평가 모드를 전환 할 수 있음

 

손실함수

1) 평균제곱오차(MSE)

제곱을 하기 때문에 큰 오차에 강하게 반응

강화학습에서는 큰 오차가 자주 발생하기 때문에 MSE는 지나치게 민감

2) 평균절대오차(MAE)

큰 오차와 작은 오차에 동일하게 반응

손실이 0인 점 근처에서도 기울기가 같으므로 잘 수렴하지 않음

3) 후버손실(Huber Loss)

MSE와 MAE의 장점을 합친 것

4) 경사절단법(gradient clipping)

후버 손실의 구현 방법

MSE를 사용하고, 경사가 정해진 최대값을 넘으면 최대값으로 바꿔줌

 

이중 DQN

DQN도 Q학습이므로 최대화 편향이 있음

이중 Q 학습으로 해결 가능

단, 신경망을 2개 학습시켜야하기 때문에 학습이 느려지는 문제

DQN에는 목표망(Q)가 있으므로 따로 학습시키지 않아도 됨

stable-baselines3에는 구현 X

 

듀얼링(Dueling)

행동 가치를 학습하는 대신, 상태 가치와 이득을 학습

한 행동의 경험이 같은 상태의 다른 행동에도 영향

실제로는 행동 가치에서 상태 가치와 이득을 분리할 수 없음

학습의 안정성을 위해 이득의 평균을 뺌

 

 

 

## DQN 학습
import gymnasium as gym
from stable_baselines3 import DQN
# 환경
env = gym.make("CartPole-v1", render_mode="human")
# 모델
model = DQN("MlpPolicy", env, verbose=1)
# 총 100,000회에 걸쳐 학습
model.learn(total_timesteps=100_000, log_interval=4, progress_bar=True)


## 애니메이션, 모델 저장 및 불러오기
# 애니메이션
render_episode(env, model)
# 저장
model.save("dqn_cartpole")
# 삭제
del model
# 불러오기
model = DQN.load("dqn_cartpole")

 

## QNetwork
class QNetwork(BasePolicy): # DQN을 위한 행동-가치(Q-Value) 네트워크
action_space: spaces.Discrete # 행동 공간은 이산적(discrete)이어야 함
def __init__(
self,
observation_space: spaces.Space, # 관측 공간 정의
action_space: spaces.Discrete, # 행동 공간 정의 (이산적)
features_extractor: BaseFeaturesExtractor, # 특징 추출기
features_dim: int, # 특징 차원
net_arch: Optional[List[int]] = None, # 네트워크 구조 (옵션)
activation_fn: Type[nn.Module] = nn.ReLU, # 활성화 함수 (기본값: ReLU)
normalize_images: bool = True, # 이미지 정규화 여부 (기본값: True, 255.0으로 나눔)
) -> None:

## QNetwork.__init__
# 부모 클래스 초기화
super().__init__(
observation_space,
action_space,
features_extractor=features_extractor,
normalize_images=normalize_images,
)
# 네트워크 구조가 지정되지 않았다면 기본값 [64, 64] 사용
if net_arch is None:
net_arch = [64, 64]

# 클래스 변수 설정
self.net_arch = net_arch
self.activation_fn = activation_fn
self.features_dim = features_dim
action_dim = int(self.action_space.n) # 행동의 수
# Q-네트워크 생성 (MLP: 다층 퍼셉트론)
q_net = create_mlp(self.features_dim, action_dim, self.net_arch, self.activation_fn)
self.q_net = nn.Sequential(*q_net) # Sequential 모델로 구성

## QNetwork.forward
def forward(self, obs: PyTorchObs) -> th.Tensor:
    """
    Q-값을 예측합니다.
    :param obs: 관측값
    :return: 각 행동에 대한 예측 Q-값
    """
    # 특징을 추출하고 Q-네트워크를 통과시켜 Q-값 계산
    # (기본적으로는 아무 특징도 추출하지 않음)
    return self.q_net(self.extract_features(obs, self.features_extractor))

## QNetwork._predict
def _predict(self, observation: PyTorchObs, deterministic: bool = True) -> th.Tensor:
    # Q-값 계산
    q_values = self(observation)
    # 최대 Q-값을 가진 행동 선택 (탐욕적 방식)
    action = q_values.argmax(dim=1).reshape(-1)
    return action

 

## DQNPolicy
class DQNPolicy(BasePolicy):
    """
    DQN을 위한 Q-Value 네트워크와 목표 네트워크를 포함하는 정책 클래스
    :param observation_space: 관측 공간
    :param action_space: 행동 공간
    :param lr_schedule: 학습률 스케줄 (고정값일 수도 있음)
    :param net_arch: 정책 및 가치 네트워크의 구조
    :param activation_fn: 활성화 함수
    :param features_extractor_class: 사용할 특징 추출기 클래스
    :param features_extractor_kwargs: 특징 추출기에 전달할 키워드 인자
    :param normalize_images: 이미지 정규화 여부 (기본값: True, 255.0으로 나눔)
    :param optimizer_class: 사용할 최적화 알고리즘 (기본값: Adam)
    :param optimizer_kwargs: 최적화 알고리즘에 전달할 추가 키워드 인자 (학습률 제외)
    """
    q_net: QNetwork # 메인 Q-네트워크
    q_net_target: QNetwork # 목표 Q-네트워크
   

## DQNPolicy._build
def _build(self, lr_schedule: Schedule) -> None:
    """
    __init__에서 네트워크와 최적화 알고리즘을 생성
    :param lr_schedule: 학습률 스케줄 (lr_schedule(1)이 초기 학습률)
    """
    # 메인 Q-네트워크와 목표 Q-네트워크 생성
    self.q_net = self.make_q_net()
    self.q_net_target = self.make_q_net()
    # 목표 네트워크의 가중치를 메인 네트워크와 동일하게 초기화
    self.q_net_target.load_state_dict(self.q_net.state_dict())
    self.q_net_target.set_training_mode(False) # 목표 네트워크를 평가 모드로 설정
    self.optimizer = self.optimizer_class(… # 최적화 알고리즘 설정

 

## DQN
stable_baselines3/dqn/policies.py
class DQN(OffPolicyAlgorithm):
    """
    :param policy: 사용할 정책 모델 (MlpPolicy, CnnPolicy, ...)
    :param env: 학습할 환경 (Gym에 등록된 경우 문자열로 지정 가능)
    :param learning_rate: 학습률, 현재 진행 상황에 따른 함수일 수 있음 (1에서 0으로)
    :param buffer_size: 리플레이 버퍼의 크기
    :param learning_starts: 학습 시작 전 수집할 트랜지션 수
    :param batch_size: 각 그래디언트 업데이트의 미니배치 크기
    :param tau: 소프트 업데이트 계수 ("Polyak 업데이트", 0과 1 사이) 기본값 1은 하드 업데이트
    :param gamma: 할인 계수
    :param train_freq: 모델 업데이트 주기. (5, "step") 또는 (2, "episode") 같은 튜플로 지정 가능
    :param gradient_steps: 각 롤아웃 후 수행할 그래디언트 스텝 수
    :param replay_buffer_class: 사용할 리플레이 버퍼 클래스 (예: HerReplayBuffer)
    :param replay_buffer_kwargs: 리플레이 버퍼 생성 시 전달할 키워드 인자
    :param optimize_memory_usage: 메모리 효율적인 리플레이 버퍼 변형 사용 여부
    :param target_update_interval: 타겟 네트워크 업데이트 주기 (환경 스텝 단위)
    :param exploration_fraction: 탐험률을 줄이는 전체 학습 기간의 비율
    :param exploration_initial_eps: 초기 랜덤 행동 확률
    :param exploration_final_eps: 최종 랜덤 행동 확률
    :param max_grad_norm: 그래디언트 클리핑의 최대값
    :param stats_window_size: 롤아웃 로깅을 위한 윈도우 크기
    :param tensorboard_log: 텐서보드 로그 위치
    :param policy_kwargs: 정책 생성 시 전달할 추가 인자
    :param verbose: 출력 레벨 (0: 없음, 1: 정보 메시지, 2: 디버그 메시지)
    :param seed: 의사 난수 생성기의 시드
    :param device: 코드를 실행할 장치 (cpu, cuda, ...)
    :param _init_setup_model: 인스턴스 생성 시 네트워크를 구축할지 여부

 

## DQN.learn
# OffPolicyAlgorithm.learn을 상속
# stable_baselines3/common/off_policy_algorithm.py

def learn(…):
    callback.on_training_start(locals(), globals()) # 트레이닝 시작 콜백
    …
    while self.num_timesteps < total_timesteps:
    	rollout = self.collect_rollouts(…) # 롤아웃(현재 정책으로 행동하여 데이터 수집)
    	…
    	self.train(…) # 수집된 데이터에 훈련
    callback.on_training_end() # 트레이닝 종료 콜백
    return self

## DQN.train
def train(self, gradient_steps: int, batch_size: int = 100) -> None:
    self.policy.set_training_mode(True) # 훈련 모드로 전환 (배치 정규화 / 드롭아웃에 영향)
    self._update_learning_rate(self.policy.optimizer) # 스케줄에 따라 학습률 업데이트
    losses = []
    for _ in range(gradient_steps):
        # 리플레이 버퍼에서 샘플 추출
        replay_data = self.replay_buffer.sample(batch_size, env=self._vec_normalize_env)
        
## DQN.train: TD 타겟 계산
# TD 타겟으로 오차가 역전파 되지 않도록 경사 계산에서 제외
with th.no_grad():
    # 타겟 네트워크를 사용하여 다음 Q-값 계산
    next_q_values = self.q_net_target(replay_data.next_observations)
    next_q_values, _ = next_q_values.max(dim=1) # 탐욕적 정책: 가장 높은 값 선택
    next_q_values = next_q_values.reshape(-1, 1)
    
    # 1-스텝 TD 타겟
    target_q_values = replay_data.rewards + (1 - replay_data.dones) * self.gamma * next_q_values
    
## DQN.train: 파라미터 업데이트
# 현재 Q-값 추정치 얻기
current_q_values = self.q_net(replay_data.observations)

# 리플레이 버퍼의 행동에 대한 q-값 검색
current_q_values = th.gather(current_q_values, dim=1, index=replay_data.actions.long())

# Huber 손실
loss = F.smooth_l1_loss(current_q_values, target_q_values)
losses.append(loss.item())

# 오차 역전파
self.policy.optimizer.zero_grad()
loss.backward()

 

## DQN.predict
def predict(…
    """
    입실론-그리디 탐색
    """
    if not deterministic and np.random.rand() < self.exploration_rate: # 랜덤 행동 선택 (탐색)
    if self.policy.is_vectorized_observation(observation):
        if isinstance(observation, dict):
        	n_batch = observation[next(iter(observation.keys()))].shape[0]
        else:
        	n_batch = observation.shape[0]
        action = np.array([self.action_space.sample() for _ in range(n_batch)])
	else:
        action = np.array(self.action_space.sample())
    else: # 정책에 따른 행동 선택
    	action, state = self.policy.predict(observation, state, episode_start, deterministic