08-强化学习基础¶

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.distributions import Categorical

class ActorCritic(nn.Module):  # 继承nn.Module定义神经网络层
    """共享底层特征，分别输出策略和价值的Actor-Critic网络"""
    def __init__(self, state_dim, action_dim, hidden_dim=128):  # __init__构造方法，创建对象时自动调用
        super().__init__()  # super()调用父类方法
        # 共享特征提取层：Actor和Critic共用，减少参数量
        self.shared = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
        )
        # Actor头：输出各动作的概率分布
        self.actor = nn.Linear(hidden_dim, action_dim)
        # Critic头：输出当前状态的价值估计V(s)
        self.critic = nn.Linear(hidden_dim, 1)

    def forward(self, state):
        features = self.shared(state)                           # 提取共享特征
        action_probs = F.softmax(self.actor(features), dim=-1)  # 动作概率（softmax归一化）
        state_value = self.critic(features)                     # 状态价值V(s)
        return action_probs, state_value

    def select_action(self, state):
        """根据策略分布采样动作，同时返回log概率和价值（用于后续梯度计算）"""
        action_probs, value = self.forward(state)
        dist = Categorical(action_probs)  # 构建离散分布
        action = dist.sample()            # 按概率采样动作
        return action.item(), dist.log_prob(action), value  # .item()将单元素张量转为Python数值

def train_actor_critic(env, model, optimizer, n_episodes=1000, gamma=0.99):
    """训练Actor-Critic模型"""
    for episode in range(n_episodes):
        state, _ = env.reset()
        log_probs, values, rewards = [], [], []  # 存储轨迹数据

        # 采样一个完整episode
        done = False
        while not done:
            state_tensor = torch.FloatTensor(state)
            action, log_prob, value = model.select_action(state_tensor)

            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            log_probs.append(log_prob)   # 动作的log概率
            values.append(value)          # Critic的价值估计
            rewards.append(reward)        # 环境奖励
            state = next_state

        # 从后往前计算每步的折扣回报G_t，并标准化
        returns = []
        G = 0
        for r in reversed(rewards):
            G = r + gamma * G
            returns.insert(0, G)
        returns = torch.tensor(returns)
        returns = (returns - returns.mean()) / (returns.std() + 1e-8)  # 标准化降低方差

        # 计算Actor和Critic的损失
        policy_loss = 0
        value_loss = 0
        for log_prob, value, G in zip(log_probs, values, returns):  # zip按位置配对多个可迭代对象
            advantage = G - value.item()                       # 优势 = 实际回报 - 价值估计
            policy_loss -= log_prob * advantage                # Actor损失：增大高优势动作的概率
            value_loss += F.smooth_l1_loss(value.squeeze(), G) # Critic损失：让V(s)逼近实际回报

        # 联合优化：policy_loss + 0.5 * value_loss
        loss = policy_loss + 0.5 * value_loss
        optimizer.zero_grad()  # 清零梯度，防止梯度累积
        loss.backward()  # 反向传播计算梯度
        optimizer.step()  # 根据梯度更新模型参数

        if (episode + 1) % 100 == 0:
            print(f"Episode {episode+1}, Total Reward: {sum(rewards):.1f}")

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from collections import deque
import random

class DQN(nn.Module):
    """深度Q网络：用神经网络近似Q(s,a)函数"""
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        # 三层全连接网络：输入状态 → 输出每个动作的Q值
        self.net = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),  # 输出维度=动作数
        )

    def forward(self, x):
        return self.net(x)

class ReplayBuffer:
    """经验回放缓冲区：存储历史转移，随机采样打破时间相关性"""
    def __init__(self, capacity=10000):
        self.buffer = deque(maxlen=capacity)  # 固定容量的双端队列

    def push(self, state, action, reward, next_state, done):
        """存储一条转移记录 (s, a, r, s', done)"""
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        """随机采样一批数据，转换为Tensor格式"""
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (
            torch.FloatTensor(np.array(states)),  # np.array创建NumPy数组
            torch.LongTensor(actions),
            torch.FloatTensor(rewards),
            torch.FloatTensor(np.array(next_states)),
            torch.FloatTensor(dones),
        )

    def __len__(self):  # __len__定义len()的行为
        return len(self.buffer)

