Practice · Agent Orchestration

Keep a supervisor in control

Write the three parts that make a supervisor agent safe to leave running, hand-offs, a turn budget and an approval gate, and check your work against six tests. A scripted model means no API key.

Open on GitHub

Python 3.10+No API key

Run it

From a clone of stackunseen/examples (on Windows, type python where it says python3):

bash

git clone https://github.com/stackunseen/examples
cd examples
cd practice-supervisor-control
python3 -m unittest -v

What you build

A supervisor asks the model what to do next, hands the work to a specialist and feeds the result back until the model answers. Getting that loop to run is the easy part. You write the three things that make it safe to leave running:

Hand-offs that come back. Call the specialist the model names and give the model what it returned. A name that is not a specialist goes back to the model as a message, not a crash.
A turn budget. Stop after max_turns, however keen the model is to carry on.
An approval gate. An action that changes something outside, here sending an email, runs only when approve() says yes.

The model is scripted and the specialists are stubs, so it runs offline in a second. The supervisor is yours.

Checking your work

Six checks in test_supervisor.py say what the supervisor must do, one behaviour each. The first passes before you start; the others fail with the TODO they need. A missing turn budget fails its check after 50 turns instead of running for ever. The code tour below gives one hint per TODO when you want one.

When you are done

solution.py in the same folder has one way to do it, and PRACTICE_SOLUTION=1 python3 -m unittest -v runs the same checks against it. Then open Multi-agent orchestration, which builds the same supervisor with the OpenAI Agents SDK: find where its turn limit is set and where an approval step would go.

The code

Code tour · 4 stops

python

1"""Practice: keep a supervisor in control.
2
3A supervisor asks the model what to do next, hands the work to a specialist and feeds
4the result back, until the model answers. Getting that loop to run is the easy part.
5The three things that make it safe to leave running are yours to write:
6
7  TODO 1  Hand the work to the specialist the model names; give the model what came back.
8  TODO 2  Stop at the turn budget, however keen the model is to carry on.
9  TODO 3  Run an action that changes something outside only when approve() says yes.
10
11Record each step in the trace as you go; it is what you read when a run goes wrong.
12
13  What happened                         Trace entry
14  a specialist was called               ("call", name)
15  the model named no such specialist    ("unknown", name)
16  an approved action ran                ("ran", name)
17  an action was refused                 ("denied", name)
18  the model answered                    ("answer", "")
19  the turn budget ran out               ("stopped", "budget")
20
21Check your work:  python -m unittest -v
22The model is scripted (fake_model.py), so this runs offline with no API key.
23solution.py has one way to do it.
24"""
25
26from __future__ import annotations
27
28from dataclasses import dataclass, field
29from typing import Callable, Literal, Protocol
30
31
32@dataclass
33class Step:
34    """What the model wants next: call a specialist, take an action, or answer."""
35
36    kind: Literal["call", "act", "answer"]
37    name: str = ""  # the specialist or the action
38    text: str = ""  # what to pass on, or the answer
39
40
41class Model(Protocol):
42    def next_step(self, history: list[str]) -> Step: ...
43
44
45@dataclass
46class Run:
47    status: Literal["answered", "budget_exhausted"]
48    answer: str | None
49    turns: int
50    trace: list[tuple[str, str]] = field(default_factory=list)
51
52
53def run_supervisor(
54    task: str,
55    model: Model,
56    specialists: dict[str, Callable[[str], str]],
57    actions: dict[str, Callable[[str], str]],
58    approve: Callable[[Step], bool],
59    max_turns: int = 6,
60) -> Run:
61    """Work on the task until the model answers or the turn budget runs out.
62
63    One turn is one call to the model.
64    """
65    history = [f"Task: {task}"]
66    trace: list[tuple[str, str]] = []
67    turns = 0
68    while True:
69        # TODO 2: once the model has had max_turns turns, stop. Record ("stopped", "budget")
70        #         and return a Run with status "budget_exhausted" and no answer.
71
72        step = model.next_step(history)
73        turns += 1
74
75        if step.kind == "answer":
76            trace.append(("answer", ""))
77            return Run("answered", step.text, turns, trace)
78
79        if step.kind == "call":
80            # TODO 1: run the specialist called step.name on step.text, record ("call", name)
81            #         and add what it returned to history, so the model sees it next turn.
82            #         A name that is not a specialist must not crash the run: record
83            #         ("unknown", name) and say so in history, so the model can choose again.
84            raise NotImplementedError("TODO 1: hand the work to the specialist")
85
86        if step.kind == "act":
87            # TODO 3: ask approve(step) before anything runs. If it says yes, run the action
88            #         and record ("ran", name). If it says no, run nothing, record
89            #         ("denied", name) and tell the model in history.
90            raise NotImplementedError("TODO 3: hold the action for approval")

Go further

Full example on GitHubgithub.com/stackunseen/examples/tree/main/practice-supervisor-control On the mapAI Application → Agent