强化学习-python案例

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

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

1. 奖励（Reward）：
   r t = R ( s t , a t ) r_t = R(s_t, a_t) rt​=R(st​,at​)
   其中 r t r_t rt​ 是在时间步 t t t 时，智能体在状态 s t s_t st​ 下采取动作 a t a_t at​ 所获得的奖励。

2. 状态价值函数（State Value Function）：
   V ( s ) = E [ ∑ t = 0 ∞ γ t r t ∣ s 0 = s ] V(s) = \mathbb{E} \left[ \sum_{t=0}^{\infty} \gamma^t r_t \mid s_0 = s \right] V(s)=E[t=0∑∞​γtrt​∣s0​=s]
   其中 V ( s ) V(s) V(s) 是状态 s s s 的价值， γ \gamma γ 是折扣因子 ( 0 ≤ γ < 1 ( 0 \leq \gamma < 1 (0≤γ<1），表示未来奖励的重要性。

3. 动作价值函数（Action Value Function）：
   Q ( s , a ) = E [ ∑ t = 0 ∞ γ t r t ∣ s 0 = s , a 0 = a ] Q(s, a) = \mathbb{E} \left[ \sum_{t=0}^{\infty} \gamma^t r_t \mid s_0 = s, a_0 = a \right] Q(s,a)=E[t=0∑∞​γtrt​∣s0​=s,a0​=a]
   其中 Q ( s , a ) Q(s, a) Q(s,a) 是在状态 s s s 下采取动作 a a a 的价值。

4. 贝尔曼方程（Bellman Equation）：

   • 状态价值函数的贝尔曼方程：
     V ( s ) = ∑ a π ( a ∣ s ) ∑ s ′ , r P ( s ′ , r ∣ s , a ) [ r + γ V ( s ′ ) ] V(s) = \sum_{a} \pi(a \mid s) \sum_{s', r} P(s', r \mid s, a) \left[ r + \gamma V(s') \right] V(s)=a∑​π(a∣s)s′,r∑​P(s′,r∣s,a)[r+γV(s′)]
   • 动作价值函数的贝尔曼方程：
     Q ( s , a ) = ∑ s ′ , r P ( s ′ , r ∣ s , a ) [ r + γ max ⁡ a ′ Q ( s ′ , a ′ ) ] Q(s, a) = \sum_{s', r} P(s', r \mid s, a) \left[ r + \gamma \max_{a'} Q(s', a') \right] Q(s,a)=s′,r∑​P(s′,r∣s,a)[r+γa′max​Q(s′,a′)]

5. 策略（Policy）：
   π ( a ∣ s ) = P ( a ∣ s ) \pi(a \mid s) = P(a \mid s) π(a∣s)=P(a∣s)
   其中 π ( a ∣ s ) \pi(a \mid s) π(a∣s) 是在状态 s s s 下选择动作 a a a 的概率。

目标函数

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

损失函数

2. 策略损失函数（使用REINFORCE算法）：
   L ( θ ) = − E τ ∼ π θ [ ∑ t = 0 T r t log ⁡ π θ ( a t ∣ s t ) ] L(\theta) = -\mathbb{E}_{\tau \sim \pi_\theta} \left[ \sum_{t=0}^{T} r_t \log \pi_\theta(a_t \mid s_t) \right] L(θ)=−Eτ∼πθ​​[t=0∑T​rt​logπθ​(at​∣st​)]

   • 说明：这个损失函数的目的是最小化负的期望总奖励。通过优化该损失函数，可以最大化目标函数 J ( θ ) J(\theta) J(θ)。这里的 log ⁡ π θ ( a t ∣ s t ) \log \pi_\theta(a_t \mid s_t) logπθ​(at​∣st​) 是对策略的对数概率，表示在状态 s t s_t st​ 下采取动作 a t a_t at​ 的可能性。

3. 价值函数损失（对于Q-learning）：
   L ( θ ) = E [ ( r t + γ max ⁡ a ′ Q ( s ′ , a ′ ; θ ) − Q ( s , a ; θ ) ) 2 ] L(\theta) = \mathbb{E} \left[ \left( r_t + \gamma \max_{a'} Q(s', a'; \theta) - Q(s, a; \theta) \right)^2 \right] L(θ)=E[(rt​+γa′max​Q(s′,a′;θ)−Q(s,a;θ))2]

   • 说明：该损失函数用于最小化当前动作价值函数 Q ( s , a ; θ ) Q(s, a; \theta) Q(s,a;θ) 和目标价值 r t + γ max ⁡ a ′ Q ( s ′ , a ′ ; θ ) r_t + \gamma \max_{a'} Q(s', a'; \theta) rt​+γmaxa′​Q(s′,a′;θ) 之间的均方误差。通过最小化该损失，更新网络参数 θ \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)

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

代码结构

1. 环境（Environment）

   • SimpleEnv 类：模拟一个简单的环境，包含 reset 和 step 方法。
     • reset()：初始化并返回一个随机状态。
     • step(action)：根据所采取的动作返回下一个状态、奖励和是否结束标志。
     • 奖励和结束状态是随机生成的，模拟了一个非常简化的环境。

2. 策略网络（Policy Network）

   • PolicyNetwork 类：定义一个神经网络，用于近似策略。
     • 使用全连接层，输入状态维度为 4（环境状态的维度），输出动作概率的维度为 2（假设有两个可能的动作）。
     • forward 方法通过 softmax 函数输出每个动作的概率。

3. 折扣奖励计算

   • compute_discounted_rewards(rewards, discount_factor=0.99)：计算每个时间步的折扣奖励。
     • 从后往前遍历奖励列表，使用折扣因子更新累计奖励，生成折扣奖励列表。

4. 训练函数（Training Function）

   • train(env, policy_net, optimizer, episodes=1000)：进行训练的主函数。
     • 循环执行指定的回合数：
       • 重置环境，初始化奖励和日志概率列表。
       • 在回合中循环，使用当前状态选择动作并记录日志概率和奖励。
       • 计算并更新策略网络的损失，使用反向传播更新参数。
       • 每个回合结束后打印总奖励，帮助监控训练进度。

5. 测试函数（Testing Function）

   • test(env, policy_net, episodes=10)：用于评估训练后模型表现的函数。
     • 重置环境并执行多个测试回合，选择最大概率的动作。
     • 累计并打印每个测试回合的总奖励，评估训练的效果。

6. 主程序

   • 创建环境和策略网络实例，定义优化器（Adam）。
   • 调用训练函数进行训练，然后调用测试函数进行评估。

整体逻辑

1. 环境设置：定义了一个非常简单的环境，主要用于演示如何应用策略梯度方法。实际应用中，可以替换为更复杂的环境，比如OpenAI的Gym库中的环境。

2. 策略学习：使用神经网络近似策略，通过与环境的交互收集状态、动作、奖励，并更新网络参数，以优化策略。

3. 输出和评估：通过在训练过程中的总奖励输出和测试过程中的评估，可以观察到模型的学习进展。

小结

这段代码是一个简单的强化学习示例，展示了如何使用策略梯度方法和PyTorch进行训练和测试。虽然环境和任务是简化的，但它提供了一个良好的基础，便于理解强化学习的核心概念和实现。