Loops and blocks

Concepts: Loops.

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

@loop(until=State.is_answered, limit=50)
def coding(agent: Agent, state: State):
    compact_if_full(agent, state)
    agent.think(state)
    if state.wants_tools():
        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_by == "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.is_finished(): stops with state.stopped_by == "finish"
2. each until function: stops with that function's name
3. the turn count against limit: stops with "limit"

Each call counts turns from zero. 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.

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. State.is_answered. A lambda works but leaves no useful name in stopped_by, 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 (state.is_answered()) or a bound method (state.is_answered), an until item is not callable, or limit is not an integer.
ValueError	limit is less than 1, or an until function is named finish or limit (reserved names).

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 context if it is over 60% full, asks the model, and runs the tools it asked for. Stops when the model answers without tool calls (State.is_answered) 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.

CompactIfFull

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

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

Call it at the start of a turn, before agent.think. compact_if_full is one with the defaults.

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 (state.context_used) 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).