Examples

Game

Exercise (intermediate)

1. Implement and test the game problem from the MDPFromMapping.

2. Implement and test the value iteration algorithm. The output of this algorithm is a policy mapping and a value mapping (the optimal ones).

Hint 2.

To keep things more readable, a function qualities_from_state is also implemented.

src

def value_iteration_empty[S: State](
    mdp: MDPProblem[S], discount: float = 1.0, convergence_limit: float = 1e-3
) -> tuple[PolicyMapping, ValuesMapping]:
    """approximate the optimal values and the optimal policy for every state."""
    ...

src

def qualities_from_state_empty[S: State](
    mdp: MDPProblem[S], state: S, values: ValuesMapping, discount: float
) -> set[tuple[Reward, Action]]:
    """compute the current approximation of the qualities from 'state'
    using the current approximations of values

    returns a set of tuples(reward, action) in order to be able to arg max on action
    to define the optimal policy"""
    ...

Correction

1. These tests are almost trivial. They are there to force you to clearly understand the defined types.

mdp_mapping_game: MDPMapping = {
   "in": {
      "stay": {"in": (2.0 / 3.0, 4.0), "out": (1.0 / 3.0, 5.0)},
      "quit": {"out": (1.0, 10.0)},
   },
   "out": {},
}

tests

@pytest.fixture
def game_mdp():
    mdp_mapping_game: MDPMapping = {
        "in": {
            "stay": {"in": (2.0 / 3.0, 4.0), "out": (1.0 / 3.0, 5.0)},
            "quit": {"out": (1.0, 10.0)},
        },
        "out": {},
    }
    return MDPFromMapping(mdp_mapping_game)

tests

def test_actions_from_state(game_mdp):
    assert game_mdp.actions_from_state("in") == {"quit", "stay"}
    assert game_mdp.actions_from_state("out") == set()


def test_outcomes_from_state_and_action(game_mdp):
    assert game_mdp.outcomes_from_state_and_action("in", "stay") == {
        "in": (2.0 / 3.0, 4.0),
        "out": (1.0 / 3.0, 5.0),  # dubious == with floats
    }
    assert game_mdp.outcomes_from_state_and_action("in", "quit") == {"out": (1.0, 10.0)}


def test_states(game_mdp):
    assert game_mdp.states == {"in", "out"}


def test_is_end_state(game_mdp):
    assert game_mdp.is_end_state("out")

src

def value_iteration[S: State](
    mdp: MDPProblem[S], discount: float = 1.0, convergence_limit: float = 1e-3
) -> tuple[PolicyMapping, ValuesMapping]:
    """approximate the optimal values and the optimal policy for every state."""
    states: set[S] = mdp.states
    values: ValuesMapping = {}.fromkeys(states, 0.0)
    policy: PolicyMapping = {}
    is_converged = False
    while not is_converged:
        is_converged = True
        for state in states:
            if not mdp.is_end_state(state):
                qualities: set[tuple[Reward, Action]] = qualities_from_state(
                    mdp, state, values, discount
                )
                quality_opt, action_opt = max(qualities)
                if abs(values[state] - quality_opt) > convergence_limit:
                    # if any value for any state changes significantly,
                    # the process has not converged
                    is_converged = False
                values[state] = quality_opt
                policy[state] = action_opt
    return policy, values

src

def qualities_from_state[S: State](
    mdp: MDPProblem[S], state: S, values: ValuesMapping, discount: float
) -> set[tuple[Reward, Action]]:
    """compute the current approximation of the qualities from 'state'
    using the current approximations of values

    returns a set of tuples(reward, action) in order to be able to arg max on action
    to define the optimal policy"""
    qualities = set()
    for action in mdp.actions_from_state(state):
        outcomes: Outcomes = mdp.outcomes_from_state_and_action(state, action)
        # for a given state: outcomes = dict[Next_State, tuple[Transition, Reward]]
        # from the definition of qualities (with the current approx of values):
        quality: Reward = sum(
            transition * (reward + discount * values[next_state])
            for next_state, (transition, reward) in outcomes.items()
        )
        qualities.add((quality, action))
    return qualities

tests

import pytest

