<jupyter_start><jupyter_text>Policy Iteration (2 step)- Policy Evaluation- Policy Improvement<jupyter_code>import gym
%matplotlib inline
env = gym.make("FrozenLake8x8-v0")

obs_space = env.observation_space
n_state = obs_space.n
print("Observation space's total states :", n_state)

act_space = env.action_space
n_act = act_space.n
print("Action space's total actions :", n_act)

# Transition Matrix for FrozenLake
P = env.unwrapped.P
env.render() # show<jupyter_output>Observation space's total states : 64
Action space's total actions : 4

[41mS[0mFFFFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG<jupyter_text>Define functions step by step Initial Policy : $\pi_0 (\centerdot| s)$<jupyter_code>def plot_policy(policy):
    plt.figure(num=0)
    plt.imshow(np.ones((8,8,3)))
    plt.title("Policy")  
    arr_len = 0.2
    for i in range(8):
        for j in range(8):
            s = i*8+j
            if policy[s][0]> 0:
                plt.arrow(j,i,-arr_len,0,color="r",alpha=policy[s][0],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
            if policy[s][1]> 0:
                plt.arrow(j,i,0,arr_len,color="r",alpha=policy[s][1],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
            if policy[s][2]> 0:
                plt.arrow(j,i,arr_len,0,color="r",alpha=policy[s][2],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
            if policy[s][3]> 0:
                plt.arrow(j,i,0,-arr_len,color="r",alpha=policy[s][3],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
    plt.grid(False)
    plt.show()
    
policy = np.random.uniform(size=(n_state,n_act)) # 0~1사이의 숫자 random sampling
policy = policy/np.sum(policy, axis=1, keepdims=True) # 합이 1이 되도록 만들어준다.

np.set_printoptions(precision=3, suppress=True)
print("Initial Policy Distribution : \n")
plot_policy(policy)
print(policy)<jupyter_output>Initial Policy Distribution :<jupyter_text>Policy Evaluation: policy($\pi$)를 고정하고, 전체 state의 value function을 계산한다. : 아래의 식을 활용 $$V_{\pi}(s) = \sum_a \pi(a|s)\sum_{s'} [r(s,a,s') + \gamma V_{\pi}(s')]P(s'|s,a)$$<jupyter_code>P.keys() # all state
P[0].keys() # all action in state 0
P[0][1]
def policy_evaluation(env, policy, gamma=0.99, epsilon=1e-6):
    """
    gamma : discount factor
    epsilon : terminal condition(이것보다 작아지면 break하여 optimal추출)
    """
    
    # Extract environment information
    obs_space = env.observation_space
    n_state = obs_space.n
    P = env.unwrapped.P
    
    # Random initialize the Value (function)
    v = np.random.uniform(size=(n_state,1))
    
    while True:
        v_prime = np.zeros((n_state,))
        for s in P.keys():
            for a in P[s].keys():
                for prob, next_s, reward, done in P[s][a]:
                    v_prime[s] += policy[s][a] * (reward + gamma*v[next_s]) * prob # 지금 가지고 있는 policy를 가지고 value를 뽑고
        
        dist = np.max(np.abs(v-v_prime))
        v = v_prime
        if dist < epsilon:
            break
    return v

def plot_results(policy, value): 
    # policy visualization
    plt.figure(num=0)
    plt.imshow(np.reshape(value,(8,8)))
    plt.title("Value Function")  
    arr_len = 0.2
    for i in range(8):
        for j in range(8):
            s = i*8+j
            if policy[s][0]> 0:
                plt.arrow(j,i,-arr_len,0,color="r",alpha=policy[s][0],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
            if policy[s][1]> 0:
                plt.arrow(j,i,0,arr_len,color="r",alpha=policy[s][1],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
            if policy[s][2]> 0:
                plt.arrow(j,i,arr_len,0,color="r",alpha=policy[s][2],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
            if policy[s][3]> 0:
                plt.arrow(j,i,0,-arr_len,color="r",alpha=policy[s][3],width=0.01,head_width=0.5,head_length=0.1,overhang=1)
    plt.grid(False)
    plt.show()

if __name__=='__main__':
    print("Policy Evaluation for random policy \n")
    value = policy_evaluation(env, policy=policy, gamma=0.99, epsilon=1e-6)
    print("Value of current policy")
    plot_results(policy=policy, value=value)
    
    # policy 초기값 넣어 한번 evaluation하고 시각화<jupyter_output>Policy Evaluation for random policy 

Value of current policy<jupyter_text>Policy Improvement: Policy Evaluation에서 계산된 value가지고, Policy($\pi$)를 업데이트하자.: 새로운 policy는 greed하게 업데이트된다. \DeclareMathOperator*{\argmax}{arg\,max}$$ \\ Q_{\pi_i}(s,a) = \sum_{s'}[r(s,a,s') + \gamma V_{\pi_i}(s')]P(s'|s,a) \\$$$$\pi_{i+1} = argmax_a Q_{\pi_i}(s,a)$$<jupyter_code>def policy_improvement(env, v, gamma=0.99):
    obs_space = env.observation_space
    n_state = obs_space.n
    act_space = env.action_space
    n_act = act_space.n
    q = np.zeros((n_state,n_act))
    
    for s in P.keys():
        for a in P[s].keys():
            for prob, next_s, reward, done in P[s][a]:
                q[s][a] += (reward + gamma*v[next_s])*prob
                
    policy = np.zeros((n_state, n_act))
    # 모든 state의 expected return(q값)이 제일 큰 action의 probability를 1로 만든다.
    policy[np.arange(n_state), np.argmax(q,axis=1)] = 1 
    
    return policy

if __name__=='__main__':
    print("Policy improvement \n")
    policy = policy_improvement(env, value) # 위에서 선언한 value
    print("Improved Policy")
    plot_results(policy=policy, value=value)<jupyter_output>Policy improvement 

Improved Policy<jupyter_text>이게 최초 업데이트된 상태의 policy 이제 이걸 iteration시킨다.(반복)  Policy IterationPolicy iteration algorithm consists of four steps: (1) Initial policy distribution is randomly initialized (2) Value function for current policy is computed by policy evaluation (3) Policy distribution is greedily updated using value function from (2) (4) If policy distribution doesn't change, then stop. Othrewise, repeat (1) ~ (3)<jupyter_code>def policy_iteration(env):
    # initial policy setting
    policy = np.random.uniform(size=(n_state, n_act))
    policy = policy / np.sum(policy, axis=1, keepdims=True)
    
    while True:
        # policy evaluation
        value = policy_evaluation(env, policy)
        # policy improvement
        policy_prime = policy_improvement(env, value)
        
        if (policy == policy_prime).all(): # it means 'nothing changed'
            print("complete the iteration, find the optimal policy!!")
            break
        policy = policy_prime # update new policy
    
    return policy, value

if __name__=='__main__':
    print("Policy Iteration")
    import time
    start = time.time()
    policy, value = policy_iteration(env)
    print("Computation Time : %.2f"%(time.time() - start))
    plot_results(policy, value)<jupyter_output>Policy Iteration
complete the iteration, find the optimal policy!!
Computation Time : 23.39<jupyter_text>Run Optimal Policy<jupyter_code>obs = env.reset()
for t in range(10000):
    print('t = %d'%t)
    env.render()
    print('')
    action = np.random.choice(n_act, 1, p=policy[obs][:])[0]
    next_obs, reward, done, info = env.step(action)
    obs = next_obs
    if done:
        break
env.render()
env.close()<jupyter_output>t = 0

[41mS[0mFFFFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 1
  (Up)
[41mS[0mFFFFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 2
  (Up)
[41mS[0mFFFFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 3
  (Up)
S[41mF[0mFFFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 4
  (Right)
SFFFFFFF
F[41mF[0mFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 5
  (Up)
S[41mF[0mFFFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 6
  (Right)
SF[41mF[0mFFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 7
  (Right)
SFF[41mF[0mFFFF
FFFFFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 8
  (Right)
SFFFFFFF
FFF[41mF[0mFFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 9
  (Up)
SFFFFFFF
FFFF[41mF[0mFFF
FFFHFFFF
FFFFFHFF
FFFHFFFF
FHHFFFHF
FHFFHFHF
FFFHFFFG

t = 10
  (Up)
SFFFFFFF
FFFFF[41mF[0mFF
FFF[...]
{'fraction_non_alphanumeric': 0.1352692975924547, 'fraction_numerical': 0.02680565897244974, 'mean_word_length': 3.5660056657223795, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 16, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '217151', 'license': 'MIT', 'n_tokens_mistral': 3419, 'n_tokens_neox': 3231, 'n_words': 651, 'path': 'fastcampus/week2/1.Policy_Iteration.ipynb', 'repo_name': 'Junhojuno/Deep-Reinforcement-Learning', 'size': '8058'}