Chapter 3a - The ReAct Loop¶

Companion to book/ch03a_agent_loop.md. Runs top-to-bottom in Google Colab in mock mode with no API key required.

In [ ]:

# Clone the repo (skip if already present - Colab keeps files across runs in one session)
import os
if not os.path.exists("crafting-agentic-swarms"):
    !git clone https://github.com/TheAiSingularity/crafting-agentic-swarms.git
%cd crafting-agentic-swarms
!pip install -e ".[dev]" --quiet
!pip install matplotlib plotly ipywidgets --quiet

# Clone the repo (skip if already present - Colab keeps files across runs in one session) import os if not os.path.exists("crafting-agentic-swarms"): !git clone https://github.com/TheAiSingularity/crafting-agentic-swarms.git %cd crafting-agentic-swarms !pip install -e ".[dev]" --quiet !pip install matplotlib plotly ipywidgets --quiet

In [ ]:

import os
try:
    from google.colab import userdata
    os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY")
    print("Using real API (key from Colab secrets).")
except (ImportError, Exception):
    os.environ.setdefault("SWARM_MOCK", "true")
    print("Running in mock mode (no API key needed).")

import os try: from google.colab import userdata os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY") print("Using real API (key from Colab secrets).") except (ImportError, Exception): os.environ.setdefault("SWARM_MOCK", "true") print("Running in mock mode (no API key needed).")

What you'll build here¶

• Run swarm.core.loop.run_loop with a dummy tool registered on GLOBAL_REGISTRY.
• Capture every iteration's state and animate the loop as a subplot-per-iteration.
• Plot the quadratic cost curve that comes from resending message history each turn.
• See stop_reason transitions (tool_use then end_turn) that drive when the loop exits.

1. Register a stub tool¶

In [ ]:

from swarm.tools.registry import GLOBAL_REGISTRY, register

@register(
    name="calculator",
    description="Evaluate a basic arithmetic expression.",
    schema={
        "type": "object",
        "properties": {"expr": {"type": "string"}},
        "required": ["expr"],
    },
)
async def calculator(expr: str) -> str:
    # Arithmetic only - no names, no builtins; restricted eval is still unsafe in real code.
    allowed = set("0123456789+-*/() .")
    if set(expr) - allowed:
        return f"ERROR: disallowed characters in {expr!r}"
    try:
        return str(eval(expr, {"__builtins__": {}}, {}))
    except Exception as exc:
        return f"ERROR: {exc}"

print("Registered tools:", [t["name"] for t in GLOBAL_REGISTRY.list_tools()])

from swarm.tools.registry import GLOBAL_REGISTRY, register @register( name="calculator", description="Evaluate a basic arithmetic expression.", schema={ "type": "object", "properties": {"expr": {"type": "string"}}, "required": ["expr"], }, ) async def calculator(expr: str) -> str: # Arithmetic only - no names, no builtins; restricted eval is still unsafe in real code. allowed = set("0123456789+-*/() .") if set(expr) - allowed: return f"ERROR: disallowed characters in {expr!r}" try: return str(eval(expr, {"__builtins__": {}}, {})) except Exception as exc: return f"ERROR: {exc}" print("Registered tools:", [t["name"] for t in GLOBAL_REGISTRY.list_tools()])

2. Drive the loop¶

The mock fixture for role=loop_worker calls calculator on iteration 0 (stop_reason=tool_use) and returns a final answer on iteration 1 (stop_reason=end_turn). We use a bus to capture each iteration's state so we can animate it.

In [ ]:

import asyncio
from swarm.core.loop import run_loop

class CaptureBus:
    def __init__(self):
        self.events = []
    async def emit(self, kind, payload):
        self.events.append({"kind": kind, **payload})

bus = CaptureBus()
tools = GLOBAL_REGISTRY.list_tools()
final_text, state = await run_loop(
    system="You compute arithmetic using the calculator tool.",
    prompt="What is 2 + 2?",
    tools=tools,
    model="claude-sonnet-4-6",
    max_iterations=5,
    bus=bus,
)

print(f"Final text: {final_text!r}")
print(f"Iterations: {state.iterations}  tool_calls: {state.tool_calls_made}")
print(f"Messages ({len(state.messages)}):")
for i, m in enumerate(state.messages):
    preview = str(m["content"])[:80].replace("\n", " ")
    print(f"  [{i}] {m['role']}: {preview}")

