State models

Some systems can be modeled by discrete states. Some actions might allow to change state. A cost is potentially associated with each action. A common problem is: what is the sequence of actions to go from a starting state to a target state at minimum cost? Even if mentioning this at this stage of the course might seem strange, note that the result of an action is deterministic. Both the next state and the cost to reach it are known deterministic quantities. This assumption will be dropped later in the course.

A state problem might be represented by a graph: every node is a state, every edge is associated with an action and a cost. That is one of many application of minimum path in a graph that was the point of a previous lecture about graph traversals.

Magic tram example

A classical example is to solve a Rubik’s cube: what is the fastest sequence of rotations to go from whatever state to a solved state? Let us start on a simpler example, the magic tramway example (inspired from the previously mentioned [LS]). Imagine a tramway and its stops. The stops are numbered with integers. You want to travel from a first stop (the starting point) to a second stop (the destination) only moving forward. You have the choice to either walk between two stops at a (time) cost of +1; or to take the tramway. However, the tramway is magic. For a cost of +2, it transports you from your current stop \(S\) to the \(2\times S\).

magic tram

Magic tram

This problem will be decomposed in two clearly distinct codes:

  1. the state problem itself: the actions, the outcome of the actions etc

  2. the solving algorithms: uniform cost search, dynamic programming etc

But first, create a brand new project “ai” where the state models, the reinforcement learning and the deep learning will be studied during the next few weeks.

Exercise (intermediate)

  1. Draw a graph representing the state problem between states 2 and 8.

  2. Implement and test the magic tram problem based on the following:

    src
     1from typing import Protocol
 2
 3# to be able to define the type State as either str int or tuple[int, ...]
 4# the notion of generic type is required
 5# a generic type is define with square brackets
 6# the convention is to used a single upper case letter (here S)
 7type State = str | int | tuple[int, ...]
 8type Action = str
 9type NextState = State
10type Cost = float
11type Outcome[S] = tuple[S, Cost]
12
13type StatePath[S] = list[S]
14type ActionPath = list[Action]
15
16
17class StateProblem[S: State](Protocol):
18    # whatever S is between str, int or tuple[int,...]
19    # this type is used consistently everywhere S appears
20    starting: S
21    destination: S
22
23    def actions_from_state(self, state: S) -> set[Action]: ...
24
25    def outcome_from_state_and_action(self, state: S, action: Action) -> Outcome: ...
26
27    def is_end_state(self, state: S) -> bool: ...

  3. Implement and test a solving algorithm based on the uniform cost search. Base your code on our previous version of UCS below and adapt it to state problems.

    In order to return also the actions to take, the output type should be changed to: tuple[Cost, StatePath, ActionPath]

    src
     1def uniform_cost_search_iterative_path_and_distance(
 2    graph: DirectedWeightedGraph, starting_node: Node, target_node: Node
 3) -> tuple[PathGraph, Weight]:
 4    visited = set()
 5    node = starting_node
 6    starting_distance: Weight = 0.0
 7    index: int = 0
 8    queue = [(starting_distance, index, [starting_node])]
 9    inf = float("inf")
10
11    while queue:
12        distance, _, nodes = heapq.heappop(queue)
13        node: Node = nodes[-1]
14        if node == target_node:
15            return nodes, distance
16        if node in visited:
17            continue
18
19        visited.add(node)
20        for child, child_distance in graph[node].items():
21            index += 1
22            total_distance: Weight = child_distance + distance
23            heapq.heappush(queue, (total_distance, index, nodes + [child]))
24
25    return [], inf

  4. Implement and test a dynamic programming solving algorithm. Base your code on our previous version of the shortest path in DAG below and adapt it to state problems.

    In order to return also the actions to take, the output type should be changed to: tuple[Cost, StatePath, ActionPath]

    src
     1def shortest_distance_and_path_DAG(
 2    graph: DirectedWeightedGraph, source: Node, target: Node
 3) -> tuple[Weight, PathGraph]:
 4    @cache
 5    def shortest_distance_and_path_to_target_from(
 6        current: Node,
 7    ) -> tuple[Weight, PathGraph]:
 8        """shortest distance and path from current to target"""
 9        if current == target:
10            return 0.0, [target]
11        # assuming that the shortest distance from child / neighbour
12        # to target is known (from recursive call),
13        # then the shortest distance from parent to target is
14        # the minimum of the distances of the paths through every child.
15        distances_and_paths: list[tuple[Weight, PathGraph]] = []
16        for neighbour, distance_to_neighbour in graph[current].items():
17            future_distance, future_path = shortest_distance_and_path_to_target_from(
18                neighbour
19            )
20            if future_path:
21                distances_and_paths.append(
22                    (distance_to_neighbour + future_distance, [current] + future_path)
23                )
24        if distances_and_paths:
25            return min(distances_and_paths)
26
27        return float("inf"), []
28
29    return shortest_distance_and_path_to_target_from(source)
Hint 3.

  • import the relevant state problem types

  • change the types in order to use state problem types

  • the starting and destination are now attributes of the state problem instance

  • the ‘child’ state are now obtained throught the outcomes of the available actions at a given state

