目录

1. 概述
2. 基本概念
3. 马尔可夫决策过程
4. 动态规划
5. 蒙特卡洛方法
6. 时序差分学习
7. Q-Learning与SARSA
8. 深度Q网络(DQN)
9. 策略梯度方法
10. Actor-Critic方法
11. PPO算法
12. 模型基础方法
13. 多智能体强化学习
14. 逆强化学习
15. 强化学习应用
16. 完整代码实现
17. 参考资料

---

1. 概述

1.1 什么是强化学习

强化学习 (Reinforcement Learning, RL) 是机器学习的一个分支，研究智能体如何在环境中通过试错来学习最优策略，以最大化累积奖励。

1.2 核心要素

要素	说明
智能体 (Agent)	学习和决策的主体
环境 (Environment)	智能体交互的外部世界
状态 (State)	环境的描述
动作 (Action)	智能体可以执行的操作
奖励 (Reward)	环境对动作的反馈
策略 (Policy)	状态到动作的映射
价值函数 (Value Function)	状态或动作的长期价值

1.3 与其他机器学习的区别

监督学习: 从标注数据学习输入到输出的映射
无监督学习: 发现数据中的模式
强化学习: 从交互中学习，延迟奖励，探索-利用权衡

1.4 发展历程

1950s: 动态规划 (Bellman)
1989: Q-Learning (Watkins)
1992: TD-Gammon (Tesauro)
2013: DQN (DeepMind)
2016: AlphaGo
2017: PPO
2019: AlphaStar, OpenAI Five
2022-至今: RLHF用于大语言模型

---

2. 基本概念

2.1 智能体与环境交互

┌─────────┐     动作a_t      ┌─────────┐
│         │ ───────────────→ │         │
│  智能体  │                  │   环境   │
│         │ ←─────────────── │         │
└─────────┘  状态s_t, 奖励r_t └─────────┘

循环:
1. 智能体观察状态 s_t
2. 智能体选择动作 a_t
3. 环境返回新状态 s_{t+1} 和奖励 r_t
4. 重复

2.2 策略

确定性策略：
$$a=\pi (s)$$

随机策略：
$$a\sim \pi (\cdot |s)$$

2.3 价值函数

状态价值函数：
$$V^{\pi }(s)=E_{\pi }\left[\sum _{t=0}^{\infty }{\gamma }^t{r}_t|s_0=s\right]$$

动作价值函数：
$$Q^{\pi }(s,a)=E_{\pi }\left[\sum _{t=0}^{\infty }{\gamma }^t{r}_t|s_0=s,a_0=a\right]$$

优势函数：
$$A^{\pi }(s,a)=Q^{\pi }(s,a)-V^{\pi }(s)$$

2.4 折扣因子

$$G_t=\sum _{k=0}^{\infty }{\gamma }^k{r}_{t+k}$$

• $\gamma =0$: 只关注即时奖励
• $\gamma =1$: 平等对待所有未来奖励
• 通常 $\gamma \in [0.9,0.99]$

---

3. 马尔可夫决策过程

3.1 定义

MDP是一个五元组 $(S,A,P,R,\gamma )$：

符号	含义
$S$	状态集合
$A$	动作集合
$P(s’	s, a)$
$R(s,a)$	奖励函数
$\gamma$	折扣因子

3.2 马尔可夫性质

$$P(s_{t+1}|s_t,a_t,s_{t-1},a_{t-1},...)=P(s_{t+1}|s_t,a_t)$$

未来只依赖于当前状态，与历史无关。

3.3 回合制与持续任务

回合制任务：有终止状态（如游戏结束）
持续任务：无终止状态（如机器人控制）

3.4 贝尔曼方程

贝尔曼期望方程：
$$V^{\pi }(s)=\sum _a\pi (a|s)\sum _{s^{\prime }}P(s^{\prime }|s,a)[R(s,a)+\gamma V^{\pi }(s^{\prime })]$$

$$Q^{\pi }(s,a)=\sum _{s^{\prime }}P(s^{\prime }|s,a)[R(s,a)+\gamma \sum _{a^{\prime }}\pi (a^{\prime }|s^{\prime })Q^{\pi }(s^{\prime },a^{\prime })]$$

贝尔曼最优方程：
$$V^{\ast }(s)=\underset{a}{\max }\sum _{s^{\prime }}P(s^{\prime }|s,a)[R(s,a)+\gamma V^{\ast }(s^{\prime })]$$

$$Q^{\ast }(s,a)=\sum _{s^{\prime }}P(s^{\prime }|s,a)[R(s,a)+\gamma \underset{a^{\prime }}{\max }{Q}^{\ast }(s^{\prime },a^{\prime })]$$

---

4. 动态规划

4.1 策略评估

给定策略 $\pi$，计算其价值函数：