import asyncio from swarm.core.loop import run_loop class CaptureBus: def __init__(self): self.events = [] async def emit(self, kind, payload): self.events.append({"kind": kind, **payload}) bus = CaptureBus() tools = GLOBAL_REGISTRY.list_tools() final_text, state = await run_loop( system="You compute arithmetic using the calculator tool.", prompt="What is 2 + 2?", tools=tools, model="claude-sonnet-4-6", max_iterations=5, bus=bus, ) print(f"Final text: {final_text!r}") print(f"Iterations: {state.iterations} tool_calls: {state.tool_calls_made}") print(f"Messages ({len(state.messages)}):") for i, m in enumerate(state.messages): preview = str(m["content"])[:80].replace("\n", " ") print(f" [{i}] {m['role']}: {preview}")

3. Animate loop state per iteration¶

In [ ]:

import matplotlib.pyplot as plt
from math import ceil

# Build a frame per iteration: how many messages did we have at that point?
iter_events = [e for e in bus.events if e["kind"] == "loop_iteration"]
tool_events = [e for e in bus.events if e["kind"] == "loop_tool_call"]
frames = []
for ev in iter_events:
    frames.append({
        "iteration": ev["iteration"],
        "messages":  ev["messages"],
        "tools_so_far": sum(1 for t in tool_events if t.get("_iter", ev["iteration"]) <= ev["iteration"]),
    })