Correction

  1. The actions and costs are omitted for clarity. Every straight arrow represents walking to the next state associated with a cost of +1. Every curved arrow represents taking the magic tram from \(S\) to \(2\times S\) with a cost of +2.

    tram graph

    Tramway graph

  2. src
     1@dataclass(frozen=True)
 2class MagicTramProblem(StateProblem[TramState]):
 3    # indicates that the State type is TramState which is an int
 4    """magic tram example from CS221.
 5    from starting point to destination only moving forward, walking has a cost and makes you move +1.
 6    Taking the tram has a higher cost, but makes you move (state x tram_factor)"""
 7
 8    starting: TramState
 9    destination: TramState
10    tram_factor: int = 2
11    tram_cost: float = 2.0
12    walk_cost: float = 1.0
13
14    def __post_init__(self):
15        if self.starting > self.destination:
16            raise ValueError(
17                f"starting {self.starting} > destination {self.destination}"
18            )
19        if self.starting <= 0:
20            raise ValueError(f"starting {self.starting} <= 0")
21
22    @override
23    def actions_from_state(self, state: TramState) -> set[Action]:
24        if state >= self.destination:
25            return set()
26        # must not risk to overshoot with tram
27        if state * self.tram_factor <= self.destination:
28            return {"walk", "tram"}
29        return {"walk"}
30
31    @override
32    def outcome_from_state_and_action(
33        self, state: TramState, action: Action
34    ) -> Outcome:
35        return (
36            (state + 1, self.walk_cost)
37            if action == "walk"
38            else (state * self.tram_factor, self.tram_cost)
39        )
40
41    @override
42    def is_end_state(self, state: TramState) -> bool:
43        return state == self.destination
44
45    def to_string(self, states: StatePath, actions: ActionPath) -> str:
46        total_cost: Cost = 0.0
47        table = []
48        for state, action in zip(states, actions):
49            if action == "tram":
50                table.append(
51                    [state, Fore.RED + action, f"{total_cost:.1f}" + Style.RESET_ALL]  # type: ignore
52                )
53                total_cost += self.tram_cost
54            else:
55                table.append([state, action, f"{total_cost:.1f}"])
56                total_cost += self.walk_cost
57        table.append([states[-1], "", f"{total_cost:.1f}"])
58        return tabulate(
59            table, headers=["state", "action", "cumulated cost"], tablefmt="grid"
60        )
    tests
     1import pytest
 2
 3from ai.state_models.state_example_tram import MagicTramProblem
 4
 5
 6@pytest.fixture
 7def problem():
 8    starting = 1
 9    return MagicTramProblem(starting=starting, destination=4 * starting)
10
11
12def test_destination_lower_than_starting():
13    with pytest.raises(ValueError, match="starting 2 > destination 1"):
14        MagicTramProblem(starting=2, destination=1)
15
16
17def test_destination_negative():
18    with pytest.raises(ValueError, match="starting -1 <= 0"):
19        MagicTramProblem(starting=-1, destination=1)
20
21
22def test_actions_from_state(problem):
23    assert problem.actions_from_state(problem.starting) == {"walk", "tram"}
24    assert problem.actions_from_state(problem.destination) == set()
25    assert problem.actions_from_state(problem.destination - 1) == {"walk"}
26
27
28def test_outcome_from_state_and_action(problem):
29    assert problem.outcome_from_state_and_action(problem.starting, "walk") == (
30        problem.starting + 1,
31        problem.walk_cost,
32    )
33    assert problem.outcome_from_state_and_action(problem.starting, "tram") == (
34        problem.starting * problem.tram_factor,
35        problem.tram_cost,
36    )
37
38
39def test_is_end_state(problem):
40    assert problem.is_end_state(problem.destination)

  3. Here, the UCS is implemented without a predecessor tree (which is less memory efficient). For the test, an obvious but not trivial example should be used. The magic tram from 2 to 13 meets this requirement.

    src
     1def ucs_states_and_actions_with_paths(
 2    problem: StateProblem,
 3) -> tuple[Cost, StatePath, ActionPath]:
 4    cost_from_start: Cost = 0.0
 5    states: StatePath = [problem.starting]
 6    actions: ActionPath = []
 7    visited: set[State] = set()
 8    index: int = 0
 9    inf = float("inf")
