【超简单之强化学习入门】基于Sarsa表格方法的出租车调度

!pip install -U pip --user >log.log
!pip install gym >log.log

!pip list|grep gym

gym                    0.12.1

+---------+
|R: | : :G|
| : | : : |
| : : : : |
| | : | : |
|Y| : |B: |

# -*- coding: utf-8 -*-
import numpy as np
class SarsaAgent(object):
    def __init__(self,
                 obs_n,
                 act_n,
                 learning_rate=0.01,
                 gamma=0.9,
                 e_greed=0.1):
        self.act_n = act_n  # 动作维度，有几个动作可选
        self.lr = learning_rate  # 学习率
        self.gamma = gamma  # reward的衰减率
        self.epsilon = e_greed  # 按一定概率随机选动作
        self.Q = np.zeros((obs_n, act_n))
    # 根据输入观察值，采样输出的动作值，带探索
    def sample(self, obs):
        if np.random.uniform(0, 1) < (1.0 - self.epsilon):  #根据table的Q值选动作
            action = self.predict(obs)
        else:
            action = np.random.choice(self.act_n)  #有一定概率随机探索选取一个动作
        return action
    # 根据输入观察值，预测输出的动作值
    def predict(self, obs):
        Q_list = self.Q[obs, :]
        maxQ = np.max(Q_list)
        action_list = np.where(Q_list == maxQ)[0]  # maxQ可能对应多个action
        action = np.random.choice(action_list)
        return action
    # 学习方法，也就是更新Q-table的方法
    def learn(self, obs, action, reward, next_obs, next_action, done):
        """ on-policy
            obs: 交互前的obs, s_t
            action: 本次交互选择的action, a_t
            reward: 本次动作获得的奖励r
            next_obs: 本次交互后的obs, s_t+1
            next_action: 根据当前Q表格, 针对next_obs会选择的动作, a_t+1
            done: episode是否结束
        """
        predict_Q = self.Q[obs, action]
        if done:
            target_Q = reward  # 没有下一个状态了
        else:
            target_Q = reward + self.gamma * self.Q[next_obs,
                                                    next_action]  # Sarsa
        self.Q[obs, action] += self.lr * (target_Q - predict_Q)  # 修正q
    def save(self):
        npy_file = './q_table.npy'
        np.save(npy_file, self.Q)
        print(npy_file + ' saved.')
    def restore(self, npy_file='./q_table.npy'):
        self.Q = np.load(npy_file)
        print(npy_file + ' loaded.')

# -*- coding: utf-8 -*-
import gym
import numpy as np
np.random.seed(0)
import gym
from agent import SarsaAgent
import time
def run_episode(env, agent, is_render=False):
    total_steps = 0  # 记录每个episode走了多少step
    total_reward = 0
    obs = env.reset()  # 重置环境, 重新开一局（即开始新的一个episode）
    action = agent.sample(obs)  # 根据算法选择一个动作
    while True:
        next_obs, reward, done, _ = env.step(action)  # 与环境进行一个交互
        next_action = agent.sample(next_obs)  # 根据算法选择一个动作
        # 训练 Sarsa 算法
        agent.learn(obs, action, reward, next_obs, next_action, done)
        action = next_action
        obs = next_obs  # 存储上一个观察值
        total_reward += reward
        total_steps += 1  # 计算step数
        if is_render:
            env.render()  # 渲染新的一帧图形
        if done:
            break
    return total_reward, total_steps
def test_episode(env, agent):
    total_reward = 0
    obs = env.reset()
    while True:
        action = agent.predict(obs)  # greedy
        next_obs, reward, done, _ = env.step(action)
        total_reward += reward
        obs = next_obs
        time.sleep(0.5)
        env.render()
        if done:
            print('test reward = %.1f' % (total_reward))
            break
def main():
    # 环境使用
    env = gym.make('Taxi-v3')
    env.reset(seed=1024)
    print('观察空间 = {}'.format(env.observation_space.n))
    print('动作空间 = {}'.format(env.action_space.n))
    print('状态数量 = {}'.format(env.observation_space.n))
    print('动作数量 = {}'.format(env.action_space.n))
    state = env.reset()
    taxirow, taxicol, passloc, destidx = env.unwrapped.decode(state)
    print(taxirow, taxicol, passloc, destidx)
    print('的士位置 = {}'.format((taxirow, taxicol)))
    print('乘客位置 = {}'.format(env.unwrapped.locs[passloc]))
    print('目标位置 = {}'.format(env.unwrapped.locs[destidx]))
    env.render()
    env.step(0)
    env.render()
    # agent定义
    agent = SarsaAgent(obs_n=env.observation_space.n, act_n=env.action_space.n)
    # 训练
    episodes = 3000
    episode_rewards = []
    is_render = False
    # 开始训练
    for episode in range(episodes):
        ep_reward, ep_steps = run_episode(env, agent, is_render)
        print('Episode %s: steps = %s , reward = %.1f' % (episode, ep_steps,
                                                          ep_reward))
        # 每隔20个episode渲染一下看看效果
        if episode % 20 == 0:
            is_render = True
        else:
            is_render = False
    # 训练结束，查看算法效果
    test_episode(env, agent)
if __name__ == "__main__":
    main()

Episode 1493: steps = 200 , reward = -290.0
Episode 1494: steps = 200 , reward = -344.0
Episode 1495: steps = 113 , reward = -110.0
Episode 1496: steps = 19 , reward = 2.0
Episode 1497: steps = 10 , reward = 2.0
Episode 1498: steps = 118 , reward = -169.0
Episode 1499: steps = 108 , reward = -87.0
Episode 1500: steps = 200 , reward = -281.0