from ai.reinforcement.mdp_types import Reward, Action, State, ValuesMapping
from ai.reinforcement.mdp_iteration import qualities_from_state, value_iteration


@pytest.mark.parametrize(
    "values, expected",
    [
        ({"in": 0.0, "out": 0.0}, {"stay": 13.0 / 3.0, "quit": 10.0}),
        ({"in": 10.0, "out": 0.0}, {"stay": 11.0, "quit": 10.0}),
    ],
)
def test_qualities_from_state_in(
    game_mdp, values: ValuesMapping, expected: dict[Action, Reward]
):
    # stay: (4 + 1*0) * 2/3 + (5 + 1*0) * 1/3 = (8+5)/3 = 13/3
    # quit: (10 + 1*0) * 1 = 10
    # stay: (4 + 1*10) * 2/3 + (5 + 1*0) * 1/3 = (28+5)/3 = 11
    # quit: (10 + 1*0) * 1 = 10
    qualities_in: set[tuple[Reward, Action]] = qualities_from_state(
        game_mdp, "in", values, 1.0
    )
    for value, action in qualities_in:
        assert pytest.approx(value) == expected[action]


def test_value_iteration(game_mdp):
    policy, values = value_iteration(game_mdp, 1.0, 1e-12)
    assert policy == {"in": "stay"}
    assert pytest.approx(values["in"]) == 13.0
    assert pytest.approx(values["out"]) == 0.0

Magic tramway

Extending the magic tram problem from the state problem section, a failure probability is introduced when taking the magic tram. Taking the tram has still a higher cost than walking, still can make you move (state x tram_factor) but now with a fail probability to leave you at your current state.

Magic tram fail

Exercise (intermediate)

1. Implement a MDPMagicTram problem from the MDPProblem and test it.

2. Interpret the value iteration algorithm for different values of the failure probability.

Correction

1. src

   @dataclass(frozen=True)
   class MDPMagicTram(MDPProblem[StateMagicTram]):
       """already encountered magic tram example from CS221.
       from starting point to destination only moving forward, walking has a cost and makes you move +1.
       Taking the tram has a higher cost, but makes you move (state x tram_factor) with a fail probability to leave
       you at your current state"""
   
       starting_point: StateMagicTram
       destination: StateMagicTram
       fail_probability: float = 0.0
       tram_factor: int = 2
       tram_cost: float = 2.0
       walk_cost: float = 1.0
   
       @override
       def actions_from_state(self, state: StateMagicTram) -> set[Action]:
           if state == self.destination:
               return set()
           # must not risk to overshoot with tram
           if state * self.tram_factor <= self.destination:
               return {"walk", "tram"}
           return {"walk"}
   
       @override
       def outcomes_from_state_and_action(
           self, state: StateMagicTram, action: Action
       ) -> Outcomes:
           if action == "walk":
               return {state + 1: (1.0, -self.walk_cost)}
   
           outcomes: Outcomes = {
               state * self.tram_factor: (1.0 - self.fail_probability, -self.tram_cost),
               state: (self.fail_probability, -self.tram_cost),
           }
           return outcomes
   
       @property
       @override
       def states(self) -> set[StateMagicTram]:
           return set(range(self.starting_point, self.destination + 1))
   
       @override
       def is_end_state(self, state: State) -> bool:
           return state == self.destination
   
       def to_string(
           self,
           policy: PolicyMapping,
           values: ValuesMapping,
       ) -> str:
           table = []
           for state, action in policy.items():
               if action == "tram":
                   table.append(
                       [state, Fore.RED + action + Style.RESET_ALL, f"{values[state]:.1f}"]  # type: ignore
                   )
               else:
                   table.append([state, action, f"{values[state]:.1f}"])
           table.append([self.destination, "", 0.0])
           return tabulate(table, headers=["state", "action", "value"], tablefmt="grid")
   

   tests

   import pytest
   
   
   from ai.reinforcement.mdp_example_tram import MDPMagicTram
   
   
   @pytest.fixture
   def tram():
       return MDPMagicTram(2, 5, 0.1)
   
   
   def test_actions_from_state(tram):
       assert tram.actions_from_state(2) == {"walk", "tram"}
       assert tram.actions_from_state(5) == set()
       assert tram.actions_from_state(4) == {"walk"}
   
   
   def test_outcomes_from_state_and_action(tram):
       assert tram.outcomes_from_state_and_action(2, "tram") == {
           2: (tram.fail_probability, -tram.tram_cost),
           4: (1.0 - tram.fail_probability, -tram.tram_cost),  # dubious == with floats
       }
       assert tram.outcomes_from_state_and_action(2, "walk") == {3: (1.0, -1.0)}
   
   
   def test_is_end_state(tram):
       assert tram.is_end_state(tram.destination)
   