10
11    # index allows to break ties for equal total_cost using the push order
12    # instead of only the current state to schedule a visit,
13    # the entire state path is used, to get the state, just take the last
14    queue = [(cost_from_start, index, states, actions)]
15    while queue:
16        cost_from_start, _, states, actions = heapq.heappop(queue)
17        state = states[-1]
18        # do not use a direct comparison between destination and current state
19        # this is the job of is_end_state
20        if problem.is_end_state(state):
21            return cost_from_start, states, actions
22
23        if state in visited:
24            continue
25
26        visited.add(state)
27
28        # no longer loop directly on children, from the actions available
29        # get the outcome of these actions
30        for action in problem.actions_from_state(state):
31            index += 1
32            next_state, next_cost = problem.outcome_from_state_and_action(state, action)
33            total_cost: Cost = cost_from_start + next_cost
34            heapq.heappush(
35                queue, (total_cost, index, states + [next_state], actions + [action])
36            )
37
38    return inf, [], []
    tests
    1@pytest.fixture
2def tram_problem():
3    return MagicTramProblem(starting=2, destination=13)
    tests
    1def test_ucs_states_and_actions_tram(tram_problem):
2    cost, states, actions = ucs_states_and_actions_with_paths(tram_problem)
3    assert pytest.approx(cost) == 6.0
4    assert states == [2, 3, 6, 12, 13]
5    assert actions == ["walk", "tram", "tram", "walk"]
  4. src
     1def dynamic_programming_state_cost_states_actions(
 2    problem: StateProblem,
 3) -> tuple[Cost, StatePath, ActionPath]:
 4    @cache
 5    def minimum_cost_state_and_action_path_to_end_from(
 6        state: State,
 7    ) -> tuple[Cost, StatePath, ActionPath]:
 8        """minimum cost solution from current state to end state"""
 9        if problem.is_end_state(state):
10            return 0.0, [state], []
11
12        cost_states_actions = []
13        for action in problem.actions_from_state(state):
14            next_state, next_cost = problem.outcome_from_state_and_action(state, action)
15
16            future_cost, future_states, future_actions = (
17                minimum_cost_state_and_action_path_to_end_from(next_state)
18            )
19            if future_states:
20                cost_states_actions.append(
21                    (
22                        next_cost + future_cost,
23                        [state] + future_states,
24                        [action] + future_actions,
25                    )
26                )
27        if cost_states_actions:
28            return min(cost_states_actions)
29
30        return float("inf"), [], []
31
32    return minimum_cost_state_and_action_path_to_end_from(problem.starting)
    tests
    1def test_dynamic_programming_state_cost_states_actions(tram_problem):
2    cost, states, actions = dynamic_programming_state_cost_states_actions(tram_problem)
3    assert pytest.approx(cost) == 6.0
4    assert states == [2, 3, 6, 12, 13]
5    assert actions == ["walk", "tram", "tram", "walk"]

    For example from 2 to 101, an optimal path is:

    state

    action

    cumulated cost

    2

    walk

    0

    3

    tram

    1

    6

    tram

    3

    12

    tram

    5

    24

    walk

    7

    25

    tram

    8

    50

    tram

    10

    100

    walk

    12

    101

    13

Swamp island example

An other example of state problem search is to find an optimal path on a rectangular the isle. The goal is to reach the end point at minimum 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).

Exercise (intermediate)

  1. Draw a graph representing the state problem. Choose a case where the solution is obvious.

  2. Implement and test the swamp isle problem.

  3. Test the solving algorithm based on the uniform cost search.

  4. As implemented, the dynamic programming solving algorithm fail. Can you explain why?

Correction

  1. The thicker edges indicate a higher cost. The green nodes are swamps. Start from top left, end at bottom right:

    Swamp isle

    Swamp isle

  2. src
     1@dataclass(frozen=True)
 2class SwampIsleProblem(StateProblem[IsleState]):
 3    """the isle is rectangular, there are some dry land and swamps.
 4    The goal is to reach the end point at minimum cost.
 5    However, walking on dry land has a cost and walking in swamps has an even higher cost.
 6    One can walk either vertically or horizontally (not diagonally).
 7    """
 8
 9    nb_rows: int
