액터-크리틱(AC, Actor-Critic)

정책을 의미하는 액터와 가치 함수를 의미하는 크리틱을 같이 두는 방법론으로 크리틱은 가치 함수를 추정하며 정책을 평가하는 역할을 합니다.

 

 

정책 그래디언트 정리와 REINFORCE 알고리즘의 $\nabla J(\theta)$ 식은 다음과 같습니다.

 

$\begin{aligned}  
\nabla J(\theta) & \propto \sum_s \mu(s) \sum_a q_{\pi}(s,a) \nabla \pi (a|s, \theta) & \scriptstyle {\text { Policy Gradient Theorem }} \\\\

& = \mathbb{E}_\pi [\color {red} {G_t} \nabla \ln \pi(a|s, \theta)] & \scriptstyle {\text{REINFORCE}} & \\ 
\end{aligned}$

 

REINFORCE 알고리즘에서는 따로 가치망을 두지 않고 한 에피소드의 리턴 $G_t$ 을 이용해서 정책을 평가하고 개선했습니다. 리턴이 양수라면 취했던 행동의 확률은 증가하고 음수라면 확률을 감소하는 방향으로 정책 파라미터를 업데이트하는 것입니다.

 

이러한 REINFORCE 알고리즘은 직관적이고 틀릴 수 없는 알고리즘이지만 episodic 환경에서만 사용할 수 있으며 리턴을 이용하므로 분산이 크다는 단점을 갖습니다. 따라서 유연하며 확장성이 좋은 액터-크리틱 방법론을 알아보도록 하겠습니다.

 

먼저 알아볼 알고리즘은 QAC 입니다.

 

$Q$ 가 붙은 이유는 행동 가치 함수 $Q(s,a)$ 를 학습하기 때문입니다. $\nabla J(\theta)$ 계산은 정책 그래디언트 정리의 $q_{\pi}$ 가 아닌 추정 $Q_w$ 로 사용합니다. 이는 액터의 파라미터 $\theta$ 가 아닌 크리틱의 파라미터 $w$ 라는 의미입니다. QAC 알고리즘은 이러한 $Q_w$ 를 이용해 정책을 평가하고 개선하는 것입니다.

 

 

QAC 알고리즘은 다음과 같습니다.

 

 

 

구현

QAC 알고리즘을 테스트할 환경은 OpenAI Gym 의 CartPole-v0 입니다.

 

 

CartPole 은 좌우로 카트를 이동하여 막대기가 쓰러지지 않도록 오래 유지하는 문제입니다. 

 

환경으로부터 받을 수 있는 정보를 출력합니다.

import gym

env = gym.make('CartPole-v0')

print('obs:', env.observation_space.shape[0])
print('act:', env.action_space.n)

 

상태 정보는 4차원의 벡터로 연속적인 공간을 갖으며 에이전트가 취할 수 있는 행동은 스칼라로 2개의 범위를 갖는 이산적인 공간입니다.

obs: 4
act: 2

 

 

액터 신경망을 정의합니다.

class Actor:
    def __init__(self):
        with tf.variable_scope('actor'):
            with tf.name_scope('input'):
                self.s = tf.placeholder(tf.float32, [None, N_S], name='state')
                self.a = tf.placeholder(tf.int32, name='action')
                self.q = tf.placeholder(tf.float32, name='q')

            with tf.variable_scope('layer'):
                self.fc = tf.layers.dense(self.s, 32, tf.nn.relu)

            with tf.variable_scope('output'):
                self.pi = tf.layers.dense(self.fc, N_A, tf.nn.softmax)

            with tf.name_scope('loss'):
                self.loss = -tf.log(self.pi[0, self.a]) * self.q[self.a]

            with tf.name_scope('optimizer'):
                self.train_op = tf.train.AdamOptimizer(LR_A).minimize(self.loss)

 

상태, 행동, $Q(s,a)$ 를 입력으로 받습니다.

with tf.name_scope('input'):
    self.s = tf.placeholder(tf.float32, [None, N_S], name='state')
    self.a = tf.placeholder(tf.int32, name='action')
    self.q = tf.placeholder(tf.float32, name='q')

 

출력층으로 입력으로 받은 상태에 대한 행동의 확률을 출력합니다.

with tf.variable_scope('output'):
    self.pi = tf.layers.dense(self.fc, N_A, tf.nn.softmax)

 

소프트맥스 활성화 함수를 이용하여 출력의 합이 1인 확률 분포를 출력하는 것입니다.

 

손실 함수를 정의합니다.

