<jupyter_start><jupyter_text>Temporal Difference: On-policy n-Tuple Expected Sarsa, Stochastic<jupyter_code>import numpy as np<jupyter_output><empty_output><jupyter_text>Create environment<jupyter_code>def create_environment_states():
    """Creates environment states.

    Returns:
        num_states: int, number of states.
        num_terminal_states: int, number of terminal states.
        num_non_terminal_states: int, number of non terminal states.
    """
    num_states = 16
    num_terminal_states = 2
    num_non_terminal_states = num_states - num_terminal_states

    return num_states, num_terminal_states, num_non_terminal_states
def create_environment_actions(num_non_terminal_states):
    """Creates environment actions.

    Args:
        num_non_terminal_states: int, number of non terminal states.

    Returns:
        max_num_actions: int, max number of actions possible.
        num_actions_per_non_terminal_state: array[int], number of actions per
            non terminal state.
    """
    max_num_actions = 4

    num_actions_per_non_terminal_state = np.repeat(
        a=max_num_actions, repeats=num_non_terminal_states)

    return max_num_actions, num_actions_per_non_terminal_state
def create_environment_successor_counts(num_states, max_num_actions):
    """Creates environment successor counts.

    Args:
        num_states: int, number of states.
        max_num_actions: int, max number of actions possible.
    Returns:
        num_state_action_successor_states: array[int], number of successor
            states s' that can be reached from state s by taking action a.
    """
    num_state_action_successor_states = np.repeat(
        a=1, repeats=num_states * max_num_actions)

    num_state_action_successor_states = np.reshape(
        a=num_state_action_successor_states,
        newshape=(num_states, max_num_actions))

    return num_state_action_successor_states
def create_environment_successor_arrays(
        num_non_terminal_states, max_num_actions):
    """Creates environment successor arrays.

    Args:
        num_non_terminal_states: int, number of non terminal states.
        max_num_actions: int, max number of actions possible.
    Returns:
        sp_idx: array[int], state indices of new state s' of taking action a
            from state s.
        p: array[float], transition probability to go from state s to s' by
            taking action a.
        r: array[float], reward from new state s' from state s by taking
            action a.
    """
    sp_idx = np.array(
        object=[1, 0, 14, 4,
                2, 1, 0, 5,
                2, 2, 1, 6,
                4, 14, 3, 7,
                5, 0, 3, 8,
                6, 1, 4, 9,
                6, 2, 5, 10,
                8, 3, 7, 11,
                9, 4, 7, 12,
                10, 5, 8, 13,
                10, 6, 9, 15,
                12, 7, 11, 11,
                13, 8, 11, 12,
                15, 9, 12, 13],
        dtype=np.int64)

    p = np.repeat(
        a=1.0, repeats=num_non_terminal_states * max_num_actions * 1)

    r = np.repeat(
        a=-1.0, repeats=num_non_terminal_states * max_num_actions * 1)

    sp_idx = np.reshape(
        a=sp_idx,
        newshape=(num_non_terminal_states, max_num_actions, 1))
    p = np.reshape(
        a=p,
        newshape=(num_non_terminal_states, max_num_actions, 1))
    r = np.reshape(
        a=r,
        newshape=(num_non_terminal_states, max_num_actions, 1))

    return sp_idx, p, r
