Workflows: small, auditable, resumable pipelines

State in, state out

A node receives the shared state and returns a new state. That is the whole contract. Here is a two-step procedure that drafts an answer and then tightens it, with the state carrying the work from one step to the next:

wf <- agent_workflow("draft_then_polish") |>
  add_node("draft", function(state) {
    a <- agent("Drafter", cfg)
    state$draft <- a$reply(paste("Draft two sentences on:", state$topic))
    state
  }) |>
  add_node("polish", function(state) {
    a <- agent("Editor", cfg)
    state$final <- a$reply(paste("Tighten this to one sentence:", state$draft))
    state
  }) |>
  add_edge("draft", "polish")

run <- run_workflow(wf, input = list(topic = "why peer review is slow"))
run$state$final

run_workflow() returns a record of what happened: the sequence of nodes, the state at each step, and a status. Because every node’s state is hashed, the run is checkable, which is what makes the next two features possible.

Branches and loops

An edge can carry a condition. The runtime takes the first edge whose condition holds, so you can branch on what a node produced, or loop until something is good enough. Loops are bounded by max_steps: a runaway procedure stops rather than spends, raising a typed error you can catch.

wf2 <- agent_workflow("until_short_enough") |>
  add_node("write", function(state) {
    a <- agent("Writer", cfg)
    state$tries <- (state$tries %||% 0L) + 1L
    state$text <- a$reply(paste0("Write a one-line summary of: ", state$topic,
                                 " (attempt ", state$tries, ")"))
    state
  }) |>
  add_node("done", function(state) state) |>
  add_edge("write", "write", when = function(state) nchar(state$text) > 120 && state$tries < 3) |>
  add_edge("write", "done", when = function(state) nchar(state$text) <= 120 || state$tries >= 3)

run2 <- run_workflow(wf2, input = list(topic = "the history of the printing press"),
                     max_steps = 10)
run2$state$tries   # how many attempts it took

Checkpoints, resume, and fork

Give run_workflow() a checkpoint_dir and it writes the state after every node as it goes. If the run fails partway, from a network interruption or a model timeout, you resume from where it stopped rather than starting over, and completed nodes do not run again.

dir <- file.path(tempdir(), "long_run")
run3 <- run_workflow(wf, input = list(topic = "open access publishing"),
                     checkpoint_dir = dir)
# had it failed, resume_workflow(run3, wf) would continue from the last good node

fork_workflow() branches an existing run at a chosen point, so you can try a different continuation from a shared prefix without recomputing it – useful for asking “what if the editor had been stricter” without paying for the draft twice.

Replay verification

A run records a state hash at each step. replay_run() re-executes the procedure and compares. The comparison accounts for nondeterminism: a deterministic node must reproduce its hash exactly, while a model node is checked for design identity rather than identical text, because a model does not return the same words twice. A divergence names the first step that differs.

rp <- replay_run(run, wf, verify = "structural")
rp$steps[, c("node", "replay_match")]

Use verify = "strict" when the whole procedure is meant to be deterministic and any difference is a problem. Use the default "structural" when model nodes are involved and you want to confirm the structure of the run without demanding identical wording. Either way, a tampered checkpoint is caught: the hash will not match.

When to reach for this

If deliberate() or agent_pipeline() already does what you need, use them; the workflow runtime is not a replacement and workflow_from_pipeline() exists only to show that the engine can express them. Reach for agent_workflow() when you need a branch, a loop, a human gate, or durability across a long run. The added setup adds checkpointing, resumption, forking, and replay verification: a pipeline you can leave and return to, and one a reader can verify ran the way you say it did.