몬테카를로법으로 정책 평가

몬테카를로법으로 정책 평가

• 몬테카를로법으로 가치 함수를 추정할 것 임.
  즉, 정책 pi가 주어졌을때, 이 정책의 가치함수를 몬테카를로법으로 계산할거임

 

 

 

• 위의 식은 가치 함수를 몬테카를로법으로 계산한 것 이다.
  • 위의 식을 수행하기 위해서 agent는 정책pi에 따라서 실제 action을 하도록한다.
    이렇게 하여 얻은 수익G가 sample data이며, 이러한 sample data를 모아서 평균을 구한다. 이것이 몬테카를로법이다.
• 상태 s에서 시작해서 얻은 수익을 G로 표기함. i번째 얻은 수익을 G(i)로 표기한다. 
  그래서 몬테카를로법으로 계산하기 위해서는 episode를 n번 수행하여 얻은 sample data의 평균을 구하면 된다.
• 몬테카를로법을 이용한 것은 일회성 과제에서만 사용가능...
  • why?
    • 지속적인 과제에는 수익의 sample data가 확정되지 않기 때문이다.
• ex)
  • agent가 총 3번의 action을 수행하였으며 얻은 보상이 1,0,2인 경우, G(1)=1+0+2=3이다.
    (할인률을 1로 가정하였음)
    2번째는 0,0,1,1의 보상을 얻었으며, G(2)=0+0+1+1=2이다.
  • ==> 해당 시점에서 가치 함수 V(s)는 2.5이다. ( 3+2/2 = 2.5 )
  • 위와 같이 실제로 행동을 하여, 수익의 평균을 구하며 가치함수를 근사하는 것이 가능하다.
    ( 시도 횟수를 늘릴수록, 정확도는 높아짐)
• 각각의 state에 대해서 가치 함수를 구하면 비효율적이다...
  • 예시로 state가 A,B,C의 3가지가 있다고 하자... 그러면 아래와 같이 각각 구해야한다.
    • state에 따라서 A->B->C로 전이함을 전재로 하자...

저러한 것이 더 많아지면 비효율적이다...(중복되는 계산이 많다...)

• 위의 식을 개선해보면 아래와 같다.

이렇게 하면 중복되는 계산이 사라짐

 

 

 

import numpy as np
from collections import defaultdict

class GridWorld:
    def __init__( self ):
        self.action_space = [ 0, 1, 2 , 3 ]  # 행동 할 수 있는 가지 수
        self.action_meaning = { 0:'UP' , 1:'DOWN' , 2:'LEFT' , 3:'RIGHT' } # 위에서 정의한 행동 번호가 의미 하는 바
        self.reward_map = np.array( # 게임 맵
                    [ [ 0 , 0    , 0 ,  1.0 ] ,
                      [ 0 , None , 0 , -1.0 ] ,
                      [ 0 , 0    , 0 ,  0   ] ] )
        self.goal_state = ( 0 , 3 ) # 목표 지점
        self.wall_state = ( 1 , 1 ) # 벽 지점
        self.start_state = ( 2 , 0 ) # 시작 지점
        self.agent_state = self.start_state # 에이전트 초기 상태

    @property # 맵의 행 갯수
    def height( self ):
        return len( self.reward_map )

    @property # 맵의 컬럼 갯수
    def width( self ):
        return len( self.reward_map[0] )

    @property # 맵의 쉐잎
    def shape( self ):
        return self.reward_map.shape

    def actions( self ): # 행동할수 있는 범위 
        return self.action_space

    def states( self ):
       for h in range ( self.height):
            for w in range ( self.width):
                yield ( h , w ) # yield가 있는 라인은 함수를 호출한 쪽으로 프로그램의 제어를 넘겨줌을 의미 
                # yield는 return과 다르게 함수 종료가 아니라 일지중지이다. 그래서 다음에 함수 실행시 이전 상태를 기억하고 이어서 호출이 가능
    
    def next_state( self , state , action ):
        action_move_map = [ ( -1 , 0 ) , ( 1, 0 ) , ( 0 , -1 ) , ( 0 , 1 ) ]
        move = action_move_map[ action ] 
        next_state = ( state[0] + move[0] , state[1] + move[1] ) # 위치 이동에 대해서 계산을 수행
        ny , nx = next_state

        if nx < 0 or nx >= self.width or ny < 0 or ny >= self.height: # agent가 맵을 벗어나려는 경우
            next_state = state
        elif next_state == self.wall_state: # 벽에 부딪히는 경우
            next_state = state
        return next_state

    def reward( self , state , action , next_state ):
        return self.reward_map[ next_state ]

    def reset( self ):
        self.agent_state = self.start_state
        return self.agent_state

    def step( self , action ):
        state = self.agent_state
        next_state = self.next_state( state , action ) # 전달받은 action으로 다음 state를 구함.
        reward = self.reward( state , action , next_state )# 위ㅣ치를 이동하였을때 보상을 구함
        done = ( next_state == self.goal_state ) # goal에 도달하였는지 확인
        self.agent_state = next_state # 다음 위치를 저장
        return next_state , reward , done

class RandomAgent:
    def __init__( self ):
        self.gamma = 0.9
        self.action_size = 4
        random_actions = { 0:0.25 , 1:0.25 , 2:0.25 , 3:0.25 }
        self.pi = defaultdict( lambda: random_actions ) # 정책을 상하좌우 각각 0.25확률로 초기화 시킨다.
        self.V = defaultdict( lambda: 0 )
        self.cnts = defaultdict( lambda: 0 )
        self.memory = []

    def get_action( self , state ): # 현재 상태(위치)를 입력받습니다.
        action_probabilities = self.pi[ state ] # 현재 위치에 대해서 정책을 가져옵니다.
        actions = list( action_probabilities.keys() ) # 정책 상하좌우의 키값인 0,1,2,3을 가져옵니다.
        probabilities = list( action_probabilities.values() )# 정책의 확률값을 가져옵니다.
        return np.random.choice( actions , p=probabilities ) # probabilites의 확률에 따라서 actions중 하나를 뽑음

    def add( self , state , action , reward ):
        data = ( state , action , reward )
        self.memory.append( data )

    def reset( self ):
        self.memory.clear()

    def eval( self ):
        G = 0
        for data in reversed( self.memory ):
            state , action , reward = data
            print(data)
            print("---------------")
            G = self.gamma * G + reward
            self.cnts[ state ] += 1
            self.V[ state ] += ( G-self.V[state] ) / self.cnts[ state ]