Lengthy-horizon brokers and sturdy execution

The agentic loop

Most brokers, from coding brokers similar to Claude Code to non-public
assistants similar to OpenClaw, are constructed across the identical loop, normally
known as the agentic loop. It has 4 elements:

• a objective: the end result we wish the agent to realize. For a
  coding agent, this is perhaps a program to jot down; for Inventory
  Replenisher, an element saved in inventory.
• instruments by means of which the agent acts on this planet. Consider
  read_file, edit_file, and
  bash for a coding agent; get_quotes and
  place_order for Inventory Replenisher.
• state: all the things the agent is aware of about its progress
  towards the objective to this point.
• a mannequin that, at every flip, seems on the present state and
  chooses which software to make use of subsequent, and what to move it.

At every flip, the mannequin chooses a software, and the code round it runs
the software and updates the state with the consequence. That code is commonly
known as the harness; Claude Code, for instance, is a harness
round Claude. The loop ends when the agent reaches the objective or offers
up. When the agent wants enter from an individual or an exterior supply, it
asks and waits for a solution.

Right here is Inventory Replenisher’s loop in code. (All code examples are in
Python.)

GOAL = ("Deliver inventory again above security inventory: order from an appropriate "
        "provider, get each value authorized by the supervisor earlier than cash "
        "is dedicated, and see the order by means of to supply.")
MAX_TURNS = 30

TOOLS = [
    GetQuotes(), DraftPurchaseOrder(), RequestApproval(),
    WaitForApproval(), PlaceOrder(), WaitForSupplierReply(),
    ReadSupplierReply(), WaitForGoods(), ClosePurchaseOrder(), Done(),
    ...
]

class Replenish:
    async def run(self, inventory: StockLevel) -> ReplenishmentResult:
        state = State(inventory.sku, inventory.on_hand, inventory.safety_stock)
        for flip in vary(MAX_TURNS):
            allowed_tools = [t for t in TOOLS if t.allowed(state)]
            software = await self.decider.resolve(GOAL, state, allowed_tools)
            if isinstance(software, Finished):
                break
            if isinstance(software, Wait):
                attempt:
                    payload = await asyncio.wait_for(
                        self.inbox[tool.event].get(),
                        timeout=timedelta(days=software.days).total_seconds())
                    software.apply(state, payload)
                besides TimeoutError:
                    software.apply(state, None)
            else:
                consequence = await software.execute(state, self.context)
                software.apply(state, consequence)
        return end result(state)

decider.resolve is the decision to the mannequin (in our case, a
classification mannequin Jev); it picks the software to run in
the present flip primarily based on the GOAL, the present state of the replenishment run
(state), and a prefiltered listing of allowed instruments
allowed_tools. The loop ends when the mannequin selects
Finished or after MAX_TURNS. The
end result exhibits how the replenishment concluded: acquired,
cancelled, rejected by the supervisor, or given up.

Every software has a reputation and outline, which the mannequin makes use of to make
its choice. The allowed methodology states whether or not the software
can be utilized given the replenishment’s present state. This manner the
enterprise guidelines that matter are in code, not within the immediate. For
occasion, the place_order software is not obtainable till the
supervisor approves the worth on the desk. The execute
methodology runs the software, e.g., makes API calls, writes an e mail, and so on., and
apply updates state primarily based on the
results of execute.