2. For the case of 2 to 13:

   • For a failure probability of 0, the problem is identical to the state problem already solved during a previous session. The value of state 2 is supposed to be identical to the total cost of the state problem.

   • For a failure probability of 1, it is obvious that one should not take the tram.

   • For a failure probability of 0.2, the value of state 2 is logically larger than a failure probability of 0. However, it is still worth it to take the tram at state 3 and 6.

   • For a failure probability of 0.4, the value of state 2 is even higher than a failure probability of 0.2. However, now it is not worth it to take the tram at state 3, the risk of failure and the associated cost is to large regarding the length of the jump. The tram at state 6 is still worth it.

   failure probability 0.0

   state	action	value
   2	walk	-6
   3	tram	-5
   4	walk	-5
   5	walk	-4
   6	tram	-3
   7	walk	-6
   8	walk	-5
   9	walk	-4
   10	walk	-3
   11	walk	-2
   12	walk	-1
   13		0

   failure probability 1.0

   state	action	value
   2	walk	-11
   3	walk	-10
   4	walk	-9
   5	walk	-8
   6	walk	-7
   7	walk	-6
   8	walk	-5
   9	walk	-4
   10	walk	-3
   11	walk	-2
   12	walk	-1
   13		0

   failure probability 0.2

   state	action	value
   2	walk	-7
   3	tram	-6
   4	walk	-5.5
   5	walk	-4.5
   6	tram	-3.5
   7	walk	-6
   8	walk	-5
   9	walk	-4
   10	walk	-3
   11	walk	-2
   12	walk	-1
   13		0

   failure probability 0.4

   state	action	value
   2	walk	-8.3
   3	walk	-7.3
   4	walk	-6.3
   5	walk	-5.3
   6	tram	-4.3
   7	walk	-6
   8	walk	-5
   9	walk	-4
   10	walk	-3
   11	walk	-2
   12	walk	-1
   13		0

Swamp isle

The swamp isle problem is similar as the one presented in the state model section. The isle is rectangular, there are some swamps. The goal is to reach the end point limiting the cost. However, walking on dry land has a cost and walking in swamps has an even higher cost. One can walk either vertically or horizontally (not diagonally).

The novelty is that the problem is now a MDP because every action can lead to an equiprobable orientation failure in any of the other potential directions: the chance of no failure is 1-failure probability.

Exercise (intermediate)

1. Implement a MDPSwampIsle problem from the MDPProblem and test it.

2. Interpret the value iteration algorithm for different values of the failure probability.

Correction

