ReAct from scratch

Open in Colab

ReAct from scratch#

Track 04 - Agents · Notebook 01 · Runtime: ~5s on CPU

Prerequisites: comfortable with regex and dataclasses.

A ReAct agent (Yao et al. 2022) is a three-line loop: ask the LLM, run the tool, append the observation, repeat. Two refinements over a bare loop — emit a Thought: line before each Action:, and constrain actions to a strict name[arg] grammar so a parser can dispatch reliably.

We build the parser, the tools, the loop, and the scoring against recorded claude-haiku-4-5 ReAct traces in _fixtures/01_react_traces.json. The model’s strings are replayed; the parser, tools, and loop run for real. LIVE mode (ANTHROPIC_API_KEY set) hits the real API.

For the production-grade version that uses native tool_use instead of regex parsing, see 08_production/03_tool_use_agent.

from llm_systems_cookbook.nb import bootstrap
from llm_systems_cookbook._utils import repo_root

import ast
import json
import operator as op
import os
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path

s = bootstrap("04_agents_01_react_from_scratch")

LIVE = bool(os.environ.get("ANTHROPIC_API_KEY"))
MODEL = os.environ.get("MODEL_ANTHROPIC", "claude-haiku-4-5")
FIXTURE = json.loads((Path(repo_root()) / "notebooks/04_agents/_fixtures/01_react_traces.json").read_text())
print(f"mode={'LIVE' if LIVE else 'REPLAY'}  model={MODEL}  trajectories={len(FIXTURE['trajectories'])}")

Step 1 - The parser#

Whatever the LLM generates, we need to turn it back into a structured object with fields like thought, action, action_input, or final_answer. Two regexes are enough:

  • Action:\s*(\w+)\[(.*?)\] matches Action: calculator[2 + 2].

  • Final Answer:\s*(.*) matches the terminator.

If neither matches, the output is a parse error - the agent didn’t follow its own format. Parse errors are tracked as a separate metric because a model that reasons brilliantly but can’t stay in format is useless in production.

ACTION_RE = re.compile(r"Action:\s*(\w+)\[(.*?)\]", re.DOTALL)
FINAL_RE = re.compile(r"Final\s*Answer:\s*(.*)", re.DOTALL)


@dataclass
class ParsedStep:
    thought: str | None
    action: str | None
    action_input: str | None
    final_answer: str | None
    raw: str

    @property
    def is_action(self) -> bool: return self.action is not None

    @property
    def is_final(self) -> bool: return self.final_answer is not None


def parse_llm_output(text: str) -> ParsedStep:
    thought_match = re.search(r"Thought:\s*(.+?)(?=\n(?:Action|Final Answer):|$)", text, re.DOTALL)
    thought = thought_match.group(1).strip() if thought_match else None
    final_match = FINAL_RE.search(text)
    if final_match:
        return ParsedStep(thought, None, None, final_match.group(1).strip(), text)
    action_match = ACTION_RE.search(text)
    if action_match:
        return ParsedStep(thought, action_match.group(1), action_match.group(2).strip(), None, text)
    return ParsedStep(thought, None, None, None, text)


# Smoke test: does the parser find the action in a well-formed output?
example = "Thought: I need to compute 2+2.\nAction: calculator[2+2]"
parsed = parse_llm_output(example)
print(f"action = {parsed.action!r}  input = {parsed.action_input!r}")
assert parsed.action == "calculator"

Tools — three Python functions#

Every tool is a str → str function. The loop calls TOOLS[action](action_input) and feeds the return value back as the next Observation:.

  • calculator — AST walk (never eval) over + - * / ** %.

  • wiki_search — dict lookup over 25 facts. In production this would hit retrieval; here it’s a dict for determinism.

  • get_datetime — relative to a frozen now so the test assertions don’t drift by day.

_ALLOWED_OPS = {
    ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv,
    ast.Pow: op.pow, ast.USub: op.neg, ast.UAdd: op.pos, ast.Mod: op.mod,
}


def _eval_ast(node: ast.AST) -> float:
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return node.value
    if isinstance(node, ast.BinOp):
        return _ALLOWED_OPS[type(node.op)](_eval_ast(node.left), _eval_ast(node.right))
    if isinstance(node, ast.UnaryOp):
        return _ALLOWED_OPS[type(node.op)](_eval_ast(node.operand))
    raise ValueError(f"unsupported node {type(node).__name__}")


def calculator(expression: str) -> str:
    try:
        value = _eval_ast(ast.parse(expression.strip(), mode="eval").body)
    except Exception as e:  # noqa: BLE001 - tool outputs must never raise
        return f"ERROR: {type(e).__name__}: {e}"
    if isinstance(value, float) and value.is_integer():
        return str(int(value))
    return str(value) if not isinstance(value, float) else f"{value:.6f}".rstrip("0").rstrip(".")