def policy_evaluation(env, policy, gamma=0.99, theta=1e-8):
    """
    策略评估：计算给定策略的价值函数

    Args:
        env: 环境
        policy: 策略 π(a|s)
        gamma: 折扣因子
        theta: 收敛阈值

    Returns:
        V: 状态价值函数
    """
    V = {s: 0.0 for s in env.states}

    while True:
        delta = 0
        for s in env.states:
            v = V[s]

            # 计算新价值
            new_v = 0
            for a in env.actions:
                for prob, next_state, reward in env.transitions(s, a):
                    new_v += policy[s][a] * prob * (reward + gamma * V[next_state])

            V[s] = new_v
            delta = max(delta, abs(v - V[s]))

        if delta < theta:
            break

    return V

4.2 策略改进

def policy_improvement(env, V, gamma=0.99):
    """
    策略改进：基于价值函数改进策略

    Args:
        env: 环境
        V: 状态价值函数
        gamma: 折扣因子

    Returns:
        policy: 改进后的策略
    """
    policy = {}

    for s in env.states:
        # 计算每个动作的价值
        action_values = {}
        for a in env.actions:
            value = 0
            for prob, next_state, reward in env.transitions(s, a):
                value += prob * (reward + gamma * V[next_state])
            action_values[a] = value

        # 选择最优动作
        best_action = max(action_values, key=action_values.get)
        policy[s] = {a: 1.0 if a == best_action else 0.0 for a in env.actions}

    return policy

4.3 策略迭代

def policy_iteration(env, gamma=0.99):
    """
    策略迭代：交替进行策略评估和策略改进

    Returns:
        policy: 最优策略
        V: 最优价值函数
    """
    # 初始化随机策略
    policy = {s: {a: 1/len(env.actions) for a in env.actions} for s in env.states}

    while True:
        # 策略评估
        V = policy_evaluation(env, policy, gamma)

        # 策略改进
        new_policy = policy_improvement(env, V, gamma)

        # 检查是否收敛
        if new_policy == policy:
            break

        policy = new_policy

    return policy, V

4.4 价值迭代

def value_iteration(env, gamma=0.99, theta=1e-8):
    """
    价值迭代：直接迭代贝尔曼最优方程

    Returns:
        policy: 最优策略
        V: 最优价值函数
    """
    V = {s: 0.0 for s in env.states}

    while True:
        delta = 0
        for s in env.states:
            v = V[s]

            # 计算最优价值
            action_values = []
            for a in env.actions:
                value = 0
                for prob, next_state, reward in env.transitions(s, a):
                    value += prob * (reward + gamma * V[next_state])
                action_values.append(value)

            V[s] = max(action_values)
            delta = max(delta, abs(v - V[s]))

        if delta < theta:
            break

    # 提取最优策略
    policy = {}
    for s in env.states:
        action_values = {}
        for a in env.actions:
            value = 0
            for prob, next_state, reward in env.transitions(s, a):
                value += prob * (reward + gamma * V[next_state])
            action_values[a] = value

        best_action = max(action_values, key=action_values.get)
        policy[s] = {a: 1.0 if a == best_action else 0.0 for a in env.actions}

    return policy, V

---

5. 蒙特卡洛方法

5.1 核心思想

通过采样轨迹来估计价值函数，不需要知道环境模型。

5.2 首次访问MC

def first_visit_mc(env, policy, num_episodes=1000, gamma=0.99):
    """
    首次访问蒙特卡洛方法

    Args:
        env: 环境
        policy: 策略
        num_episodes: 采样episode数量
        gamma: 折扣因子

    Returns:
        V: 状态价值函数估计
    """
    V = {}
    returns = {s: [] for s in env.states}

    for _ in range(num_episodes):
        # 采样轨迹
        trajectory = []
        state = env.reset()

        done = False
        while not done:
            action = policy(state)
            next_state, reward, done, _ = env.step(action)
            trajectory.append((state, action, reward))
            state = next_state

        # 计算累积回报
        G = 0
        visited_states = set()

        for state, action, reward in reversed(trajectory):
            G = reward + gamma * G

            # 首次访问
            if state not in visited_states:
                visited_states.add(state)
                returns[state].append(G)
                V[state] = np.mean(returns[state])

    return V

5.3 每次访问MC

def every_visit_mc(env, policy, num_episodes=1000, gamma=0.99):
    """
    每次访问蒙特卡洛方法
    """
    V = {}
    returns = {s: [] for s in env.states}

    for _ in range(num_episodes):
        trajectory = []
        state = env.reset()

        done = False
        while not done:
            action = policy(state)
            next_state, reward, done, _ = env.step(action)
            trajectory.append((state, action, reward))
            state = next_state

        G = 0
        for state, action, reward in reversed(trajectory):
            G = reward + gamma * G
            returns[state].append(G)
            V[state] = np.mean(returns[state])

    return V

5.4 MC控制