with tf.name_scope('loss'):
    self.loss = tf.log(self.pi[0, self.a]) * self.q[self.a]

 

해당 타임스텝에서 취한 행동에 대해서 추정 $Q(s,a)$ 가 양수라면 취한 행동의 확률은 증가하고 음수이면 그 확률은 감소할 것입니다. 또한 log 함수를 통해 높은 확률을 갖는 행동에 대해서는 변화량이 적으며 낮은 확률을 갖는 행동의 변화량은 클 것입니다.

 

마이너스를 붙여 손실값이 최대가 되는 방향으로 학습하도록 설정합니다.

with tf.name_scope('optimizer'):
    self.train_op = tf.train.AdamOptimizer(LR_A).minimize(-self.loss)

 

크리틱 신경망을 정의합니다.

class Critic:
    def __init__(self):
        with tf.variable_scope('critic'):
            with tf.name_scope('input'):
                self.s = tf.placeholder(tf.float32, [None, N_S], name='state')
                self.a = tf.placeholder(tf.int32, name='action')
                self.td_target = tf.placeholder(tf.float32, name='td_target')

            with tf.variable_scope('layer'):
                self.fc = tf.layers.dense(self.s, 32, tf.nn.relu)

            with tf.variable_scope('output'):
                self.q = tf.layers.dense(self.fc, N_A)

            with tf.name_scope('loss'):
                self.loss = tf.losses.mean_squared_error(self.td_target, self.q[0, self.a])

            with tf.name_scope('optimizer'):
                self.train_op = tf.train.AdamOptimizer(LR_C).minimize(self.loss)

 

상태, 행동, TD 타겟을 입력으로 받습니다.

with tf.name_scope('input'):
    self.s = tf.placeholder(tf.float32, [None, N_S], name='state')
    self.a = tf.placeholder(tf.int32, name='action')
    self.td_target = tf.placeholder(tf.float32, name='td_target')

 

출력층으로 입력으로 받은 상태에 대한 $Q(s,a)$ 를 출력합니다. 

with tf.variable_scope('output'):
    self.q = tf.layers.dense(self.fc, N_A)

 

손실 함수를 정의합니다.

with tf.name_scope('loss'):
    self.loss = tf.losses.mean_squared_error(self.td_target, self.q[0, self.a])

 

타겟과 예측 사이의 평균 제곱 오차(mean suqared error, MSE)로 TD 타겟과 추정 $Q(s,a)$ 간의 계산입니다.

 

손실값을 최소화하는 방향으로 학습하도록 설정합니다.

with tf.name_scope('optimizer'):
    self.train_op = tf.train.AdamOptimizer(LR_C).minimize(self.loss)

 

에이전트와 환경의 상호작용입니다.

num_episode = 200
for i in range(num_episode):
    state = env.reset()
    action = agent.get_action(state)
    done = False
    while not done:
        next_state, reward, done, _ = env.step(action)
        next_action = agent.get_action(next_state)
        
        agent.update_critic(done, reward, state, next_state, next_action, action)
        agent.update_actor(state, action)
        
        state = next_state
        action = next_action

 

에피소드를 진행하며 매 타임스텝마다 액터와 크리틱의 신경망을 업데이트합니다.

 

추정 $Q_t(s,a)$ 를 입력으로 하여 액터 신경망을 업데이트합니다.

def update_actor(self, state, action):
    q = self.get_q(state)
    self.sess.run(self.actor.train_op, {self.actor.s: state[np.newaxis, :], self.actor.a: action, self.actor.q: q})

def get_q(self, s):
    return self.sess.run(self.critic.q, {self.critic.s: s[np.newaxis, :]})[0]

 

크리틱 신경망을 업데이트합니다.

def update_critic(self, done, reward, state, next_state, next_action, action):
    if done:
        action_value = 0
    else:
        q = self.get_q(next_state)
        action_value = q[next_action]

    td_target = reward + GAMMA * action_value

    self.sess.run(self.critic.train_op, {self.critic.s: state[np.newaxis, :], self.critic.a: action, self.critic.td_target: td_target})

 

에피소드가 종료되지 않았다면 다음 상태와 다음 행동으로 추정 $Q(s',a')$ 를 이용하여 1-스텝 TD 타겟을 계산합니다.

 

 

500번의 에피소드 동안 학습한 결과입니다.

 

 

저장된 모델을 불러와 테스트합니다.

