<jupyter_start><jupyter_text>Consider a measurable space $(\Omega, \mathcal{F})$. A map$Q\colon\Omega \times \mathcal{F} \to [0, +\infty)$ is a transitionkernel if* $Q(\cdot, A)$ is measurable map for any $A \in \mathcal{F}$* $Q(\omega, \cdot)$ is a finite measure on $(\Omega, \mathcal{F})$   for any $\omega \in \Omega$ For any measurable map $f$ we can define a measurable map$$    T f    \colon x \mapsto \int f(\omega) \, Q(x, d\omega)    \,. $$For a measure $\lambda$ on $(\Omega, \mathcal{F})$ we can similarlydefine a measure$$    T^* \lambda    \colon A \mapsto \int Q(\omega, A) \lambda(d\omega)    \,. $$ A dual pairing:$$\langle    f, \lambda\rangle = \int f(\omega) \lambda(d\omega)    \,. $$We can show that $\langle    T f, \lambda\rangle = \langle        f, T^* \lambda    \rangle$ via Fubini (?) theorem:$$\int T f(x) \lambda(dx)    = \int \int f(\omega) \, Q(x, d\omega) \lambda(dx)    % = \iint f(\omega) \lambda(dx) Q(x, d\omega)    = \int f(\omega) \int \lambda(dx) Q(x, d\omega)    = \int f(\omega) (T^* \lambda)(d\omega)    \,. $$<jupyter_code>import numpy as np

%matplotlib inline
import matplotlib.pyplot as plt
import torch
import torch.nn.functional as F
import gym<jupyter_output><empty_output><jupyter_text>A display device<jupyter_code>from IPython.display import clear_output

from time import sleep

def display(env, fps=15):
    if fps > 0:
        clear_output(wait=True)
        env.render()
        sleep(1. / fps)<jupyter_output><empty_output><jupyter_text>Environment simulation loop:<jupyter_code>from tools.delayed import DelayedKeyboardInterrupt

def run(env, policy, *, fps=15):
    state, terminate = env.reset(), False

    display(env, fps)
    with DelayedKeyboardInterrupt("ignore") as stop:
        while not (terminate or stop):
    
            # take an action and get a response form the environment
            action = policy(env, state)
            state_prime, reward, terminate, info = env.step(action)

            # return the result
            yield state, action, reward, state_prime, terminate
            state = state_prime

            # render the enviroment
            display(env, fps)

        env.close()<jupyter_output><empty_output><jupyter_text><jupyter_code>from gym.envs.toy_text import FrozenLakeEnv, CliffWalkingEnv

env = FrozenLakeEnv(map_name="8x8", is_slippery=True)
# env = gym.make('CliffWalking-v0')<jupyter_output><empty_output><jupyter_text>Random policy<jupyter_code>def random_policy(env):
    return {
        state: [(1. / env.nA, action) for action in kernel]
        for state, kernel in env.P.items()
    }

def random(env, state=None):
    return env.action_space.sample()