def mc_control_epsilon_greedy(env, num_episodes=10000, gamma=0.99, epsilon=0.1):
    """
    蒙特卡洛控制（ε-贪婪策略）

    Returns:
        Q: 动作价值函数
        policy: 最优策略
    """
    Q = {}
    returns = {}
    policy = {}

    for episode in range(num_episodes):
        # ε-贪婪策略
        def epsilon_greedy(state):
            if state not in policy:
                policy[state] = {a: 1/len(env.actions) for a in env.actions}

            if np.random.random() < epsilon:
                return np.random.choice(env.actions)
            else:
                return max(Q.get(state, {a: 0 for a in env.actions}),
                          key=Q.get(state, {a: 0 for a in env.actions}).get)

        # 采样
        trajectory = []
        state = env.reset()

        done = False
        while not done:
            action = epsilon_greedy(state)
            next_state, reward, done, _ = env.step(action)
            trajectory.append((state, action, reward))
            state = next_state

        # 更新Q
        G = 0
        visited = set()

        for state, action, reward in reversed(trajectory):
            G = reward + gamma * G

            if (state, action) not in visited:
                visited.add((state, action))

                if (state, action) not in returns:
                    returns[(state, action)] = []
                returns[(state, action)].append(G)

                if state not in Q:
                    Q[state] = {}
                Q[state][action] = np.mean(returns[(state, action)])

                # 改进策略
                best_action = max(Q[state], key=Q[state].get)
                policy[state] = {
                    a: 1 - epsilon + epsilon/len(env.actions) if a == best_action
                    else epsilon/len(env.actions)
                    for a in env.actions
                }

    return Q, policy

---

6. 时序差分学习

6.1 核心思想

结合MC和DP的优点：

• 像MC一样从经验学习
• 像DP一样使用自举（bootstrapping）

6.2 TD(0)预测

def td0_prediction(env, policy, num_episodes=1000, alpha=0.1, gamma=0.99):
    """
    TD(0)预测

    更新规则: V(s) ← V(s) + α[r + γV(s') - V(s)]

    Args:
        env: 环境
        policy: 策略
        num_episodes: episode数量
        alpha: 学习率
        gamma: 折扣因子

    Returns:
        V: 状态价值函数
    """
    V = {s: 0.0 for s in env.states}

    for _ in range(num_episodes):
        state = env.reset()

        done = False
        while not done:
            action = policy(state)
            next_state, reward, done, _ = env.step(action)

            # TD更新
            if done:
                V[state] += alpha * (reward - V[state])
            else:
                V[state] += alpha * (reward + gamma * V[next_state] - V[state])

            state = next_state

    return V

6.3 TD(λ)与资格迹

def td_lambda_prediction(env, policy, num_episodes=1000, alpha=0.1, gamma=0.99, lambda_=0.9):
    """
    TD(λ)预测（使用资格迹）

    Args:
        lambda_: 衰减参数
            λ=0: 等价于TD(0)
            λ=1: 等价于MC
    """
    V = {s: 0.0 for s in env.states}
    eligibility = {s: 0.0 for s in env.states}

    for _ in range(num_episodes):
        state = env.reset()

        # 重置资格迹
        for s in eligibility:
            eligibility[s] = 0.0

        done = False
        while not done:
            action = policy(state)
            next_state, reward, done, _ = env.step(action)

            # TD误差
            if done:
                td_error = reward - V[state]
            else:
                td_error = reward + gamma * V[next_state] - V[state]

            # 更新资格迹
            eligibility[state] += 1

            # 更新所有状态
            for s in env.states:
                V[s] += alpha * td_error * eligibility[s]
                eligibility[s] *= gamma * lambda_

            state = next_state

    return V

---

7. Q-Learning与SARSA

7.1 SARSA

def sarsa(env, num_episodes=10000, alpha=0.1, gamma=0.99, epsilon=0.1):
    """
    SARSA: 在线策略TD控制

    更新规则: Q(s,a) ← Q(s,a) + α[r + γQ(s',a') - Q(s,a)]

    Args:
        env: 环境
        num_episodes: episode数量
        alpha: 学习率
        gamma: 折扣因子
        epsilon: 探索率

    Returns:
        Q: 动作价值函数
    """
    Q = {s: {a: 0.0 for a in env.actions} for s in env.states}

    def epsilon_greedy(state):
        if np.random.random() < epsilon:
            return np.random.choice(env.actions)
        else:
            return max(Q[state], key=Q[state].get)

    for _ in range(num_episodes):
        state = env.reset()
        action = epsilon_greedy(state)

        done = False
        while not done:
            next_state, reward, done, _ = env.step(action)
            next_action = epsilon_greedy(next_state)

            # SARSA更新
            if done:
                Q[state][action] += alpha * (reward - Q[state][action])
            else:
                Q[state][action] += alpha * (
                    reward + gamma * Q[next_state][next_action] - Q[state][action]
                )

            state = next_state
            action = next_action

    return Q

7.2 Q-Learning

def q_learning(env, num_episodes=10000, alpha=0.1, gamma=0.99, epsilon=0.1):
    """
    Q-Learning: 离线策略TD控制

    更新规则: Q(s,a) ← Q(s,a) + α[r + γ max_a' Q(s',a') - Q(s,a)]

    Args:
        env: 环境
        num_episodes: episode数量
        alpha: 学习率
        gamma: 折扣因子
        epsilon: 探索率

    Returns:
        Q: 动作价值函数
    """
    Q = {s: {a: 0.0 for a in env.actions} for s in env.states}

    def epsilon_greedy(state):
        if np.random.random() < epsilon:
            return np.random.choice(env.actions)
        else:
            return max(Q[state], key=Q[state].get)

    for _ in range(num_episodes):
        state = env.reset()

        done = False
        while not done:
            action = epsilon_greedy(state)
            next_state, reward, done, _ = env.step(action)

            # Q-Learning更新
            if done:
                Q[state][action] += alpha * (reward - Q[state][action])
            else:
                best_next = max(Q[next_state].values())
                Q[state][action] += alpha * (
                    reward + gamma * best_next - Q[state][action]
                )

            state = next_state

    return Q