FACTS = {
    "capital of france": "Paris", "capital of japan": "Tokyo", "capital of brazil": "Brasilia",
    "capital of canada": "Ottawa", "capital of kenya": "Nairobi", "capital of australia": "Canberra",
    "capital of south africa": "Pretoria", "capital of egypt": "Cairo",
    "capital of argentina": "Buenos Aires", "capital of mexico": "Mexico City",
    "tallest mountain": "Mount Everest at 8848 metres.",
    "longest river": "The Nile is 6650 km long.",
    "speed of light": "Approximately 299,792,458 metres per second in vacuum.",
    "electron charge": "Approximately 1.602e-19 coulombs.",
    "planck constant": "Approximately 6.626e-34 joule-seconds.",
    "number of continents": "Seven.",
    "chemical symbol for gold": "Au.", "chemical symbol for iron": "Fe.",
    "chemical symbol for silver": "Ag.", "chemical symbol for tungsten": "W.",
    "author of hamlet": "William Shakespeare.", "author of the iliad": "Homer.",
    "author of war and peace": "Leo Tolstoy.", "author of the great gatsby": "F. Scott Fitzgerald.",
    "painter of the mona lisa": "Leonardo da Vinci.",
}


def wiki_search(query: str) -> str:
    q = query.strip().lower().rstrip("?").strip()
    if q in FACTS:
        return FACTS[q]
    for key, val in FACTS.items():
        if key in q or q in key:
            return val
    return "No matching result in the local knowledge base."


FIXED_NOW = datetime(2026, 4, 17, 12, 0, tzinfo=timezone.utc)


def get_datetime(spec: str) -> str:
    spec = spec.strip().lower()
    if spec in {"", "now", "today", "current date"}: return FIXED_NOW.date().isoformat()
    if spec == "year":     return str(FIXED_NOW.year)
    if spec == "month":    return FIXED_NOW.strftime("%B")
    if spec == "day":      return FIXED_NOW.strftime("%A")
    if spec == "tomorrow": return (FIXED_NOW + timedelta(days=1)).date().isoformat()
    if spec == "yesterday":return (FIXED_NOW - timedelta(days=1)).date().isoformat()
    return f"ERROR: unknown date spec {spec!r}"


TOOLS: dict[str, callable] = {
    "calculator": calculator, "wiki_search": wiki_search, "get_datetime": get_datetime,
}
print("tools:", list(TOOLS))

The policy — the only piece that talks to an LLM#

A Policy has one method: step(task, scratch) → str. Given the task and the running transcript, return the next ReAct-format string. We ship two:

  • RecordedPolicy — replays the recorded claude-haiku-4-5 outputs in the fixture, one round per call. Used for deterministic scoring.

  • AnthropicPolicy — calls the real Anthropic API in LIVE mode with the same system prompt the fixtures were recorded with.

The parser, tools, loop, and scoring don’t change between the two — the only difference is where the strings come from.

class Policy:
    def step(self, prompt: str, scratch: str) -> str:
        raise NotImplementedError


class RecordedPolicy(Policy):
    """Replays per-task assistant outputs from the fixture, one per call."""

    def __init__(self, trajectories: list[dict]) -> None:
        self._by_task = {t["task"]: list(t["rounds"]) for t in trajectories}
        self._cursor: dict[str, int] = {}

    def step(self, prompt: str, scratch: str) -> str:
        rounds = self._by_task.get(prompt, [])
        i = self._cursor.get(prompt, 0)
        self._cursor[prompt] = i + 1
        if i < len(rounds):
            return rounds[i]
        # Out of recorded rounds — return a parse-error sentinel; the loop
        # will count it but won't crash.
        return "Thought: out of recorded rounds."


class AnthropicPolicy(Policy):
    """Calls the real Anthropic API with the same system prompt as the fixture."""

    def __init__(self, model: str, system_prompt: str) -> None:
        self.model = model
        self.system = system_prompt
        import anthropic  # noqa: PLC0415
        self._client = anthropic.Anthropic()

    def step(self, prompt: str, scratch: str) -> str:
        resp = self._client.messages.create(
            model=self.model, max_tokens=200, system=self.system,
            messages=[{"role": "user", "content": prompt + "\n" + scratch}],
        )
        return "".join(b.text for b in resp.content if b.type == "text")


policy = AnthropicPolicy(MODEL, FIXTURE["system_prompt"]) if LIVE else RecordedPolicy(FIXTURE["trajectories"])
print(f"using {type(policy).__name__}")

Step 4 - The ReAct loop#

Six lines of actual logic: call policy, parse output, dispatch to tool, append observation, stop if final answer or max steps. Everything else is bookkeeping.

@dataclass
class EpisodeResult:
    task: str
    expected: str
    final: str | None
    steps: int
    parse_errors: int
    transcript: str


def run_episode(task: str, expected: str, policy: Policy, max_steps: int = 6) -> EpisodeResult:
    scratch = ""
    parse_errors = 0
    for step in range(max_steps):
        out = policy.step(task, scratch)
        parsed = parse_llm_output(out)

        if parsed.is_final:
            return EpisodeResult(task, expected, parsed.final_answer, step + 1, parse_errors, scratch + out)

        if not parsed.is_action:
            parse_errors += 1
            scratch += f"\n{out}\nObservation: parse error; expected an Action or Final Answer.\n"
            continue

        tool = TOOLS.get(parsed.action)
        if tool is None:
            parse_errors += 1
            scratch += f"\n{out}\nObservation: unknown tool {parsed.action!r}\n"
            continue

        obs = tool(parsed.action_input or "")
        scratch += f"\n{out}\nObservation: {obs}\n"

    return EpisodeResult(task, expected, None, max_steps, parse_errors, scratch)


