Loops and blocks

Learn: Write your own loop. Concepts: Loops and stop rules. Guide: Stop a run.

loop

loop(fn: Any = None, /, *, until: Any = None, limit: Any = None) -> Any

Turns a one-turn function into a loop that takes (agent, state) and returns the answer.

until and limit are both required, so how the loop stops is always written next to it. The body returns nothing; to set the answer yourself, call state.finish(answer). An async def body makes an async loop for agent.arun.

Example

from alpineagents import waiting_for_user

@loop(until=waiting_for_user, limit=50)
def coding(agent: Agent, state: State):
    compact_if_full(agent, state)
    agent.think(state)
    if state.pending_calls:
        agent.use_tools(state)

Parameters:

Name	Type	Description	Default
until	Any	A State -> bool function or a list of them, checked before every turn.	None
limit	Any	The most turns one run takes. Reaching it stops quietly with state.stopped == StoppedByLimit(limit).	None

Returns:

Type	Description
Any	A decorator that turns the body into a Loop.

Raises:

Type	Description
TypeError	@loop is used bare, without until and limit. See Loop for the other checks.

Loop

Loop(body: Callable[[Any, State], Any], *, until: Any, limit: Any)

A loop made by @loop. Calling loop(agent, state) runs the body turn by turn and returns state.answer.

Before every turn it checks, in order:

1. state.stopped already set by state.finish() (StoppedByFinish) or by a permission's Denied(..., stop=True) (StoppedByPermission(call, ...)): stops and keeps it
2. each until function: stops with state.stopped == StoppedByUntil(name), the function's name
3. the turn count against limit: stops with state.stopped == StoppedByLimit(limit)

A stop the loop decides (2 and 3) is recorded in the history as a StopEntry. Each call counts turns from zero. A nested loop's until/limit stop ends only that loop: the outer loop checks its own until and limit and, if it goes on, records StopEntry(None), so state.stopped goes back to None. Exceptions from the body or the until functions propagate as is. An async loop (async def body) returns a coroutine that does the same; its until functions stay plain functions.

The body must return None: a returned value raises TypeError (set the answer with state.finish(answer)), and so does an awaitable returned by a sync loop. The checks run before each turn, so an until function sees the State after the previous turn. A loop written without @loop makes none of these checks: it tests state.stopped itself, and after a @loop called inside it stopped on its own until or limit (which leaves state.stopped set), it tests only for StoppedByFinish and StoppedByPermission or is itself written with @loop. An until function can read state.snapshot() first so that all the values it reads belong to one moment.

Usually created with @loop. Every check happens here, when the decorator is applied.

Parameters:

Name	Type	Description	Default
body	Callable[[Any, State], Any]	The one-turn function (agent, state) -> None, or an async def one.	required
until	Any	A State -> bool function or a list of them. Pass the function itself, e.g. waiting_for_user. A lambda works but leaves no useful name in state.stopped, so it warns.	required
limit	Any	The most turns, an integer of 1 or more.	required

Raises:

Type	Description
TypeError	until or limit is missing, until got a call result (waiting_for_user(state)), an until item is not callable, or limit is not an integer.
ValueError	limit is less than 1.

body instance-attribute

body: Callable[[Any, State], Any] = body

The one-turn function.

until instance-attribute

until: tuple[Condition, ...] = tuple(checked)

The stop conditions, each a State -> bool function.

limit instance-attribute

limit: int = limit

The most turns one call runs.

is_async instance-attribute

is_async: bool = is_async_callable(body)

True if the body is async def. Calling the loop then returns a coroutine.

copy

copy(*, until: Any = ..., limit: Any = ...) -> Loop

Returns a new Loop with until or limit changed. The original is unchanged.

Example

quick = coding.copy(limit=5)

Parameters:

Name	Type	Description	Default
until	Any	New stop conditions. Omit to keep the current ones.	...
limit	Any	A new turn limit. Omit to keep the current one.	...

Raises:

Type	Description
TypeError	A new value fails the same checks as @loop.
ValueError	A new value fails the same checks as @loop.

default_loop

default_loop(agent: Agent, state: State)

The loop agent.run uses when the Agent has no loop of its own.

Each turn compacts the messages if the context is over 60% full, asks the model, and runs the tools it asked for. Stops when waiting_for_user is true (stopped by waiting_for_user) or after 50 turns. Copy it as a starting point for your own loop.

adefault_loop async

adefault_loop(agent: Agent, state: State)

The async version of default_loop, used by agent.arun when the Agent has no loop of its own.

waiting_for_user

waiting_for_user(state: State) -> bool

True when the model has handed the turn back: nothing waits for a tool result, and the last message is the model's reply without tool calls.

The default loops stop on it, and so can yours: @loop(until=waiting_for_user, limit=30). It says the model is waiting for the user, not that the task is done; to end a task for good, call state.finish(answer). A message added after the reply (state.add_message) makes it false again, so the loop goes on. For another rule, write your own State -> bool function.

Reads one snapshot, so the checks see the same State even while another thread adds a message.

Example

from alpineagents import Agent, State, loop, waiting_for_user

@loop(until=waiting_for_user, limit=30)
def coding(agent: Agent, state: State):
    agent.think(state)
    if state.pending_calls:
        agent.use_tools(state)

CompactIfFull

CompactIfFull(at: float = 0.6, instructions: str | None = None)

A block that compacts the messages when the context is fuller than at.

Call it at the start of a turn, before agent.think. compact_if_full is one with the defaults (at=0.6, no instructions). Compaction replaces the messages with the first message and a summary and rebuilds the provider's cache. In an async loop use acall (or acompact_if_full): calling it in an async run raises TypeError the first time it compacts, which can be many turns in.

Example

CompactIfFull(at=0.5, instructions="Keep file paths and test results")(agent, state)

Parameters:

Name	Type	Description	Default
at	float	The fraction of the model's context window (agent.context_used(state)) above which to compact.	0.6
instructions	str | None	What the summary must keep, added to the default summary prompt.	None

at instance-attribute

at = at

The fraction of the context window above which to compact.

instructions instance-attribute

instructions = instructions

What the summary should keep, or None.

acall async

acall(agent: Agent, state: State)

The async version, for async loops: await CompactIfFull(at=0.5).acall(agent, state).

compact_if_full module-attribute

compact_if_full = CompactIfFull()

Compacts the context when it is more than 60% full. Call it as compact_if_full(agent, state) at the start of a turn. The same as CompactIfFull().

acompact_if_full module-attribute

acompact_if_full = compact_if_full.acall

The async version of compact_if_full, for async loops: await acompact_if_full(agent, state).