import gymnasium as gym
import numpy as np
import json
env = gym.make('CliffWalking-v1')
N_STATES = env.observation_space.n
N_ACTIONS = env.action_space.n
GAMMA = 0.99
def sarsa(env, n_episodes=20000, alpha=0.1, epsilon=0.1, gamma=GAMMA):
Q = np.zeros((N_STATES, N_ACTIONS))
rewards_history = []
for ep in range(n_episodes):
state, _ = env.reset()
if np.random.random() < epsilon:
action = env.action_space.sample()
else:
action = int(np.argmax(Q[state]))
total_reward = 0
done = False
while not done:
next_state, reward, terminated, truncated, _ = env.step(action)
# 选择下一个动作(同策略)
if np.random.random() < epsilon:
next_action = env.action_space.sample()
else:
next_action = int(np.argmax(Q[next_state]))
# SARSA更新
target = reward + gamma * Q[next_state, next_action] * (1 - terminated)
Q[state, action] += alpha * (target - Q[state, action])
state = next_state
action = next_action
total_reward += reward
done = terminated or truncated
rewards_history.append(total_reward)
return Q, rewards_history
def q_learning(env, n_episodes=20000, alpha=0.1, epsilon=0.1, gamma=GAMMA):
Q = np.zeros((N_STATES, N_ACTIONS))
rewards_history = []
for ep in range(n_episodes):
state, _ = env.reset()
total_reward = 0
done = False
while not done:
if np.random.random() < epsilon:
action = env.action_space.sample()
else:
action = int(np.argmax(Q[state]))
next_state, reward, terminated, truncated, _ = env.step(action)
best_next = np.max(Q[next_state])
Q[state, action] += alpha * (reward + gamma * best_next * (1 - terminated) - Q[state, action])
state = next_state
total_reward += reward
done = terminated or truncated
rewards_history.append(total_reward)
return Q, rewards_history
# 运行对比
Q_sarsa, rewards_sarsa = sarsa(env)
Q_ql, rewards_ql = q_learning(env)
# 测试两种策略
def test_policy(Q, env, n_episodes=1000, epsilon=0.1):
total_rewards = []
cliff_falls = 0
for ep in range(n_episodes):
s, _ = env.reset()
done = False
total_r = 0
while not done:
if np.random.random() < epsilon:
a = env.action_space.sample()
else:
a = int(np.argmax(Q[s]))
s, r, terminated, truncated, _ = env.step(a)
total_r += r
if r == -100:
cliff_falls += 1
done = terminated or truncated
total_rewards.append(total_r)
return np.mean(total_rewards), cliff_falls
sarsa_avg, sarsa_falls = test_policy(Q_sarsa, env)
ql_avg, ql_falls = test_policy(Q_ql, env)
window = 500
sarsa_smooth = [np.mean(rewards_sarsa[max(0,i-window):i+1]) for i in range(len(rewards_sarsa))]
ql_smooth = [np.mean(rewards_ql[max(0,i-window):i+1]) for i in range(len(rewards_ql))]
print("=== SARSA vs Q-Learning 对比 ===")
print(f"{'指标':<20} | {'SARSA':>10} | {'Q-Learning':>10}")
print("-" * 45)
print(f"{'测试平均奖励':<20} | {sarsa_avg:>10.2f} | {ql_avg:>10.2f}")
print(f"{'悬崖掉落次数':<20} | {sarsa_falls:>10d} | {ql_falls:>10d}")
print(f"{'最后500回合奖励':<20} | {np.mean(rewards_sarsa[-500:]):>10.2f} | {np.mean(rewards_ql[-500:]):>10.2f}")
print("\\nSARSA学到更安全的路径(远离悬崖)，Q-Learning学到更短但有风险的最优路径")
result = {
"sarsa": {"avg_reward": round(float(sarsa_avg),2), "cliff_falls": sarsa_falls},
"q_learning": {"avg_reward": round(float(ql_avg),2), "cliff_falls": ql_falls},
"sarsa_smooth_10": [round(sarsa_smooth[i], 2) for i in range(0, len(sarsa_smooth), len(sarsa_smooth)//10)],
"ql_smooth_10": [round(ql_smooth[i], 2) for i in range(0, len(ql_smooth), len(ql_smooth)//10)]
}
with open("/var/www/ttl/rl/lesson08_result.json", "w") as f:
json.dump(result, f)
print("✅验证通过 - SARSA学得保守安全策略，Q-Learning学得最优但冒险的策略")
env.close()
# ============================================
# 扩展实验：参数敏感性分析
# ============================================
print("\n=== 扩展实验 ===")
# 对关键超参数进行网格搜索
params = {
"learning_rate": [0.001, 0.01, 0.1],
"epsilon": [0.05, 0.1, 0.2],
"gamma": [0.9, 0.95, 0.99]
}
print("超参数搜索空间:")
for k, v in params.items():
print(f" {k}: {v}")
print("共{}种组合".format(1))
for k, v in params.items():
print(f" {k}: {len(v)}种选择")
total = 1
for k, v in params.items():
total *= len(v)
print(f"总计: {total}种超参数组合")
print("扩展实验框架验证成功 - ✅")
📝 算法伪代码：SARSA
SARSA核心步骤:
1. 初始化参数/网络
2. FOR episode = 1 TO N:
3. 初始化环境状态 s
4. WHILE NOT done:
5. 根据当前策略选择动作 a
6. 执行动作, 观察奖励 r 和新状态 s'
7. 存储经验 (s, a, r, s')
8. 采样mini-batch更新参数
9. s = s'
10. END WHILE
11. 更新探索率/目标网络(如适用)
12. END FOR
13. RETURN 训练好的策略/值函数