exp/1_gridworld.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Gridworld\n",
    "\n",
    "`通过这个实验，了解MDP的Dynamic Programming解法`\n",
    "\n",
    "所有的实验源代码都在`lib`目录下，来自[dennybritz](https://github.com/dennybritz/reinforcement-learning)，这里只做解读和归总。\n",
    "\n",
    "## 实验目录\n",
    "\n",
    "- [Gridworld](https://applenob.github.io/gridworld.html)：对应MDP的Dynamic Programming\n",
    "- [Blackjack](https://applenob.github.io/black_jack.html)：对应Model Free的Monte Carlo的Planning和Controlling\n",
    "- [Windy Gridworld](https://applenob.github.io/windy_gridworld.html)：对应Model Free的Temporal Difference的On-Policy Controlling，SARSA。\n",
    "- [Cliff Walking](https://applenob.github.io/cliff_walking.html)：对应Model Free的Temporal Difference的Off-Policy Controlling，Q-learning。\n",
    "- [Mountain Car](https://applenob.github.io/mountain_car.html)：对应Q-Learning with Linear Function Approximation。\n",
    "- [Atari](https://applenob.github.io/atari.html)：对应Deep-Q Learning。\n",
    "\n",
    "\n",
    "## 本文目录\n",
    "\n",
    "- [问题介绍](#问题介绍)\n",
    "- [Policy Evaluation](#Policy-Evaluation)\n",
    "- [Policy Iteration](#Policy-Iteration)\n",
    "- [Value Iteration](#Value-Iteration)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "## 问题介绍\n",
    "\n",
    "![](https://github.com/applenob/rl_learn/raw/master/res/grid_world.png)\n",
    "\n",
    "你的agent在一个M×N的格子世界里，想办法走到左上角或者右下角的终点。\n",
    "\n",
    "这个实验用来解释如何在MDP的环境里，使用动态规划（DP），来寻找最佳策略$\\pi_*$。方法是：`Policy Iteration`和`Value Iteration`。\n",
    "\n",
    "为什么说这个实验是MDP呢？回顾下MDP的五元组：$<\\textbf{S}, \\textbf{A}, \\textbf{P}, \\textbf{R}, \\gamma>$，分别对应：\n",
    "\n",
    "- $\\textbf{S}$：状态集合，上图一共有16种状态，对应16个格子。\n",
    "- $\\textbf{A}$：动作集合，上下左右四种。\n",
    "- $\\textbf{P}$：状态转移概率矩阵，即当前状态是$s$且动作是$a$时，下一个状态是$s+1$的概率。这里如果动作选定，下一个状态是唯一的。\n",
    "- $\\textbf{R}$：奖励函数，即在某一个状态$s$，下一时刻能收获的即时奖励的期望。这里假设除了最终状态，每到另一个状态，即时奖励都是-1。\n",
    "- $\\gamma$：折扣系数，题目没有明确给出。\n",
    "\n",
    "来总结一下：五元组中，$\\textbf{S}$和$\\textbf{R}$用来描述环境，即环境有什么样的状态，不同的状态下有什么回报。$\\textbf{A}$描述agent的可操作动作范围。$\\textbf{P}$最重要，它描述**agent和环境如何交互**，即，在某状态下采取某个动作下一步会进到什么状态。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```python\n",
    "import numpy as np\n",
    "import sys\n",
    "from gym.envs.toy_text import discrete\n",
    "\n",
    "# 定义动作\n",
    "UP = 0\n",
    "RIGHT = 1\n",
    "DOWN = 2\n",
    "LEFT = 3\n",
    "\n",
    "class GridworldEnv(discrete.DiscreteEnv):\n",
    "    \"\"\"\n",
    "    Grid World environment from Sutton's Reinforcement Learning book chapter 4.\n",
    "    You are an agent on an MxN grid and your goal is to reach the terminal\n",
    "    state at the top left or the bottom right corner.\n",
    "\n",
    "    For example, a 4x4 grid looks as follows:\n",
    "\n",
    "    T  o  o  o\n",
    "    o  x  o  o\n",
    "    o  o  o  o\n",
    "    o  o  o  T\n",
    "\n",
    "    x is your position and T are the two terminal states.\n",
    "\n",
    "    You can take actions in each direction (UP=0, RIGHT=1, DOWN=2, LEFT=3).\n",
    "    Actions going off the edge leave you in your current state.\n",
    "    You receive a reward of -1 at each step until you reach a terminal state.\n",
    "    \"\"\"\n",
    "\n",
    "    metadata = {'render.modes': ['human', 'ansi']}\n",
    "\n",
    "    def __init__(self, shape=[4,4]):\n",
    "        if not isinstance(shape, (list, tuple)) or not len(shape) == 2:\n",
    "            raise ValueError('shape argument must be a list/tuple of length 2')\n",
    "\n",
    "        self.shape = shape\n",
    "\n",
    "        nS = np.prod(shape)\n",
    "        nA = 4\n",
    "\n",
    "        MAX_Y = shape[0]\n",
    "        MAX_X = shape[1]\n",
    "\n",
    "        # 用一个字典表示状态转移概率\n",
    "        P = {}\n",
    "        grid = np.arange(nS).reshape(shape)\n",
    "        it = np.nditer(grid, flags=['multi_index'])\n",
    "\n",
    "        # 初始化P\n",
    "        while not it.finished:\n",
    "            s = it.iterindex\n",
    "            y, x = it.multi_index\n",
    "\n",
    "            P[s] = {a : [] for a in range(nA)}\n",
    "\n",
    "            is_done = lambda s: s == 0 or s == (nS - 1)\n",
    "            reward = 0.0 if is_done(s) else -1.0\n",
    "\n",
    "            # We're stuck in a terminal state\n",
    "            if is_done(s):\n",
    "                P[s][UP] = [(1.0, s, reward, True)]\n",
    "                P[s][RIGHT] = [(1.0, s, reward, True)]\n",
    "                P[s][DOWN] = [(1.0, s, reward, True)]\n",
    "                P[s][LEFT] = [(1.0, s, reward, True)]\n",
    "            # Not a terminal state\n",
    "            else:\n",
    "                ns_up = s if y == 0 else s - MAX_X\n",
    "                ns_right = s if x == (MAX_X - 1) else s + 1\n",
    "                ns_down = s if y == (MAX_Y - 1) else s + MAX_X\n",
    "                ns_left = s if x == 0 else s - 1\n",
    "                P[s][UP] = [(1.0, ns_up, reward, is_done(ns_up))]\n",
    "                P[s][RIGHT] = [(1.0, ns_right, reward, is_done(ns_right))]\n",
    "                P[s][DOWN] = [(1.0, ns_down, reward, is_done(ns_down))]\n",
    "                P[s][LEFT] = [(1.0, ns_left, reward, is_done(ns_left))]\n",
    "\n",
    "            it.iternext()\n",
    "\n",
    "        # Initial state distribution is uniform\n",
    "        isd = np.ones(nS) / nS\n",
    "\n",
    "        # We expose the model of the environment for educational purposes\n",
    "        # This should not be used in any model-free learning algorithm\n",
    "        self.P = P\n",
    "\n",
    "        super(GridworldEnv, self).__init__(nS, nA, P, isd)\n",
    "\n",
    "    def _render(self, mode='human', close=False):\n",
    "        if close:\n",
    "            return\n",
    "\n",
    "        outfile = StringIO() if mode == 'ansi' else sys.stdout\n",
    "\n",
    "        grid = np.arange(self.nS).reshape(self.shape)\n",
    "        it = np.nditer(grid, flags=['multi_index'])\n",
    "        while not it.finished:\n",
    "            s = it.iterindex\n",
    "            y, x = it.multi_index\n",
    "\n",
    "            if self.s == s:\n",
    "                output = \" x \"\n",
    "            elif s == 0 or s == self.nS - 1:\n",
    "                output = \" T \"\n",
    "            else:\n",
    "                output = \" o \"\n",
    "\n",
    "            if x == 0:\n",
    "                output = output.lstrip() \n",
    "            if x == self.shape[1] - 1:\n",
    "                output = output.rstrip()\n",
    "\n",
    "            outfile.write(output)\n",
    "\n",
    "            if x == self.shape[1] - 1:\n",
    "                outfile.write(\"\\n\")\n",
    "\n",
    "            it.iternext()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- P : [state_num, action_num, 4]，其中最后一维的4位分别表示： `prob`, `next_state`, `reward`, `is_done`。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from lib.envs.gridworld import GridworldEnv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "env = GridworldEnv()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Policy Evaluation\n",
    "\n",
    "![](https://github.com/applenob/rl_learn/raw/master/res/iterative_policy_evaluation.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def policy_eval(policy, env, discount_factor=1.0, theta=0.00001):\n",
    "    \"\"\"\n",
    "    策略评估\n",
    "    Evaluate a policy given an environment and a full description of the environment's dynamics.\n",
    "    \n",
    "    Args:\n",
    "        policy: [S, A] shaped matrix representing the policy.\n",
    "        env: OpenAI env. env.P represents the transition probabilities of the environment.\n",
    "            env.P[s][a] is a list of transition tuples (prob, next_state, reward, done).\n",
    "        theta: We stop evaluation once our value function change is less than theta for all states.\n",
    "        discount_factor: gamma discount factor.\n",
    "    \n",
    "    Returns:\n",
    "        Vector of length env.nS representing the value function.\n",
    "    \"\"\"\n",
    "    # Start with a random (all 0) value function\n",
    "    V = np.zeros(env.nS)\n",
    "    while True:\n",
    "        delta = 0\n",
    "        # For each state, perform a \"full backup\"\n",
    "        for s in range(env.nS):\n",
    "            v = 0\n",
    "            # Look at the possible next actions\n",
    "            for a, action_prob in enumerate(policy[s]):\n",
    "                # For each action, look at the possible next states...\n",
    "                for  prob, next_state, reward, done in env.P[s][a]:\n",
    "                    # Calculate the expected value\n",
    "                    v += action_prob * prob * (reward + discount_factor * V[next_state])\n",
    "            # How much our value function changed (across any states)\n",
    "            delta = max(delta, np.abs(v - V[s]))\n",
    "            V[s] = v\n",
    "        # Stop evaluating once our value function change is below a threshold\n",
    "        if delta < theta:\n",
    "            break\n",
    "    return np.array(V)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- **Bellman Equation**:\n",
    "- 公式：$v_{\\pi}(s) = \\sum_a\\pi(a|s)\\sum_{s',r}p(s',r|s,a)[r+\\gamma v_{\\pi}(s')]\\;\\;\\forall s \\in S$\n",
    "- 代码：\n",
    "```python\n",
    "for a, action_prob in enumerate(policy[s]):\n",
    "                # For each action, look at the possible next states...\n",
    "                for  prob, next_state, reward, done in env.P[s][a]:\n",
    "                    # Calculate the expected value\n",
    "                    v += action_prob * prob * (reward + discount_factor * V[next_state])\n",
    "```\n",
    "- 公式和代码正好一一对应。\n",
    "- 测试期望的value和评估的value误差小于一定的范围：\n",
    "- `abs(desired-actual) < 1.5 * 10**(-decimal)`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "random_policy = np.ones([env.nS, env.nA]) / env.nA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "均匀分布的随机策略"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "v = policy_eval(random_policy, env)\n",
    "expected_v = np.array([0, -14, -20, -22, -14, -18, -20, -20, -20, -20, -18, -14, -22, -20, -14, 0])\n",
    "np.testing.assert_array_almost_equal(v, expected_v, decimal=2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "实际的价值函数和评估的价值函数的误差通过测试。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Policy Iteration\n",
    "\n",
    "![](https://github.com/applenob/rl_learn/raw/master/res/policy_iteration.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def policy_iteration(env, policy_eval_fn=policy_eval, discount_factor=1.0):\n",
    "    \"\"\"\n",
    "    Policy Iteration Algorithm. Iteratively evaluates and improves a policy\n",
    "    until an optimal policy is found.\n",
    "    \n",
    "    Args:\n",
    "        env: The OpenAI envrionment.\n",
    "        policy_eval_fn: Policy Evaluation function that takes 3 arguments:\n",
    "            policy, env, discount_factor.\n",
    "        discount_factor: Lambda discount factor.\n",
    "        \n",
    "    Returns:\n",
    "        A tuple (policy, V). \n",
    "        policy is the optimal policy, a matrix of shape [S, A] where each state s\n",
    "        contains a valid probability distribution over actions.\n",
    "        V is the value function for the optimal policy.\n",
    "        \n",
    "    \"\"\"\n",
    "    # Start with a random policy\n",
    "    policy = np.ones([env.nS, env.nA]) / env.nA\n",
    "    \n",
    "    while True:\n",
    "        # Evaluate the current policy\n",
    "        # 策略评估\n",
    "        V = policy_eval_fn(policy, env, discount_factor)\n",
    "        \n",
    "        # Will be set to false if we make any changes to the policy\n",
    "        policy_stable = True\n",
    "        \n",
    "        # 策略改善\n",
    "        # For each state...\n",
    "        for s in range(env.nS):\n",
    "            # The best action we would take under the currect policy\n",
    "            chosen_a = np.argmax(policy[s])\n",
    "            \n",
    "            # Find the best action by one-step lookahead\n",
    "            # Ties are resolved arbitarily\n",
    "            action_values = np.zeros(env.nA)\n",
    "            for a in range(env.nA):\n",
    "                for prob, next_state, reward, done in env.P[s][a]:\n",
    "                    action_values[a] += prob * (reward + discount_factor * V[next_state])\n",
    "            # 取value最大的action\n",
    "            best_a = np.argmax(action_values)\n",
    "            \n",
    "            # Greedily update the policy\n",
    "            if chosen_a != best_a:\n",
    "                policy_stable = False\n",
    "            policy[s] = np.eye(env.nA)[best_a]\n",
    "        \n",
    "        # If the policy is stable we've found an optimal policy. Return it\n",
    "        if policy_stable:\n",
    "            return policy, V"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这里的核心代码是：\n",
    "\n",
    "```python\n",
    "# Find the best action by one-step lookahead\n",
    "# Ties are resolved arbitarily\n",
    "action_values = np.zeros(env.nA)\n",
    "for a in range(env.nA):\n",
    "    for prob, next_state, reward, done in env.P[s][a]:\n",
    "        action_values[a] += prob * (reward + discount_factor * V[next_state])\n",
    "# 取value最大的action\n",
    "best_a = np.argmax(action_values)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[['UP'], ['LEFT'], ['LEFT'], ['DOWN'], ['UP'], ['UP'], ['UP'], ['DOWN'], ['UP'], ['UP'], ['RIGHT'], ['DOWN'], ['UP'], ['RIGHT'], ['RIGHT'], ['UP']]\n"
     ]
    }
   ],
   "source": [
    "policy, v = policy_iteration(env)\n",
    "actions = [\"UP\", \"RIGHT\", \"DOWN\", \"LEFT\"]\n",
    "print([[action for num, action in zip(one, actions) if num == 1.0] for one in policy])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](https://github.com/applenob/rl_learn/raw/master/res/policy_evaluation_gw1.png)\n",
    "\n",
    "![](https://github.com/applenob/rl_learn/raw/master/res/policy_evaluation_gw2.png)\n",
    "\n",
    "可以看到输出的policy和图上的policy基本一致。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Value Iteration\n",
    "\n",
    "![](https://github.com/applenob/rl_learn/raw/master/res/value_iteration.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def value_iteration(env, theta=0.0001, discount_factor=1.0):\n",
    "    \"\"\"\n",
    "    Value Iteration Algorithm.\n",
    "    \n",
    "    Args:\n",
    "        env: OpenAI environment. env.P represents the transition probabilities of the environment.\n",
    "        theta: Stopping threshold. If the value of all states changes less than theta\n",
    "            in one iteration we are done.\n",
    "        discount_factor: lambda time discount factor.\n",
    "        \n",
    "    Returns:\n",
    "        A tuple (policy, V) of the optimal policy and the optimal value function.\n",
    "    \"\"\"\n",
    "    \n",
    "    def one_step_lookahead(state, V):\n",
    "        \"\"\"\n",
    "        Helper function to calculate the value for all action in a given state.\n",
    "        \n",
    "        Args:\n",
    "            state: The state to consider (int)\n",
    "            V: The value to use as an estimator, Vector of length env.nS\n",
    "        \n",
    "        Returns:\n",
    "            A vector of length env.nA containing the expected value of each action.\n",
    "        \"\"\"\n",
    "        A = np.zeros(env.nA)\n",
    "        for a in range(env.nA):\n",
    "            for prob, next_state, reward, done in env.P[state][a]:\n",
    "                A[a] += prob * (reward + discount_factor * V[next_state])\n",
    "        return A\n",
    "    \n",
    "    V = np.zeros(env.nS)\n",
    "    while True:\n",
    "        # Stopping condition\n",
    "        delta = 0\n",
    "        # Update each state...\n",
    "        for s in range(env.nS):\n",
    "            # Do a one-step lookahead to find the best action\n",
    "            A = one_step_lookahead(s, V)\n",
    "            best_action_value = np.max(A)\n",
    "            # Calculate delta across all states seen so far\n",
    "            delta = max(delta, np.abs(best_action_value - V[s]))\n",
    "            # Update the value function\n",
    "            V[s] = best_action_value        \n",
    "        # Check if we can stop \n",
    "        if delta < theta:\n",
    "            break\n",
    "    \n",
    "    # Create a deterministic policy using the optimal value function\n",
    "    policy = np.zeros([env.nS, env.nA])\n",
    "    for s in range(env.nS):\n",
    "        # One step lookahead to find the best action for this state\n",
    "        A = one_step_lookahead(s, V)\n",
    "        best_action = np.argmax(A)\n",
    "        # Always take the best action\n",
    "        policy[s, best_action] = 1.0\n",
    "    \n",
    "    return policy, V"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "整体思路：每次通过one-step lookahead更新v(s)，找到让v(s)最大的action。直到找到最终的$v_*(s)$，每次选取的动作可以构成一个deterministic policy。\n",
    "\n",
    "更新公式：$v_{k+1}(s) = \\underset{a}{max}\\sum_{s',r}p(s',r|s,a)[r+\\gamma v_k(s')]$，one-step lookahead对应：$\\sum_{s',r}p(s',r|s,a)[r+\\gamma v_k(s')]$"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[['UP'], ['LEFT'], ['LEFT'], ['DOWN'], ['UP'], ['UP'], ['UP'], ['DOWN'], ['UP'], ['UP'], ['RIGHT'], ['DOWN'], ['UP'], ['RIGHT'], ['RIGHT'], ['UP']]\n"
     ]
    }
   ],
   "source": [
    "policy, v = value_iteration(env)\n",
    "print([[action for num, action in zip(one, actions) if num == 1.0] for one in policy])"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}