从零构建悬崖漫步AI：Sarsa算法实战与深度调优指南

如果你已经对强化学习的基本概念有所了解，比如知道什么是状态、动作和奖励，但每次看到那些抽象的数学公式和理论推导，总觉得离真正的代码实现还隔着一层纱，那么这篇文章就是为你准备的。我们今天不谈复杂的收敛性证明，也不做泛泛的概念科普，而是直接打开Jupyter Notebook，从一个经典的“悬崖漫步”环境入手，手把手地带你实现Sarsa算法。你会发现，这个看似简单的“走一步看一步”的算法，背后藏着许多影响最终性能的魔鬼细节。无论是学习率的微妙调整，还是探索与利用的永恒博弈，我们都会在代码的运行和结果的对比中，获得最直观的体会。本文面向的是希望将理论知识转化为实际项目能力的Python开发者，我们将一起构建一个完整的、可运行的智能体，并深入探讨如何让它学得更快、走得更稳。

1. 环境搭建与核心概念具象化

在开始写任何算法代码之前，我们必须先理解智能体所处的“世界”。悬崖漫步环境是一个极其精巧的简化模型，它完美地封装了强化学习中的几个核心挑战：稀疏奖励、致命风险和寻找最优路径。

想象一个4行12列的网格世界。智能体从左下角(3, 0)出发，目标是到达右下角(3, 11)的终点。最底下那一行的中间10个格子(3,1)到(3,10)，是万丈悬崖。规则很简单：

• 每走一步，只要没掉下悬崖也没到达终点，就收到-1的奖励（鼓励用最短路径到达）。
• 如果掉下悬崖，则收到-100的巨额惩罚，并被瞬间传送回起点。
• 到达终点则获得0奖励，并结束本轮游戏。

这个环境之所以经典，是因为它迫使智能体在“冒险抄近路”和“保守绕远路”之间做出抉择。一条贴着悬崖边走的路径步数最少，但风险极高；而一条从上方安全绕行的路径虽然步数多，但更为稳妥。我们的Sarsa智能体将通过学习，自己找到这个平衡点。

下面我们用Python来定义这个世界。注意，为了后续与大多数强化学习库的惯例保持一致，我们将二维的网格坐标转换为一维的状态索引，这是一个非常实用的技巧。

import numpy as np

class CliffWalkingEnv:
    """悬崖漫步环境"""
    def __init__(self, ncol=12, nrow=4):
        self.ncol = ncol
        self.nrow = nrow
        self.start = (nrow - 1, 0)  # 起点：左下角
        self.goal = (nrow - 1, ncol - 1)  # 终点：右下角
        # 悬崖是最后一排从第1列到第10列的所有格子
        self.cliff = [(self.nrow - 1, j) for j in range(1, self.ncol - 1)]
        self.reset()

    def reset(self):
        """重置环境，返回初始状态索引"""
        self.state = self.start
        return self._state_to_index(self.state)

    def step(self, action):
        """执行动作（0:上, 1:右, 2:下, 3:左），返回(next_state, reward, done, info)"""
        x, y = self.state
        # 根据动作移动，并处理边界（不让走出网格）
        if action == 0:   # 上
            x = max(x - 1, 0)
        elif action == 1: # 右
            y = min(y + 1, self.ncol - 1)
        elif action == 2: # 下
            x = min(x + 1, self.nrow - 1)
        elif action == 3: # 左
            y = max(y - 1, 0)

        self.state = (x, y)
        next_state_idx = self._state_to_index(self.state)

        # 判断奖励和是否终止
        if self.state == self.goal:
            reward = 0
            done = True
        elif self.state in self.cliff:
            reward = -100
            done = False  # 注意：掉下悬崖不算终止，但会重置状态
            self.state = self.start  # 传送回起点
        else:
            reward = -1
            done = False

        return next_state_idx, reward, done, {}

    def _state_to_index(self, state):
        """将(row, col)的二维坐标映射到一维索引，方便Q表存储"""
        return state[0] * self.ncol + state[1]

    def render(self):
        """简单打印当前网格状态，用于调试"""
        grid = [['.' for _ in range(self.ncol)] for _ in range(self.nrow)]
        for (cx, cy) in self.cliff:
            grid[cx][cy] = 'C'  # 悬崖
        grid[self.goal[0]][self.goal[1]] = 'G'  # 终点
        grid[self.state[0]][self.state[1]] = 'A'  # 智能体
        for row in grid:
            print(' '.join(row))
        print()