INFO:tensorflow:Restoring parameters from ./tmp/qac_cartpole/model/model
episode: 0 / step: 146
episode: 1 / step: 200
episode: 2 / step: 150
episode: 3 / step: 31
episode: 4 / step: 180

 

 

어드밴티지 액터-크리틱(Advantage Actor-Critic, A2C)

QAC 는 추정 $Q(s,a)$ 를 이용해서 정책을 평가하고 개선했습니다.

 

하지만 이보다 더욱 효과적인 방법이 있는데 바로 어드밴티지(advantage)를 이용하는 방법입니다.

 

$A(s,a) = Q_w(s,a) - V_v(s)$

 

상태 $s$ 에서 행동 $a$ 를 취했을 때의 추가되는 이득만 고려하는 것이라고 할 수 있습니다. 따라서 추정치의 분산을 작게 만드는 효과가 있습니다. 하지만 위의 식은 $\pi_\theta$, $V_\phi$, $Q_w$ 를 사용하기 때문에 번거롭습니다.

 

좀더 효율적으로 어드밴티지를 구할 수 있는데 다음 식을 통해 알아보겠습니다.

 

$\delta = r + \gamma V(s') - V(s)$

 

$\delta$ 는 1-스텝 TD 오차이며 크리틱 신경망에서는 이 TD 오차를 감소하는 방향으로 학습한다는 것을 알아보았습니다. 또한 $\delta$ 를 어드밴티지로 사용할 수 있는데 이는 $\delta$ 의 기댓값이 $A(s,a)$ 이기 때문입니다. 따라서 $Q_w$ 대신 $r + \gamma V(s')$ 를 이용해서 어드밴티지를 계산할 수 있는 것입니다. 

 

이러한 TD 오차 어드밴티지 방식을 TD Actor-Critic 이라고 하며 일반적으로 Advantage Actor-Critic 이라고 합니다.

 

 

A2C 알고리즘은 다음과 같습니다.

 

 

 

구현

A2C 알고리즘 테스트 환경은 OpenAI Gym 의 Acrobot-v1 입니다.

 

 

Acrobot 은 두 개의 관절과 두 개의 연결된 막대를 갖으며 주어진 높이까지 스윙하는 문제입니다. 한 에피소드에 500번의 타임스텝이 주어지며 단축된 스텝으로 클리어하는 것을 목표로 합니다.

 

환경으로부터 받을 수 있는 정보를 출력합니다.

import gym

env = gym.make('Acrobot-v1')

print('obs:', env.observation_space.shape[0])
print('act:', env.action_space.n)

 

상태 정보는 6차원의 벡터로 연속적인 공간을 갖으며 에이전트가 취할 수 있는 행동은 3개의 범위를 갖는 이산적인 공간입니다.

obs: 6
act: 3

 

 

액터 신경망을 정의합니다.

class Actor:
    def __init__(self):
        with tf.variable_scope('actor'):
            with tf.name_scope('input'):
                self.s = tf.placeholder(tf.float32, [None, N_S], name='state')
                self.a = tf.placeholder(tf.int32, name='action')
                self.adv = tf.placeholder(tf.float32, name='advantage')

            with tf.variable_scope('layer'):
                self.fc = tf.layers.dense(self.s, 64, tf.nn.relu)

            with tf.variable_scope('output'):
                self.pi = tf.layers.dense(self.fc, N_A, tf.nn.softmax)

            with tf.name_scope('loss'):
                probs = tf.reduce_sum(self.pi * tf.one_hot(self.a, N_A), axis=1, keep_dims=True)
                self.loss = -tf.log(probs) * self.adv

            with tf.name_scope('optimizer'):
                self.train_op = tf.train.AdamOptimizer(LR_A).minimize(self.loss)

 

상태, 행동, 어드밴티지를 입력으로 받습니다.

with tf.name_scope('input'):
    self.s = tf.placeholder(tf.float32, [None, N_S], name='state')
    self.a = tf.placeholder(tf.int32, name='action')
    self.adv = tf.placeholder(tf.float32, name='advantage')

 

손실 함수를 정의합니다.

with tf.name_scope('loss'):
    probs = tf.reduce_sum(self.pi * tf.one_hot(self.a, N_A), axis=1, keep_dims=True)
    self.loss = tf.log(probs) * self.adv

 

여기서는 n-스텝 TD 를 계산하기 때문에 행동 시퀀스를 입력으로 받습니다. 이것은 정수 인코딩된 값이므로 원-핫 인코딩을 통해 소프트맥스 함수를 통해 출력된 행렬에서 해당 인덱스의 확률 값을 계산하는 것입니다. 또한 TD 오차 어드밴티지를 사용하여 손실값을 계산합니다.

 

위의 손실값이 최대화(maximize) 하는 방향으로 학습해야 하므로 - 를 취합니다.

with tf.name_scope('optimizer'):
    self.train_op = tf.train.AdamOptimizer(LR_A).minimize(-self.loss)

 

크리틱 신경망을 정의합니다.

class Critic:
    def __init__(self):
        with tf.variable_scope('critic'):
            with tf.name_scope('input'):
                self.s = tf.placeholder(tf.float32, [None, N_S], name='state')
                self.td_target = tf.placeholder(tf.float32, [None, 1], name='td_target')

            with tf.variable_scope('layer'):
                self.fc = tf.layers.dense(self.s, 64, tf.nn.relu)

            with tf.variable_scope('output'):
                self.v = tf.layers.dense(self.fc, 1)

            with tf.name_scope('td_error'):
                self.td_error = self.td_target - self.v
                
            with tf.name_scope('loss'):
                self.loss = tf.reduce_mean(tf.square(self.td_error))

            with tf.name_scope('optimizer'):
                self.train_op = tf.train.AdamOptimizer(LR_C).minimize(self.loss)

 

손실 함수를 정의합니다.

with tf.name_scope('td_error'):
    self.td_error = self.td_target - self.v

with tf.name_scope('loss'):
    self.loss = tf.reduce_mean(tf.square(self.td_error))

 

TD 오차를 계산한 후 제곱의 평균을 계산하는 것으로 먼저 계산된 TD 오차는 어드밴티지로 사용합니다.

 

에피소드를 진행하여 상태 행동 보상을 메모리에 저장합니다.

for i in range(MAX_EPI + 1):
    memory = []
    
    state = env.reset()
    done = False
    while not done:
        action = agent.get_action(state)
        next_state, reward, done, _ = env.step(action)
        
        memory.append((state, action, reward))
        
        ...

 

n-스텝의 TD 타겟을 계산하여 업데이트를 수행합니다.

T_MAX = 8

...

if done or len(memory) == T_MAX:
    memory = np.array(memory)

    if done:
        v_ = 0
    else:
        v_ = agent.get_v(next_state)

    td_target = []
    for r in reversed(memory[:, 2]):
        v_ = r + GAMMA * v_
        td_target.append(v_)
    td_target.reverse()
    
    td_error = agent.update_critic(memory[:, 0], td_target)
    agent.update_actor(memory[:, 0], memory[:, 1], td_error)

    memory = []

 

에피소드가 종료되었다면 다음 상태의 추정 가치 $V(s')$ 는 0으로 하여 리턴 계산과 동일하며 에피소드가 종료되지 않았다면 다음 상태의 추정 가치로부터 계산을 시작하는 것입니다.

 

저장된 보상 시퀀스를 역방향으로 n-스텝 리턴 또는 TD 타겟을 계산합니다.

td_target = []
for r in reversed(memory[:, 2]):
    v_ = r + GAMMA * v_
    td_target.insert(0, v_)

 

TD 타겟을 이용하여 크리틱 신경망과 액터 신경망을 업데이트합니다.

td_error = agent.update_critic(memory[:, 0], td_target)
agent.update_actor(memory[:, 0], memory[:, 1], td_error)

memory = []

 

크리틱 신경망에서는 TD 타겟과 추정 $V(s)$ 간의 오차를 감소하는 방향으로 학습하는 것이고 액터 신경망은 크리틱에서 계산된 오차를 이용하여 정책을 평가하고 개선하는 방향으로 학습하는 것입니다.

 

def update_critic(self, state, td_target):
    _, td_error = self.sess.run([self.critic.train_op, self.critic.td_error], {self.critic.s: state.tolist(), self.critic.td_target: np.vstack(td_target)})
    return td_error

def update_actor(self, state, action, td_error):
    self.sess.run(self.actor.train_op, {self.actor.s: state.tolist(), self.actor.a: action, self.actor.adv: td_error})

def get_v(self, s):
    return self.sess.run(self.critic.v, {self.critic.s: s[np.newaxis, :]})[0, 0]

 

 

300번의 에피소드 동안 학습한 결과입니다.

 

 

 

저장된 모델을 불러와 테스트합니다.

INFO:tensorflow:Restoring parameters from ./tmp/a2c_acrobot/model/model
episode: 0 / step: 117
episode: 1 / step: 84
episode: 2 / step: 143
episode: 3 / step: 124
episode: 4 / step: 114