Architecture

The interesting constraint in this project is that the roster is not known when the code is written. There is no class per agent and no branch per destination. The graph is assembled at load time from whatever the config declares, which means the same module compiles a five-agent travel network or a five-agent consulting firm without knowing the difference.

Shared state, and why it needs a reducer

Every node reads and writes one state object. Most of its fields are ordinary, written by one node at a time. agent_outputs is not: the specialists run in parallel and all write to it at once, so LangGraph needs to be told how to combine their writes.

def _merge(left: dict | None, right: dict | None) -> dict:
    """Reducer letting parallel agent nodes write into one dict without clobbering."""
    return {**(left or {}), **(right or {})}

class AgentState(TypedDict, total=False):
    input: str
    scenario_id: int
    broker_plan: list[str]
    agent_outputs: Annotated[dict[str, str], _merge]
    final_response: str
    requires_approval: bool
    approval_decision: str

The Annotated wrapper attaches the reducer to that one field. Without it, parallel writes to the same key are a conflict, and with a naive reducer the specialists would overwrite each other and the broker would synthesise from whichever happened to finish last.

Nodes are generated, not written

Each agent in the config becomes a node through a closure that captures its own definition. The node itself does almost nothing: it looks up the canned reply for the current scenario and returns it under the agent's name.

def specialist(agent_id: str):
    agent = config.agents[agent_id]

    def node(state: AgentState) -> dict:
        reply = agent.reply_for(state["scenario_id"])
        channel.status(f"Consulting {agent.name} ...")
        channel.log(f"{agent.name} returned: {reply}")
        return {"agent_outputs": {agent.name: reply}}

    return node

The edges are wired the same way, in a loop over the roster. The conditional edge out of the broker has to declare every possible destination up front, which is the one place the roster has to be enumerated.

graph.set_entry_point("broker")
graph.add_conditional_edges(
    "broker", route, list(config.agents) + ["human_approval", END]
)
for agent_id in config.agents:
    graph.add_edge(agent_id, "broker")
# The gate is terminal. Going back to the broker would re-synthesise.
graph.add_edge("human_approval", END)

Routing, and admitting when it fails

The broker asks the model which specialists to consult and expects a JSON list of agent ids back. Models do not always oblige, so the parser treats three failures as distinct: output that is not JSON, JSON that is not a list, and names that are not in the roster.

def _parse_plan(raw: str, config: NetworkConfig, channel: Channel) -> list[str]:
    """Read the router's JSON list of agent ids, and say so when it is malformed.

    Falls back to a single agent rather than raising, but logs every reason it
    did: unparseable output, a non-list, or names that are not in the roster.
    An unlogged fallback is indistinguishable from a real routing decision.
    """
    cleaned = raw.replace("```json", "").replace("```", "").strip()
    fallback = [next(iter(config.agents))]
    try:
        plan = json.loads(cleaned)
    except json.JSONDecodeError as exc:
        channel.log(f"router returned unparseable JSON ({exc}); falling back. Raw: {raw!r}")
        return fallback

    if not isinstance(plan, list):
        channel.log(f"router returned {type(plan).__name__}, expected a list; falling back.")
        return fallback

    known = [a for a in plan if a in config.agents]
    if unknown := [a for a in plan if a not in config.agents]:
        channel.log(f"router named agents that do not exist, ignoring: {unknown}")
    if not known:
        channel.log("router named no valid agents; falling back.")
        return fallback
    return known

All three fall back to consulting one agent rather than raising, because a demo that stops on a bad parse is less useful than one that degrades. Each is logged on the way past. The third case matters most: a model that invents a plausible agent name produces a run that looks entirely successful, and without the log there is nothing to distinguish it from a real routing decision.

Scenario selection

Which canned reply an agent returns depends on the scenario, matched by keyword against the question. It is the least clever part of the system and the place where imprecision is hardest to notice.

def scenario_for(self, prompt: str) -> int:
    """Pick a scenario by keyword.

    Keywords match on word boundaries with an optional plural, so "crib"
    matches "cribs" while "ai" matches "AI" and "AI-driven" but not "retail",
    "explain" or "email". Substring matching would catch all three and route
    them to the wrong scenario without raising anything. First match in
    document order wins, so order the scenarios from most to least specific.
    """
    text = prompt.lower()
    for sc in self.scenarios:
        for kw in sc.keywords:
            if re.search(rf"\b{re.escape(kw)}(?:s|es)?\b", text):
                return sc.id
    return self.default_scenario