提示：在step函数中，掉下悬崖后done=False是一个关键设计。这意味着本轮游戏继续，智能体必须从起点重新开始尝试。这模拟了“生命不止，学习不息”的过程，让智能体有更多机会从错误中学习。

有了环境，我们就可以初始化一个智能体进去看看。先手动控制它走几步，感受一下环境的反馈：

env = CliffWalkingEnv()
state = env.reset()
print(f"初始状态索引: {state}")
env.render()

# 尝试向右走一步
next_state, reward, done, _ = env.step(1)
print(f"向右走一步后: 状态={next_state}, 奖励={reward}, 是否结束={done}")
env.render()

2. Sarsa算法核心：五元组更新与代码实现

现在进入正题：Sarsa算法。这个名字来源于它更新价值函数时所依赖的五个元素：(S, A, R, S', A')，即当前状态、当前动作、获得的奖励、下一个状态、以及在下一个状态根据当前策略选择的动作。这最后一个A'，是Sarsa与其近亲Q-learning最本质的区别。Sarsa是一种同策略算法，它评估和改进的是它自己正在执行的那个策略。

它的核心更新公式，用一句话概括就是：用“当前奖励”加上“对下一步的估值”来更新“当前动作的价值”。具体公式如下：

[ Q(S, A) \leftarrow Q(S, A) + \alpha [R + \gamma Q(S', A') - Q(S, A)] ]

其中：

• α 是学习率，控制新信息覆盖旧信息的强度。
• γ 是折扣因子，决定未来奖励在当前有多重要。
• R + γQ(S', A') 被称为时序差分目标，R + γQ(S', A') - Q(S, A) 被称为时序差分误差。

这个公式的直觉非常清晰：如果执行A后得到的即时奖励R加上下一个状态-动作对的预期价值Q(S',A')，比我们之前认为的Q(S,A)要高，那么我们就向上调整Q(S,A)；反之则向下调整。

接下来，我们实现Sarsa智能体。我们需要一个Q表来存储所有状态-动作对的价值，以及一个选择动作的策略。这里我们采用最经典的ε-贪婪策略。

class SarsaAgent:
    """Sarsa算法智能体"""
    def __init__(self, env, learning_rate=0.1, discount_factor=0.95, epsilon=0.1):
        self.env = env
        self.lr = learning_rate       # 学习率 α
        self.gamma = discount_factor  # 折扣因子 γ
        self.epsilon = epsilon        # 探索概率 ε

        self.n_states = env.nrow * env.ncol
        self.n_actions = 4  # 上、右、下、左
        # 初始化Q表，可以尝试用小的随机数初始化，打破对称性
        self.Q = np.random.uniform(low=-0.01, high=0.01, size=(self.n_states, self.n_actions))

    def choose_action(self, state):
        """基于ε-贪婪策略选择动作"""
        if np.random.random() < self.epsilon:
            # 探索：随机选择一个动作
            return np.random.choice(self.n_actions)
        else:
            # 利用：选择当前状态下Q值最大的动作
            # np.argmax 默认返回第一个最大值索引，这正合我意
            return np.argmax(self.Q[state])

    def update(self, state, action, reward, next_state, next_action, done):
        """执行Sarsa核心更新"""
        current_q = self.Q[state, action]
        # 如果游戏已经结束，则没有下一个状态-动作对的价值
        next_q = self.Q[next_state, next_action] if not done else 0.0

        # 计算时序差分目标
        td_target = reward + self.gamma * next_q
        # 计算时序差分误差
        td_error = td_target - current_q
        # 更新Q值
        self.Q[state, action] += self.lr * td_error

    def train_one_episode(self, env):
        """在环境中运行一个完整的回合，并持续学习"""
        state = env.reset()
        action = self.choose_action(state)
        total_reward = 0
        steps = 0
        done = False

        while not done:
            # 执行动作，与环境交互
            next_state, reward, done, _ = env.step(action)
            # 根据当前策略选择下一个动作（关键！）
            next_action = self.choose_action(next_state)

            # 使用五元组 (s, a, r, s', a') 更新Q表
            self.update(state, action, reward, next_state, next_action, done)

            # 转移到下一个状态和动作
            state, action = next_state, next_action
            total_reward += reward
            steps += 1

        return total_reward, steps

    def get_policy(self):
        """从训练好的Q表中提取贪婪策略（每个状态下选择最优动作）"""
        policy = np.argmax(self.Q, axis=1)
        return policy

注意：在update函数中，当done=True时，我们将next_q设为0。这是因为在终止状态后，没有后续的状态和动作，其未来累积奖励为0。这是处理回合终止的标准做法。

