Skip to content

Algorithm API

Dynamic programming

gym_classics2.algorithms.dynamic_programming

This file implements dynamic programming algorithms for solving Markov Decision Processes (MDPs) in gym-classics environments with model access. The algorithms include value iteration and policy iteration, which are fundamental methods in reinforcement learning for computing optimal policies and value functions.

backup 

backup(env, discount, V, state, action)

Computes the Bellman backup for a given state and action.

Parameters:

Name Type Description Default
env

A gym-classics environment with model access.

 required
discount

The discount factor.

 required
V

The current value function.

 required
state

The current state.

 required
action

The action to evaluate.

 required

Returns: The computed Q-value for the given state and action.

Source code in gym_classics2/algorithms/dynamic_programming.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def backup(env, discount, V, state, action):
    """Computes the Bellman backup for a given state and action.

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor.
        V: The current value function.
        state: The current state.
        action: The action to evaluate.
    Returns:
        The computed Q-value for the given state and action.
    """

    V = np.array(V)

    next_states, rewards, terminals, probs = env.model(state, action)
    bootstraps = (1.0 - terminals) * V[next_states]
    return np.sum(probs * (rewards + discount * bootstraps))

value_iteration 

value_iteration(env, discount, precision=0.001, history=False, verbose=False)

Performs value iteration for the given environment.

Parameters:

Name Type Description Default
env

A gym-classics environment with model access.

 required
discount

The discount factor (0 <= discount <= 1).

 required
precision

The precision for convergence (default: 1e-3).

 0.001
history

If True, returns a list of intermediate value functions.

 False
verbose

If True, prints progress information.

 False

Returns:

Type Description

The optimal value function V. If history is True, returns a list of intermediate value functions.
Source code in gym_classics2/algorithms/dynamic_programming.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def value_iteration(env, discount, precision=1e-3, history = False, verbose = False):
    """Performs value iteration for the given environment.

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        precision: The precision for convergence (default: 1e-3).
        history: If True, returns a list of intermediate value functions.
        verbose: If True, prints progress information.

    Returns:
        The optimal value function V. If history is True, returns a list of intermediate value functions.
    """

    assert isinstance(env, GymClassicsBaseEnv), "Value iteration requires a gym-classics environment with model access to the environment." 
    assert 0.0 <= discount <= 1.0
    assert precision > 0.0

    V = np.zeros(len(env.states()), dtype=np.float64)  
    if history:
        V_list = []
        V_list.append(V.copy())

    sweeps = 0
    progress = tqdm(total=None, desc="Value Iteration", disable=verbose)
    while True:
        progress.update()
        if verbose:
            print('.', end = '')
            sweeps += 1

        V_old = V.copy()

        for s in env.states():
            Q_values = [backup(env, discount, V, s, a) for a in range(len(env.actions()))]
            V[s] = np.max(Q_values)

        if history:
            V_list.append(V.copy())

        if np.abs(V - V_old).max() <= precision:
            break

    if verbose:
        print(f'\nConverged after {sweeps} sweeps.')

    if history:
        return V_list 

    return V

policy_evaluation 

policy_evaluation(env, discount, policy, precision=0.001, max_backups=1000)

Evaluates a given policy to compute its value function.

Parameters:

Name Type Description Default
env

A gym-classics environment with model access.

 required
discount

The discount factor (0 <= discount <= 1).

 required
policy

The policy to evaluate.

 required
precision

The precision for convergence (default: 1e-3).

 0.001
max_backups

Maximum number of backups to perform to prevent infinite loops (default: 1000).

 1000

Returns:

Type Description

The value function for the given policy.
Source code in gym_classics2/algorithms/dynamic_programming.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def policy_evaluation(env, discount, policy, precision=1e-3, max_backups=1000):
    """Evaluates a given policy to compute its value function.

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        policy: The policy to evaluate.
        precision: The precision for convergence (default: 1e-3).
        max_backups: Maximum number of backups to perform to prevent infinite loops (default: 1000).

    Returns:
        The value function for the given policy.
    """

    assert isinstance(env, GymClassicsBaseEnv), "Value iteration requires a gym-classics environment with model access to the environment." 
    assert 0.0 <= discount <= 1.0
    assert precision > 0.0

    V = np.zeros(len(policy), dtype=np.float64)

    while True:
        V_old = V.copy()

        for s in env.states():
            V[s] = backup(env, discount, V, s, policy[s])

        if np.abs(V - V_old).max() <= precision or max_backups <= 0:
            break

        max_backups -= 1
    return V

policy_improvement 

policy_improvement(policy, V_policy, env, discount, precision=0.001)

Improves the policy based on the given value function.

Parameters:

Name Type Description Default
policy

The current policy to improve.

 required
V_policy

The value function of the current policy.

 required
env

A gym-classics environment with model access.

 required
discount

The discount factor (0 <= discount <= 1).

 required
precision

The precision for determining stability (default: 1e-3).

 0.001

Returns:

Type Description

A tuple (improved_policy, stable) where stable is True if the policy did not change.
Source code in gym_classics2/algorithms/dynamic_programming.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def policy_improvement(policy, V_policy, env, discount, precision=1e-3):
    """Improves the policy based on the given value function.

    Args:
        policy: The current policy to improve.
        V_policy: The value function of the current policy.
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        precision: The precision for determining stability (default: 1e-3).

    Returns:
        A tuple (improved_policy, stable) where stable is True if the policy did not change.
    """

    policy_old = policy.copy()
    V_old = V_policy.copy()

    for s in env.states():
        Q_values = [backup(env, discount, V_policy, s, a) for a in env.actions()]
        policy[s] = np.argmax(Q_values)
        V_policy[s] = max(Q_values)

    stable = np.logical_or(
        policy == policy_old,
        np.abs(V_policy - V_old).max() <= precision,
    ).all()

    return policy, stable

policy_iteration 

policy_iteration(env, discount, precision=0.001, max_backups=1000, history=False, verbose=False, rng=None)

Performs policy iteration for the given environment.

Parameters:

Name Type Description Default
env

A gym-classics environment with model access.

 required
discount

The discount factor (0 <= discount <= 1).

 required
precision

The precision for convergence (default: 1e-3).

 0.001
max_backups

Maximum number of iterations used in policy evaluation. Note: this prevents an infinite loop for policies that do not reach a terminal state.

 1000
history

If True, returns lists of intermediate policies and value functions.

 False
verbose

If True, prints progress information.

 False
rng

NumPy generator or integer seed used to initialize the policy.

 None

Returns:

Type Description

The optimal policy. If history is True, returns a tuple (policy_list, V_list) containing lists of intermediate policies and value functions.
Source code in gym_classics2/algorithms/dynamic_programming.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def policy_iteration(env, discount, precision=1e-3, max_backups=1000, history=False, verbose=False, rng=None):
    """Performs policy iteration for the given environment.

    Args:
        env: A gym-classics environment with model access.
        discount: The discount factor (0 <= discount <= 1).
        precision: The precision for convergence (default: 1e-3).
        max_backups: Maximum number of iterations used in policy evaluation. Note: this prevents an infinite loop for policies that do not reach a terminal state.
        history: If True, returns lists of intermediate policies and value functions.
        verbose: If True, prints progress information.
        rng: NumPy generator or integer seed used to initialize the policy.

    Returns:
        The optimal policy. If history is True, returns a tuple (policy_list, V_list) containing lists of intermediate policies and value functions.
    """

    assert isinstance(env, GymClassicsBaseEnv), "Value iteration requires a gym-classics environment with model access to the environment." 
    assert 0.0 <= discount <= 1.0
    assert precision > 0.0

    policy = random_policy(env, rng=rng)

    if history:
        pol_list = []
        pol_list.append(policy.copy())
        V_list = []

    iterations = 0
    progress = tqdm(total=None, desc="Policy Iteration", disable=verbose)
    while True:
        progress.update()
        if verbose:
            print('.', end = '')
            iterations += 1

        V_policy = policy_evaluation(env, discount, policy, precision, max_backups)
        if history:
            V_list.append(V_policy.copy())

        policy, stable = policy_improvement(policy, V_policy, env, discount, precision)

        if stable:
            break

        if history:
            pol_list.append(policy.copy())

    if verbose:
        print(f'\nConverged after {iterations} iterations.')

    if history:
        return pol_list, V_list

    return policy

Policy helpers

gym_classics2.algorithms.policy

This file implements different policy representations and functions for working with policies in gym-classics environments. It includes functions for creating random policies, encoding policies for display, and computing greedy policies based

make_multidiscrete_policy 

make_multidiscrete_policy(policy, env)

Converts a tabular policy vector to a multi-discrete tabular policy stored in a dictionary that can be used for sample.

Source code in gym_classics2/algorithms/policy.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
def make_multidiscrete_policy(policy, env):
    """
    Converts a tabular policy vector to a multi-discrete tabular policy stored in a dictionary that can be used for sample.
    """
    assert isinstance(env.observation_space, gym.spaces.MultiDiscrete), "Requires an environment with a multi-discrete state spaces."

    if isinstance(policy, dict):
        return policy

    if isinstance(env, GymClassicsBaseEnv):
        return dict(zip(env.id2state(list(env.states())), [policy[s] for s in env.states()]))
    else:
        raise ValueError("Unsupported environment type")

random_policy 

random_policy(env, rng=None)

Create a random policy for the given environment. The policy is represented as a numpy array where each entry corresponds to an action for a state. rng may be a NumPy generator or an integer seed.

