easy-rl/notebooks/Value Iteration/value_iteration.ipynb

值迭代算法

作者：stzhao github: https://github.com/zhaoshitian

一、定义环境

In [30]:

import numpy as np
import sys,os
curr_path = os.path.abspath('')
parent_path = os.path.dirname(curr_path)
sys.path.append(parent_path)
from envs.simple_grid import DrunkenWalkEnv

In [63]:

def all_seed(env,seed = 1):
    ## 这个函数主要是为了固定随机种子
    import numpy as np
    import random
    import os
    env.seed(seed) 
    np.random.seed(seed)
    random.seed(seed)
    os.environ['PYTHONHASHSEED'] = str(seed) 

In [32]:

env = DrunkenWalkEnv(map_name="theAlley")
all_seed(env, seed = 1) # 设置随机种子为1

二、价值迭代算法

In [40]:

def value_iteration(env, theta=0.005, discount_factor=0.9):
    Q = np.zeros((env.nS, env.nA)) # 初始化一个Q表格
    count = 0
    while True:
        delta = 0.0
        Q_tmp = np.zeros((env.nS, env.nA))
        for state in range(env.nS):
            for a in range(env.nA):
                accum = 0.0
                reward_total = 0.0
                for prob, next_state, reward, done in env.P[state][a]:
                    accum += prob* np.max(Q[next_state, :])
                    reward_total += prob * reward
                Q_tmp[state, a] = reward_total + discount_factor * accum
                delta = max(delta, abs(Q_tmp[state, a] - Q[state, a]))
        Q = Q_tmp
        
        count += 1
        if delta < theta or count > 100: # 这里设置了即使算法没有收敛，跑100次也退出循环
            break 
    return Q

In [41]:

Q = value_iteration(env)
print(Q)

[[2.25015697e+22 2.53142659e+22 4.50031394e+22 2.53142659e+22]
 [2.81269621e+22 5.41444021e+22 1.01257064e+23 5.41444021e+22]
 [6.32856648e+22 1.21824905e+23 2.27828393e+23 1.21824905e+23]
 [1.42392746e+23 2.74106036e+23 5.12613885e+23 2.74106036e+23]
 [3.20383678e+23 5.76690620e+23 1.15338124e+24 5.76690620e+23]
 [7.20863276e+23 1.38766181e+24 2.59510779e+24 1.38766181e+24]
 [1.62194237e+24 3.12223906e+24 5.83899253e+24 3.12223906e+24]
 [3.64937033e+24 7.02503789e+24 1.31377332e+25 7.02503789e+24]
 [8.21108325e+24 1.47799498e+25 2.95598997e+25 1.47799498e+25]
 [1.84749373e+25 3.55642543e+25 6.65097743e+25 3.55642543e+25]
 [4.15686089e+25 8.00195722e+25 1.49646992e+26 8.00195722e+25]
 [9.35293701e+25 1.80044037e+26 3.36705732e+26 1.80044037e+26]
 [5.89235032e+26 7.36543790e+26 7.57587898e+26 7.36543790e+26]]

In [69]:

policy = np.zeros([env.nS, env.nA]) # 初始化一个策略表格
for state in range(env.nS):
    best_action = np.argmax(Q[state, :]) #根据价值迭代算法得到的Q表格选择出策略
    policy[state, best_action] = 1

policy = [int(np.argwhere(policy[i]==1)) for i in range(env.nS) ]
print(policy)

[2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2]

三、测试

In [78]:

num_episode = 1000 # 测试1000次
def test(env,policy):
    
    rewards = []  # 记录所有回合的奖励
    success = []  # 记录该回合是否成功走到终点
    for i_ep in range(num_episode):
        ep_reward = 0  # 记录每个episode的reward
        state = env.reset()  # 重置环境, 重新开一局（即开始新的一个回合） 这里state=0
        while True:
            action = policy[state]  # 根据算法选择一个动作
            next_state, reward, done, _ = env.step(action)  # 与环境进行一个交互
            state = next_state  # 更新状态
            ep_reward += reward
            if done:
                break
        if state==12: # 即走到终点
            success.append(1)
        else:
            success.append(0)
        rewards.append(ep_reward)
    acc_suc = np.array(success).sum()/num_episode
    print("测试的成功率是：", acc_suc)
    

In [79]:

test(env, policy)

测试的成功率是： 0.64