1
2
3

conda create --name rl python=3.12 # 创建虚拟环境
conda activate rl # 切换环境
pip install "gymnasium[all]" # 安装 gymnasium，国内环境可以加一个加速镜像源 -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 对于这个案例，你可以使用 `pip install "gymnasium[classic-control]"` 安装 gym 需要的特性（像我一样 all 了之后可以忽略这行）
import gymnasium as gym

# 创建我们的训练环境 —— 一个小车顶着一个杆子，要平衡杆子不让它倒下
env = gym.make("CartPole-v1", render_mode="human")

# 重置环境到起始状态，每次 env.step 前都一定要 reset 哦
observation, info = env.reset()
# observation 是智能体从环境中能获得的信息，比如小车位置、小车速度、杆子角度、……
# info 提供了一些额外 debug 信息，在我们入门的旅程中，不需要在意它

print(f"Starting observation: {observation}")
# 一个可能的输出: [ 0.01234567 -0.00987654  0.02345678  0.01456789]
# 对应： [小车位置, 小车速度, 杆子角度, 杆子旋转角速度]

episode_over = False
total_reward = 0

while not episode_over:
    # 从所有可能的动作中选一个: 0 = 小车向左, 1 = 小车向右
    action = env.action_space.sample()  # 目前先随机一个动作吧

    # 采取这个动作，看看会发生什么
    observation, reward, terminated, truncated, info = env.step(action)

    # 在这个案例中：杆子每保持直立一步，reward 就增加 1
    # 当杆子落得太远，terminated 会置为真 (智能体失败了)
    # 当到达时间限制的时候 truncated 会置 (默认 500 步)

    total_reward += reward
    episode_over = terminated or truncated

print(f"Episode finished! Total reward: {total_reward}")
env.close()

1
2
3
4
5
6
7
8
9

import gymnasium as gym
env = gym.make(
    'FrozenLake-v1',
    desc=None,
    map_name="4x4",
    is_slippery=False,
    reward_schedule=(10,-10,-1),
    render_mode="human"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

from collections import defaultdict
import gymnasium as gym
import numpy as np
class FrozenLakeAgent:
    def __init__(
        self,
        env: gym.Env,
        learning_rate: float = 0.1,
        initial_epsilon: float = 1,
        epsilon_decay: float = 0.999,
        final_epsilon: float = 0.1,
        discount_factor: float = 1,
    ):
        """初始化一个 Q-Learning 智能体
        
        参数:
            env: gym 环境
            learning_rate: 学习率
            initial_epsilon: 初始探索率
            epsilon_decay: 探索率衰减
            final_epsilon: 最终探索率
            discount_factor: 折扣因子
        """
        self.env = env
        # Q 值表，使用 defaultdict 初始化为 0 以便自动初始化未见过的状态
        self.qValues = defaultdict(lambda: np.zeros(env.action_space.n))
        
        self.lr = learning_rate # 学习率
        self.discount_factor = discount_factor # 折扣因子
        
        # 探索率相关参数
        self.epsilon = initial_epsilon
        self.epsilon_decay = epsilon_decay
        self.final_epsilon = final_epsilon
        
        # 跟踪训练过程
        self.training_error = []
    
    def get_action(self, obsv: int) -> int:
        """基于 ε-贪婪策略选择动作
        
        参数:
            obsv: 当前观察到的状态
            
        返回:
            选择的动作
        """
        if np.random.rand() < self.epsilon:
            return self.env.action_space.sample()  # 探索Explore：随机选择动作
        else:
            return np.argmax(self.qValues[obsv])  # 利用Exploit：选择 Q 值最高的动作
    
    def update(
        self,
        obsv: int,
        action: int,
        reward: float,
        terminated: bool,
        next_obsv: int
    ):
        """使用 Q-Learning 更新 Q 值
        
        参数:
            obsv: 当前观察到的状态
            action: 采取的动作
            reward: 获得的奖励
            terminated: 是否终止
            next_obsv: 下一个观察到的状态
        """
        current_q = self.qValues[obsv][action]
        max_future_q = np.max(self.qValues[next_obsv]) if not terminated else 0
        # Q-Learning 更新公式
        new_q = (1 - self.lr) * current_q + self.lr * (reward + self.discount_factor * max_future_q)
        self.qValues[obsv][action] = new_q
        
        # 记录训练误差
        self.training_error.append(abs(new_q - current_q))
    
    def decay_epsilon(self):
        """衰减探索率 ε"""
        self.epsilon = max(self.final_epsilon, self.epsilon * self.epsilon_decay) if self.epsilon > self.final_epsilon else self.epsilon

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

n_episodes = 10000
env = gym.make(
    'FrozenLake-v1',
    desc=None,
    is_slippery=False,
    success_rate=0.9,
    reward_schedule=(10,-10,-1),
    map_name="4x4"
)
env = gym.wrappers.RecordEpisodeStatistics(env, buffer_length=n_episodes)
observation, info = env.reset()
tot_reward = 0

agent = FrozenLakeAgent(env)

from tqdm import tqdm
for episode in tqdm(range(n_episodes)):
    observation, info = env.reset()
    done = False
    while not done:
        action = agent.get_action(observation)
        next_observation, reward, terminated, truncated, info = env.step(action)
        agent.update(observation, action, reward, terminated, next_observation)
        done = terminated or truncated
        observation = next_observation
    agent.decay_epsilon()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

total_rewards = []
old_epsilon = agent.epsilon
agent.epsilon = 0.0  # 禁用探索以测试学习到的策略

for _ in range(100):
    obs, info = env.reset()
    episode_reward = 0
    done = False
    while not done:
        action = agent.get_action(obs)
        obs, reward, terminated, truncated, info = env.step(action)
        episode_reward += reward
        done = terminated or truncated
    total_rewards.append(episode_reward)

win_rate = np.mean(np.array(total_rewards) > 0)
print(f"Win rate over 100 test episodes: {win_rate * 100}%")
print(f"Average reward over 100 test episodes: {np.mean(total_rewards)}")
print(f"Standard deviation of reward over 100 test episodes: {np.std(total_rewards)}")
# 下面是展示训练得到的 qValues
print(f"Trained Q-values:")
for state, actions in agent.qValues.items():
    print(f"State {state}: {actions}")