深度Q网络（DQN）算法技术博客

news2024/10/7 9:16:39

深度Q网络（DQN）是一种将深度学习与强化学习相结合的算法，用于解决高维状态空间的强化学习问题。本文将详细介绍DQN算法的基本原理，关键公式以及具体的代码实现。

一、DQN算法的基本原理

DQN算法是Q学习的一种扩展，利用神经网络来逼近Q值函数。其核心思想是通过不断地与环境交互，从而学习到一个策略，使得在每个状态下的累积奖励最大化。Q值函数的定义如下：

$Q(s, a) = \mathbb{E}[r_t + \gamma \max_{a'} Q(s_{t+1}, a') | s_t = s, a_t = a]$

其中：

$s$ 是状态
$a$ 是动作
$r$ 是奖励
$\gamma$ γ是折扣因子（0 <= $\gamma$ < 1）

DQN通过使用两个神经网络来稳定训练过程：

策略网络（Policy Network）：用来生成动作的Q值。
目标网络（Target Network）：用来生成目标Q值，其参数定期从策略网络复制。

二、DQN算法的关键步骤

经验回放（Experience Replay）：为了打破数据之间的相关性，DQN使用了经验回放技术，将经验存储在一个记忆库中，并从中随机采样一批用于训练。
目标Q值的计算： $y_i = r_i + \gamma \max_{a'} Q'(s_{i+1}, a'; \theta^{-})$ 其中 $\theta^{-}$ 是目标网络的参数， $\theta$ 是策略网络的参数。
损失函数的定义： $L(\theta) = \mathbb{E}_{(s, a, r, s') \sim D} [(y_i - Q(s, a; \theta))^2]$ 通过最小化上述损失函数，来更新策略网络的参数。

三、DQN算法的代码实现

以下是一个简单的DQN算法在OpenAI Gym的CartPole环境中的实现。

import gym
import numpy as np
import random
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque

# 定义Q网络
class QNetwork(nn.Module):
    def __init__(self, state_size, action_size):
        super(QNetwork, self).__init__()
        self.fc1 = nn.Linear(state_size, 24)
        self.fc2 = nn.Linear(24, 24)
        self.fc3 = nn.Linear(24, action_size)
    
    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

# DQN算法类
class DQNAgent:
    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.memory = deque(maxlen=2000)
        self.gamma = 0.95  # 折扣因子
        self.epsilon = 1.0  # 探索率
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995
        self.learning_rate = 0.001
        self.model = QNetwork(state_size, action_size)
        self.target_model = QNetwork(state_size, action_size)
        self.optimizer = optim.Adam(self.model.parameters(), lr=self.learning_rate)
        self.update_target_model()
    
    def update_target_model(self):
        self.target_model.load_state_dict(self.model.state_dict())
    
    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))
    
    def act(self, state):
        if np.random.rand() <= self.epsilon:
            return random.randrange(self.action_size)
        state = torch.FloatTensor(state)
        act_values = self.model(state)
        return np.argmax(act_values.detach().numpy())
    
    def replay(self, batch_size):
        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            target = self.model(torch.FloatTensor(state)).detach().numpy()
            if done:
                target[action] = reward
            else:
                t = self.target_model(torch.FloatTensor(next_state)).detach().numpy()
                target[action] = reward + self.gamma * np.amax(t)
            target_f = self.model(torch.FloatTensor(state))
            target_f[action] = torch.FloatTensor([target[action]])
            self.model.zero_grad()
            loss = nn.MSELoss()(target_f, torch.FloatTensor(target))
            loss.backward()
            self.optimizer.step()
        
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

# 训练DQN模型
if __name__ == "__main__":
    env = gym.make("CartPole-v1")
    state_size = env.observation_space.shape[0]
    action_size = env.action_space.n
    agent = DQNAgent(state_size, action_size)
    episodes = 1000
    batch_size = 32
    
    for e in range(episodes):
        state = env.reset()
        state = np.reshape(state, [1, state_size])
        for time in range(500):
            action = agent.act(state)
            next_state, reward, done, _ = env.step(action)
            reward = reward if not done else -10
            next_state = np.reshape(next_state, [1, state_size])
            agent.remember(state, action, reward, next_state, done)
            state = next_state
            if done:
                agent.update_target_model()
                print(f"Episode: {e}/{episodes}, Score: {time}, Epsilon: {agent.epsilon:.2}")
                break
            if len(agent.memory) > batch_size:
                agent.replay(batch_size)