7.3 SARSA vs Q-Learning

特性	SARSA	Q-Learning
策略类型	在线策略	离线策略
更新目标	Q(s’,a’)	max Q(s’,a’)
探索影响	考虑探索	不考虑探索
收敛性	更保守	更激进
安全性	更安全	可能风险大

---

8. 深度Q网络(DQN)

8.1 核心思想

用神经网络近似Q函数：
$$Q(s,a;\theta )\approx Q^{\ast }(s,a)$$

8.2 DQN架构

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):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        return self.fc3(x)

8.3 经验回放

class ReplayBuffer:
    def __init__(self, capacity=100000):
        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(np.array(states)),
            torch.LongTensor(actions),
            torch.FloatTensor(rewards),
            torch.FloatTensor(np.array(next_states)),
            torch.FloatTensor(dones)
        )

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

8.4 完整DQN实现

class DQNAgent:
    def __init__(
        self,
        state_dim,
        action_dim,
        hidden_dim=128,
        lr=1e-3,
        gamma=0.99,
        epsilon_start=1.0,
        epsilon_end=0.01,
        epsilon_decay=0.995,
        buffer_size=100000,
        batch_size=64,
        target_update=10
    ):
        self.action_dim = action_dim
        self.gamma = gamma
        self.epsilon = epsilon_start
        self.epsilon_end = epsilon_end
        self.epsilon_decay = epsilon_decay
        self.batch_size = batch_size
        self.target_update = target_update

        # 网络
        self.q_net = DQN(state_dim, action_dim, hidden_dim)
        self.target_net = DQN(state_dim, action_dim, hidden_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(buffer_size)

        # 计数器
        self.steps = 0

    def select_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_net(state)
            return q_values.argmax().item()

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

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

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

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

        # 损失
        loss = F.mse_loss(q_values, target_q_values)

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

        # 更新epsilon
        self.epsilon = max(self.epsilon_end, self.epsilon * self.epsilon_decay)

        # 更新目标网络
        self.steps += 1
        if self.steps % self.target_update == 0:
            self.target_net.load_state_dict(self.q_net.state_dict())

        return loss.item()

    def train(self, env, num_episodes=1000):
        rewards_history = []

        for episode in range(num_episodes):
            state = env.reset()
            total_reward = 0

            done = False
            while not done:
                action = self.select_action(state)
                next_state, reward, done, _ = env.step(action)

                self.buffer.push(state, action, reward, next_state, done)
                self.update()

                state = next_state
                total_reward += reward

            rewards_history.append(total_reward)

            if episode % 100 == 0:
                avg_reward = np.mean(rewards_history[-100:])
                print(f"Episode {episode}, Avg Reward: {avg_reward:.2f}, Epsilon: {self.epsilon:.3f}")

        return rewards_history

8.5 Double DQN

def update_double_dqn(self):
    """Double DQN: 减少过估计"""
    states, actions, rewards, next_states, dones = self.buffer.sample(self.batch_size)

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

    # Double DQN: 用当前网络选择动作，用目标网络评估
    with torch.no_grad():
        next_actions = self.q_net(next_states).argmax(1)
        next_q_values = self.target_net(next_states).gather(1, next_actions.unsqueeze(1)).squeeze(1)
        target_q_values = rewards + self.gamma * next_q_values * (1 - dones)

    loss = F.mse_loss(q_values, target_q_values)

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

8.6 Dueling DQN

class DuelingDQN(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()

        # 共享层
        self.fc1 = nn.Linear(state_dim, hidden_dim)

        # 价值流
        self.value_fc = nn.Linear(hidden_dim, hidden_dim)
        self.value = nn.Linear(hidden_dim, 1)

        # 优势流
        self.advantage_fc = nn.Linear(hidden_dim, hidden_dim)
        self.advantage = nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))

        # 价值
        value = F.relu(self.value_fc(x))
        value = self.value(value)

        # 优势
        advantage = F.relu(self.advantage_fc(x))
        advantage = self.advantage(advantage)

        # 组合
        q_values = value + advantage - advantage.mean(dim=1, keepdim=True)

        return q_values

---

9. 策略梯度方法

9.1 核心思想

直接优化策略：
$${\nabla }_{\theta }J(\theta )=E_{{\pi }_{\theta }}\left[{\nabla }_{\theta }\log {\pi }_{\theta }(a|s)\cdot Q^{{\pi }_{\theta }}(s,a)\right]$$

9.2 REINFORCE算法

