5. 몬테카를로 Prediction

독서/기초부터 시작하는 강화학습 신경망 알고리즘

5. 몬테카를로 Prediction

studylida 2025. 4. 2. 22:25

몬테카를로 방법을 이용한 상태가치 함수 추정 (Prediction)

강화학습에서 상태가치를 구하기 위한 수식은 다음과 같다:

V_π(s) = ∑_a π(a|s) ∑_{s'} P(s'|s,a) [r(s,a,s')] + γV_π(s')

이 수식을 계산하려면 다음 항목이 필요하다:

정책 π(a|s)
상태전이확률 P(s'|s,a)
보상 r(s,a,s')
감가율 γ
다음 상태의 가치 V_π(s')

특히 상태전이확률을 알 수 없다면 계산이 불가능하다.

동적계획법과 실제 환경의 차이

동적계획법은 환경의 정보를 완전히 알고 있으며, 결정론적 환경(P(s'|s,a) = 1)을 전제한다. 하지만 실제 환경은 불확실성이 존재한다. 예를 들어, 빙판길에서 앞으로 한 걸음 내딛었을 때 발생할 수 있는 결과는 다음과 같다:

P(전진 | 빙판길, 앞으로) = ?
P(미끄러짐 | 빙판길, 앞으로) = ?
...

이처럼 모든 전이 확률을 열거하거나 알 수 없기 때문에, 동적계획법은 현실에 적용하기 어렵다.

모델 프리 강화학습의 필요성

환경 모델 없이도 학습할 수 있는 모델 프리(Model-Free) 알고리즘이 필요하다. 대표적인 방법으로는 몬테카를로 방법과 시간차 학습(Temporal Difference) 이 있다. 이들은 다음과 같이 나뉜다:

Prediction: 주어진 정책에 대한 가치 함수 계산
Control: 최적 정책을 찾기 위한 과정

기존에 사용했던 동적계획법과 같이 상태전이확률을 알아야 사용할 수 있는 알고리즘을 모델 기반 알고리즘이라 한다.

몬테카를로 방법의 핵심 아이디어

몬테카를로 방법은 환경의 모델 없이, 정책을 따라 에피소드를 생성하고 수익을 관측하여 상태가치를 추정한다. 동적계획법처럼 상태전이확률을 이용하지 않고, 경험 기반의 샘플링으로 학습한다.

몬테카를로 Prediction 과정

에이전트는 도착지점을 제외한 s0부터 s7까지의 모든 상태에서 에피소드를 시작한다.
가능한 행동 중 무작위로 선택하여 이동한다.
지정된 횟수만큼 에피소드를 반복하며, 각 에피소드에서 수익 G를 저장한다.
수익 G들의 평균을 각 상태에 대해 계산하여 상태가치로 저장한다.

V(s) = average(G₁, G₂, ..., Gₙ)

중복 상태가 있는 경우

에피소드 내에 동일한 상태가 여러 번 등장할 수 있기 때문에, 몬테카를로 방법에서는 두 가지 방식의 수익 계산법을 제안한다:

First-Visit MC: 상태가 처음 등장했을 때의 수익만 계산
Every-Visit MC: 등장한 모든 시점의 수익을 계산

예시

First-Visit MC의 경우:

G = r(s0, a2, s3) + γr(s3, a2, s6) + γ²r(s6, a0, s3) + γ³r(s7, a1, s8)

s3, s6의 첫 방문만 계산에 포함된다.

Every-Visit MC의 경우:

G = r(s0, a2, s3) + γr(s3, a2, s6) + γ²r(s6, a0, s3) + γ³r(s3, a2, s6) + γ⁴r(s6, a1, s7) + γ⁵r(s7, a1, s8)

모든 방문을 수익 계산에 포함한다.

두 방식은 모두 사용 가능하며, 상황에 따라 선택된다.

First-Visit MC Prediction 알고리즘

알고리즘: First-visit MC와 Prediction

입력:
초기화:
    π <- 평가할 정책
    V <- 임의의 상태 가치 함수
    Return(s) <- 빈 리스트(모든 s ∈ S에 대해)

반복:
    정책 π를 이용해 에피소드 생성
    에피소드에 출현한 각 상태 s에 대해:
        G <- 처음 s에 의해 발생한 수익
        G를 Return(s)에 추가(append)
        V(s) <- average(Return(s))

이 알고리즘은 정책 π에 따라 에피소드를 생성하고, 수익을 상태별로 누적하여 평균값을 상태가치로 추정한다.

몬테카를로 방법의 학습 조건

모든 상태에서 에피소드 시작 가능
→ 모든 상태를 경험하지 못하면 상태가치 추정이 불가능
에피소드가 반드시 종료되어야 함
→ 수익 계산이 종료 시점 기준으로 수행되기 때문에, 무한 에피소드는 적용 어려움

에피소드 생성 함수

def generate_episode(env, agent, first_visit):
    gamma = 0.09
    # 에피소드를 저장할 리스트
    episode = []
    # 이전에 방문여부 체크
    visit = np.zeros((env.reward.shape[0], env.reward.shape[1]))

    # 에이전트가 모든 상태에서 출발할 수 있게 출발지점을 무작위로 설정
    i = np.random.randint(0,env.reward.shape[0])
    j = np.random.randint(0,env.reward.shape[1])
    agent.set_pos([i,j])    
    #에피소드의 수익을 초기화
    G = 0
    #감쇄율의 지수
    step = 0
    max_step = 100
    # 에피소드 생성
    for k in range(max_step):
        pos = agent.get_pos()            
        action = np.random.randint(0,len(agent.action))            
        observaetion, reward, done = env.move(agent, action)    

        if first_visit:
            # 에피소드에 첫 방문한 상태인지 검사 :
            # visit[pos[0],pos[1]] == 0 : 첫 방문
            # visit[pos[0],pos[1]] == 1 : 중복 방문
            if visit[pos[0],pos[1]] == 0:   
                # 에피소드가 끝날때까지 G를 계산
                G += gamma**(step) * reward        
                # 방문 이력 표시
                visit[pos[0],pos[1]] = 1
                step += 1               
                # 방문 이력 저장(상태, 행동, 보상)
                episode.append((pos,action, reward))
        else:
            G += gamma**(step) * reward
            step += 1                   
            episode.append((pos,action,reward))            

        # 에피소드가 종료했다면 루프에서 탈출
        if done == True:                
            break        

    return i, j, G, episode

first_visit: First-Visit MC 여부 설정
반환값:
- (i, j): 출발 상태 좌표
- G: 수익
- episode: (상태, 행동, 보상)의 리스트

몬테카를로 Prediction 알고리즘 구현

# first-visit MC and every-visit MC prediction
np.random.seed(0)
# 환경, 에이전트를 초기화
env = Environment()
agent = Agent()

# 임의의 상태 가치 함수𝑉
v_table = np.zeros((env.reward.shape[0], env.reward.shape[1]))

# 상태별로 에피소드 출발횟수를 저장하는 테이블
v_start = np.zeros((env.reward.shape[0], env.reward.shape[1]))

# 상태별로 도착지점 도착횟수를 저장하는 테이블
v_success = np.zeros((env.reward.shape[0], env.reward.shape[1]))

# 𝑅𝑒𝑡𝑢𝑟𝑛(𝑠)←빈 리스트 (모든 s∈𝑆에 대해)
Return_s = [[[] for j in range(env.reward.shape[1])] for i in range(env.reward.shape[0])]

# 최대 에피소드 수를 지정
max_episode = 100000

# first visit 를 사용할지 every visit를 사용할 지 결정
# first_visit = True : first visit
# first_visit = False : every visit
first_visit = True
if first_visit:
    print("start first visit MC")
else : 
    print("start every visit MC")
print()

for epi in tqdm(range(max_episode)):

    i,j,G,episode = generate_episode(env, agent, first_visit)

    # 수익 𝐺를 𝑅𝑒𝑡𝑢𝑟𝑛(𝑠)에 추가(append)
    Return_s[i][j].append(G)

    # 에피소드 발생 횟수 계산
    episode_count = len(Return_s[i][j])
    # 상태별 발생한 수익의 총합 계산
    total_G = np.sum(Return_s[i][j])
    # 상태별 발생한 수익의 평균 계산
    v_table[i,j] = total_G / episode_count

  # 도착지점에 도착(reward = 1)했는지 체크    
    # episode[-1][2] : 에피소드 마지막 상태의 보상
    if episode[-1][2] == 1:
        v_success[i,j] += 1

# 에피소드 출발 횟수 저장 
for i in range(env.reward.shape[0]):
    for j in range(env.reward.shape[1]):
        v_start[i,j] = len(Return_s[i][j])

print("V(s)")
show_v_table(np.round(v_table,2),env)
print("V_start_count(s)")
show_v_table(np.round(v_start,2),env)
print("V_success_pr(s)")
show_v_table(np.round(v_success/v_start,2),env)

결과 해석

상태가치는 도착지점에 가까울수록 증가
상태전이확률 없이도 가치 추정 가능
First-visit과 Every-visit의 결과는 거의 동일 (계산 시간은 차이 있음)
각 상태에서의 출발 횟수는 전체적으로 고르게 분포
도착 확률도 도착지점에 가까운 상태일수록 높음

메모리 효율을 고려한 몬테카를로 Prediction 개선

앞서 구현한 몬테카를로 Prediction에서는 각 상태별로 발생한 수익들을 Return_s 리스트에 모두 저장하고, 에피소드가 끝난 뒤 평균을 계산하는 방식이었다. 이 방식은 상태 수가 적은 경우에는 문제가 없지만, 상태가 많은 경우에는 메모리 사용량이 커져 비효율적이다.

예를 들어, 어떤 상태 s에서 얻은 수익이 다음과 같다고 하자:

수익 리스트 = [-10, -20, 15]
average = (-10 - 20 + 15) / 3 = -5

4번째 수익 G₄ = 3이 추가되면 리스트에 저장한 후 다시 평균을 계산해야 한다. 이처럼 수익 리스트를 계속 유지하면서 평균을 계산하는 것은 메모리 측면에서 비효율적이다.

점진적 평균 (Incremental Mean)

이 문제를 해결하기 위해 평균 계산식을 다음과 같이 변형할 수 있다.

n번째까지의 평균 Qₙ:

Qₙ = (G₁ + G₂ + ... + Gₙ) / n

새로운 수익 Gₙ₊₁이 추가되었을 때 평균 Qₙ₊₁은 다음과 같이 갱신된다:

Qₙ₊₁ = Qₙ + (Gₙ₊₁ - Qₙ) / (n + 1)

전체 유도 과정은 다음과 같다.

```
Q_{n+1} = ∑_{i=1}^{n+1} G_i / (n+1)
	= (G_{n+1} + ∑_{i=1}^{n} G_i)) / (n+1)
	= (G_{n+1} + ∑_{i=1}^{n} G_i) / n * n ) / (n+1)
	= (G_{n+1} + n*Q_n ) / (n+1)
	= (G_{n+1} + n*Q_n  + Q_n - Q_n) / (n+1)
	= (G_{n+1} + (n+1)*Q_n - Q_n ) / (n+1)
	= Q_n + G_{n+1}/(n+1) - Q_n/(n+1)
	= Q_n + (G_{n+1} - Q_n)/(n+1)
```

이는 다음 형태로 해석할 수 있다:

새로운 평균 ← 이전 평균 + 스텝 사이즈 × (새로운 수익 - 이전 평균)

여기서 (1 / (n + 1))이 스텝 사이즈가 되며, 새로 들어온 데이터가 평균에 미치는 영향을 조절한다. 이 식은 점진적 평균(Incremental Mean) 이라고 하며, 강화학습뿐 아니라 머신러닝에서도 널리 활용된다.

Incremental Mean 기반 몬테카를로 Prediction

앞서 사용했던 평균 계산 방식:

V(s) ← average(Return(s))
Return(s) = (G_0 + G_1 + G_2, ...)

이제는 다음과 같이 바꿔 쓸 수 있다:

V(Sₜ) ← V(Sₜ) + α[Gₜ - V(Sₜ)]

여기서 α는 스텝 사이즈이며, 수익 Gₜ와 현재 상태가치 V(Sₜ)의 차이를 줄이도록 업데이트하는 방식이다. 이는 Gₜ를 V(Sₜ)의 목표값으로 보고 그 차이를 줄이는 방식의 학습으로 이해할 수 있다. 따라서 [G_t - V(S_t)] = 0이 되도록 상태가치 V(S_t)를 학습한다.

Incremental Mean 기반 코드 구현

#Incremental mean 을 이용하는 몬테카를로 Prediction 알고리즘
np.random.seed(0)
# 환경, 에이전트를 초기화
env = Environment()
agent = Agent()

# 임의의 상태 가치 함수𝑉
v_table = np.zeros((env.reward.shape[0], env.reward.shape[1]))

# 추가
# 상태를 방문한 횟수를 저장하는 테이블
v_visit = np.zeros((env.reward.shape[0], env.reward.shape[1]))

# 최대 에피소드 수와 에피소드 최대 길이지정
max_episode = 100000

# first visit을 사용할지 every visit을 사용할 지 결정
# first_visit = True : first visit
# first_visit = False : every visit
first_visit = True
if first_visit:
    print("start first visit MC")
else : 
    print("start every visit MC")
print()

for epi in tqdm(range(max_episode)):

    i,j,G,episode = generate_episode(env, agent, first_visit)

    # Incremental mean  평균을 계산
    v_visit[i,j] += 1
    v_table[i,j] += 1 / v_visit[i,j] * (G - v_table[i,j])


print("V(s)")
show_v_table(np.round(v_table,2),env)
print()

이제 수익 리스트를 따로 저장할 필요 없이, 상태별 방문 횟수와 현재 평균만으로 상태가치를 업데이트할 수 있다.

한 상태에서만 시작할 때의 문제

몬테카를로 방법이 제대로 작동하려면 모든 상태에서 에피소드를 시작할 수 있어야 한다는 전제조건이 필요하다. 하지만 미로 환경처럼 시작 상태가 고정되어 있는 경우, 출발 상태 이외의 상태에서는 상태가치 함수를 추정할 수 없다.

이 문제를 해결하기 위해 하나의 에피소드를 서브 에피소드로 분할하여 학습할 수 있다.

서브 에피소드 방식의 적용

예를 들어, 에피소드가 다음 상태들을 순서대로 거쳤다고 하자:

s0 → s1 → s2 → 절벽

이 에피소드를 다음과 같이 나눌 수 있다:

s0 → s1 → s2 → 절벽
s1 → s2 → 절벽
s2 → 절벽

각 서브 에피소드를 독립적인 에피소드로 간주하고, 해당 시점 이후의 수익 G를 계산하여 상태가치를 업데이트한다.

서브 에피소드 방식의 코드 구현

# 에피소드 분리를 이용하는 몬테카를로 방법의 Prediction 알고리즘
def generate_episode(env, agent, first_visit):
    gamma = 0.09
    # 에피소드를 저장할 리스트
    episode = []
    # 이전에 방문여부 체크
    visit = np.zeros((env.reward.shape[0], env.reward.shape[1]))

    # 에이전트는 항상 (0,0)에서 출발
    i = 0
    j = 0
    agent.set_pos([i,j])    
    #에피소드의 수익을 초기화
    G = 0
    #감쇄율의 지수
    step = 0
    max_step = 100
    # 에피소드 생성
    for k in range(max_step):
        pos = agent.get_pos()            
        action = np.random.randint(0,len(agent.action))            
        observaetion, reward, done = env.move(agent, action)    

        if first_visit:
            if visit[pos[0],pos[1]] == 0:   
                G += gamma**(step) * reward        
                visit[pos[0],pos[1]] = 1
                step += 1               
                episode.append((pos,action, reward))
        else:
            G += gamma**(step) * reward
            step += 1                   
            episode.append((pos,action, reward))            

        if done == True:                
            break        

    return i, j, G, episode

#에피소드 분리를 이용하는 몬테카를로 Prediction 알고리즘
np.random.seed(0)
# 환경, 에이전트를 초기화
env = Environment()
agent = Agent()

# 임의의 상태 가치 함수𝑉
v_table = np.zeros((env.reward.shape[0], env.reward.shape[1]))

# 상태를 방문한 횟수를 저장하는 테이블
v_visit = np.zeros((env.reward.shape[0], env.reward.shape[1]))

# 최대 에피소드 수와 에피소드 최대 길이를 지정
max_episode = 99999

# first visit 를 사용할지 every visit를 사용할 지 결정
# first_visit = True : first visit
# first_visit = False : every visit
first_visit = True
if first_visit:
    print("start first visit MC")
else : 
    print("start every visit MC")
print()

gamma = 0.09
for epi in tqdm(range(max_episode)):

    i,j,G,episode = generate_episode(env, agent, first_visit)
    for step_num in range(len(episode)):
        G = 0
        
        # episode[step_num][0][0] : step_num번째 방문한 상태의 x 좌표
        # episode[step_num][0][1] : step_num번째 방문한 상태의 y 좌표
        # episode[step_num][1] : step_num번째 상태에서 선택한 행동
        i = episode[step_num][0][0]
        j = episode[step_num][0][1]
        
        # 에피소드 시작점을 카운트
        v_visit[i,j] += 1
        
        # 서브 에피소드 (episode[step_num:])의 출발부터 끝까지 보수 G를 계산
        # k[2] : episode[step_num][2] 과 같으며 step_num 번째 받은 보상
        # step : 감쇄율
        for step, k in enumerate(episode[step_num:]):
            G += gamma**(step)*k[2]

        # Incremental mean  평균을 계산
        v_table[i,j] += 1 / v_visit[i,j] * (G - v_table[i,j])

print("V(s)")
show_v_table(np.round(v_table,2),env)
print("V_start_count(s)")
show_v_table(np.round(v_visit,2),env)

결과 해석

서브 에피소드 방식을 적용하면, 한 상태에서만 시작하는 환경에서도 모든 상태의 가치 함수 추정이 가능해진다.

(0,0)에서만 에피소드가 시작되었기 때문에 출발 상태의 에피소드 수는 최대
도착지점으로 갈수록 에피소드 수는 감소하지만, 상태가치의 추세는 여전히 도착지점으로 향해 증가하는 형태를 유지
서브 에피소드 방식으로도 초기 방식과 거의 동일한 결과를 얻을 수 있음

'독서 > 기초부터 시작하는 강화학습 신경망 알고리즘' 카테고리의 다른 글

7. 시간차 학습의 Prediction과 Control (0)	2025.04.11
6. 몬테카를로 Control (0)	2025.04.11
4.2 정책 개선, 정책 반복, 그리고 가치 반복 (0)	2025.04.01
4.1 정책 평가와 반복 정책 평가 (0)	2025.03.31
3.2 가치함수: 행동 가치 계산 (0)	2025.03.31

현재글5. 몬테카를로 Prediction

studylida

주식, 마이크로서비스, 네트워크, 연결, IEMG, 로드 밸런서, 모놀리, cpp, 게시글 작성 규칙, 인터페이스, 어드레싱, OSI 참조 모델, MAC 주소 필터링, SPTL, API 게이트웨이, 작성 규칙, 프로그래밍, fill_n, INDA, 리얼티인컴,

Today :
Yesterday :

일	월	화	수	목	금	토
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30

내 블로그 - 관리자 홈 전환	`Q` `Q`
새 글 쓰기	`W` `W`

글 수정 (권한 있는 경우)	`E` `E`
댓글 영역으로 이동	`C` `C`

이 페이지의 URL 복사	`S` `S`
맨 위로 이동	`T` `T`
티스토리 홈 이동	`H` `H`
단축키 안내	`Shift` + `/` `⇧` + `/`

studylida