Hierarchical Decision Processes

Similar to how complex programs are decomposed into functions, a complex Decision Process (DP) is often decomposed into multiple DPs. Since a Cognitive Agent (Cogent) has a single “root” DP, process decomposition thus naturally produces a hierarchy. And as with function decomposition, hierarchical DPs commonly produces multiple benefits, including…

  • DP re-use: a single decision process can be called upon in multiple contexts, possibly with re-parameterization (e.g., via supplied arguments)

  • State/logic simplification: each sub-process focuses solely on those state features that are necessary

    • As a bi-product, testing becomes simplified as well!

In the code below, an agent assists a Bingo caller in keeping track of drawn balls:

  1"""
  2Illustrates a simple hierarchical
  3decomposition of decision processes
  4in a Bingo caller assistant.
  5
  6Sub-problem of getting valid input
  7takes care of both letters and numbers,
  8while the play process handles de-
  9duplication and interactive game state.
 10
 11Integrated is a simple example of testing
 12the input sub-process in isolation.
 13"""
 14
 15from __future__ import annotations
 16
 17import contextlib
 18import sys
 19from collections.abc import Sequence
 20from contextlib import contextmanager
 21from dataclasses import dataclass, field
 22from io import StringIO
 23from random import shuffle
 24from string import ascii_uppercase
 25from typing import ClassVar, cast, override
 26
 27from cognition import Cogent, DecisionProcess, IOContainer, Operator
 28
 29# ===
 30
 31
 32@dataclass
 33class InputState:
 34    """
 35    Sub-problem of getting a valid value
 36    """
 37
 38    utterance: str | None = None
 39    """supplied input; None for nothing yet"""
 40
 41    validated: bool = False
 42    """has the utterance been checked yet"""
 43
 44    @staticmethod
 45    def done(state: InputState, _io: IOContainer) -> bool:
 46        """is the input problem complete?"""
 47
 48        return (state.utterance is not None) and state.validated
 49
 50
 51# ===
 52
 53dp_input = DecisionProcess(InputState).add_termination_check(InputState.done)
 54
 55
 56# START encapsulate
 57ARG_LETTER: str = "letter"
 58"""name of the argument to supply to the sub-problem"""
 59
 60
 61def get_valid_input(letter: str | None) -> str:
 62    """
 63    convenience method to provide the
 64    letter as the appropriate argument
 65    to the sub-dp and extract the result
 66    """
 67
 68    dp_input.reinit()
 69    result = dp_input(max_cycles=None, suppress_errors=True, **{ARG_LETTER: letter})
 70
 71    assert result is not None
 72
 73    return cast(str, result.utterance)
 74
 75
 76# END encapsulate
 77
 78# ====
 79
 80
 81def _get_letter(io: IOContainer) -> str | None:
 82    """arg getter"""
 83
 84    return cast(str | None, io.args[ARG_LETTER])
 85
 86
 87@dp_input.operator("get")
 88class GetInput(Operator[InputState]):
 89    """gets the utterance"""
 90
 91    @override
 92    def can_perform(self, state: InputState, _io: IOContainer) -> bool:
 93        return state.utterance is None
 94
 95    @override
 96    def perform(self, state: InputState, io: IOContainer) -> None:
 97        prompt = f"Please provide the { "letter" if _get_letter(io) is None else "number" }: "
 98        state.utterance = input(prompt)
 99