让我们立刻运行一个简单的训练循环，看看这个智能体最初的表现如何。我们将记录每轮的总奖励和步数。

def train_sarsa(env, agent, episodes=500):
    """训练循环"""
    reward_history = []
    step_history = []

    for episode in range(episodes):
        total_reward, steps = agent.train_one_episode(env)
        reward_history.append(total_reward)
        step_history.append(steps)

        # 每100轮输出一次进度
        if (episode + 1) % 100 == 0:
            avg_reward = np.mean(reward_history[-100:])
            avg_steps = np.mean(step_history[-100:])
            print(f"Episode {episode+1:4d} | Avg Reward (last 100): {avg_reward:7.2f} | Avg Steps: {avg_steps:5.2f}")

    return reward_history, step_history

# 创建环境和智能体
env = CliffWalkingEnv()
agent = SarsaAgent(env, learning_rate=0.1, discount_factor=0.95, epsilon=0.1)

print("开始训练Sarsa智能体...")
rewards, steps = train_sarsa(env, agent, episodes=500)

运行这段代码，你可能会看到智能体的平均奖励从最初的-1000多（不断掉下悬崖）逐渐提升到-100左右，平均步数也从几十步逐渐下降。这说明它在学习！但它学到的策略是最优的吗？我们接下来就需要一套方法来评估和可视化它的学习成果。

3. 可视化、评估与策略分析

训练完成后，一堆数字可能不够直观。我们需要“看见”智能体学到了什么。我们将从三个维度进行可视化：1）学习曲线；2）最终策略网格图；3）Q值热力图。

首先，绘制奖励和步数随训练轮次变化的曲线。这是观察算法是否收敛、学习过程是否稳定的最基本工具。

import matplotlib.pyplot as plt

def plot_training_progress(rewards, steps, window=50):
    """绘制训练进度图，包含滑动平均曲线"""
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10))

    # 绘制原始奖励曲线和滑动平均
    ax1.plot(rewards, alpha=0.3, label='Raw Reward', color='lightblue')
    reward_smooth = np.convolve(rewards, np.ones(window)/window, mode='valid')
    ax1.plot(range(window-1, len(rewards)), reward_smooth, label=f'Moving Avg (window={window})', color='blue', linewidth=2)
    ax1.set_xlabel('Training Episode')
    ax1.set_ylabel('Total Reward per Episode')
    ax1.set_title('Sarsa Learning Curve - Reward')
    ax1.legend()
    ax1.grid(True, alpha=0.3)

    # 绘制原始步数曲线和滑动平均
    ax2.plot(steps, alpha=0.3, label='Raw Steps', color='lightcoral')
    steps_smooth = np.convolve(steps, np.ones(window)/window, mode='valid')
    ax2.plot(range(window-1, len(steps)), steps_smooth, label=f'Moving Avg (window={window})', color='red', linewidth=2)
    ax2.set_xlabel('Training Episode')
    ax2.set_ylabel('Steps per Episode')
    ax2.set_title('Sarsa Learning Curve - Steps')
    ax2.legend()
    ax2.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

plot_training_progress(rewards, steps)

接着，让我们看看智能体最终学到的策略是什么样子。我们将每个状态下的最优动作用箭头表示出来。

def visualize_policy(env, agent):
    """在网格中可视化策略"""
    policy = agent.get_policy()
    action_symbols = ['↑', '→', '↓', '←']

    grid = []
    for i in range(env.nrow):
        row_display = []
        for j in range(env.ncol):
            state_idx = i * env.ncol + j
            if (i, j) == env.goal:
                row_display.append(' G ')  # 终点
            elif (i, j) in env.cliff:
                row_display.append(' C ')  # 悬崖
            elif (i, j) == env.start:
                row_display.append(' S ')  # 起点
            else:
                row_display.append(f' {action_symbols[policy[state_idx]]} ')
        grid.append(row_display)

    print("学习到的最优策略（箭头表示从该格子出发应走的方向）:")
    print("-" * (env.ncol * 4 + 1))
    for row in grid:
        print('|' + '|'.join(row) + '|')
        print("-" * (env.ncol * 4 + 1))

visualize_policy(env, agent)

一个训练良好的Sarsa智能体，通常会学到一个非常保守的策略：从起点开始向上走，沿着最上面一行走到最右端，再向下走到终点。这个策略完美地避开了所有悬崖格子，虽然步数不是绝对最少（贴着悬崖走的那条路更短），但绝对安全。这正是Sarsa“同策略”特性在风险敏感环境中的体现——它评估的是自己实际要走的路径，因此会对危险动作给出极低的估值，从而避免选择它们。