class PlaceOrder(Name):
    identify = "place_order"
    description = "Ship the authorized buy order to the provider."

    def allowed(self, state):
        return (state.po just isn't None and state.order is None
                and state.approved_up_to just isn't None
                and state.value 

state holds all the things the agent is aware of about one
        replenishment: the quotes, buy order, authorized value vary, and
        order. Solely the apply methodology updates it primarily based on the
        results of every flip.

@dataclass
class State:
    sku: str
    on_hand: int
    safety_stock: int
    quotes: listing[Quote] | None = None
    po: PurchaseOrder | None = None
    value: float | None = None
    approved_up_to: float | None = None
    order: OrderAck | None = None
    items: GoodsReceipt | None = None
    ...

Wait can be a software that waits for a message, such because the

        supervisor's choice, or till its time runs out. This loop works as

        lengthy as the method working it stays up, however quite a bit can go improper throughout

        lengthy wait instances. And the agent must get better and proceed its

        process.

What can go improper

It helps to consider a long-horizon agent as a distributed system.
Lots of the failure modes we’re conversant in in distributed programs
occur right here too: calls to different programs fail, processes crash whereas the
agent waits, and so forth.

The primary 4 failures are the identical in any area. A sturdy
execution runtime handles them with general-purpose primitives in order that
the harness can deal with domain-specific logic. With out the runtime,
you would need to write a number of glue code to deal with the failures
your self: file what has been finished, retry, set timers, and choose up
the place you left off.

However treating the agent as simply one other distributed system misses
one thing. What makes it a unique beast is the mannequin at its coronary heart,
which might resolve how one can get better from failures and hold the duty going.
When a failure happens, the harness’s job is to supply the mannequin with
the appropriate enter so it will probably resolve what to do subsequent.

The subsequent sections have a look at sturdy execution intimately. After that,
we return to the failures the harness should deal with explicitly.

Sturdy execution

Sturdy execution is a programming mannequin through which the
runtime retains a file of the results of every step that interacts with
the skin world in order that, when a failure happens, this system runs
once more, accomplished steps are skipped, and this system’s state is rebuilt
from the file.

Temporal, Restate, Inngest, DBOS, and Cloudflare Workflows all
implement it. We constructed Inventory Replenisher on two of them, Temporal and
Cloudflare Workflows. The code on this article comes from the Temporal model.
Cloudflare seems the place the 2 differ materially.

The definition rests on three concepts: steps, a log, and replay.

A step is a name to the skin world that the runtime
makes on this system’s behalf: the LLM name, a name to the ERP, a
message to the provider. Something that interacts with the skin world
goes by means of a step. The code between steps solely reads from and writes
to native state. In Inventory Replenisher, decider.resolve and every
software’s execute are steps; allowed and
apply aren’t.

Determine 2: Each name
to the skin world is a step

Earlier than this system proceeds to the following step, the runtime writes
each the step’s enter and output to a persistent log.

Determine 3: A step,
close-up

After a crash, this system restarts from the highest (i.e., it’s
replayed). Any step with a recorded output returns it from
the log reasonably than working once more. The code
between steps runs once more and rebuilds this system state, which in
Inventory Replenisher is state. Step one and not using a
file runs for actual.

Determine 4: After a crash, this system runs once more; recorded steps do not

This manner this system can simply rebuild its state from the log after
a crash.

  from temporalio import workflow
  
  @workflow.defn
  class Replenish:
      @workflow.run
      async def run(self, inventory: StockLevel) -> ReplenishmentResult:
          state = State(inventory.sku, inventory.on_hand, inventory.safety_stock)
          for flip in vary(MAX_TURNS):
              allowed_tools = [t for t in TOOLS if t.allowed(state)]
              software = await workflow.execute_activity(
                  Decider.resolve, args=[GOAL, state, allowed_tools],
                  activity_id=f"decide-{flip}", ...)
              if isinstance(software, Finished):
                  break
              if isinstance(software, Wait):
                  attempt:
                      await workflow.wait_condition(
                          lambda: self.inbox[tool.event],
                          timeout=timedelta(days=software.days))
                      software.apply(state, self.inbox[tool.event].pop(0))
                  besides TimeoutError:
                      software.apply(state, None)
              else:
                  consequence = await workflow.execute_activity(
                      run_tool, args=[tool, state],
                      activity_id=f"{software.identify}-{flip}", ...)
                  software.apply(state, consequence)
          return end result(state)

The entity and the workflow

We have seen the sturdy agentic loop. The subsequent query is the place
that loop lives and the way it maps to the enterprise area.

Take inventory replenishment: there are two loops at play. One handles
replenishment for a single SKU and follows the work by means of till the
items are acquired. The opposite is the SKU-level loop. It checks inventory
in opposition to the replenishment coverage and triggers replenishment when
wanted, separately. It's also the SKU's level of contact: messages
from the supervisor, the suppliers, and the warehouse come to it. The SKU
shall be replenished many instances over its life.

Determine 5: One
agent per SKU; one workflow at a time

That is the excellence between a course of workflow and an
entity workflow.

A course of workflow is a sturdy workflow that carries
out one piece of labor (e.g., a replenishment) after which finishes.

An entity workflow is a sturdy workflow that
represents one enterprise entity (e.g., a SKU) for the lifetime of that
entity. It maintains the entity's state and initiates course of
workflows.

In a subscription enterprise, prospects match the identical sample. The
buyer's entity workflow maintains the shopper context and
initiates the suitable course of workflows (onboarding, renewal,
help, and so on.). Every of these course of workflows completes, however the
buyer entity workflow continues.

The entity workflow's first job is to observe the inventory and set off
replenishments.

the entity workflow: one per SKU

  @workflow.defn
  class SkuAgent:
      @workflow.run
      async def run(self, sku: str, outcomes: listing[ReplenishmentResult]):
          inventory = await workflow.execute_activity(get_stock, sku, ...)
          if inventory.on_hand 
              self.lively = f"replenish-{sku}-{len(outcomes) + 1}"
              attempt:
                  consequence = await workflow.execute_child_workflow(
                      Replenish.run, inventory, id=self.lively)
              besides ChildWorkflowError as e:
                  consequence = ReplenishmentResult(sku, "failed", str(e.trigger))
              outcomes.append(consequence)
              self.lively = None
          else:
              await workflow.sleep(CHECK_EVERY)
          workflow.continue_as_new(args=[sku, outcomes])

Every run of the entity workflow checks the inventory as soon as. If the SKU is
under security inventory, it begins a replenishment and waits for it to
end; in any other case it sleeps till the following test. Then
continue_as_new begins a contemporary run with the identical ID and an
empty historical past, which receives solely what it's given, right here the SKU and
the outcomes to this point.

The entity workflow's second job is to obtain messages for the SKU
from the skin world. It has a steady, distinctive ID,
agent-SKU-123, for the SKU's life. Every request the agent
sends out, similar to an approval request or a purchase order order, carries the
SKU or the acquisition order quantity in order that the reply will be despatched again
to agent-SKU-123. The supervisor approving or
the provider replying would not have to know which replenishment is
working; the entity passes the message on to the present one.

the supervisor's approval, despatched to SKU-123

  await shopper.get_workflow_handle("agent-SKU-123").sign(
      "ship", args=["approval", {"approved": True}])

the entity's handle: messages from exterior

  class SkuAgent:
      ...
      @workflow.sign
      async def ship(self, occasion: str, payload: dict):
          if self.lively:
              await workflow.get_external_workflow_handle(self.lively).sign(
                  Replenish.ship, args=[event, payload])

the replenishment's inbox: messages from the entity

  @workflow.defn
  class Replenish:
      @workflow.run
      async def run(self, inventory: StockLevel) -> ReplenishmentResult:
          state = State(inventory.sku, inventory.on_hand, inventory.safety_stock)
          for flip in vary(MAX_TURNS):
              ...
          return end result(state)
  
      @workflow.sign
      def ship(self, occasion: str, payload: dict):
          self.inbox[event].append(payload)

Each message for SKU-123 goes by means of SKU-123's entity workflow
to the working replenishment, which applies them separately.

Within the code, Inventory Replenisher is now two courses:
SkuAgent, the entity workflow, and Replenish,
the method workflow that runs the loop from earlier.

That brings us to the ready.

Ready for folks and programs

As we have seen to this point, long-horizon brokers spend a lot of their time
ready for people or exterior programs. The sturdy execution runtime
supplies primitives for ready effectively. Whereas these brokers wait,
the runtime ensures they do not maintain a course of in reminiscence or use
compute.

This issues as a result of, with the entity-workflow sample we noticed
earlier, each SKU could have its personal workflow, and that workflow will
run so long as the SKU is stocked. A producer with 1000's of
elements could have 1000's of brokers. At any given time, virtually all of
them shall be idle or ready. If a wait holds a course of or reminiscence, then
the price will increase in proportion to the variety of SKUs and the size
of the waits, not in proportion to the work truly finished.

workflow.wait_condition suspends the workflow till a
message of the appropriate kind is within the inbox, or the timeout expires. The
wait's timer can be logged, and so is the message that ends it. So a
wait survives a crash simply as a step does.

Determine 6: The wait's timer and the message each go into the log

Waking up has a value too. Rebuilding a workflow means replaying the
log, and that value grows with the log's size. If wake-up is sluggish or
costly, then having 1000's of brokers ready is simply as
infeasible as preserving 1000's of processes open for weeks. In our
runs, waking a workflow took beneath three seconds and tens of
milliseconds of CPU.

If replay will get too sluggish, the following factor to attempt is a snapshot, so solely the log after it needs to be
replayed. This issues most for the entity, whose log would hold
rising so long as the SKU is stocked. On Temporal,
continue_as_new has the identical impact: it carries the
entity's state right into a contemporary run with an empty historical past. On Cloudflare,
the entity is a Sturdy Object, which reads its present state from
storage and has no log to replay. A replenishment workflow would not want
a snapshot. Its log stays brief as a result of the run ends.

A wait ends in one among two methods: a message arrives, or the timeout
expires. If the provider hasn't received again to us inside seven days, then
wait_for_supplier_reply will mark the order as
overdue:

class WaitForSupplierReply(Wait):
    occasion, days = "supplier-reply", 7
    ...
    def apply(self, state, payload):
        if payload is None:
            state.supplier_overdue = True
        else:
            state.supplier_reply = SupplierReply(**payload).message

class ChaseSupplier(Name):
    ...
    def allowed(self, state):
        return (state.supplier_overdue and state.chases 
                and is_open(state))

Which means chase_supplier is allowed, and the mannequin
will see it as an choice on its subsequent flip.

On this run, the provider goes quiet after the order is positioned. Look
for the wait that ends with a timeout seven days later, then the chase
that follows it, and the reply a day after the chase.

Retrying failed calls, at the very least as soon as

Calls to exterior programs can fail due to community points or
as a result of the receiving system is down. Sturdy execution runtimes
present primitives to retry failed calls with configurable retry
insurance policies, so they do not overload the receiving system.

RETRY = RetryPolicy(maximum_attempts=6,
                    initial_interval=timedelta(seconds=2),
                    backoff_coefficient=2)

@workflow.run
async def run(self, inventory: StockLevel) -> ReplenishmentResult:
    ...
    consequence = await workflow.execute_activity(
        run_tool, args=[tool, state],
        activity_id=f"{software.identify}-{flip}", retry_policy=RETRY)
    ...

What the runtime cannot do is make a retry secure. If the provider
takes the order and the response is misplaced, no result's recorded, so
the runtime calls the provider once more. This is called at-least-once semantics, as
against exactly-once. This might end in duplicate
orders if not dealt with correctly.

To get the impact of precisely as soon as for a write, the receiver should
recognise repeats utilizing an idempotency key, which the harness should
hold the identical on each try and replay.

Within the place_order software, the PO quantity turns into the
idempotency key.

class PlaceOrder(Name):
    ...
    async def execute(self, state, context):
        ...
        key = state.po.po_number
        ack = await context.provider.place_order(po, idempotency_key=key)
        return PlaceOrderResult(positioned=True, value=value, ack=ack)

Within the run under, the provider's system is down on the primary
try. The runtime retries with the identical PO quantity as its
idempotency key. Search for the 2 makes an attempt on place_order and the
single order on the provider.