Agent Loop Control: Step Budgets, Loop Detection, and Stop Conditions That Actually Work
Most runaway agent bills come from one missing piece: a loop with no brakes. Here's how to add step, token, and wall-clock budgets, detect repeated tool calls, and fail gracefully, with working Python.
Every agent is, at its core, a while loop: call the model, run the tool it asked for, feed the result back, repeat. The model decides when to stop. That's the problem. A model that is confused, rate-limited into empty tool results, or stuck retrying the same failing query will happily keep going until something external kills it, usually your cloud bill or a user closing the tab.
This post is about the brakes. We'll build a small loop controller in Python that enforces budgets, detects repetition, and ends runs in a way you can debug afterwards.
The problem: three failure modes of unbounded loops
- The infinite retry. A tool returns an error. The model re-issues the identical call. Forever.
- The slow drift. The agent makes "progress" (each step is different) but never converges. Context grows, cost per step rises, and step 60 costs ten times step 6.
- The ping-pong. Two tools, or two sub-agents, hand work back and forth. Each step looks valid in isolation.
A simple max_iterations=10 stops the first and third eventually, but it's a blunt instrument. It also truncates legitimate long tasks, and it gives you no information about why the run ended.
Explanation: layers of control
Good loop control has four independent layers. Each catches something the others miss.
| Layer | What it limits | Catches | Typical default |
|---|---|---|---|
| Step budget | Number of model calls | Runaway loops | 15โ40 depending on task |
| Token/cost budget | Cumulative tokens or dollars | Context-bloat drift | A hard $ cap per run |
| Wall-clock deadline | Elapsed seconds | Slow tools, hung calls | 60โ300 s for interactive, longer for batch |
| Repetition detector | Identical or cyclic tool calls | Retry loops, ping-pong | 3 identical calls, or a repeating cycle |
The key design principle: the controller lives outside the model. Never rely on the prompt saying "stop after 10 steps". Models miscount, and a prompt injection can talk them out of it. The loop code enforces limits; the prompt merely informs the model about them so it can wrap up gracefully.
Code: a loop controller
Below is a provider-agnostic controller. The call_model and run_tool functions are placeholders for your SDK of choice; the control logic is the point.
import hashlib
import json
import time
from dataclasses import dataclass, field
from enum import Enum
class StopReason(str, Enum):
DONE = "done"
MAX_STEPS = "max_steps"
MAX_TOKENS = "max_tokens"
DEADLINE = "deadline"
REPETITION = "repetition"
TOOL_ERRORS = "consecutive_tool_errors"
@dataclass
class Limits:
max_steps: int = 25
max_total_tokens: int = 150_000
deadline_seconds: float = 120.0
max_identical_calls: int = 3
max_consecutive_errors: int = 4
cycle_window: int = 6 # look at the last N calls for A,B,A,B patterns
@dataclass
class RunState:
started: float = field(default_factory=time.monotonic)
steps: int = 0
tokens: int = 0
consecutive_errors: int = 0
call_fingerprints: list[str] = field(default_factory=list)
def fingerprint(tool_name: str, args: dict) -> str:
# sort_keys makes {"a":1,"b":2} and {"b":2,"a":1} identical
payload = json.dumps([tool_name, args], sort_keys=True, default=str)
return hashlib.sha256(payload.encode()).hexdigest()[:16]
def detect_repetition(fps: list[str], limits: Limits) -> bool:
# 1) the same call N times in a row
n = limits.max_identical_calls
if len(fps) >= n and len(set(fps[-n:])) == 1:
return True
# 2) a short cycle, e.g. A,B,A,B,A,B
w = limits.cycle_window
tail = fps[-w:]
if len(tail) == w:
for period in (2, 3):
if w % period == 0 and tail == tail[:period] * (w // period):
return True
return False
def check_budgets(state: RunState, limits: Limits) -> StopReason | None:
if state.steps >= limits.max_steps:
return StopReason.MAX_STEPS
if state.tokens >= limits.max_total_tokens:
return StopReason.MAX_TOKENS
if time.monotonic() - state.started >= limits.deadline_seconds:
return StopReason.DEADLINE
if state.consecutive_errors >= limits.max_consecutive_errors:
return StopReason.TOOL_ERRORS
if detect_repetition(state.call_fingerprints, limits):
return StopReason.REPETITION
return None
def run_agent(task: str, limits: Limits = Limits()):
state = RunState()
messages = [{"role": "user", "content": task}]
while True:
reason = check_budgets(state, limits)
if reason:
return finalize(messages, reason, state)
response = call_model(messages) # your SDK call
state.steps += 1
state.tokens += response.usage_total_tokens
messages.append(response.as_message())
if not response.tool_calls:
return {"reason": StopReason.DONE, "answer": response.text}
for call in response.tool_calls:
state.call_fingerprints.append(fingerprint(call.name, call.args))
try:
result = run_tool(call.name, call.args)
state.consecutive_errors = 0
except Exception as exc: # tool failures are data for the model
result = f"ERROR: {type(exc).__name__}: {exc}"
state.consecutive_errors += 1
messages.append({"role": "tool", "id": call.id, "content": result})
Notice what happens at the top of each iteration: budgets are checked before spending another model call, not after. Checking afterwards means you always overshoot by one expensive step.
Ending gracefully instead of just dying
When a limit trips, don't return a stack trace. Make one final, tool-free model call asking for a best-effort summary of what was done and what's left. That turns a hard failure into a partial result the user (or a calling system) can act on.
def finalize(messages, reason: StopReason, state: RunState):
wrap_up = {
"role": "user",
"content": (
f"Stopping early ({reason.value}). Do not call any tools. "
"Summarize what you completed, what remains, and any "
"partial results the user can use."
),
}
# tools disabled for this call so the model can't keep looping
response = call_model(messages + [wrap_up], tools=None)
return {
"reason": reason,
"partial_answer": response.text,
"steps": state.steps,
"tokens": state.tokens,
}
Always log the StopReason, step count, tokens, and the last few fingerprints. When you later ask "why did 3% of runs end in repetition?", you'll be able to answer from data instead of guessing.
Telling the model about its budget
Models behave better when they know the constraints. Inject a short status line into each turn:
def budget_hint(state: RunState, limits: Limits) -> str:
left = limits.max_steps - state.steps
return f"[Budget: {left} steps remaining. Prefer finishing over exploring.]"
This is advisory only. The controller still enforces the limit. But in practice it reduces how often you hit the hard stop, because the model starts consolidating as the budget shrinks.
Choosing limits: a quick checklist
- Measure the p95 step count of successful runs on your eval set. Set
max_stepsto roughly 2x that. - Set a dollar cap per run from your unit economics, then derive the token limit from it.
- Set the deadline from the user's patience (interactive) or your queue's visibility timeout (batch).
- Make limits configurable per task type. A code-refactor agent and a FAQ agent shouldn't share them.
- Expose a cancel signal so a user or supervisor can stop a run mid-flight.
- Alert on the rate of each stop reason, not just on errors.
Common pitfalls
Fingerprinting too strictly. If a tool call includes a timestamp or request ID in its arguments, no two calls ever match and your repetition detector is blind. Hash only the semantically meaningful arguments.
Fingerprinting too loosely. Polling a job-status endpoint three times is legitimate. Whitelist idempotent polling tools, or give them their own higher threshold.
Counting steps, ignoring context growth. Ten steps with a 100k-token context cost far more than ten steps with a 5k context. Track tokens, not just iterations. If you use prompt caching, the effective cost differs, but the cap should still be on what you're willing to spend.
Retrying inside the tool and in the loop. Tool-level retries with backoff multiply with the agent's own retries. Decide which layer owns retries. Usually transient network errors belong in the tool, and "this query returned nothing, try another" belongs to the model.
Treating every stop as failure. A max_steps stop with a useful partial answer is often a fine outcome for open-ended research tasks. Define success per stop reason in your evals.
Forgetting sub-agents. If an orchestrator spawns workers, each needs its own limits and the parent needs an aggregate budget. Otherwise a fan-out of 10 workers with 25 steps each is a 250-step run.
A note on frameworks
Most agent frameworks expose some of this (a max-iterations parameter, a recursion limit, a timeout), but the details vary and change between versions. Check your framework's documentation for what it actually enforces, and whether it stops before or after the offending step. Even when you use a framework, wrapping its run call with your own deadline and token accounting is cheap insurance.
Next steps
- Add the
StopReasonenum and logging to your existing agent today. Even without changing any limits, the distribution of stop reasons will tell you where your loops go wrong. - Replay your last 100 production traces through
detect_repetitionoffline. You'll likely find loops you never noticed. - Add eval cases that are designed to loop (a tool that always errors, a search that always returns empty) and assert the agent stops with the right reason and a usable partial answer.
- Pair this with cost attribution so you know which tasks and customers consume the budget.
If you want to build agents with production-grade guardrails like these, hands-on with reviewed code and real projects, take a look at the Joinloop engineering cohort.
Want to build this yourself?
Join a live cohort and go from reading about AI to shipping it, with practitioners who do it daily.