episode = [*run(env, random)]<jupyter_output><empty_output><jupyter_text>From Whiteson lecture on MLSS 2019 MDP -- classical formal model of a sequential decision problem:* fully-observable, stationary, and possibly stochastic environment* discrete states $S$ and actions $A_s$ for each $s \in S$* transition kernel $s\to z \colon z\sim q(z\mid s, a)$ on $S$* reward distributuion $q(r\mid s, a, s')$ when transitioning $s \to s'$ under $a$* aplanning horizon or a discount factor $\gamma \in (0, 1)$ Markov property$p(z_{t+1}, r_{t+1}\mid s_t, a_t) = p(z_{t+1}, r_{t+1}\mid s_t, a_t, s_{:t}, a_{:t})$* Reactive policies $a\sim \pi(a\mid s)$* deterministic policies **State-value** of a policy $v^\pi(s) = \mathbb{E}_\pi \bigl(    \sum_{k\geq 1} \gamma r_{t+k+1}    \big\vert s_t = s\bigr)$ and **action-value** $Q^\pi(s, a) = \mathbb{E}_\pi \bigl(    \sum_{k\geq 1} \gamma r_{t+k+1}    \big\vert s_t = s, a_t = a\bigr)$ Bellman fixed-point equation for $v^\pi$:$$v^\pi(s)    = \mathbb{E}_{a\sim \pi(s)} \mathbb{E}_{s' \sim q(s'\mid s, a)}        r(s, a, s') + \gamma v^\pi(s')    \,, $$and for $q^\pi$$$q^\pi(s, a)    = \mathbb{E}_{s' \sim q(s'\mid s, a)}        r(s, a, s') + \gamma \mathbb{E}_{a'\sim \pi(s')} q^\pi(s', a')    = \mathbb{E}_{s' \sim q(s'\mid s, a)}        r(s, a, s') + \gamma v^\pi(s')    \,. $$ Policies can be partially oredered by their value function. Andall optimal policies share the same optimal state valeu function$v^*(\cdot) = \max_\pi v^\pi(\cdot)$. The Bellamn optimality conditions for $v^*$ and are$$v^*(s)    = \max_{a\in A_s} \mathbb{E}_{s' \sim q(s'\mid s, a)}        r(s, a, s') + \gamma v^*(s')    \,, $$and$$q^*(s, a)    = \mathbb{E}_{z \sim q(s'\mid s, a)}        r(s, a, s') + \gamma \max_{a'\in A_{s'}}  q^*(s', a')    \,, $$respectively. The optimal policy is greedy with respect to $q$:$$\pi^*(s)    = \delta_{a^*_s}    \,, \text{ for }    a^*_s = \arg\max_{a\in A_s} q^*(s, a)    \,. $$   Policy Evaluation Bellman operator for a policy $\pi\colon \mathcal{S} \to \Delta_A$$$T_\pi(v)\colon s \mapsto \mathbb{E}_{a\sim \pi(a\mid s)}    \mathbb{E}_{z\sim q(z\mid s, a)} r(s, a, z) + \gamma v(z)    \,. $$<jupyter_code>def expected_state_reward(states, value, gamma=1.0):
    # kernel -- list of next state-rewards with probabilities
    return sum(
        prob * (reward + gamma * value[state])
        for prob, state, reward, term in states
    )

def expected_action_reward(actions, kernel, value, gamma=1.0):
    # policy -- list of actions with probabilities
    return sum(
        prob * expected_state_reward(kernel[action], value, gamma)
        for prob, action in actions
    )<jupyter_output><empty_output><jupyter_text>The policy evaluation is performed via the fixed point iterations:* repeat $v_{t+1} \leftarrow T_\pi(v_t)$ until convergence in $\|\cdot\|_\infty$<jupyter_code>def evaluate_policy(env, policy, gamma=1.0, atol=1e-8):
    value, delta = {state: 0. for state in env.P}, float("+inf")
    while delta > atol:
        Tv = {
            state: expected_action_reward(policy[state], kernel, value, gamma)
            for state, kernel in env.P.items()
        }
        
        delta = max(abs(a - b) for a, b in zip(Tv.values(), value.values()))
        value = Tv

    return value<jupyter_output><empty_output><jupyter_text>The $q$-function of $v$ is$$q_\infty(s, a)    = \mathbb{E}_{s'\sim q(s'\mid s, a)}        r(s, a, s') + \gamma v_\infty(s')    \,. $$<jupyter_code>def q_fun(env, value, gamma=1.0):
    return {
        state: {
            action: expected_state_reward(states, value, gamma)
            for action, states in kernel.items()
        }
        for state, kernel in env.P.items()
    }<jupyter_output><empty_output><jupyter_text>Let's evaluate a random exploration policy:<jupyter_code>gamma = 0.9

policy = random_policy(env)
value = evaluate_policy(env, policy, gamma=gamma)
value<jupyter_output><empty_output><jupyter_text>The stationary $v$-function induces the following policy (consistent with it):$$\pi(s)    = \delta_{a_s}    \,, \text{ for } a_s = \arg \max_{a \in A_s} Q(s, a)     \,. $$<jupyter_code>def get_greedy_from_q(q):
    def _policy(env, state):
        actions, expected = zip(*q[state].items())
        return actions[np.argmax(expected)]

    return _policy
greedy = get_greedy_from_q(q_fun(env, value, gamma=gamma))
episode = [*run(env, greedy)]<jupyter_output><empty_output><jupyter_text>Policy improvement The fixed point $v^\pi$ is the true value function of $\pi$. The associated $q$ function ca be used to reason about improvements in the policy $\pi$:if at some $s\in S$ we have $q^\pi(s, a_s) > v^\pi(s)$ for some $a_s \in A_s$ then the new policy $\hat{\pi}(\cdot) = \pi(\cdot)$ but $\hat{\pi}(s) = \delta_{a_s}$ is strictly better than $\pi$ (w.r.t $v$-function) Applying this to all states yields the **greedy** policy improvement:$$\pi_{t+1}(s) \in \arg\max_{a\in A_s} q^{\pi_t}(s, a)    \,. $$<jupyter_code>def better_policy(q):
    policy = {}
    for state, value in q.items():
        # put equal mass on the actions with the maximal expected reward
        v_max = max(value.values())
        action = [a for a, v in value.items() if v >= v_max]
        policy[state] = [(1. / len(action), a) for a in action]
    
    return policy<jupyter_output><empty_output><jupyter_text>If $\pi_{t+1} = \pi_t$ then $v^{\pi_t} = v^{\pi_{t+1}} = v$, whichsatisfies the Bellamn Optimiality principle:* $T(v) = v$ for$$T(v)\colon S \to \mathbb{R}\colon s \mapsto \max_{a\in A_s}    \mathbb{E}_{z\sim q(s'\mid s, a)} r(s, a, s') + \gamma v(s')    \,. $$<jupyter_code>def policy_iteration(env, gamma=1.0, atol=1e-8):
    policy = random_policy(env)
    value, delta = evaluate_policy(env, policy, gamma), float("+inf")
    while delta > atol:
        policy = better_policy(q_fun(env, value, gamma))
        new = evaluate_policy(env, policy, gamma)

        delta = max(abs(a - b) for a, b in zip(new.values(), value.values()))
        value = new

    return value, policy
value, policy = policy_iteration(env, gamma=gamma)<jupyter_output><empty_output><jupyter_text>Use a truly stochastic policy<jupyter_code>def get_random_from_pi(policy):
    def _policy(env, state):
        probs, actions = zip(*policy[state])
        return np.random.choice(actions, p=probs)

    return _policy
episode = [*run(env, get_random_from_pi(policy))]<jupyter_output><empty_output><jupyter_text>Takes a stochastic policy and uses MAP action prediction<jupyter_code>def get_greedy_from_pi(policy):
    def _policy(env, state):
        probs, actions = zip(*policy[state])
        return actions[np.argmax(probs)]

    return _policy
episode = [*run(env, get_greedy_from_pi(policy))]<jupyter_output><empty_output><jupyter_text>Value iteration The operator in Bellaman's optimiality conditions acts on $v$ thus$$T(v)\colon S \to \mathbb{R}\colon s \mapsto \max_{a\in A_s} \mathbb{E}_{s'\sim q(s'\mid s, a)}    r(s, a, s') + \gamma v(s')    \,. $$<jupyter_code>def optimal_action_reward(kernel, value, gamma=1.0):
    return max(
        expected_state_reward(states, value, gamma)
        for action, states in kernel.items()
    )<jupyter_output><empty_output><jupyter_text>It can be shown that $T$ is a contraction mapping w.r.t $\|\cdot\|_\infty$ for $\gamma \in (0, 1)$,and thus the fixed point iteration converges to a $v_*$:* repeat $v_{t+1} \leftarrow T(v_t)$ until convergence in $\|\cdot\|_\infty$<jupyter_code>def value_iteration(env, gamma=1.0):
    value, delta = {state: 0. for state in env.P}, float("+inf")
    while delta > 1e-8:
        # compute the operator
        new = {state: optimal_action_reward(kernel, value, gamma)
               for state, kernel in env.P.items()}
        
        delta = max(abs(a - b) for a, b in zip(value.values(), new.values()))
        value = new

    return value<jupyter_output><empty_output><jupyter_text>Compute the $q$-function implied by the converged value function:$$Q(s, a)    = \mathbb{E}_{z\sim q(z\mid s, a)} r(s, a, s') + \gamma v^*(z)    \,. $$<jupyter_code>value = value_iteration(env, gamma=gamma)

q = q_fun(env, value, gamma=gamma)<jupyter_output><empty_output><jupyter_text>The optimal value function implies an optimal (greedy) poilcy:$$\pi(s) \in \arg\max_{a\in A_s}    \mathbb{E}_{z\sim q(z\mid s, a)} r(s, a, z) + \gamma v^*(z)    \,. $$<jupyter_code>episode = [*run(env, get_greedy_from_q(q))]
assert False<jupyter_output><empty_output><jupyter_text>Monte Carlo methods Provides amethod for finding an optimal policy without knowing the underlying MDP:* Learn $v^\pi$ without a model* unlike DP, MC uses the entire episode, and does not bootstrap (with $v^\pi(s')$* unlike DP, MC observes only one choice at each state $s\in S$Learn $q^\pi$ by averaging returns obtained when following $\pi$ after taking $a \in A_s$ at $s$:1. (**Monte Carlo**) policy evaluation: use MC method to get $q^\pi$2. (**greedy**) policy improvement: refine $\pi(s) \leftarrow g(s) = \arg \max_{a\in A_s} q^\pi(s, a)$   * soften by mixing with a uniform policy:$    \pi(a\vert s) = (1 - \varepsilon) \, \delta_{g(s)}(a) + \varepsilon \, \mathrm{U}_{A_s}(a)$PI theorem guarantees that and $\varepsilon$-greedy policy improves soft $\varepsilon$-greedy policy.<jupyter_code>def soften(policy, epsilon=1e-1):
    def _policy(env, state):
        if np.random.uniform(1) <= epsilon:
            return random(env, state)

        return policy(env, state)

    return _policy<jupyter_output><empty_output><jupyter_text>Off-policy MC control* **estimation policy** is evaluated on samples from**behaviour policy**, provided it is sufficiently exploratory* use importance sampling to re-weight returns:$\mathbb{E}_{z \sim P} h    = \mathbb{E}_{z \sim Q} w \, h% \mathbb{E}_{z \sim P(z)} h(z)%     = \mathbb{E}_{z \sim Q(z)} w_z h(z)$for $w = \tfrac{dP}{dQ}$ provided $P \ll Q$  * $P$ comes from the estimation policy, $Q$ -- from the behavioural policy $$v^\pi    \approx \frac{        \hat{\mathbb{E}}_{i\sim D_s} \tfrac{p_i(s)}{p_i'(s)} R_i(s)    }{        \hat{\mathbb{E}}_{i\sim D_s} \tfrac{p_i(s)}{p_i'(s)}    }    \,, $$for$$\frac{p_i(s)}{p_i'(s)}    = \frac{        \prod_{k=t}^{T_i(s)-1} \pi(a_k \vert s_k) q(s_{k+1}\vert a_k, s_k)    }{        \prod_{k=t}^{T_i(s)-1} \pi'(a_k \vert s_k) q(s_{k+1}\vert a_k, s_k)    }    = \prod_{k=t}^{T_i(s)-1} \frac{\pi(a_k \vert s_k)}{\pi'(a_k \vert s_k)}    \,, $$  TD(0) algorithmValue function estimation from the experience (given by simulating from the behavioural policy)* $v(s_t) \leftarrow v(s_t) + \alpha (R_t + \gamma v(s_{t+1}) - v(s_t))$-- bootstrap from the existing estimateSamples and bootstraps<jupyter_code>from collections import defaultdict

def td_predict(env, policy, n_episodes=100, alpha=0.95, gamma=1.0, fps=15):
    value = defaultdict(float)
    # for each episode ...
    with DelayedKeyboardInterrupt("ignore") as stop:
        for n_episode in range(n_episodes):
            # (lazily) play out the policy ...
            episode = run(env, policy, fps=fps)

            # run TD(0) updates ...
            for state, action, reward, next_state, terminal in episode:
                # ... for the value function of the policy
                bootstrap = reward + gamma * value[next_state]
                value[state] += alpha * (bootstrap - value[state])

            if stop:
                break
    return value
piter = get_random_from_pi(policy)
# policy = soften(get_greedy_from_q(q), epsilon=1e-1)
policy = soften(piter, epsilon=1e-1)

value = td_predict(env, policy, fps=0)<jupyter_output><empty_output><jupyter_text>TD: control SARSASarsa algorithm<jupyter_code>def sarsa_evaluate(env, n_episodes=10000, alpha=0.01, gamma=1.0, fps=15):
    q_fun = defaultdict(lambda: defaultdict(float))

    def q_policy(env, state):
        if q_fun[state]:
            actions, expected = zip(*q_fun[state].items())
            return actions[np.argmax(expected)]
        return random(env, state)

    # for each episode ...
    for n_episode in range(n_episodes):

        _feedback = soften(q_policy, epsilon=1e-2)
        
        # play out the policy ...
        episode = run(env, _feedback, fps=fps)

        # run TD(0) updates ...
        state, action, reward, next_state, terminal = next(episode)
        while not terminal:
            # ... for the q-function function of the policy
            next_state, next_action, *rest = next(episode)

            bootstrap = reward + gamma * q_fun[next_state][next_action]
            q_fun[state][action] += alpha * (bootstrap - q_fun[state][action])

            state, action = next_state, next_action
            reward, next_state, terminal = rest
    
    return q_fun
q_fun = sarsa_evaluate(env, fps=0)
q_fun
grp = {}
for (k, a), v in q_fun.items():
    grp.setdefault(k , []).append((a, v))
q_fun = {k: v for k, v in zip(grp, map(dict, grp.values()))}
q_fun
assert False<jupyter_output><empty_output><jupyter_text>Policy improvement and Policy Iteration<jupyter_code>from matplotlib.collections import LineCollection
s, *_ = zip(*episode)

segs = np.c_[np.unravel_index(s, (8, 8))]
assert False<jupyter_output><empty_output><jupyter_text><jupyter_code># !pip install gym
import numpy as np
%matplotlib inline
import matplotlib.pyplot as plt
import torch
import torch.nn.functional as F
import gym
from gym.envs.classic_control import MountainCarEnv
from gym import spaces

class ModifiedMountainCarEnv(MountainCarEnv):
    def __init__(self, goal_velocity = 0):
        self.min_position = -2.5
        self.max_position = 2.5
        self.max_speed = 0.07
        self.goal_position = 2.0
        self.goal_velocity = goal_velocity

        self.force = 0.001
        self.gravity = 0.0025

        self.low = np.array([self.min_position, -self.max_speed])
        self.high = np.array([self.max_position, self.max_speed])

        self.viewer = None

        self.action_space = spaces.Discrete(3)
        self.observation_space = spaces.Box(self.low, self.high, dtype=np.float32)

        self.seed()
env = ModifiedMountainCarEnv()
def run(env, agent, fps=15):
    state, terminated = env.reset(), False
    history = []
    while not terminated:
        state, reward, terminated, info = env.step(agent(state))
        history.append((state, reward))

    return history
class BaseAgent(object):
    def __init__(self, env):
        self.env = env
    
    def reset(self):
        pass
    
    def update(self, state, action, reward, next_state, terminated=False):
        pass

    def __call__(self, state=None):
        return env.action_space.sample()
space = env.observation_space
from gym.spaces import Space
shape = (51, 71)
state = 0, 0
unit = (state - space.low) / (space.high - space.low)
class Discretizer(object):
    def __init__(self, space, shape):
        assert isinstance(space, Box) and len(space.shape) == 1
        assert space.is_bounded

        if not isinstance(n_states, (list, tuple)):
            shape = space.shape[0] * [shape]

        assert shape == space.shape[0]
        self.space, self.shape = space, shape

    def to_ix(self, state, flatten=False):
        unit = (state - space.low) / (space.high - space.low)
        ix = (unit * shape + 0.5).astype(int)

        if flatten:
            return np.unravel_index(ix, shape=self.shape)
        return ix

    def from_ix(self, *index):
        return np.array(index) * (space.high - space.low) + space.low
def rescale(obs, env):
    space = env.observation_space
    return (obs - space.low) / (space.high - space.low)

class TabularQLearner(BaseAgent):
    def __init__(self, env, n_states=51):
        super().__init__(env)
        self.n_states, self.n_actions = n_states, env.action_space.n

        self.reset()
    
    def reset(self):
        self.q_table = torch.zeros(self.n_states, self.n_states, n_actions)

    def __call__(self, state=None):
        
        return env.action_space.sample()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1531177704847474, 'fraction_numerical': 0.007585947167375047, 'mean_word_length': 3.28690036900369, 'pattern_counts': {'":': 0, '<': 94, '<?xml version=': 0, '>': 99, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '504728', 'license': 'MIT', 'n_tokens_mistral': 6712, 'n_tokens_neox': 6444, 'n_words': 1748, 'path': 'coding/notes_on_rl.ipynb', 'repo_name': 'ivannz/general-scribbles', 'size': '18587'}