100
101@dp_input.operator("val_letter")
102class ValidateInputLetter(Operator[InputState]):
103    """validates the utterance as a bingo letter"""
104
105    @override
106    def can_perform(self, state: InputState, io: IOContainer) -> bool:
107        return (state.utterance is not None) and (_get_letter(io) is None)
108
109    @override
110    def perform(self, state: InputState, io: IOContainer) -> None:
111        if state.utterance in ("B", "I", "N", "G", "O"):
112            state.validated = True
113        else:
114            state.utterance = None
115
116
117@dp_input.operator("val_num")
118class ValidateInputNumber(Operator[InputState]):
119    """validates the utterance as a bingo number"""
120
121    # START valid_bounds
122    BOUNDS: ClassVar[dict[str, tuple[int, int]]] = {
123        "B": (1, 15),
124        "I": (16, 30),
125        "N": (31, 45),
126        "G": (46, 60),
127        "O": (61, 75),
128    }
129    # END valid_bounds
130
131    @override
132    def can_perform(self, state: InputState, io: IOContainer) -> bool:
133        return (state.utterance is not None) and (_get_letter(io) is not None)
134
135    @override
136    def perform(self, state: InputState, io: IOContainer) -> None:
137        letter = cast(str, _get_letter(io))
138
139        try:
140            number = int(cast(str, state.utterance))
141
142            lower, upper = self.BOUNDS[letter]
143            if lower <= number <= upper:
144                state.validated = True
145            else:
146                state.utterance = None
147
148        except ValueError:
149            state.utterance = None
150
151
152# ===
153
154# pylint: disable=invalid-name
155# START sub_problem_testing
156tests_passed: int = 0
157
158
159@contextmanager
160def _simulate_stdin(inputs: Sequence[str]):
161    """Context manager to mock sys.stdin with a sequence of strings."""
162
163    # Combine inputs with newlines to simulate pressing 'Enter'
164    mock_input = "\n".join(inputs) + "\n"
165
166    # Save the original stdin so we can restore it later
167    original_stdin = sys.stdin
168    try:
169        sys.stdin = StringIO(mock_input)
170        yield
171    finally:
172        sys.stdin = original_stdin
173
174
175def _test_input_dp(letter: str | None, expected: str, *inputs: str) -> None:
176    """
177    wrapper around assert to mock stdin/out,
178    then call get_valid_input with supplied args
179    """
180
181    with _simulate_stdin(inputs), contextlib.redirect_stdout(StringIO()):
182        actual = get_valid_input(letter)
183
184    assert (
185        actual == expected
186    ), f"letter={letter}, inputs={inputs}; actual ({actual}) != expected ({expected})"
187
188    global tests_passed  # pylint: disable=global-statement
189    tests_passed += 1
190
191
192def _input_dp_tests() -> None:
193    """
194    tests...
195    * bad letters
196    * good letters
197    * good + bad letter/number combos
198      (with significant input mistakes)
199    """
200
201    bad_letters = [l for l in ascii_uppercase if l not in ValidateInputNumber.BOUNDS]
202
203    for l, (low, high) in ValidateInputNumber.BOUNDS.items():
204        # easy good letter
205        _test_input_dp(None, l, l)
206
207        # variable known bad letters
208        shuffle(bad_letters)
209        _test_input_dp(None, l, *(bad_letters + [l]))
210        bad_letters.pop()
211
212        bad = [str(b) for b in range(101) if not low <= b <= high]
213
214        for n in range(low, high + 1):
215            # easy good letter + number
216            _test_input_dp(l, str(n), str(n))
217
218            # given letter; variable known bad numbers
219            shuffle(bad)
220            _test_input_dp(l, str(n), *(bad + [str(n)]))
221            bad.pop()
222
223
224_input_dp_tests()
225# END sub_problem_testing
226
227# ===
228
229
230@dataclass
231class PlayState:
232    """
233    Game of Bingo
234    """
235
236    called: set[str] = field(default_factory=set)
237    """prior spots called"""
238
239    next_call: str | None = None
240    """next spot; or None for pending"""
241
242    done: bool = False
243    """has someone won?"""
244
245    @staticmethod
246    def game_over(state: PlayState, _io: IOContainer) -> bool:
247        """func for detecting end of game play"""
248
249        return state.done
250
251
252# ===
253
254dp_play = DecisionProcess(PlayState).add_termination_check(PlayState.game_over)
255
256
257@dp_play.operator("draw")
258class PlayDraw(Operator[PlayState]):
259    """draw the next spot"""
260
261    @override
262    def can_perform(self, state: PlayState, _io: IOContainer) -> bool:
263        return state.next_call is None
264
265    @override
266    def perform(self, state: PlayState, _io: IOContainer) -> None:
267        while True:
268            letter = get_valid_input(None)
269            number = get_valid_input(letter)
270
271            candidate = f"{letter}{number}"
272            if candidate not in state.called:
273                break
274
275            print(f"Already called: {candidate}")
276
277        state.next_call = candidate
278
279
280@dp_play.operator("announce")
281class PlayAnnounce(Operator[PlayState]):
282    """announce the draw"""
283
284    @override
285    def can_perform(self, state: PlayState, _io: IOContainer) -> bool:
286        return state.next_call is not None
287
288    @override
289    def perform(self, state: PlayState, _io: IOContainer) -> None:
290
291        # excuse the nested operator logic
292        # for a shorter example
293        result = input(f"Draw {state.next_call} (bingo?): ")
294        if result.strip().lower() == "bingo":
295            print(f"Prior: {state.called}")
296
297            confirm = input("Confirmed (y/n)? ")
298            if confirm.strip().lower() == "y":
299                state.done = True
300
301        state.called.add(cast(str, state.next_call))
302        state.next_call = None
303
304
305# ===
306
307if __name__ == "__main__":
308    print(f"Tests passed: {tests_passed}")
309    print()
310
311    Cogent(dp_play, name="bingo")()  # let's play!