最后，我们可以窥探一下智能体“大脑”的内部——Q表。通过热力图观察不同状态-动作对的价值，能加深我们对算法工作原理的理解。

import seaborn as sns

def plot_q_value_heatmap(env, agent, action_name='Right'):
    """绘制特定动作在所有状态下的Q值热图"""
    action_idx = ['Up', 'Right', 'Down', 'Left'].index(action_name)
    q_values_for_action = agent.Q[:, action_idx].reshape(env.nrow, env.ncol)

    plt.figure(figsize=(env.ncol, env.nrow))
    sns.heatmap(q_values_for_action, annot=True, fmt=".1f", cmap="RdYlGn", center=0,
                cbar_kws={'label': f'Q-value for action: {action_name}'})
    plt.title(f'Q-value Heatmap for Action "{action_name}"')
    plt.xlabel('Column')
    plt.ylabel('Row')
    plt.show()

# 例如，查看“向右”动作的Q值分布
plot_q_value_heatmap(env, agent, action_name='Right')

在“向右”动作的Q值热图中，你应该能看到在最上面一行，越靠近右侧的格子，其“向右”的Q值越高，因为离终点越近。而在悬崖上方的格子，其“向下”动作的Q值会非常低。这些直观的图表，是将抽象算法具象化的有力工具。

4. 超参数调优：寻找最佳学习配置

到目前为止，我们用的都是一组默认的超参数。但在实际项目中，超参数的选择往往对算法性能有决定性影响。Sarsa主要有三个关键超参数：学习率α、折扣因子γ和探索率ε。我们来系统地探索一下它们的影响。

我们将设计一个实验，在合理的范围内网格搜索这些参数，并用平均奖励作为性能指标。为了节省时间，我们只训练较少的轮次，但足以看出趋势。

def parameter_grid_search(env, param_grid, eval_episodes=200, train_episodes=300):
    """
    对Sarsa进行超参数网格搜索
    param_grid: 字典，例如 {'lr': [0.01, 0.1, 0.5], 'gamma': [0.9, 0.95, 0.99], 'epsilon': [0.05, 0.1, 0.2]}
    """
    results = []
    configs = []

    # 遍历所有参数组合
    import itertools
    keys, values = zip(*param_grid.items())
    for combination in itertools.product(*values):
        params = dict(zip(keys, combination))
        print(f"Testing config: {params}")

        # 使用当前参数创建并训练智能体
        agent = SarsaAgent(env, learning_rate=params['lr'],
                           discount_factor=params['gamma'],
                           epsilon=params['epsilon'])
        # 训练阶段
        train_rewards, _ = train_sarsa(env, agent, episodes=train_episodes)

        # 评估阶段：关闭探索，用贪婪策略运行多个回合
        agent.epsilon = 0.0  # 评估时不再随机探索
        eval_rewards = []
        for _ in range(eval_episodes):
            state = env.reset()
            done = False
            total_reward = 0
            while not done:
                action = agent.choose_action(state)  # 此时是纯贪婪策略
                next_state, reward, done, _ = env.step(action)
                state = next_state
                total_reward += reward
            eval_rewards.append(total_reward)

        avg_eval_reward = np.mean(eval_rewards)
        results.append(avg_eval_reward)
        configs.append(params)
        print(f"  -> Average evaluation reward: {avg_eval_reward:.2f}\n")

    # 找出最佳配置
    best_idx = np.argmax(results)
    best_config = configs[best_idx]
    best_score = results[best_idx]

    print("="*50)
    print(f"Grid Search Completed.")
    print(f"Best Configuration: {best_config}")
    print(f"Best Average Evaluation Reward: {best_score:.2f}")

    return best_config, best_score, results, configs

# 定义要搜索的参数网格
param_grid = {
    'lr': [0.01, 0.1, 0.3],       # 学习率太小学得慢，太大不稳定
    'gamma': [0.85, 0.95, 0.99],   # 折扣因子影响未来奖励的权重
    'epsilon': [0.05, 0.1, 0.2]    # 探索率影响探索与利用的平衡
}

# 注意：完整网格搜索（3x3x3=27种组合）需要一定时间，你可以先小规模测试
best_config, best_score, all_results, all_configs = parameter_grid_search(env, param_grid, eval_episodes=50, train_episodes=150)

根据实验结果，你可能会发现一些规律：

