Gymnasium 自定义环境

最近因为项目需要接触到 Gymnasium 这个开源强化学习环境库。它提供的环境 API 很规范，对于自己自定义的强化学习环境有很好的参考价值。因此，写这篇博文来详细介绍 Gymnasium 环境框架。

简介

gymnasium 的 Env 基类提供以下函数用于与环境交互，自定义时需要重写：

step(action) -> obs, reward, terminated, truncated, info 使用动作更新环境，其返回下一个智能体观测、采取该动作获得的奖励、环境是否因最简的动作而终止或截断，以及来自该环境的关于该步骤的信息
reset(*, seed, options) -> obs, info 将环境重置为初始内部状态，返回初始观测和信息。

除了 step 和 reset 这两个核心函数，还有 render() 渲染智能体的可视化的内容以及 close() 关闭环境。

另外还要定义 action_space 确定动作数量，observation_space 确定返回观测数据的数量。

一个实现

项目学习需要实现简单的 2 相位单路口的仿真环境，为了追求规范性，所以采用继承 gymnasium.Env 基类构建环境。
两相位单路口分东西向和南北向，直接就对车辆进入路口的随机量而不再考虑复杂的物理建模。
首先确定输入的动作有两个：南北向绿灯或者东西向绿灯，仅此而已，非红即绿。观测的状态就是两向的等待车辆数以及当前的相位。等待车辆数仅为 0 ~ 4，5 种状态。

以此实现构造函数： __init__(self)：

1
2
3
4
5
6


def __init__(self) -> None:
    super().__init__()
    self.action_space = spaces.Discrete(2)  # 动作：0/1
    self.observation_space = spaces.MultiDiscrete([5, 5, 2])  # 状态：(5, 5, 2)
    self.phase = 0  # 当前绿灯相位 0/1
    self.queue = [0, 0]  # 两个方向的排队分档

两向的车辆数实际是存在 self.queue 内，而 self.observation_space 为符合 gym API 规定，但目前还不是很清楚，该环境也用不上。

再实现一个环境初始化函数，主要是对 phase 和 queue 复位，还要迎合 gym API 传入种子之类。

1
2
3
4
5


    def reset(self, *, seed=None, options=None):
        super().reset(seed=seed)  # 初始化 self.np_random
        self.phase = 0
        self.queue = [0, 0]
        return self._obs(), {}

gymnasium 使用 numpy 的随机数生成器，形成函数 np_random()，该函数会返回一个 numpy 随机数生成器。

接下来是 step() 要实现对环境的交互，传入动作输出观测值。函数内部做一步环境的更新，对于单路口仿真环境就是随机进入的车辆以及放行绿灯向的车辆，还有奖励和终止状态的判断。

step 这里实现按：车辆到达->决策相位->绿灯放行->奖励 + 终止状态进行。

对于观测值的返回额外实现了函数，观测值主要是两向的车辆数以及当前绿灯相位。

1
2


def _obs(self):
    return np.array([self.queue[0], self.queue[1], self.phase], dtype=int)

所以 step 可以做如下实现：

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


    def step(self, action):
        # 车辆到达
        for i in range(2):
            self.queue[i] = min(4, int(self.queue[i] + self.np_random.integers(0, 2)))

        # 决策相位
        if action == 1:
            self.phase = 1 - self.phase

        # 绿灯放行
        self.queue[self.phase] = max(0, self.queue[self.phase] - 1)

        # 奖励 + 终止状态
        reward = sum(self.queue) * -1
        terminated = False # 终止
        truncated = False  # 截断
        info = {} # 附加调试信息字典

        return self._obs(), reward, terminated, truncated, info

一般环境实例化对象为 env，智能体(强化学习模型)实例化对象为 agent。
智能体一般会实现 choose_action 和 update 方法与环境进行交互，大致是智能体选择一个动作给环境做一次交互，环境返回观测值给智能体更新这样一个过程：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


env = TwoPhaseIntersection()
agent = QLearning()

for ep in range(5001):
    obs, _ = env.reset()
    total = 0
    for _ in range(100):
        action = agent.choose_action(obs)
        next_obs, reward, terminated, truncated, _ = env.step(action)
        agent.update(obs, action, reward, next_obs)
        obs = next_obs
        total += reward
        if terminated or truncated:
            break

    agent.delay_espilon()

简介

一个实现

参考资料