Loops and stop rules¶
A loop decides the steps of one turn and when to stop. To write one, see Write your own loop. This page has the rules.
When a loop stops¶
state.stopped says why the run is set to stop. It is None until something decides:
| Who decides | When state.stopped is set |
state.stopped |
|---|---|---|
state.finish() |
Right away, when it is called | StoppedByFinish(answer) |
A permission denies a tool call with stop=True |
Right away, when use_tools records the results |
StoppedByPermission(call, permission) |
An until function returns True |
When the loop checks it, before a turn | StoppedByUntil(name), for example StoppedByUntil("waiting_for_user") |
limit turns have run in this call |
When the loop checks it, before a turn | StoppedByLimit(turns), where turns is the limit |
The default loop stops on waiting_for_user.
Before every turn, the loop stops if state.stopped is already set. Otherwise it checks the until functions, then
limit. Each stop is a StopEntry in state.history.
- The checks run before each turn, so an
untilfunction sees the State after the previous turn. - Reaching
limitdoes not raise. Checkisinstance(state.stopped, StoppedByLimit)when you need to know. runrecords aRunStartEntry, which resetsstate.stoppedtoNone. After a run that ended with an exception, it isNonetoo.state.finish(answer)ends every loop, and it can be called from the loop body or from a tool. It ends the State for good: see State and history. A permission'sstop=Trueends every loop too.
until¶
until takes a function from State to bool, or a list of them. The loop stops when any of them returns True.
from alpineagents import Agent, State, loop, waiting_for_user
def spent_too_much(state: State) -> bool:
return state.usage.output_tokens > 20_000
@loop(until=[waiting_for_user, spent_too_much], limit=30)
def careful(agent: Agent, state: State):
agent.think(state)
if state.pending_calls:
agent.use_tools(state)
agent = Agent(model="claude-sonnet-5", loop=careful)
quick = careful.copy(limit=5)
- Pass the function itself:
until=waiting_for_user. Notwaiting_for_user(state). - Use a named function. Its name becomes
state.stopped.name. Alambdaworks, but warns because it has no useful name. - An
untilfunction that reads several values can takesnap = state.snapshot()first, so they all belong to one moment even if another thread adds a message in between. copyreturns a new loop with differentuntilorlimit. The original does not change.
Loops without @loop¶
A loop is any function that takes (agent, state). One written without @loop makes none of the checks above, so
check state.stopped yourself:
from alpineagents import Agent, State, waiting_for_user
def my_loop(agent: Agent, state: State):
while state.stopped is None and not waiting_for_user(state):
agent.think(state)
if state.pending_calls:
agent.use_tools(state)
agent = Agent(model="claude-sonnet-5", loop=my_loop)
Nested loops¶
A loop inside another loop: its until or limit ends only the inner loop. The outer @loop checks its own, and if
it goes on, state.stopped goes back to None. That is a StopEntry with content=None, which history shows as a
stop that was cleared. finish() and a permission's stop end every loop.
A loop written without @loop does not clear what a @loop inside it leaves behind. After the inner loop stops on
its own until or limit, state.stopped stays set, so state.stopped is None would end the outer loop too. When
it calls a @loop, check only the stops that end every loop, or write the outer loop with @loop as well:
from alpineagents import Agent, State, StoppedByFinish, StoppedByPermission, loop, tool, waiting_for_user
@tool
def look(topic: str) -> str:
"""Look something up"""
return f"Notes on {topic}"
@loop(until=waiting_for_user, limit=5)
def research(agent: Agent, state: State):
agent.think(state)
if state.pending_calls:
agent.use_tools(state)
def three_rounds(agent: Agent, state: State):
for _ in range(3):
# Not `state.stopped is None`: research may have left StoppedByLimit(5) there.
if isinstance(state.stopped, (StoppedByFinish, StoppedByPermission)) or waiting_for_user(state):
return
research(agent, state)
agent = Agent(model="claude-sonnet-5", tools=[look], loop=three_rounds)
print(agent.run("Research how the cache works"))
Rules¶
| Rule | Detail |
|---|---|
The body must return None |
A returned value raises TypeError. Set the answer with state.finish(answer) |
| Each call of the loop counts turns from zero | state.turn keeps counting across calls, the loop's count does not |
Exceptions from the body or an until function propagate |
Nothing is swallowed. See Errors and interruptions |
An async def body makes an async loop |
Run it with agent.arun. See Use async |
A block is a function that takes (agent, state) and does one step of a turn, such as compact_if_full. See
Write your own loop.