note/A3C.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A3C"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A3C是Asynchronous Advantage Actor-Critic Model的简称，即异步优势演员-评论家模型，A3C并不是一种像Policy Gradient或DQN这样具体的算法，而是一种解决问题的思想，它的核心精神是，在强化学习的训练过程中，我们可以并行地训练多个Agent，在训练的过程中，各个Agent是参数共享的。更具体一些，我们可能有N个Agent并行地在环境采样1回合后计算1次梯度，我们还有1个或多个Agent在这个过程中什么都不做，当N个Agent中的1个或者M个或者N个采样完成，并反向传播或者基于时间的反向传播计算完1回合的梯度后，这些Agent会将梯度异步地分发给那1个或者多个什么都不做的Agent，然后这些什么都不做的Agent执行一次参数更新，再将更新后的参数分发给这个分发梯度的Agent，然后一直重复这个过程。当然分发梯度和分发参数这个过程是否是异步或者同步，也是可以大做文章的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A3C with TensorFlow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "那么具体要怎么操作呢？这里以TensorFlow为框架实现了一个DPPO，即Distributed Proximal Policy Optimization，分布式近端策略优化模型。这个和A3C有什么关系呢？在上文提到，A3C并不是一个具体的算法，它的核心精神是一套异步训练模型、同步或者异步更新参数的思想。或者换句话说，不管是DQN、Policy Gradient、PPO、ACER，还是基于它们的一系列改进，我们都可以用A3C的思想去改进它们。好在TensorFlow已经为我们做了大部分的底层工作，我们只需要几十行代码，就可以把一个单进程的训练过程改进为分布式训练过程。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Distributed TensorFlow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "为了避免让文章沦为文档翻译，所以这里仅仅对分布式TensorFlow做非常简短的说明，详细的文档可以参考：\n",
    "> [Distributed TensorFlow](https://www.tensorflow.org/deploy/distributed)     "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "首先，需要构造集群描述对象让集群待命，可以通过如下方法构造集群描述对象：\n",
    "```\n",
    "cluster = tf.train.ClusterSpec({\n",
    "    'worker': [\n",
    "        'localhost:8001',\n",
    "        'localhost:8002',\n",
    "        'localhost:8003',\n",
    "    ],\n",
    "    'ps': [\n",
    "        'localhost:8000'\n",
    "    ]\n",
    "})\n",
    "```\n",
    "可以看出，集群描述对象是一个键为job_name（任务名），值为ip:port的字典，至于 job_name 的定义稍后会做解释。然后通过如下语句启动集群中的一个节点，并让节点待命：\n",
    "```\n",
    "server = tf.train.Server(cluster, job_name=role, task_index=task_index)\n",
    "if role == 'ps':\n",
    "    logging.warning('Parameter server started.')\n",
    "    server.join()\n",
    "else:\n",
    "    pass\n",
    "    # do some sth later.\n",
    "```\n",
    "至此，一个节点就被启动并待命了，可以看到一个节点会被抽象为一个server对象，其中job_name对应了节点的任务名，ps是Parameter Server，即参数服务器，worker即计算服务器，它们的用途会在下文提到。根据集群中的每个节点是否会完整地构建自己的计算图，TensorFlow提供了两种方案，分别是 In-graph replication 和 Between-graph replication，每个节点是否会构建自己的计算图，也决定了每个节点的工作方式。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### In-graph replication"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在这种方案中，集群中的每一个节点不会完整地构建自己的计算图，每一个节点仅仅是单纯地利用自己的算力通过以下的语句执行任务：\n",
    "```\n",
    "with tf.device(\"/job:ps/task:0\"):\n",
    "    # Define vars.\n",
    "    \n",
    "with tf.device(\"/job:worker/task:0\"):\n",
    "    # Do computations.\n",
    "```\n",
    "通常，只需要提前启动集群，然后构造一个Session，然后根据节点分配计算图中的各个结点，然后进行训练就可以了，非常地直觉。这样做有一个缺点是数据会在各个结点分发，如果数据非常大，这样是得不偿失的。在下文实现的DPPO中，我们将不会采用这套方案。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Between-graph replication"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "与In-graph replication不同的是，集群中的每一个节点会完整地构建自己的计算图，可以说这种方案就是为了A3C而设计的，在这种方案中，我们会有一个或者多个参数节点（Parameters Server），多个计算节点（Worker Server），每个计算节点完成梯度计算后，会异步地将梯度分发到参数节点，然后参数节点会同步或者异步地用梯度更新参数，然后分发最新的参数到一个或者多个计算节点。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Parameters Server & Worker Server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "ps，即参数节点，在Between-graph replication的方案中，它通常什么都不做，节点启动后即调用`join()`待命，worker，即计算节点，在Between-graph replication方案中，这些节点定义了完整的计算图并执行这些计算，在计算节点完成一次梯度计算后，梯度会被异步分发给参数节点，参数节点更新参数后，分发参数给计算节点。这个过程可以既可以是异步的也可以是同步的，在Between-graph replication方案中，默认是异步的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PPO"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在前一篇文章中已经实现了一个PPO，学习笔记：\n",
    "> [PPO Note](https://github.com/Ceruleanacg/Learning-Notes/blob/master/note/PPO.ipynb)    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "源码：\n",
    "> [PPO Code](https://github.com/Ceruleanacg/Learning-Notes/blob/master/playground/PPO.py)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DPPO in Action"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "首先实现一个方法，它用来启动集群的各个节点，并根据节点类型待命或者定义并执行计算图："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/Users/shuyu/anaconda3/envs/quant/lib/python3.5/importlib/_bootstrap.py:222: RuntimeWarning: compiletime version 3.6 of module 'tensorflow.python.framework.fast_tensor_util' does not match runtime version 3.5\n",
      "  return f(*args, **kwds)\n"
     ]
    }
   ],
   "source": [
    "# coding=utf-8\n",
    "\n",
    "import sys\n",
    "sys.path.append('..')\n",
    "\n",
    "import multiprocessing as mp\n",
    "import tensorflow as tf\n",
    "import logging\n",
    "import gym\n",
    "\n",
    "from base.model import *\n",
    "from playground import PPO\n",
    "from utility.launcher import start_game\n",
    "\n",
    "\n",
    "def start_a3c(cluster, role, task_index):\n",
    "    # 根据集群描述对象启动节点\n",
    "    server = tf.train.Server(cluster, job_name=role, task_index=task_index)\n",
    "    if role == 'ps':\n",
    "        # 如果是参数节点，则join待命\n",
    "        logging.warning('Parameter server started.')\n",
    "        server.join()\n",
    "    else:\n",
    "        # 如果是计算节点，定义计算图，计算梯度\n",
    "        worker_device = \"/job:worker/task:{}\".format(task_index)\n",
    "        logging.warning('Worker: {},  server stated.'.format(worker_device))\n",
    "        # 根据集群描述对象分配节点\n",
    "        with tf.device(tf.train.replica_device_setter(cluster=cluster)):\n",
    "            # Make env.\n",
    "            env = gym.make('CartPole-v0')\n",
    "            env.seed(1)\n",
    "            env = env.unwrapped\n",
    "            # Init session.\n",
    "            session = tf.Session(server.target)\n",
    "            # session = tf.Session()\n",
    "            # Init agent.\n",
    "            agent = PPO.Agent(env.action_space.n, env.observation_space.shape[0], **{\n",
    "                KEY_SESSION: session,\n",
    "                KEY_MODEL_NAME: 'PPO',\n",
    "                KEY_TRAIN_EPISODE: 1000\n",
    "            })\n",
    "            start_game(env, agent, task_index)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "然后定义集群描述对象："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster = tf.train.ClusterSpec({\n",
    "        'worker': [\n",
    "            'localhost:8001',\n",
    "            'localhost:8002',\n",
    "            'localhost:8003',\n",
    "        ],\n",
    "        'ps': [\n",
    "            'localhost:8000'\n",
    "        ]\n",
    "    })\n",
    "\n",
    "role_task_index_map = [\n",
    "    ('ps', 0),\n",
    "    ('worker', 0),\n",
    "    ('worker', 1),\n",
    "    ('worker', 2),\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "启动A3C并训练："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pool = mp.Pool(processes=4)\n",
    "\n",
    "for role, task_index in role_task_index_map:\n",
    "    pool.apply_async(start_a3c, args=(cluster, role, task_index, ))\n",
    "pool.close()\n",
    "pool.join()"
   ]
  }
 ],
 "metadata": {
  "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.5.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}