强化学习（RL）
强化学习（Reinforcement Learning，简称RL）是机器学习的一个重要分支，它主要关注如何在环境中采取行动以最大化某种累积奖励。强化学习的核心是智能体（Agent）通过与环境（Environment）的交互来学习最佳策略，即在给定状态下选择最佳动作的规则。
关键概念
- 状态（State）：智能体所处的环境状态。
- 动作（Action）：智能体在给定状态下可以执行的可能动作。
- 奖励（Reward）：智能体执行动作后从环境中获得的即时反馈。
- 策略（Policy）：从状态到动作的映射，指导智能体在特定状态下应采取的动作。
- 价值函数（Value Function）：预测采取某个策略后获得的累积奖励。
- Q函数（Q-Function）：预测在给定状态下采取特定动作的期望累积奖励。
DQN
原理
DQN （Deep Q-Network）核心思想是利用 Bellman 公式的 bootstrap 特性，不断迭代优化一个 Q(s,a)函数。Q(s,a)函数拟合的是一对状态 - 动作的长期收益评估。DQN 探索和利用的平衡靠的是一种称为ε-greedy 的策略，针对最新的 Q(s,a)函数和当前的输入状态 s，agent 做决策时以概率ε随机选择 action，而以 1-ε的概率选择使 Q(s,a)最大的 action，随着ε从大到小变化，DQN 也相应地从 “强探索弱利用” 过渡到“弱探索强利用”。
DQN结合了深度学习与Q学习，使用神经网络来近似Q函数。
前提条件
DQN 原理使其天然地适合离散动作空间，也就是 action 可以穷举。比如走迷宫的 agent 只允许前后左右 4 个动作，下围棋的 AlphaGo 只允许 19*19=361 个落子位置（实际还要排除已经落子的网格点）。这是一个重要的特征，如果你手上是一个连续控制任务，action 在某区间内有无数种可能，那就不适合用 DQN 了。
实现步骤
- 初始化：初始化DQN网络和经验回放缓冲区。
- 探索与利用：智能体在每个时间步选择一个动作，这通常通过ε-贪心策略来实现，即以ε的概率随机选择动作，以1-ε的概率选择当前估计的最优动作。
- 执行动作：智能体执行选择的动作，并观察环境的反馈，即下一个状态和奖励。
- 存储经验：将当前状态、选择的动作、获得的奖励和下一个状态存储到经验回放缓冲区。
- 样本采样：从经验回放缓冲区中随机采样一批数据。
- 计算目标Q值：使用目标网络计算采样数据的目标Q值。
- 损失计算与网络更新：计算当前网络输出的Q值与目标Q值之间的差异，并使用梯度下降算法更新网络权重。
- 目标网络更新：定期更新目标网络的参数，通常是将主网络的参数复制给目标网络。
- 重复步骤2-8：直到智能体学会最优策略。
import random
import torch
import torch.nn as nn
import torch.optim as optim
# 定义DQN网络结构
class DQN(nn.Module):
def __init__(self, input_size, output_size):
super(DQN, self).__init__()
self.network = nn.Sequential(
nn.Linear(input_size, 128),
nn.ReLU(),
nn.Linear(128, 128),
nn.ReLU(),
nn.Linear(128, output_size)
)
def forward(self, x):
return self.network(x)
# 环境参数
input_size = 4 # 假设状态空间的大小为4
output_size = 2 # 假设动作空间的大小为2
# 初始化DQN网络
dqn = DQN(input_size, output_size)
# 定义超参数
learning_rate = 0.01
batch_size = 32
gamma = 0.99 # 折扣因子
epsilon = 1.0 # 探索率
epsilon_min = 0.01
epsilon_decay = 0.995
# 经验回放缓冲区
memory = []
# 优化器和损失函数
optimizer = optim.Adam(dqn.parameters(), lr=learning_rate)
criterion = nn.MSELoss()
# 目标网络
target_dqn = DQN(input_size, output_size)
target_dqn.load_state_dict(dqn.state_dict())
# 训练循环
for episode in range(1000): # 假设我们进行1000个训练周期
state = env.reset() # 假设env是一个环境对象，可以重置环境状态
done = False
while not done:
# 探索与利用
if random.uniform(0, 1) < epsilon:
action = random.randint(0, output_size - 1) # 随机选择动作
else:
action = dqn(torch.tensor(state, dtype=torch.float32)).argmax().item() # 选择最优动作
# 执行动作并获取反馈
next_state, reward, done, _ = env.step(action) # 假设env.step返回下一个状态和奖励
# 存储经验
memory.append((state, action, reward, next_state, done))
# 经验回放
if len(memory) > batch_size:
sample = random.sample(memory, batch_size)
states, actions, rewards, next_states, dones = zip(*sample)
# 计算当前Q值和目标Q值
current_q_values = dqn(torch.tensor(states, dtype=torch.float32)).squeeze(1)
next_q_values = target_dqn(torch.tensor(next_states, dtype=torch.float32)).detach().squeeze(1)
expected_q_values = rewards + (gamma * next_q_values * (1 - dones))
# 计算损失并更新网络
loss = criterion(current_q_values, expected_q_values)
optimizer.zero_grad()
loss.backward()
optimizer.step()
# 更新探索率
epsilon = max(epsilon_min, epsilon * epsilon_decay)
# 每100个周期更新一次目标网络
if episode % 100 == 0:
target_dqn.load_state_dict(dqn.state_dict())
state = next_state
# 保存训练好的模型
torch.save(dqn.state_dict(), 'dqn_model.ckpt')
PPO
PPO（Proximal Policy Optimization）是一种策略梯度方法，是一种无模型的强化学习方法，旨在优化策略网络，使得期望奖励最大化。PPO的核心思想是在每一步更新中，限制策略更新的步长，确保更新后的策略与当前策略足够接近，从而避免大幅度的更新导致性能下降。
前提条件
PPO是一种先进的策略梯度强化学习算法，它特别适用于需要长期依赖历史经验的复杂任务。PPO算法的核心优势在于其稳定性和快速收敛性，这使得它在多个领域都有广泛的应用。
实现步骤
- 初始化：初始化策略网络和价值网络（可选）。
- 收集数据：通过当前策略网络与环境交互，收集状态、动作、奖励和下一个状态的数据。
- 计算目标：使用重要性采样计算每个样本的目标值。
- 截断策略：计算PPO-Clip的目标函数，并应用截断策略。
- 更新网络：使用目标函数和截断策略来更新策略网络。
- 重复步骤2-5：直到策略网络收敛或达到预定的迭代次数。
import gym
import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical
# 定义PPO的策略网络和价值网络
class ActorCriticNetwork(nn.Module):
def __init__(self, state_dim, action_dim):
super(ActorCriticNetwork, self).__init__()
self.fc = nn.Sequential(
nn.Linear(state_dim, 64),
nn.ReLU(),
nn.Linear(64, 64),
nn.ReLU(),
nn.Linear(64, action_dim), # 策略网络输出层
nn.Softmax(dim=-1)
)
self.value_fc = nn.Sequential(
nn.Linear(state_dim, 64),
nn.ReLU(),
nn.Linear(64, 64),
nn.ReLU(),
nn.Linear(64, 1) # 价值网络输出层
)
def forward(self, state):
probs = self.fc(state)
action_log_probs = torch.log(probs)
state_values = self.value_fc(state)
return probs, action_log_probs, state_values
# PPO智能体
class PPOAgent:
def __init__(self, env, gamma=0.99, K_epochs=4, eps_clip=0.2, lr=0.0003):
self.gamma = gamma
self.K_epochs = K_epochs
self.eps_clip = eps_clip
self.env = env
self.policy_net = ActorCriticNetwork(env.observation_space.shape[0], env.action_space.n)
self.optimizer = optim.Adam(self.policy_net.parameters(), lr=lr)
self.MseLoss = nn.MSELoss()
def select_action(self, state):
state = torch.FloatTensor(state)
probs, action_log_probs, state_values = self.policy_net(state)
dist = Categorical(probs)
action = dist.sample()
return action.item(), action_log_probs[action].item(), state_values.item()
def update(self, memory):
# 转换为PyTorch张量
states, actions, rewards, next_states, dones = map(torch.tensor, memory)
old_probs, old_action_log_probs, old_state_values = self.policy_net(states)
# 计算PPO的目标函数
with torch.no_grad():
new_probs, new_action_log_probs, new_state_values = self.policy_net(states)
ratios = torch.exp(new_action_log_probs - old_action_log_probs)
surr1 = ratios * old_action_log_probs * rewards
surr2 = torch.clamp(ratios, 1-self.eps_clip, 1+self.eps_clip) * old_action_log_probs * rewards
policy_loss = -torch.min(surr1, surr2).mean()
# 更新策略网络
self.optimizer.zero_grad()
policy_loss.backward()
self.optimizer.step()
# 计算价值网络的损失
value_loss = self.MseLoss(new_state_values, rewards)
self.optimizer.zero_grad()
value_loss.backward()
self.optimizer.step()
def train(self):
memory = []
states = self.env.reset()
for _ in range(100): # 假设我们进行100个训练周期
# 采样和执行动作
action, action_log_prob, state_value = self.select_action(states)
next_states, rewards, dones, _ = self.env.step(action)
# 存储经验
memory.append((states, action, rewards, next_states, dones))
# 转换经验到张量
for i in range(len(memory)):
states = memory[i][3]
if memory[i][4]: # 如果完成，则结束
break
# 训练PPO
for _ in range(self.K_epochs):
self.update(memory)
# 清除内存
memory.clear()
states = next_states
# 创建环境和智能体
env = gym.make('CartPole-v1')
ppo_agent = PPOAgent(env)
ppo_agent.train()