class REINFORCE:
    def __init__(self, state_dim, action_dim, hidden_dim=128, lr=1e-3, gamma=0.99):
        self.gamma = gamma

        # 策略网络
        self.policy = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )

        self.optimizer = torch.optim.Adam(self.policy.parameters(), lr=lr)

    def select_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()

    def train(self, env, num_episodes=1000):
        rewards_history = []

        for episode in range(num_episodes):
            state = env.reset()
            log_probs = []
            rewards = []

            done = False
            while not done:
                action, log_prob = self.select_action(state)
                next_state, reward, done, _ = env.step(action)

                log_probs.append(log_prob)
                rewards.append(reward)

                state = next_state

            self.update(log_probs, rewards)

            total_reward = sum(rewards)
            rewards_history.append(total_reward)

            if episode % 100 == 0:
                avg_reward = np.mean(rewards_history[-100:])
                print(f"Episode {episode}, Avg Reward: {avg_reward:.2f}")

        return rewards_history

9.3 带基线的REINFORCE

class REINFORCEWithBaseline:
    def __init__(self, state_dim, action_dim, hidden_dim=128, lr=1e-3, gamma=0.99):
        self.gamma = gamma

        # 策略网络
        self.policy = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )

        # 价值网络（基线）
        self.value = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

        self.policy_optimizer = torch.optim.Adam(self.policy.parameters(), lr=lr)
        self.value_optimizer = torch.optim.Adam(self.value.parameters(), lr=lr)

    def select_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, states, log_probs, rewards):
        # 计算累积回报
        returns = []
        G = 0
        for r in reversed(rewards):
            G = r + self.gamma * G
            returns.insert(0, G)
        returns = torch.FloatTensor(returns)

        states = torch.FloatTensor(np.array(states))

        # 计算基线
        values = self.value(states).squeeze()
        advantages = returns - values.detach()

        # 策略损失
        policy_loss = 0
        for log_prob, advantage in zip(log_probs, advantages):
            policy_loss -= log_prob * advantage

        # 价值损失
        value_loss = F.mse_loss(values, returns)

        # 更新策略
        self.policy_optimizer.zero_grad()
        policy_loss.backward()
        self.policy_optimizer.step()

        # 更新价值网络
        self.value_optimizer.zero_grad()
        value_loss.backward()
        self.value_optimizer.step()

---

10. Actor-Critic方法

10.1 核心思想

同时学习：

• Actor（演员）: 策略网络 ${\pi }_{\theta }(a|s)$
• Critic（评论家）: 价值网络 $V_{\varphi }(s)$

10.2 A2C (Advantage Actor-Critic)

class ActorCritic(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()

        # 共享层
        self.shared = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU()
        )

        # Actor头
        self.actor = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )

        # Critic头
        self.critic = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

    def forward(self, x):
        shared = self.shared(x)
        action_probs = self.actor(shared)
        value = self.critic(shared)
        return action_probs, value


class A2CAgent:
    def __init__(self, state_dim, action_dim, lr=1e-3, gamma=0.99, entropy_coeff=0.01):
        self.gamma = gamma
        self.entropy_coeff = entropy_coeff

        self.model = ActorCritic(state_dim, action_dim)
        self.optimizer = torch.optim.Adam(self.model.parameters(), lr=lr)

    def select_action(self, state):
        state = torch.FloatTensor(state).unsqueeze(0)
        action_probs, value = self.model(state)

        dist = torch.distributions.Categorical(action_probs)
        action = dist.sample()

        return action.item(), dist.log_prob(action), value

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

        log_probs = torch.stack(log_probs)
        values = torch.stack(values).squeeze()

        # 优势
        advantages = returns - values.detach()

        # Actor损失
        actor_loss = -(log_probs * advantages).mean()

        # Critic损失
        critic_loss = F.mse_loss(values, returns)

        # 熵奖励
        entropy = -(torch.exp(log_probs) * log_probs).mean()

        # 总损失
        loss = actor_loss + 0.5 * critic_loss - self.entropy_coeff * entropy

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

        return loss.item()

10.3 A3C (异步优势Actor-Critic)

import torch.multiprocessing as mp

def worker(worker_id, global_model, optimizer, env_name, gamma=0.99):
    """A3C工作进程"""
    env = gym.make(env_name)
    local_model = ActorCritic(env.observation_space.shape[0], env.action_space.n)

    while True:
        # 同步参数
        local_model.load_state_dict(global_model.state_dict())

        # 采样
        log_probs, values, rewards = [], [], []
        state = env.reset()

        for _ in range(20):  # 最多20步
            state = torch.FloatTensor(state).unsqueeze(0)
            action_probs, value = local_model(state)

            dist = torch.distributions.Categorical(action_probs)
            action = dist.sample()

            next_state, reward, done, _ = env.step(action.item())

            log_probs.append(dist.log_prob(action))
            values.append(value)
            rewards.append(reward)

            state = next_state

            if done:
                break

        # 计算损失并更新全局模型
        # ... (类似A2C的更新逻辑)

---

11. PPO算法

11.1 PPO-Clip

