<jupyter_start><jupyter_text>Windy Gridworld Playground环境介绍- observation为格子所在的编号;- action的组成: 有4个动作, 分别是上下左右;    - 0, UP    - 1, RIGHT    - 2, DOWN    - 3, LEFT- reward: 每走一步reward=-1, reward越大也就是走的步数越少;      初始化环境<jupyter_code>environment = WindyGridworldEnv()
# 这个环境中可能动作的个数
nA = environment.action_space.n
print(nA)<jupyter_output>4<jupyter_text>策略的定义- $\mu$使用$\epsilon$-Greedy策略- $\pi$使用greedy策略<jupyter_code>def mu_policy(Q, epsilon, nA):
    """
    这是一个随机的策略, 执行每一个action的概率都是相同的.
    """
    def policy_fn(observation):
        # 看到这个state之后, 采取不同action获得的累计奖励
        action_values = Q[observation]
        # 使用获得奖励最大的那个动作
        greedy_action = np.argmax(action_values)
        # 是的每个动作都有出现的可能性
        probabilities = np.ones(nA) /nA * epsilon
        # 最好的那个动作的概率会大一些
        probabilities[greedy_action] = probabilities[greedy_action] + (1 - epsilon)
        
        return probabilities

    return policy_fn
def pi_policy(Q):
    """
    这是greedy policy, 每次选择最优的动作
    """
    def policy_fn(observation):
        action_values = Q[observation]
        best_action = np.argmax(action_values) # 选择最优的动作
        return np.eye(len(action_values))[best_action] # 返回的是两个动作出现的概率
        
    return policy_fn<jupyter_output><empty_output><jupyter_text>Importance Sampling for Off-Policy 2-step TD<jupyter_code>def td_control_importance_sampling(env, num_episodes, discount_factor=1.0, alpha=0.1, epsilon=0.2):
    # 环境中所有动作的数量
    nA = env.action_space.n

    # 初始化Q表
    Q = defaultdict(lambda: np.zeros(nA))
    
    # Keeps track of useful statistics
    stats = plotting.EpisodeStats(
        episode_lengths=np.zeros(num_episodes+1),
        episode_rewards=np.zeros(num_episodes+1)) 
    
    # 初始化police, 因为是off-policy, 所以有两个策略
    behaviour_policy = mu_policy(Q, epsilon, nA) # 这是我们实际执行action时候采取的策略, 这里使用随机游走
    policy = pi_policy(Q)

    for i_episode in range(1, num_episodes + 1):
        # 开始一轮游戏
        state = env.reset()        
        action = np.random.choice(nA, p=behaviour_policy(state)) # 从实际执行的policy, 选择action
        for t in itertools.count():
            env.s = state
            next_state, reward, done, _ = env.step(action) # 执行action, 返回reward和下一步的状态
            next_action = np.random.choice(nA, p=behaviour_policy(next_state)) # 选择下一步的动作
            next_next_state, next_reward, done, _ = env.step(next_action)
            next_action_pi = np.argmax(policy(next_next_state))

            # 计算两个概率
            pi_p = policy(next_state)[next_action] # 这里只有0或是1两个值, 完全确定的policy
            mu_p = behaviour_policy(next_state)[next_action]


            # 更新Q
            Q[state][action] = Q[state][action] + alpha * pi_p/mu_p * (reward + discount_factor*next_reward + (discount_factor**2)*Q[next_next_state][next_action_pi] - Q[state][action]) # 收敛
            # Q[state][action] = Q[state][action] + alpha * (reward + pi_p/mu_p*discount_factor*next_reward + (discount_factor**2)*Q[next_next_state][next_action_pi] - Q[state][action]) # 收敛
            # Q[state][action] = Q[state][action] + alpha * (pi_p/mu_p*(reward + discount_factor*next_reward + (discount_factor**2)*Q[next_next_state][next_action_pi]) - Q[state][action]) # 没收敛
            # Q[state][action] = Q[state][action] + alpha * (reward + discount_factor*next_reward + (discount_factor**2)*Q[next_next_state][next_action_pi] - Q[state][action]) # 收敛

            # 计算统计数据
            stats.episode_rewards[i_episode] += reward # 计算累计奖励
            stats.episode_lengths[i_episode] = t # 查看每一轮的时间

            if done:
                break
            
            if t > 500:
                break

            state = next_state
            action = next_action
        
        if i_episode % 100 == 0:
            # 打印
            print("\rEpisode {}/{}. | ".format(i_episode, num_episodes), end="")
    
    return Q, policy, stats<jupyter_output><empty_output><jupyter_text>开始模拟<jupyter_code>Q, policy, stats = td_control_importance_sampling(environment, num_episodes=10000, discount_factor=0.9, alpha=0.3, epsilon=0.5)
plotting.plot_episode_stats(stats)<jupyter_output><empty_output><jupyter_text>使用最终策略执行<jupyter_code>state = environment.reset()
action = np.argmax(policy(state))

for t in itertools.count():
    state, reward, done, _ = environment.step(action) # 执行action, 返回reward和下一步的状态 
    action = np.argmax(policy(state)) # 查看新的action

    print('-Step:{}-'.format(str(t)))
    environment._render() # 显示结果

    if done:
        break

    if t > 50:
        break<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11764705882352941, 'fraction_numerical': 0.009951348960636886, 'mean_word_length': 3.1193078324225865, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '108701', 'license': 'MIT', 'n_tokens_mistral': 1873, 'n_tokens_neox': 1918, 'n_words': 250, 'path': 'Reinforcement_Learning/Week_05/05_Importance_Sampling_2-step-Q-Learning_Windy_Gridworld.ipynb', 'repo_name': 'wmn7/PyTorch-Practice', 'size': '4522'}