def grade(result: EpisodeResult) -> bool:
    if result.final is None: return False
    return result.expected.lower() in result.final.lower()

Step 5 - 20 tasks across three kinds of queries#

Eight arithmetic, eight lookup, four temporal. Ground truth for each is a single substring that should appear in the final answer.

Three metrics:

  • Success rate - fraction of tasks solved correctly. Floor: 70 %.

  • Parse-error rate - fraction of LLM outputs that didn’t follow the format. Ceiling: 10 %.

  • Avg steps per task - how many loop iterations it takes on average. Ceiling: 5. (If your agent is bouncing around in 10+ steps per task, it’s wandering.)

TASKS: list[tuple[str, str]] = [(t["task"], t["expected"]) for t in FIXTURE["trajectories"]]

results = [run_episode(task, expected, policy) for task, expected in TASKS]

n_success = sum(grade(r) for r in results)
total_parse_errors = sum(r.parse_errors for r in results)
avg_steps = sum(r.steps for r in results) / len(results)
success_rate = n_success / len(results)
parse_err_rate = total_parse_errors / max(sum(r.steps for r in results), 1)

for r in results[:5]:
    print(f"{r.task:<45}  expected={r.expected!r:<12}  final={r.final!r}")
print(f"\nsuccess {n_success}/{len(results)} ({success_rate:.0%})  "
      f"avg steps {avg_steps:.2f}  parse errors {parse_err_rate:.1%}")

s.check("success_rate_at_least_70pct", lambda: success_rate >= 0.70, msg=f"{success_rate:.2%}")
s.check("parse_error_rate_at_most_10pct", lambda: parse_err_rate <= 0.10, msg=f"{parse_err_rate:.2%}")
s.check("avg_steps_at_most_5", lambda: avg_steps <= 5.0, msg=f"{avg_steps:.2f}")
s.check("every_task_has_finite_trace", lambda: all(r.steps >= 1 for r in results))
s.check(
    "parser_extracts_action_from_well_formed_output",
    lambda: parse_llm_output("Thought: x\nAction: calculator[2+2]").action == "calculator",
)

What the numbers look like#

Two views of the same 20 episodes. On the left, the step-budget sweep shows how many tasks would have solved if we capped the loop earlier; the curve plateaus once every well-formed trajectory has completed. On the right, a stacked histogram of trajectory lengths split by outcome makes it obvious whether failures are concentrated at the step cap (wandering) or at short lengths (parse errors). Both are derived from results - no extra episodes run.

import matplotlib.pyplot as plt

outcomes = {"success": [], "failure": []}
for r in results:
    outcomes["success" if grade(r) else "failure"].append(r.steps)

budgets = list(range(1, 7))
rates = [sum(1 for r in results if grade(r) and r.steps <= b) / len(results) for b in budgets]

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(9, 3.2))
ax1.plot(budgets, rates, marker="o", color="tab:blue")
ax1.axhline(success_rate, color="tab:gray", linestyle=":", label=f"final {success_rate:.0%}")
ax1.set_xlabel("allowed step budget")
ax1.set_ylabel("success rate")
ax1.set_ylim(0, 1.05)
ax1.set_title("success rate vs step budget")
ax1.legend()
ax1.grid(True, alpha=0.3)

bins = range(1, max(r.steps for r in results) + 2)
ax2.hist([outcomes["success"], outcomes["failure"]], bins=bins,
         stacked=True, color=["tab:green", "tab:red"], label=["success", "failure"])
ax2.set_xlabel("steps taken")
ax2.set_ylabel("task count")
ax2.set_title(f"trajectory length per outcome (avg {avg_steps:.2f})")
ax2.legend()
fig.tight_layout()
plt.show()

Exercises#

  1. Live mode. Set ANTHROPIC_API_KEY and rerun. Watch how the parse- error rate changes when the model improvises around the format.

  2. A new tool. Add a search_pypi(package) function and route it from the policy. (For LIVE mode, update the system prompt’s tool list.)

  3. Force format with constrained decoding. 02_structured_outputs_three_ways shows how grammar-constrained decoding eliminates the parse-error class entirely; 08_production/03_tool_use_agent shows the production equivalent via native tool_use.

  4. Multi-hop queries. Add “What is the population of the capital of Japan?” — what state does the policy need to chain two tool calls?

Further reading#

  • Yao et al. 2022, ReAct: Synergizing Reasoning and Acting in Language Models (arxiv 2210.03629).

  • Schick et al. 2023, Toolformer.

  • Anthropic 2024, Building Effective Agents.

Next#

02_structured_outputs_three_ways — same agent, but the model is forced to emit valid JSON via grammar-constrained decoding, validate+retry, or native function calling.

s.summary()
s.save()