class PPO:
    def __init__(
        self,
        state_dim,
        action_dim,
        hidden_dim=64,
        lr=3e-4,
        gamma=0.99,
        lam=0.95,
        clip_epsilon=0.2,
        entropy_coeff=0.01,
        value_coeff=0.5,
        max_grad_norm=0.5,
        ppo_epochs=4,
        mini_batch_size=64
    ):
        self.gamma = gamma
        self.lam = lam
        self.clip_epsilon = clip_epsilon
        self.entropy_coeff = entropy_coeff
        self.value_coeff = value_coeff
        self.max_grad_norm = max_grad_norm
        self.ppo_epochs = ppo_epochs
        self.mini_batch_size = mini_batch_size

        # Actor-Critic网络
        self.actor = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, action_dim),
            nn.Softmax(dim=-1)
        )

        self.critic = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, 1)
        )

        self.optimizer = torch.optim.Adam(
            list(self.actor.parameters()) + list(self.critic.parameters()),
            lr=lr
        )

    def select_action(self, state):
        state = torch.FloatTensor(state).unsqueeze(0)
        action_probs = self.actor(state)
        value = self.critic(state)

        dist = torch.distributions.Categorical(action_probs)
        action = dist.sample()

        return action.item(), dist.log_prob(action), value.item()

    def compute_gae(self, rewards, values, dones):
        """计算广义优势估计"""
        advantages = []
        gae = 0

        for t in reversed(range(len(rewards))):
            if t == len(rewards) - 1:
                next_value = 0
            else:
                next_value = values[t + 1]

            delta = rewards[t] + self.gamma * next_value * (1 - dones[t]) - values[t]
            gae = delta + self.gamma * self.lam * (1 - dones[t]) * gae
            advantages.insert(0, gae)

        returns = [adv + val for adv, val in zip(advantages, values)]
        return advantages, returns

    def update(self, states, actions, old_log_probs, advantages, returns):
        states = torch.FloatTensor(np.array(states))
        actions = torch.LongTensor(actions)
        old_log_probs = torch.FloatTensor(old_log_probs)
        advantages = torch.FloatTensor(advantages)
        returns = torch.FloatTensor(returns)

        # 归一化优势
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)

        for _ in range(self.ppo_epochs):
            # 随机打乱
            indices = torch.randperm(len(states))

            for start in range(0, len(states), self.mini_batch_size):
                end = start + self.mini_batch_size
                idx = indices[start:end]

                # 获取批次数据
                batch_states = states[idx]
                batch_actions = actions[idx]
                batch_old_log_probs = old_log_probs[idx]
                batch_advantages = advantages[idx]
                batch_returns = returns[idx]

                # 当前策略
                action_probs = self.actor(batch_states)
                values = self.critic(batch_states).squeeze()

                dist = torch.distributions.Categorical(action_probs)
                new_log_probs = dist.log_prob(batch_actions)
                entropy = dist.entropy().mean()

                # 比率
                ratio = torch.exp(new_log_probs - batch_old_log_probs)

                # PPO-Clip损失
                surr1 = ratio * batch_advantages
                surr2 = torch.clamp(ratio, 1 - self.clip_epsilon, 1 + self.clip_epsilon) * batch_advantages
                actor_loss = -torch.min(surr1, surr2).mean()

                # 价值损失
                critic_loss = F.mse_loss(values, batch_returns)

                # 总损失
                loss = actor_loss + self.value_coeff * critic_loss - self.entropy_coeff * entropy

                # 更新
                self.optimizer.zero_grad()
                loss.backward()
                nn.utils.clip_grad_norm_(
                    list(self.actor.parameters()) + list(self.critic.parameters()),
                    self.max_grad_norm
                )
                self.optimizer.step()

    def train(self, env, num_iterations=1000, steps_per_iteration=2048):
        rewards_history = []

        for iteration in range(num_iterations):
            # 采样
            states, actions, rewards, dones, old_log_probs, values = [], [], [], [], [], []

            state = env.reset()
            episode_reward = 0

            for _ in range(steps_per_iteration):
                action, log_prob, value = self.select_action(state)
                next_state, reward, done, _ = env.step(action)

                states.append(state)
                actions.append(action)
                rewards.append(reward)
                dones.append(done)
                old_log_probs.append(log_prob)
                values.append(value)

                episode_reward += reward
                state = next_state

                if done:
                    rewards_history.append(episode_reward)
                    episode_reward = 0
                    state = env.reset()

            # 计算GAE
            advantages, returns = self.compute_gae(rewards, values, dones)

            # 更新
            self.update(states, actions, old_log_probs, advantages, returns)

            # 记录
            if iteration % 10 == 0:
                avg_reward = np.mean(rewards_history[-100:]) if rewards_history else 0
                print(f"Iteration {iteration}, Avg Reward: {avg_reward:.2f}")

        return rewards_history

---

12. 模型基础方法

12.1 Dyna-Q