def create_environment():
    """Creates environment.

    Returns:
        num_states: int, number of states.
        num_terminal_states: int, number of terminal states.
        num_non_terminal_states: int, number of non terminal states.
        max_num_actions: int, max number of actions possible.
        num_actions_per_non_terminal_state: array[int], number of actions per
            non terminal state.
        num_state_action_successor_states: array[int], number of successor
            states s' that can be reached from state s by taking action a.
        sp_idx: array[int], state indices of new state s' of taking action a
            from state s.
        p: array[float], transition probability to go from state s to s' by
            taking action a.
        r: array[float], reward from new state s' from state s by taking
            action a.
    """
    (num_states,
     num_terminal_states,
     num_non_terminal_states) = create_environment_states()

    (max_num_actions,
     num_actions_per_non_terminal_state) = create_environment_actions(
        num_non_terminal_states)

    num_state_action_successor_states = create_environment_successor_counts(
        num_states, max_num_actions)

    (sp_idx,
     p,
     r) = create_environment_successor_arrays(
        num_non_terminal_states, max_num_actions)

    return (num_states,
            num_terminal_states,
            num_non_terminal_states,
            max_num_actions,
            num_actions_per_non_terminal_state,
            num_state_action_successor_states,
            sp_idx,
            p,
            r)<jupyter_output><empty_output><jupyter_text>Set hyperparameters<jupyter_code>def set_hyperparameters():
    """Sets hyperparameters.

    Returns:
        num_episodes: int, number of episodes to train over.
        maximum_episode_length: int, max number of timesteps for an episode.
        num_qs: int, number of state-action-value functions Q_i(s, a).
        alpha: float, alpha > 0, learning rate.
        epsilon: float, 0 <= epsilon <= 1, exploitation-exploration trade-off,
            higher means more exploration.
        gamma: float, 0 <= gamma <= 1, amount to discount future reward.
    """
    num_episodes = 10000
    maximum_episode_length = 200
    num_qs = 3
    alpha = 0.1
    epsilon = 0.1
    gamma = 1.0

    return num_episodes, maximum_episode_length, num_qs, alpha, epsilon, gamma<jupyter_output><empty_output><jupyter_text>Create value function and policy arrays<jupyter_code>def create_value_function_arrays(num_qs, num_states, max_num_actions):
    """Creates value function arrays.

    Args:
        num_qs: int, number of state-action-value functions Q_i(s, a).
        num_states: int, number of states.
        max_num_actions: int, max number of actions possible.
    Returns:
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
    """
    q = np.repeat(a=0.0, repeats=num_qs * num_states * max_num_actions)
    q = np.reshape(a=q, newshape=(num_qs, num_states, max_num_actions))

    return q
def create_policy_arrays(num_non_terminal_states, max_num_actions):
    """Creates policy arrays.

    Args:
        num_non_terminal_states: int, number of non terminal states.
        max_num_actions: int, max number of actions possible.
    Returns:
        policy: array[float], learned stochastic policy of which
            action a to take in state s.
    """
    policy = np.repeat(
        a=1.0 / max_num_actions,
        repeats=num_non_terminal_states * max_num_actions)

    policy = np.reshape(
        a=policy,
        newshape=(num_non_terminal_states, max_num_actions))

    return policy<jupyter_output><empty_output><jupyter_text>Create algorithm<jupyter_code># Set random seed so that everything is reproducible
np.random.seed(seed=0)
def initialize_epsiode(num_non_terminal_states):
    """Initializes epsiode with initial state.

    Args:
        num_non_terminal_states: int, number of non terminal states.
    Returns:
        init_s_idx: int, initial state index from set of non terminal states.
    """
    # Randomly choose an initial state from all non-terminal states
    init_s_idx = np.random.randint(
        low=0, high=num_non_terminal_states, dtype=np.int64)

    return init_s_idx
def epsilon_greedy_policy_from_state_action_function(
        max_num_actions, q, epsilon, s_idx, policy):
    """Create epsilon-greedy policy from state-action value function.

    Args:
        max_num_actions: int, max number of actions possible.
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
        epsilon: float, 0 <= epsilon <= 1, exploitation-exploration trade-off,
            higher means more exploration.
        s_idx: int, current state index.
        policy: array[float], learned stochastic policy of which action a to
            take in state s.
    Returns:
        policy: array[float], learned stochastic policy of which action a to
            take in state s.
    """
    # Combine state-action value functions
    q = np.sum(a=q[:, s_idx, :], axis=0)

    # Save max state-action value and find the number of actions that have the
    # same max state-action value
    max_action_value = np.max(a=q)
    max_action_count = np.count_nonzero(a=q == max_action_value)

    # Apportion policy probability across ties equally for state-action pairs
    # that have the same value and zero otherwise
    if max_action_count == max_num_actions:
        max_policy_prob_per_action = 1.0 / max_action_count
        remain_prob_per_action = 0.0
    else:
        max_policy_prob_per_action = (1.0 - epsilon) / max_action_count
        remain_prob_per_action = epsilon / (max_num_actions - max_action_count)

    policy[s_idx, :] = np.where(
        q == max_action_value,
        max_policy_prob_per_action,
        remain_prob_per_action)

    return policy