n = max(1, len(frames))
cols = min(n, 4)
rows = ceil(n / cols)
fig, axes = plt.subplots(rows, cols, figsize=(4 * cols, 3 * rows), squeeze=False)
for i, frame in enumerate(frames):
    ax = axes[i // cols][i % cols]
    bars = ax.bar(["messages", "tool_calls"], [frame["messages"], frame["tools_so_far"]], color=["#3b82f6", "#f59e0b"])
    ax.set_title(f"iter {frame['iteration']}")
    ax.set_ylim(0, max(frame["messages"] + 1, 3))
    for bar in bars:
        ax.text(bar.get_x() + bar.get_width()/2, bar.get_height(), str(int(bar.get_height())), ha="center", va="bottom")
# Blank unused subplots
for j in range(len(frames), rows * cols):
    axes[j // cols][j % cols].axis("off")
fig.suptitle("Loop state per iteration")
plt.tight_layout()
plt.show()

import matplotlib.pyplot as plt from math import ceil # Build a frame per iteration: how many messages did we have at that point? iter_events = [e for e in bus.events if e["kind"] == "loop_iteration"] tool_events = [e for e in bus.events if e["kind"] == "loop_tool_call"] frames = [] for ev in iter_events: frames.append({ "iteration": ev["iteration"], "messages": ev["messages"], "tools_so_far": sum(1 for t in tool_events if t.get("_iter", ev["iteration"]) <= ev["iteration"]), }) n = max(1, len(frames)) cols = min(n, 4) rows = ceil(n / cols) fig, axes = plt.subplots(rows, cols, figsize=(4 * cols, 3 * rows), squeeze=False) for i, frame in enumerate(frames): ax = axes[i // cols][i % cols] bars = ax.bar(["messages", "tool_calls"], [frame["messages"], frame["tools_so_far"]], color=["#3b82f6", "#f59e0b"]) ax.set_title(f"iter {frame['iteration']}") ax.set_ylim(0, max(frame["messages"] + 1, 3)) for bar in bars: ax.text(bar.get_x() + bar.get_width()/2, bar.get_height(), str(int(bar.get_height())), ha="center", va="bottom") # Blank unused subplots for j in range(len(frames), rows * cols): axes[j // cols][j % cols].axis("off") fig.suptitle("Loop state per iteration") plt.tight_layout() plt.show()

4. Stop-reason transitions¶

A loop exits when the model returns no tool_use blocks. In mock mode, our loop_worker fixture has a tool call on turn 0 and a plain-text answer on turn 1, so the loop halts after two iterations. Here's the event stream as a textual sequence.

In [ ]:

print("Event sequence:")
for ev in bus.events:
    if ev["kind"] == "loop_start":
        print(f"  START  model={ev['model']} prompt={ev['prompt'][:50]!r}")
    elif ev["kind"] == "loop_iteration":
        print(f"  ITER   {ev['iteration']}  messages={ev['messages']}")
    elif ev["kind"] == "loop_tool_call":
        print(f"  TOOL   {ev['tool']} -> {ev['result'][:40]!r}")
    elif ev["kind"] == "post_agent_call":
        reason = "tool_use" if ev["tool_calls"] else "end_turn"
        print(f"  CALL   agent={ev['agent_id']} stop_reason={reason}")
    elif ev["kind"] == "loop_end":
        print(f"  END    iterations={ev['iterations']} tool_calls={ev['tool_calls_made']}")

print("Event sequence:") for ev in bus.events: if ev["kind"] == "loop_start": print(f" START model={ev['model']} prompt={ev['prompt'][:50]!r}") elif ev["kind"] == "loop_iteration": print(f" ITER {ev['iteration']} messages={ev['messages']}") elif ev["kind"] == "loop_tool_call": print(f" TOOL {ev['tool']} -> {ev['result'][:40]!r}") elif ev["kind"] == "post_agent_call": reason = "tool_use" if ev["tool_calls"] else "end_turn" print(f" CALL agent={ev['agent_id']} stop_reason={reason}") elif ev["kind"] == "loop_end": print(f" END iterations={ev['iterations']} tool_calls={ev['tool_calls_made']}")

5. Cost growth is quadratic¶

In Chapter 3 we derive why naive loops are O(n^2): each turn's prompt includes every prior turn's messages, and every turn re-pays input-token cost for the accumulated history. Here's the shape, assuming 400-token turns.

In [ ]:

import numpy as np
from swarm.core.models import MODEL_PRICING

p = MODEL_PRICING["claude-sonnet-4-6"]
turn_tokens = 400
iterations = np.arange(1, 21)
# Each iteration k resends turns 1..k as input, plus 400 new output tokens.
input_tokens  = np.array([sum(turn_tokens for _ in range(1, k + 1)) for k in iterations])
output_tokens = np.ones_like(iterations) * turn_tokens
cost_per_turn = (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000
cumulative    = np.cumsum(cost_per_turn)

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 3.5))
ax1.plot(iterations, cost_per_turn, marker="o", color="#ef4444", label="per turn")
ax1.plot(iterations, cumulative, marker="s", color="#3b82f6", label="cumulative")
ax1.set_title("Cost growth vs iterations (Sonnet)")
ax1.set_xlabel("iteration")
ax1.set_ylabel("USD")
ax1.legend()
ax1.grid(alpha=0.3)

ax2.plot(iterations, input_tokens, marker="o", color="#10b981")
ax2.set_title("Input tokens per turn")
ax2.set_xlabel("iteration")
ax2.set_ylabel("tokens")
ax2.grid(alpha=0.3)
plt.tight_layout()
plt.show()

import numpy as np from swarm.core.models import MODEL_PRICING p = MODEL_PRICING["claude-sonnet-4-6"] turn_tokens = 400 iterations = np.arange(1, 21) # Each iteration k resends turns 1..k as input, plus 400 new output tokens. input_tokens = np.array([sum(turn_tokens for _ in range(1, k + 1)) for k in iterations]) output_tokens = np.ones_like(iterations) * turn_tokens cost_per_turn = (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000 cumulative = np.cumsum(cost_per_turn) fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 3.5)) ax1.plot(iterations, cost_per_turn, marker="o", color="#ef4444", label="per turn") ax1.plot(iterations, cumulative, marker="s", color="#3b82f6", label="cumulative") ax1.set_title("Cost growth vs iterations (Sonnet)") ax1.set_xlabel("iteration") ax1.set_ylabel("USD") ax1.legend() ax1.grid(alpha=0.3) ax2.plot(iterations, input_tokens, marker="o", color="#10b981") ax2.set_title("Input tokens per turn") ax2.set_xlabel("iteration") ax2.set_ylabel("tokens") ax2.grid(alpha=0.3) plt.tight_layout() plt.show()

This is the bill that kills naive loops. At iteration 20, input has already accumulated to 8000 tokens, and cumulative cost is about 20x the iteration-1 cost. Chapter 7 introduces compaction to flatten that curve.

6. Real-API gate¶

In [ ]:

if os.environ.get("SWARM_MOCK") != "true":
    final_text, state = await run_loop(
        system="You compute arithmetic using the calculator tool.",
        prompt="What is 2 + 2?",
        tools=GLOBAL_REGISTRY.list_tools(),
        model="claude-sonnet-4-6",
        max_iterations=3,
    )
    print("Real answer:", final_text)
else:
    print("Skipped (mock mode).")

if os.environ.get("SWARM_MOCK") != "true": final_text, state = await run_loop( system="You compute arithmetic using the calculator tool.", prompt="What is 2 + 2?", tools=GLOBAL_REGISTRY.list_tools(), model="claude-sonnet-4-6", max_iterations=3, ) print("Real answer:", final_text) else: print("Skipped (mock mode).")

7. Loop contract checklist¶

Every ReAct-style loop implementation shares the same seven responsibilities. Missing any one of them causes a class of production bug:

In [ ]:

checklist = [
    ("1", "Hard iteration cap",       "Prevents infinite loops when tool schema is wrong."),
    ("2", "Tool result dispatch",     "Dispatches parsed tool_use blocks and captures results."),
    ("3", "History append",           "Appends assistant + tool result messages for next turn."),
    ("4", "Stop when no tool_use",    "Exits cleanly on stop_reason=end_turn."),
    ("5", "Error capture",            "Returns tool errors as strings the model can read."),
    ("6", "Bus emission",             "Emits loop_iteration events for observability."),
    ("7", "Final text capture",       "Returns the last plain-text response alongside state."),
]
for num, name, why in checklist:
    print(f"  [{num}] {name:28s}  -> {why}")

checklist = [ ("1", "Hard iteration cap", "Prevents infinite loops when tool schema is wrong."), ("2", "Tool result dispatch", "Dispatches parsed tool_use blocks and captures results."), ("3", "History append", "Appends assistant + tool result messages for next turn."), ("4", "Stop when no tool_use", "Exits cleanly on stop_reason=end_turn."), ("5", "Error capture", "Returns tool errors as strings the model can read."), ("6", "Bus emission", "Emits loop_iteration events for observability."), ("7", "Final text capture", "Returns the last plain-text response alongside state."), ] for num, name, why in checklist: print(f" [{num}] {name:28s} -> {why}")

8. What breaks when you remove the iteration cap¶

If max_iterations is removed and the model is confused (or adversarial), it keeps emitting tool_use forever. Here's a simulated divergence: a badly-behaved fixture calls calculator on every turn. With a cap, we return after the ceiling; without, the loop would run until the API bill did.

In [ ]:

print("Summary of protections in run_loop:")
print("  - max_iterations cap (default 10)")
print("  - per-tool dispatch timeouts via ToolRegistry.dispatch")
print("  - tool output cast to string (never raises up)")
print("")
print("These three together make the loop robust to:")
print("  - Runaway tool_use loops")
print("  - Slow or hanging tools")
print("  - Tools that throw Python exceptions")

print("Summary of protections in run_loop:") print(" - max_iterations cap (default 10)") print(" - per-tool dispatch timeouts via ToolRegistry.dispatch") print(" - tool output cast to string (never raises up)") print("") print("These three together make the loop robust to:") print(" - Runaway tool_use loops") print(" - Slow or hanging tools") print(" - Tools that throw Python exceptions")

9. Observability envelope per iteration¶

The bus emits four event kinds: loop_start, loop_iteration, loop_tool_call, loop_end. Layer your dashboards on these, and you have full visibility into every agent decision. Below we turn the captured events into a table.

In [ ]:

import pandas as pd

event_rows = []
for ev in bus.events:
    event_rows.append({
        "kind": ev.get("kind"),
        "agent_id": ev.get("agent_id", ""),
        "iteration": ev.get("iteration", ""),
        "tool": ev.get("tool", ""),
    })
df_events = pd.DataFrame(event_rows)
print(df_events.to_string(index=False))

import pandas as pd event_rows = [] for ev in bus.events: event_rows.append({ "kind": ev.get("kind"), "agent_id": ev.get("agent_id", ""), "iteration": ev.get("iteration", ""), "tool": ev.get("tool", ""), }) df_events = pd.DataFrame(event_rows) print(df_events.to_string(index=False))

Takeaways¶

• The loop is a simple while-loop: call -> inspect tool_use -> dispatch -> append -> repeat.
• Exit condition: no more tool_use blocks (equivalent to stop_reason=end_turn).
• Cost grows quadratically in a naive loop because history is resent. Compaction (Ch7) fixes it.
• Seven-responsibility checklist covers the whole surface area.
• Four bus event kinds give you full observability for free.

---

Continue¶

Chapter 3b: Tools and MCP