1. src

   @dataclass(frozen=True)
   class MDPSwampIsle(MDPProblem[IsleState]):
       """the isle is rectangular, there are some swamps.
       The goal is to reach the end point limiting the cost.
       However, walking on dry land has a cost and walking in swamps has an even higher cost.
       One can walk either vertically or horizontally (not diagonally).
       Every action can lead to an orientation failure in any of the other potential directions: the chance of no failure is 1-failure probability.
       """
   
       nb_rows: int
       nb_cols: int
       end_point: Coordinates
       swamp_spots: set[Coordinates] = field(default_factory=set)
       fail_probability: float = 0.0
       walking_cost: Reward = 1.0
       swamp_cost: Reward = 20.0
   
       @override
       def actions_from_state(self, state: IsleState) -> set[Action]:
           coordinates = state
   
           if self.is_end_state(state):
               return set()
   
           LAST_COL = self.nb_cols - 1
           LAST_ROW = self.nb_rows - 1
           actions = {"N", "E", "S", "W"}
   
           # can not cross the domain edge
           if coordinates[0] == 0:
               actions.remove("N")
           if coordinates[0] == LAST_ROW:
               actions.remove("S")
           if coordinates[1] == 0:
               actions.remove("W")
           if coordinates[1] == LAST_COL:
               actions.remove("E")
   
           return actions
   
       @override
       def outcomes_from_state_and_action(
           self, state: IsleState, action: Action
       ) -> Outcomes:
           coordinates = state
   
           if coordinates in self.swamp_spots:
               reward = -self.swamp_cost
           else:
               reward = -self.walking_cost
   
           deltas: dict[Action, Coordinates] = {
               "N": (-1, 0),
               "E": (0, 1),
               "S": (1, 0),
               "W": (0, -1),
           }
   
           # target state
           delta = deltas[action]
           target: IsleState = (coordinates[0] + delta[0], coordinates[1] + delta[1])
   
           # keep potential fail directions
           actions: set[Action] = self.actions_from_state(state)
           actions_to_drop = set(deltas).difference(actions)
           actions_to_drop.add(action)
           for action_to_drop in actions_to_drop:
               del deltas[action_to_drop]
   
           # build outcomes
           outcomes: Outcomes = {target: (1.0 - self.fail_probability, reward)}
           nb_slip_directions = len(deltas)
           # the failing probability is uniformly spread on every fail direction
           fail_in_one_direction_probability = self.fail_probability / nb_slip_directions
           for delta in deltas.values():
               next_state: IsleState = (
                   coordinates[0] + delta[0],
                   coordinates[1] + delta[1],
               )
               outcomes[next_state] = (
                   fail_in_one_direction_probability,
                   reward,  # type: ignore
               )
           return outcomes
   
       @property
       @override
       def states(self) -> set[IsleState]:
           return {
               (row, col) for row in range(self.nb_rows) for col in range(self.nb_cols)
           }
   
       @override
       def is_end_state(self, state: IsleState) -> bool:
           coordinates = state
           return coordinates == self.end_point
   
       def to_string(
           self,
           policy: PolicyMapping | None = None,
           values: ValuesMapping | None = None,
       ) -> str:
           isle: list[list[str]] = []
           for row in range(self.nb_rows):
               isle.append(self.nb_cols * [""])
   
           # indicate swamps
           for row, col in self.swamp_spots:
               isle[row][col] += Back.GREEN + " X " + Style.RESET_ALL + "\n"  # type: ignore
   
           # indicate potential actions
           for row in range(self.nb_rows):
               for col in range(self.nb_cols):
                   coordinates: Coordinates = (row, col)
                   state: IsleState = coordinates
                   actions = self.actions_from_state(state)
                   if self.is_end_state(state):
                       actions = Back.RED + "END" + Style.RESET_ALL + "\n"  # type: ignore
   
                   isle[row][col] += f"{actions}\n"
   
           # indicate best action and expected value
           if policy is not None and values is not None:
               for state_, action in policy.items():
                   row, col = cast(IsleState, state_)
                   value: Reward = values[state_]
                   isle[row][col] += (
                       Fore.RED + f"[{action}] {value:.1f}" + Style.RESET_ALL  # type: ignore
                   )
   
           return tabulate(isle, tablefmt="grid")
   

   tests

   import pytest
   
   from ai.reinforcement.mdp_example_swamp import MDPSwampIsle
   
   
   @pytest.fixture
   def problem():
       return MDPSwampIsle(3, 4, (2, 3), {(0, 2), (0, 3), (1, 1), (1, 2), (2, 2)}, 0.1)
   
   
   def test_actions_from_state(problem):
       assert problem.actions_from_state((0, 0)) == {"S", "E"}
       assert problem.actions_from_state((0, 1)) == {"S", "E", "W"}
       assert problem.actions_from_state((1, 0)) == {"S", "E", "N"}
       assert problem.actions_from_state((1, 1)) == {"N", "S", "E", "W"}
       assert problem.actions_from_state((problem.nb_rows - 1, 0)) == {"N", "E"}
       assert problem.actions_from_state((problem.nb_rows - 1, 1)) == {"N", "E", "W"}
       assert problem.actions_from_state((0, problem.nb_cols - 1)) == {"S", "W"}
       assert problem.actions_from_state((1, problem.nb_cols - 1)) == {"S", "W", "N"}
       assert problem.actions_from_state(problem.end_point) == set()
   
   
   def test_outcomes_from_state_and_action(problem):
       assert problem.outcomes_from_state_and_action((0, 0), "S") == {
           (1, 0): (
               1.0 - problem.fail_probability,
               -problem.walking_cost,
           ),
           (0, 1): (
               problem.fail_probability,
               -problem.walking_cost,
           ),  # dubious == with floats
       }
       assert problem.outcomes_from_state_and_action((1, 0), "E") == {
           (0, 0): (
               problem.fail_probability / 2,
               -problem.walking_cost,
           ),
           (2, 0): (
               problem.fail_probability / 2,
               -problem.walking_cost,
           ),
           (1, 1): (
               (1.0 - problem.fail_probability),
               -problem.walking_cost,
           ),  # dubious == with floats
       }
   
   
   def test_is_end_state(problem):
       assert problem.is_end_state(problem.end_point)
   

