Problem 119 · Agents

Minimal Agent Loop with Budgets and Stop Reasons

MediumAI engineering

01Problem

Write `run_agent(model, tools, task, max_steps, max_tokens, max_consecutive_errors, max_repeats)`. Each step calls `model(transcript)`, which returns `{"type": "final", "content": ...}` or `{"type": "tool", "name": ..., "args": {...}}` plus an optional `usage` token count. Run tools and append their output (or the exception text, flagged `is_error`) to the transcript. Stop with an explicit reason: `final`, `max_steps`, `token_budget`, `repeated_call` (the same tool with identical arguments more than `max_repeats` times) or `tool_errors` (more than `max_consecutive_errors` failures in a row).

02Key insight

The loop itself is ten lines; the engineering is in how it stops. Every unbounded dimension (steps, tokens, identical retries, error streaks) needs a budget, and every exit needs a named reason so dashboards can tell 'answered' from 'gave up' from 'looped'. Tool exceptions become observations instead of crashes, because the model can often recover (fix the argument, pick another tool), but a streak of errors means it cannot. Canonicalising arguments (sorted-key JSON) is what makes loop detection actually fire.

03Solution

solution.pypython46 lines

import json
from collections import Counter
from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class AgentResult:
    stop_reason: str
    output: str | None
    steps: int
    transcript: list[dict] = field(default_factory=list)


def run_agent(model: Callable[[list[dict]], dict], tools: dict[str, Callable[..., Any]], task: str,
              max_steps: int = 8, max_tokens: int = 10_000, max_consecutive_errors: int = 2,
              max_repeats: int = 2) -> AgentResult:
    transcript: list[dict] = [{"role": "user", "content": task}]
    tokens = 0
    error_streak = 0
    calls: Counter[tuple[str, str]] = Counter()
    for step in range(1, max_steps + 1):
        action = model(transcript)
        tokens += int(action.get("usage", 0))
        if tokens > max_tokens:
            return AgentResult("token_budget", None, step, transcript)
        if action["type"] == "final":
            transcript.append({"role": "assistant", "content": action["content"]})
            return AgentResult("final", action["content"], step, transcript)
        name, args = action["name"], action.get("args", {})
        signature = (name, json.dumps(args, sort_keys=True))
        calls[signature] += 1
        if calls[signature] > max_repeats:
            return AgentResult("repeated_call", None, step, transcript)
        transcript.append({"role": "assistant", "tool": name, "args": args})
        try:
            if name not in tools:
                raise LookupError(f"unknown tool {name}")
            observation, is_error = str(tools[name](**args)), False
        except Exception as e:
            observation, is_error = f"{type(e).__name__}: {e}", True
        transcript.append({"role": "tool", "name": name, "content": observation, "is_error": is_error})
        error_streak = error_streak + 1 if is_error else 0
        if error_streak > max_consecutive_errors:
            return AgentResult("tool_errors", None, step, transcript)
    return AgentResult("max_steps", None, max_steps, transcript)

Try it first.

TimeO(max_steps) model and tool calls

SpaceO(transcript length)

tests.py · 8 passing

tests.pypython27 lines

tools = {"add": lambda a, b: a + b, "div": lambda a, b: a / b, "search": lambda q: f"results for {q}"}
def solver(tr):
    obs = [m for m in tr if m["role"] == "tool"]
    if not obs:
        return {"type": "tool", "name": "add", "args": {"b": 3, "a": 2}}
    return {"type": "final", "content": obs[-1]["content"]}
r = run_agent(solver, tools, "2+3?")
assert (r.stop_reason, r.output, r.steps) == ("final", "5", 2)
r = run_agent(lambda tr: {"type": "tool", "name": "search", "args": {"q": "x"}}, tools, "loop")
assert r.stop_reason == "repeated_call" and r.steps == 3
r = run_agent(lambda tr: {"type": "tool", "name": "search", "args": {"q": len(tr)}}, tools, "wander", max_steps=5)
assert r.stop_reason == "max_steps" and r.steps == 5
r = run_agent(lambda tr: {"type": "tool", "name": "div", "args": {"a": len(tr), "b": 0}}, tools, "fail")
assert r.stop_reason == "tool_errors" and r.steps == 3
assert all(m["is_error"] for m in r.transcript if m["role"] == "tool")
r = run_agent(lambda tr: {"type": "tool", "name": "search", "args": {"q": len(tr)}, "usage": 4000}, tools, "spend")
assert r.stop_reason == "token_budget" and r.steps == 3
def recovers(tr):
    obs = [m for m in tr if m["role"] == "tool"]
    if not obs:
        return {"type": "tool", "name": "nope", "args": {}}
    if obs[-1]["is_error"]:
        return {"type": "tool", "name": "div", "args": {"a": 6, "b": 3}}
    return {"type": "final", "content": obs[-1]["content"]}
r = run_agent(recovers, tools, "recover")
assert (r.stop_reason, r.output) == ("final", "2.0")
assert "LookupError" in r.transcript[2]["content"]

04Staff-level follow-up

What changes in production? Persist the transcript after every step so a crashed worker resumes instead of re-running side effects, give each tool call an idempotency key, add wall-clock and dollar budgets, and compact or summarise old observations when the transcript nears the context limit. Measure the stop-reason distribution per release: a rise in `max_steps` or `repeated_call` is the earliest regression signal you will get.