强化学习-python案例

强化学习是一种机器学习方法，旨在通过与环境的交互来学习最优策略。它的核心概念是智能体（agent）在环境中采取动作，从而获得奖励或惩罚。智能体的目标是最大化长期奖励，通过试错的方式不断改进其决策策略。

在强化学习中，智能体观察当前状态，选择动作，并根据环境反馈（奖励和下一个状态）调整其策略。常见的强化学习算法包括Q-learning、策略梯度方法和深度强化学习等。强化学习广泛应用于游戏、机器人控制、推荐系统等领域。

奖励（Reward）：
$r_t = R(s_t, a_t)$
其中 $r_t$ 是在时间步 $t$ 时，智能体在状态 $s_t$ 下采取动作 $a_t$ 所获得的奖励。
状态价值函数（State Value Function）：
$\mathbb{E} \left[ \sum_{t=0}^{\infty} \gamma^t r_t \mid s_0 = s \right]$
其中 $V (s)$ 是状态 $s$ 的价值， $\gamma$ 是折扣因子 $\leq \gamma < 1$ ），表示未来奖励的重要性。
动作价值函数（Action Value Function）：
$\mathbb{E} \left[ \sum_{t=0}^{\infty} \gamma^t r_t \mid s_0 = s, a_0 = a \right]$
其中 $Q (s, a)$ 是在状态 $s$ 下采取动作 $a$ 的价值。
贝尔曼方程（Bellman Equation）：
- 状态价值函数的贝尔曼方程：
  $\sum_{a} \pi(a \mid s) \sum_{s', r} P(s', r \mid s, a) \left[ r + \gamma V(s') \right]$
- 动作价值函数的贝尔曼方程：
  $\sum_{s', r} P(s', r \mid s, a) \left[ r + \gamma \max_{a'} Q(s', a') \right]$
策略（Policy）：
$\pi(a \mid s) = P(a \mid s)$
其中 $\pi(a \mid s)$ 是在状态 $s$ 下选择动作 $a$ 的概率。

目标函数

策略梯度目标函数：
$J(\theta) = \mathbb{E}_{\tau \sim \pi_\theta} \left[ \sum_{t=0}^{T} r_t \right]$
- 说明： $J(\theta)$ 是关于策略参数 $\theta$ 的目标函数，表示在策略 $\pi_\theta$ 下，执行轨迹 $\tau$ 的预期总奖励。目标是最大化该期望值，通常通过梯度上升方法进行优化。

损失函数

策略损失函数（使用REINFORCE算法）：
$L(\theta) = -\mathbb{E}_{\tau \sim \pi_\theta} \left[ \sum_{t=0}^{T} r_t \log \pi_\theta(a_t \mid s_t) \right]$
- 说明：这个损失函数的目的是最小化负的期望总奖励。通过优化该损失函数，可以最大化目标函数 $J(\theta)$ 。这里的 $\log \pi_\theta(a_t \mid s_t)$ 是对策略的对数概率，表示在状态 $s_t$ 下采取动作 $a_t$ 的可能性。
价值函数损失（对于Q-learning）：
$L(\theta) = \mathbb{E} \left[ \left( r_t + \gamma \max_{a'} Q(s', a'; \theta) - Q(s, a; \theta) \right)^2 \right]$
- 说明：该损失函数用于最小化当前动作价值函数 $\theta)$ 和目标价值 $r_t + \gamma \max_{a'} Q(s', a'; \theta)$ 之间的均方误差。通过最小化该损失，更新网络参数 $\theta$ 以更准确地预测价值。

细节总结

目标函数：用于衡量当前策略的性能，指导优化过程。强化学习的目标是通过更新策略来最大化期望奖励。
损失函数：是优化过程中实际最小化的函数，直接反映模型的学习效果。损失函数的设计直接影响学习的效率和效果。

这些公式是强化学习中策略优化和价值评估的核心，理解它们有助于深入掌握强化学习的理论基础和应用。

代码

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np

# 环境假设
class SimpleEnv:
    def reset(self):
        return np.random.rand(4)  # 随机状态

    def step(self, action):
        next_state = np.random.rand(4)
        reward = np.random.rand()  # 随机奖励
        done = np.random.rand() > 0.9  # 随机结束
        return next_state, reward, done

# 策略网络
class PolicyNetwork(nn.Module):
    def __init__(self):
        super(PolicyNetwork, self).__init__()
        self.fc = nn.Sequential(
            nn.Linear(4, 128),
            nn.ReLU(),
            nn.Linear(128, 2),  # 假设有两个动作
        )

    def forward(self, x):
        return torch.softmax(self.fc(x), dim=-1)

# 计算折扣奖励
def compute_discounted_rewards(rewards, discount_factor=0.99):
    discounted_rewards = []
    cumulative_reward = 0
    for r in reversed(rewards):
        cumulative_reward = r + cumulative_reward * discount_factor
        discounted_rewards.insert(0, cumulative_reward)
    return discounted_rewards

# 训练函数
def train(env, policy_net, optimizer, episodes=1000):
    for episode in range(episodes):
        state = env.reset()
        rewards = []
        log_probs = []
        
        while True:
            state_tensor = torch.FloatTensor(state)
            probs = policy_net(state_tensor)
            action = np.random.choice(len(probs), p=probs.detach().numpy())
            log_prob = torch.log(probs[action])
            
            next_state, reward, done = env.step(action)
            log_probs.append(log_prob)
            rewards.append(reward)
            state = next_state
            
            if done:
                break
        
        # 计算折扣奖励
        discounted_rewards = compute_discounted_rewards(rewards)
        
        # 更新策略
        optimizer.zero_grad()
        loss = -sum(log_prob * reward for log_prob, reward in zip(log_probs, discounted_rewards))
        loss.backward()
        optimizer.step()

        # 输出每个回合的总奖励
        total_reward = sum(rewards)
        print(f"Episode {episode + 1}, Total Reward: {total_reward:.2f}")

# 测试函数
def test(env, policy_net, episodes=10):
    for episode in range(episodes):
        state = env.reset()
        total_reward = 0
        
        while True:
            state_tensor = torch.FloatTensor(state)
            with torch.no_grad():
                probs = policy_net(state_tensor)
            action = torch.argmax(probs).item()
            next_state, reward, done = env.step(action)
            total_reward += reward
            state = next_state
            
            if done:
                break
        
        print(f"Test Episode {episode + 1}, Total Reward: {total_reward:.2f}")

# 主程序
env = SimpleEnv()
policy_net = PolicyNetwork()
optimizer = optim.Adam(policy_net.parameters(), lr=0.01)

train(env, policy_net, optimizer)
test(env, policy_net)

在这里插入图片描述

训练奖励图：显示每个训练回合的总奖励变化，帮助评估模型在训练过程中的学习效果。
测试奖励图：展示在测试回合中模型的总奖励，反映训练后的表现。

代码结构

环境（Environment）
- SimpleEnv 类：模拟一个简单的环境，包含 reset 和 step 方法。
  - reset()：初始化并返回一个随机状态。
  - step(action)：根据所采取的动作返回下一个状态、奖励和是否结束标志。
  - 奖励和结束状态是随机生成的，模拟了一个非常简化的环境。
策略网络（Policy Network）
- PolicyNetwork 类：定义一个神经网络，用于近似策略。
  - 使用全连接层，输入状态维度为 4（环境状态的维度），输出动作概率的维度为 2（假设有两个可能的动作）。
  - forward 方法通过 softmax 函数输出每个动作的概率。
折扣奖励计算
- compute_discounted_rewards(rewards, discount_factor=0.99)：计算每个时间步的折扣奖励。
  - 从后往前遍历奖励列表，使用折扣因子更新累计奖励，生成折扣奖励列表。
训练函数（Training Function）
- train(env, policy_net, optimizer, episodes=1000)：进行训练的主函数。
  - 循环执行指定的回合数：
    - 重置环境，初始化奖励和日志概率列表。
    - 在回合中循环，使用当前状态选择动作并记录日志概率和奖励。
    - 计算并更新策略网络的损失，使用反向传播更新参数。
    - 每个回合结束后打印总奖励，帮助监控训练进度。
测试函数（Testing Function）
- test(env, policy_net, episodes=10)：用于评估训练后模型表现的函数。
  - 重置环境并执行多个测试回合，选择最大概率的动作。
  - 累计并打印每个测试回合的总奖励，评估训练的效果。
主程序
- 创建环境和策略网络实例，定义优化器（Adam）。
- 调用训练函数进行训练，然后调用测试函数进行评估。