10    nb_cols: int
11    starting: Coordinates
12    destination: Coordinates
13    swamp_spots: set[Coordinates] = field(
14        default_factory=set
15    )  # never use mutable directly
16    walking_cost: Cost = 1.0
17    swamp_cost: Cost = 20.0
18
19    @override
20    def actions_from_state(self, state: IsleState) -> set[Action]:
21        coordinates = state
22
23        if self.is_end_state(state):
24            return set()
25
26        LAST_COL = self.nb_cols - 1
27        LAST_ROW = self.nb_rows - 1
28        actions = {"N", "E", "S", "W"}
29
30        # can not cross the domain edge
31        if coordinates[0] == 0:
32            actions.remove("N")
33        if coordinates[0] == LAST_ROW:
34            actions.remove("S")
35        if coordinates[1] == 0:
36            actions.remove("W")
37        if coordinates[1] == LAST_COL:
38            actions.remove("E")
39
40        return actions
41
42    @override
43    def outcome_from_state_and_action(
44        self, state: IsleState, action: Action
45    ) -> Outcome:
46        coordinates = state
47        if coordinates in self.swamp_spots:
48            cost = self.swamp_cost
49        else:
50            cost = self.walking_cost
51
52        deltas: dict[Action, Coordinates] = {
53            "N": (-1, 0),
54            "E": (0, 1),
55            "S": (1, 0),
56            "W": (0, -1),
57        }
58        delta = deltas[action]
59        target: IsleState = (coordinates[0] + delta[0], coordinates[1] + delta[1])
60
61        return (target, cost)
62
63    @override
64    def is_end_state(self, state: IsleState) -> bool:
65        coordinates = state
66        return coordinates == self.destination
67
68    def to_string(self, states: StatePath, actions: ActionPath | None = None) -> str:
69        table = []
70        for row in range(self.nb_rows):
71            table.append([])
72            for col in range(self.nb_cols):
73                coordinates: Coordinates = (row, col)
74                content = ""
75                if coordinates == self.starting:
76                    content += Fore.BLUE + "start " + Style.RESET_ALL  # type: ignore
77                elif coordinates == self.destination:
78                    content += Fore.BLUE + "end " + Style.RESET_ALL  # type: ignore
79                if coordinates in self.swamp_spots:
80                    content += Back.GREEN + "*" + Style.RESET_ALL  # type: ignore
81                table[row].append(content)
82
83        if actions is not None:
84            actions.append("")
85            total_cost: Cost = 0.0
86            for state, action in zip(states, actions):
87                coordinates = state  # type: ignore
88                row, col = coordinates
89                table[row][col] += action
90                table[row][col] += Fore.RED + f" {total_cost:.1f}" + Style.RESET_ALL  # type: ignore
91                if coordinates in self.swamp_spots:
92                    total_cost += self.swamp_cost
93                else:
94                    total_cost += self.walking_cost
95
96        return tabulate(table, tablefmt="grid")
    tests
     1import pytest
 2
 3from ai.state_models.state_example_isle import SwampIsleProblem
 4
 5
 6@pytest.fixture
 7def problem():
 8    return SwampIsleProblem(
 9        3, 4, (0, 0), (2, 3), {(0, 2), (0, 3), (1, 1), (1, 2), (2, 2)}
10    )
11
12
13def test_actions_from_state(problem):
14    assert problem.actions_from_state((0, 0)) == {"S", "E"}
15    assert problem.actions_from_state((0, 1)) == {"S", "E", "W"}
16    assert problem.actions_from_state((1, 0)) == {"S", "E", "N"}
17    assert problem.actions_from_state((1, 1)) == {"N", "S", "E", "W"}
18    assert problem.actions_from_state((problem.nb_rows - 1, 0)) == {"N", "E"}
19    assert problem.actions_from_state((problem.nb_rows - 1, 1)) == {"N", "E", "W"}
20    assert problem.actions_from_state((0, problem.nb_cols - 1)) == {"S", "W"}
21    assert problem.actions_from_state((1, problem.nb_cols - 1)) == {"S", "W", "N"}
22    assert problem.actions_from_state(problem.destination) == set()
23
24
25def test_outcome_from_state_and_action(problem):
26    assert problem.outcome_from_state_and_action((0, 0), "S") == (
27        (1, 0),
28        problem.walking_cost,
29    )
30    assert problem.outcome_from_state_and_action((0, 2), "S") == (
31        (1, 2),
32        problem.swamp_cost,
33    )
34
35
36def test_is_end_state(problem):
37    assert problem.is_end_state(problem.destination)
  3. tests
    1def test_ucs_states_and_actions_isle():
2    problem = SwampIsleProblem(
3        3, 4, (0, 0), (2, 3), {(0, 2), (0, 3), (1, 1), (1, 2), (2, 2)}
4    )
5    cost, states, actions = ucs_states_and_actions_with_paths(problem)
6    assert pytest.approx(cost) == 24.0
7    assert states == [(0, 0), (1, 0), (2, 0), (2, 1), (2, 2), (2, 3)]
8    assert actions == ["S", "S", "E", "E", "E"]

    From start to e,d, the numbers represent the cumulated cost, the letters are the actions, the dots are the swamps:

    S 1.0

    start W 0.0

    S 2.0

    E 3.0

    E 4.0

    • E 5.0

    end 25.0

  4. As implemented, the dynamic programming algorithm falls into infinite recursion. Indeed, imagine from the starting position top left corner, the first recursive call is for (1,0) from action “S”, entering this call the next recursive call is (0,0) from action “N”. Here, the caching does not help. The previous states keep being recursively added.