• 学习率α：0.1通常是一个不错的起点。过小（如0.01）导致学习缓慢；过大（如0.5）可能导致Q值震荡，难以收敛。
• 折扣因子γ：在悬崖漫步中，因为每步都有-1的即时惩罚，一个较高的γ（如0.95或0.99）能让智能体更看重尽快到达终点以停止惩罚，从而促进其找到较短路径。
• 探索率ε：0.1是一个经典值。太小的ε（如0.01）可能导致智能体过早陷入局部最优（比如某条次优路径）；太大的ε（如0.3）则会导致智能体一直进行低效的随机探索，无法充分利用学到的知识。

我们可以将不同参数组合的性能用表格来清晰对比：

学习率 (α)	折扣因子 (γ)	探索率 (ε)	平均评估奖励	备注
0.01	0.95	0.10	~ -120	学习过慢，尚未充分收敛
0.10	0.85	0.10	~ -105	未来奖励权重不足，可能选择绕远路
0.10	0.95	0.05	~ -98	探索不足，可能偶尔会掉下悬崖
0.10	0.95	0.10	~ -85	稳健且表现良好的默认配置
0.10	0.99	0.10	~ -88	与0.95相近，有时略优
0.30	0.95	0.10	~ -90 到 -110	学习不稳定，奖励波动大
0.10	0.95	0.20	~ -95	探索过多，影响策略利用效率

注意：表格中的奖励值是示例，实际运行结果会因随机种子等因素有浮动。但相对优劣关系通常保持一致。

调参不仅仅是找到一组数字，更是理解每个参数如何影响智能体的“性格”和学习过程。这个过程充满了实验的乐趣，也是机器学习工程实践中的重要一环。

5. 进阶技巧：从单步Sarsa到多步Sarsa与工程实践

我们目前实现的Sarsa是“单步”的，它只往前看一步(S', A')就进行更新。这种方法简单高效，但有时学习速度不够快。多步Sarsa是单步Sarsa的自然扩展，它在更新当前值函数时，会考虑未来多步的奖励，然后再用后续状态的估计值来“收尾”。这就像不是每走一步就反思一次，而是走了一段路（比如3步）后，把这段路的收获和接下来的计划结合起来，做一次更全面的总结。

多步Sarsa的更新目标（n步回报）定义为： [ G_{t:t+n} = R_{t+1} + \gamma R_{t+2} + ... + \gamma^{n-1} R_{t+n} + \gamma^n Q(S_{t+n}, A_{t+n}) ] 然后用这个G去更新Q(S_t, A_t)。

实现多步Sarsa需要维护一个最近n步的轨迹（状态、动作、奖励）。代码比单步复杂，但能显著提升在某些环境下的样本效率（即用更少的环境交互学到更好的策略）。由于篇幅限制，这里不展开完整代码，但理解其思想至关重要：它在单步TD方法和蒙特卡洛方法之间做了一个折中，平衡了估计的偏差和方差。

除了算法本身的扩展，在实际工程化过程中，我们还需要考虑以下几点：

1. 经验回放：虽然标准的Sarsa是同策略的，无法直接使用异策略的经验回放，但在一些变体中，可以结合使用。对于复杂的项目，设计一个高效的数据存储和采样机制是必须的。
2. 模型保存与加载：训练一个好的模型可能需要很长时间。我们需要将最终的Q表或策略网络参数保存下来，以便后续部署或继续训练。

   import pickle
   # 保存智能体
   with open('sarsa_agent.pkl', 'wb') as f:
       pickle.dump(agent, f)
   # 加载智能体
   with open('sarsa_agent.pkl', 'rb') as f:
       loaded_agent = pickle.load(f)
   

3. TensorBoard可视化：对于更复杂的训练过程，使用TensorBoard或Weights & Biases等工具来实时跟踪损失、奖励、探索率衰减等指标，是进行高效调试和实验管理的标准做法。
4. 封装为可调用API：最终，我们需要将训练好的智能体封装成一个简单的函数或类方法，接收状态，返回动作，以便集成到更大的应用系统中。

在悬崖漫步这个简单环境里摸爬滚打一遍后，你应该对Sarsa算法有了肌肉记忆般的理解。下次当你遇到一个需要序列决策的新问题时，比如简单的游戏AI、广告排序策略的模拟测试，或是机器人控制的基础原型，不妨从搭建一个类似的环境和实现一个Sarsa智能体开始。从这些看似简单的基石出发，你才能更扎实地走向深度Q网络、策略梯度等更现代的强化学习算法。