node

Adds a node named name that runs action and returns a reference to it for use with then.

Throws

Adds a node that is tried again when it fails, that has a time limit, or both.

val fetch = node("fetch", retry = Retry(attempts = 3), timeout = 30.seconds) { order ->
    order.copy(stock = warehouse.stock(order.item))
}

With a retry, a node that fails runs again on the same state, in the same step: see Retry for how often and with which pause. Only when its last try fails does the run fail, with a NodeExecutionException. Nothing is saved between two tries, and the other nodes of the step go on while this one waits for its next try. A stream gets a GraphEvent.NodeRetry for each try that failed.

With a timeout, a try that takes longer is cancelled and fails with a NodeTimeoutException, which a retry treats like any other failure. The limit stops code that suspends, such as a call with a Ktor client. Code that blocks its thread goes on until it returns, and the node fails then.

A node that is tried again starts from its first line, so make its side effects safe to repeat.

Parameters

Throws

Adds a node whose job is split in two, so that it can run next to other nodes without a Reducer. work does the slow part, such as a call to a model or a service, and returns a result. update writes that result into the state.

val web = node("web", work = { searchWeb(it.question) }) { state, found ->
    state.copy(findings = state.findings + found)
}

When several nodes run in the same step, their work runs in parallel on the same state. Their updates are then applied one after another, in the order the nodes were added to the graph, each to the state that the previous one produced. No change is lost, and if two nodes write the same property, the node that was added later wins.

update must only build the new state. The engine may call it more than once for one run of the node.

Throws

Adds a node with a work and an update that is tried again when it fails, that has a time limit, or both. See the overload of a node that returns a whole state for what retry and timeout do.

val classify = node(
    "classify",
    work = { email -> model.ask<Verdict>("Classify this email: ${email.body}") },
    retry = Retry(attempts = 3),
) { email, verdict -> email.copy(category = verdict.category) }

A try is the work of the node. In a step with several nodes, the update runs once, after the work of every node of the step has succeeded.

Parameters

Throws