强化学习-值迭代代码

1、构建一个初始值向量 values，为所有状态的大小

values = np.zeros([N_STATES])

2、求出每个状态所有动作的收益，找出在状态s下，四个动作的最大收益，赋给value[s]

以状态s为例，
状态s下，所有动作的最大收益赋给V[1]

原理：

# s为当前状态，s1为下一状态
values[s] = max([ sum([P_a[s, s1, a]*(rewards[s] + gamma*values_tmp[s1]) for s1 in range(N_STATES)]) for a in range(N_ACTIONS)])

这个代码看起来很复杂，简化为

A = np.zeros(N_ACTIONS)
for a in range(N_ACTIONS):
	for s1 in range(N_STATES):
        A[a] += P_a[s, s1, a]*(rewards[s] + gamma*values_tmp[s1]
best_action_value = np.max(A)

迭代更新，直到v收敛

import math
import numpy as np

#定义值迭代
def value_iteration(P_a, rewards, gamma, error=0.01, deterministic=True):
  """
  static value iteration function. Perhaps the most useful function in this repo
  
  inputs:
  #状态转移矩阵
    P_a         NxNxN_ACTIONS transition probabilities matrix - 
                              P_a[s0, s1, a] is the transition prob of 
                              landing at state s1 when taking action 
                              a at state s0
  #所有状态的回报函数
    rewards     Nx1 matrix - rewards for all the states
    #RL的折扣因子
    gamma       float - RL discount
    error       float - threshold for a stop
    deterministic   bool - to return deterministic policy or stochastic policy
  
  returns:
  #所有状态的值
    values    Nx1 matrix - estimated values
  #策略，以概率矩阵表示
    policy    Nx1 (NxN_ACTIONS if non-det) matrix - policy
  """
  N_STATES, _, N_ACTIONS = np.shape(P_a)

  # 构建了初始的q值 100 x 1
  values = np.zeros([N_STATES])
  
  # estimate values
  while True:
    values_tmp = values.copy()
    #对于所有状态
    for s in range(N_STATES):
      v_s = []
      #值迭代Qi+1=E[r+max(Qi)]
      values[s] = max([sum([P_a[s, s1, a]*(rewards[s] + gamma*values_tmp[s1]) for s1 in range(N_STATES)]) for a in range(N_ACTIONS)])
      #if s == 14 or s == 19 or s == 13 or s == 20 or s == 22:
        #values[s] = 0
    #当Q的变化很小时，停止迭代
    if max([abs(values[s] - values_tmp[s]) for s in range(N_STATES)]) < error:
      break

  #如果确定性
  if deterministic:
    # generate deterministic policy
    policy = np.zeros([N_STATES])
    for s in range(N_STATES):
      policy[s] = np.argmax([sum([P_a[s, s1, a]*(rewards[s]+gamma*values[s1]) 
                                  for s1 in range(N_STATES)]) 
                                  for a in range(N_ACTIONS)])

    return values, policy

  # 随机策略
  else:
    #不确定的
    #NxN_ACTIONS if non-det，按（s,a）来
    # generate stochastic policy
    policy = np.zeros([N_STATES, N_ACTIONS])
    for s in range(N_STATES):
      v_s = np.array([sum([P_a[s, s1, a]*(rewards[s] + gamma*values[s1])
                           for s1 in range(N_STATES)])
                           for a in range(N_ACTIONS)])
      policy[s,:] = np.transpose(v_s/np.sum(v_s))
    return values, policy