一文读懂时序差分算法（Sarsa算法）—强化学习(3)
·
目录
1、核心理论：在线时序差分学习（On-policy TD Learning）
1、核心理论：在线时序差分学习（On-policy TD Learning）
Sarsa 算法是强化学习中时序差分（TD）学习的一种经典实现，属于在线策略（On-policy） 算法。其核心思想是：在与环境交互的过程中，通过不断更新动作价值函数（Q 函数）来学习最优策略，且学习的策略与正在执行的策略保持一致（即 “在线” 更新）。
- 动作价值函数（Q 函数）：定义为Q(s,a)，表示在状态s下执行动作a后，遵循当前策略所能获得的累积折扣奖励的期望。
- 目标：通过学习使Q(s,a)收敛到最优动作价值函数
，此时对应的策略即为最优策略。
边试边学，走一步看一步
Sarsa 算法是强化学习里一种 “边行动边学习” 的方法。它的核心逻辑很简单：你在做一件事的时候，每走一步，就根据接下来的情况调整对 “当前这一步好坏” 的判断，而且判断的依据和你实际要走的下一步是绑定的。
打个比方：你在迷宫里找出口，每走一步（比如向东），会得到反馈（比如吃到金币或撞到墙），然后你会看下一步打算往哪走（比如接下来向北），再根据 “当前步的反馈 + 下一步的计划”，来修正 “向东这一步到底好不好” 的想法。下次再到这个位置，就知道该怎么选了。
2、基本原理：“S-A-R-S'-A'” 的循环更新
Sarsa 算法的名称来源于其更新过程中涉及的五个关键要素： 当前状态（State, s）→ 当前动作（Action, a）→ 奖励（Reward, r）→ 下一状态（Next State, s'）→ 下一动作（Next Action, a'），即 “” 的序列。
算法的核心流程如下：
- 初始化：设定初始状态s，并根据当前策略（如
- 贪婪策略）选择初始动作a。
- 交互与观察：执行动作a，获得环境反馈的奖励r和下一状态s'。
- 确定下一动作：根据当前策略，从下一状态s'中选择动作a'。
- 更新 Q 函数：利用时序差分误差（TD error）更新Q(s,a)。
- 迭代：将s'设为当前状态s，a'设为当前动作a，重复步骤 2-4，直到达到终止状态（如任务完成或失败）。
具体过程：“走 - 看 - 评 - 改” 的循环
Sarsa 的名字来源于它每次学习都离不开 5 个关键信息：
现在在哪（状态 s）→ 现在做什么（动作 a）→ 得到什么反馈（奖励 r）→ 接下来到哪（新状态 s'）→ 接下来做什么（新动作 a'）
整个过程像个循环：
- 开始：比如你在迷宫的某个位置（s），凭当前的经验选一个动作（a，比如向右走）。
- 行动与反馈：执行动作 a，得到奖励 r（比如 + 10 分），并走到新位置 s'。
- 计划下一步：根据当前的经验，从新位置 s' 再选一个下一步的动作 a'（比如接下来向上走）。
- 更新经验：用 “刚才的奖励 + 下一步的预期”，修正对 “在 s 做 a” 的评价（也就是 Q 值）。
- 重复：把新位置 s' 当成现在的位置，新动作 a' 当成现在要做的动作，继续循环，直到到达终点（比如走出迷宫）。
3、核心公式：Q 函数的更新规则
Sarsa 算法通过时序差分目标（TD target） 与当前 Q 值的差异来更新Q(s,a)，公式如下：
-
时序差分目标（TD Target） 表示在状态s执行动作a后，基于 “当前策略” 所能预期的累积奖励，计算公式为：
其中：
- r：当前动作a获得的即时奖励；
：折扣因子，用于权衡即时奖励和未来奖励的重要性；
- Q(s',a')：下一状态s'和下一动作a'的当前 Q 值（基于当前策略）。
-
时序差分误差（TD Error） 衡量当前 Q 值与预期目标的差异：
-
Q 函数更新公式 利用 TD 误差更新Q(s,a)，公式为：
其中：
合并后可写为：
：学习率，控制每次更新的幅度（值越大，更新越激进；值越小，收敛越稳定）。
简单公式：怎么修正 “好坏评价”？
我们用 Q (s,a) 表示 “在状态 s 做动作 a” 的好坏（值越大越好）。Sarsa 的核心就是更新这个 Q 值，公式可以通俗写成：
新的 Q (s,a) = 原来的 Q (s,a) + 一点点修正
这个 “修正” 怎么来？看两点：
- 刚才的奖励 r（比如吃到金币 + 10，撞到墙 - 5）；
- 下一步的预期：也就是 “到了新位置 s'，做动作 a'” 的好坏（Q (s',a')），再打个折扣（比如未来的奖励没现在重要，打 8 折）。
所以修正量是：（r + 折扣 × 下一步的 Q 值） - 原来的 Q 值
最后公式写成： Q(s,a) = Q(s,a) + α × [r + γ × Q(s',a') - Q(s,a)]
这里：
- α 是 “学习率”（比如 0.1），控制每次修正的幅度（别改太猛）；
- γ 是 “折扣因子”（比如 0.9），表示未来的奖励有多重要（越接近 1，越看重长远）。
4、多步 Sarsa 算法核心
多步 Sarsa 与单步 Sarsa 的核心区别是：利用未来 n 步的累积奖励更新 Q 值，而非仅依赖单步奖励，从而更高效地学习策略。
- 核心思想：通过累积未来 n 步的奖励（而非单步奖励）计算更新目标，平衡偏差和方差，加速收敛。
- 关键公式：n 步回报（n-step return）
其中，t为当前时间步，n为步数，R为奖励，
为折扣因子。
- 更新规则：
5、策略与收敛性
- 策略选择：Sarsa 通常采用
的概率选择当前 Q 值最大的动作（ exploitation），以
- 收敛性：在满足一定条件（如学习率
随时间递减且总和无穷、策略渐进收敛到贪婪策略）时，Sarsa 算法可保证Q(s,a)收敛到最优值
Sarsa 最特别的地方是：它学习的 “经验” 和实际 “行动” 是一致的。比如它计划下一步要向上走（a'），就会用 “向上走的预期” 来修正现在的判断，而不是空想 “如果下一步走最好的路会怎样”。这种 “说到做到” 的特点，让它在需要 “谨慎规划” 的任务（比如悬崖边走路，不能随便尝试危险动作）中更稳妥。
6、完整代码
6.1、一步Sarsa算法
6.1.1、完整代码
"""
文件名: 5.1
作者: 墨尘
日期: 2025/7/22
项目名: d2l_learning
备注: Sarsa算法在悬崖漫步环境中的实现与可视化
"""
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.table import Table
class CliffWalkingEnv:
"""悬崖漫步环境（确定性动作）"""
def __init__(self, ncol=12, nrow=4):
"""初始化环境参数"""
self.ncol = ncol # 环境宽度（列数）
self.nrow = nrow # 环境高度（行数）
self.reset() # 重置环境状态到起点
# 定义悬崖位置：最后一行的第1列到第10列
self.cliff = [(nrow - 1, j) for j in range(1, ncol - 1)]
self.goal = (nrow - 1, ncol - 1) # 目标位置：右下角
def reset(self):
"""重置环境到初始状态"""
self.state = (self.nrow - 1, 0) # 起点：左下角(3,0)
return self._state_to_index(self.state) # 返回一维状态索引
def step(self, action):
"""执行动作并返回下一状态、奖励和终止标志"""
# 动作编码：0=上, 1=右, 2=下, 3=左
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) # 更新当前状态
# 计算奖励和终止条件
if self.state == self.goal:
reward = 0 # 到达目标获得0奖励
done = True # 终止回合
elif self.state in self.cliff:
reward = -100 # 掉入悬崖获得-100惩罚
done = False # 回合不终止，但回到起点
self.state = (self.nrow - 1, 0) # 重置到起点
else:
reward = -1 # 每步移动获得-1奖励（鼓励最短路径）
done = False
return self._state_to_index(self.state), reward, done, {}
def _state_to_index(self, state):
"""将二维坐标状态转换为一维索引"""
return state[0] * self.ncol + state[1]
class SarsaAgent:
"""Sarsa算法实现"""
def __init__(self, env, alpha=0.5, gamma=0.9, epsilon=0.1):
"""初始化Sarsa智能体参数"""
self.env = env # 环境实例
self.n_states = env.nrow * env.ncol # 状态总数
self.n_actions = 4 # 动作总数（上右下左）
self.alpha = alpha # 学习率：控制更新步长
self.gamma = gamma # 折扣因子：权衡即时奖励和未来奖励
self.epsilon = epsilon # ε-贪婪策略参数
# 初始化Q表：状态×动作矩阵，存储每个状态-动作对的价值
self.Q = np.zeros((self.n_states, self.n_actions))
def epsilon_greedy_policy(self, state):
"""ε-贪婪策略：以ε概率随机探索，以1-ε概率利用当前最优动作"""
if np.random.random() < self.epsilon:
return np.random.randint(0, self.n_actions) # 随机动作（探索）
else:
return np.argmax(self.Q[state]) # 最优动作（利用）
def update(self, state, action, reward, next_state, next_action, done):
"""Sarsa更新公式：基于五元组(s,a,r,s',a')更新Q表"""
current_q = self.Q[state, action] # 当前状态-动作对的Q值
# 下一状态-动作对的Q值（若回合结束则为0）
next_q = self.Q[next_state, next_action] if not done else 0
# 计算TD目标：即时奖励 + 折扣后的下一状态Q值
td_target = reward + self.gamma * next_q
# 计算TD误差：TD目标与当前Q值的差
td_error = td_target - current_q
# 更新Q表：当前Q值向TD目标移动α步长
self.Q[state, action] += self.alpha * td_error
def train(self, episodes=500):
"""训练智能体指定轮数"""
rewards_history = [] # 记录每轮总奖励
steps_history = [] # 记录每轮步数
for episode in range(episodes):
state = self.env.reset() # 重置环境，获取初始状态
action = self.epsilon_greedy_policy(state) # 选择初始动作
done = False # 终止标志
total_reward = 0 # 累积奖励
steps = 0 # 步数
while not done:
# 执行动作，获取下一状态、奖励和终止标志
next_state, reward, done, _ = self.env.step(action)
# 基于下一状态选择下一动作（ε-贪婪策略）
next_action = self.epsilon_greedy_policy(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 # 增加步数
# 记录本轮训练结果
rewards_history.append(total_reward)
steps_history.append(steps)
# 每100轮打印训练进度
if (episode + 1) % 100 == 0:
print(f"Episode {episode + 1}/{episodes}, "
f"Total Reward: {total_reward}, "
f"Steps: {steps}")
# 训练结束后，提取最优策略（每个状态下Q值最大的动作）
self.policy = np.argmax(self.Q, axis=1)
return rewards_history, steps_history
def visualize_policy(agent, env):
"""可视化最优策略（动作方向）"""
action_symbols = ['↑', '→', '↓', '←'] # 动作符号映射
grid = np.empty((env.nrow, env.ncol), dtype=object)
# 构建策略网格
for i in range(env.nrow):
for j in range(env.ncol):
state_idx = i * env.ncol + j
if (i, j) == env.goal:
grid[i, j] = 'G' # 目标位置标记
elif (i, j) in env.cliff:
grid[i, j] = 'C' # 悬崖位置标记
else:
# 根据Q表选择最优动作
action = agent.policy[state_idx]
grid[i, j] = action_symbols[action]
# 打印策略网格
print("\n最优策略：")
for row in grid:
print(' '.join([f"{cell:^3}" for cell in row]))
def plot_training_history(rewards, steps):
"""绘制训练历史曲线（奖励和步数随轮次的变化）"""
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))
# 绘制奖励曲线
ax1.plot(rewards)
ax1.set_title('Episode Rewards')
ax1.set_xlabel('Episode')
ax1.set_ylabel('Total Reward')
# 绘制步数曲线
ax2.plot(steps)
ax2.set_title('Episode Steps')
ax2.set_xlabel('Episode')
ax2.set_ylabel('Steps')
plt.tight_layout() # 调整布局
plt.show()
if __name__ == "__main__":
# 创建悬崖漫步环境（12列×4行）
env = CliffWalkingEnv(ncol=12, nrow=4)
# 初始化Sarsa智能体
agent = SarsaAgent(env, alpha=0.5, gamma=0.9, epsilon=0.1)
# 训练智能体500轮
print("开始训练Sarsa智能体...")
rewards, steps = agent.train(episodes=500)
# 可视化最优策略和训练历史
visualize_policy(agent, env)
plot_training_history(rewards, steps)6.1.2、实验结果
6.2、多步Sarsa算法
6.2.1、完整代码
"""
文件名: 5.2
作者: 墨尘
日期: 2025/7/22
项目名: d2l_learning
备注: 多步Sarsa算法在悬崖漫步环境中的实现与可视化
"""
import numpy as np
import matplotlib.pyplot as plt
class CliffWalkingEnv:
"""悬崖漫步环境（确定性动作）"""
def __init__(self, ncol=12, nrow=4):
self.ncol = ncol # 环境宽度（列数）
self.nrow = nrow # 环境高度（行数）
self.reset() # 重置环境状态到起点
self.cliff = [(nrow - 1, j) for j in range(1, ncol - 1)] # 悬崖位置
self.goal = (nrow - 1, ncol - 1) # 目标位置：右下角
def reset(self):
self.state = (self.nrow - 1, 0) # 起点：左下角(3,0)
return self._state_to_index(self.state)
def step(self, action):
# 动作编码：0=上, 1=右, 2=下, 3=左
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)
# 奖励与终止条件
if self.state == self.goal:
reward = 0
done = True
elif self.state in self.cliff:
reward = -100
done = False
self.state = (self.nrow - 1, 0) # 回起点
else:
reward = -1
done = False
return self._state_to_index(self.state), reward, done, {}
def _state_to_index(self, state):
return state[0] * self.ncol + state[1]
class NStepSarsaAgent:
"""多步Sarsa算法实现"""
def __init__(self, env, n=3, alpha=0.5, gamma=0.9, epsilon=0.1):
self.env = env
self.n = n # 多步Sarsa的步数（核心参数）
self.n_states = env.nrow * env.ncol
self.n_actions = 4 # 上右下左
self.alpha = alpha # 学习率
self.gamma = gamma # 折扣因子
self.epsilon = epsilon # ε-贪婪策略
self.Q = np.zeros((self.n_states, self.n_actions)) # Q表
def epsilon_greedy_policy(self, state):
"""ε-贪婪策略：平衡探索与利用"""
if np.random.random() < self.epsilon:
return np.random.randint(0, self.n_actions) # 随机探索
else:
return np.argmax(self.Q[state]) # 利用最优动作
def update(self, trajectory, tau):
"""根据轨迹更新Q值（核心：计算n步回报）"""
# trajectory格式：[(s0,a0,r0), (s1,a1,r1), ..., (st,at,rt)]
# tau：待更新的时间步（t-n ≤ tau < t）
G = 0.0 # n步回报
# 累积n步奖励：R_{tau+1} + γR_{tau+2} + ... + γ^{n-1}R_{tau+n}
for i in range(tau + 1, min(tau + self.n + 1, len(trajectory))):
# 奖励的时间步为i，折扣因子为γ^(i - tau - 1)
G += (self.gamma ** (i - tau - 1)) * trajectory[i][2]
# 若未到轨迹终点，加上γ^n * Q(S_{tau+n}, A_{tau+n})
if tau + self.n < len(trajectory) - 1:
s_n = trajectory[tau + self.n][0] # S_{tau+n}
a_n = trajectory[tau + self.n][1] # A_{tau+n}
G += (self.gamma ** self.n) * self.Q[s_n, a_n]
# 更新Q值：Q(S_tau, A_tau) += α[G - Q(S_tau, A_tau)]
s_tau = trajectory[tau][0]
a_tau = trajectory[tau][1]
self.Q[s_tau, a_tau] += self.alpha * (G - self.Q[s_tau, a_tau])
def train(self, episodes=500):
"""训练智能体（多步Sarsa核心逻辑）"""
rewards_history = []
steps_history = []
for episode in range(episodes):
trajectory = [] # 存储轨迹：[(s0,a0,r0), (s1,a1,r1), ...]
state = self.env.reset() # 初始状态
action = self.epsilon_greedy_policy(state) # 初始动作
trajectory.append((state, action, 0.0)) # 初始奖励为0
done = False
total_reward = 0
steps = 0
while not done:
# 执行动作，获取下一状态和奖励
next_state, reward, done, _ = self.env.step(action)
steps += 1
total_reward += reward
# 选择下一动作
next_action = self.epsilon_greedy_policy(next_state)
trajectory.append((next_state, next_action, reward))
# 更新时间步（当前已执行steps步，时间步t=steps）
t = steps
# 待更新的时间步tau：从t-n开始，直到t-1
tau = t - self.n
if tau >= 0:
self.update(trajectory, tau) # 用n步回报更新Q值
# 转移到下一状态和动作
state, action = next_state, next_action
# 处理轨迹末尾剩余的tau（确保所有时间步都被更新）
while tau < len(trajectory) - 1:
tau += 1
if tau >= self.n:
self.update(trajectory, tau)
rewards_history.append(total_reward)
steps_history.append(steps)
if (episode + 1) % 100 == 0:
print(f"Episode {episode + 1}/{episodes}, "
f"Total Reward: {total_reward}, "
f"Steps: {steps}")
# 提取最优策略
self.policy = np.argmax(self.Q, axis=1)
return rewards_history, steps_history
def visualize_policy(agent, env):
"""可视化最优策略"""
action_symbols = ['↑', '→', '↓', '←']
grid = np.empty((env.nrow, env.ncol), dtype=object)
for i in range(env.nrow):
for j in range(env.ncol):
state_idx = i * env.ncol + j
if (i, j) == env.goal:
grid[i, j] = 'G'
elif (i, j) in env.cliff:
grid[i, j] = 'C'
else:
action = agent.policy[state_idx]
grid[i, j] = action_symbols[action]
print("\n最优策略：")
for row in grid:
print(' '.join([f"{cell:^3}" for cell in row]))
def plot_training_history(rewards, steps):
"""绘制训练曲线"""
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))
ax1.plot(rewards)
ax1.set_title('Episode Rewards (n-step Sarsa)')
ax1.set_xlabel('Episode')
ax1.set_ylabel('Total Reward')
ax2.plot(steps)
ax2.set_title('Episode Steps (n-step Sarsa)')
ax2.set_xlabel('Episode')
ax2.set_ylabel('Steps')
plt.tight_layout()
plt.show()
if __name__ == "__main__":
# 创建环境
env = CliffWalkingEnv(ncol=12, nrow=4)
# 初始化多步Sarsa智能体（n=3步）
agent = NStepSarsaAgent(env, n=3, alpha=0.5, gamma=0.9, epsilon=0.1)
# 训练500轮
print("开始训练多步Sarsa智能体...")
rewards, steps = agent.train(episodes=500)
# 可视化结果
visualize_policy(agent, env)
plot_training_history(rewards, steps)6.2.2、实验结果
更多推荐
15
0- 0
已为社区贡献5条内容
所有评论(0)