本文尝试解析Python实现的Policy Iteration算法，代码来自Github某大神的库– Github地址。其实现代码解决了下图中方块行走的问题，即控制红色方块，走到蓝色球的位置上算通关，碰到绿色三角要减分。

算法描述

Policy Iteration直译成中文是策略迭代，言下之意就是通过不停的更新策略使策略达到最优解。Policy Iteration算法的主要步骤分为2步，先是执行Policy Evaluation（策略评估），然后执行Policy Improve（策略改进），然后通过循环以上两个步骤得出最优解。整个过程中涉及到的概念有：

代理（agent）

即受控制的物件，在本例子中是红色方块。

状态（state）

即应用场景的当前状态，在这个问题中指的是红色方块的当前位置坐标，如上图的状态为(0,0)。

策略（policy）

当前状态选择了某种action即是一种策略。此处代码中self.policy_table记录了所有状态下的策略。

动作（action）

从字面意思就知道是一个操作，在此处可以是上下左右移动。本例子的action设置：

POSSIBLE_ACTIONS = [0, 1, 2, 3] # up, down, left, right ACTIONS = [(-1, 0), (1, 0), (0, -1), (0, 1)] # actions in coordinates

奖励（reward）

agent达到某个状态的奖励，在本例子中，达到蓝色圆形位置奖励为1，达到绿色三角处奖励为-1：

self.reward[2][2] = 1 # reward 1 for circle self.reward[1][2] = -1 # reward -1 for triangle self.reward[2][1] = -1 # reward -1 for triangle

价值（value）

即agent执行策略后达到某个状态的价值，每个状态都对应一个价值。

衰减因子（discount_factor）

即在某个状态下执行策略达到下一个状态，下个状态的策略价值对前一个状态的策略价值的影响系数，本例子为0.9。

核心代码

def policy_evaluation(self): next_value_table = [[0.00] * self.env.width for _ in range(self.env.height)] # Bellman Expectation Equation for the every states for state in self.env.get_all_states(): value = 0.0 ... for action in self.env.possible_actions: next_state = self.env.state_after_action(state, action) reward = self.env.get_reward(state, action) next_value = self.get_value(next_state) value += (self.get_policy(state)[action] * (reward + self.discount_factor * next_value)) next_value_table[state[0]][state[1]] = round(value, 2) self.value_table = next_value_table

上面的代码就是上文算法描述中提到的Policy Evaluation的实现（为了方便理解我去掉了不必要的代码，大家自己阅读源码的时候结合上下文看就知道为什么了），其作用是评估当前的策略，评估结果记录到next_value_table变量中，并最终赋值给self.value_table，更新当前策略价值。从注释中可以看出评估策略价值使用的方法是贝尔曼方程，其核心为下面这段。它将当前状态执行可能发生的action之后得到的奖励值reward和下一个状态的策略价值next_value乘以一个衰减因子discount_factor的和乘以当前状态下策略发生此action的概率，再对所有可能发生的action执行上面的操作进行求和。

for action in self.env.possible_actions: value += (self.get_policy(state)[action] * (reward + self.discount_factor * next_value))

然后就是改进策略Policy Improvement的代码了：

def policy_improvement(self): next_policy = self.policy_table for state in self.env.get_all_states(): value = -99999 max_index = [] result = [0.0, 0.0, 0.0, 0.0] # initialize the policy # for every actions, calculate # [reward + (discount factor) * (next state value function)] for index, action in enumerate(self.env.possible_actions): next_state = self.env.state_after_action(state, action) reward = self.env.get_reward(state, action) next_value = self.get_value(next_state) temp = reward + self.discount_factor * next_value # We normally can't pick multiple actions in greedy policy. # but here we allow multiple actions with same max values if temp == value: max_index.append(index) elif temp > value: value = temp max_index.clear() max_index.append(index) # probability of action prob = 1 / len(max_index) for index in max_index: result[index] = prob next_policy[state[0]][state[1]] = result self.policy_table = next_policy

上面的函数将策略价值记录到next_policy变量中，并最终更新的到self.policy_table中。算法使用的是贪心策略，即评估当前策略价值并记录能够获取最大价值的action的index，最后更新策略表，将概率平分到能够获取到最大策略价值的action上。