class DQNAgent:
    """DQN智能体：集成Q网络、目标网络、经验回放和ε-贪心探索"""
    def __init__(self, state_dim, action_dim, lr=1e-3, gamma=0.99,
                 epsilon_start=1.0, epsilon_end=0.01, epsilon_decay=0.995):
        self.action_dim = action_dim
        self.gamma = gamma               # 折扣因子
        self.epsilon = epsilon_start      # 探索率（从高到低衰减）
        self.epsilon_end = epsilon_end
        self.epsilon_decay = epsilon_decay

        self.q_net = DQN(state_dim, action_dim)         # 在线Q网络（实时训练）
        self.target_net = DQN(state_dim, action_dim)    # 目标网络（延迟更新，稳定训练）
        self.target_net.load_state_dict(self.q_net.state_dict())  # 初始参数同步

        self.optimizer = torch.optim.Adam(self.q_net.parameters(), lr=lr)
        self.buffer = ReplayBuffer()

    def select_action(self, state):
        """ε-贪心策略：以ε概率随机探索，1-ε概率选Q值最大的动作"""
        if random.random() < self.epsilon:
            return random.randrange(self.action_dim)  # 随机探索
        with torch.no_grad():  # 禁用梯度计算，节省内存
            q_values = self.q_net(torch.FloatTensor(state))
            return q_values.argmax().item()           # 贪心选择

    def update(self, batch_size=64):
        """从经验池采样一批数据，执行一步梯度更新"""
        if len(self.buffer) < batch_size:
            return  # 经验不足，跳过

        states, actions, rewards, next_states, dones = self.buffer.sample(batch_size)

        # 计算当前Q值：Q(s,a;θ)
        current_q = self.q_net(states).gather(1, actions.unsqueeze(1)).squeeze()

        # 计算TD目标：r + γ * max_a' Q(s',a';θ⁻)，使用目标网络避免自举不稳
        with torch.no_grad():
            max_next_q = self.target_net(next_states).max(dim=1)[0]
            target_q = rewards + self.gamma * max_next_q * (1 - dones)

        # 最小化TD误差（Huber损失更鲁棒）
        loss = F.smooth_l1_loss(current_q, target_q)
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        # 逐步衰减探索率：从探索为主过渡到利用为主
        self.epsilon = max(self.epsilon_end, self.epsilon * self.epsilon_decay)

    def update_target(self):
        """将在线网络的参数同步到目标网络（硬更新）"""
        self.target_net.load_state_dict(self.q_net.state_dict())

def train_dqn(env_name='CartPole-v1', n_episodes=500, target_update=10):
    """DQN训练主循环：在CartPole环境中训练智能体"""
    import gymnasium as gym
    env = gym.make(env_name)

    state_dim = env.observation_space.shape[0]   # 状态维度（CartPole=4）
    action_dim = env.action_space.n              # 动作数量（CartPole=2：左/右）
    agent = DQNAgent(state_dim, action_dim)

    rewards_history = []

    for episode in range(n_episodes):
        state, _ = env.reset()
        total_reward = 0
        done = False

        # 与环境交互，收集经验并实时学习
        while not done:
            action = agent.select_action(state)      # ε-贪心选动作
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            agent.buffer.push(state, action, reward, next_state, float(done))  # 存入经验池
            agent.update()  # 从经验池采样并更新Q网络

            state = next_state
            total_reward += reward

        # 每隔target_update个episode同步目标网络
        if (episode + 1) % target_update == 0:
            agent.update_target()

        rewards_history.append(total_reward)

        # 每50个episode打印训练进度
        if (episode + 1) % 50 == 0:
            avg_reward = np.mean(rewards_history[-50:])  # 切片操作取子序列
            print(f"Episode {episode+1}, Avg Reward: {avg_reward:.1f}, "
                  f"Epsilon: {agent.epsilon:.3f}")

    env.close()
    return agent

# agent = train_dqn()