Source code in gym_classics2/algorithms/policy.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def random_policy(env, rng=None):
    """
    Create a random policy for the given environment. 
    The policy is represented as a numpy array where each entry corresponds to an action for a state.
    ``rng`` may be a NumPy generator or an integer seed.
    """

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        return rng.integers(env.action_space.n, size=env.observation_space.n)
    else:
        assert isinstance(env, GymClassicsBaseEnv), "Only gym-classics environments are supported for random policies with multi-discrete state spaces."
        return make_multidiscrete_policy(rng.integers(env.action_space.n, size=len(env.states())), env)

encode_policy 

encode_policy(policy, env, type='text')

Encode a policy for display. The policy is represented as a numpy array where each entry corresponds to an action for a state. The function returns a list of action names corresponding to the actions in the policy.

Source code in gym_classics2/algorithms/policy.py 
43
44
45
46
47
48
49
50
51
def encode_policy(policy, env, type = "text"):
    """
    Encode a policy for display. The policy is represented as a numpy array where each entry corresponds to an action for a state.
    The function returns a list of action names corresponding to the actions in the policy.
    """

    assert isinstance(env, GymClassicsBaseEnv)

    return [env.unwrapped.id2action(a, type = type) for a in policy]

greedy_policy 

greedy_policy(V, env, discount=1, rng=None)

Calculate a greedy policy from a state-value function.

Parameters:

Name Type Description Default
V

One-dimensional state-value array.

 required
env

Environment with a discrete state space and model access.

 required
discount

Discount factor in [0, 1].

 1
rng

NumPy generator or integer seed for random tie-breaking.

 None

Returns:

Type Description

Integer action ID selected for each state.
Source code in gym_classics2/algorithms/policy.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def greedy_policy(V, env, discount=1, rng=None):
    """Calculate a greedy policy from a state-value function.

    Args:
        V: One-dimensional state-value array.
        env: Environment with a discrete state space and model access.
        discount: Discount factor in ``[0, 1]``.
        rng: NumPy generator or integer seed for random tie-breaking.

    Returns:
        Integer action ID selected for each state.
    """
    assert isinstance(env, GymClassicsBaseEnv), "greedy_policy requires a gym-classics environment with discrete state space."
    assert isinstance(env.action_space, gym.spaces.Discrete)
    assert isinstance(env.observation_space, gym.spaces.Discrete)
    assert 0.0 <= discount <= 1.0

    policy = np.zeros(len(env.states()), dtype=np.int64)
    rng = get_rng(rng)

    env = env.unwrapped

    for s in env.states():
        Q_values = [_backup(env, discount, V, s, a) for a in range(env.action_space.n)]
        policy[s] = random_argmax(Q_values, rng=rng)

    return policy

greedy_policy_Q 

greedy_policy_Q(Q, env, discount=1, rng=None)

Calculate a greedy policy from an action-value function.

Parameters:

Name Type Description Default
Q

Two-dimensional action-value array indexed by state and action.

 required
env

Environment with discrete observation and action spaces.

 required
discount

Unused; retained for API compatibility with greedy_policy.

 1
rng

NumPy generator or integer seed for random tie-breaking.

 None

Returns:

Type Description

Integer action ID selected for each state.
Source code in gym_classics2/algorithms/policy.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def greedy_policy_Q(Q, env, discount=1, rng=None):
    """Calculate a greedy policy from an action-value function.

    Args:
        Q: Two-dimensional action-value array indexed by state and action.
        env: Environment with discrete observation and action spaces.
        discount: Unused; retained for API compatibility with ``greedy_policy``.
        rng: NumPy generator or integer seed for random tie-breaking.

    Returns:
        Integer action ID selected for each state.
    """

    assert isinstance(env.action_space, gym.spaces.Discrete)
    assert isinstance(env.observation_space, gym.spaces.Discrete)
    assert 0.0 <= discount <= 1.0

    return random_argmax(Q, axis=1, rng=rng)

epsilon_greedy_action 

epsilon_greedy_action(policy, state=None, epsilon=0, rng=None)

Select an epsilon-greedy action from tabular action values.

Parameters:

Name Type Description Default
policy

Action values for all actions, optionally indexed first by state.

 required
state

Current state index. If omitted, policy is treated as the action-value vector for the current state.

 None
epsilon

Probability of selecting a uniformly random action.

 0
rng

NumPy generator or integer seed.

 None

Returns:

Type Description

Selected integer action ID.
Source code in gym_classics2/algorithms/policy.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def epsilon_greedy_action(policy, state=None, epsilon=0, rng=None):
    """Select an epsilon-greedy action from tabular action values.

    Args:
        policy: Action values for all actions, optionally indexed first by state.
        state: Current state index. If omitted, ``policy`` is treated as the
            action-value vector for the current state.
        epsilon: Probability of selecting a uniformly random action.
        rng: NumPy generator or integer seed.

    Returns:
        Selected integer action ID.
    """

    rng = get_rng(rng)

    if epsilon > 0 and rng.random() < epsilon:
        return rng.integers(len(policy)) if state is None else rng.integers(len(policy[state]))

    if state is None:
        return random_argmax(policy, rng=rng)
    else:
        return random_argmax(policy[state], rng=rng)

Monte Carlo methods

gym_classics2.algorithms.monte_carlo_methods

This file implements tabular Monte Carlo methods for policy evaluation and control in gym-classics environments with discrete state spaces.

sample_episode 

sample_episode(env, policy=None, start_state=None, start_action=None, epsilon=0, max_len=1000, verbose=False, rng=None)

Samples an episode from the environment using the given policy and starting conditions.

Parameters:

Name Type Description Default
env

The environment to sample from.

 required
policy

A mapping from states to actions. For discrete state spaces, use an action list in state order. If None, a random policy is used.

 None
start_state

