Loops

loop runs an inner step over and over, accumulating typed results, until your predicate returns True. Iteration is bounded — there is a hard cap, and exhausting it raises LoopExhaustedError.

Signature

from bridle import loop

loop(
    prompt: str,
    *,
    schema: type[T],
    until: Callable[[list[T]], bool],
    context: Any = None,
    tools: Sequence[Tool] = (),
    max_iterations: int = 32,
    label: str | None = None,
) -> list[T]

Returns a Call typed as list[T]. Resolves on iteration, length, or attribute access.

The shape

sources = loop(
    "gather a source on Mars weather",
    schema=Source,
    until=lambda acc: len(acc) >= 3,
    tools=[search],
)

Each iteration:

1. Runs an inner step with the original context plus a running accumulator.
2. Appends the result to the accumulator.
3. Calls until(accumulator). If it returns True, returns the accumulator.

The accumulator is built in pure Python. The model never sees the predicate — it just keeps producing the next value.

Bounded by design

max_iterations (default 32) caps the number of steps. If the predicate has not fired by then, loop raises LoopExhaustedError carrying the partial accumulator:

from bridle import LoopExhaustedError

try:
    sources = loop(
        "gather a source on a niche topic",
        schema=Source,
        until=lambda acc: len(acc) >= 100,
        max_iterations=10,
    )
except LoopExhaustedError as exc:
    partial: list = exc.accumulator
    # Decide what to do with the partial result.

Set max_iterations to whatever ceiling you actually want. Smaller is safer.

What until should do

Three rules.

Pure Python, no model calls. The predicate runs after each iteration; making it a model call would defeat the point. Hold any model judgment inside the inner step's prompt.

Read the accumulator, not state outside it. The predicate sees only the list of results so far. If you need to compare against external state, capture it via closure:

seen_urls: set[str] = set()

def enough(acc: list[Source]) -> bool:
    for s in acc:
        seen_urls.add(s.url)
    return len(seen_urls) >= 5

sources = loop("gather a source", schema=Source, until=enough, tools=[search])

Cheap. It runs after every iteration. Don't do anything expensive in it.

Each iteration sees previous results

The inner step's context is {"original_context": ..., "previous_results": [...], "iteration": N}. The model sees what has been gathered so far and can avoid duplicates. If duplicates do slip through, dedupe in the predicate or post-process the accumulator after the loop returns.

Pitfalls

Predicates that never return True. If the predicate cannot be satisfied with the schema's values (e.g. "until the model produces something with summary='exact match'"), the loop runs to max_iterations and raises. Make the predicate something the next iteration can plausibly achieve.

Schemas that drift over iterations. The schema is fixed for the loop. If you need the structure to change ("first three are summaries, last one is a synthesis"), you want a step after the loop, not a different schema mid-loop.

Using loop for parallelism. Loop iterations are sequential. Concurrent gathering arrives with the parallel primitive in v0.2.

A complete example

import bridle
from bridle import agent, loop, tool
from bridle.models.anthropic import install
from pydantic import BaseModel


class Source(BaseModel):
    url: str
    summary: str


@tool
def search(query: str) -> list[str]:
    """Search the web. Returns up to 10 result URLs."""
    slug = query.lower().replace(" ", "-")[:40]
    return [f"https://example.com/{slug}/{i}" for i in range(5)]


@agent(input=str, output=list[Source], model="claude-sonnet-4-6")
def gather(topic: str) -> list[Source]:
    return loop(
        f"gather a distinct source on {topic}",
        schema=Source,
        until=lambda acc: len({s.url for s in acc}) >= 3,
        tools=[search],
        max_iterations=8,
    )


install()
sources = bridle.resolve(gather("Mars weather"))
for s in sources:
    print(f"- {s.url}: {s.summary}")

Next