Prebuilt patterns¶

Ready-made graph patterns and node wrappers. See the Prebuilt patterns guide. Quality gates and idempotency helpers have their own reference page.

agentflow.prebuilt ¶

Prebuilt graph patterns built entirely from core primitives.

Tool `dataclass` ¶

A callable the model may invoke, plus the schema advertised to it.

fn may be sync or async and is called with the model-supplied argument dict as keyword arguments. Its return value becomes the tool result content (stringified for the model).

Source code in agentflow/prebuilt/tool_loop.py

@dataclass(frozen=True)
class Tool:
    """A callable the model may invoke, plus the schema advertised to it.

    ``fn`` may be sync or async and is called with the model-supplied argument
    dict as keyword arguments. Its return value becomes the tool result
    content (stringified for the model).
    """

    name: str
    fn: Callable[..., Any]
    description: str = ""
    schema: dict[str, Any] = field(default_factory=dict)

    def spec(self) -> ToolSpec:
        return ToolSpec(name=self.name, description=self.description, schema=self.schema)

ToolLoopState ¶

Bases: State

Conversation plus loop bookkeeping.

Source code in agentflow/prebuilt/tool_loop.py

class ToolLoopState(State):
    """Conversation plus loop bookkeeping."""

    messages: Annotated[list, append]  # the running chat transcript
    turns: Annotated[int, add]  # model calls made
    _pending: Annotated[list, last]  # tool calls awaiting execution
    status: Annotated[str, last]  # terminal status (see tool_loop docs)

Candidate `dataclass` ¶

One turn's result. done lets the worker end the loop itself.

Source code in agentflow/prebuilt/loop.py

@dataclass(frozen=True)
class Candidate:
    """One turn's result. ``done`` lets the worker end the loop itself."""

    score: float
    payload: Any = None
    done: bool = False

iterate_until_converged ¶

iterate_until_converged(work: Worker, *, perfect_score: float = 100.0, patience: int = 4, max_iterations: int = 20, checkpointer: Any = None)

Compile a loop graph around work.

Stop conditions, in priority order:

the worker returns Candidate(done=True) -> stop_reason="done"
best score reaches perfect_score -> stop_reason="optimal"
no improvement for patience rounds -> stop_reason="converged"
max_iterations reached -> stop_reason="exhausted"

The final state carries best_score, best_payload, stop_reason, iterations, and history (a list of per-turn scores).

Source code in agentflow/prebuilt/loop.py

def iterate_until_converged(
    work: Worker,
    *,
    perfect_score: float = 100.0,
    patience: int = 4,
    max_iterations: int = 20,
    checkpointer: Any = None,
):
    """Compile a loop graph around ``work``.

    Stop conditions, in priority order:

    - the worker returns ``Candidate(done=True)`` -> ``stop_reason="done"``
    - best score reaches ``perfect_score`` -> ``stop_reason="optimal"``
    - no improvement for ``patience`` rounds -> ``stop_reason="converged"``
    - ``max_iterations`` reached -> ``stop_reason="exhausted"``

    The final state carries ``best_score``, ``best_payload``, ``stop_reason``,
    ``iterations``, and ``history`` (a list of per-turn scores).
    """

    async def step(state: dict, ctx) -> dict:
        candidate = await work(state)
        prev_best = state.get("best_score", float("-inf"))
        improved = candidate.score > prev_best

        update: dict[str, Any] = {
            "iterations": 1,
            "last_score": candidate.score,
            "history": candidate.score,
        }
        if improved:
            update["best_score"] = candidate.score
            update["best_payload"] = candidate.payload
            update["streak"] = 0
        else:
            update["streak"] = state.get("streak", 0) + 1

        best_now = candidate.score if improved else prev_best
        iters = state.get("iterations", 0) + 1

        if candidate.done:
            update["stop_reason"] = "done"
            update["_done"] = True
        elif best_now >= perfect_score:
            update["stop_reason"] = "optimal"
            update["_done"] = True
        elif update.get("streak", 0) >= patience:
            update["stop_reason"] = "converged"
            update["_done"] = True
        elif iters >= max_iterations:
            update["stop_reason"] = "exhausted"
            update["_done"] = True
        else:
            update["_done"] = False

        return update

    def route(state: Mapping[str, Any]) -> str:
        return "stop" if state.get("_done") else "again"

    g = Graph(_LoopState)
    g.add_node("step", step)
    g.add_edge(START, "step")
    g.add_conditional_edges("step", route, {"again": "step", "stop": END})
    # A generous engine-level guard well above max_iterations; the loop's own
    # counter is the real terminator.
    return g.compile(checkpointer=checkpointer, step_limit=max_iterations + 5)

with_retry ¶

with_retry(node: Node, *, retries: int = 2, backoff: float = 0.1, backoff_factor: float = 2.0, max_backoff: float = 30.0, on: tuple[type[BaseException], ...] = (Exception,), jitter: float = 0.0) -> Node

Wrap node to retry on failure with exponential backoff.

retries is the number of additional attempts after the first, so the node runs up to retries + 1 times.
on selects which exception types trigger a retry; anything else propagates immediately.
:class:InterruptError is always excluded, even if on would match it, so HITL suspends are never retried.
After the last attempt the final exception is re-raised unchanged.

backoff is the first delay; each retry multiplies it by backoff_factor up to max_backoff. jitter adds up to that many seconds of uniform random delay (0 disables it, keeping tests deterministic).

Source code in agentflow/prebuilt/resilience.py

def with_retry(
    node: Node,
    *,
    retries: int = 2,
    backoff: float = 0.1,
    backoff_factor: float = 2.0,
    max_backoff: float = 30.0,
    on: tuple[type[BaseException], ...] = (Exception,),
    jitter: float = 0.0,
) -> Node:
    """Wrap ``node`` to retry on failure with exponential backoff.

    - ``retries`` is the number of *additional* attempts after the first, so
      the node runs up to ``retries + 1`` times.
    - ``on`` selects which exception types trigger a retry; anything else
      propagates immediately.
    - :class:`InterruptError` is always excluded, even if ``on`` would match
      it, so HITL suspends are never retried.
    - After the last attempt the final exception is re-raised unchanged.

    ``backoff`` is the first delay; each retry multiplies it by
    ``backoff_factor`` up to ``max_backoff``. ``jitter`` adds up to that many
    seconds of uniform random delay (0 disables it, keeping tests deterministic).
    """
    if retries < 0:
        raise ValueError("retries must be >= 0")

    async def wrapped(state: Mapping[str, Any], ctx: Any) -> Mapping[str, Any] | None:
        import random

        attempt = 0
        delay = backoff
        while True:
            try:
                return await _maybe_await(node(state, ctx))
            except InterruptError:
                raise  # never retry a human-in-the-loop suspend
            except on as exc:  # noqa: B902 - configurable exception set
                attempt += 1
                if attempt > retries:
                    raise
                wait = delay + (random.uniform(0, jitter) if jitter else 0.0)
                if wait > 0:
                    await asyncio.sleep(wait)
                delay = min(delay * backoff_factor, max_backoff)
                # loop to retry; keep last exc bound only for clarity
                del exc

    wrapped.__name__ = f"with_retry({getattr(node, '__name__', 'node')})"
    return wrapped

with_timeout ¶

with_timeout(node: Node, seconds: float) -> Node

Wrap node so it fails if it runs longer than seconds.

Raises :class:asyncio.TimeoutError on expiry (the runtime wraps it as a NodeError). An :class:InterruptError still propagates: a suspended node is not a timeout.