从Q-learning到DQN：用Python和PyTorch一步步搭建你的第一个智能体（附完整代码）

从Q-learning到DQN用Python和PyTorch搭建智能体的实战指南1. 强化学习基础与环境搭建在开始构建智能体之前我们需要先理解强化学习的核心概念。强化学习是一种通过与环境交互来学习最优行为的机器学习方法。智能体通过尝试不同的动作观察环境的反馈奖励逐步调整策略以获得最大累积奖励。让我们先创建一个简单的网格世界环境这将作为我们智能体的训练场import numpy as np class GridWorld: def __init__(self, size5): self.size size self.goal (size-1, size-1) self.obstacles [(1,1), (2,3), (3,1)] self.reset() def reset(self): self.state (0, 0) return self.state def step(self, action): x, y self.state if action 0: # 上 x max(0, x-1) elif action 1: # 右 y min(self.size-1, y1) elif action 2: # 下 x min(self.size-1, x1) elif action 3: # 左 y max(0, y-1) self.state (x, y) if self.state in self.obstacles: reward -10 done True elif self.state self.goal: reward 100 done True else: reward -1 done False return self.state, reward, done, {}这个5×5的网格世界包含起点(0,0)终点(4,4)到达获得100奖励障碍物(1,1)、(2,3)、(3,1)碰到会结束并获得-10奖励每移动一步获得-1奖励鼓励智能体尽快到达终点提示在设计强化学习环境时奖励函数的设计至关重要。过大的负奖励可能导致智能体过于保守而过小的负奖励又可能导致学习缓慢。2. Q-learning算法实现Q-learning是一种基于表格的强化学习算法它通过学习状态-动作对的Q值来找到最优策略。Q值表示在特定状态下采取特定动作的长期收益。2.1 Q表初始化与更新首先我们实现Q-learning的核心逻辑class QLearningAgent: def __init__(self, state_size, action_size, learning_rate0.1, discount_factor0.95, epsilon0.1): self.q_table np.zeros((state_size, state_size, action_size)) self.lr learning_rate self.gamma discount_factor self.epsilon epsilon def get_action(self, state): if np.random.rand() self.epsilon: return np.random.randint(0, len(self.q_table[0][0])) else: return np.argmax(self.q_table[state[0]][state[1]]) def update(self, state, action, reward, next_state, done): current_q self.q_table[state[0]][state[1]][action] if done: target_q reward else: target_q reward self.gamma * np.max(self.q_table[next_state[0]][next_state[1]]) self.q_table[state[0]][state[1]][action] self.lr * (target_q - current_q)关键参数说明learning_rate控制新信息覆盖旧信息的速度discount_factor未来奖励的折扣率epsilon探索概率控制智能体尝试随机动作的频率2.2 训练过程实现下面是完整的训练循环def train_q_learning(env, agent, episodes1000): rewards [] for episode in range(episodes): state env.reset() total_reward 0 done False while not done: action agent.get_action(state) next_state, reward, done, _ env.step(action) agent.update(state, action, reward, next_state, done) state next_state total_reward reward rewards.append(total_reward) if episode % 100 0: print(fEpisode {episode}, Total Reward: {total_reward}) return rewards训练完成后我们可以可视化Q表来理解智能体学到了什么def visualize_q_table(q_table): directions [↑, →, ↓, ←] for i in range(q_table.shape[0]): row [] for j in range(q_table.shape[1]): best_action np.argmax(q_table[i][j]) row.append(directions[best_action]) print( .join(row))注意Q-learning在状态空间较小时效果很好但当状态空间增大时Q表会变得非常庞大且难以维护。这正是我们需要深度Q网络(DQN)的原因。3. 深度Q网络(DQN)实现DQN使用神经网络来近似Q函数解决了Q-learning在高维状态空间中的局限性。我们将使用PyTorch来实现DQN。3.1 神经网络架构import torch import torch.nn as nn import torch.optim as optim class DQN(nn.Module): def __init__(self, input_size, output_size): super(DQN, self).__init__() self.fc1 nn.Linear(input_size, 64) self.fc2 nn.Linear(64, 64) self.fc3 nn.Linear(64, output_size) def forward(self, x): x torch.relu(self.fc1(x)) x torch.relu(self.fc2(x)) return self.fc3(x)这个网络结构包含输入层接收状态表示两个隐藏层64个神经元使用ReLU激活函数输出层输出每个动作的Q值3.2 经验回放缓冲区经验回放是DQN的关键组件它存储智能体的经验(状态、动作、奖励、新状态)并从中随机采样进行训练from collections import deque import random class ReplayBuffer: def __init__(self, capacity): self.buffer deque(maxlencapacity) def push(self, state, action, reward, next_state, done): self.buffer.append((state, action, reward, next_state, done)) def sample(self, batch_size): return random.sample(self.buffer, batch_size) def __len__(self): return len(self.buffer)3.3 DQN智能体实现class DQNAgent: def __init__(self, state_size, action_size): self.state_size state_size self.action_size action_size self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.policy_net DQN(state_size*2, action_size).to(self.device) self.target_net DQN(state_size*2, action_size).to(self.device) self.target_net.load_state_dict(self.policy_net.state_dict()) self.target_net.eval() self.optimizer optim.Adam(self.policy_net.parameters(), lr1e-3) self.memory ReplayBuffer(10000) self.batch_size 64 self.gamma 0.95 self.epsilon 1.0 self.epsilon_min 0.01 self.epsilon_decay 0.995 def get_action(self, state): if np.random.rand() self.epsilon: return np.random.randint(self.action_size) state torch.FloatTensor(state).unsqueeze(0).to(self.device) with torch.no_grad(): q_values self.policy_net(state) return q_values.argmax().item() def update(self): if len(self.memory) self.batch_size: return batch self.memory.sample(self.batch_size) states, actions, rewards, next_states, dones zip(*batch) states torch.FloatTensor(np.array(states)).to(self.device) actions torch.LongTensor(np.array(actions)).unsqueeze(1).to(self.device) rewards torch.FloatTensor(np.array(rewards)).unsqueeze(1).to(self.device) next_states torch.FloatTensor(np.array(next_states)).to(self.device) dones torch.FloatTensor(np.array(dones)).unsqueeze(1).to(self.device) current_q self.policy_net(states).gather(1, actions) next_q self.target_net(next_states).max(1)[0].unsqueeze(1) target_q rewards (1 - dones) * self.gamma * next_q loss nn.MSELoss()(current_q, target_q.detach()) self.optimizer.zero_grad() loss.backward() self.optimizer.step() if self.epsilon self.epsilon_min: self.epsilon * self.epsilon_decay def update_target(self): self.target_net.load_state_dict(self.policy_net.state_dict())4. 训练与优化技巧4.1 完整训练流程def train_dqn(env, agent, episodes1000, target_update10): rewards [] for episode in range(episodes): state env.reset() state np.array(state) total_reward 0 done False while not done: action agent.get_action(state) next_state, reward, done, _ env.step(action) next_state np.array(next_state) agent.memory.push(state, action, reward, next_state, done) state next_state total_reward reward agent.update() rewards.append(total_reward) if episode % target_update 0: agent.update_target() if episode % 100 0: print(fEpisode {episode}, Total Reward: {total_reward}, Epsilon: {agent.epsilon:.2f}) return rewards4.2 常见问题与解决方案在训练DQN时经常会遇到以下问题训练不稳定使用目标网络减少目标Q值的波动适当调整学习率增加批量大小奖励稀疏重新设计奖励函数提供更密集的反馈使用好奇心驱动探索尝试分层强化学习过估计问题实现Double DQN使用Dueling DQN架构4.3 性能优化技巧优先经验回放给重要的经验样本更高的采样概率学习率调度随着训练进行逐步降低学习率梯度裁剪防止梯度爆炸多步学习使用n步回报而不是单步回报# 优先经验回放示例 class PrioritizedReplayBuffer(ReplayBuffer): def __init__(self, capacity, alpha0.6): super().__init__(capacity) self.priorities deque(maxlencapacity) self.alpha alpha def push(self, *args): super().push(*args) self.priorities.append(max(self.priorities, default1.0)) def sample(self, batch_size, beta0.4): priorities np.array(self.priorities) probs priorities ** self.alpha probs / probs.sum() indices np.random.choice(len(self.buffer), batch_size, pprobs) samples [self.buffer[idx] for idx in indices] weights (len(self.buffer) * probs[indices]) ** (-beta) weights / weights.max() return samples, indices, np.array(weights, dtypenp.float32)5. 结果可视化与分析训练完成后我们可以通过多种方式评估智能体的表现5.1 训练曲线import matplotlib.pyplot as plt def plot_training(rewards, window100): plt.figure(figsize(10,5)) moving_avg np.convolve(rewards, np.ones(window)/window, modevalid) plt.plot(rewards, alpha0.3, labelEpisode Reward) plt.plot(moving_avg, labelfMoving Average ({window} episodes)) plt.xlabel(Episode) plt.ylabel(Total Reward) plt.legend() plt.show()5.2 策略可视化def visualize_policy(env, agent): state env.reset() done False trajectory [state] while not done: state_tensor torch.FloatTensor(state).unsqueeze(0).to(agent.device) with torch.no_grad(): action agent.policy_net(state_tensor).argmax().item() state, _, done, _ env.step(action) trajectory.append(state) grid np.zeros((env.size, env.size)) for x, y in trajectory: grid[x][y] 1 plt.imshow(grid, cmapBlues) for (x, y) in env.obstacles: plt.text(y, x, X, hacenter, vacenter, colorred) plt.text(env.goal[1], env.goal[0], G, hacenter, vacenter, colorgreen) plt.xticks([]) plt.yticks([]) plt.title(Agent Trajectory) plt.show()5.3 Q值分布分析def plot_q_distribution(agent): q_values [] for i in range(env.size): for j in range(env.size): state torch.FloatTensor([i,j]).to(agent.device) with torch.no_grad(): q agent.policy_net(state).cpu().numpy() q_values.append(q.max()) plt.hist(q_values, bins20) plt.xlabel(Max Q Value) plt.ylabel(Frequency) plt.title(Q Value Distribution Across States) plt.show()在实际项目中我发现调整探索率(epsilon)的衰减速度对训练效果影响很大。过快的衰减可能导致智能体过早停止探索而过慢的衰减又会导致训练效率低下。经过多次实验我发现从1.0开始以0.995的速率衰减到0.01左右是一个不错的起点。