class DynaQ:
    def __init__(self, state_dim, action_dim, planning_steps=10, alpha=0.1, gamma=0.99, epsilon=0.1):
        self.action_dim = action_dim
        self.planning_steps = planning_steps
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon

        # Q表
        self.Q = np.zeros((state_dim, action_dim))

        # 环境模型
        self.model = {}  # (state, action) -> (next_state, reward)

    def select_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.action_dim)
        return np.argmax(self.Q[state])

    def update(self, state, action, reward, next_state, done):
        # 真实经验更新
        if done:
            self.Q[state, action] += self.alpha * (reward - self.Q[state, action])
        else:
            self.Q[state, action] += self.alpha * (
                reward + self.gamma * np.max(self.Q[next_state]) - self.Q[state, action]
            )

        # 存储模型
        self.model[(state, action)] = (next_state, reward)

        # 规划（使用模型）
        for _ in range(self.planning_steps):
            # 随机选择之前经历的状态-动作对
            (s, a), (next_s, r) = random.choice(list(self.model.items()))

            # 模型更新
            self.Q[s, a] += self.alpha * (
                r + self.gamma * np.max(self.Q[next_s]) - self.Q[s, a]
            )

12.2 MCTS (蒙特卡洛树搜索)

class MCTSNode:
    def __init__(self, state, parent=None, action=None):
        self.state = state
        self.parent = parent
        self.action = action
        self.children = []
        self.visits = 0
        self.value = 0.0
        self.untried_actions = None

    def ucb1(self, c=1.41):
        """UCB1选择"""
        if self.visits == 0:
            return float('inf')
        return self.value / self.visits + c * np.sqrt(np.log(self.parent.visits) / self.visits)


class MCTS:
    def __init__(self, env, num_simulations=1000):
        self.env = env
        self.num_simulations = num_simulations

    def search(self, state):
        root = MCTSNode(state)
        root.untried_actions = list(range(self.env.action_space.n))

        for _ in range(self.num_simulations):
            node = root
            env_copy = copy.deepcopy(self.env)

            # 选择
            while node.untried_actions == [] and node.children:
                node = max(node.children, key=lambda n: n.ucb1())

            # 扩展
            if node.untried_actions:
                action = random.choice(node.untried_actions)
                node.untried_actions.remove(action)
                next_state, reward, done, _ = env_copy.step(action)
                child = MCTSNode(next_state, parent=node, action=action)
                node.children.append(child)
                node = child

            # 模拟
            total_reward = 0
            done = False
            while not done:
                action = self.env.action_space.sample()
                _, reward, done, _ = env_copy.step(action)
                total_reward += reward

            # 回溯
            while node is not None:
                node.visits += 1
                node.value += total_reward
                node = node.parent

        # 返回访问次数最多的动作
        return max(root.children, key=lambda n: n.visits).action

---

13. 多智能体强化学习

13.1 独立Q-Learning

class IndependentQLearning:
    def __init__(self, num_agents, state_dim, action_dim, lr=0.1, gamma=0.99, epsilon=0.1):
        self.num_agents = num_agents
        self.agents = [
            QLearningAgent(state_dim, action_dim, lr, gamma, epsilon)
            for _ in range(num_agents)
        ]

    def select_actions(self, states):
        return [agent.select_action(state) for agent, state in zip(self.agents, states)]

    def update(self, states, actions, rewards, next_states, dones):
        for i, agent in enumerate(self.agents):
            agent.update(states[i], actions[i], rewards[i], next_states[i], dones[i])

13.2 通信学习

class CommNet(nn.Module):
    """带有通信的多智能体网络"""
    def __init__(self, state_dim, action_dim, comm_dim=64, hidden_dim=128):
        super().__init__()

        # 编码器
        self.encoder = nn.Linear(state_dim, hidden_dim)

        # 通信层
        self.comm = nn.Linear(hidden_dim + comm_dim, comm_dim)

        # 策略头
        self.policy = nn.Linear(hidden_dim + comm_dim, action_dim)

    def forward(self, states, messages):
        # states: [num_agents, state_dim]
        # messages: [num_agents, comm_dim]

        encoded = F.relu(self.encoder(states))

        # 通信：平均消息
        avg_message = messages.mean(dim=0, keepdim=True).expand_as(messages)

        # 处理
        combined = torch.cat([encoded, avg_message], dim=-1)
        new_messages = F.relu(self.comm(combined))

        # 动作
        policy_input = torch.cat([encoded, new_messages], dim=-1)
        action_probs = F.softmax(self.policy(policy_input), dim=-1)

        return action_probs, new_messages

---

14. 逆强化学习

14.1 核心思想

从专家示范中学习奖励函数。

14.2 最大熵IRL

class MaxEntIRL:
    def __init__(self, state_dim, action_dim, hidden_dim=64, lr=1e-3):
        # 奖励网络
        self.reward_net = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

        self.optimizer = torch.optim.Adam(self.reward_net.parameters(), lr=lr)

    def compute_reward(self, state):
        return self.reward_net(torch.FloatTensor(state))

    def train(self, expert_trajectories, env, num_iterations=100):
        for iteration in range(num_iterations):
            # 1. 计算专家特征期望
            expert_features = self.compute_feature_expectations(expert_trajectories)

            # 2. 学习当前奖励下的策略（使用RL）
            policy = self.learn_policy(env)

            # 3. 采样轨迹
            rl_trajectories = self.sample_trajectories(env, policy)

            # 4. 计算RL特征期望
            rl_features = self.compute_feature_expectations(rl_trajectories)

            # 5. 更新奖励网络
            loss = -(expert_features - rl_features).mean()

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