Matching is on word boundaries with an optional plural, so crib catches "cribs" while ai catches "AI" and "AI-driven" but not retail, explain or email. The precision is worth the regular expression, because a mis-selected scenario does not raise anything: the agents return confident, well-formed answers about the wrong situation. First match in document order wins, so scenarios are ordered from most specific to least.

The graph does not know where its output goes

The nodes never print and never call input(). They talk to a channel, handed in when the network is built, and that is the only route anything takes out of the graph or back into it.

class Channel:
    """Interface between the running graph and whoever is watching it."""

    def status(self, message: str) -> None:
        """One-line progress, meant to be overwritten by the next one."""

    def log(self, message: str) -> None:
        """Detail for the trace log."""

    def chat(self, message: str) -> None:
        """A message addressed to the user."""

    def ask_approval(self, proposal: str) -> bool:
        """Block until a human approves or rejects. Return True to proceed."""
        raise NotImplementedError

Three implementations. ConsoleChannel prints and can answer the approval gate automatically, which is what lets the notebook run unattended. QueueChannel pushes messages onto a queue the desktop UI polls, and blocks the worker thread on an Event until a person answers. A third, used by the tests, records everything and never prints.

That indirection is what makes the tests possible. The suite runs the real graph with a stub model and a recording channel: no API key, no network, no cost.

The human gate

The synthesis prompt asks the model to score its own recommendation, and the config supplies the rule. In the travel network, changing prices or publishing content needs a person. In the consulting network it is layoffs, restructuring, or budget moves over $1M. The broker sets requires_approval when the answer contains RISK: YES, and the router sends it to the gate.

The gate is terminal. It has one outgoing edge, to END, and it hands back exactly the text it showed the human.

def human_approval(state: AgentState) -> dict:
    proposal = state["final_response"]
    channel.status("Waiting for human approval ...")
    channel.log(f"proposal put to a human:\n{proposal}")

    if channel.ask_approval(proposal):
        channel.log("human approved")
        # Deliver exactly the text that was approved. Routing back through the
        # broker would re-run synthesis and hand the user a different answer
        # from the one they authorised.
        final = proposal + "\n\n[STATUS: APPROVED BY HUMAN, EXECUTED]"
        channel.chat(final)
        channel.status("Ready")
        return {"approval_decision": "APPROVED", "final_response": final}

    channel.log("human rejected")
    final = "ACTION CANCELLED. The human operator rejected the proposal.\n\nWhat was proposed:\n" + proposal
    channel.chat(final)
    channel.status("Ready")
    return {"approval_decision": "REJECTED", "final_response": final}

That the gate does not route back into the broker is the whole design, and it is worth being explicit about why. The broker's first action is to check whether agent outputs exist. After a fan-out they do, so any path that re-enters the broker runs synthesis a second time and produces a fresh answer. The person would then have authorised one piece of text and received another, generated after they clicked yes, potentially carrying a different risk verdict.

Nothing about that looks wrong from outside: the system stops, a person approves, and a plausible answer appears. An approval gate that regenerates its answer is not an approval gate, so the test suite asserts that the proposal text survives verbatim into the final output, and that synthesis runs exactly once.

<agent id="market_scout" name="Market Scout">
    <description>events, weather</description>
    <mock_data default="No significant market alerts.">
        <response scenario_id="1">{"event": "Tech Summit", "status": "CANCELLED", "impact": "negative demand"}</response>
        <response scenario_id="2">{"alert": "City Construction", "location": "Main St", "duration": "2 weeks"}</response>
        <response scenario_id="3">{"event": "Taylor Swift Eras Tour", "status": "NEW_ANNOUNCEMENT", "impact": "massive demand spike"}</response>
        <response scenario_id="4">{"demographic": "Families", "flight_capacity": "+20%", "origin": "Domestic"}</response>
        <response scenario_id="5">{"market_status": "stable", "competitor_activity": "aggressive discounting"}</response>
    </mock_data>
</agent>

def reply_for(self, scenario_id: int) -> str:
    """The canned reply this agent gives for a scenario.

    Every reply is written into the config in advance. No agent calls a real
    API, so no live business data passes through this network.
    """
    return self.replies.get(scenario_id, self.default_reply)