One Coordinator, Many Specialists

# labs/agent_team.py
import queue
import re
import sqlite3
import threading
import time

from glados.knowledge import knowledge_context                    # chapter 85
from labs.command_loop import action_catalog, execute_action, extract_action

MEMORY_DB = "glados/data/memory.db"

def house_context(_: str) -> str:
    return action_catalog()                       # the action names the registry will accept

def recall_context(request: str) -> str:
    with sqlite3.connect(MEMORY_DB) as conn:
        return knowledge_context(conn, request)   # her own memory, searched by meaning

SPECIALISTS: dict[str, dict] = {
    "facts": {
        "keywords": {"what", "who", "when", "where", "how", "why", "explain", "define"},
        "context": None,
        "executes": False,
        "system": ("Answer one factual question in at most two sentences. If you do not "
                   "know, say so in three words. Never mention this house."),
    },
    "house": {
        "keywords": {"light", "lights", "lamp", "fan", "door", "lock", "camera",
                     "thermostat", "heater"},
        "context": house_context,
        "executes": True,
        "system": ("You control this house. Reply with exactly one action name from the "
                   "catalog above and nothing else. If none of them fit, reply NONE."),
    },
    "recall": {
        "keywords": {"remember", "recall", "yesterday", "earlier", "history", "said", "told"},
        "context": recall_context,
        "executes": False,
        "system": ("Answer from the retrieved lines above and nothing else. If they do not "
                   "contain the answer, reply exactly: NOT IN MEMORY."),
    },
}

WORD = re.compile(r"[a-z]+")

def score(request: str) -> dict[str, int]:
    words = set(WORD.findall(request.lower()))
    return {name: len(words & spec["keywords"]) for name, spec in SPECIALISTS.items()}

def pick(request: str, floor: int = 1) -> str:
    scores = score(request)
    best = max(scores, key=lambda name: scores[name])
    return best if scores[best] >= floor else "general"

REQUESTS = [
    "Turn off the kitchen lights.",
    "Show me the camera feed.",
    "What did I ask you to remember yesterday?",
    "Explain why the hallway lamp keeps turning itself off.",
    "Nice weather today.",
]

if __name__ == "__main__":
    print("  ".join(f"{name:>6}" for name in SPECIALISTS) + "   picked   request")
    for request in REQUESTS:
        scores = score(request)
        cells = "  ".join(f"{scores[name]:>6}" for name in SPECIALISTS)
        print(f"{cells}   {pick(request):<8} {request}")

$ uv run python -m labs.agent_team
 facts   house  recall   picked   request
     0       1       0   house    Turn off the kitchen lights.
     0       1       0   house    Show me the camera feed.
     1       0       2   recall   What did I ask you to remember yesterday?
     2       1       0   facts    Explain why the hallway lamp keeps turning itself off.
     0       0       0   general  Nice weather today.

Chapter 20's classifier returned the first intent whose keyword appeared anywhere in the sentence, which made dict order a routing policy nobody wrote down. Counting replaces it with a number per specialist, and numbers can be printed, diffed and asserted on in a test suite. Row three is the case first-match got wrong: "what" is a facts keyword and it is right there in the sentence, so first-match hands a memory question to the facts agent, while two points against one send it where it belongs.

The last row scores zero everywhere and lands on general, which is not a specialist at all but one ordinary call in her ordinary voice. The floor of 1 keeps pick total, the way chapter 20's registered fallback kept its router total: every sentence gets an answer. This fallback has the stronger claim, though, because that single plain call is the design the rest of the chapter has to beat.

Argue with row four. "Explain why the hallway lamp keeps turning itself off" scores facts 2, house 1, and the router is doing exactly what it was told, and the answer is probably wrong: the lamp is the subject and the question word is only the wrapper. Exercise 2 fixes it with weights. Keep the sentence in mind for the rule at the end, because a request that seems to want two specialists usually wants neither.

def score(request: str) -> dict[str, int]:
    lowered = request.lower()
    return {name: sum(1 for kw in spec["keywords"] if kw in lowered)   # BUG: substrings
            for name, spec in SPECIALISTS.items()}