def loop_through_episode(
        num_non_terminal_states,
        max_num_actions,
        num_state_action_successor_states,
        sp_idx,
        p,
        r,
        num_qs,
        q,
        policy,
        alpha,
        epsilon,
        gamma,
        maximum_episode_length,
        s_idx):
    """Loops through episode to iteratively update policy.

    Args:
        num_non_terminal_states: int, number of non terminal states.
        max_num_actions: int, max number of actions possible.
        num_state_action_successor_states: array[int], number of successor
            states s' that can be reached from state s by taking action a.
        sp_idx: array[int], state indices of new state s' of taking action a
            from state s.
        p: array[float], transition probability to go from state s to s' by
            taking action a.
        r: array[float], reward from new state s' from state s by taking
            action a.
        num_qs: int, number of state-action-value functions Q_i(s, a).
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
        policy: array[float], learned stochastic policy of which
            action a to take in state s.
        alpha: float, alpha > 0, learning rate.
        epsilon: float, 0 <= epsilon <= 1, exploitation-exploration trade-off,
            higher means more exploration.
        gamma: float, 0 <= gamma <= 1, amount to discount future reward.
        maximum_episode_length: int, max number of timesteps for an episode.
        s_idx: int, current state index.
    Returns:
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
        policy: array[float], learned stochastic policy of which
            action a to take in state s.
    """
    # Loop through episode steps until termination
    for t in range(0, maximum_episode_length):
        # Choose policy for chosen state by epsilon-greedy choosing from the
        # state-action-value function
        policy = epsilon_greedy_policy_from_state_action_function(
            max_num_actions, q, epsilon, s_idx, policy)

        # Get epsilon-greedy action
        a_idx = np.random.choice(
            a=max_num_actions, p=policy[s_idx, :])

        # Get reward
        successor_state_transition_idx = np.random.choice(
            a=num_state_action_successor_states[s_idx, a_idx],
            p=p[s_idx, a_idx, :])

        reward = r[s_idx, a_idx, successor_state_transition_idx]

        # Get next state
        next_s_idx = sp_idx[s_idx, a_idx, successor_state_transition_idx]

        # Update state action value equally randomly selecting from the
        # state-action-value functions
        updating_q_idx = np.random.randint(low=0, high=num_qs, dtype=np.int64)

        q, policy, s_idx = update_q(
            num_non_terminal_states,
            max_num_actions,
            policy,
            alpha,
            epsilon,
            gamma,
            s_idx,
            a_idx,
            reward,
            next_s_idx,
            updating_q_idx,
            num_qs,
            q)

        if next_s_idx >= num_non_terminal_states:
            break  # episode terminated since we ended up in a terminal state

    return q, policy
def update_q(
        num_non_terminal_states,
        max_num_actions,
        policy,
        alpha,
        epsilon,
        gamma,
        s_idx,
        a_idx,
        reward,
        next_s_idx,
        updating_q_idx,
        num_qs,
        q):
    """Updates state-action-value function using multiple estimates.

    Args:
        num_non_terminal_states: int, number of non terminal states.
        max_num_actions: int, max number of actions possible.
        policy: array[float], learned stochastic policy of which
            action a to take in state s.
        alpha: float, alpha > 0, learning rate.
        epsilon: float, 0 <= epsilon <= 1, exploitation-exploration trade-off,
            higher means more exploration.
        gamma: float, 0 <= gamma <= 1, amount to discount future reward.
        s_idx: int, current state index.
        a_idx: int, current action index.
        reward: float, current reward from taking action a_idx in state s_idx.
        next_s_idx: int, next state index.
        updating_q_idx: int, index to which Q_i(s, a) we'll be updating.
        num_qs: int, number of state-action-value functions Q_i(s, a).
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
    Returns:
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
        policy: array[float], learned stochastic policy of which
            action a to take in state s.
        s_idx: int, new current state index.
    """
    # Check to see if we actioned into a terminal state
    if next_s_idx >= num_non_terminal_states:
        delta = reward - q[updating_q_idx, s_idx, a_idx]
        q[updating_q_idx, s_idx, a_idx] += alpha * delta
    else:
        # Get next action, using expectation value
        q_indices = np.arange(num_qs)

        not_updating_q_idx = np.random.choice(
            a=np.extract(condition=q_indices != updating_q_idx, arr=q_indices))

        not_updating_v_expected_value_on_policy = np.sum(
            a=policy[next_s_idx, :] * q[not_updating_q_idx, next_s_idx, :])

        # Calculate state-action-function expectation
        delta = gamma * not_updating_v_expected_value_on_policy
        delta -= q[updating_q_idx, s_idx, a_idx]
        q[updating_q_idx, s_idx, a_idx] += alpha * (reward + delta)

        # Update state and action to next state and action
        s_idx = next_s_idx

    return q, policy, s_idx