---

15. 强化学习应用

15.1 游戏

# Atari游戏
import gym

env = gym.make('PongNoFrameskip-v4')
agent = DQNAgent(env.observation_space.shape, env.action_space.n)
agent.train(env, num_episodes=10000)

15.2 机器人控制

# 连续动作空间
import gym

env = gym.make('HalfCheetah-v4')
agent = SAC(
    state_dim=env.observation_space.shape[0],
    action_dim=env.action_space.shape[0],
    action_bounds=(env.action_space.low, env.action_space.high)
)

15.3 推荐系统

class RecSysRL:
    def __init__(self, num_items, embedding_dim=64):
        self.state_encoder = nn.LSTM(embedding_dim, embedding_dim)
        self.policy = nn.Linear(embedding_dim, num_items)

    def recommend(self, user_history):
        # 编码历史
        state, _ = self.state_encoder(user_history)

        # 选择推荐项
        action_probs = F.softmax(self.policy(state[-1]), dim=-1)
        return torch.multinomial(action_probs, 1)

15.4 RLHF (语言模型对齐)

# 见RLHF文档
# 使用PPO训练语言模型

---

16. 完整代码实现

16.1 Gym环境示例

import gymnasium as gym
import numpy as np

def train_q_learning():
    env = gym.make('FrozenLake-v1', is_slippery=False)

    # 超参数
    num_episodes = 10000
    learning_rate = 0.1
    gamma = 0.99
    epsilon = 0.1

    # Q表
    Q = np.zeros([env.observation_space.n, env.action_space.n])

    # 训练
    for episode in range(num_episodes):
        state, _ = env.reset()
        done = False

        while not done:
            # ε-贪婪
            if np.random.random() < epsilon:
                action = env.action_space.sample()
            else:
                action = np.argmax(Q[state])

            # 执行动作
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            # Q-Learning更新
            Q[state, action] += learning_rate * (
                reward + gamma * np.max(Q[next_state]) * (1 - terminated) - Q[state, action]
            )

            state = next_state

        if episode % 1000 == 0:
            # 评估
            total_rewards = []
            for _ in range(100):
                state, _ = env.reset()
                done = False
                total_reward = 0
                while not done:
                    action = np.argmax(Q[state])
                    state, reward, terminated, truncated, _ = env.step(action)
                    done = terminated or truncated
                    total_reward += reward
                total_rewards.append(total_reward)
            print(f"Episode {episode}, Avg Reward: {np.mean(total_rewards):.2f}")

    return Q

16.2 PyTorch完整示例

import torch
import torch.nn as nn
import torch.optim as optim
import gymnasium as gym
import numpy as np
from collections import deque

def train_dqn_cartpole():
    env = gym.make('CartPole-v1')

    # 超参数
    state_dim = env.observation_space.shape[0]
    action_dim = env.action_space.n
    hidden_dim = 128
    lr = 1e-3
    gamma = 0.99
    epsilon_start = 1.0
    epsilon_end = 0.01
    epsilon_decay = 0.995
    buffer_size = 10000
    batch_size = 64
    target_update = 10
    num_episodes = 500

    # 创建智能体
    agent = DQNAgent(
        state_dim=state_dim,
        action_dim=action_dim,
        hidden_dim=hidden_dim,
        lr=lr,
        gamma=gamma,
        epsilon_start=epsilon_start,
        epsilon_end=epsilon_end,
        epsilon_decay=epsilon_decay,
        buffer_size=buffer_size,
        batch_size=batch_size,
        target_update=target_update
    )

    # 训练
    rewards_history = agent.train(env, num_episodes)

    return agent, rewards_history

---

17. 参考资料

核心教材

1. Sutton & Barto: “Reinforcement Learning: An Introduction” (2018)
2. Szepesvári: “Algorithms for Reinforcement Learning” (2010)

核心论文

1. DQN: “Playing Atari with Deep Reinforcement Learning” (2013)
2. PPO: “Proximal Policy Optimization Algorithms” (2017)
3. A3C: “Asynchronous Methods for Deep Reinforcement Learning” (2016)
4. SAC: “Soft Actor-Critic” (2018)
5. AlphaGo: “Mastering the game of Go with deep neural networks and tree search” (2016)
6. AlphaStar: “Grandmaster level in StarCraft II using multi-agent reinforcement learning” (2019)

开源库

• Stable Baselines3: https://github.com/DLR-RM/stable-baselines3
• RLlib: https://github.com/ray-project/ray
• CleanRL: https://github.com/vwxyzjn/cleanrl
• Gymnasium: https://github.com/Farama-Foundation/Gymnasium

推荐资源

• David Silver RL课程: https://www.davidsilver.uk/teaching/
• OpenAI Spinning Up: https://spinningup.openai.com
• DeepMind x UCL RL课程

---