우리는 Policy Gradient Theorem을 통해 정책 기반 강화학습에서 policy를 직접 학습할 수 있는 방법을 알았다.
Policy Gradient Theorem은 다음과 같은 형태로 표현된다.
이 식은 매우 중요한 의미를 가지는데,
식을 통해 우리는 policy의 gradient를 계산하여 policy parameter θ를 직접 업데이트할 수 있다.
하지만 여기서 한 가지 현실적인 문제가 등장한다.
를 실제로 계산할 수 있는가?
일반적인 환경에서는 정확한 Action-value Qπ(s,a)를 알 수 없다.
따라서 문제를 해결하기 위해 등장한 가장 간단한 알고리즘이 바로 REINFORCE 알고리즘이다.
Policy Gradient Theorm과 REINFORCE 알고리즘
REINFORCE 알고리즘의 핵심 아이디어
REINFORCE 알고리즘의 핵심은 매우 단순하다.
즉 다음과 같이 바꾸는 것이다.
Q 대신 Return (G_t)를 사용할 수 있는 이유
왜 Qπ(s,a) 대신 G_t를 사용해도 괜찮을까?
Action-value 함수는 다음과 같이 정의된다.
이때 Qπ(s,a)는 return G_t의 기댓값이다.
Policy Gradient 식에는 기댓값 연산자 𝐸[⋅]가 존재하기 때문에 여기에 𝐺𝑡를 넣어도 문제가 없다.
여러 에피소드에서 얻은 𝐺𝑡들을 평균내면 결국 Qπ(s,a)로 수렴하게 된다.
왜 Return을 사용하는가?
또 하나 중요한 이유는 Qπ(s,a)를 실제로 계산하기 어렵기 때문이다.
우리가 알고 있는 것은 다음뿐이다.
- 현재 policy π
- 환경과 상호작용하면서 얻는 reward
하지만 Qπ(s,a)는 상태 s에서 행동 a를 했을 때 미래에 받을 모든 reward의 기댓값을 알아야 한다.
반면 Return Gt는 다음의 방법으로 쉽게 얻을 수 있다.
- policy를 환경에 적용한다
- episode를 끝까지 진행한다
- reward를 모두 더한다
따라서 REINFORCE 알고리즘은 "Q 대신 실제로 관측 가능한 Return을 사용하는 방법"이라고 볼 수 있다.
REINFORCE 알고리즘
REINFORCE 알고리즘의 pseudo code
1. Policy Network 초기화
- 먼저 policy network를 정의하고 파라미터 θ를 랜덤하게 초기화한다.
- 이 네트워크는 상태 s를 입력받아 행동 확률 분포 출력한다. (πθ(a∣s))
2. 에피소드 생성
- 현재 policy를 사용하여 환경과 상호작용한다.
(예를 들어 다음과 같은 trajectory가 생성될 수 있다: (s0,a0,r0),(s1,a1,r1),(s2,a2,r2),...) - 이 과정은 episode가 끝날 때까지 계속된다.
3. Return 계산
- 각 시점 t에서의 Return은 현재 시점부터 미래까지 받을 reward의 합으로 정의된다. (뒤에서부터 계산하면 효율적)
4. Policy Gradient Update
- 각 시점에 대해 다음과 같이 파라미터를 업데이트한다.
이 과정을 반복하면 policy가 점점 개선된다.
REINFORCE 업데이트의 직관적 의미
REINFORCE의 파라미터 업데이트 식은 다음과 같다.
먼저 다음 부분을 보자.
이 값은 행동 a의 확률을 증가시키는 방향을 의미한다.
즉 이 방향으로 파라미터를 업데이트하면 πθ(a∣s)가 증가한다.
Return이 양수일 때
만약 Return G_t가 크다면 "+G_t"가 곱해지므로 해당 행동의 확률을 더 크게 증가시킨다.
즉, 좋은 결과를 만든 행동을 더 자주 하도록 만든다.
반면, Return이 음수일 때
반대로 Return이 음수라면 업데이트 방향이 반대로 바뀐다.
즉, 해당 행동의 확률을 감소시킨다.
이를 한 문장으로 정리하면
REINFORCE 알고리즘은 다음 규칙을 따른다.
결과가 좋았던 에피소드의 행동은 확률을 높이고
결과가 나빴던 에피소드의 행동은 확률을 낮춘다.
즉, “좋은 행동은 더 자주, 나쁜 행동은 덜 하도록 학습한다.”
REINFORCE 알고리즘의 loss 함수
그러나 실제 구현에서는 위의 업데이트 식을 직접 사용하지 않는다.
딥러닝 프레임워크(PyTorch 등)는 Loss Function을 최소화하는 방식으로 학습하기 때문이다.
우리가 원하는 업데이트는 다음이다.
이 gradient를 얻기 위해서는 다음 값을 미분하면 된다.
하지만 딥러닝에서는 Loss를 최소화해야 하므로 다음과 같이 음수를 붙인다.
이 Loss를 최소화하면 실제로는 log πθ(s,a) * G_t를 최대화하는 것과 동일하다.
REINFORCE 알고리즘 구현
앞서 살펴본 것처럼 REINFORCE 알고리즘은 Policy Gradient Theorem을 가장 직접적으로 구현한 알고리즘이다.
이 알고리즘의 장점은 구조가 매우 단순하다는 점이다.
이전에 구현했던 DQN의 경우 Replay Buffer나 Target Network와 같은 추가 구성요소들이 필요했다.
반면 REINFORCE는 Policy Network 하나만으로 학습이 가능하기 때문에 훨씬 간단하게 구현할 수 있다.
라이브러리 및 하이퍼파라미터 설정
import gym
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.distributions import Categorical
#Hyperparameters
learning_rate = 0.0002
gamma = 0.98
이번 구현에서 사용한 주요 파라미터는 learning rate 0.0002와 discount factor γ = 0.98이다.
이 값들은 이론적으로 정답이 있는 값이 아니라 실험적으로 안정적으로 동작하는 값을 선택한 것이다.
Policy Network 정의
REINFORCE 알고리즘에서는 Q Network 대신 Policy Network를 사용한다.
즉 상태를 입력받아 각 행동의 확률을 출력하는 신경망을 구성한다.
class Policy(nn.Module):
def __init__(self):
super(Policy, self).__init__()
self.data = []
self.fc1 = nn.Linear(4, 128)
self.fc2 = nn.Linear(128, 2)
self.optimizer = optim.Adam(self.parameters(), lr=learning_rate)
구조는 매우 단순하다.
- 입력: 상태(state)
- 은닉층: 128 차원 fully connected layer
- 출력층: 행동 확률 분포
CartPole 환경의 상태는 4차원 벡터이다.
그리고 가능한 행동은 왼쪽 또는 오른쪽 두 가지이므로 출력 차원은 2가 된다.
이 네트워크의 출력은 다음과 같이 확률 분포가 되어야 한다.
따라서 마지막 layer에는 Softmax 함수가 적용된다.
def forward(self, x):
x = F.relu(self.fc1(x))
x = F.softmax(self.fc2(x), dim=0)
return x
Softmax를 사용하는 이유는 Policy는 확률 분포여야 하기 때문이다.
즉 출력값들은 다음 조건을 만족해야 한다.
- 각 값은 0 이상
- 전체 합은 1
그래야 행동을 확률적으로 샘플링할 수 있다.
Episode 데이터 저장
REINFORCE 알고리즘은 한 에피소드가 끝난 후에 학습을 진행한다.
따라서 에피소드 동안 생성된 데이터를 잠시 저장해 둘 공간이 필요하다.
이를 위해 다음과 같은 리스트를 사용한다.
self.data = []
그리고 데이터를 저장하는 함수는 다음과 같다.
def put_data(self, item):
self.data.append(item)
여기서 저장되는 데이터는 다음 두 가지이다.
- reward rr
- 실제로 선택한 행동의 확률 πθ(s,a)
즉 (reward, prob) 형태로 저장된다.
REINFORCE 업데이트
REINFORCE 알고리즘의 핵심 학습 과정은 train_net() 함수에 구현되어 있다.
def train_net(self):
R = 0
self.optimizer.zero_grad()
for r, prob in self.data[::-1]:
R = r + gamma * R
loss = -torch.log(prob) * R
loss.backward()
self.optimizer.step()
self.data = []
이 함수의 핵심은 Return GtG_t 를 계산하는 부분이다.
Return은 다음과 같이 정의된다.
즉 현재 시점부터 미래에 받을 reward의 누적값이다.
이때 코드를 보면 데이터를 뒤에서부터 순회하고 있다.
self.data[::-1]
예를 들어 하나의 에피소드에서 다음과 같은 reward가 발생했다고 가정해보자.
그러면 각 상태에서의 Return은 다음과 같이 계산된다.
이처럼 앞에서부터 계산하면 매번 미래 reward를 다시 전부 더해야 한다.
하지만 식을 자세히 보면 중요한 패턴이 하나 보인다.
즉 현재 Return은 다음 Return을 이용해서 계산할 수 있다.
이를 일반적으로 표현하면 다음과 같다.
이 성질을 이용하면 Return을 뒤에서부터 계산하는 것이 매우 효율적이라는 사실을 확인할 수 있다.
Loss 함수
Loss 함수는 위에서 설명한 것처럼 다음과 같이 정의한다.
코드에서는 다음과 같이 표현된다.
loss = -torch.log(prob) * R
여기서
- prob : 선택된 행동의 확률
- R : Return
이 식을 통해 Return이 양수이면 해당 행동 확률이 증가하고, Return이 음수이면 해당 행동 확률이 감소한다.
Environment와 상호작용
이제 실제로 환경과 상호작용하며 데이터를 생성하는 부분을 살펴보자.
env = gym.make('CartPole-v1')
pi = Policy()
CartPole 환경을 생성하고 policy network를 초기화한다.
그리고 각 에피소드마다 다음 과정을 반복한다.
prob = pi(torch.from_numpy(s).float())
m = Categorical(prob)
a = m.sample()
여기서 중요한 부분은 행동을 확률적으로 샘플링한다는 점이다.
Categorical 분포를 사용하면 policy가 출력한 확률에 따라 행동을 선택할 수 있다.
만약 action이 왼쪽과 오른쪽이 있고 policy가 각각 0.3과 0.7의 확률을 출력했다면, 30% 확률로 왼쪽, 70% 확률로 오른쪽 행동이 선택된다.
데이터 저장
환경에서 행동을 수행하면 다음 값을 얻는다.
s_prime, r, done, truncated, info = env.step(a.item())
이때 REINFORCE 학습에 필요한 정보는 다음 두 가지이다.
- reward
- 선택한 action의 확률
따라서 다음과 같이 저장한다.
pi.put_data((r, prob[a]))
여기서 prob[a]는 실제로 선택한 action의 확률을 의미한다.
학습 진행 과정
에피소드가 종료되면 다음 코드가 실행된다.
pi.train_net()
이 과정에서 아래의 순서로 학습이 이루어진다.
- Return 계산
- Loss 계산
- Gradient 계산
- Policy Network 업데이트
이 과정을 10000 episode 동안 반복한다.
전체 코드
import gym
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.distributions import Categorical
#Hyperparameters
learning_rate = 0.0002
gamma = 0.98
class Policy(nn.Module):
def __init__(self):
super(Policy, self).__init__()
self.data = []
self.fc1 = nn.Linear(4, 128)
self.fc2 = nn.Linear(128, 2)
self.optimizer = optim.Adam(self.parameters(), lr=learning_rate)
def forward(self, x):
x = F.relu(self.fc1(x))
x = F.softmax(self.fc2(x), dim=0)
return x
def put_data(self, item):
self.data.append(item)
def train_net(self):
R = 0
self.optimizer.zero_grad()
for r, prob in self.data[::-1]:
R = r + gamma * R
loss = -torch.log(prob) * R
loss.backward()
self.optimizer.step()
self.data = []
def main():
env = gym.make('CartPole-v1')
pi = Policy()
score = 0.0
print_interval = 20
for n_epi in range(10000):
s, _ = env.reset()
done = False
while not done: # CartPole-v1 forced to terminates at 500 step.
prob = pi(torch.from_numpy(s).float())
m = Categorical(prob)
a = m.sample()
s_prime, r, done, truncated, info = env.step(a.item())
pi.put_data((r,prob[a]))
s = s_prime
score += r
pi.train_net()
if n_epi%print_interval==0 and n_epi!=0:
print("# of episode :{}, avg score : {}".format(n_epi, score/print_interval))
score = 0.0
env.close()
if __name__ == '__main__':
main()
실행 결과
학습이 진행되면 CartPole의 점수가 점점 증가하는 것을 볼 수 있다.
초기에는 평균 점수가 약 20점 정도에서 시작하지만 학습이 진행될수록 점수가 상승한다.
하지만 학습 도중 점수가 갑자기 떨어지는 경우도 발생한다.
그 이유는 다음과 같다.
- 행동 선택이 확률적이다
- 환경의 반응도 확률적이다
즉 같은 policy라도 운에 따라 결과가 달라질 수 있다.
Policy 확률의 변화
이번에는 각 episode마다 출력되는 (1초마다 Policy Neural Network에 의해 출력되는) policy의 확률분포를 보자.
while not done: # CartPole-v1 forced to terminates at 500 step.
prob = pi(torch.from_numpy(s).float())
if n_epi % 20 == 0:
print("# of episode :{}, prob : {}".format(n_epi, prob)) // prob 출력!
m = Categorical(prob)
a = m.sample()
s_prime, r, done, truncated, info = env.step(a.item())
pi.put_data((r,prob[a]))
s = s_prime
score += r
episode가 20번밖에 진행되지 않은 학습 초기에 policy network의 출력 확률을 보면 다음과 같은 값을 확인할 수 있다.
거의 두 행동이 50대 50으로 동일한 확률로 선택된다.
그 결과 평균 점수도 20점대에 머무른다.
하지만 episode가 2500을 넘을 정도로 학습이 진행되면 다음과 같이 변화한다.
이 의미는 각 step마다 Return이 좋았던 행동의 확률이 점차 조정되고 있음을 뜻한다.
즉 REINFORCE 알고리즘이 실제로 좋은 행동을 강화(reinforce) 하고 있음을 확인할 수 있다.
정리
REINFORCE 알고리즘은 Policy Gradient를 가장 단순하게 구현한 알고리즘이다.
핵심 아이디어는 다음과 같다.
- Policy Network를 이용해 행동을 선택한다
- Episode를 실행해 Return을 얻는다
- Return을 이용해 행동 확률을 업데이트한다
즉
좋은 결과를 만든 행동은 확률을 높이고
나쁜 결과를 만든 행동은 확률을 낮춘다.
이 단순한 아이디어만으로도 CartPole 문제를 성공적으로 해결할 수 있다.