# @save
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any, Callable
@dataclass(frozen=True)
class ToolSpec:
"""One tool the harness is willing to execute on the model's behalf.
The schema maps argument names to Python types; it is the tool's public
contract, advertised to the model as data and enforced by the harness
before the handler runs. The handler is ordinary code with real authority,
which is why the effectful flag marks tools that change the world instead
of reading it.
"""
description: str
schema: dict[str, type]
handler: Callable[..., Any]
effectful: bool = False16 Anatomy of an Agent: The Loop, Tools, and Workflow Patterns
A customer writes, “Order A-17 arrived damaged. Please make this right.” A language model can compose a sympathetic reply. Give it a lookup tool and it can fetch the order. Give it a refund tool and it can propose a remedy. None of those additions answers the engineering question that matters: what stands between the proposal and the movement of money? Suppose the system prompt says “never refund more than 50 dollars without approval” but the Python handler accepts any integer. The model emits amount_cents=999999, the caller invokes the handler, and the payment provider accepts it. The prompt expressed an intention; the executable path granted the authority. The real limit was never 50 dollars — it was whatever the handler and the payment account permitted.
We are now ready to build the thing that closes that gap: a complete agent in bare Python, assembled in small executed steps. The parts are (i) a tool registry with declared, machine-checkable contracts; (ii) the tool-call wire format a chat model actually emits, and a parser that survives it being malformed; (iii) the loop itself — propose, gate, execute, observe — whose transcript we print in full; (iv) a policy gate that turns an over-limit refund into a denial the model can read and recover from; (v) budgets and typed endings that make every run finish; (vi) the workflow canon — chain, router, parallel, evaluator, agent — implemented as control policies over the same kernel and compared with measured numbers; (vii) a ReAct trace and a plan-then-execute worked example; and (viii) verification that checks the environment rather than believing the answer. The whole artifact is about 150 lines, built from the raw message format of Chapter 12 with no framework and no SDK, because every framework we meet in Chapter 19 is a wrapper around exactly this loop.
One honesty note before we start. The book executes offline and deterministically, so the “model” in these pages is a scripted stub: a small object that replays a visible script of assistant messages, speaking the same wire format a hosted model speaks. You will see its script in full every time we use it — nothing is generated, and we never pretend otherwise. Everything around the stub is the real mechanism: the schemas, the parse, the gate, the budgets, and every measured number are exactly what a live model would plug into, and one eval: false cell shows the genuine network call for when you want to swap the stub out.
16.1 Tools and their contracts
An agent’s tools are ordinary functions. What makes them tools is that they carry a declared contract — a schema of argument names and types — that is advertised to the model as data and enforced by our code before anything runs. We start with the contract type.
Our running environment is a tiny support desk: one order table, a policy document, and three tools. The important design choice is the effect log — an append-only list that records every refund actually issued. Reads return data; only refund_order appends a receipt. For the rest of the chapter, every safety claim we make will be an assertion on this list, never on anything the model says.
# @save
def make_environment() -> tuple[dict[str, ToolSpec], list[dict[str, Any]]]:
"""Create the support-desk world: three tools and an append-only effect log.
The effect log is the ground truth of this small world. Reads return data;
only ``refund_order`` appends a receipt. Every safety claim we make later
is an assertion on this list, never on anything the model says.
Returns:
The tool registry and the (initially empty) effect log.
"""
orders = {"A-17": {"order_id": "A-17", "status": "damaged", "paid_cents": 4999}}
effects: list[dict[str, Any]] = []
def lookup_order(order_id: str) -> dict[str, Any]:
if order_id not in orders:
raise KeyError(f"no such order: {order_id}")
return orders[order_id]
def read_policy() -> str:
return "Damaged orders: refund the amount paid, up to 5000 cents automatically."
def refund_order(order_id: str, amount_cents: int) -> dict[str, Any]:
receipt = {"order_id": order_id, "refunded_cents": amount_cents}
effects.append(receipt)
return receipt
tools = {
"lookup_order": ToolSpec(
"Fetch an order's status and the amount paid.", {"order_id": str}, lookup_order
),
"read_policy": ToolSpec("Read the refund policy.", {}, read_policy),
"refund_order": ToolSpec(
"Issue a refund. Moves real money.",
{"order_id": str, "amount_cents": int},
refund_order,
effectful=True,
),
}
return tools, effectsWatch what the raw handlers do when we call them directly, with no model anywhere in sight.
tools, effects = make_environment()
print(tools["lookup_order"].handler(order_id="A-17"))
print(tools["refund_order"].handler(order_id="A-17", amount_cents=999999))
print("effect log:", effects){'order_id': 'A-17', 'status': 'damaged', 'paid_cents': 4999}
{'order_id': 'A-17', 'refunded_cents': 999999}
effect log: [{'order_id': 'A-17', 'refunded_cents': 999999}]The second line is the chapter’s whole problem in one receipt: the handler cheerfully refunded 999,999 cents — two hundred times what the customer paid — because nothing in Python stops it. The authority lives in the handler, and today the handler grants it to any caller. Before we let a model near this function, we need two layers of judgment. The first is validation: is this call well formed? A proposal names a tool and carries arguments, and validation checks it against the declared contract.
# @save
@dataclass(frozen=True)
class ToolCall:
"""A parsed proposal to run one tool. It carries no authority by itself.
``arguments`` is ``None`` when the model's argument string failed to parse
as JSON — a proposal can be broken before it can be judged.
"""
call_id: str
name: str
arguments: dict[str, Any] | None
def validate_call(call: ToolCall, tools: dict[str, ToolSpec]) -> str | None:
"""Check one proposal against the registry's declared contracts.
Validation answers "is this call well formed?" — the tool exists, the
argument names match exactly, and every value has the declared type. It
deliberately says nothing about whether the call is *permitted*; that is
the gate's question, asked later.
Args:
call: The parsed proposal.
tools: The registry of declared contracts.
Returns:
None when the proposal is well formed, otherwise a reason string the
model can read and act on.
"""
if call.arguments is None:
return "arguments were not valid JSON"
tool = tools.get(call.name)
if tool is None:
return f"unknown tool: {call.name}"
if set(call.arguments) != set(tool.schema):
return f"{call.name} expects exactly the fields {sorted(tool.schema)}"
for arg_name, expected in tool.schema.items():
if not isinstance(call.arguments[arg_name], expected):
return f"{arg_name} must be {expected.__name__}"
return Nonegood = ToolCall("c1", "refund_order", {"order_id": "A-17", "amount_cents": 4999})
bad_type = ToolCall("c2", "refund_order", {"order_id": "A-17", "amount_cents": "lots"})
bad_name = ToolCall("c3", "delete_database", {})
for call in (good, bad_type, bad_name):
print(f"{call.name:16} -> {validate_call(call, tools)}")refund_order -> None
refund_order -> amount_cents must be int
delete_database -> unknown tool: delete_databaseNote what validation did not catch: the 999,999-cent refund is perfectly well formed — right tool, right fields, right types. Well-formed and permissible are different questions, and we keep them in different functions on purpose. Our exact name -> type schema is a teaching miniature; production tools use JSON Schema with ranges, enumerations, and nested objects, but the division of labor is identical. The second question — may this exact call run now? — waits for the gate in Section 16.4.
The last thing tools need is a way to reach the model. They travel as data: one JSON declaration per tool, carried in the request. The model never receives the handlers — only these descriptions — so everything it “knows” about a tool is what we choose to advertise.
# @save
def tool_schema_json(tools: dict[str, ToolSpec]) -> list[dict[str, Any]]:
"""Render the registry in the JSON wire shape a chat API expects.
This is how tools reach the model: as data in the request, one JSON
schema per tool. The model never receives the handlers — only these
descriptions — so everything it "knows" about a tool is what we choose
to advertise here.
Args:
tools: The registry to advertise.
Returns:
A list of function-tool declarations in OpenAI-compatible form.
"""
type_names = {str: "string", int: "integer", float: "number", bool: "boolean"}
return [
{
"type": "function",
"function": {
"name": name,
"description": spec.description,
"parameters": {
"type": "object",
"properties": {
arg: {"type": type_names[kind]} for arg, kind in spec.schema.items()
},
"required": sorted(spec.schema),
},
},
}
for name, spec in tools.items()
]print(json.dumps(tool_schema_json(tools)[2], indent=2)){
"type": "function",
"function": {
"name": "refund_order",
"description": "Issue a refund. Moves real money.",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string"
},
"amount_cents": {
"type": "integer"
}
},
"required": [
"amount_cents",
"order_id"
]
}
}
}That block of JSON is the entire interface between the model and the refund handler. Writing tool descriptions that models use well is a craft of its own — Chapter 17 spends a whole chapter on it.
16.2 The model side of the wire
What does a model actually send back when it wants to use a tool? Not a function call — text, structured as an assistant message that carries a list of proposed tool calls. Two helpers build the two message shapes we need, and their exact structure is the point, so read them as data formats rather than code.
# @save
def tool_call_message(
call_id: str, name: str, arguments: dict[str, Any], thought: str | None = None
) -> dict[str, Any]:
"""Build an assistant message that proposes one tool call, wire-shaped.
The arguments travel as a JSON *string*, exactly as chat APIs deliver
them: the model emitted text that happens to look like JSON, and the
harness must parse it — and must survive the parse failing.
Args:
call_id: Correlation id the tool result must echo back.
name: The tool being proposed.
arguments: The arguments, serialized to the wire string.
thought: Optional visible reasoning text carried in ``content``.
Returns:
An assistant message dict in OpenAI-compatible wire shape.
"""
return {
"role": "assistant",
"content": thought,
"tool_calls": [
{
"id": call_id,
"type": "function",
"function": {"name": name, "arguments": json.dumps(arguments)},
}
],
}
def answer_message(text: str) -> dict[str, Any]:
"""Build a plain assistant message that answers and proposes nothing."""
return {"role": "assistant", "content": text}print(json.dumps(tool_call_message("call_1", "lookup_order", {"order_id": "A-17"}), indent=2)){
"role": "assistant",
"content": null,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "lookup_order",
"arguments": "{\"order_id\": \"A-17\"}"
}
}
]
}Three details deserve a hard look. The arguments value is a string containing JSON, not a JSON object — the model generated characters, and someone has to parse them. The id is a correlation handle: when we send the tool’s result back, we must echo this id so the model (and our own verifier) can match result to proposal; position in a list is not a stable identity once calls can be parallel or retried. And content can coexist with tool_calls — that slot is where visible reasoning text rides along, which is how we will print a ReAct trace in Section 16.7.
Now the stand-in model. It holds a script of assistant turns and plays them in order, with one twist that earns its keep for the rest of the chapter: before consuming the next turn, it checks the most recent tool result against a small table of reactions, so a script can respond to what the loop feeds back — retry after a denial, for example — while staying fully deterministic.
# @save
@dataclass
class ScriptedModel:
"""A stand-in for a chat model whose entire behavior is a visible script.
``chat`` plays the scripted turns in order. Before consuming the next
turn it checks the most recent tool result against ``reactions``: if a
key occurs in that result, the paired message is returned instead. That
one rule lets a script react to what the loop feeds back — retry after
a denial, for example — while staying deterministic and inspectable.
Nothing here is a language model; the point is that the loop cannot tell
the difference, because both speak the same wire format.
"""
turns: list[dict[str, Any]]
reactions: dict[str, dict[str, Any]] = field(default_factory=dict)
def chat(self, messages: list[dict[str, Any]]) -> dict[str, Any]:
"""Return the next assistant message, given the conversation so far.
Args:
messages: The transcript; only the latest tool message is
consulted, and only to match ``reactions`` keys.
Returns:
The next scripted assistant message, a reaction, or a default
answer once the script is exhausted.
"""
last = messages[-1]
if last["role"] == "tool":
for needle, reply in self.reactions.items():
if needle in last["content"]:
return reply
if not self.turns:
return answer_message("I have nothing further to add.")
return self.turns.pop(0)model = ScriptedModel(
turns=[
tool_call_message("call_1", "lookup_order", {"order_id": "A-17"}),
answer_message("Order A-17 arrived damaged; the customer paid 4999 cents."),
]
)
history = [{"role": "user", "content": "What happened with order A-17?"}]
first = model.chat(history)
print("turn 1:", json.dumps(first))
history += [first, {"role": "tool", "tool_call_id": "call_1",
"content": json.dumps({"status": "damaged", "paid_cents": 4999})}]
print("turn 2:", json.dumps(model.chat(history)))turn 1: {"role": "assistant", "content": null, "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "lookup_order", "arguments": "{\"order_id\": \"A-17\"}"}}]}
turn 2: {"role": "assistant", "content": "Order A-17 arrived damaged; the customer paid 4999 cents."}The interface is one method: messages in, assistant message out. A hosted model has exactly the same interface, which is why swapping the stub for the real thing touches nothing else in the chapter. Here is that swap, shown but not executed — it needs a network and a served model — followed by a representative response so you can see the wire shape survives contact with reality.
import httpx
payload = {
"model": "an-8b-instruct",
"messages": [
{"role": "system", "content": "You are a support agent. Use the tools."},
{"role": "user", "content": "Order A-17 arrived damaged. Please make this right."},
],
"tools": tool_schema_json(tools),
"temperature": 0, "seed": 7,
}
resp = httpx.post("http://localhost:8000/v1/chat/completions", json=payload).json()
print(json.dumps(resp["choices"][0]["message"], indent=2))Representative response (an 8B-class instruct model behind an OpenAI-compatible endpoint):
{
"role": "assistant",
"content": null,
"tool_calls": [
{
"id": "chatcmpl-tool-9f2a41",
"type": "function",
"function": {
"name": "lookup_order",
"arguments": "{\"order_id\": \"A-17\"}"
}
}
]
}Tool-call envelopes are vendor-specific and drift. The OpenAI-compatible shape above — tool_calls with arguments as a JSON string — is the most widely copied, but Anthropic’s Messages API carries tool_use content blocks whose input arrives already parsed, and both vendors have revised these envelopes several times since 2023. The mechanism this chapter teaches (proposals as data, parse, gate, correlated results) is stable; the field names are not. Verify live: your provider’s tool-use API reference for the current envelope (checked 2026-07-20).
Because the arguments are model-generated text, parsing them is a step that can fail, and a loop that crashes on bad JSON hands the model a denial-of-service button. Our parser therefore never raises: a call whose arguments do not parse still comes back as a ToolCall, with arguments=None, and validate_call already knows what to say about it.
# @save
def parse_tool_calls(message: dict[str, Any]) -> list[ToolCall]:
"""Parse the tool calls, if any, out of an assistant message.
The arguments field is model-generated text, so it can be broken JSON;
a call whose arguments fail to parse still comes back as a ``ToolCall``,
with ``arguments=None``, so the loop can reject it through the normal
validation path instead of crashing.
Args:
message: An assistant message in wire shape.
Returns:
One ``ToolCall`` per entry in ``tool_calls``; empty for a plain answer.
"""
calls: list[ToolCall] = []
for item in message.get("tool_calls") or []:
function = item["function"]
try:
arguments = json.loads(function["arguments"])
except json.JSONDecodeError:
arguments = None
calls.append(ToolCall(item["id"], function["name"], arguments))
return callsprint(parse_tool_calls(tool_call_message("call_7", "refund_order",
{"order_id": "A-17", "amount_cents": 4999})))
broken = {"role": "assistant", "content": None,
"tool_calls": [{"id": "call_8", "type": "function",
"function": {"name": "refund_order",
"arguments": '{"order_id": A-17}'}}]}
print(parse_tool_calls(broken))
print(validate_call(parse_tool_calls(broken)[0], tools))[ToolCall(call_id='call_7', name='refund_order', arguments={'order_id': 'A-17', 'amount_cents': 4999})]
[ToolCall(call_id='call_8', name='refund_order', arguments=None)]
arguments were not valid JSONA parse failure is just another way a proposal can be invalid, so it flows down the same rejection path as a wrong type or an unknown tool. One code path for every malformed input is easier to test than special cases, and the model gets a readable reason either way.
16.3 The loop: propose, gate, execute, observe
We have tools with contracts and a model that proposes calls in wire format. The loop that connects them has four verbs, and keeping them separate matters more than any framework choice. Propose: the model returns an assistant message. Gate: our code decides whether each proposed call may run. Execute: only an allowed call reaches a handler. Observe: the result — success, denial, or failure — travels back to the model as an ordinary tool message. First, the typed endings and budgets the loop will enforce.
# @save
from enum import Enum
class Stop(str, Enum):
"""The typed ways a run can end. Callers branch on these, not on prose."""
ANSWERED = "answered"
STEP_LIMIT = "step_limit"
TOKEN_LIMIT = "token_limit"
NO_PROGRESS = "no_progress"
@dataclass(frozen=True)
class Limits:
"""Resource ceilings the loop enforces regardless of what the model wants.
``max_turns`` bounds model calls; ``max_prompt_tokens`` bounds the
estimated context we are willing to send on any single call; and
``repeated_denials`` bounds how often the identical proposal may be
denied before the loop declares no progress.
"""
max_turns: int = 8
max_prompt_tokens: int = 4000
repeated_denials: int = 2Next, the observation type and a cost estimator. An observation is the typed result of attempting one call, correlated to the proposal that caused it; the kind field distinguishes a denial from a validation failure from a handler error, because a strategy that only sees “failed” can do nothing smarter than retry blindly. The token estimator is deliberately crude — four characters per token — but it is monotone in the thing that matters: the transcript, and with it the price of every further turn, only ever grows.
# @save
@dataclass(frozen=True)
class Observation:
"""The typed result of attempting one tool call.
``kind`` is one of ``result`` (the handler ran), ``invalid`` (validation
failed), ``denied`` (the gate refused), or ``error`` (the handler raised
an expected exception). A strategy can branch on the kind; "failed" alone
invites blind retries.
"""
call_id: str
ok: bool
kind: str
content: Any
def estimate_tokens(messages: list[dict[str, Any]]) -> int:
"""Estimate the token cost of a message list at four characters per token.
Crude, but monotone in the thing that matters: the transcript — and with
it the price of every further turn — only ever grows.
Args:
messages: Wire-format messages about to be sent.
Returns:
The estimated prompt token count.
"""
return sum(len(json.dumps(message)) for message in messages) // 4Now the function that carries one proposal across the effect boundary — or refuses to. A gate is any function from a validated call to an allow-or-deny decision with a reason. We start with the gate that every system has before anyone writes policy: the one that allows everything.
# @save
Gate = Callable[[ToolCall], tuple[bool, str]]
def allow_all(call: ToolCall) -> tuple[bool, str]:
"""A gate that approves everything — the absence of policy, made explicit."""
return True, "no policy configured"
def execute_call(call: ToolCall, tools: dict[str, ToolSpec], gate: Gate) -> Observation:
"""Carry one proposal across the effect boundary — or refuse to.
Validation asks whether the call is well formed; the gate asks whether
this exact call may run now. Only when both pass does the handler
execute. Expected handler failures come back as typed observations
instead of ending the run, and a denial is information, not an exception.
Args:
call: The parsed proposal.
tools: The registry of executable contracts.
gate: The policy consulted immediately before execution.
Returns:
A typed, correlated ``Observation``.
"""
error = validate_call(call, tools)
if error is not None:
return Observation(call.call_id, False, "invalid", error)
allowed, reason = gate(call)
if not allowed:
return Observation(call.call_id, False, "denied", reason)
try:
result = tools[call.name].handler(**call.arguments)
except (KeyError, ValueError, TimeoutError) as exc:
return Observation(call.call_id, False, "error", str(exc))
return Observation(call.call_id, True, "result", result)Three probes — a good call, a missing order, an unknown tool — show the three failure textures a strategy can tell apart.
tools, effects = make_environment()
print(execute_call(ToolCall("c1", "lookup_order", {"order_id": "A-17"}), tools, allow_all))
print(execute_call(ToolCall("c2", "lookup_order", {"order_id": "Z-99"}), tools, allow_all))
print(execute_call(ToolCall("c3", "delete_database", {}), tools, allow_all))Observation(call_id='c1', ok=True, kind='result', content={'order_id': 'A-17', 'status': 'damaged', 'paid_cents': 4999})
Observation(call_id='c2', ok=False, kind='error', content="'no such order: Z-99'")
Observation(call_id='c3', ok=False, kind='invalid', content='unknown tool: delete_database')The missing order came back as an error observation rather than a crashed run, because a KeyError from a lookup is an expected, informative failure. The catch list is deliberately short: an assertion failure or a corrupted registry should still crash loudly, because a loop that converts every exception into a polite observation invites the model to improvise around broken safety code. Figure 16.1 traces the path one proposal takes.
sequenceDiagram
autonumber
participant M as Model
participant L as Loop
participant G as Validate + gate
participant T as Tool handler
participant E as Environment
M->>L: assistant message with tool_calls
L->>G: ToolCall(id, name, arguments)
alt malformed or denied
G-->>L: reason
L-->>M: tool message (kind=invalid / denied)
else allowed
G-->>L: allow
L->>T: handler(**arguments)
T->>E: read or effect
E-->>T: result or expected error
T-->>L: typed result
L-->>M: tool message (kind=result / error)
end
The loop’s state is worth a pause before we write the loop. It is nothing but plain data — the wire-format transcript, the typed observations, and two counters. That plainness is a feature with a name: if you persist messages, you can resume the run on another machine, replay it into an evaluator, or hand it to a human, because there is no hidden state anywhere else.
# @save
@dataclass
class RunState:
"""Everything the loop knows, as plain serializable data.
``messages`` is the transcript in wire format: persist it and the run can
resume anywhere, because there is no hidden state elsewhere. The typed
observations and per-turn prompt sizes exist so a verifier can replay
what happened without re-running anything.
"""
messages: list[dict[str, Any]]
observations: list[Observation] = field(default_factory=list)
prompt_token_log: list[int] = field(default_factory=list)
turns: int = 0
@dataclass(frozen=True)
class RunResult:
"""A typed ending, the final answer if any, and the full state behind it."""
stop: Stop
answer: str | None
state: RunStateHere is the loop entire — no elisions, because this is the raw mechanism every agent framework wraps. The model chooses what to propose; this function owns everything else: the transcript, the budgets, the effect boundary, and the decision to stop.
# @save
SYSTEM_PROMPT = (
"You are a support agent. Use the tools to resolve the customer's problem. "
"A denied call is information: read the reason and adjust."
)
def run_agent(
task: str,
model: Any,
tools: dict[str, ToolSpec],
gate: Gate,
limits: Limits = Limits(),
) -> RunResult:
"""Run one task through the propose-gate-execute-observe loop to a typed end.
Each turn sends the whole transcript to the model, appends whatever it
returns, executes any proposed calls through ``execute_call``, and appends
the typed results as tool messages. The model chooses what to propose;
this function owns everything else — the transcript, the budgets, the
effect boundary, and the decision to stop.
Args:
task: The user's goal. Text only; it grants no authority.
model: Anything with ``chat(messages) -> assistant message``.
tools: The registry of executable contracts.
gate: Policy consulted immediately before every handler call.
limits: Step, token, and no-progress ceilings.
Returns:
A ``RunResult`` with the stop reason, the answer text when the model
answered, and the complete state for verification.
"""
state = RunState(messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": task},
])
denials: dict[str, int] = {}
while state.turns < limits.max_turns:
prompt_tokens = estimate_tokens(state.messages)
if prompt_tokens > limits.max_prompt_tokens:
return RunResult(Stop.TOKEN_LIMIT, None, state)
state.prompt_token_log.append(prompt_tokens)
reply = model.chat(state.messages)
state.messages.append(reply)
state.turns += 1
calls = parse_tool_calls(reply)
if not calls:
return RunResult(Stop.ANSWERED, reply.get("content"), state)
for call in calls:
observation = execute_call(call, tools, gate)
state.observations.append(observation)
state.messages.append({
"role": "tool",
"tool_call_id": call.call_id,
"content": json.dumps({
"ok": observation.ok,
"kind": observation.kind,
"content": observation.content,
}),
})
if observation.kind == "denied":
key = f"{call.name}:{json.dumps(call.arguments, sort_keys=True)}"
denials[key] = denials.get(key, 0) + 1
if denials[key] >= limits.repeated_denials:
return RunResult(Stop.NO_PROGRESS, None, state)
return RunResult(Stop.STEP_LIMIT, None, state)It is worth reading this function slowly, because you will recognize it inside every agent product you ever open up. To watch it run, we want the transcript in a readable form, so a small printer first.
# @save
def print_transcript(state: RunState) -> None:
"""Print a run's transcript, one aligned line per message or tool call.
Args:
state: The finished (or in-flight) run state to display.
"""
labels = {"system": "system |", "user": "user |", "tool": "tool |"}
for message in state.messages:
if message["role"] == "assistant":
if message.get("content"):
print(f"model | {message['content']}")
for item in message.get("tool_calls") or []:
function = item["function"]
print(f"model | -> {function['name']} {function['arguments']}")
else:
print(f"{labels[message['role']]} {message['content']}")Now the first complete run: a well-behaved script that looks up the order, refunds what was paid, and answers.
task = "Order A-17 arrived damaged. Please make this right."
tools, effects = make_environment()
scripted = ScriptedModel(turns=[
tool_call_message("call_1", "lookup_order", {"order_id": "A-17"}),
tool_call_message("call_2", "refund_order", {"order_id": "A-17", "amount_cents": 4999}),
answer_message("Order A-17 arrived damaged, so I refunded the 4999 cents you paid."),
])
result = run_agent(task, scripted, tools, allow_all)
print_transcript(result.state)
print(f"\nstop={result.stop.value} turns={result.state.turns} effects={effects}")system | You are a support agent. Use the tools to resolve the customer's problem. A denied call is information: read the reason and adjust.
user | Order A-17 arrived damaged. Please make this right.
model | -> lookup_order {"order_id": "A-17"}
tool | {"ok": true, "kind": "result", "content": {"order_id": "A-17", "status": "damaged", "paid_cents": 4999}}
model | -> refund_order {"order_id": "A-17", "amount_cents": 4999}
tool | {"ok": true, "kind": "result", "content": {"order_id": "A-17", "refunded_cents": 4999}}
model | Order A-17 arrived damaged, so I refunded the 4999 cents you paid.
stop=answered turns=3 effects=[{'order_id': 'A-17', 'refunded_cents': 4999}]Three turns, two tool calls, one receipt, a typed answered ending — and every line of it sitting in result.state.messages as plain data. This is a good moment to name the separation the transcript makes visible, because the words model, harness, and environment will carry weight for the rest of the book. The model proposes: it maps a transcript to a next message and has no other power. The harness is everything we just wrote — prompts, tool registry, gate, budgets, the loop — and it owns every decision. The environment is the world the tools touch: the order table and the effect log, which record what actually happened regardless of what anyone said. Figure 16.2 locates the three.
flowchart LR
Model(["Model<br/>proposes next message"])
subgraph Harness["Harness — this chapter's ~150 lines"]
Loop["run_agent loop<br/>budgets + transcript"]
Gate["gate<br/>policy on exact arguments"]
Reg["tool registry<br/>declared contracts"]
end
Env[("Environment<br/>orders, money, effect log")]
Model -->|assistant message| Loop
Loop -->|tool message| Model
Loop --> Gate --> Reg -->|effect| Env
Env -->|observation| Loop
16.4 The effect boundary: a denial that must not happen
The happy path ran with allow_all, and that should make you nervous. Let us make the danger concrete with a model script that over-reaches: it looks up the order, proposes a 9,999-cent refund — twice what the customer paid — and, if it ever sees a denial, falls back to refunding exactly the paid amount. The fallback lives in the script’s reaction table, so this stub reacts to what the loop feeds back the way a capable instructed model would.
# @save
def overreaching_model() -> ScriptedModel:
"""Build the chapter's over-reaching support model, fresh for each run.
Its script looks up the order, proposes a 9999-cent refund — twice the
amount paid — and, if it ever sees a denial, falls back to refunding
exactly what the order cost. A capable model behaves this way when its
instructions say to read denials; here the reaction is scripted so every
run of the book reproduces it.
Returns:
A fresh ``ScriptedModel`` with the same three turns and one reaction.
"""
return ScriptedModel(
turns=[
tool_call_message("call_1", "lookup_order", {"order_id": "A-17"}),
tool_call_message("call_2", "refund_order",
{"order_id": "A-17", "amount_cents": 9999}),
answer_message("I have issued the refund for order A-17."),
],
reactions={
"denied": tool_call_message("call_3", "refund_order",
{"order_id": "A-17", "amount_cents": 4999}),
},
)First, the world as it is today: the same loop, no policy.
tools, effects_ungated = make_environment()
result_ungated = run_agent(task, overreaching_model(), tools, allow_all)
print(f"stop={result_ungated.stop.value} turns={result_ungated.state.turns}")
print("effect log:", effects_ungated)stop=answered turns=3
effect log: [{'order_id': 'A-17', 'refunded_cents': 9999}]The 9,999-cent refund landed. Note that the system prompt was present the whole time — it says to use the tools responsibly — and it changed nothing, because a prompt is an input to a distribution, not a constraint on execution. The fix is not a sterner prompt. It is a few lines of policy evaluated against the exact arguments immediately before execution.
# @save
AUTO_REFUND_LIMIT_CENTS = 5000
def refund_gate(call: ToolCall) -> tuple[bool, str]:
"""The support desk's policy, applied at the effect boundary.
It examines the exact arguments about to run — not what the prompt said,
not what the model intended — and refuses any refund above the automatic
authority limit. Everything else passes.
Args:
call: The validated proposal about to be executed.
Returns:
``(allowed, reason)``; the reason travels back to the model either way.
"""
if call.name == "refund_order" and call.arguments["amount_cents"] > AUTO_REFUND_LIMIT_CENTS:
return False, f"refund exceeds automatic authority of {AUTO_REFUND_LIMIT_CENTS} cents"
return True, "within policy"Same task, same over-reaching model, one changed argument: the gate.
tools, effects_gated = make_environment()
result_gated = run_agent(task, overreaching_model(), tools, refund_gate)
print_transcript(result_gated.state)
print(f"\nstop={result_gated.stop.value} turns={result_gated.state.turns}")
print("effect log:", effects_gated)system | You are a support agent. Use the tools to resolve the customer's problem. A denied call is information: read the reason and adjust.
user | Order A-17 arrived damaged. Please make this right.
model | -> lookup_order {"order_id": "A-17"}
tool | {"ok": true, "kind": "result", "content": {"order_id": "A-17", "status": "damaged", "paid_cents": 4999}}
model | -> refund_order {"order_id": "A-17", "amount_cents": 9999}
tool | {"ok": false, "kind": "denied", "content": "refund exceeds automatic authority of 5000 cents"}
model | -> refund_order {"order_id": "A-17", "amount_cents": 4999}
tool | {"ok": true, "kind": "result", "content": {"order_id": "A-17", "refunded_cents": 4999}}
model | I have issued the refund for order A-17.
stop=answered turns=4
effect log: [{'order_id': 'A-17', 'refunded_cents': 4999}]Read the middle of that transcript carefully — it is the most important trace in the chapter. The 9,999-cent proposal was denied, and the denial did not crash anything: it became an ordinary tool message carrying a readable reason. The model’s next turn (the scripted reaction) used that information to propose 4,999 instead, which passed. Adaptation happened without weakening the gate: the model stayed free to propose any value, and our code stayed free to refuse it. The effect log — the only witness that cannot be talked out of its story — shows exactly one receipt at 4,999 cents. Figure 16.3 puts the two runs side by side.
Show the code that draws this figure
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(5.8, 3.2))
amounts = [effects_ungated[0]["refunded_cents"], effects_gated[0]["refunded_cents"]]
bars = ax.bar(["allow_all gate", "refund_gate"], amounts, width=0.5,
color=["#b13f3f", "#2a7f9e"])
ax.axhline(AUTO_REFUND_LIMIT_CENTS, ls="--", color="0.4")
ax.text(1.32, AUTO_REFUND_LIMIT_CENTS + 120, "policy limit (5000)", fontsize=8,
ha="right", color="0.3")
for bar, amount in zip(bars, amounts):
ax.text(bar.get_x() + bar.get_width() / 2, amount + 120, str(amount),
ha="center", fontsize=9)
ax.set_ylabel("cents actually refunded")
fig.tight_layout()
plt.show()
Honesty about scope: refund_gate proves one value-limit property and nothing broader. It does not check that the caller owns A-17, that the currency matches, that this order was not already refunded, or that 4,999 cents is the right remedy — a production authorization layer owns those, and Chapter 24 builds it. What the small gate does establish is the architectural point: within this process, a denied call cannot reach a handler through run_agent, no matter what the model writes. When we want to test that claim, we assert on effects_gated, never on the model’s apology — the model saying “I was denied” tests obedience, while the empty log tests containment.
NoteIn the interview
Q. Your agent’s system prompt says “never refund more than $50.” A red-teamer gets the model to emit amount_cents=999999. What happens, and what should? What happens depends on one thing only: whether code inspects the arguments at the execution boundary. A prompt is an input to a probability distribution — it shifts behavior, it enforces nothing. The answer they want: policy lives in a gate that evaluates the exact arguments immediately before the handler runs, and the test asserts on the effect log (no receipt appears), not on the model’s reply. The trap is describing better prompting, or testing that the model says it refused.
16.5 Termination, budgets, and typed endings
A gate can contain every action while the loop still fails operationally: a model can wander across harmless reads forever, or repeat a denied proposal until the caller’s budget is gone. “The model will eventually stop” is not a termination condition — our while loop needs its own. The Limits we wrote in Section 16.3 give it three, and we can trigger each one on demand. First the wanderer, a script that reacts to every lookup result by looking up again.
tools, _ = make_environment()
lookup_again = tool_call_message("call_w", "lookup_order", {"order_id": "A-17"})
wanderer = ScriptedModel(turns=[lookup_again], reactions={"order_id": lookup_again})
result_w = run_agent("Investigate order A-17.", wanderer, tools, refund_gate,
Limits(max_turns=5))
print(f"stop={result_w.stop.value} turns={result_w.state.turns} "
f"observations={len(result_w.state.observations)}")stop=step_limit turns=5 observations=5Each iteration consumes one turn, so the loop must return within max_turns decisions — a liveness guarantee that holds even for a model that never chooses to stop. Next, the stubborn model: it proposes the over-limit refund and reacts to every denial by proposing it again, unchanged.
tools, effects_stub = make_environment()
insist = tool_call_message("call_s", "refund_order",
{"order_id": "A-17", "amount_cents": 9999})
stubborn = ScriptedModel(turns=[insist], reactions={"denied": insist})
result_s = run_agent("Refund order A-17.", stubborn, tools, refund_gate)
print(f"stop={result_s.stop.value} turns={result_s.state.turns}")
print("denied observations:",
sum(1 for o in result_s.state.observations if o.kind == "denied"))
print("effect log:", effects_stub)stop=no_progress turns=2
denied observations: 2
effect log: []The loop fingerprints each denied proposal by tool name and canonical arguments; the second identical denial trips repeated_denials and the run ends no_progress after two turns instead of burning six more. The detector is deliberately narrow — a model alternating between 9,998 and 9,999 would evade it, and Exercise 3 asks you to close that gap without punishing a legitimate retry. Third, the budget that bites in production: tokens. Every turn resends the whole transcript, so cost per turn grows even when the model does nothing new.
tools, _ = make_environment()
wanderer = ScriptedModel(turns=[lookup_again], reactions={"order_id": lookup_again})
result_t = run_agent("Investigate order A-17.", wanderer, tools, refund_gate,
Limits(max_turns=40, max_prompt_tokens=1200))
print(f"stop={result_t.stop.value} turns={result_t.state.turns}")
print("prompt tokens per turn:", result_t.state.prompt_token_log[:6], "...",
result_t.state.prompt_token_log[-2:])stop=token_limit turns=14
prompt tokens per turn: [54, 142, 229, 317, 404, 492] ... [1104, 1192]Fourteen turns in, the next prompt would exceed 1,200 estimated tokens and the loop refuses to admit another model call. The per-turn log climbs from 54 to 1,192 in nearly even steps of about 88 tokens — the price of appending one proposal and one observation — and Figure 16.4 plots that climb against the four-turn denial run, whose transcript stayed small because it finished.
Show the code that draws this figure
fig, ax = plt.subplots(figsize=(6.2, 3.2))
ax.plot(range(1, len(result_t.state.prompt_token_log) + 1),
result_t.state.prompt_token_log, marker="o", label="wanderer (no progress)")
ax.plot(range(1, len(result_gated.state.prompt_token_log) + 1),
result_gated.state.prompt_token_log, marker="s", label="denial run (finished)")
ax.axhline(1200, ls="--", color="0.4")
ax.text(1.1, 1240, "max_prompt_tokens = 1200", fontsize=8, color="0.3")
ax.set_xlabel("model call number")
ax.set_ylabel("estimated prompt tokens")
ax.legend(loc="center right", fontsize=8)
fig.tight_layout()
plt.show()
The four endings we have now produced are not interchangeable failures, and that is why Stop is an enum rather than a string. answered can proceed to outcome verification; no_progress should surface the last denial to a human; step_limit can return partial findings; token_limit should stop admitting paid calls before the next one, not after. A caller who receives “agent failed” has lost the information needed to choose among those responses. Figure 16.5 draws the full machine — every path from Ready reaches an absorbing state, which is the property the budgets buy.
stateDiagram-v2
[*] --> Ready
Ready --> Proposing: budget admits another call
Proposing --> Answered: plain assistant message
Proposing --> Executing: gate allows
Proposing --> Denied: invalid or denied
Executing --> Ready: typed observation appended
Denied --> Ready: below repeat threshold
Denied --> NoProgress: identical proposal denied again
Ready --> StepLimit: max_turns reached
Ready --> TokenLimit: next prompt exceeds budget
Answered --> [*]
NoProgress --> [*]
StepLimit --> [*]
TokenLimit --> [*]
One production caveat belongs here in plain text: checking budgets between calls prevents admitting another call after a limit; it does not preempt a model or tool already in flight, and it says nothing about an effect that may have completed remotely when a timeout fired. Durable, exactly-once effects are Chapter 26’s problem, and they are much harder than this loop makes termination look.
16.6 The workflow canon: strategies over one kernel
The same refund task can be organized under very different control policies, and the industry has settled on a small canon of named patterns: the prompt chain, the router, the parallel fan-out, the evaluator–optimizer, and the adaptive agent (Anthropic 2024). Implementing them as five separate runtimes would duplicate message handling, dispatch, and safety behavior — and let the copies drift. The discipline that keeps a codebase sane is to fix what must be shared and vary only what genuinely differs. What must be shared is the effect boundary: every pattern executes tools through the same execute_call and the same refund_gate. What varies is exactly one thing: who chooses the next step. To compare the patterns fairly we also need shared accounting.
# @save
@dataclass
class Meter:
"""Measure what a pattern actually spends.
Wraps model calls and tool executions so every pattern is charged in the
same three currencies: model calls, tool calls, and estimated prompt
tokens. The differences between patterns then come from control policy,
not from accounting.
"""
model_calls: int = 0
tool_calls: int = 0
prompt_tokens: int = 0
def chat(self, model: Any, messages: list[dict[str, Any]]) -> dict[str, Any]:
"""Charge one model call and forward it to the model.
Args:
model: The model being consulted.
messages: The prompt about to be sent; its size is charged.
Returns:
The assistant message the model produced.
"""
self.model_calls += 1
self.prompt_tokens += estimate_tokens(messages)
return model.chat(messages)
def execute(self, call: ToolCall, tools: dict[str, ToolSpec], gate: Gate) -> Observation:
"""Charge one tool call and route it through the shared effect boundary.
Args:
call: The proposal to execute.
tools: The registry.
gate: The policy gate — the same one every pattern must pass.
Returns:
The typed observation from ``execute_call``.
"""
self.tool_calls += 1
return execute_call(call, tools, gate)The chain is the pattern to reach for when the path is known in advance: code owns the whole sequence and consults the model exactly once, for the only genuinely semantic step.
# @save
def run_chain(model: Any, tools: dict[str, ToolSpec], gate: Gate, meter: Meter) -> str:
"""The chain pattern: code owns the whole sequence.
Look up the order, refund exactly what was paid, then consult the model
once — for the only genuinely semantic step, drafting the reply. When the
path is known in advance, this is the cheapest and most testable shape.
Args:
model: Drafting model (one call).
tools: The registry.
gate: The shared policy gate.
meter: The accounting for this run.
Returns:
The drafted customer reply.
"""
order = meter.execute(ToolCall("c1", "lookup_order", {"order_id": "A-17"}),
tools, gate).content
receipt = meter.execute(
ToolCall("c2", "refund_order",
{"order_id": "A-17", "amount_cents": order["paid_cents"]}),
tools, gate).content
reply = meter.chat(model, [{"role": "user", "content":
f"Draft one sentence for the customer citing this receipt: "
f"{json.dumps(receipt)}"}])
return reply["content"]The parallel fan-out dispatches independent, effect-free reads together and joins the typed results before acting — safe here precisely because the two reads cannot conflict.
# @save
def run_parallel(model: Any, tools: dict[str, ToolSpec], gate: Gate, meter: Meter) -> str:
"""The parallel fan-out pattern: independent reads dispatched together.
Code fires both reads, joins the typed results, refunds the amount the
order reports, and consults the model once with everything in view. The
fan-out is safe because the reads are independent and effect-free.
Args:
model: Drafting model (one call).
tools: The registry.
gate: The shared policy gate.
meter: The accounting for this run.
Returns:
The drafted customer reply.
"""
reads = [ToolCall("p1", "lookup_order", {"order_id": "A-17"}),
ToolCall("p2", "read_policy", {})]
joined = {c.name: meter.execute(c, tools, gate).content for c in reads}
receipt = meter.execute(
ToolCall("p3", "refund_order",
{"order_id": "A-17",
"amount_cents": joined["lookup_order"]["paid_cents"]}),
tools, gate).content
reply = meter.chat(model, [{"role": "user", "content":
f"Order and policy: {json.dumps(joined)}. "
f"Draft one sentence citing {json.dumps(receipt)}."}])
return reply["content"]The router grants the model one bounded decision — classify the request — and code owns everything after it. The failure mode is proportionate to the grant: a misroute contaminates the entire branch, which is why the branch set stays small and the routing decision is worth logging.
# @save
def run_router(task: str, model: Any, tools: dict[str, ToolSpec], gate: Gate,
meter: Meter) -> str:
"""The router pattern: one bounded model decision picks a code-owned branch.
The model answers a single classification question; code owns everything
after it. A misroute contaminates the whole branch, which is why the
branch set stays small and the routing decision is worth logging.
Args:
task: The customer request being routed.
model: The routing (and drafting) model.
tools: The registry.
gate: The shared policy gate.
meter: The accounting for this run.
Returns:
The branch's final reply.
"""
route = meter.chat(model, [{"role": "user", "content":
f"Answer 'billing' or 'policy' only. Request: {task}"}])
branch = route["content"].strip()
if branch == "policy":
policy = meter.execute(ToolCall("r1", "read_policy", {}), tools, gate).content
return f"Our policy: {policy}"
return run_chain(model, tools, gate, meter)The evaluator–optimizer alternates generation with a check, and the check is what makes it more than retry-until-tired: code owns an acceptance criterion — here, that the draft states the exact refunded amount — and each rejection feeds a concrete critique back into the next attempt. A weak criterion rewards cosmetic revision, so the criterion deserves as much design attention as the prompt.
# @save
def run_evaluator(model: Any, tools: dict[str, ToolSpec], gate: Gate, meter: Meter,
max_rounds: int = 3) -> str:
"""The evaluator-optimizer pattern: generate, check, revise until it passes.
Code performs the transaction, then loops the model against an acceptance
criterion owned by code — here, that the draft states the exact refunded
amount. The criterion is what makes revision meaningful; a weak one
rewards cosmetic edits.
Args:
model: The drafting model, consulted once per round.
tools: The registry.
gate: The shared policy gate.
meter: The accounting for this run.
max_rounds: The revision budget.
Returns:
The first draft that passes the criterion, or the last draft.
"""
order = meter.execute(ToolCall("e1", "lookup_order", {"order_id": "A-17"}),
tools, gate).content
receipt = meter.execute(
ToolCall("e2", "refund_order",
{"order_id": "A-17", "amount_cents": order["paid_cents"]}),
tools, gate).content
critique = ""
draft = ""
for _ in range(max_rounds):
prompt = f"Draft a reply for this receipt: {json.dumps(receipt)}.{critique}"
draft = meter.chat(model, [{"role": "user", "content": prompt}])["content"]
if str(receipt["refunded_cents"]) in draft:
return draft
critique = " Critique: the draft must state the exact refunded amount."
return draftTwo of these deserve to be seen actually exercising their distinctive move. The router, given two different requests (with routing turns scripted accordingly), takes two different branches — note the second task never touches the refund tool at all.
for routed_task, script in [
("Order A-17 arrived damaged. Please make this right.",
ScriptedModel(turns=[answer_message("billing"),
answer_message("We refunded 4999 cents for order A-17.")])),
("What is your refund policy for damaged orders?",
ScriptedModel(turns=[answer_message("policy")])),
]:
tools, effects_r = make_environment()
meter = Meter()
reply = run_router(routed_task, script, tools, refund_gate, meter)
print(f"task: {routed_task}")
print(f" -> {reply}")
print(f" model_calls={meter.model_calls} tool_calls={meter.tool_calls} "
f"effects={len(effects_r)}")task: Order A-17 arrived damaged. Please make this right.
-> We refunded 4999 cents for order A-17.
model_calls=2 tool_calls=2 effects=1
task: What is your refund policy for damaged orders?
-> Our policy: Damaged orders: refund the amount paid, up to 5000 cents automatically.
model_calls=1 tool_calls=1 effects=0And the evaluator, given a scripted drafter whose first attempt omits the amount, visibly spends a second round to satisfy the criterion.
tools, effects_e = make_environment()
meter = Meter()
draft_model = ScriptedModel(turns=[
answer_message("We are sorry about the damage - your refund is on its way."),
answer_message("We are sorry about the damage - we refunded 4999 cents for order A-17."),
])
final = run_evaluator(draft_model, tools, refund_gate, meter)
print("accepted draft:", final)
print(f"model_calls={meter.model_calls} (round 1 failed the criterion)")accepted draft: We are sorry about the damage - we refunded 4999 cents for order A-17.
model_calls=2 (round 1 failed the criterion)Now the comparison. We run all five patterns — the fifth being the adaptive agent, our run_agent loop with the over-reaching model from Section 16.4 — on the same task against fresh environments, and read off what each spent and whether the correct refund landed. A small helper keeps the accounting uniform.
def measure(name: str, run_fn: Callable[..., Any]) -> dict[str, Any]:
"""Run one pattern against a fresh environment and record what it spent."""
tools, fx = make_environment()
meter = Meter()
run_fn(tools, meter)
return {"pattern": name, "model_calls": meter.model_calls,
"tool_calls": meter.tool_calls, "prompt_tokens": meter.prompt_tokens,
"refund_landed": fx == [{"order_id": "A-17", "refunded_cents": 4999}]}
def drafter() -> ScriptedModel:
"""A one-turn scripted drafter for the patterns that consult the model once."""
return ScriptedModel(turns=[
answer_message("We refunded 4999 cents for damaged order A-17."),
])rows = [
measure("chain", lambda t, m: run_chain(drafter(), t, refund_gate, m)),
measure("router", lambda t, m: run_router(task, ScriptedModel(
turns=[answer_message("billing"), answer_message("We refunded 4999 cents for A-17.")]),
t, refund_gate, m)),
measure("parallel", lambda t, m: run_parallel(drafter(), t, refund_gate, m)),
measure("evaluator", lambda t, m: run_evaluator(ScriptedModel(
turns=[answer_message("We are sorry - your refund is on its way."),
answer_message("We refunded 4999 cents for damaged order A-17.")]),
t, refund_gate, m)),
]
tools_a, fx_a = make_environment()
result_a = run_agent(task, overreaching_model(), tools_a, refund_gate)
rows.append({"pattern": "agent", "model_calls": result_a.state.turns,
"tool_calls": len(result_a.state.observations),
"prompt_tokens": sum(result_a.state.prompt_token_log),
"refund_landed": fx_a == [{"order_id": "A-17", "refunded_cents": 4999}]})
for row in rows:
print(f"{row['pattern']:10} model_calls={row['model_calls']} "
f"tool_calls={row['tool_calls']} prompt_tokens={row['prompt_tokens']:4} "
f"refund_landed={row['refund_landed']}")chain model_calls=1 tool_calls=2 prompt_tokens= 34 refund_landed=True
router model_calls=2 tool_calls=2 prompt_tokens= 65 refund_landed=True
parallel model_calls=1 tool_calls=3 prompt_tokens= 78 refund_landed=True
evaluator model_calls=2 tool_calls=2 prompt_tokens= 71 refund_landed=True
agent model_calls=4 tool_calls=3 prompt_tokens= 774 refund_landed=TrueEvery pattern reached the same correct outcome — same receipt, same gate — but at visibly different prices, and the price differences are structural, not incidental. The chain spent one model call and 34 estimated prompt tokens because code carried all the control flow. The agent spent four model calls and 774 tokens — roughly twenty times the chain’s context bill — because each turn resends the growing transcript, and one of those turns was the denied over-reach. That ratio is measured on scripted drafts, so it says nothing about output quality; what it measures honestly is the overhead each control policy pays before quality even enters the question. Figure 16.6 draws both currencies.
Show the code that draws this figure
names = [row["pattern"] for row in rows]
x = range(len(names))
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(7.6, 3.2))
ax1.bar([i - 0.18 for i in x], [row["model_calls"] for row in rows], width=0.36,
label="model calls", color="#2a7f9e")
ax1.bar([i + 0.18 for i in x], [row["tool_calls"] for row in rows], width=0.36,
label="tool calls", color="#b7791f")
ax1.set_xticks(list(x)); ax1.set_xticklabels(names, rotation=25, fontsize=8)
ax1.set_ylabel("calls"); ax1.legend(fontsize=8)
ax2.bar(list(x), [row["prompt_tokens"] for row in rows], width=0.5, color="#557a3c")
ax2.set_xticks(list(x)); ax2.set_xticklabels(names, rotation=25, fontsize=8)
ax2.set_ylabel("estimated prompt tokens")
fig.tight_layout()
plt.show()
The canon in one view, with the question that actually selects among them — who chooses the next step — and the failure each choice buys:
| Pattern | Who chooses the next step | Best fit | Characteristic failure |
|---|---|---|---|
| Chain | Code, fixed in advance | Stable, known sequence | Cannot adapt to an unmodeled case |
| Router | One bounded model decision, then code | Distinct task families | Misroute contaminates the branch |
| Parallel fan-out | Code-owned dispatcher | Independent, effect-free subtasks | Cost growth; conflicting writes |
| Evaluator–optimizer | Code loops; criterion decides acceptance | Output with a checkable rubric | Weak criterion rewards cosmetic revision |
| Adaptive agent | Model, within the gated action space | Path depends on observations | Wandering, compounding cost, larger attack surface |
The table’s rows are not rungs of sophistication — they are positions on the autonomy slider from Section 1.8, and the slider is set per transition, not per system. A production support desk might route deterministically, run an agentic read phase, and then execute the refund as a fixed, gated transaction. The engineering question is never “workflow or agent?” as an identity; it is “which transition benefits from model choice, and what measured evidence pays for that choice?”
16.7 ReAct, planning, and replanning
The adaptive agent we have run so far acts without narrating. ReAct — reason and act — added a simple, influential discipline: interleave visible reasoning with actions and observations, so that each consequential step is grounded in what the environment just said rather than in a plan hallucinated whole at turn one (Yao et al. 2023). Our wire format already has the slot for it — the content field that rides alongside tool_calls — so a ReAct turn is an assistant message carrying both a thought and an action. Here is a full trace through the gated loop; the thoughts are scripted, like every model output in this chapter, but the interleaving machinery — thought and action arriving together, each next step conditioned on a fresh observation — is exactly the mechanism, and the trace below is really executed.
tools, effects_react = make_environment()
react_model = ScriptedModel(turns=[
tool_call_message("t1", "lookup_order", {"order_id": "A-17"},
thought="Thought: before promising anything, see what A-17 actually is."),
tool_call_message("t2", "read_policy", {},
thought="Thought: damaged, paid 4999 cents. Check what I may do automatically."),
tool_call_message("t3", "refund_order", {"order_id": "A-17", "amount_cents": 4999},
thought="Thought: 4999 is within the 5000-cent limit; refund exactly what was paid."),
answer_message("Your order arrived damaged, so I refunded the full 4999 cents you paid."),
])
result_react = run_agent(task, react_model, tools, refund_gate)
print_transcript(result_react.state)
print(f"\nstop={result_react.stop.value} effects={effects_react}")system | You are a support agent. Use the tools to resolve the customer's problem. A denied call is information: read the reason and adjust.
user | Order A-17 arrived damaged. Please make this right.
model | Thought: before promising anything, see what A-17 actually is.
model | -> lookup_order {"order_id": "A-17"}
tool | {"ok": true, "kind": "result", "content": {"order_id": "A-17", "status": "damaged", "paid_cents": 4999}}
model | Thought: damaged, paid 4999 cents. Check what I may do automatically.
model | -> read_policy {}
tool | {"ok": true, "kind": "result", "content": "Damaged orders: refund the amount paid, up to 5000 cents automatically."}
model | Thought: 4999 is within the 5000-cent limit; refund exactly what was paid.
model | -> refund_order {"order_id": "A-17", "amount_cents": 4999}
tool | {"ok": true, "kind": "result", "content": {"order_id": "A-17", "refunded_cents": 4999}}
model | Your order arrived damaged, so I refunded the full 4999 cents you paid.
stop=answered effects=[{'order_id': 'A-17', 'refunded_cents': 4999}]Read the rhythm: thought, action, observation, next thought conditioned on that observation. The engineering value of ReAct is not that the thought text becomes trustworthy — a model can narrate one plan and execute another, which is why our gate never reads the thoughts at all. The value is grounding (the refund amount came from a lookup, not from the model’s prior) plus an operational record: a concise stated reason next to each structured action is easier to audit and redact than raw hidden reasoning, and Chapter 25 examines how far such narrations can be trusted as evidence. ReAct’s successors keep the skeleton and extend the feedback: Reflexion, for example, feeds verbal self-critique from failed episodes back into the next attempt, a memory-shaped idea we build properly in Chapter 18 (Shinn et al. 2023).
Planning goes one step further than a per-turn thought: the model proposes a future — several steps at once. The temptation is to treat an approved-looking plan as permission to execute it. Resist that: a plan is a hypothesis about dependencies, not an execution grant, and the world can change between planning and acting. Mechanically that means each planned step passes the same gate at execution time, and the executor stops at the first failure so the planner can revise with fresh information.
# @save
def execute_plan(plan: list[dict[str, Any]], tools: dict[str, ToolSpec],
gate: Gate) -> tuple[list[Observation], int]:
"""Execute a model-proposed plan step by step through the normal gate.
A plan is a hypothesis about the future, not an execution grant: each
step is gated at execution time, and the executor stops at the first
step that fails so the planner can revise with fresh information.
Args:
plan: Steps of the form ``{"tool": name, "arguments": {...}}``.
tools: The registry.
gate: The same policy gate every other pattern uses.
Returns:
The observations for the executed prefix, and the index of the first
failed step (``len(plan)`` when every step succeeded).
"""
observations: list[Observation] = []
for index, step in enumerate(plan):
observation = execute_call(
ToolCall(f"s{index}", step["tool"], step["arguments"]), tools, gate)
observations.append(observation)
if not observation.ok:
return observations, index
return observations, len(plan)Our scripted planner drafts its plan before looking anything up, and guesses the refund at 9,999 cents — a stale-information mistake real planners make constantly. Watch the plan survive parsing, start executing, and hit the gate mid-plan.
tools, effects_plan = make_environment()
planner = ScriptedModel(turns=[
answer_message(json.dumps([
{"tool": "lookup_order", "arguments": {"order_id": "A-17"}},
{"tool": "refund_order", "arguments": {"order_id": "A-17", "amount_cents": 9999}},
])),
answer_message(json.dumps([
{"tool": "refund_order", "arguments": {"order_id": "A-17", "amount_cents": 4999}},
])),
])
plan = json.loads(planner.chat([{"role": "user", "content": task}])["content"])
print("plan v1:", json.dumps(plan, indent=2))
observations, failed_at = execute_plan(plan, tools, refund_gate)
for observation in observations:
print(observation)
print(f"stopped at step {failed_at}; effect log: {effects_plan}")plan v1: [
{
"tool": "lookup_order",
"arguments": {
"order_id": "A-17"
}
},
{
"tool": "refund_order",
"arguments": {
"order_id": "A-17",
"amount_cents": 9999
}
}
]
Observation(call_id='s0', ok=True, kind='result', content={'order_id': 'A-17', 'status': 'damaged', 'paid_cents': 4999})
Observation(call_id='s1', ok=False, kind='denied', content='refund exceeds automatic authority of 5000 cents')
stopped at step 1; effect log: []Step 0 executed and step 1 was denied, so the effect log is still empty — the bad half of the plan simply never ran. Now the replanning move: we hand the planner the failure and the fresh observation, and it produces a corrected plan.
replan_prompt = (f"Step {failed_at} failed: {observations[-1].content}. "
f"The order shows paid_cents={observations[0].content['paid_cents']}. Replan.")
plan2 = json.loads(planner.chat([{"role": "user", "content": replan_prompt}])["content"])
print("plan v2:", json.dumps(plan2))
observations2, failed_at2 = execute_plan(plan2, tools, refund_gate)
print(f"completed {failed_at2}/{len(plan2)} steps; effect log: {effects_plan}")plan v2: [{"tool": "refund_order", "arguments": {"order_id": "A-17", "amount_cents": 4999}}]
completed 1/1 steps; effect log: [{'order_id': 'A-17', 'refunded_cents': 4999}]The trigger for replanning was a code-visible event — a step failed — not a vibe. That is the durable rule: replan on a failed precondition, a denied step, a rejected outcome, or a crossed budget, because those are events a runtime can log and an evaluator can count; “reflect when uncertain” gives neither one anything to inspect. The three planning shapes you will meet are all variations on this cell: plan-then-execute (one plan, executed with verification — what we just did), a rolling plan (revise a short horizon after every observation — more model calls, less staleness), and planner–executor–verifier (separate roles, which reduce correlated error only when the verifier gets independent evidence rather than a third read of the same context). Search over candidate plans with verifier guidance is Chapter 8’s machinery; the systems rule it inherits here is that search consumes explicit budget and every surviving step still passes the gate one at a time. More thinking never enlarges authority.
16.8 A specification you can test against
Now that the loop exists and has run, the formal skeleton underneath it is cheap to state and worth owning, because it tells us what to test. A run is a growing history — our messages list — and each iteration applies one step relation:
h_{t+1} \;=\; h_t \oplus \bigl(a_t,\, o_t\bigr), \qquad a_t = \pi(h_t), \qquad o_t = E\bigl(a_t,\, G(a_t)\bigr), \tag{16.1}
where h_t is the transcript at turn t, \pi is the proposal policy (the model — ScriptedModel.chat here, a hosted model in production), a_t its proposed action, G the gate, E the executor that produces observation o_t (a denial when G refuses), and \oplus appends both to the history. The endings in Figure 16.5 are the absorbing states of the machine Equation 16.1 defines, and the budgets guarantee one is reached. Five invariants make the specification testable, and each one is a line of code you have already read: effect authorization — no handler runs without validation and gate approval (execute_call is the only call site); correlation — every observation carries the id of the proposal that caused it (the tool_call_id echo); boundedness — the loop returns within its budgets (the while condition and the token check); typed endings — every return path names a Stop; and outcome separation — “the model answered” and “the task succeeded” are different predicates, which the rest of this section makes concrete. Because the state is plain data, checking the first four needs no re-running:
# @save
def check_run(result: RunResult, effects: list[dict[str, Any]],
limits: Limits) -> dict[str, bool]:
"""Check a finished run's mechanical invariants from its own record.
These are the properties the loop must guarantee no matter how the model
behaved: every effect was authorized through an allowed observation,
every tool message correlates to a proposed call id, the turn budget was
respected, and the ending is one of the typed stops.
Args:
result: The finished run.
effects: The environment's effect log for the same run.
limits: The limits the run was given.
Returns:
A mapping from invariant name to whether it held.
"""
state = result.state
proposed_ids = {item["id"]
for message in state.messages if message.get("tool_calls")
for item in message["tool_calls"]}
return {
"effects_authorized": all(
any(o.ok and o.content == receipt for o in state.observations)
for receipt in effects),
"observations_correlated": all(
message["tool_call_id"] in proposed_ids
for message in state.messages if message["role"] == "tool"),
"bounded": state.turns <= limits.max_turns,
"typed_ending": isinstance(result.stop, Stop),
}print(check_run(result_gated, effects_gated, Limits()))
print(check_run(result_s, effects_stub, Limits())){'effects_authorized': True, 'observations_correlated': True, 'bounded': True, 'typed_ending': True}
{'effects_authorized': True, 'observations_correlated': True, 'bounded': True, 'typed_ending': True}Both the successful denial-recovery run and the no-progress run satisfy all four — which is the point of a specification: the unsuccessful run is still a correct run, because failing inside the rules is an outcome the system was designed to produce. The fifth invariant needs the environment. answered is a control-flow fact; whether the customer was actually made whole is a question only the effect log can answer, and conflating the two is the most common measurement error in agent systems. The verifier is three lines:
# @save
def verify_refund(effects: list[dict[str, Any]], order_id: str, amount_cents: int) -> bool:
"""Decide task success from the environment, not from the answer text.
Success is exactly one receipt for the right order and the right amount.
A polite "your refund is complete" beside an empty effect log fails this
check, which is the entire point of grounding verification in state the
model cannot narrate its way around.
Args:
effects: The environment's append-only effect log.
order_id: The order that should have been refunded.
amount_cents: The exact amount that should have moved.
Returns:
True when the log contains exactly the one intended receipt.
"""
return effects == [{"order_id": order_id, "refunded_cents": amount_cents}]And here is the model it exists to catch — one that announces success without ever proposing a tool call:
tools, effects_liar = make_environment()
liar = ScriptedModel(turns=[
answer_message("Good news - your refund for order A-17 is complete!"),
])
result_liar = run_agent(task, liar, tools, refund_gate)
print(f"stop={result_liar.stop.value}")
print(f"answer: {result_liar.answer}")
print(f"effect log: {effects_liar}")
print(f"verified: {verify_refund(effects_liar, 'A-17', 4999)}")stop=answered
answer: Good news - your refund for order A-17 is complete!
effect log: []
verified: FalseThe run ended answered — truthfully, as a statement about control flow — and verification failed, truthfully, as a statement about the world. Keeping both signals is what lets an evaluator score the run zero while the loop reports it terminated normally. In practice verification comes in three layers of increasing grip: invariant checks like check_run (did the system stay inside its own rules?), process checks (did required evidence exist — was the order read before the refund, does the answer cite the receipt?), and outcome checks like verify_refund (is the environment in the goal state?). Outcome checks are the strongest when the environment exposes a deterministic predicate — database state, passing tests, a compiled artifact. Model-based judges cover semantic qualities code cannot express, at the price of their own calibration problem, which is Chapter 22’s subject; Chapter 8 already showed why an external verifier outranks a model revising its own text.
Step back and the chapter’s design stance has a public name: it is most of 12-Factor Agents, a practitioner distillation of what survives contact with production (Horthy 2025). Own your control flow (the while loop is ours, not a framework’s); own your context and state (the transcript is plain data we can persist, replay, and resume); treat tool calls as structured outputs to parse and judge, never as capabilities the model holds. Everything this kernel deliberately does not do — authenticate tenants, isolate untrusted code, bind human approvals, survive crashes mid-effect, reconcile ambiguous remote failures — has an owner in the chapters ahead: the harness in Chapter 17, protocols and frameworks in Chapter 19, security in Chapter 24, durable execution in Chapter 26. The raw loop is worth having built precisely so that each of those can be judged as an addition to a mechanism you already understand.
16.9 Summary
An agent is a model proposing typed actions inside a loop that code owns. We built the whole thing: tools with declared contracts, the tool-call wire format and its failure-tolerant parse, the propose-gate-execute-observe loop, and the gate that turned a 9,999-cent over-reach into a denial the model recovered from — with the effect log, not the model’s prose, as witness. Budgets gave every run a typed ending; the workflow canon became five control policies over one kernel, measured at 1–4 model calls and 34–774 prompt tokens; verification checked the environment instead of believing the answer. Chapter 17 builds the harness around this loop.
16.10 Exercises
- Trace by hand. Before running anything: for the happy-path script of Section 16.3 under
Limits(max_turns=2), predict the stop reason,turns, the length ofobservations, and the effect log. Run it and reconcile every difference. - Find the layer. Construct three bad proposals:
amount_centsas the string"4999", an arguments field that is not valid JSON, and a refund of 999,999 cents. Predict which layer — parse, validation, or gate — converts each into whichObservation.kind, then confirm withexecute_call. Explain why the three layers should not be merged into one check. - Detect oscillation. Defeat the no-progress counter with a script that alternates denied refunds of 9,998 and 9,999 cents (use
reactionskeyed on"denied"plus a mutating closure, or subclassScriptedModel). Extendrun_agentto end the run when any recent window of denials repeats, while still allowing one legitimate retry after anerrorobservation. State the window size you chose and what it costs in false positives. - Parallel proposals. The wire format carries a list of tool calls, and
run_agentalready executes all of them in order.- Build one assistant message proposing
lookup_orderandread_policytogether and run it; confirm both observations appear andcheck_runpasses. - Reverse the order in which the two tool messages are appended and re-run
check_run. Which invariant would break if results were matched by list position instead oftool_call_id? - Add a second
refund_orderto the same message and decide: should the loop execute effectful calls in parallel batches at all? Defend your answer with the effect log.
- Build one assistant message proposing
- Completion versus success. Extend the liar of Section 16.8 so it fabricates a plausible receipt id in its answer. Write a process check that requires a successful
lookup_orderobservation to precede any refund receipt, and show it rejecting a run where the refund happened without a prior lookup. - Budget the evaluator. Give
run_evaluatora drafter whose scripted drafts never satisfy the criterion. Measure model calls and prompt tokens asmax_roundssweeps 1 to 6, plot the growth, and decide whichStopvalue an exhausted revision budget should map to — then argue where that budget belongs: inside the pattern, or in the caller’sLimits. - Prompt versus gate. A reviewer proposes moving the 5,000-cent rule out of
refund_gateand intoSYSTEM_PROMPTso policy can change without a deploy. Using the two runs of Section 16.4 as evidence, write the rebuttal; then design the compromise — the limit as configuration read by the gate at call time — and name the one residual risk (stale configuration) your design still carries.