The state to start the episode from (if None, the environment's default starting state will be used).

 None
start_action

The action to take in the first step of the episode (if None, the action will be chosen according to the policy or randomly if no policy is given).

 None
epsilon

The probability of taking a random action instead of the policy's action at each step (for epsilon-greedy exploration).

 0
max_len

The maximum length of the episode to prevent infinite loops.

 1000
verbose

If True, prints the state transitions and rewards for each step in the episode.

 False
rng

NumPy generator or integer seed for policy and exploration choices.

 None

Returns:

Type Description

A list of (state, action, reward, next_state) tuples representing the episode.

The episode ends when a terminal state is reached or when max_len steps have been taken.

Note that the last tuple in the episode will have a next_state that is either terminal or the

state at which the episode was truncated due to max_len.
Source code in gym_classics2/algorithms/monte_carlo_methods.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def sample_episode(env, policy=None, start_state=None, start_action=None, epsilon=0,
                   max_len=1000, verbose=False, rng=None):
    """
    Samples an episode from the environment using the given policy and starting conditions.

    Args: 
        env: The environment to sample from.
        policy: A mapping from states to actions. For discrete state spaces, use
            an action list in state order. If ``None``, a random policy is used.
        start_state: The state to start the episode from (if None, the environment's default starting state will be used).
        start_action: The action to take in the first step of the episode (if None, the action will be chosen according to the policy or randomly if no policy is given).
        epsilon: The probability of taking a random action instead of the policy's action at each step (for epsilon-greedy exploration).
        max_len: The maximum length of the episode to prevent infinite loops.
        verbose: If True, prints the state transitions and rewards for each step in the episode.
        rng: NumPy generator or integer seed for policy and exploration choices.

    Returns:
        A list of (state, action, reward, next_state) tuples representing the episode. 
        The episode ends when a terminal state is reached or when max_len steps have been taken.    
        Note that the last tuple in the episode will have a next_state that is either terminal or the 
        state at which the episode was truncated due to max_len.  
    """

    assert isinstance(env.observation_space, gym.spaces.Discrete) or isinstance(env.observation_space, gym.spaces.MultiDiscrete), "Tabular methods require discrete state space."  
    assert 0.0 <= epsilon <= 1.0
    assert max_len > 0

    rng = get_rng(rng)

    if policy is None:
        policy = random_policy(env, rng=rng)
        epsilon = 1.0
        if verbose:
            print("*** No policy given, sampling using random actions!")

    if isinstance(env.observation_space, gym.spaces.MultiDiscrete):
        policy = make_multidiscrete_policy(policy, env)

    episode = []
    s, r = env.reset()

    # force custom start state could be a state or an state index.
    if not start_state is None:
        if isinstance(env, BaseEnv) and env.tabular:
            env.state = env.id2state(start_state)
        elif isinstance(env.observation_space, gym.spaces.Discrete):
            env.state = int(start_state)
        else:
            raise ValueError("Custom start state is only supported for tabular gym-classics environments and environments with a MultiDiscrete observation space.")
        s = start_state

    step = 0
    done = False
    while not done and step < max_len:
        a = policy[s]

        # epsilon greedy choice?
        if epsilon > 0 and rng.random() < epsilon:
            a = rng.integers(env.action_space.n)

        # for exploring starts
        if step == 0 and not start_action is None:
            a = start_action

        sp, reward, done, _, _ = env.step(a)

        episode.append((s,a,reward,sp))

        if verbose:
            print ("Step", step, "(s,a,r,s')=",s,a,reward,sp)

        s = sp
        step += 1

    return(episode)

on_policy_state_distribution 

on_policy_state_distribution(policy, env, discount=1, epsilon=0, n=100, verbose=False, rng=None)

Estimate a policy's state distribution using rng for exploration.

Source code in gym_classics2/algorithms/monte_carlo_methods.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def on_policy_state_distribution(policy, env, discount=1, epsilon=0, n=100,
                                 verbose=False, rng=None):
    """Estimate a policy's state distribution using ``rng`` for exploration."""

    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."   
    assert 0.0 <= epsilon <= 1.0
    assert 0.0 < discount <= 1.0
    assert n > 0

    rng = get_rng(rng)
    state_counts = np.zeros(env.observation_space.n)

    for _ in tqdm(range(n), desc="Sampling Episodes", disable=verbose):
        episode = np.array(sample_episode(env, policy=policy, epsilon=epsilon, rng=rng))
        states = np.append(episode[:,0], episode[-1,3])
        discounts = np.array([discount**i for i in range(len(states))])
        for s, d in zip(states, discounts):
            state_counts[int(s)] += d

    state_prob = state_counts / sum(state_counts)
    return state_prob

MC_prediction 

MC_prediction(policy, env, discount, n=100, max_episode_len=100, verbose=False, rng=None)

Estimate a policy's state values with first-visit Monte Carlo prediction.

Parameters:

Name Type Description Default
policy

Array-like mapping from state IDs to action IDs.

 required
env

Gymnasium environment with a discrete observation space.

 required
discount

Reward discount factor.

 required
n

Number of episodes to sample.

 100
max_episode_len

Maximum sampled steps per episode.

 100
verbose

Print episode progress when true.

 False
rng

NumPy generator or integer seed for episode sampling.

 None

Returns:

Type Description

A NumPy value array with one entry per state. Unvisited states contain

numpy.nan.
Source code in gym_classics2/algorithms/monte_carlo_methods.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def MC_prediction(policy, env, discount, n=100, max_episode_len=100,
                  verbose=False, rng=None):
    """Estimate a policy's state values with first-visit Monte Carlo prediction.

    Args:
        policy: Array-like mapping from state IDs to action IDs.
        env: Gymnasium environment with a discrete observation space.
        discount: Reward discount factor.
        n: Number of episodes to sample.
        max_episode_len: Maximum sampled steps per episode.
        verbose: Print episode progress when true.
        rng: NumPy generator or integer seed for episode sampling.

    Returns:
        A NumPy value array with one entry per state. Unvisited states contain
        ``numpy.nan``.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  
    assert n > 0
    assert max_episode_len > 0

    rng = get_rng(rng)
    Returns = defaultdict(list) # a list for each s

    for i in tqdm(range(n), desc="MC Prediction", disable=verbose):
        if verbose:
            print("episode", i," of ", n)

        episode = sample_episode(env, policy, max_len=max_episode_len, rng=rng)
        G = 0

        # for first visit check for s
        visited = [s for s,a,r,sp in episode]

        # process episode in reverse order
        for t in range(len(episode)-1, -1, -1):
            s,a,r,sp = episode[t]
            G = discount * G + r

            # use first visit of s only
            if not s in visited[0:t]:
                Returns[s].append(G)

    Vs = np.array([np.mean(Returns[s]) if len(Returns[s]) else np.nan for s in range(env.observation_space.n)])
    return Vs

MC_control_ES_textbook 

MC_control_ES_textbook(env, discount, n=100, Q=None, max_episode_len=100, history=False, verbose=False, rng=None)

Monte Carlo control with exploring starts and stored sample returns.

This direct textbook implementation stores every first-visit return. Prefer :func:MC_control_ES for larger experiments because its running-average update uses less memory.

Parameters:

Name Type Description Default
env

A tabular gym_classics2 environment.

 required
discount

Reward discount factor.

 required
n

Number of episodes to sample.

 100
Q

Optional initial action-value array.

 None
max_episode_len

Maximum sampled steps per episode.

 100
history

Retain intermediate policies, Q arrays, episodes, and returns.

 False
verbose

Print episode details; values greater than one print transitions.

 False
rng

NumPy generator or integer seed for all algorithm choices.

 None

Returns:

Type Description

(policy, Q). If history=True, a third item contains the history dictionary with policies, Q_values, episodes, and returns.
Source code in gym_classics2/algorithms/monte_carlo_methods.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def MC_control_ES_textbook(env, discount, n=100, Q=None, max_episode_len=100,
                           history=False, verbose=False, rng=None):
    """Monte Carlo control with exploring starts and stored sample returns.

    This direct textbook implementation stores every first-visit return. Prefer
    :func:`MC_control_ES` for larger experiments because its running-average
    update uses less memory.

    Args:
        env: A tabular ``gym_classics2`` environment.
        discount: Reward discount factor.
        n: Number of episodes to sample.
        Q: Optional initial action-value array.
        max_episode_len: Maximum sampled steps per episode.
        history: Retain intermediate policies, Q arrays, episodes, and returns.
        verbose: Print episode details; values greater than one print transitions.
        rng: NumPy generator or integer seed for all algorithm choices.

    Returns:
        ``(policy, Q)``. If ``history=True``, a third item contains the history dictionary with ``policies``, ``Q_values``, ``episodes``, and ``returns``.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  
    assert n > 0
    assert max_episode_len > 0
    assert Q is None or Q.shape == (len(env.states()), env.action_space.n)

    rng = get_rng(rng)
    policy = random_policy(env, rng=rng)

    if Q is None:
        Q = np.zeros((len(env.states()), env.action_space.n))

    # lists that grow are very slow. We should use a running average instead. But this is easier to read and understand.
    Returns = defaultdict(list) # a list for each (s,a)

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        pol_list = []
        pol_list.append(policy.copy())
        ep_list = []
        return_list = []


    for i in tqdm(range(n), desc="MC Control", disable=verbose):
        if verbose:
            print("episode", i," of ", n)

        # Sample starting (s,a)
        s = rng.integers(env.observation_space.n)
        a = rng.integers(env.action_space.n)

        episode = sample_episode(env, policy, start_state=s, start_action=a,
                                 max_len=max_episode_len, verbose=verbose > 1,
                                 rng=rng)
        G = 0

        # for first visit check for (s,a)
        visited = [(s,a) for s,a,r,sp in episode]

        # process episode in reverse order
        for t in range(len(episode)-1, -1, -1):
            s,a,r,sp = episode[t]
            G = discount * G + r

            # use first visit of (s,a) only
            if not (s,a) in visited[0:t]:
                Returns[(s,a)].append(G)

                # update policy
                Q[s,a] = np.mean(Returns[(s,a)])
                policy[s] = random_argmax(Q[s, :], rng=rng)

        if history:
            pol_list.append(policy.copy())
            Q_list.append(Q.copy())
            ep_list.append(episode.copy())
            return_list.append(G)

    if history:
        return policy, Q, {'policies': pol_list, 'Q_values': Q_list, 'episodes': ep_list, 'returns': return_list}

    return policy, Q

MC_control_ES 

MC_control_ES(env, discount, n=100, Q=None, max_episode_len=100, history=False, verbose=False, rng=None)

Monte Carlo Control with Exploring Starts (incremental version). This algorithm estimates the optimal action-value function Q and the corresponding greedy policy by sampling episodes with exploring starts. It uses incremental updates to compute the average returns for each (s,a) pair, which is more memory efficient than storing all returns.

Parameters:

Name Type Description Default
env

A tabular gym_classics2 environment.

 required
discount

Reward discount factor.

 required
n

Number of episodes to sample.

 100
Q

Optional initial action-value array.

 None
max_episode_len

Maximum sampled steps per episode.

 100
history

Retain intermediate policies, Q arrays, episodes, and returns. Note: retaining the history may require a lot of memory.

 False
verbose

Print episode details; values greater than one print transitions.

 False
rng

NumPy generator or integer seed for all algorithm choices.

 None

Returns:

Type Description

(policy, Q). If history=True, a third item contains the history dictionary with policies, Q_values, episodes, and returns.
Source code in gym_classics2/algorithms/monte_carlo_methods.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def MC_control_ES(env, discount, n=100, Q=None, max_episode_len=100,
                  history=False, verbose=False, rng=None):
    """Monte Carlo Control with Exploring Starts (incremental version).
    This algorithm estimates the optimal action-value function Q and the corresponding greedy policy by sampling episodes with exploring starts. It uses incremental updates to compute the average returns for each (s,a) pair, which is more memory efficient than storing all returns.

    Args:
        env: A tabular ``gym_classics2`` environment.
        discount: Reward discount factor.
        n: Number of episodes to sample.
        Q: Optional initial action-value array.
        max_episode_len: Maximum sampled steps per episode.
        history: Retain intermediate policies, Q arrays, episodes, and returns. Note: retaining the history may require a lot of memory.
        verbose: Print episode details; values greater than one print transitions.
        rng: NumPy generator or integer seed for all algorithm choices.

    Returns:
        ``(policy, Q)``. If ``history=True``, a third item contains the history dictionary with ``policies``, ``Q_values``, ``episodes``, and ``returns``.
    """

    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  
    assert n > 0
    assert max_episode_len > 0
    assert Q is None or Q.shape == (env.observation_space.n, env.action_space.n)

    rng = get_rng(rng)
    policy = random_policy(env, rng=rng)

    if Q is None:
        Q = np.zeros((env.observation_space.n, env.action_space.n))

    # Count how many first-visit returns have been used for each (s, a)
    N = np.zeros((env.observation_space.n, env.action_space.n), dtype=int)

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        pol_list = []
        pol_list.append(policy.copy())
        ep_list = []
        return_list = []

    for i in tqdm(range(n), desc="MC Control (Incremental)", disable=verbose):
        if verbose:
            print("episode", i, "of", n)

        # Exploring starts
        s = rng.integers(env.observation_space.n)
        a = rng.integers(env.action_space.n)

        episode = sample_episode(
            env,
            policy,
            start_state=s,
            start_action=a,
            max_len=max_episode_len,
            verbose=verbose > 1,
            rng=rng,
        )

        G = 0

        # Process episode backward
        visited = set()
        for t in range(len(episode) - 1, -1, -1):
            s, a, r, sp = episode[t]
            G = discount * G + r

            # First-visit MC: only update the first occurrence from the start
            if (s, a) not in visited:
                visited.add((s, a))

                N[s, a] += 1
                Q[s, a] += (G - Q[s, a]) / N[s, a]

                # Improve policy greedily
                policy[s] = random_argmax(Q[s, :], rng=rng)

        if history:
            pol_list.append(policy.copy())
            Q_list.append(Q.copy())
            ep_list.append(episode.copy())
            return_list.append(G)

    if history:
        return policy, Q, {'policies': pol_list, 'Q_values': Q_list, 'episodes': ep_list, 'returns': return_list}

    return policy, Q

Temporal-difference learning

gym_classics2.algorithms.temporal_difference_learning

This file implements temporal difference learning algorithms for policy evaluation and control in gym-classics
environments with discrete state spaces. The algorithms include Sarsa(0) and Q-learning, which are fundamental methods in reinforcement learning for learning value functions and optimal policies from experience without requiring a model of the environment.

Sarsa_0 

Sarsa_0(env, discount, alpha, epsilon, Q=None, n=100, verbose=False, history=False, rng=None)

Learn action values with one-step on-policy Sarsa.

Parameters:

Name Type Description Default
env

Gymnasium environment with discrete observation and action spaces.

 required
discount

Reward discount factor in [0, 1].

 required
alpha

Scalar step size or Schedule evaluated once per episode.

 required
epsilon

Scalar exploration probability or Schedule evaluated once per episode.

 required
Q

Optional initial action-value array shaped (env.observation_space.n, env.action_space.n). The array is updated in place.

 None
n

Number of training episodes.

 100
verbose

Print individual updates when true.

 False
history

Retain Q arrays, discounted episode returns, and episode lengths.

 False
rng

NumPy generator or integer seed for exploration and tie-breaking.

 None

Returns:

Type Description

The learned Q array. If history=True, returns (Q, history_dict);

the dictionary contains Qs, returns, and ep_lens.

Raises:

Type Description
AssertionError

If the observation space is not discrete.
Source code in gym_classics2/algorithms/temporal_difference_learning.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def Sarsa_0(env, discount, alpha, epsilon, Q=None, n=100, verbose=False,
            history=False, rng=None):
    """Learn action values with one-step on-policy Sarsa.

    Args:
        env: Gymnasium environment with discrete observation and action spaces.
        discount: Reward discount factor in ``[0, 1]``.
        alpha: Scalar step size or ``Schedule`` evaluated once per episode.
        epsilon: Scalar exploration probability or ``Schedule`` evaluated once
            per episode.
        Q: Optional initial action-value array shaped
            ``(env.observation_space.n, env.action_space.n)``. The array is updated
            in place.
        n: Number of training episodes.
        verbose: Print individual updates when true.
        history: Retain Q arrays, discounted episode returns, and episode lengths.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        The learned Q array. If ``history=True``, returns ``(Q, history_dict)``;
        the dictionary contains ``Qs``, ``returns``, and ``ep_lens``.

    Raises:
        AssertionError: If the observation space is not discrete.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  

    rng = get_rng(rng)

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if Q is None:
        Q = np.zeros((env.observation_space.n, env.action_space.n))

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        return_list = []
        ep_len_list = []


    for i in tqdm(range(n), desc="Sarsa", disable=verbose):
        s, r = env.reset()

        if verbose:
            print(f"--- Episode {i} ---")      

        if rng.random() > epsilon(i):
            a = random_argmax(Q[s, :], rng=rng)
        else:
            a = rng.integers(env.action_space.n)

        t = 0
        done = False
        G = 0
        while not done:            
            sp, r, done, _, _ = env.step(a)
            G += r * pow(discount, t)
            t += 1


            if rng.random() > epsilon(i):
                ap = random_argmax(Q[sp, :], rng=rng)
            else:
                ap = rng.integers(env.action_space.n)

            Q[s,a] = Q[s,a] + alpha(i) * (r + discount * Q[sp,ap] - Q[s,a])

            if verbose:
                print(f"{t} - sarsa: {s},{a},{r},{sp},{ap}, - new Q(s,a): {Q[s,a]}")
                if done:
                    print("Total return:", G)

            s = sp
            a = ap

        if history:
            Q_list.append(Q.copy())
            return_list.append(G)   
            ep_len_list.append(t)

    if history:
        return Q, {'Qs': Q_list, 'returns': return_list, 'ep_lens': ep_len_list}

    return Q

Q_learning 

Q_learning(env, discount, alpha, epsilon, Q=None, n=100, verbose=False, history=False, rng=None)

Learn action values with one-step off-policy Q-learning.

Parameters:

Name Type Description Default
env

Gymnasium environment with discrete observation and action spaces.

 required
discount

Reward discount factor in [0, 1].

 required
alpha

Scalar step size or Schedule evaluated once per episode.

 required
epsilon

Scalar exploration probability or Schedule evaluated once per episode.

 required
Q

Optional initial action-value array shaped (env.observation_space.n, env.action_space.n). The array is updated in place.

 None
n

Number of training episodes.

 100
verbose

Print progress information when true.

 False
history

Retain Q arrays, discounted episode returns, episode lengths, and state-visit counts.

 False
rng

NumPy generator or integer seed for exploration and tie-breaking.

 None

Returns:

Type Description

The learned Q array. If history=True, returns (Q, history_dict);

the dictionary contains Qs, returns, ep_lens, and

state_visits.

Raises:

Type Description
AssertionError

If the observation space is not discrete.
Source code in gym_classics2/algorithms/temporal_difference_learning.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def Q_learning(env, discount, alpha, epsilon, Q=None, n=100, verbose=False,
               history=False, rng=None):
    """Learn action values with one-step off-policy Q-learning.

    Args:
        env: Gymnasium environment with discrete observation and action spaces.
        discount: Reward discount factor in ``[0, 1]``.
        alpha: Scalar step size or ``Schedule`` evaluated once per episode.
        epsilon: Scalar exploration probability or ``Schedule`` evaluated once
            per episode.
        Q: Optional initial action-value array shaped
            ``(env.observation_space.n, env.action_space.n)``. The array is updated
            in place.
        n: Number of training episodes.
        verbose: Print progress information when true.
        history: Retain Q arrays, discounted episode returns, episode lengths, and
            state-visit counts.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        The learned Q array. If ``history=True``, returns ``(Q, history_dict)``;
        the dictionary contains ``Qs``, ``returns``, ``ep_lens``, and
        ``state_visits``.

    Raises:
        AssertionError: If the observation space is not discrete.
    """
    assert isinstance(env.observation_space, gym.spaces.Discrete), "Tabular methods require discrete state space."  

    rng = get_rng(rng)

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if Q is None:
        Q = np.zeros((env.observation_space.n, env.action_space.n))

    if history:
        Q_list = []
        Q_list.append(Q.copy())
        return_list = []
        ep_len_list = []
        #state_visits = np.zeros(env.observation_space.n, dtype=int)
        # Note we use float so visualization works better
        state_visits = np.zeros(env.observation_space.n, dtype=float)

    for i in tqdm(range(n), desc="Q-Learning", disable=verbose):
        s, r = env.reset()

        if history:
            state_visits[s] += 1

        done = False
        G = 0
        t = 0   
        while not done:
            # epsilon-greedy choice w.r.t. Q 
            if rng.random() > epsilon(i):
                a = random_argmax(Q[s, :], rng=rng)
            else:
                a = rng.integers(env.action_space.n)

            sp, r, done, _, _ = env.step(a)

            if history:
                state_visits[sp] += 1

            if history:
                G += r * pow(discount, t)
                t += 1

            Q[s,a] = Q[s,a] + alpha(i) * (r + discount * np.max(Q[sp,:]) - Q[s,a])

            s = sp

        if history:
            Q_list.append(Q.copy())
            return_list.append(G)
            ep_len_list.append(t)

    if history:
        return Q, {'Qs': Q_list, 'returns': return_list, 'ep_lens': ep_len_list, 'state_visits': state_visits}

    return Q

Linear function approximation

gym_classics2.algorithms.linear_approximation

Linear function approximation algorithms for policy evaluation and control.

The algorithms do not require discrete state spaces. Callers provide a state_features(state, env) function that converts states to state feature vectors.

state_action_features 

state_action_features(s, a, env, state_features)

Construct the state-action feature vector x(s, a).

The state feature vector is expected to have the form [1, x1, ..., xd], with a leading intercept. The returned vector keeps one shared intercept and has a separate block of the remaining state features for each action. Only the intercept and the weights for a are selected; all other elements are set to zero.

Parameters:

Name Type Description Default
s

State to represent.

 required
a

Integer ID of the action to represent.

 required
env

Environment providing the discrete action space.

 required
state_features

Callable that returns [1, x1, ..., xd] for (s, env).

 required

Returns:

Type Description

Block-coded NumPy feature vector for the state-action pair (s, a).
Source code in gym_classics2/algorithms/linear_approximation.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def state_action_features(s, a, env, state_features):
    """Construct the state-action feature vector ``x(s, a)``.

    The state feature vector is expected to have the form
    ``[1, x1, ..., xd]``, with a leading intercept. The returned vector keeps
    one shared intercept and has a separate block of the remaining state
    features for each action. Only the intercept and the weights for ``a`` are
    selected; all other elements are set to zero.

    Args:
        s: State to represent.
        a: Integer ID of the action to represent.
        env: Environment providing the discrete action space.
        state_features: Callable that returns ``[1, x1, ..., xd]`` for
            ``(s, env)``.

    Returns:
        Block-coded NumPy feature vector for the state-action pair ``(s, a)``.
    """
    def active_weights(a, sf_len):
        return [0] + list(range(a*sf_len+1, a*sf_len+sf_len+1))

    s = state_features(s, env)
    x = np.zeros(1+len(s)*env.action_space.n)
    x[active_weights(a, len(s)-1)] = s
    return x

v_hat 

v_hat(s, w, env, state_features)

Compute the linear state-value approximation v_hat(s, w).

Approximates the state value as w^T x(s), the weighted sum of the components in the state feature vector x(s). The weight and feature vectors must have the same length.

Parameters:

Name Type Description Default
s

State to evaluate.

 required
w

Weight vector of the linear approximator.

 required
env

Environment containing the state.

 required
state_features

Callable returning the feature vector x(s) for (s, env).

 required

Returns:

Type Description

Scalar estimate of the expected return from s.
Source code in gym_classics2/algorithms/linear_approximation.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def v_hat(s, w, env, state_features):
    """Compute the linear state-value approximation ``v_hat(s, w)``.

    Approximates the state value as ``w^T x(s)``, the weighted sum of the components in the
    state feature vector ``x(s)``. The weight and feature vectors must have the
    same length.

    Args:
        s: State to evaluate.
        w: Weight vector of the linear approximator.
        env: Environment containing the state.
        state_features: Callable returning the feature vector ``x(s)`` for
            ``(s, env)``.

    Returns:
        Scalar estimate of the expected return from ``s``.
    """
    return np.dot(w, state_features(s, env))

q_hat 

q_hat(s, a, w, env, state_features)

Compute the linear action-value approximation q_hat(s, a, w).

Estimates the q-value as w^T x(s, a) using the block-coded state-action features produced by state_action_features(). The weight vector must have the same length as that feature vector.

Parameters:

Name Type Description Default
s

State to evaluate.

 required
a

Integer ID of the action to evaluate.

 required
w

Weight vector of the linear approximator.

 required
env

Environment providing the discrete action space.

 required
state_features

Callable returning a state feature vector with a leading intercept for (s, env).

 required

Returns:

Type Description

Scalar estimate of the expected return from taking a in s.
Source code in gym_classics2/algorithms/linear_approximation.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def q_hat(s, a, w, env, state_features):
    """Compute the linear action-value approximation ``q_hat(s, a, w)``.

    Estimates the q-value as ``w^T x(s, a)`` using the block-coded state-action
    features produced by `state_action_features()`. The weight vector must
    have the same length as that feature vector.

    Args:
        s: State to evaluate.
        a: Integer ID of the action to evaluate.
        w: Weight vector of the linear approximator.
        env: Environment providing the discrete action space.
        state_features: Callable returning a state feature vector with a
            leading intercept for ``(s, env)``.

    Returns:
        Scalar estimate of the expected return from taking ``a`` in ``s``.
    """    
    x = state_action_features(s, a, env, state_features)
    return np.dot(w, x)

epsilon_greedy_action_w 

epsilon_greedy_action_w(s, w, env, state_features, epsilon=0, rng=None)

Select an epsilon-greedy action from approximate action values.

Parameters:

Name Type Description Default
s

Current state.

 required
w

Weight vector for the action-value approximator.

 required
env

Environment providing the discrete action space.

 required
state_features

Callable converting (state, env) to a feature vector.

 required
epsilon

Probability of selecting a uniformly random action.

 0
rng

NumPy generator or integer seed.

 None

Returns:

Type Description

Selected integer action ID.
Source code in gym_classics2/algorithms/linear_approximation.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def epsilon_greedy_action_w(
    s, w, env, state_features, epsilon=0, rng=None
):
    """Select an epsilon-greedy action from approximate action values.

    Args:
        s: Current state.
        w: Weight vector for the action-value approximator.
        env: Environment providing the discrete action space.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        epsilon: Probability of selecting a uniformly random action.
        rng: NumPy generator or integer seed.

    Returns:
        Selected integer action ID.
    """

    rng = get_rng(rng)

    if epsilon > 0 and rng.random() < epsilon:
        return rng.integers(env.action_space.n)

    return random_argmax(
        [
            q_hat(s, a, w, env, state_features)
            for a in range(env.action_space.n)
        ],
        rng=rng,
    )

MSVE 

MSVE(V, V_true, weight=None)

Calculate the weighted mean squared value error.

Parameters:

Name Type Description Default
V

Estimated value for each state.

 required
V_true

Reference value for each state.

 required
weight

Weight for each state, typically its stationary visitation probability. If omitted, use unit weights.

 None

Returns:

Type Description

Weighted sum of squared value errors.
Source code in gym_classics2/algorithms/linear_approximation.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def MSVE(V, V_true, weight=None):
    """Calculate the weighted mean squared value error.

    Args:
        V: Estimated value for each state.
        V_true: Reference value for each state.
        weight: Weight for each state, typically its stationary visitation
            probability. If omitted, use unit weights.

    Returns:
        Weighted sum of squared value errors.
    """
    if weight is None:
        weight = np.ones(len(V))

    return np.sum(weight * (V - V_true)**2)

semi_gradient_TD0_estimation 

semi_gradient_TD0_estimation(env, state_features, policy, n, alpha, gamma, max_episode_length=1000, verbose=False)

Estimate state values with semi-gradient TD(0).

This function runs TD(0) learning with function approximation over multiple episodes generated from a given policy and environment. Updates are performed using the semi-gradient of the value function approximation.

Parameters:

Name Type Description Default
env

Episodic Gymnasium environment used to generate experience.

 required
state_features

Callable converting (state, env) to a feature vector.

 required
policy

Deterministic policy indexed by state.

 required
n

Number of training episodes.

 required
alpha

Step size or schedule.

 required
gamma

Discount factor in [0, 1].

 required
max_episode_length

Maximum number of steps per episode.

 1000
verbose

Whether to print step-by-step diagnostics.

 False

Returns:

Type Description

Learned weight vector for the approximate value function.
Source code in gym_classics2/algorithms/linear_approximation.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def semi_gradient_TD0_estimation(
    env,
    state_features,
    policy,
    n,
    alpha,
    gamma,
    max_episode_length=1000,
    verbose=False,
):
    """Estimate state values with semi-gradient TD(0).

    This function runs TD(0) learning with function approximation over multiple
    episodes generated from a given policy and environment. Updates are performed
    using the semi-gradient of the value function approximation.

    Args:
        env: Episodic Gymnasium environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        policy: Deterministic policy indexed by state.
        n: Number of training episodes.
        alpha: Step size or schedule.
        gamma: Discount factor in ``[0, 1]``.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.

    Returns:
        Learned weight vector for the approximate value function.
    """
    assert gamma >= 0 and gamma <= 1, "Gamma must be in [0,1]"
    assert n > 0, "Number of episodes must be positive"
    assert max_episode_length > 0, "Max episode length must be positive"

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)

    state, _ = env.reset()
    w = np.zeros(len(state_features(state, env)))  # Initialize weights (intercept + x and y)

    for episode in tqdm(range(n), desc="Semi-Gradient TD(0)", disable=verbose):
        state, _ = env.reset()
        done = False

        i = 0
        while not done and i < max_episode_length:
            action = policy[state]  # follow policy
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            # Semi-gradient TD(0) update
            # Note: v_hat(terminal, w) needs to be 0
            if terminated:
                w += alpha(episode) * (reward - v_hat(state, w, env, state_features)) * state_features(state, env)
            else: 
                w += alpha(episode) * (reward + gamma * v_hat(next_state, w, env, state_features) - v_hat(state, w, env, state_features)) * state_features(state, env)

            if verbose:
                print (f"Episode {episode+1}, Step {i+1}: S={state}, A={action}, R={reward}, S'={next_state}, w={w}")

            state = next_state
            i += 1

    return w

semi_gradient_Sarsa_0 

semi_gradient_Sarsa_0(env, state_features, n, epsilon, alpha, gamma, w=None, max_episode_length=1000, verbose=False, history=False, rng=None)

Run semi-gradient Sarsa(0) with function approximation.

Implements the semi-gradient Sarsa(0) algorithm for estimating the optimal action-value function q_*(s, a) using a differentiable function approximator q̂(s, a, w). Actions are selected according to an ε-greedy policy derived from the current action-value estimate.

Episodes are truncated after max_episode_length time steps.

Parameters:

Name Type Description Default
env

Episodic environment used to generate experience.

 required
state_features

Callable converting (state, env) to a feature vector.

 required
n

Number of training episodes.

 required
epsilon

Exploration rate or schedule for the epsilon-greedy policy.

 required
alpha

Step size or schedule.

 required
gamma

Discount factor in [0, 1].

 required
w

Initial action-value weight vector. If omitted, initialize it to zeros.

 None
max_episode_length

Maximum number of steps per episode.

 1000
verbose

Whether to print step-by-step diagnostics.

 False
history

Whether to return weights, returns, and episode lengths collected during training.

 False
rng

NumPy generator or integer seed for exploration and tie-breaking.

 None

Returns:

Type Description

Learned weight vector. If history is true, returns (w, history).
Source code in gym_classics2/algorithms/linear_approximation.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def semi_gradient_Sarsa_0(env, state_features, n, epsilon, alpha, gamma, w=None,
                          max_episode_length=1000, verbose=False,
                          history=False, rng=None):
    """Run semi-gradient Sarsa(0) with function approximation.

    Implements the **semi-gradient Sarsa(0)** algorithm for estimating the optimal
    action-value function q_*(s, a) using a differentiable function approximator
    q̂(s, a, w). Actions are selected according to an ε-greedy policy derived
    from the current action-value estimate.

    Episodes are truncated after `max_episode_length` time steps.

    Args:
        env: Episodic environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        n: Number of training episodes.
        epsilon: Exploration rate or schedule for the epsilon-greedy policy.
        alpha: Step size or schedule.
        gamma: Discount factor in ``[0, 1]``.
        w: Initial action-value weight vector. If omitted, initialize it to zeros.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.
        history: Whether to return weights, returns, and episode lengths collected
            during training.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        Learned weight vector. If ``history`` is true, returns ``(w, history)``.
    """

    assert gamma >= 0 and gamma <= 1, "Gamma must be in [0,1]"
    assert n > 0, "Number of episodes must be positive"
    assert max_episode_length > 0, "Max episode length must be positive"

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if w is None:
        state, _ = env.reset()
        w = np.zeros(len(state_action_features(state, 0, env, state_features)))

    if history:
        ws = []
        ws.append(w.copy())
        returns = []
        ep_lens = []

    for episode in tqdm(range(n), desc="Semi-Gradient SARSA(0)", disable=verbose):
        state, _ = env.reset()
        action = epsilon_greedy_action_w(
            state, w, env, state_features, epsilon(episode), rng=rng
        )
        done = False

        i = 0
        if history:
            G = 0
        while not done and i < max_episode_length:


            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            x = state_action_features(state, action, env, state_features)

            if terminated:
                next_action = None
                w += alpha(episode) * (reward - q_hat(state, action, w, env, state_features)) * x

            else:
                next_action = epsilon_greedy_action_w(
                    next_state, w, env, state_features, epsilon(episode), rng=rng
                )
                w += alpha(episode) * (reward + gamma * q_hat(next_state, next_action, w, env, state_features) - q_hat(state, action, w, env, state_features)) * x

            if verbose:
                print (f"Episode {episode+1}, Step {i+1}: S={state}, A={action}, R={reward}, S'={next_state}, w={w}")

            state = next_state
            action = next_action
            i += 1

            if history:
                G += reward * (gamma ** (i-1))

        if history:
            ws.append(w.copy())
            returns.append(G)
            ep_lens.append(i)

    if history:
        return w, {'ws': ws, 'returns': returns, 'ep_lens': ep_lens}

    return w

create_fourier_basis_coefs 

create_fourier_basis_coefs(dim, order)

Create Fourier basis coefficient vectors.

Parameters:

Name Type Description Default
dim

Number of state-feature dimensions.

 required
order

Maximum coefficient in each dimension.

 required

Returns:

Type Description

Array containing the Cartesian product of coefficients from zero through

order in each dimension.
Source code in gym_classics2/algorithms/linear_approximation.py 
307
308
309
310
311
312
313
314
315
316
317
318
def create_fourier_basis_coefs(dim, order): 
    """Create Fourier basis coefficient vectors.

    Args:
        dim: Number of state-feature dimensions.
        order: Maximum coefficient in each dimension.

    Returns:
        Array containing the Cartesian product of coefficients from zero through
        ``order`` in each dimension.
    """
    return np.array(list(product(range(order+1), repeat=dim)))

transformation_fourier_basis 

transformation_fourier_basis(min, max, order)

Create a Fourier basis feature transformation.

Parameters:

Name Type Description Default
min

Minimum state value in each dimension.

 required
max

Maximum state value in each dimension.

 required
order

Maximum Fourier coefficient in each dimension.

 required

Returns:

Type Description

Callable that normalizes a state to the unit hypercube and returns its

Fourier basis features.
Example 
transform = transformation_fourier_basis([0, 0], [1, 1], order=3)

def state_features(state, env):
    return transform(state)
Source code in gym_classics2/algorithms/linear_approximation.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def transformation_fourier_basis(min, max, order):
    """Create a Fourier basis feature transformation.

    Args:
        min: Minimum state value in each dimension.
        max: Maximum state value in each dimension.
        order: Maximum Fourier coefficient in each dimension.

    Returns:
        Callable that normalizes a state to the unit hypercube and returns its
        Fourier basis features.

    Example:
        ```python
        transform = transformation_fourier_basis([0, 0], [1, 1], order=3)

        def state_features(state, env):
            return transform(state)
        ```
    """

    min = np.array(min)
    max = np.array(max)
    coefs = create_fourier_basis_coefs(len(min), order)

    def fourier_basis(s):
        # normalize state to [0,1]
        s = np.array(s)
        assert s.shape == min.shape, "State dimension does not match Fourier basis dimension"

        norm_s = (s - min) / (max - min)
        return np.cos(np.pi * np.dot(coefs, norm_s))

    return fourier_basis

Eligibility traces

gym_classics2.algorithms.eligibility_traces

Semi-gradient Sarsa(lambda) with linear function approximation.

Callers provide a state_features(state, env) function that converts states to feature vectors.

semi_gradient_Sarsa_lambda 

semi_gradient_Sarsa_lambda(env, state_features, n, epsilon, alpha, gamma, lam, w=None, max_episode_length=1000, verbose=False, history=False, rng=None)

Run semi-gradient Sarsa(lambda) with linear function approximation.

Parameters:

Name Type Description Default
env

Episodic environment used to generate experience.

 required
state_features

Callable converting (state, env) to a feature vector.

 required
n

Number of episodes.

 required
epsilon

Exploration rate or schedule for the epsilon-greedy policy.

 required
alpha

Step size or schedule.

 required
gamma

Discount factor in [0, 1].

 required
lam

Trace-decay parameter lambda in [0, 1].

 required
w

Initial weight vector. If omitted, initialize it to zeros.

 None
max_episode_length

Maximum number of steps per episode.

 1000
verbose

Whether to print step-by-step diagnostics.

 False
history

Whether to return weights, returns, and episode lengths collected during training.

 False
rng

NumPy generator or integer seed for exploration and tie-breaking.

 None

Returns:

Type Description

Learned weight vector. If history is true, returns (w, history).
Source code in gym_classics2/algorithms/eligibility_traces.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def semi_gradient_Sarsa_lambda(
    env,
    state_features,
    n,
    epsilon,
    alpha,
    gamma,
    lam,
    w=None,
    max_episode_length=1000,
    verbose=False,
    history=False,
    rng=None,
):
    """Run semi-gradient Sarsa(lambda) with linear function approximation.

    Args:
        env: Episodic environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        n: Number of episodes.
        epsilon: Exploration rate or schedule for the epsilon-greedy policy.
        alpha: Step size or schedule.
        gamma: Discount factor in ``[0, 1]``.
        lam: Trace-decay parameter lambda in ``[0, 1]``.
        w: Initial weight vector. If omitted, initialize it to zeros.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.
        history: Whether to return weights, returns, and episode lengths collected
            during training.
        rng: NumPy generator or integer seed for exploration and tie-breaking.

    Returns:
        Learned weight vector. If ``history`` is true, returns ``(w, history)``.
    """

    assert gamma >= 0 and gamma <= 1, "gamma must be in [0,1]"
    assert lam >= 0 and lam <= 1, "lambda must be in [0,1]"
    assert n > 0, "number of episodes must be positive"
    assert max_episode_length > 0, "max episode length must be positive"

    rng = get_rng(rng)

    if not isinstance(alpha, Schedule):
        alpha = ConstantSchedule(alpha)
    if not isinstance(epsilon, Schedule):
        epsilon = ConstantSchedule(epsilon)

    if w is None:
        state, _ = env.reset()
        w = np.zeros(len(state_action_features(state, 0, env, state_features)))

    if history:
        ws = []
        ws.append(w.copy())
        returns = []
        ep_lens = []


    for episode in tqdm(range(n), desc="Semi-Gradient SARSA(lambda)", disable=verbose):
        state, _ = env.reset()
        action = epsilon_greedy_action_w(
            state, w, env, state_features, epsilon(episode), rng=rng
        )

        # eligibility trace vector, same size as w
        z = np.zeros_like(w)
        Q_old = 0

        done = False
        i = 0

        G = 0  # for tracking returns if history is enabled

        while not done and i < max_episode_length:
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            G += reward * (gamma ** i)  # accumulate return if history is enabled

            # current feature vector for (state, action)
            x = state_action_features(state, action, env, state_features)

            # update trace
            z = gamma * lam * z + (1 - alpha(episode) * gamma * lam * np.dot(z, x)) * x

            if terminated:
                delta = reward - q_hat(state, action, w, env, state_features)
            else:
                next_action = epsilon_greedy_action_w(
                    next_state, w, env, state_features, epsilon(episode), rng=rng
                )
                delta = reward + gamma * q_hat(next_state, next_action, w, env, state_features) - q_hat(state, action, w, env, state_features)

            # semi-gradient weight update
            Q = q_hat(state, action, w, env, state_features)
            Q_prime = q_hat(next_state, next_action, w, env, state_features) if not terminated else 0
            w += alpha(episode) * (delta + Q - Q_old) * z - alpha(episode) * (Q - Q_old) * x

            Q_old = Q_prime

            if verbose:
                if terminated:
                    print(
                        f"Episode {episode+1}, Step {i+1}: "
                        f"S={state}, A={action}, R={reward}, S'={next_state}, "
                        f"delta={delta}, z={z}, w={w}"
                    )
                else:
                    print(
                        f"Episode {episode+1}, Step {i+1}: "
                        f"S={state}, A={action}, R={reward}, S'={next_state}, A'={next_action}, "
                        f"delta={delta}, z={z}, w={w}"
                    )

            if done:
                break

            state = next_state
            action = next_action
            i += 1

        if history:
            returns.append(G)
            ws.append(w.copy())
            ep_lens.append(i)


    if history:        
        return w, {'ws': ws, 'returns': returns, 'ep_lens': ep_lens}

    return w

Policy-gradient methods

gym_classics2.algorithms.policy_gradient_methods

This file implements policy gradient methods for learning parameterized policies. The main algorithm implemented is REINFORCE, which is a Monte Carlo policy gradient method that updates policy parameters based on the returns observed in sampled episodes. The policy is represented using a softmax function over linear state-action features, and the algorithm estimates the policy gradient using the log-likelihood of actions taken in the episodes. This implementation allows for learning stochastic policies that can handle exploration and exploitation in reinforcement learning tasks.

Callers provide a state_features(state, env) function that converts states to feature vectors suitable for the environment being used.

h 

h(s, a, theta, env, state_features)

Return the linear action preference for state s and action a.

Source code in gym_classics2/algorithms/policy_gradient_methods.py 
22
23
24
def h(s, a, theta, env, state_features):
    """Return the linear action preference for state ``s`` and action ``a``."""
    return np.dot(theta, state_action_features(s, a, env, state_features))

pi 

pi(s, theta, env, state_features)

Return the softmax action-probability vector for a state.

Source code in gym_classics2/algorithms/policy_gradient_methods.py 
26
27
28
29
30
31
32
def pi(s, theta, env, state_features):
    """Return the softmax action-probability vector for a state."""
    hs = np.array(
        [h(s, a, theta, env, state_features) for a in range(env.action_space.n)]
    )
    exp_hs = np.exp(hs)
    return exp_hs / np.sum(exp_hs)

sample_episode_approx_policy 

sample_episode_approx_policy(env, pi, theta, state_features, max_episode_length=1000, rng=None)

Sample an episode from a parameterized policy.

Parameters:

Name Type Description Default
env

Episodic Gymnasium environment.

 required
pi

Callable returning action probabilities for (state, theta, env).

 required
theta

Policy parameter vector.

 required
max_episode_length

Maximum number of transitions to sample.

 1000
rng

NumPy generator or integer seed used to sample actions.

 None

Returns:

Type Description

List of (state, action, reward, next_state) transitions.
Source code in gym_classics2/algorithms/policy_gradient_methods.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def sample_episode_approx_policy(
    env, pi, theta, state_features, max_episode_length=1000, rng=None
):
    """Sample an episode from a parameterized policy.

    Args:
        env: Episodic Gymnasium environment.
        pi: Callable returning action probabilities for ``(state, theta, env)``.
        theta: Policy parameter vector.
        max_episode_length: Maximum number of transitions to sample.
        rng: NumPy generator or integer seed used to sample actions.

    Returns:
        List of ``(state, action, reward, next_state)`` transitions.
    """

    rng = get_rng(rng)
    s, _ = env.reset()
    episode_data = []

    for t in range(max_episode_length):
        a = rng.choice(
            env.action_space.n, p=pi(s, theta, env, state_features)
        )
        next_s, r, done, _, _ = env.step(a)
        episode_data.append((s, a, r, next_s))
        s = next_s

        if done:
            break

    return episode_data

choose_action_w 

choose_action_w(env, pi, theta, state, state_features, rng=None)

Sample an action from a parameterized policy.

Parameters:

Name Type Description Default
env

Environment providing the discrete action space.

 required
pi

Callable returning action probabilities for (state, theta, env).

 required
theta

Policy parameter vector.

 required
state

Current state.

 required
rng

NumPy generator or integer seed used to sample the action.

 None

Returns:

Type Description

Selected integer action ID.
Source code in gym_classics2/algorithms/policy_gradient_methods.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def choose_action_w(env, pi, theta, state, state_features, rng=None):
    """Sample an action from a parameterized policy.

    Args:
        env: Environment providing the discrete action space.
        pi: Callable returning action probabilities for ``(state, theta, env)``.
        theta: Policy parameter vector.
        state: Current state.
        rng: NumPy generator or integer seed used to sample the action.

    Returns:
        Selected integer action ID.
    """
    rng = get_rng(rng)
    return rng.choice(
        env.action_space.n, p=pi(state, theta, env, state_features)
    )

REINFORCE 

REINFORCE(env, state_features, n, alpha, gamma, theta=None, max_episode_length=1000, verbose=False, history=False, rng=None)

Run REINFORCE with a linear softmax policy.

Parameters:

Name Type Description Default
env

Episodic environment used to generate experience.

 required
state_features

Callable converting (state, env) to a feature vector.

 required
n

Number of training episodes.

 required
alpha

Policy step size or schedule.

 required
gamma

Discount factor in [0, 1].

 required
theta

Initial policy parameter vector. If omitted, initialize it to zeros.

 None
max_episode_length

Maximum number of steps per episode.

 1000
verbose

Whether to print step-by-step diagnostics.

 False
history

Whether to return parameters, returns, and episode lengths collected during training.

 False
rng

NumPy generator or integer seed used to sample actions.

 None

Returns:

Type Description

Learned policy parameters. If history is true, returns

(theta, history).
Source code in gym_classics2/algorithms/policy_gradient_methods.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def REINFORCE(
    env,
    state_features,
    n,
    alpha,
    gamma,
    theta = None,
    max_episode_length=1000,
    verbose=False,
    history=False,
    rng=None,
    ):
    """Run REINFORCE with a linear softmax policy.

    Args:
        env: Episodic environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        n: Number of training episodes.
        alpha: Policy step size or schedule.
        gamma: Discount factor in ``[0, 1]``.
        theta: Initial policy parameter vector. If omitted, initialize it to
            zeros.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.
        history: Whether to return parameters, returns, and episode lengths
            collected during training.
        rng: NumPy generator or integer seed used to sample actions.

    Returns:
        Learned policy parameters. If ``history`` is true, returns
        ``(theta, history)``.
    """

    assert gamma >= 0 and gamma <= 1, "gamma must be in [0  ,1]"
    assert n > 0, "number of episodes must be positive"
    assert max_episode_length > 0, "max episode length must be positive"

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if isinstance(alpha, float):
        alpha = ConstantSchedule(alpha)

    if theta is None:
        state, _ = env.reset()
        theta = np.zeros(
            len(state_action_features(state, 0, env, state_features))
        )

    if history:
        returns = []        
        ep_lens = []
        thetas = []
        thetas.append(theta.copy())

    for episode in tqdm(range(n), desc="Episodes", disable=verbose):
        if verbose:
            print(f"Episode {episode+1}/{n}")

        # sample complete episode using pi (this is a MC method)
        episode_data = sample_episode_approx_policy(
            env, pi, theta, state_features, max_episode_length, rng=rng
        )

        for t in range(len(episode_data)):
            # update policy for each step in the episode using the return observed from that step on
            #print(episode_data[t])

            G = np.sum([e[2] for e in episode_data[t:]] * (gamma ** np.arange(len(episode_data[t:]))))
            if history and t == 0:
                returns.append(G)   

            s,a,r,next_s = episode_data[t]

            # ln policy gradient= x(s,a)- sum_b pi(b|s,theta) x(s,b)
            grad_log_pi = state_action_features(s, a, env, state_features) - sum(
                pi(s, theta, env, state_features)[b]
                * state_action_features(s, b, env, state_features)
                for b in range(env.action_space.n)
            )

            if verbose: 
                print (f"t: {t}, G: {G:.2f}, grad_log_pi: {grad_log_pi}")

            theta += alpha(episode) * (gamma**t) * G * grad_log_pi

        if history:
            thetas.append(theta.copy())
            ep_lens.append(len(episode_data))

    if history:
        return theta, {'thetas': thetas, 'returns': returns, 'ep_lens': ep_lens}

    return theta

AC 

AC(env, state_features, n, alpha_policy, alpha_value, gamma, max_episode_length=1000, verbose=False, history=False, rng=None)

Run one-step actor-critic with linear policy and value approximators.

Parameters:

Name Type Description Default
env

Episodic environment used to generate experience.

 required
state_features

Callable converting (state, env) to a feature vector.

 required
n

Number of training episodes.

 required
alpha_policy

Step size or schedule for policy updates.

 required
alpha_value

Step size or schedule for value-function updates.

 required
gamma

Discount factor in [0, 1].

 required
max_episode_length

Maximum number of steps per episode.

 1000
verbose

Whether to print step-by-step diagnostics.

 False
history

Whether to return policy parameters, returns, and episode lengths collected during training.

 False
rng

NumPy generator or integer seed used to sample actions.

 None

Returns:

Type Description

Tuple (theta, w) containing the learned policy and value parameters.

If history is true, returns (theta, w, history).
Source code in gym_classics2/algorithms/policy_gradient_methods.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def AC(
    env,
    state_features,
    n,
    alpha_policy,
    alpha_value,
    gamma,
    max_episode_length=1000,
    verbose=False,
    history=False,
    rng=None,
    ):
    """Run one-step actor-critic with linear policy and value approximators.

    Args:
        env: Episodic environment used to generate experience.
        state_features: Callable converting ``(state, env)`` to a feature vector.
        n: Number of training episodes.
        alpha_policy: Step size or schedule for policy updates.
        alpha_value: Step size or schedule for value-function updates.
        gamma: Discount factor in ``[0, 1]``.
        max_episode_length: Maximum number of steps per episode.
        verbose: Whether to print step-by-step diagnostics.
        history: Whether to return policy parameters, returns, and episode lengths
            collected during training.
        rng: NumPy generator or integer seed used to sample actions.

    Returns:
        Tuple ``(theta, w)`` containing the learned policy and value parameters.
        If ``history`` is true, returns ``(theta, w, history)``.
    """

    assert gamma >= 0 and gamma <= 1, "gamma must be in [0  ,1]"
    assert n > 0, "number of episodes must be positive"
    assert max_episode_length > 0, "max episode length must be positive"

    rng = get_rng(rng)

    if isinstance(env.observation_space, gym.spaces.Discrete):
        warnings.warn("The environment has a discrete state space. Consider using a tabular method instead of function approximation.")

    if isinstance(alpha_policy, float):
        alpha_policy = ConstantSchedule(alpha_policy)
    if isinstance(alpha_value, float):
        alpha_value = ConstantSchedule(alpha_value) 

    state, _ = env.reset()

    # for simplicity we use the same features for the value function and the policy approximation
    # value function weights
    w = np.zeros(len(state_features(state, env)))
    # policy weights
    theta = np.zeros(len(state_action_features(state, 0, env, state_features)))

    if history:
        returns = []        
        ep_lens = []
        ws = []
        ws.append(w.copy())
        thetas = []
        thetas.append(theta.copy())

    for episode in tqdm(range(n), desc="Episodes", disable=verbose):
        if verbose:
            print(f"Episode {episode+1}/{n}")

        disc_factor = 1.0
        state, _ = env.reset()
        done = False
        i = 0
        G = 0.0

        while not done and  i < max_episode_length:
            # use actor to determine next action
            a = rng.choice(
                env.action_space.n,
                p=pi(state, theta, env, state_features),
            )

            # execute action
            next_state, r, done, _, _ = env.step(a)

            # use critic to calculate  TD error
            td_error = r + gamma * np.dot(w, state_features(next_state, env)) - np.dot(w, state_features(state, env))

            # update critic
            w += alpha_value(episode) * td_error * state_features(state, env)

            # update actor
            grad_log_pi = state_action_features(
                state, a, env, state_features
            ) - sum(
                pi(state, theta, env, state_features)[b]
                * state_action_features(state, b, env, state_features)
                for b in range(env.action_space.n)
            )
            theta += alpha_policy(episode) * disc_factor * td_error * grad_log_pi

            G += disc_factor * r
            disc_factor *= gamma      
            state = next_state
            i += 1

        if history:
            thetas.append(theta.copy())
            ws.append(w.copy())
            returns.append(G)
            ep_lens.append(i)

    if history:
        return theta, w, {'thetas': thetas, 'returns': returns, 'ep_lens': ep_lens}

    return theta, w

Schedules

gym_classics2.algorithms.schedules

Scalar schedules for reinforcement-learning hyperparameters.

A schedule is a callable that maps a nonnegative episode or step index t to a floating-point value. The included algorithms use schedules to vary parameters such as the step size (alpha) and exploration rate (epsilon) during training.

For example, a linear schedule can decrease epsilon from 1.0 to 0.1 over 10 steps:

schedule = LinearDecaySchedule(1.0, min_value=0.1, decay_steps=10)
for t in (0, 5, 10, 15):
    print(t, schedule(t))
# 0 1.0
# 5 0.55
# 10 0.1
# 15 0.1

Schedule

Interface for a scalar value indexed by episode or time step.

Source code in gym_classics2/algorithms/schedules.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
class Schedule:
    """Interface for a scalar value indexed by episode or time step."""

    def __call__(self, t):
        """Evaluate the schedule.

        Args:
            t: Nonnegative episode or step index.

        Returns:
            Scheduled scalar value at ``t``.
        """
        raise NotImplementedError

__call__ 

__call__(t)

Evaluate the schedule.

Parameters:

Name Type Description Default
t

Nonnegative episode or step index.

 required

Returns:

Type Description

Scheduled scalar value at t.
Source code in gym_classics2/algorithms/schedules.py 
26
27
28
29
30
31
32
33
34
35
def __call__(self, t):
    """Evaluate the schedule.

    Args:
        t: Nonnegative episode or step index.

    Returns:
        Scheduled scalar value at ``t``.
    """
    raise NotImplementedError

ConstantSchedule

Bases: Schedule

Return the same value for every index.

Parameters:

Name Type Description Default
value

Constant value returned by the schedule.

 required
Source code in gym_classics2/algorithms/schedules.py 
37
38
39
40
41
42
43
44
45
46
47
class ConstantSchedule(Schedule):
    """Return the same value for every index.

    Args:
        value: Constant value returned by the schedule.
    """
    def __init__(self, value):
        self.value = float(value)

    def __call__(self, t):
        return self.value

StepSchedule

Bases: Schedule

Switch once from a high value to a low value.

The schedule returns high_value while t < steps and low_value from t == steps onward.

Parameters:

Name Type Description Default
high_value

Value before the switch.

 required
low_value

Value at and after the switch.

 required
steps

Index at which to switch to low_value.

 required
Source code in gym_classics2/algorithms/schedules.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class StepSchedule(Schedule):
    """Switch once from a high value to a low value.

    The schedule returns ``high_value`` while ``t < steps`` and ``low_value``
    from ``t == steps`` onward.

    Args:
        high_value: Value before the switch.
        low_value: Value at and after the switch.
        steps: Index at which to switch to ``low_value``.
    """
    def __init__(self, high_value, low_value, steps):
        self.high_value = float(high_value)
        self.low_value = float(low_value)
        self.steps = int(steps)

    def __call__(self, t):
        if t < self.steps:
            return self.high_value
        return self.low_value

LinearDecaySchedule

Bases: Schedule

Interpolate linearly from an initial value to a final value.

The schedule returns initial_value at t = 0, changes linearly through decay_steps, and returns min_value thereafter.

Parameters:

Name Type Description Default
initial_value

Value at index zero.

 required
min_value

Final value and lower endpoint of a decreasing schedule.

 required
decay_steps

Number of indices over which to interpolate.

 required
Source code in gym_classics2/algorithms/schedules.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class LinearDecaySchedule(Schedule):
    """Interpolate linearly from an initial value to a final value.

    The schedule returns ``initial_value`` at ``t = 0``, changes linearly through
    ``decay_steps``, and returns ``min_value`` thereafter.

    Args:
        initial_value: Value at index zero.
        min_value: Final value and lower endpoint of a decreasing schedule.
        decay_steps: Number of indices over which to interpolate.
    """
    def __init__(self, initial_value, min_value, decay_steps):
        self.initial_value = float(initial_value)
        self.min_value = float(min_value)
        self.decay_steps = int(decay_steps)

    def __call__(self, t):
        fraction = min(float(t) / max(1, self.decay_steps), 1.0)
        return self.initial_value + fraction * (self.min_value - self.initial_value)

ExponentialDecaySchedule

Bases: Schedule

Decay geometrically to a minimum value.

At index t, the value is max(min_value, initial_value * decay_rate**t).

Parameters:

Name Type Description Default
initial_value

Unclipped value at index zero.

 required
min_value

Lower bound for the returned value.

 required
decay_rate

Multiplicative factor applied at each successive index.

 required
Source code in gym_classics2/algorithms/schedules.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class ExponentialDecaySchedule(Schedule):
    """Decay geometrically to a minimum value.

    At index ``t``, the value is ``max(min_value, initial_value * decay_rate**t)``.

    Args:
        initial_value: Unclipped value at index zero.
        min_value: Lower bound for the returned value.
        decay_rate: Multiplicative factor applied at each successive index.
    """
    def __init__(self, initial_value, min_value, decay_rate):
        self.initial_value = float(initial_value)
        self.min_value = float(min_value)
        self.decay_rate = float(decay_rate)

    def __call__(self, t):
        return max(self.min_value, self.initial_value * (self.decay_rate ** t))

InverseDecaySchedule

Bases: Schedule

Decay in inverse proportion to the index.

The schedule returns initial_value at t = 0. For t > 0, it returns max(min_value, initial_value / t).

Parameters:

Name Type Description Default
initial_value

Value at index zero and numerator of the inverse schedule.

 required
min_value

Lower bound for the returned value.

 0.0
Source code in gym_classics2/algorithms/schedules.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
class InverseDecaySchedule(Schedule):
    """Decay in inverse proportion to the index.

    The schedule returns ``initial_value`` at ``t = 0``. For ``t > 0``, it
    returns ``max(min_value, initial_value / t)``.

    Args:
        initial_value: Value at index zero and numerator of the inverse schedule.
        min_value: Lower bound for the returned value.
    """
    def __init__(self, initial_value, min_value=0.0):
        self.initial_value = float(initial_value)
        self.min_value = float(min_value)

    def __call__(self, t):
        if t == 0:
            return self.initial_value
        return max(self.min_value, self.initial_value / t)

plot_schedule 

plot_schedule(schedule, steps=1000)

Plot scheduled values for indices 0 through steps - 1.

Example 
from gym_classics2.algorithms.schedules import (
    LinearDecaySchedule,
    plot_schedule,
)

epsilon = LinearDecaySchedule(1.0, min_value=0.1, decay_steps=1_000)
plot_schedule(epsilon, steps=1_000)

Parameters:

Name Type Description Default
schedule

Callable that accepts an integer index and returns a scalar.

 required
steps

Number of scheduled values to plot.

 1000
Source code in gym_classics2/algorithms/schedules.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def plot_schedule(schedule, steps=1000):
    """Plot scheduled values for indices ``0`` through ``steps - 1``.

    Example:
        ```python
        from gym_classics2.algorithms.schedules import (
            LinearDecaySchedule,
            plot_schedule,
        )

        epsilon = LinearDecaySchedule(1.0, min_value=0.1, decay_steps=1_000)
        plot_schedule(epsilon, steps=1_000)
        ```

    Args:
        schedule: Callable that accepts an integer index and returns a scalar.
        steps: Number of scheduled values to plot.
    """
    import matplotlib.pyplot as plt

    values = [schedule(t) for t in range(steps)]

    plt.figure(figsize=(8, 5))
    plt.plot(range(steps), values, linewidth=2)
    plt.xlabel('Step')
    plt.ylabel('Schedule Value')
    plt.title('Schedule Plot')
    plt.grid(True)
    plt.show()