Implementing the ReAct Loop from Scratch
50 min readnotebookFrom Workflows to Agents
3 of 16Agent Architectures: Loops, Planning & Memory
Up next · Quiz: Agent Loops
Every agent framework you will ever evaluate — LangGraph, the OpenAI Agents SDK, Microsoft Agent Framework — is, at its core, a loop that sends messages to a model, executes the tool calls that come back, appends the results, and repeats until something says stop. The frameworks wrap that loop in tens of thousands of lines of graph abstractions, callbacks, and integrations. Today you write the loop itself: about a hundred and fifty lines of Python that accumulate messages, dispatch tools, enforce budgets, detect when the agent is going in circles, and trace every step to a log you can actually read. This file becomes the substrate for the rest of the course — the planner of Lesson 5 and the reflection wrapper of Lesson 6 both call it.
The code targets the Anthropic Messages API, whose tool-use mechanics you know from LLM Engineering: APIs, Inference & Structured Output. Because it makes live API calls, the main loop is not runnable in the browser — read it section by section and assemble it in your own environment. The loop-detection logic, though, is pure Python, and you will run it right here.
Before writing a line, enumerate what the loop owes you. Six responsibilities, each of which will map to a visible piece of code:
tool_result block tied to its
tool_use_id, truncated to a sane size.From Tool Use, Code Execution & Sandboxing you know how to design schemas, sandbox execution, and cap output sizes. Here we need only a thin registry over three file tools — enough to give the agent something real to do (answer questions about a codebase) while keeping the lesson's focus on the loop:
# tools.py
import pathlib, subprocess
MAX_RESULT = 8_000 # characters; tool results are context you pay for
def read_file(path: str) -> str:
p = pathlib.Path(path)
if not p.is_file():
return f"error: {path} is not a file"
text = p.read_text(errors="replace")
return text[:MAX_RESULT] + ("\n...[truncated]" if len(text) > MAX_RESULT else "")
def grep(pattern: str, path: str = ".") -> str:
r = subprocess.run(["grep", "-rn", "-m", "50", pattern, path],
capture_output=True, text=True, timeout=10)
return r.stdout[:MAX_RESULT] or "no matches"
def list_dir(path: str = ".") -> str:
return "\n".join(sorted(str(p) for p in pathlib.Path(path).iterdir()))
REGISTRY = {
"read_file": (read_file, "Read a file's contents.",
{"type": "object", "properties": {"path": {"type": "string"}},
"required": ["path"]}),
"grep": (grep, "Search files for a pattern (first 50 matches).",
{"type": "object", "properties": {"pattern": {"type": "string"},
"path": {"type": "string"}},
"required": ["pattern"]}),
"list_dir": (list_dir, "List a directory.",
{"type": "object", "properties": {"path": {"type": "string"}},
"required": []}),
}
TOOL_SPECS = [{"name": name, "description": desc, "input_schema": schema}
for name, (fn, desc, schema) in REGISTRY.items()]
def dispatch(name: str, args: dict) -> str:
if name not in REGISTRY:
return f"error: unknown tool {name}"
fn = REGISTRY[name][0]
try:
return str(fn(**args))
except Exception as e:
return f"error: {type(e).__name__}: {e}" # errors are observations
Two details worth pausing on. MAX_RESULT exists
because tool results are context: an unbounded
read_file on a 2 MB log would inject half a million
tokens into every subsequent step's input bill.
And dispatch never raises — a
FileNotFoundError becomes the string
error: FileNotFoundError: ..., which the model reads
as an observation and routes around, exactly as it would a "no
matches" from grep.
The Messages API expresses tool use as content blocks. When the
model wants a tool, the response arrives with
stop_reason: "tool_use" and one or more blocks like
this in content:
{
"type": "tool_use",
"id": "toolu_01A9rXk...",
"name": "grep",
"input": {"pattern": "def dispatch", "path": "app/"}
}
Your reply — as a user message — carries the results, each tied back by id:
{
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": "toolu_01A9rXk...",
"content": "app/tools.py:31:def dispatch(name, args):"
}]
}
The alternation is strict: the assistant turn containing the
tool_use blocks must be appended to history verbatim
(the whole content list, including any thinking or
text blocks around the tool calls), and the very next message
must be the user turn with the matching results. Three
stop_reason values matter to the loop:
tool_use (execute and continue),
end_turn (the model considers itself finished — this
is your natural exit), and max_tokens (the response
was cut off; treat it as an incomplete answer, not a final one).
Now the loop itself. It is a class only because the budgets, client, and trace path are worth holding as state; there is no framework hiding in it:
# loop.py
import json, time
import anthropic
from tools import TOOL_SPECS, dispatch, MAX_RESULT
from detect import LoopDetector # Section 5
SYSTEM = (
"You are a code-analysis agent. Interleave brief reasoning with "
"tool calls. Ground every claim in a tool observation. When you "
"have enough evidence, give the final answer. If a tool result "
"starts with 'error:', adapt — do not repeat the same call."
)
class AgentLoop:
def __init__(self, model="claude-sonnet-4-6", max_steps=20,
max_tokens_total=150_000, trace_path="trace.jsonl"):
self.client = anthropic.Anthropic()
self.model = model
self.max_steps = max_steps
self.max_tokens_total = max_tokens_total
self.trace_path = trace_path
def run(self, goal: str) -> dict:
messages = [{"role": "user", "content": goal}]
detector, tokens_used = LoopDetector(), 0
for step in range(1, self.max_steps + 1):
t0 = time.perf_counter()
r = self.client.messages.create(
model=self.model, max_tokens=2048, system=SYSTEM,
tools=TOOL_SPECS, messages=messages)
tokens_used += r.usage.input_tokens + r.usage.output_tokens
self._trace(step, r, time.perf_counter() - t0, tokens_used)
if r.stop_reason != "tool_use": # natural exit
text = "".join(b.text for b in r.content if b.type == "text")
return {"status": "done", "answer": text,
"steps": step, "tokens": tokens_used}
messages.append({"role": "assistant", "content": r.content})
results = []
for block in r.content:
if block.type != "tool_use":
continue
verdict = detector.check(block.name, block.input)
out = verdict or dispatch(block.name, block.input)
detector.observe(block.name, block.input, out)
results.append({"type": "tool_result",
"tool_use_id": block.id,
"content": out[:MAX_RESULT]})
messages.append({"role": "user", "content": results})
if tokens_used > self.max_tokens_total: # budget exit
return self._wrap_up(messages, "token budget",
step, tokens_used)
return self._wrap_up(messages, "step budget",
self.max_steps, tokens_used)
def _wrap_up(self, messages, reason, steps, tokens) -> dict:
messages.append({"role": "user", "content":
f"STOP: the {reason} is exhausted. Do not call more tools. "
"Report what you established (with evidence), what remains "
"unknown, and what you would try next. Do not guess."})
r = self.client.messages.create(model=self.model, max_tokens=1024,
system=SYSTEM, messages=messages)
text = "".join(b.text for b in r.content if b.type == "text")
return {"status": f"{reason} exhausted", "answer": text,
"steps": steps, "tokens": tokens}
Read the exits. The natural one fires when
stop_reason is anything but tool_use —
the model has answered. The budget exits do not simply
return "budget exhausted": they spend one final tool-free call
asking the model to report verified findings and open questions,
explicitly forbidding invention. That distinction — between
aborting and concluding — is the difference between an agent that
fails usefully and one that fails silently; Lesson 7 builds the
full protocol.
The classic agent pathology: it greps for a symbol that does not exist, gets "no matches", and — believing the tool failed rather than the assumption — greps again. And again. A step budget will eventually stop it, but forty steps late and forty steps expensive. The detector below catches the spiral in two ways: a repeated-signature check (same tool, same canonicalized arguments, seen too many times in a sliding window) and a novelty check (results whose hash we have already seen add no new information). On detection it does not kill the run — it returns an intervention string that the loop injects as the tool result, steering the model instead of stopping it:
# detect.py
import hashlib, json
class LoopDetector:
def __init__(self, max_repeats=3, window=10):
self.max_repeats = max_repeats
self.window = window
self.calls = [] # signature history
self.seen_results = set() # result-hash history
@staticmethod
def _sig(name: str, args: dict) -> str:
canon = json.dumps(args, sort_keys=True, separators=(",", ":"))
return hashlib.sha256((name + ":" + canon).encode()).hexdigest()[:16]
def check(self, name: str, args: dict):
sig = self._sig(name, args)
if self.calls[-self.window:].count(sig) >= self.max_repeats:
return (f"error: you have already called {name} with these "
"exact arguments several times and the result will "
"not change. Take a different approach, or state "
"your best answer from the evidence you have.")
return None
def observe(self, name: str, args: dict, result: str) -> bool:
self.calls.append(self._sig(name, args))
h = hashlib.sha256(result[:2_000].encode()).hexdigest()
fresh = h not in self.seen_results
self.seen_results.add(h)
return fresh # False = no new information
Canonicalizing the arguments (sort_keys, fixed
separators) matters: without it, {"path": ".", "pattern":
"x"} and {"pattern": "x", "path": "."} hash
differently and the detector goes blind. Run the simulation below
to watch the detector catch a stuck agent — this one is pure
Python, so press Run and then experiment with the thresholds:
The detector fires on step 5 — the fourth identical grep — after
which a real loop would hand the intervention string back as the
observation. Note the [no new information] tags on
the repeated "no matches" results before the trip-wire: the
novelty signal degrades earlier and more gradually than the
signature counter, which makes it useful as a soft warning
(Lesson 7 wires it into a budget-tightening policy). Try setting
max_repeats=2, or vary the pattern slightly per call
to see why near-duplicate detection is a harder problem than
exact-duplicate detection.
An agent run is a distributed system of one process: things happen in an order you did not write, for reasons you cannot see without instrumentation. The trace is one JSONL record per step — cheap to write, and the first thing you will open every single time an agent misbehaves:
# inside AgentLoop
def _trace(self, step, r, latency, tokens_total):
rec = {
"ts": time.time(),
"step": step,
"stop_reason": r.stop_reason,
"text": "".join(b.text for b in r.content
if b.type == "text")[:300],
"tool_calls": [{"tool": b.name, "args": b.input}
for b in r.content if b.type == "tool_use"],
"in_tokens": r.usage.input_tokens,
"out_tokens": r.usage.output_tokens,
"tokens_total": tokens_total,
"latency_s": round(latency, 2),
}
with open(self.trace_path, "a") as f:
f.write(json.dumps(rec) + "\n")
A real trace excerpt, reformatted for reading — notice how the reasoning text plus the arguments tell you the agent's belief at every step, and how the token column makes the quadratic growth visible:
{"step": 1, "stop_reason": "tool_use", "text": "I need to find where dispatch is defined.",
"tool_calls": [{"tool": "grep", "args": {"pattern": "def dispatch"}}],
"in_tokens": 1094, "out_tokens": 87, "tokens_total": 1181, "latency_s": 1.9}
{"step": 2, "stop_reason": "tool_use", "text": "Found it in tools.py. Reading it.",
"tool_calls": [{"tool": "read_file", "args": {"path": "app/tools.py"}}],
"in_tokens": 1421, "out_tokens": 71, "tokens_total": 2673, "latency_s": 1.7}
{"step": 3, "stop_reason": "end_turn", "text": "dispatch() maps tool names to functions and converts exceptions to error strings...",
"in_tokens": 3866, "out_tokens": 214, "tokens_total": 6753, "latency_s": 3.4}
Assemble the three files, point the loop at a small repository,
and ask it something that needs two or three hops
(AgentLoop().run("What does the dispatch function in this
repo do when a tool raises?")). Then break it on purpose —
this is the most instructive twenty minutes in the course so far:
max_steps=3 and ask a five-hop question. Compare
the wrap-up answer with and without the "do not guess"
instruction — the difference is fabrication.-i, tweak a word).
Exact signatures stop matching; only the novelty check still
fires. Sit with that gap — you close it in Lesson 7.What you have at the end is small and real: a loop that accumulates messages correctly, dispatches tools without dying, stops for the right reasons, refuses to spiral, and leaves a trace you can read. Everything else in this course — planning, reflection, memory, persistence, human gates — is built by wrapping or extending this file, never by replacing it. The quiz next lesson checks the mechanics; then Lesson 5 teaches the loop to follow a plan.