$ uv run python -m labs.agent_team
 facts   house  recall   picked   request
     0       2       0   house    Turn off the kitchen lights.
     1       1       0   facts    Show me the camera feed.
     1       0       2   recall   What did I ask you to remember yesterday?
     2       1       0   facts    Explain why the hallway lamp keeps turning itself off.
     0       0       0   general  Nice weather today.

import json
import urllib.request

OLLAMA_URL = "http://127.0.0.1:11434/api/generate"
CHAT_MODEL = "llama3.2:3b"

def call_model(system: str, prompt: str, timeout: float = 120.0) -> str:
    payload = json.dumps({"model": CHAT_MODEL, "system": system,
                          "prompt": prompt, "stream": False}).encode("utf-8")
    request = urllib.request.Request(
        OLLAMA_URL, data=payload,
        headers={"Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(request, timeout=timeout) as reply:
        return json.loads(reply.read().decode("utf-8"))["response"].strip()

def run_specialist(name: str, prompt: str, out: queue.Queue) -> None:
    """Answer as one specialist. Always puts exactly one dict on the queue."""
    spec = SPECIALISTS[name]
    started = time.monotonic()
    try:
        context = spec["context"](prompt) if spec["context"] else ""
        text = call_model(spec["system"], f"{context}\n\nRequest: {prompt}".strip())
        out.put({"agent": name, "text": text, "error": None,
                 "seconds": time.monotonic() - started})
    except Exception as exc:                  # a thread that raises dies in silence
        out.put({"agent": name, "text": "", "error": f"{type(exc).__name__}: {exc}",
                 "seconds": time.monotonic() - started})

The system field is the entire difference between one specialist and another. Same weights, same server, same three billion parameters; different standing orders, and a different lookup pasted above the question. That is also why adding a fourth specialist costs nothing but a registry entry: you are not loading a second model, you are handing the same one a different job description.

The try is doing something a thread makes necessary. An exception raised in a worker never crosses back into the caller; the thread prints a message to stderr if anyone is watching and then it is simply gone, leaving the coordinator waiting out its entire budget for a result that stopped being possible seconds ago. Catching it and putting it on the queue turns a silent wait into a value with a name attached. Note also that the worker touches nothing except the queue it was handed, no shared dict, no global counter, which is what makes it safe to start three of these at once.

COORDINATOR_SYSTEM = (
    "You are GLaDOS. Below are reports from the specialists you asked. Rewrite them as "
    "one reply in your own voice, dry and brief. Add no fact the reports do not contain."
)
GENERAL_SYSTEM = "You are GLaDOS. Answer in one or two sentences, dry and unimpressed."

def ask_one(name: str, prompt: str, budget: float = 30.0) -> dict:
    out: queue.Queue = queue.Queue()
    worker = threading.Thread(target=run_specialist, args=(name, prompt, out), daemon=True)
    worker.start()
    worker.join(timeout=budget)
    try:
        return out.get_nowait()
    except queue.Empty:                   # the join returned on the clock, not on an answer
        return {"agent": name, "text": "", "seconds": budget,
                "error": f"no answer within {budget:.0f}s"}

def respond(request: str, budget: float = 30.0) -> str:
    name = pick(request)
    print(f"picked {name} ({', '.join(f'{k} {v}' for k, v in score(request).items())})")
    if name == "general":
        return call_model(GENERAL_SYSTEM, request)
    report = ask_one(name, request, budget)
    if report["error"]:
        return f"The {name} specialist gave me nothing: {report['error']}"
    if SPECIALISTS[name]["executes"]:                     # the tool set, enforced here
        action = extract_action(report["text"])           # chapter 35: a real name or None
        ran = execute_action(action, request) if action else False
        print(f"{name} chose {action!r}, ran {ran}")
    return call_model(COORDINATOR_SYSTEM,
                      f"Request: {request}\nReport from {name}: {report['text']}")

if __name__ == "__main__":
    import sys
    request = sys.argv[1] if len(sys.argv) > 1 else REQUESTS[0]
    budget = float(sys.argv[2]) if len(sys.argv) > 2 else 30.0
    started = time.monotonic()
    print(f"GLaDOS: {respond(request, budget)}")
    print(f"turn total {time.monotonic() - started:.1f} s")

$ uv run python -m labs.agent_team "Turn off the kitchen lights."   # her wording will differ
picked house (facts 0, house 1, recall 0)
house chose 'turn_off_lights', ran True
GLaDOS: The kitchen is dark. Try not to walk into anything expensive.
turn total 3.6 s

$ uv run python -m labs.agent_team "Turn off the kitchen lights." 2   # straight after a vision call
picked house (facts 0, house 1, recall 0)
GLaDOS: The house specialist gave me nothing: no answer within 2s
turn total 2.0 s

join(timeout=budget) bounds the wait on a service you do not run. Afterwards the join has no idea why it returned, so the queue is the only thing that can say: an item means the worker finished, and queue.Empty means the clock did. Chapter 48 caught that same exception for the opposite reason, where an empty queue meant no work had arrived; here it means no answer arrived, and the difference is entirely in what the caller does next.

The second run is chapter 86's eviction arriving as a coordinator problem. A vision call had just pushed the chat model out of the shared memory pool, so the specialist's first request paid the reload, and a two-second budget never had a chance. She says so in one line and returns. The worker thread is still out there and may well finish; it is a daemon, its queue goes out of scope with the frame that created it, and its answer is discarded on purpose. For a turn that has already been answered, late and never are the same thing.

One line carries the tool set. executes is a permission bit in the registry, and it is the only reason the house specialist's reply is allowed near execute_action while the other two are not. The reply reaches the registry as a name from the catalog, checked by chapter 35's extractor, so a specialist that invents an action gets a None and a False instead of an unhandled string.

def ask_all(tasks: list[tuple[str, str]], budget: float = 30.0) -> list[dict]:
    """Run every (specialist, prompt) task at once under one shared deadline."""
    out: queue.Queue = queue.Queue()
    workers = [threading.Thread(target=run_specialist, args=(name, prompt, out), daemon=True)
               for name, prompt in tasks]
    for worker in workers:
        worker.start()
    deadline = time.monotonic() + budget                  # one deadline, not one each
    for worker in workers:
        worker.join(timeout=max(0.0, deadline - time.monotonic()))
    reports = []
    while True:
        try:
            reports.append(out.get_nowait())
        except queue.Empty:
            break
    answered = {report["agent"] for report in reports}
    return reports + [{"agent": name, "text": "", "seconds": budget,
                       "error": f"no answer within {budget:.0f}s"}
                      for name, _ in tasks if name not in answered]

The single shared deadline is the point of the function. Give each worker its own 30-second join and three stalled specialists can hold the turn for 90 seconds; compute the deadline once before the joins and the whole fan-out is bounded by it however many workers you started. Any specialist that missed it still gets a record in the returned list, because a coordinator handed two reports where it expected three needs to know which one is absent, not just that something is.

Draining the queue with get_nowait in a loop, after every join has returned, is the other half. Reading one item per worker in start order would block on the slowest one while faster answers sat in the queue behind it. The queue has no opinion about which worker put what into it, so the reports arrive in the order they finished, and the agent name inside each dict is what tells you who is speaking.

# labs/fanout_bench.py
import time

from labs.agent_team import REQUESTS, SPECIALISTS, ask_all

def bench(request: str = REQUESTS[0]) -> None:
    alone = {name: ask_all([(name, request)])[0]["seconds"] for name in SPECIALISTS}
    started = time.monotonic()
    together = {r["agent"]: r["seconds"]
                for r in ask_all([(name, request) for name in SPECIALISTS])}
    wall = time.monotonic() - started
    print(f"{'specialist':<12}{'alone':>7}{'together':>11}")
    for name in SPECIALISTS:
        print(f"{name:<12}{alone[name]:7.1f}{together[name]:11.1f}")
    print(f"{'serial sum':<12}{sum(alone.values()):7.1f}")
    print(f"{'wall clock':<12}{'':7}{wall:11.1f}")

if __name__ == "__main__":
    bench()

$ uv run python -m labs.fanout_bench   # measured on the Jetson; your box will differ
specialist    alone   together
facts           2.4        4.3
house           1.6        3.1
recall          2.1        3.9
serial sum      6.1
wall clock                 4.4

Read the two columns against each other. Every specialist got slower in company, and the set of them finished in less time than they take one after another: 6.1 seconds of work returned in 4.4, a saving of 1.7. Not a third of the time, which is what three workers promises anyone who has not measured. The reason is what the GPU is doing while it decodes. Producing one token means reading the entire model out of memory, and it can read those weights once for three sequences about as cheaply as once for one, so three requests batched together cost barely more per step than one. Total throughput climbs, per-request latency climbs too, and the gap between them is the whole saving. One row explains itself, by the way: the recall specialist is two model calls on its own, since chapter 85's search embeds the question before a single token is generated, and it still lands between the other two.

That saving is the only thing making a team affordable on one box. Take it away and you feel it immediately: with OLLAMA_NUM_PARALLEL at 1 the server runs the requests strictly one at a time, the wall clock returns to the serial sum, and the threads have done nothing except hide the queue from you. Measure both settings before you believe any speedup, because this failure looks exactly like success from the calling code.

PLANNER_SYSTEM = (
    "Split the request into the fewest subtasks that different specialists must answer. "
    "One line each, formatted 'specialist: subtask'. Specialists: facts, house, recall. "
    "One line is a complete plan if one specialist can do the whole job."
)
PLAN_LINE = re.compile(r"^\s*(facts|house|recall)\s*:\s*(.+)$")

def plan(request: str) -> list[tuple[str, str]]:
    text = call_model(PLANNER_SYSTEM, request)
    return [(m.group(1), m.group(2).strip())
            for line in text.splitlines() if (m := PLAN_LINE.match(line))]

def team_answer(request: str, budget: float = 30.0) -> str:
    started = time.monotonic()
    tasks = plan(request)
    print(f"plan ({time.monotonic() - started:.1f} s):")
    for name, subtask in tasks:
        print(f"  {name}: {subtask}")
    reports = ask_all(tasks, budget)
    for report in sorted(reports, key=lambda r: r["seconds"]):
        print(f"{report['agent']:<7}({report['seconds']:4.1f}s): "
              f"{report['text'] or report['error']}")
    joined = "\n".join(f"{r['agent']}: {r['text'] or r['error']}" for r in reports)
    return call_model(COORDINATOR_SYSTEM, f"Request: {request}\n{joined}")

if __name__ == "__main__":                       # replaces the stage 3 block
    import sys
    args = [arg for arg in sys.argv[1:] if arg != "--team"]
    request = args[0] if args else REQUESTS[0]
    budget = float(args[1]) if len(args) > 1 else 30.0
    turn = team_answer if "--team" in sys.argv else respond
    started = time.monotonic()
    print(f"GLaDOS: {turn(request, budget)}")
    print(f"turn total {time.monotonic() - started:.1f} s")

$ uv run python -m labs.agent_team --team "Turn off the kitchen lights."   # the plan differs every run
plan (2.5 s):
  house: Turn off the lights in the kitchen.
  recall: Check when the kitchen lights were last switched.
  facts: Explain how a smart bulb receives an off command.
house  ( 3.1s): turn_off_lights
recall ( 3.9s): You last switched the kitchen lights on Tuesday evening.
facts  ( 4.3s): A smart bulb keeps a radio link to a hub and applies whatever state it is sent.
GLaDOS: The kitchen is dark, as requested. It went the same way on Tuesday evening, over a
        radio link, in case you were desperate to know.
turn total 9.2 s

Two failures in that run, neither of which raised anything. The request named one device and one verb, and the planner produced three subtasks, because a model asked for a decomposition returns a decomposition: nothing in that prompt makes "one line is enough" the easy answer, and the instruction saying so is a suggestion competing with the format of every example the model has ever seen. Nobody asked when the lights were last switched. Nobody asked how a bulb works.

The second failure is the Tuesday. Her memory database has no such entry; the semantic search came back empty and the recall specialist filled the gap, in a confident sentence, while holding an instruction that covers this exact case in its own system prompt. A prompt is a request, never an interlock. Then the coordinator did precisely what it was told, added no fact of its own, and rewrote a fabrication into her voice with her certainty behind it. That is the laundering step at the heart of the pattern: a specialist's guess goes in as a report and comes out as her claim, and nothing in the design reviews it on the way through.