def on_policy_temporal_difference_n_tuple_expected_sarsa(
        num_non_terminal_states,
        max_num_actions,
        num_state_action_successor_states,
        sp_idx,
        p,
        r,
        num_qs,
        q,
        policy,
        alpha,
        epsilon,
        gamma,
        maximum_episode_length,
        num_episodes):
    """Loops through episodes to iteratively update policy.

    Args:
        num_non_terminal_states: int, number of non terminal states.
        max_num_actions: int, max number of actions possible.
        num_state_action_successor_states: array[int], number of successor
            states s' that can be reached from state s by taking action a.
        sp_idx: array[int], state indices of new state s' of taking action a
            from state s.
        p: array[float], transition probability to go from state s to s' by
            taking action a.
        r: array[float], reward from new state s' from state s by taking
            action a.
        num_qs: int, number of state-action-value functions Q_i(s, a).
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
        policy: array[float], learned stochastic policy of which
            action a to take in state s.
        alpha: float, alpha > 0, learning rate.
        epsilon: float, 0 <= epsilon <= 1, exploitation-exploration trade-off,
            higher means more exploration.
        gamma: float, 0 <= gamma <= 1, amount to discount future reward.
        maximum_episode_length: int, max number of timesteps for an episode.
        num_episodes: int, number of episodes to train over.
    Returns:
        q: array[float], keeps track of the estimated value of each
            state-action pair Q_i(s, a).
        policy: array[float], learned stochastic policy of which
            action a to take in state s.
    """
    for episode in range(0, num_episodes):
        # Initialize episode to get initial state
        init_s_idx = initialize_epsiode(num_non_terminal_states)

        # Loop through episode and update the policy
        q, policy = loop_through_episode(
            num_non_terminal_states,
            max_num_actions,
            num_state_action_successor_states,
            sp_idx,
            p,
            r,
            num_qs,
            q,
            policy,
            alpha,
            epsilon,
            gamma,
            maximum_episode_length,
            init_s_idx)

    return q, policy<jupyter_output><empty_output><jupyter_text>Run algorithm<jupyter_code>def run_algorithm():
    """Runs the algorithm."""
    (num_states,
     num_terminal_states,
     num_non_terminal_states,
     max_num_actions,
     num_actions_per_non_terminal_state,
     num_state_action_successor_states,
     sp_idx,
     p,
     r) = create_environment()

    (num_episodes,
     maximum_episode_length,
     num_qs,
     alpha,
     epsilon,
     gamma) = set_hyperparameters()

    q = create_value_function_arrays(num_qs, num_states, max_num_actions)

    policy = create_policy_arrays(num_non_terminal_states, max_num_actions)

    # Print initial arrays
    print("\nInitial state-action value function")
    print(q)

    print("\nInitial policy")
    print(policy)

    # Run on policy temporal difference n-tuple expected sarsa
    q, policy = on_policy_temporal_difference_n_tuple_expected_sarsa(
        num_non_terminal_states,
        max_num_actions,
        num_state_action_successor_states,
        sp_idx,
        p,
        r,
        num_qs,
        q,
        policy,
        alpha,
        epsilon,
        gamma,
        maximum_episode_length,
        num_episodes)

    # Print final results
    print("\nFinal state-action value function")
    print(q)

    print("\nFinal policy")
    print(policy)
run_algorithm()<jupyter_output>Initial state-action value function
[[[0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]]

 [[0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]]

 [[0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]
  [0. 0. 0. 0.]]]

Initial policy
[[0.25 0.25 0.25 0.25]
 [0.25 0.25 0.25 0.25]
 [0.25 0.25 0.25 0.25]
 [0.25 0.25 0.25 0.25]
 [0.25 0.25 0.25 0.25]
 [0.25 0.25 0.25 0.25]
 [0.25 0.25 0.25 0.25]
 [0.25 0.25 [...]
{'fraction_non_alphanumeric': 0.08561047235080384, 'fraction_numerical': 0.021352844557264446, 'mean_word_length': 2.2765818656229615, 'pattern_counts': {'":': 0, '<': 42, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, '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': '12552', 'license': 'Apache-2.0', 'n_tokens_mistral': 6770, 'n_tokens_neox': 6252, 'n_words': 1923, 'path': 'machine_learning/reinforcement_learning/generalized_stochastic_policy_iteration/tabular/temporal_difference/np_temporal_difference/on_policy_stochastic_temporal_difference_n_tuple_expected_sarsa.ipynb', 'repo_name': 'ryangillard/artificial_intelligence', 'size': '20091'}