Root DP: Ball Draw <-> Game-Win Verification

The cogent primarily operates within a logical loop:

  1. Call the next drawn ball

    1. Verified as a valid letter-number combination

    2. Confirmed as not having been called before

  2. If “Bingo!” is claimed, providing prior draws for confirmation

    1. If confirmed, done

    2. Otherwise, back to #1

This is encoded as the dp_play DP.

Tip

In practice, decision process decomposition invites decomposition into separate files/folders (e.g., folder for the DP, with files for state, operators, support functions, etc.).

Sub-Process: Validate Call (1.1)

Nested within the root DP is the need to verify a valid letter and associated number:

    BOUNDS: ClassVar[dict[str, tuple[int, int]]] = {
        "B": (1, 15),
        "I": (16, 30),
        "N": (31, 45),
        "G": (46, 60),
        "O": (61, 75),
    }

Errors include…

  • An invalid letter

  • Given a letter, an invalid number

  • A previously called combination (signaling full re-entry needed for a typo that produced a valid combination)

This sub-process is encapsulated within a function…

ARG_LETTER: str = "letter"
"""name of the argument to supply to the sub-problem"""


def get_valid_input(letter: str | None) -> str:
    """
    convenience method to provide the
    letter as the appropriate argument
    to the sub-dp and extract the result
    """

    dp_input.reinit()
    result = dp_input(max_cycles=None, suppress_errors=True, **{ARG_LETTER: letter})

    assert result is not None

    return cast(str, result.utterance)

The letter argument parameterizes the dp_input DP’s operators:

  • The GetInput operator’s prompt (“letter” vs “number”)

  • The choice of validation operator (via can_perform): ValidateInputLetter vs ValidateInputNumber

This function is called within the perform method of the root DP’s PlayDraw operator.

Sub-Process Testing

Since the sub-process is designed to be self-contained, requiring only a single argument, it is straightforward to test:

tests_passed: int = 0


@contextmanager
def _simulate_stdin(inputs: Sequence[str]):
    """Context manager to mock sys.stdin with a sequence of strings."""

    # Combine inputs with newlines to simulate pressing 'Enter'
    mock_input = "\n".join(inputs) + "\n"

    # Save the original stdin so we can restore it later
    original_stdin = sys.stdin
    try:
        sys.stdin = StringIO(mock_input)
        yield
    finally:
        sys.stdin = original_stdin


def _test_input_dp(letter: str | None, expected: str, *inputs: str) -> None:
    """
    wrapper around assert to mock stdin/out,
    then call get_valid_input with supplied args
    """

    with _simulate_stdin(inputs), contextlib.redirect_stdout(StringIO()):
        actual = get_valid_input(letter)

    assert (
        actual == expected
    ), f"letter={letter}, inputs={inputs}; actual ({actual}) != expected ({expected})"

    global tests_passed  # pylint: disable=global-statement
    tests_passed += 1


def _input_dp_tests() -> None:
    """
    tests...
    * bad letters
    * good letters
    * good + bad letter/number combos
      (with significant input mistakes)
    """

    bad_letters = [l for l in ascii_uppercase if l not in ValidateInputNumber.BOUNDS]

    for l, (low, high) in ValidateInputNumber.BOUNDS.items():
        # easy good letter
        _test_input_dp(None, l, l)

        # variable known bad letters
        shuffle(bad_letters)
        _test_input_dp(None, l, *(bad_letters + [l]))
        bad_letters.pop()

        bad = [str(b) for b in range(101) if not low <= b <= high]

        for n in range(low, high + 1):
            # easy good letter + number
            _test_input_dp(l, str(n), str(n))

            # given letter; variable known bad numbers
            shuffle(bad)
            _test_input_dp(l, str(n), *(bad + [str(n)]))
            bad.pop()


_input_dp_tests()

Note

This very well could have been done within a standard testing framework (e.g., unittest, pytest) – the code here is illustrative of the underlying IO mocking and self-contained.