tensorflow
Q- 학습

최소 예제

Q-learning은 모델없는 강화 학습의 변형입니다. Q-learning에서 우리는 에이전트가 각 상태에서 좋은 행동을 선택할 수 있도록 (상태, 행동) 쌍이 얼마나 좋은지 추정하기를 원합니다. 이것은 아래 방정식에 맞는 action-value 함수 (Q)를 근사하여 수행됩니다.

여기서 s 와 a 는 현재 시간 단계에서 상태와 동작입니다. R 은 즉각적인 보상이며 할인 요인입니다. 그리고 s ' 는 관찰 된 다음 상태입니다.

에이전트는 환경과 상호 작용할 때 에이전트가있는 상태를보고 조치를 취하고 보상을받으며 이동 한 새 상태를 관찰합니다. 이주기는 상담원이 종료 상태에 도달 할 때까지 계속됩니다. Q-learning은 오프 정책 (off-policy) 방법이기 때문에 각 (상태, 액션, 보상, next_state)을 재생 버퍼에 경험으로 저장할 수 있습니다. 이러한 경험은 각 교육 반복에서 샘플링되어 Q의 예상치를 향상시키는 데 사용됩니다. 다음과 같은 방법이 있습니다.

1. next_state 에서 에이전트가 그 상태에서 행동을 np.max(next_state_value) 한다고 가정하여 다음 단계의 Q 값을 계산하십시오. 따라서 아래 코드에서 np.max(next_state_value) 를 선택하십시오.
2. 다음 단계의 Q 값은 할인되어 에이전트가 관찰 한 즉각적인 보상에 추가됩니다 (상태, 조치, 보상 , 주 ').
3. 상태 - 행동으로 인해 에피소드가 종료되면 위의 1 단계 및 2 단계 대신에 Q = reward 사용합니다 (일회성 학습). 따라서 버퍼에 추가되는 각 환경에 종료 플래그를 추가해야합니다. (state, action, reward, next_state, terminated)
4. 이 시점에서 우리는 reward 과 next_state 에서 계산 된 Q 값을 가지며 q 네트워크 함수 next_state 의 출력 인 또 다른 Q 값을 갖습니다. 기울기 하강을 사용하여 q- 네트워크 함수 근사의 매개 변수를 변경하고이 두 동작 값의 차이를 최소화함으로써 Q 함수 근사는 실제 동작 값을 향해 수렴합니다.

다음은 깊은 Q 네트워크 구현입니다.

import tensorflow as tf
import gym
import numpy as np

def fullyConnected(name, input_layer, output_dim, activation=None):
    """
    Adds a fully connected layer after the `input_layer`. `output_dim` is
    the size of next layer. `activation` is the optional activation 
    function for the next layer.
    """
    initializer = tf.random_uniform_initializer(minval=-.003, maxval=.003)

    input_dims = input_layer.get_shape().as_list()[1:]
    weight = tf.get_variable(name + "_w", shape=[*input_dims, output_dim],
                             dtype=tf.float32, initializer=initializer)
    bias = tf.get_variable(name + "_b", shape=output_dim, dtype=tf.float32,
                           initializer=initializer)
    next_layer = tf.matmul(input_layer, weight) + bias

    if activation:
        next_layer = activation(next_layer, name=name + "_activated")

    return next_layer

class Memory(object):
    """
    Saves experiences as (state, action, reward, next_action, 
    termination). It only supports discrete action spaces.
    """

    def __init__(self, size, state_dims):
        self.length = size

        self.states = np.empty([size, state_dims], dtype=float)
        self.actions = np.empty(size, dtype=int)
        self.rewards = np.empty((size, 1), dtype=float)
        self.states_next = np.empty([size, state_dims], dtype=float)
        self.terminations = np.zeros((size, 1), dtype=bool)

        self.memory = [self.states, self.actions,
                       self.rewards, self.states_next, self.terminations]

        self.pointer = 0
        self.count = 0

    def add(self, state, action, reward, next_state, termination):
        self.states[self.pointer] = state
        self.actions[self.pointer] = action
        self.rewards[self.pointer] = reward
        self.states_next[self.pointer] = next_state
        self.terminations[self.pointer] = termination
        self.pointer = (self.pointer + 1) % self.length
        self.count += 1

    def sample(self, batch_size):
        index = np.random.randint(
            min(self.count, self.length), size=(batch_size))
        return (self.states[index], self.actions[index],
            self.rewards[index], self.states_next[index],
            self.terminations[index])

class DQN(object):
    """
    Deep Q network agent. 
    """

    def __init__(self, state_dim, action_dim, memory_size, layer_dims,
                 optimizer):

        self.action_dim = action_dim
        self.state = tf.placeholder(
            tf.float32, [None, state_dim], "states")
        self.action_ph = tf.placeholder(tf.int32, [None], "actions")
        self.action_value_ph = tf.placeholder(
            tf.float32, [None], "action_values")
        self.memory = Memory(memory_size, state_dim)

        def _make():
            flow = self.state
            for i, size in enumerate(layer_dims):
                flow = fullyConnected(
                    "layer%i" % i, flow, size, tf.nn.relu)

            return fullyConnected(
                "output_layer", flow, self.action_dim)

        # generate the learner network
        with tf.variable_scope('learner'):
            self.action_value = _make()
        # generate the target network
        with tf.variable_scope('target'):
            self.target_action_value = _make()

        # get parameters for learner and target networks
        from_list = tf.get_collection(
            tf.GraphKeys.TRAINABLE_VARIABLES, scope='learner')
        target_list = tf.get_collection(
            tf.GraphKeys.TRAINABLE_VARIABLES, scope='target')

        # create a copy operation from parameters of learner 
        # to parameters of target network
        from_list = sorted(from_list, key=lambda v: v.name)
        target_list = sorted(target_list, key=lambda v: v.name)
        self.update_target_network = []
        for i in range(len(from_list)):
            self.update_target_network.append(target_list[i].assign(from_list[i]))

        # gather the action-values of the performed actions
        row = tf.range(0, tf.shape(self.action_value)[0])
        indexes = tf.stack([row, self.action_ph], axis=1)
        action_value = tf.gather_nd(self.action_value, indexes)

        # calculate loss of Q network
        self.single_loss = tf.square(action_value - self.action_value_ph)
        self._loss = tf.reduce_mean(self.single_loss)

        self.train_op = optimizer.minimize(self._loss)

    def train(self, session, batch=None, discount=.97):
        states, actions, rewards, next_states, terminals =\
            self.memory.sample(batch)
        next_state_value = session.run(
            self.target_action_value, {self.state: next_states})
        observed_value = rewards + discount * \
            np.max(next_state_value, 1, keepdims=True)
        observed_value[terminals] = rewards[terminals]

        _, batch_loss = session.run([self.train_op, self._loss], {
            self.state: states, self.action_ph: actions,
            self.action_value_ph: observed_value[:, 0]})
        return batch_loss

    def policy(self, session, state):
        return session.run(self.action_value, {self.state: [state]})[0]

    def memorize(self, state, action, reward, next_state, terminal):
        self.memory.add(state, action, reward, next_state, terminal)

    def update(self, session):
        session.run(self.update_target_network)

깊은 Q 네트워크 에서 에이전트의 컨버전스를 향상시키는 데 사용되는 메커니즘은 거의 없습니다. 하나는 샘플링 된 경험 간의 시간적 관계를 막기 위해 재생 버퍼의 경험을 무작위로 샘플링하는 것에 중점을 둡니다. 또 다른 메커니즘은 next_state 에 대한 Q 값을 평가할 때 대상 네트워크를 사용하는 것입니다. 대상 네트워크는 학습자 네트워크와 비슷하지만 매개 변수가 자주 수정되지 않습니다. 또한 대상 네트워크는 그라디언트 디센트에 의해 업데이트되지 않고 대신 매개 변수가 학습자 네트워크에서 복사됩니다.

아래 코드는 Cartpole 환경 에서 작업을 수행하는 방법을 학습하는 에이전트의 예입니다.

ENVIRONMENT = 'CartPole-v1'  # environment name from `OpenAI`.
MEMORY_SIZE = 50000  # how many of recent time steps should be saved in agent's memory
LEARNING_RATE = .01  # learning rate for Adam optimizer
BATCH_SIZE = 8  # number of experiences to sample in each training step
EPSILON = .1  # how often an action should be chosen randomly. This encourages exploration
EPXILON_DECAY = .99  # the rate of decaying `EPSILON`
NETWORK_ARCHITECTURE = [100] # shape of the q network. Each element is one layer
TOTAL_EPISODES = 500  # number of total episodes
MAX_STEPS = 200  # maximum number of steps in each episode
REPORT_STEP = 10  # how many episodes to run before printing a summary

env = gym.make(ENVIRONMENT)  # initialize environment
state_dim = env.observation_space.shape[
    0]  # dimensions of observation space
action_dim = env.action_space.n

optimizer = tf.train.AdamOptimizer(LEARNING_RATE)
agent = DQN(state_dim, action_dim, MEMORY_SIZE,
            NETWORK_ARCHITECTURE, optimizer)

eps = [EPSILON]

def runEpisode(env, session):
    state = env.reset()
    total_l = 0.
    total_reward = 0.
    for i in range(MAX_STEPS):
        if np.random.uniform() < eps[0]:
            action = np.random.randint(action_dim)
        else:
            action_values = agent.policy(session, state)
            action = np.argmax(action_values)

        next_state, reward, terminated, _ = env.step(action)

        if terminated:
            reward = -1

        total_reward += reward

        agent.memorize(state, action, reward, next_state, terminated)
        state = next_state
        total_l += agent.train(session, BATCH_SIZE)

        if terminated:
            break

    eps[0] *= EPXILON_DECAY
    i += 1

    return i, total_reward / i, total_l / i

session = tf.InteractiveSession()
session.run(tf.global_variables_initializer())

for i in range(1, TOTAL_EPISODES + 1):
    leng, reward, loss = runEpisode(env, session)
    agent.update(session)
    if i % REPORT_STEP == 0:
        print(("Episode: %4i " +
               "| Episod Length: %3i " +
               "| Avg Reward: %+.3f " +
               "| Avg Loss: %6.3f " +
               "| Epsilon: %.3f") %
              (i, leng, reward, loss, eps[0]))