强化学习入门¶

概述¶

强化学习（Reinforcement Learning，RL）是机器学习的重要分支，通过与环境交互学习最优策略，广泛应用于游戏、机器人、自动驾驶等领域。

核心概念¶

# 强化学习要素

智能体（Agent）：决策者
环境（Environment）：智能体所处的世界
状态（State）：当前环境的情况
动作（Action）：智能体可以执行的操作
奖励（Reward）：环境反馈的信号

# 目标

最大化累积奖励（Return）

状态-动作价值函数¶

# Q(s, a) = 在状态s下执行动作a，之后遵循最优策略的期望回报

# 贝尔曼方程
# Q(s, a) = R(s, a) + γ × max Q(s', a')
#                           a'
# 其中：
# - R(s, a): 即时奖励
# - γ: 折扣因子（0-1）
# - s': 下一个状态

Q-Learning¶

算法原理¶

import numpy as np
import random

class QLearning:
    def __init__(self, n_states, n_actions, lr=0.1, gamma=0.99, epsilon=0.1):
        self.q_table = np.zeros((n_states, n_actions))
        self.lr = lr          # 学习率
        self.gamma = gamma    # 折扣因子
        self.epsilon = epsilon  # 探索率

    def choose_action(self, state):
        if random.random() < self.epsilon:
            return random.randint(0, self.q_table.shape[1] - 1)
        return np.argmax(self.q_table[state])

    def update(self, state, action, reward, next_state):
        # Q-Learning更新规则
        best_next_action = np.argmax(self.q_table[next_state])
        td_target = reward + self.gamma * self.q_table[next_state, best_next_action]
        td_error = td_target - self.q_table[state, action]
        self.q_table[state, action] += self.lr * td_error

    def train(self, env, episodes=1000):
        rewards_per_episode = []

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

            while not done:
                action = self.choose_action(state)
                next_state, reward, done, _ = env.step(action)
                self.update(state, action, reward, next_state)
                state = next_state
                total_reward += reward

            rewards_per_episode.append(total_reward)

        return rewards_per_episode

DQN（深度Q网络）¶

神经网络近似¶

import torch
import torch.nn as nn

class DQN(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 128),
            nn.ReLU(),
            nn.Linear(128, action_dim)
        )

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

经验回放¶

from collections import deque
import random

class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.buffer = deque(maxlen=capacity)

    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (
            torch.FloatTensor(states),
            torch.LongTensor(actions),
            torch.FloatTensor(rewards),
            torch.FloatTensor(next_states),
            torch.FloatTensor(dones)
        )

    def __len__(self):
        return len(self.buffer)

DQN完整实现¶

class DQNAgent:
    def __init__(self, state_dim, action_dim):
        self.action_dim = action_dim
        self.epsilon = 1.0
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995

        # 网络
        self.q_network = DQN(state_dim, action_dim)
        self.target_network = DQN(state_dim, action_dim)
        self.target_network.load_state_dict(self.q_network.state_dict())

        # 优化器
        self.optimizer = torch.optim.Adam(self.q_network.parameters(), lr=0.001)
        self.buffer = ReplayBuffer()

    def choose_action(self, state):
        if random.random() < self.epsilon:
            return random.randint(0, self.action_dim - 1)

        with torch.no_grad():
            state = torch.FloatTensor(state).unsqueeze(0)
            q_values = self.q_network(state)
            return q_values.argmax().item()

    def update(self, batch_size=32):
        if len(self.buffer) < batch_size:
            return

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

        # 当前Q值
        q_values = self.q_network(states).gather(1, actions.unsqueeze(1))

        # 目标Q值
        with torch.no_grad():
            next_q_values = self.target_network(next_states).max(1)[0]
            target_q_values = rewards + (1 - dones) * 0.99 * next_q_values

        # 损失
        loss = nn.MSELoss()(q_values.squeeze(), target_q_values)

        # 更新
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        # 衰减探索率
        self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)

    def update_target(self):
        self.target_network.load_state_dict(self.q_network.state_dict())

策略梯度¶

REINFORCE算法¶

class PolicyNetwork(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU(),
            nn.Linear(128, action_dim),
            nn.Softmax(dim=-1)
        )

    def forward(self, state):
        return self.network(state)

class REINFORCEAgent:
    def __init__(self, state_dim, action_dim):
        self.policy = PolicyNetwork(state_dim, action_dim)
        self.optimizer = torch.optim.Adam(self.policy.parameters(), lr=0.001)
        self.gamma = 0.99

    def choose_action(self, state):
        state = torch.FloatTensor(state).unsqueeze(0)
        probs = self.policy(state)
        dist = torch.distributions.Categorical(probs)
        action = dist.sample()
        return action.item(), dist.log_prob(action)

    def update(self, log_probs, rewards):
        # 计算折扣回报
        returns = []
        G = 0
        for r in reversed(rewards):
            G = r + self.gamma * G
            returns.insert(0, G)
        returns = torch.FloatTensor(returns)

        # 标准化
        returns = (returns - returns.mean()) / (returns.std() + 1e-8)

        # 策略梯度
        loss = 0
        for log_prob, G in zip(log_probs, returns):
            loss -= log_prob * G

        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

Actor-Critic¶

A2C（优势演员-评论家）¶

class ActorCritic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.shared = nn.Sequential(
            nn.Linear(state_dim, 128),
            nn.ReLU()
        )
        self.actor = nn.Linear(128, action_dim)
        self.critic = nn.Linear(128, 1)

    def forward(self, state):
        x = self.shared(state)
        action_probs = torch.softmax(self.actor(x), dim=-1)
        state_value = self.critic(x)
        return action_probs, state_value

强化学习应用¶

# 游戏AI

AlphaGo/AlphaZero
- 围棋、国际象棋
- 搜索+学习

Atari游戏
- DQN系列
- 从像素学习

# 机器人控制

- 机械臂操作
- 步态学习
- 导航

# 自动驾驶

- 决策规划
- 路径规划

# 推荐系统

- 在线学习
- 多样性优化

# 资源调度

- 服务器负载均衡
- 网络流量优化

学习资源¶

# 教材

- 《Reinforcement Learning: An Introduction》- Sutton & Barto
- 《Deep Reinforcement Learning Hands-On》

# 开源框架

- OpenAI Gym：环境库
- Stable-Baselines3：算法库
- Ray RLlib：分布式RL

下一步学习¶

深度学习教程 - 理解神经网络
时间序列预测 - 时序决策
AI Agent智能体 - RL在Agent中的应用