2. • For a failure probability of 0, the problem is identical to the former state search problem. The difference here is that the optimal policy and value is obtained for every state, not only for a starting state. The X symbols denote the swamps. The letter between brackets represent the optimal policy. The value of 25 is from (0,1) is the same as the one obtained in the state model exercise.

   • For a failure probability of 1, you are certain of taking an other direction that the one you ask for. As a result, you can see that the optimal policy is going the opposite of the correct direction in order to hope to fail on the right direction. The cost is higher because when you have several failing direction you are not sure to fail on the correct one, and you have to pay the cost of the uncertainty.

   • The failure probability of 0.2 gives the same policy as the failure probability of 0, just at a higher cost.

   failure probability 0.0

   {‘E’, ‘S’} [S] -24.0	{‘E’, ‘W’, ‘S’} [W] -25.0	X {‘E’, ‘W’, ‘S’} [S] -41.0	X {‘W’, ‘S’} [S] -21.0
   {‘E’, ‘N’, ‘S’} [S] -23.0	X {‘E’, ‘W’, ‘N’, ‘S’} [S] -41.0	X {‘E’, ‘W’, ‘N’, ‘S’} [E] -21.0	{‘W’, ‘N’, ‘S’} [S] -1.0
   {‘E’, ‘N’} [E] -22.0	{‘E’, ‘W’, ‘N’} [E] -21.0	X {‘E’, ‘W’, ‘N’} [E] -20.0	END

   failure probability 1.0

   {‘E’, ‘S’} [E] -51.0	{‘E’, ‘W’, ‘S’} [S] -62.7	X {‘E’, ‘W’, ‘S’} [S] -72.3	X {‘W’, ‘S’} [W] -42.0
   {‘E’, ‘N’, ‘S’} [E] -50.0	X {‘E’, ‘W’, ‘N’, ‘S’} [E] -72.9	X {‘E’, ‘W’, ‘N’, ‘S’} [W] -65.8	{‘W’, ‘N’, ‘S’} [W] -22.0
   {‘E’, ‘N’} [N] -47.0	{‘E’, ‘W’, ‘N’} [N] -46.0	X {‘E’, ‘W’, ‘N’} [N] -43.0	END

   failure probability 0.2

   {‘E’, ‘S’} [S] -38.8	{‘E’, ‘W’, ‘S’} [W] -42.9	X {‘E’, ‘W’, ‘S’} [S] -56.7	X {‘W’, ‘S’} [S] -38.0
   {‘E’, ‘N’, ‘S’} [S] -36.5	X {‘E’, ‘W’, ‘N’, ‘S’} [S] -52.4	X {‘E’, ‘W’, ‘N’, ‘S’} [E] -35.8	{‘W’, ‘N’, ‘S’} [S] -8.4
   {‘E’, ‘N’} [E] -33.0	{‘E’, ‘W’, ‘N’} [E] -30.9	X {‘E’, ‘W’, ‘N’} [E] -26.7	END